import ftplib import io import ssl import subprocess import textwrap import time import test_quadlet # noqa: F401 import pytest import testinfra LEGO_CONFIG = "/etc/quadlets/lego/lego.yaml" LEGO_CERT_PATH = "/var/lib/quadlets/lego/certificates/default.crt" FTP_USER = "ftptest" FTP_UID = 1500 # vsftpd.container rewrites the home of uids 1000-1999 to /data FTP_PASSWORD = "S3cret-ftp-passw0rd" def prime_dns_cache(dns_server_ip: str, names: list[str]) -> None: """Workaround for flaky ACME validations in the test environment. Pebble resolves the domains over TCP. When the libvirt dnsmasq has to forward the AAAA query of a test domain upstream over TCP, the query may time out. A UDP query puts the (negative) answer in the dnsmasq cache, so that the subsequent TCP query of Pebble succeeds. """ for name in names: for qtype in ("A", "AAAA"): subprocess.run(["dig", "+time=2", "+tries=2", qtype, f"@{dns_server_ip}", name], capture_output=True) class ReusedSessionFTP_TLS(ftplib.FTP_TLS): """FTP_TLS client that reuses the TLS session of the control connection on the data connections. vsftpd requires TLS session reuse by default (require_ssl_reuse=YES). """ def ntransfercmd(self, cmd, rest=None): conn, size = ftplib.FTP.ntransfercmd(self, cmd, rest) if self._prot_p: conn = self.context.wrap_socket(conn, server_hostname=self.host, session=self.sock.session) return conn, size @pytest.fixture(scope="module") def dns_names() -> list[str]: """The FTP domain must be DNS-resolvable so Pebble can reach the VM on port 80 for HTTP-01 validation.""" return ["ftp"] @pytest.fixture(scope="module") def ftp_domain(top_level_domain) -> str: return f"ftp.{top_level_domain}" @pytest.fixture(scope="module") def fcos_extra_files(pebble_acme_server, ftp_domain) -> dict: """Extra files to inject into the FCOS VM for vsftpd tests.""" return { # Lego configuration pointing at the Pebble ACME test server. LEGO_CONFIG: ( textwrap.dedent(f"""\ storage: /var/lib/lego log: format: text accounts: default: server: {pebble_acme_server['directory_url']} email: test@{ftp_domain} acceptsTermsOfService: true challenges: http: http: address: ":80" certificates: default: account: default challenge: http domains: - {ftp_domain} renew: disableRandomSleep: true hooks: deploy: command: /bin/sh /etc/lego/deploy-hook.sh """), 10023, # owner: lego 10000, # group: itix-svc 0o644, ), # Pebble root CA certificate, needed by lego to authenticate the Pebble ACME server. "/etc/quadlets/lego/pebble.pem": ( pebble_acme_server["ca_cert"], 0, 0, 0o644, ), # Drop-in to trust the Pebble CA. "/etc/containers/systemd/lego.container.d/pebble.conf": ( textwrap.dedent("""\ [Container] Environment=LEGO_CA_CERTIFICATES=/etc/lego/pebble.pem """), 0, 0, 0o644, ), # vsftpd configuration: local users + TLS with the certificate provided by lego. "/etc/quadlets/vsftpd/vsftpd.conf.d/local.conf": ( textwrap.dedent(f"""\ pasv_address={ftp_domain} pasv_addr_resolve=YES local_enable=YES local_root=/data chroot_local_user=YES pam_service_name=vsftpd write_enable=YES allow_writeable_chroot=YES anonymous_enable=NO file_open_mode=0660 ssl_enable=YES ssl_request_cert=NO ssl_tlsv1_2=NO ssl_tlsv1_3=YES rsa_cert_file=/etc/vsftpd/tls/default.crt rsa_private_key_file=/etc/vsftpd/tls/default.key """), 0, 0, 0o644, ), # Storage shared through FTP. "/etc/tmpfiles.d/vsftpd-test.conf": ( f"d /var/lib/vsftpd-test 0770 {FTP_UID} {FTP_UID} -\n", 0, 0, 0o644, ), "/etc/containers/systemd/vsftpd.container.d/storage.conf": ( textwrap.dedent("""\ [Container] Volume=/var/lib/vsftpd-test:/data:z """), 0, 0, 0o644, ), } """ Verify that vsftpd serves FTPS with a certificate obtained by lego. """ class TestVsftpdQuadlet(test_quadlet.TestQuadlet): # vsftpd needs the certificate obtained by lego, so the vsftpd services are # checked once lego has obtained it (see test_create_ftp_user). expected_services = [ {"name": "lego.target", "state": "active", "exists": True}, {"name": "lego-hooks.path", "state": "active", "exists": True}, ] expected_vsftpd_services = [ {"name": "vsftpd.target", "state": "active", "exists": True}, {"name": "vsftpd.service", "state": "active", "exists": True}, ] expected_vsftpd_ports = [ {"number": 21, "protocol": "tcp", "state": "open"}, ] expected_files = [ {"path": "/etc/quadlets/lego/renew-hooks.d/vsftpd.sh", "type": "file", "owner": "root", "mode": 0o755}, ] # Certificates copied by vsftpd.service from the lego storage expected_tls_files = [ {"path": "/run/quadlets/vsftpd/tls/default.crt", "type": "file", "owner": "root", "mode": 0o600}, {"path": "/run/quadlets/vsftpd/tls/default.key", "type": "file", "owner": "root", "mode": 0o600}, ] # The vsftpd image is built on first boot (see test_create_ftp_user). expected_vsftpd_images = [ {"name": "localhost/vsftpd", "tag": "latest", "state": "present"}, ] expected_main_service = "lego.target" expected_main_service_timeout = 600 # Image pull + ACME negotiation def ftps_connect(self, host_ip: str, ca_bundle: str) -> ReusedSessionFTP_TLS: """Open an authenticated FTPS session, trusting only the Pebble CA.""" ctx = ssl.create_default_context(cadata=ca_bundle) # The VM is reached by IP: the hostname is checked separately against the certificate SAN. ctx.check_hostname = False ctx.verify_mode = ssl.CERT_REQUIRED ftp = ReusedSessionFTP_TLS(context=ctx, timeout=30) ftp.connect(host_ip, 21) ftp.login(FTP_USER, FTP_PASSWORD) ftp.prot_p() return ftp def served_certificate(self, host_ip: str, ca_bundle: str) -> dict: ftp = self.ftps_connect(host_ip, ca_bundle) try: return ftp.sock.getpeercert() finally: ftp.quit() def lego_serial(self, host) -> str: result = host.run(f"openssl x509 -in {LEGO_CERT_PATH} -noout -serial") assert result.rc == 0, f"Failed to read the lego certificate: {result.stderr}" return result.stdout.strip().removeprefix("serial=").upper() def wait_for_served_serial(self, host_ip: str, ca_bundle: str, serial: str, timeout: int = 180) -> None: deadline = time.monotonic() + timeout served = None while time.monotonic() < deadline: try: served = self.served_certificate(host_ip, ca_bundle)["serialNumber"].upper() if served == serial: return except (OSError, ftplib.Error): pass # vsftpd may be restarting time.sleep(5) raise TimeoutError(f"vsftpd serves certificate {served}, expected {serial}") # ------------------------------------------------------------------ # Test 1 — Lego obtains the certificate # ------------------------------------------------------------------ @pytest.mark.flaky(reruns=3, reruns_delay=10) def test_lego_certificate_obtained(self, fcos_host, dns_server_ip, ftp_domain): """lego.service must obtain the certificate from Pebble.""" state = self.wait_for_unit_done(fcos_host, "lego.service", timeout=600) if state == "failed": # The first attempt may fail if the DNS record of the VM was not # resolvable by Pebble when lego started: try again. prime_dns_cache(dns_server_ip, [ftp_domain]) result = fcos_host.run("systemctl start lego.service") assert result.rc == 0, ( f"lego.service failed: {result.stderr}\n" + fcos_host.run("journalctl -u lego.service --no-pager -n 50").stdout ) assert fcos_host.file(LEGO_CERT_PATH).is_file, f"{LEGO_CERT_PATH} not found" # ------------------------------------------------------------------ # Test 2 — Create a local user # ------------------------------------------------------------------ def test_create_ftp_user(self, fcos_host): """Create a local user and restart vsftpd so that it is part of its passwd cache.""" result = fcos_host.run( f"id {FTP_USER} || useradd -u {FTP_UID} -U -M -d /var/lib/vsftpd-test {FTP_USER}" ) assert result.rc == 0, f"useradd failed: {result.stderr}" result = fcos_host.run(f"echo '{FTP_USER}:{FTP_PASSWORD}' | chpasswd") assert result.rc == 0, f"chpasswd failed: {result.stderr}" # The vsftpd image build may take a while on first boot. state = self.wait_for_unit_done(fcos_host, "vsftpd-build.service", timeout=900) assert state == "inactive", ( f"vsftpd-build.service ended in unexpected state: {state!r}\n" + fcos_host.run("journalctl -u vsftpd-build.service --no-pager -n 50").stdout ) self.check_expected_podman_images(fcos_host, self.expected_vsftpd_images) result = fcos_host.run("systemctl restart vsftpd.service && systemctl start vsftpd.target") assert result.rc == 0, ( f"Failed to restart vsftpd: {result.stderr}\n" + fcos_host.run("journalctl -u vsftpd.service --no-pager -n 50").stdout ) self.check_expected_services(fcos_host, self.expected_vsftpd_services) self.check_expected_files(fcos_host, self.expected_tls_files) def test_vsftpd_ports(self, fcos_vm): """The FTP control port must be reachable.""" self.check_expected_ports(fcos_vm, self.expected_vsftpd_ports) # ------------------------------------------------------------------ # Test 3 — FTPS with the certificate obtained by lego # ------------------------------------------------------------------ @pytest.mark.flaky(reruns=6, reruns_delay=10) def test_ftps_certificate(self, fcos_vm, fcos_host, pebble_acme_server, ftp_domain): """vsftpd must serve the certificate obtained by lego from Pebble.""" cert = self.served_certificate(fcos_vm.ip, pebble_acme_server["ca_bundle"]) san = [value for kind, value in cert.get("subjectAltName", ()) if kind == "DNS"] assert ftp_domain in san, f"Expected {ftp_domain!r} in the certificate SAN, got {san}" assert cert["serialNumber"].upper() == self.lego_serial(fcos_host), ( "vsftpd does not serve the certificate obtained by lego" ) def test_ftps_upload(self, fcos_vm, fcos_host, pebble_acme_server): """An authenticated user must be able to upload and list files over FTPS.""" ftp = self.ftps_connect(fcos_vm.ip, pebble_acme_server["ca_bundle"]) try: ftp.storbinary("STOR hello.txt", io.BytesIO(b"Hello, FTPS!\n")) assert "hello.txt" in ftp.nlst(), "Uploaded file not listed" finally: ftp.quit() assert fcos_host.file("/var/lib/vsftpd-test/hello.txt").content_string == "Hello, FTPS!\n" def test_plain_ftp_rejected(self, fcos_vm): """Local users must not be able to log in without TLS.""" ftp = ftplib.FTP(timeout=30) ftp.connect(fcos_vm.ip, 21) try: with pytest.raises(ftplib.error_perm): ftp.login(FTP_USER, FTP_PASSWORD) finally: ftp.close() # ------------------------------------------------------------------ # Test 4 — Renewal: the lego hook restarts vsftpd with the new certificate # ------------------------------------------------------------------ @pytest.mark.flaky(reruns=3, reruns_delay=5) def test_certificate_renewal(self, fcos_vm, fcos_host, pebble_acme_server, dns_server_ip, ftp_domain): """When lego renews the certificate, vsftpd must serve the new one.""" prime_dns_cache(dns_server_ip, [ftp_domain]) serial_before = self.lego_serial(fcos_host) # Force the renewal: a negative number of days always renews the certificate. result = fcos_host.run( f"cp -a {LEGO_CONFIG} {LEGO_CONFIG}.orig && " f"sed -i 's/disableRandomSleep: true/disableRandomSleep: true\\n days: -1/' {LEGO_CONFIG}" ) assert result.rc == 0, f"Failed to edit lego.yaml: {result.stderr}" try: result = fcos_host.run("systemctl start lego.service") assert result.rc == 0, ( f"lego.service failed: {result.stderr}\n" + fcos_host.run("journalctl -u lego.service --no-pager -n 50").stdout ) finally: fcos_host.run(f"mv {LEGO_CONFIG}.orig {LEGO_CONFIG}") serial_after = self.lego_serial(fcos_host) assert serial_after != serial_before, "lego did not renew the certificate" self.wait_for_served_serial(fcos_vm.ip, pebble_acme_server["ca_bundle"], serial_after) # ------------------------------------------------------------------ # Test 5 — Reboot survival (must run last) # ------------------------------------------------------------------ def test_survives_reboot(self, fcos_vm, fcos_host, test_ssh_key, pebble_acme_server): """After a reboot, vsftpd must serve FTPS again (the tmpfs holding the certificates is repopulated).""" serial_before = self.lego_serial(fcos_host) # Schedule the reboot so that the SSH command returns before the connection drops. result = fcos_host.run("systemd-run --on-active=3 systemctl reboot") assert result.rc == 0, f"Failed to schedule the reboot: {result.stderr}" time.sleep(15) # let the reboot begin before we start polling SSH fcos_vm.wait_ssh(ssh_key=test_ssh_key, timeout=300) # The module-scoped fcos_host fixture is stale after reboot; create a fresh connection. new_host = testinfra.get_host( f"ssh://root@{fcos_vm.ip}", ssh_extra_args=( f"-i {test_ssh_key}" " -o StrictHostKeyChecking=no" " -o UserKnownHostsFile=/dev/null" ), ) self.wait_for_service(new_host, "vsftpd.target", timeout=300) self.check_expected_services(new_host, self.expected_services + self.expected_vsftpd_services) self.check_expected_files(new_host, self.expected_files + self.expected_tls_files) self.wait_for_served_serial(fcos_vm.ip, pebble_acme_server["ca_bundle"], serial_before)