#!/usr/bin/env python3 """Exercise ownership, partial failure and secret handling without production.""" import contextlib import io import json import os from pathlib import Path import subprocess import tempfile import threading import time import unittest from unittest.mock import MagicMock, patch import public_tunnel_acceptance as acceptance from public_tunnel_credentials import Credential class Api: def __init__(self): self.residual = False def call(self, method, path): assert method == "GET" and path == "/api/pipe" return {"config": {"quickTunnels": [{"id": "orphan"}] if self.residual else []}} class AcceptanceTest(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) self.store = Path(self.directory.name) / "addresses.json" self.credentials = [Credential(label="qa-test-a", token="a" * 64, path="/private/a"), Credential(label="qa-test-b", token="b" * 64, path="/private/b")] self.services = [{"id": "a", "spec": {"domain": "a.example.test"}}, {"id": "b", "spec": {"domain": "b.example.test"}}] self.api = Api() self.calls = [] self.failed_connect = None self.failed_delete = None self.revoked = [] self.updater_checks = [] self.evidence = {} def run_cli(cli, *args, credential=None): self.calls.append((args, credential.label if credential else None)) if credential: entries = self.read_bindings() entries = [entry for entry in entries if entry["subdomain"] != credential.label] entries.append({"subdomain": credential.label, "token": credential.token, "ownedConnector": True, "connectorId": credential.label + "-connector"}) self.write_bindings(entries) if self.failed_connect == credential.label: raise acceptance.AcceptanceError("original connect failure") if args[:2] == ("address", "delete"): if self.failed_delete == args[2]: raise acceptance.AcceptanceError("delete failed") self.write_bindings([item for item in self.read_bindings() if item["subdomain"] != args[2]]) patches = [ patch.object(acceptance, "STORE", self.store), patch.object(acceptance.shutil, "which", return_value="/test/firecrab"), patch.object(acceptance, "fixture_targets", return_value=["172.29.1.2:8080", "172.29.1.3:8080"]), patch.object(acceptance, "verify_registration", return_value=None), patch.object(acceptance, "verify_revoked", side_effect=lambda credential: self.revoked.append(credential.label)), patch.object(acceptance, "run_cli", side_effect=run_cli), patch.object(acceptance, "ready", return_value=True), patch.object(acceptance, "wait_offline", return_value=True), patch.object(acceptance, "restart_api", return_value=3), patch.object(acceptance, "verify_updater_rejection", side_effect=lambda api, wait, credential, peer: self.updater_checks.append((credential.label, peer.label if peer else None))), patch("builtins.input", return_value=""), ] for item in patches: item.start() self.addCleanup(item.stop) def read_bindings(self): if not self.store.exists(): return [] return json.loads(self.store.read_text())["bindings"] def write_bindings(self, bindings): self.store.write_text(json.dumps({"endpoint": acceptance.ENDPOINT, "protocol": "register_v1", "bindings": bindings})) def run_acceptance(self): with contextlib.redirect_stdout(io.StringIO()): acceptance.run_public(self.api, lambda *a, **kw: True, *self.services, self.credentials, self.evidence) def test_complete_sequence_checks_both_revocations_and_cleans_connections(self): self.run_acceptance() result = self.evidence["publicTunnel"] self.assertEqual(result["status"], "pending_operator_verification") self.assertEqual(result["connectionChecks"], "passed") self.assertEqual(result["dashboardRevocations"], "operator_verification_required") self.assertEqual(self.revoked, ["qa-test-a", "qa-test-a", "qa-test-b", "qa-test-a", "qa-test-b"]) self.assertEqual(self.updater_checks, [("qa-test-a", "qa-test-b"), ("qa-test-b", None)]) self.assertEqual(result["tokenAuthentication"], {"qa-test-a": "registration_rejected", "qa-test-b": "registration_rejected"}) self.assertEqual(result["dashboardReservations"], "operator_cleanup_required") self.assertEqual(self.read_bindings(), []) self.assertIn((("address", "connect", "qa-test-a", "--nginx", "b.example.test"), None), self.calls) self.assertNotIn("a" * 64, json.dumps(self.evidence)) def test_partial_connect_is_cleaned_even_when_command_fails(self): self.failed_connect = "qa-test-b" with self.assertRaisesRegex(acceptance.AcceptanceError, "original connect failure"): self.run_acceptance() self.assertEqual(self.read_bindings(), []) self.assertEqual(self.evidence["publicTunnel"]["status"], "failed") self.assertEqual(self.evidence["publicTunnel"]["tokenAuthentication"]["qa-test-a"], "accepted") def test_cleanup_failure_does_not_hide_original_failure(self): self.failed_connect = "qa-test-b" self.failed_delete = "qa-test-b" with self.assertRaisesRegex(acceptance.AcceptanceError, "original connect failure"): self.run_acceptance() self.assertEqual(self.evidence["publicTunnel"]["cleanupErrors"], ["local connection cleanup failed: qa-test-b"]) def test_foreign_ownership_is_never_deleted(self): credential = self.credentials[0] self.write_bindings([{"subdomain": credential.label, "token": "f" * 64, "ownedConnector": True, "connectorId": "foreign"}]) with self.assertRaises(acceptance.AcceptanceError): acceptance.owned_binding(self.store, credential) self.assertEqual(len(self.read_bindings()), 1) def test_orphaned_connector_makes_cleanup_fail(self): self.api.residual = True with self.assertRaisesRegex(acceptance.AcceptanceError, "cleanup failed"): self.run_acceptance() self.assertEqual(self.evidence["publicTunnel"]["status"], "failed") self.assertIn("connector cleanup is incomplete or unverified", self.evidence["publicTunnel"]["cleanupErrors"]) def test_failed_token_check_cannot_be_reported_as_revocation(self): with patch.object(acceptance, "verify_revoked", side_effect=acceptance.AcceptanceError("token still accepted")): with self.assertRaisesRegex(acceptance.AcceptanceError, "token still accepted"): self.run_acceptance() result = self.evidence["publicTunnel"] self.assertEqual(result["status"], "failed") self.assertEqual(result["connectionChecks"], "not_completed") self.assertEqual(result["tokenAuthentication"]["qa-test-a"], "accepted") self.assertEqual(self.read_bindings(), []) def test_failed_updater_check_cannot_pass_acceptance(self): with patch.object(acceptance, "verify_updater_rejection", side_effect=acceptance.AcceptanceError("updater restored access")): with self.assertRaisesRegex(acceptance.AcceptanceError, "updater restored access"): self.run_acceptance() self.assertEqual(self.evidence["publicTunnel"]["status"], "failed") self.assertEqual(self.evidence["publicTunnel"]["connectionChecks"], "not_completed") self.assertEqual(self.read_bindings(), []) class UpdaterRejectionTest(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) self.store = Path(self.directory.name) / "addresses.json" self.primary = Credential("qa-test-a", "/private/a", "a" * 64) self.peer = Credential("qa-test-b", "/private/b", "b" * 64) self.store.write_text(json.dumps({"endpoint": acceptance.ENDPOINT, "protocol": "register_v1", "bindings": [ {"subdomain": self.primary.label, "token": self.primary.token, "ownedConnector": True, "connectorId": "connector-a"}, {"subdomain": self.peer.label, "token": self.peer.token, "ownedConnector": True, "connectorId": "connector-b"}, ]})) self.active = True self.pid = "123" self.now = 0.0 self.commands = [] self.snapshots = [ {"subdomain": self.primary.label, "connectorId": "connector-a", "connected": False, "error": acceptance.REVOKED_ERROR}, {"subdomain": self.peer.label, "connectorId": "connector-b", "connected": True}, ] self.api = MagicMock() self.api.call.side_effect = lambda *args: {"firecrabTunnels": self.snapshots} self.wait_count = 0 def systemctl(self, command, **kwargs): self.commands.append(command[1]) if command[1] == "stop": self.active = False elif command[1] == "start": self.active = True if "--property=MainPID" in command: return MagicMock(stdout=(self.pid if self.active else "0") + "\n") return MagicMock(stdout=("active" if self.active else "inactive") + "\n") def update_store(self): # The real updater atomically replaces its saved ownership file. replacement = self.store.with_suffix(".new") replacement.write_text(self.store.read_text()) replacement.replace(self.store) def wait_for(self, check, timeout, message): self.wait_count += 1 self.assertFalse(check(), "an unchanged file cannot prove an updater refresh") self.update_store() if self.wait_count == 2: self.now += 30 result = check() if not result: raise acceptance.AcceptanceError(message) return result def run_check(self): with patch.object(acceptance, "STORE", self.store), \ patch.object(acceptance.time, "monotonic", side_effect=lambda: self.now), \ patch.object(acceptance.subprocess, "run", side_effect=self.systemctl): acceptance.verify_updater_rejection(self.api, self.wait_for, self.primary, self.peer) def test_restart_and_next_refresh_reject_token_and_keep_peer(self): self.run_check() self.assertEqual(self.wait_count, 2) self.assertEqual([command for command in self.commands if command in ("stop", "start")], ["stop", "start"]) def test_network_error_cannot_pass_as_token_rejection(self): self.snapshots[0]["error"] = "Cannot reach the Firecrab connection service" with self.assertRaisesRegex(acceptance.AcceptanceError, "did not reject"): self.run_check() self.assertTrue(self.active) def test_peer_disconnection_fails_the_check(self): self.snapshots[1]["connected"] = False with self.assertRaisesRegex(acceptance.AcceptanceError, "did not reject"): self.run_check() def test_second_refresh_must_still_reject(self): original_wait = self.wait_for def wait_for(check, timeout, message): if self.wait_count == 1: self.snapshots[0]["connected"] = True return original_wait(check, timeout, message) self.wait_for = wait_for with self.assertRaisesRegex(acceptance.AcceptanceError, "next refresh"): self.run_check() def test_changed_snapshot_during_api_read_is_not_accepted(self): def race(*args): self.update_store() return {"firecrabTunnels": self.snapshots} self.api.call.side_effect = race with self.assertRaisesRegex(acceptance.AcceptanceError, "did not reject"): self.run_check() def test_updater_auto_restart_cannot_count_as_periodic_refresh(self): original_wait = self.wait_for def wait_for(check, timeout, message): if self.wait_count == 1: self.pid = "456" return original_wait(check, timeout, message) self.wait_for = wait_for with self.assertRaisesRegex(acceptance.AcceptanceError, "restarted during"): self.run_check() def test_early_write_cannot_count_as_a_periodic_refresh(self): original_wait = self.wait_for def wait_for(check, timeout, message): if self.wait_count == 0: return original_wait(check, timeout, message) self.wait_count += 1 self.now += 1 self.update_store() self.assertFalse(check()) self.now += 29 self.assertFalse(check(), "an early write cannot pass just by waiting") self.update_store() return check() self.wait_for = wait_for self.run_check() self.assertEqual(self.wait_count, 2) def test_timed_out_stop_still_attempts_restart(self): original_systemctl = self.systemctl def timeout(command, **kwargs): if command[1] == "stop": self.active = False raise subprocess.TimeoutExpired(command, 120) return original_systemctl(command, **kwargs) self.systemctl = timeout with self.assertRaisesRegex(acceptance.AcceptanceError, "service command failed"): self.run_check() self.assertTrue(self.active) self.assertIn("start", self.commands) class SecretTransportTest(unittest.TestCase): @unittest.skipUnless(hasattr(os, "memfd_create"), "Linux memfd required") def test_token_is_in_inherited_file_not_arguments_or_errors(self): with tempfile.TemporaryDirectory() as directory: cli = Path(directory) / "firecrab" cli.write_text("""#!/usr/bin/env python3 import sys value = 'a' * 64 assert not any(value in arg for arg in sys.argv) assert open(sys.argv[sys.argv.index('--token-file') + 1]).read() == value print(value, file=sys.stderr) sys.exit(9) """) cli.chmod(0o755) credential = Credential(label="qa-test-a", token="a" * 64, path="/private/a") with self.assertRaises(acceptance.AcceptanceError) as error: acceptance.run_cli(str(cli), "address", "connect", "--app", "127.0.0.1:8080", credential=credential) self.assertNotIn(credential.token, str(error.exception)) def test_public_polling_is_limited(self): calls = [] poll = acceptance.limited(lambda: calls.append(1) or False) with patch.object(acceptance.time, "monotonic", side_effect=[100, 101, 104, 105]): for _ in range(4): poll() self.assertEqual(len(calls), 2) class OfflineResponseTest(unittest.TestCase): def test_only_cloudflare_1033_is_accepted_for_http_530(self): credential = Credential(label="qa-test-a", token="a" * 64, path="/private/a") cases = [ (530, "cloudflare", b"error code: 1033\n", True), (530, "CloudFlare", b"error code: 1033\n", True), (530, "cloudflare", b"error code: 1000\n", False), (530, "cloudflare", b"error code: 1033", False), (530, "cloudflare", b"error code: 1033\nextra", False), (530, "other", b"error code: 1033\n", False), (200, "cloudflare", b"error code: 1033\n", False), (404, "cloudflare", b"not found", True), ] for status, server, body, expected in cases: with self.subTest(status=status, server=server, body=body): response = MagicMock(status=status) response.getheader.return_value = server response.read.return_value = body connection = MagicMock() connection.getresponse.return_value = response with patch.object(acceptance.http.client, "HTTPSConnection", return_value=connection): self.assertEqual(acceptance.public_offline(credential), expected) connection.request.assert_called_once_with("GET", "/app", headers={"Cache-Control": "no-cache"}) class RestartTimingTest(unittest.TestCase): def setUp(self): self.credentials = [Credential("qa-test-a", "/private/a", "a" * 64), Credential("qa-test-b", "/private/b", "b" * 64)] self.calls = {credential.label: 0 for credential in self.credentials} self.lock = threading.Lock() self.evidence = {} @staticmethod def wait_for(check, timeout, message): deadline = time.monotonic() + min(timeout, 3) while time.monotonic() < deadline: if check(): return True time.sleep(.005) raise AssertionError(message) def test_each_address_has_a_successful_request_overlapping_restart(self): started = {credential.label: threading.Event() for credential in self.credentials} release = threading.Event() def check(host, path, body, context): with self.lock: self.calls[host] += 1 count = self.calls[host] if count == 2: started[host].set() if not release.wait(2): raise RuntimeError("test restart did not release request") class Process: def wait(self, timeout=None): if not all(event.wait(2) for event in started.values()): raise RuntimeError("both requests did not start") release.set() return 0 api = MagicMock() with patch.object(acceptance.probe, "validate_host", side_effect=lambda label: label), \ patch.object(acceptance.probe, "check_https", side_effect=check), \ patch.object(acceptance.ssl, "create_default_context", return_value=object()), \ patch.object(acceptance.subprocess, "Popen", return_value=Process()): acceptance.restart_api(api, self.wait_for, self.credentials, (b"a", b"b"), self.evidence) self.assertEqual(self.evidence["overlappingSuccessfulRequests"], {"qa-test-a": 1, "qa-test-b": 1}) self.assertLess(self.evidence["restartStarted"], self.evidence["restartFinished"]) for credential in self.credentials: rows = self.evidence["requests"][credential.label] self.assertEqual(len(rows), 2) self.assertTrue(all(row["success"] and row["started"] <= row["finished"] for row in rows)) def test_fast_restart_without_overlapping_requests_fails(self): gate = threading.Event() validations = {credential.label: 0 for credential in self.credentials} def host(label): with self.lock: validations[label] += 1 count = validations[label] if count == 2 and not gate.wait(2): raise RuntimeError("test did not release late request") return label class Process: def wait(self, timeout=None): return 0 class Api: def ready(self): timer = threading.Timer(.05, gate.set) timer.start() with patch.object(acceptance.probe, "validate_host", side_effect=host), \ patch.object(acceptance.probe, "check_https"), \ patch.object(acceptance.ssl, "create_default_context", return_value=object()), \ patch.object(acceptance.subprocess, "Popen", return_value=Process()): with self.assertRaisesRegex(acceptance.AcceptanceError, "overlapping"): acceptance.restart_api(Api(), self.wait_for, self.credentials, (b"a", b"b"), self.evidence) self.assertEqual(self.evidence["overlappingSuccessfulRequests"], {"qa-test-a": 0, "qa-test-b": 0}) class HostGuardTest(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) self.store = Path(self.directory.name) / "addresses.json" self.services = [{"id": "1", "spec": {"name": "tunnel-qa-a", "domain": "a.example.test", "port": 8080}}, {"id": "2", "spec": {"name": "tunnel-qa-b", "domain": "b.example.test", "port": 8080}}] self.values = { "/api/micro-services": [{"id": service["id"], **service["spec"]} for service in self.services], "/api/vms": [{"id": "vm1"}, {"id": "vm2"}], "/api/micro-services/1/deployment": {"id": "d1", "phase": "running", "instances": [{"vmId": "vm1", "ipv4": "172.29.1.2"}]}, "/api/micro-services/2/deployment": {"id": "d2", "phase": "running", "instances": [{"vmId": "vm2", "ipv4": "172.29.1.3"}]}, "/api/pipe": {"config": {"routes": [ {"host": "a.example.test", "owner": {"serviceId": "1", "deploymentId": "d1"}}, {"host": "tunnel-qa-a.internal", "owner": {"serviceId": "1", "deploymentId": "d1"}}, {"host": "b.example.test", "owner": {"serviceId": "2", "deploymentId": "d2"}}, {"host": "tunnel-qa-b.internal", "owner": {"serviceId": "2", "deploymentId": "d2"}}, ]}}, } class FakeApi: def call(_, method, path): return self.values[path] self.api = FakeApi() original_exists = Path.exists patches = [patch.object(acceptance, "STORE", self.store), patch.object(acceptance.os, "geteuid", return_value=0), patch.object(Path, "exists", lambda path: True if str(path) == "/dev/kvm" else original_exists(path)), patch.object(acceptance, "expect_fixture")] for item in patches: item.start() self.addCleanup(item.stop) def test_accepts_only_the_two_declared_fixture_services_and_vms(self): self.assertEqual(acceptance.fixture_targets(self.api, self.services), ["172.29.1.2:8080", "172.29.1.3:8080"]) self.values["/api/vms"].append({"id": "foreign"}) with self.assertRaises(acceptance.AcceptanceError): acceptance.fixture_targets(self.api, self.services) def test_existing_tunnel_and_unrelated_route_are_rejected(self): self.values["/api/pipe"]["firecrabTunnels"] = [{"subdomain": "existing"}] with self.assertRaises(acceptance.AcceptanceError): acceptance.fixture_targets(self.api, self.services) del self.values["/api/pipe"]["firecrabTunnels"] self.values["/api/pipe"]["config"]["routes"].append({"host": "other.example.test"}) with self.assertRaises(acceptance.AcceptanceError): acceptance.fixture_targets(self.api, self.services) def test_internal_route_with_another_owner_is_rejected(self): self.values["/api/pipe"]["config"]["routes"][1]["owner"]["serviceId"] = "foreign" with self.assertRaises(acceptance.AcceptanceError): acceptance.fixture_targets(self.api, self.services) if __name__ == "__main__": unittest.main()