diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/e2e/test_tor_unlock_e2e.py b/tests/e2e/test_tor_unlock_e2e.py new file mode 100644 index 0000000..33c3ad5 --- /dev/null +++ b/tests/e2e/test_tor_unlock_e2e.py @@ -0,0 +1,104 @@ +"""End-to-end test of the Tor onion LUKS-unlock path (rootless). + +It reproduces the operator-visible half of the decryption process that the +"tor" mkinitcpio hook drives at boot: + + 1. generate the v3 onion keys offline (as lim.image.tor does in the chroot), + 2. start a real Tor onion service from a torrc mirroring the baked-in one, + forwarding the virtual port 22 to a local dropbear stand-in, + 3. connect to the .onion address through Tor and deliver the passphrase, + 4. assert the passphrase reached the unlock endpoint and it "unlocked". + +The physical flashing half (loop device, cryptsetup, mount, chroot, +mkinitcpio) needs root and is out of scope here by design. + +Requires the real `tor` binary and live Tor network access, so it is +opt-in and skipped unless LIM_E2E_TOR=1: + + LIM_E2E_TOR=1 pytest tests/e2e/test_tor_unlock_e2e.py -v +""" + +import os +import shutil +import time + +import pytest + +from lim.image import tor as tor_module +from tests.e2e import tor_harness + +pytestmark = pytest.mark.skipif( + os.environ.get("LIM_E2E_TOR") != "1" or shutil.which("tor") is None, + reason="needs LIM_E2E_TOR=1 and the tor binary (live network, slow).", +) + +PASSPHRASE = b"correct horse battery staple" + + +@pytest.fixture +def workdir(tmp_path): + (tmp_path / "onion").mkdir() + (tmp_path / "data").mkdir() + return tmp_path + + +def test_offline_keygen_matches_production_flags(): + """Guard: the harness keygen mirrors the flags production actually runs.""" + script = tor_module._KEYGEN_SCRIPT + assert "--DisableNetwork 1" in script + assert "HiddenServicePort" in script + assert "22 127.0.0.1:22" in script + assert "--SocksPort 0" in script + + +def test_onion_keygen_is_deterministic_and_offline(workdir): + """Keys generate without network and the .onion address is stable.""" + address = tor_harness.generate_onion_keys( + workdir / "onion", workdir / "data" + ) + assert address.endswith(".onion") + assert len(address) == len("v" * 56) + len(".onion") # v3 = 56 base32 chars + for name in ("hs_ed25519_secret_key", "hs_ed25519_public_key", "hostname"): + assert (workdir / "onion" / name).is_file() + + +def test_unlock_passphrase_travels_over_onion(workdir): + """Full rootless round-trip: client -> Tor -> onion -> dropbear stand-in.""" + onion_dir = workdir / "onion" + data_dir = workdir / "data" + onion_address = tor_harness.generate_onion_keys(onion_dir, data_dir) + + backend_port = tor_harness.free_port() + socks_port = tor_harness.free_port() + torrc = workdir / "torrc" + log_path = workdir / "tor.log" + tor_harness.write_test_torrc(torrc, data_dir, onion_dir, backend_port) + + dropbear = tor_harness.FakeDropbear(backend_port) + dropbear.start() + service = tor_harness.start_onion_service(torrc, socks_port, log_path) + try: + tor_harness.wait_bootstrapped(log_path, service) + + # Onion descriptors need a moment to publish after bootstrap; retry. + last_error = None + for _ in range(5): + try: + stream = tor_harness.socks5_connect(socks_port, onion_address, 22) + break + except (OSError, RuntimeError) as exc: + last_error = exc + time.sleep(2) + else: + pytest.fail(f"Could not reach {onion_address} via Tor: {last_error}") + + with stream: + stream.sendall(PASSPHRASE + b"\n") + reply = stream.recv(64) + finally: + service.terminate() + service.wait(timeout=15) + dropbear.stop() + + assert reply.strip() == b"UNLOCKED" + assert dropbear.received == PASSPHRASE diff --git a/tests/e2e/tor_harness.py b/tests/e2e/tor_harness.py new file mode 100644 index 0000000..8597158 --- /dev/null +++ b/tests/e2e/tor_harness.py @@ -0,0 +1,178 @@ +"""Rootless helpers to stand up a real Tor onion service and reach it. + +These reproduce, without root/block-devices/chroot, what the "tor" +mkinitcpio hook does at LUKS-decryption time: generate v3 onion keys +offline, run a real Tor onion service that forwards the virtual port 22 +to a local "dropbear", and let a client reach that service through Tor. +""" + +import socket +import struct +import subprocess +import threading +import time +from pathlib import Path + +# Matches lim.image.tor._KEYGEN_SCRIPT: an empty -f torrc keeps the host's +# /etc/tor/torrc out of the run, --DisableNetwork 1 makes the keys appear +# without ever touching the network. +KEYGEN_TIMEOUT = 30 +BOOTSTRAP_TIMEOUT = 180 +ONION_CONNECT_TIMEOUT = 90 + + +def free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def generate_onion_keys(onion_dir: Path, data_dir: Path) -> str: + """Create v3 onion keys offline; return the .onion hostname. + + Mirrors lim.image.tor._KEYGEN_SCRIPT, run directly on the host instead + of inside the image chroot (the tor invocation is identical). + """ + onion_dir.mkdir(parents=True, exist_ok=True) + onion_dir.chmod(0o700) + empty_torrc = data_dir.parent / "keygen-torrc" + empty_torrc.write_text("") + process = subprocess.Popen( + [ + "tor", "-f", str(empty_torrc), "--DisableNetwork", "1", + "--DataDirectory", str(data_dir), + "--HiddenServiceDir", str(onion_dir), + "--HiddenServicePort", "22 127.0.0.1:22", + "--SocksPort", "0", "--RunAsDaemon", "0", "--Log", "notice stderr", + ], + stderr=subprocess.DEVNULL, + ) + hostname = onion_dir / "hostname" + try: + deadline = time.monotonic() + KEYGEN_TIMEOUT + while time.monotonic() < deadline: + if hostname.is_file() and hostname.read_text().strip(): + break + time.sleep(0.5) + finally: + process.terminate() + process.wait(timeout=10) + if not (hostname.is_file() and hostname.read_text().strip()): + raise RuntimeError("Offline onion key generation produced no hostname.") + return hostname.read_text().strip() + + +def start_onion_service( + torrc_path: Path, socks_port: int, log_path: Path +) -> subprocess.Popen: + """Start Tor with the given torrc (service side) plus a SOCKS port. + + The torrc carries the HiddenServiceDir/HiddenServicePort just like the + baked-in image torrc; --SocksPort lets the same instance also act as the + client's entry point so the test needs only one bootstrap. + """ + return subprocess.Popen( + [ + "tor", "-f", str(torrc_path), + "--SocksPort", str(socks_port), + "--Log", f"notice file {log_path}", + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + +def wait_bootstrapped(log_path: Path, process: subprocess.Popen) -> None: + deadline = time.monotonic() + BOOTSTRAP_TIMEOUT + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError(f"Tor exited early (code {process.returncode}).") + if log_path.is_file() and "Bootstrapped 100%" in log_path.read_text(): + return + time.sleep(1) + raise RuntimeError("Tor did not bootstrap in time.") + + +def _socks5_handshake(sock: socket.socket, onion_host: str, onion_port: int) -> None: + sock.sendall(b"\x05\x01\x00") # VER, 1 method, no-auth + if sock.recv(2) != b"\x05\x00": + raise RuntimeError("SOCKS5 handshake rejected.") + host = onion_host.encode() + request = ( + b"\x05\x01\x00\x03" + + bytes([len(host)]) + host + + struct.pack(">H", onion_port) + ) + sock.settimeout(ONION_CONNECT_TIMEOUT) + sock.sendall(request) + header = sock.recv(4) + if len(header) < 2 or header[1] != 0x00: + code = header[1] if len(header) >= 2 else -1 + raise RuntimeError(f"SOCKS5 CONNECT failed (reply code {code}).") + # Drain the bound-address tail so the stream starts at real payload. + atyp = header[3] + tail = {0x01: 4, 0x04: 16}.get(atyp) + if tail is None: # domain name + tail = sock.recv(1)[0] + sock.recv(tail + 2) + + +def socks5_connect(socks_port: int, onion_host: str, onion_port: int) -> socket.socket: + """Open a stream to onion_host:onion_port through Tor's SOCKS5 proxy.""" + sock = socket.create_connection(("127.0.0.1", socks_port), timeout=30) + try: + _socks5_handshake(sock, onion_host, onion_port) + except (OSError, RuntimeError): + sock.close() + raise + return sock + + +class FakeDropbear: + r"""Stand-in for the initramfs dropbear+cryptsetup unlock endpoint. + + Accepts one connection on 127.0.0.1:, reads a passphrase line, and + replies "UNLOCKED\n" — the observable proof that a passphrase delivered + over the onion connection reached the unlock shell. + """ + + def __init__(self, port: int) -> None: + self.port = port + self.received: bytes | None = None + self._server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._server.bind(("127.0.0.1", port)) + self._server.listen(1) + self._thread = threading.Thread(target=self._serve, daemon=True) + + def start(self) -> None: + self._thread.start() + + def _serve(self) -> None: + try: + self._server.settimeout(BOOTSTRAP_TIMEOUT + ONION_CONNECT_TIMEOUT) + conn, _ = self._server.accept() + with conn: + conn.settimeout(ONION_CONNECT_TIMEOUT) + self.received = conn.recv(4096).strip() + conn.sendall(b"UNLOCKED\n") + except OSError: + pass + + def stop(self) -> None: + self._server.close() + self._thread.join(timeout=5) + + +def write_test_torrc(path: Path, data_dir: Path, onion_dir: Path, backend_port: int) -> None: + """Write a torrc mirroring the baked-in one, with test-writable paths. + + The virtual onion port stays 22 (as in production); only the loopback + backend and the data/key directories move to rootless temp locations. + """ + path.write_text( + f"DataDirectory {data_dir}\n" + f"HiddenServiceDir {onion_dir}\n" + f"HiddenServicePort 22 127.0.0.1:{backend_port}\n" + "SocksPort 0\n" + )