scripts/test_public_tunnel_acceptance.py
Rama main · réplica de lectura del repositorio original.
#!/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()