scripts/public_tunnel_acceptance.py
Rama main · réplica de lectura del repositorio original.
"""Public Tunnel acceptance for two fixture services on a disposable host.
The caller owns fixture VM, Nginx, and dashboard reservation cleanup. This
module manages only the two connections whose tokens it can verify locally.
"""
import contextlib
import http.client
import importlib.util
import json
import os
from pathlib import Path
import shutil
import ssl
import subprocess
import sys
import threading
import time
from public_tunnel_credentials import verify_registration, verify_revoked
_probe_spec = importlib.util.spec_from_file_location(
"public_tunnel_probe", Path(__file__).with_name("check-public-tunnel-app.py")
)
probe = importlib.util.module_from_spec(_probe_spec)
_probe_spec.loader.exec_module(probe)
STORE = Path("/etc/firecrab/pipe/firecrab-addresses.json")
ENDPOINT = "https://tunnel.firecrab.dev"
UPDATER = "firecrab-address-updater.service"
REVOKED_ERROR = "The connection token is invalid or revoked"
class AcceptanceError(Exception):
"""The acceptance failed without exposing a token or response body."""
def owned_binding(store, credential):
"""Refuse to delete an address unless the committed ownership matches."""
if not store.exists():
return None
try:
value = json.loads(store.read_text())
if value.get("endpoint") != ENDPOINT or value.get("protocol") != "register_v1":
raise AcceptanceError("unexpected address store endpoint or protocol")
matches = [entry for entry in value["bindings"] if entry["subdomain"] == credential.label]
if len(matches) > 1:
raise AcceptanceError("duplicate address ownership")
if not matches:
return None
entry = matches[0]
if entry.get("token") != credential.token or entry.get("ownedConnector") is not True:
raise AcceptanceError("address ownership does not match the test token")
return entry["connectorId"]
except (OSError, ValueError, KeyError, TypeError) as error:
raise AcceptanceError("cannot safely inspect address ownership") from error
def run_cli(cli, *args, credential=None):
"""Pass a token through an inherited anonymous file, never arguments or logs."""
env = {"PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"), "HOME": "/root"}
descriptor = None
try:
if credential is not None:
descriptor = os.memfd_create("firecrab-qa-token", flags=0)
os.fchmod(descriptor, 0o600)
os.write(descriptor, credential.token.encode("ascii"))
os.lseek(descriptor, 0, os.SEEK_SET)
args = (*args, "--token-file", f"/proc/self/fd/{descriptor}")
output = subprocess.run([cli, "pipe", *args], env=env, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, timeout=120,
pass_fds=(descriptor,) if descriptor is not None else ())
if output.returncode:
raise AcceptanceError("Firecrab address command failed")
except (OSError, subprocess.SubprocessError) as error:
raise AcceptanceError("could not run Firecrab address command") from error
finally:
if descriptor is not None:
os.close(descriptor)
def expect_fixture(ip, port, expected):
try:
with contextlib.closing(http.client.HTTPConnection(ip, port, timeout=5)) as connection:
connection.request("GET", "/app")
response = connection.getresponse()
if response.status != 200 or response.read(100) != expected:
raise AcceptanceError("local VM fixture did not match")
except (OSError, http.client.HTTPException) as error:
raise AcceptanceError("local VM fixture is unavailable") from error
def fixture_targets(api, services):
"""Require exactly the two declared test services and their active VMs."""
if os.geteuid() != 0 or not Path("/dev/kvm").exists():
raise AcceptanceError("root and a disposable Linux KVM host are required")
expected_ids = {service["id"] for service in services}
if len(expected_ids) != 2:
raise AcceptanceError("two different fixture services are required")
listed = api.call("GET", "/api/micro-services")
if {item["id"] for item in listed} != expected_ids:
raise AcceptanceError("refusing a host with other services")
listed_by_id = {item["id"]: item for item in listed}
deployments, vm_ids, route_owners = [], set(), {}
for service, label in zip(services, ("A", "B")):
spec = service["spec"]
saved = listed_by_id[service["id"]]
if any(saved.get(key) != spec[key] for key in ("name", "domain", "port")):
raise AcceptanceError("fixture identity does not match saved service")
if not spec["name"].startswith("tunnel-qa-") or not spec["domain"].endswith(".example.test"):
raise AcceptanceError("refusing a non-test service or domain")
deployment = api.call("GET", "/api/micro-services/" + service["id"] + "/deployment")
instances = deployment["instances"]
if deployment["phase"] != "running" or len(instances) != 1 or not instances[0].get("ipv4"):
raise AcceptanceError("each fixture needs one running VM")
if not deployment.get("id"):
raise AcceptanceError("fixture deployment identity is missing")
owner = {"serviceId": service["id"], "deploymentId": deployment["id"]}
route_owners[spec["domain"]] = owner
route_owners[spec["name"] + ".internal"] = owner
instance = instances[0]
vm_ids.add(instance["vmId"])
expect_fixture(instance["ipv4"], spec["port"], ("fixture-" + label).encode())
deployments.append(instance["ipv4"] + ":" + str(spec["port"]))
if len(vm_ids) != 2 or {vm["id"] for vm in api.call("GET", "/api/vms")} != vm_ids:
raise AcceptanceError("refusing a host with other VMs")
snapshot = api.call("GET", "/api/pipe")
config = snapshot.get("config") or {}
if snapshot.get("firecrabTunnels") or config.get("quickTunnels") or config.get("quickTunnel") or config.get("tunnel"):
raise AcceptanceError("refusing existing tunnel connections")
routes = config.get("routes", [])
if (len(routes) != len(route_owners) or {route.get("host") for route in routes} != set(route_owners)
or any(route.get("owner") != route_owners.get(route.get("host"))
or route.get("pathPrefix", "/") != "/" for route in routes)):
raise AcceptanceError("refusing unexpected or unowned Nginx routes")
if STORE.exists():
try:
store = json.loads(STORE.read_text())
if store.get("endpoint") != ENDPOINT or store.get("bindings"):
raise AcceptanceError("refusing existing or non-production address store")
except (OSError, ValueError) as error:
raise AcceptanceError("cannot safely inspect the address store") from error
return deployments
def public_ok(credential, expected):
host = probe.validate_host(credential.label)
context = ssl.create_default_context()
probe.check_https(host, "/app", expected, context)
probe.check_websocket(host, "/ws", context)
return True
def public_offline(credential):
host = probe.validate_host(credential.label)
try:
with contextlib.closing(http.client.HTTPSConnection(host, timeout=15,
context=ssl.create_default_context())) as connection:
connection.request("GET", "/app", headers={"Cache-Control": "no-cache"})
response = connection.getresponse()
body = response.read(4096)
if response.status == 530:
return (response.getheader("Server", "").strip().lower() == "cloudflare"
and body == b"error code: 1033\n")
return response.status in (404, 410, 502, 503, 504)
except (OSError, http.client.HTTPException):
return False
def ready(wait_for, credential, expected):
def check():
try:
return public_ok(credential, expected)
except probe.CheckError:
return False
return wait_for(limited(check), timeout=180, message="public fixture did not become ready")
def wait_offline(wait_for, credential):
return wait_for(limited(lambda: public_offline(credential)), timeout=180,
message="public address did not become offline")
def verify_updater_rejection(api, wait_for, credential, peer):
"""Observe two updater writes after restart, including a periodic retry."""
connector = owned_binding(STORE, credential)
peer_connector = owned_binding(STORE, peer) if peer else None
if connector is None or (peer and peer_connector is None):
raise AcceptanceError("test address ownership disappeared before updater restart")
def systemctl(*args):
try:
result = subprocess.run(["systemctl", *args, UPDATER], check=True,
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
text=True, timeout=120)
return result.stdout.strip()
except (OSError, subprocess.SubprocessError) as error:
raise AcceptanceError("address updater service command failed") from error
if systemctl("show", "--property=ActiveState", "--value") != "active":
raise AcceptanceError("address updater was not active")
try:
systemctl("stop")
if systemctl("show", "--property=ActiveState", "--value") != "inactive":
raise AcceptanceError("address updater did not stop")
previous_inode = STORE.stat().st_ino
except BaseException as error:
try:
systemctl("start")
except AcceptanceError:
raise AcceptanceError("updater stop or verification failed; service restart also failed") from error
raise
else:
systemctl("start")
pid = systemctl("show", "--property=MainPID", "--value")
if not pid.isdigit() or int(pid) == 0:
raise AcceptanceError("address updater did not start")
def updated_after(inode):
try:
current_inode = STORE.stat().st_ino
except FileNotFoundError:
return False
if current_inode == inode:
return False
addresses = api.call("GET", "/api/pipe").get("firecrabTunnels") or []
matches = [item for item in addresses if item.get("subdomain") == credential.label]
if (len(matches) != 1 or matches[0].get("connectorId") != connector
or matches[0].get("connected") is not False
or matches[0].get("error") != REVOKED_ERROR):
return False
if peer:
peers = [item for item in addresses if item.get("subdomain") == peer.label]
if (len(peers) != 1 or peers[0].get("connectorId") != peer_connector
or peers[0].get("connected") is not True):
return False
try:
if STORE.stat().st_ino != current_inode:
return False
except FileNotFoundError:
return False
if systemctl("show", "--property=MainPID", "--value") != pid:
raise AcceptanceError("address updater restarted during token verification")
return current_inode
first = wait_for(lambda: updated_after(previous_inode), timeout=180,
message="updater did not reject the token after restart")
first_observed = time.monotonic()
def periodic_refresh():
nonlocal first
inode = updated_after(first)
if not inode:
return False
first = inode
return inode if time.monotonic() - first_observed >= 30 else False
wait_for(periodic_refresh, timeout=180,
message="updater did not reject the token on its next refresh")
if (systemctl("show", "--property=ActiveState", "--value") != "active"
or systemctl("show", "--property=MainPID", "--value") != pid):
raise AcceptanceError("address updater did not remain active")
def limited(check):
"""Respect production request quotas while the caller polls readiness."""
next_request = 0.0
def poll():
nonlocal next_request
now = time.monotonic()
if now < next_request:
return False
next_request = now + 5
return check()
return poll
def restart_api(api, wait_for, credentials, expected, evidence):
"""Measure real request intervals for both addresses across an API restart."""
release = threading.Event()
stop = threading.Event()
baseline_done = [threading.Event() for _ in credentials]
samples = {credential.label: [] for credential in credentials}
evidence.update({"clock": "monotonic_seconds", "requests": samples,
"restartStarted": None, "restartFinished": None})
def sample(credential, body, context):
host = probe.validate_host(credential.label)
started = time.monotonic()
success = False
try:
probe.check_https(host, "/app", body, context)
success = True
except (probe.CheckError, OSError):
pass
finally:
samples[credential.label].append({"started": started, "finished": time.monotonic(),
"success": success})
def monitor(index, credential, body):
try:
context = ssl.create_default_context()
sample(credential, body, context)
except Exception:
samples[credential.label].append({"started": time.monotonic(),
"finished": time.monotonic(), "success": False})
return
finally:
baseline_done[index].set()
release.wait()
if not stop.is_set():
sample(credential, body, context)
workers = [threading.Thread(target=monitor, args=(index, credential, body), daemon=True)
for index, (credential, body) in enumerate(zip(credentials, expected))]
for worker in workers:
worker.start()
try:
wait_for(lambda: all(event.is_set() for event in baseline_done), timeout=30,
message="restart traffic monitor did not start")
if any(not samples[credential.label] or not samples[credential.label][0]["success"]
for credential in credentials):
raise AcceptanceError("public traffic failed before API restart")
evidence["restartStartedUnix"] = time.time()
evidence["restartStarted"] = time.monotonic()
try:
process = subprocess.Popen(["systemctl", "restart", "firecrab-api.service"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
release.set()
try:
status = process.wait(timeout=120)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
raise
if status:
raise AcceptanceError("API restart command failed")
api.ready()
finally:
evidence["restartFinished"] = time.monotonic()
evidence["restartFinishedUnix"] = time.time()
finally:
stop.set()
release.set()
for worker in workers:
worker.join(timeout=20)
if any(worker.is_alive() for worker in workers):
raise AcceptanceError("restart traffic monitor did not finish")
overlaps = {}
for credential in credentials:
rows = samples[credential.label]
if not rows or not all(row["success"] for row in rows):
raise AcceptanceError("public traffic failed during API restart")
overlaps[credential.label] = sum(
row["success"] and row["started"] < evidence["restartFinished"]
and row["finished"] > evidence["restartStarted"] for row in rows
)
evidence["overlappingSuccessfulRequests"] = overlaps
if not all(overlaps.values()):
raise AcceptanceError("both addresses need a successful request overlapping the API restart")
def run_public(api, wait_for, service_a, service_b, credentials, evidence):
"""Run acceptance after the caller creates two exclusive fixture services.
The operator must revoke both test tokens at the prompts. Empty input only
resumes the test: an actual rejected registration and offline response are
required. Dashboard reservations and VM cleanup remain the caller's duty.
"""
metrics = {"status": "failed", "stages": [], "cleanupErrors": [],
"dashboardReservations": "operator_cleanup_required",
"dashboardRevocations": "operator_verification_required",
"connectionChecks": "not_completed",
"tokenAuthentication": {credential.label: "not_verified" for credential in credentials}}
evidence["publicTunnel"] = metrics
if len(credentials) != 2 or credentials[0].label == credentials[1].label:
raise AcceptanceError("two distinct test credentials are required")
cli = shutil.which("firecrab")
if not cli:
raise AcceptanceError("installed Firecrab CLI is required")
targets = fixture_targets(api, [service_a, service_b])
for credential in credentials:
verify_registration(credential)
metrics["tokenAuthentication"][credential.label] = "accepted"
metrics["stages"].append("authenticated registration for both test addresses")
attempted = []
primary, peer = credentials
try:
for credential, target, body in zip(credentials, targets, (b"fixture-A", b"fixture-B")):
attempted.append(credential)
run_cli(cli, "address", "connect", "--app", target, credential=credential)
ready(wait_for, credential, body)
metrics["stages"].append("two independent VM app connections with HTTPS and WebSocket echo")
run_cli(cli, "address", "connect", primary.label, "--nginx", service_b["spec"]["domain"])
ready(wait_for, primary, b"fixture-B")
ready(wait_for, peer, b"fixture-B")
run_cli(cli, "address", "connect", primary.label, "--app", targets[0])
ready(wait_for, primary, b"fixture-A")
metrics["stages"].append("retarget to saved Nginx route and back without disturbing peer")
metrics["restart"] = {}
restart_api(api, wait_for, credentials, (b"fixture-A", b"fixture-B"), metrics["restart"])
for credential, body in zip(credentials, (b"fixture-A", b"fixture-B")):
ready(wait_for, credential, body)
metrics["stages"].append("HTTPS continuity across API restart and WebSocket echo afterward")
if owned_binding(STORE, primary) is None:
raise AcceptanceError("primary local ownership disappeared")
run_cli(cli, "address", "delete", primary.label)
wait_offline(wait_for, primary)
ready(wait_for, peer, b"fixture-B")
run_cli(cli, "address", "connect", "--app", targets[0], credential=primary)
ready(wait_for, primary, b"fixture-A")
metrics["stages"].append("disconnect and reconnect preserve the peer address")
rejected = []
for credential, other in ((primary, peer), (peer, None)):
print(f"Revoke test token for {credential.label} in the dashboard, then press Enter.", flush=True)
input()
verify_revoked(credential)
metrics["tokenAuthentication"][credential.label] = "registration_rejected"
wait_offline(wait_for, credential)
if other:
ready(wait_for, other, b"fixture-B")
verify_updater_rejection(api, wait_for, credential, other)
rejected.append(credential)
for prior in rejected:
verify_revoked(prior)
wait_offline(wait_for, prior)
if other:
ready(wait_for, other, b"fixture-B")
metrics["stages"].append("registration rejected and public address offline after operator action: " + credential.label)
metrics["stages"].append("updater restart and periodic retry retained token rejection: " + credential.label)
metrics["connectionChecks"] = "passed"
metrics["status"] = "pending_operator_verification"
finally:
original_error = sys.exc_info()[0] is not None
for credential in reversed(attempted):
try:
if owned_binding(STORE, credential) is not None:
run_cli(cli, "address", "delete", credential.label)
if owned_binding(STORE, credential) is not None:
raise AcceptanceError("local address remains")
except Exception:
metrics["cleanupErrors"].append("local connection cleanup failed: " + credential.label)
try:
config = (api.call("GET", "/api/pipe").get("config") or {})
if config.get("quickTunnels") or config.get("quickTunnel"):
metrics["cleanupErrors"].append("connector cleanup is incomplete or unverified")
except Exception:
metrics["cleanupErrors"].append("connector cleanup is unverified")
if metrics["cleanupErrors"]:
metrics["status"] = "failed"
if not original_error:
raise AcceptanceError("public test connection cleanup failed")