#!/usr/bin/env python3 """Real API -> helper -> CLI -> Nginx test on a disposable Linux systemd host.""" import argparse import http.client import http.server import json import os from pathlib import Path import socket import shutil import subprocess import tempfile import threading import time def main(args): assert os.geteuid() == 0 and Path("/run/systemd/system").is_dir() unit = "firecrab-ingress.service" assert subprocess.check_output(["systemctl", "show", "--property=LoadState", "--value", unit], text=True).strip() == "not-found", "refusing existing ingress" class Backend(http.server.BaseHTTPRequestHandler): def do_GET(self): self.send_response(200); self.end_headers(); self.wfile.write(b"ready") def log_message(self, *args): pass backend = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Backend) threading.Thread(target=backend.serve_forever, daemon=True).start() def port(): with socket.socket() as listener: listener.bind(("127.0.0.1", 0)); return listener.getsockname()[1] api_port, ingress_port = port(), port() with tempfile.TemporaryDirectory(prefix="fc-pipe-api-", dir="/var/lib") as temporary: root = Path(temporary); root.chmod(0o755) binaries = root / "bin"; binaries.mkdir() for name, source in [("firecrab-api", args.api), ("firecrab-helper", args.helper), ("firecrab", str(Path(args.helper).with_name("firecrab")))]: shutil.copyfile(source, binaries / name); (binaries / name).chmod(0o755) shutil.copyfile(args.ingress, binaries / "firecrab-ingress") (binaries / "firecrab-ingress").chmod(0o755) working = root / "api"; working.mkdir(); os.chown(working, 65534, 65534) socket_path = root / "helper.sock" environment = dict(os.environ, FIRECRAB_LIBDIR=str(binaries), CONFDIR=str(root / "config"), UNITDIR="/run/systemd/system", FIRECRAB_NET_HELPER_SOCK=str(socket_path), FIRECRAB_NET_HELPER_ALLOWED_UID="65534", FIRECRAB_BIND_ADDR=f"127.0.0.1:{api_port}", FIRECRAB_IMAGE_ROOT=str(working / "images"), FIRECRAB_BOOTSTRAP_WARMUP="0", HOME=str(working)) helper_log = (root / "helper.log").open("w+") api_log = (root / "api.log").open("w+") helper_unit = root.name + "-helper.service" config_dir = root / "config"; config_dir.mkdir(); config_dir.chmod(0o750); os.chown(config_dir, 0, 65534) shared = root / "share"; shared.mkdir() source = Path(__file__).resolve().parent.parent template = (source / "packaging/systemd/firecrab-helper.service").read_text() for key, value in {"@FIRECRAB_GROUP@": "nogroup", "@FIRECRAB_UID@": "65534", "@PREFIX@": str(root), "@LIBDIR@": str(binaries), "@SHAREDIR@": str(shared), "@DATADIR@": str(working), "@CONFDIR@": str(config_dir), "@UNITDIR@": "/run/systemd/system"}.items(): template = template.replace(key, value) template = template.replace("RuntimeDirectory=firecrab", "RuntimeDirectory=" + root.name).replace("/run/firecrab ", "/run/" + root.name + " ") template = template.replace("[Install]", "Environment=FIRECRAB_NET_HELPER_SOCK=" + str(socket_path) + "\n\n[Install]") if profile := os.environ.get("LLVM_PROFILE_FILE"): # systemd expands % specifiers; LLVM must receive them literally. profile = profile.replace("\\", "\\\\").replace('"', '\\"').replace("%", "%%") template = template.replace("[Install]", 'Environment="LLVM_PROFILE_FILE=' + profile + '"\n\n[Install]') helper_path = Path("/run/systemd/system", helper_unit); helper_path.write_text(template) subprocess.run(["systemctl", "daemon-reload"], check=True) subprocess.run(["systemctl", "start", helper_unit], check=True) helper = None api = None def start_api(): def unprivileged(): os.setgroups([]); os.setgid(65534); os.setuid(65534) return subprocess.Popen([str(binaries / "firecrab-api")], cwd=working, env=environment, stdout=api_log, stderr=api_log, preexec_fn=unprivileged) def request(method, path, body=None, ingress=False): connection = http.client.HTTPConnection("127.0.0.1", ingress_port if ingress else api_port, timeout=30) try: connection.request(method, path, body=None if body is None else json.dumps(body), headers={"Host": "app.example.com", "Content-Type": "application/json"}) response = connection.getresponse(); return response.status, response.read() finally: connection.close() def ready(): for _ in range(100): try: if request("GET", "/api/pipe")[0] == 200: return except OSError: pass time.sleep(.1) raise AssertionError("API did not become ready") try: for _ in range(100): if socket_path.exists(): break assert subprocess.run(["systemctl", "is-active", "--quiet", helper_unit]).returncode == 0, "helper exited" time.sleep(.05) os.chown(socket_path, 0, 65534) api = start_api(); ready() config = {"listen": f"127.0.0.1:{ingress_port}", "routes": [{"host": "app.example.com", "pathPrefix": "/", "upstreams": [f"127.0.0.1:{backend.server_port}"]}], "tunnel": None} status, body = request("PUT", "/api/pipe", {"config": config, "credentials": None}) assert status == 200, body snapshot = json.loads(body); original = snapshot["units"][0]["mainPid"] assert original > 0 and request("GET", "/", ingress=True) == (200, b"ready") assert os.stat(root / "config/pipe/manifest.json").st_uid == 0 print("PASS unprivileged API applies routes through the sandboxed systemd helper", flush=True) rejected = json.loads(json.dumps(config)); rejected["routes"][0]["pathPrefix"] = '/api";return 200;' assert request("PUT", "/api/pipe", {"config": rejected, "credentials": None})[0] == 400 api.terminate(); api.wait(timeout=10) for _ in range(10): assert request("GET", "/", ingress=True) == (200, b"ready") api = start_api(); ready() snapshot = json.loads(request("GET", "/api/pipe")[1]) assert snapshot["units"][0]["mainPid"] == original and snapshot["config"] == config print("PASS real API stop/restart preserves ingress PID, traffic and saved routes", flush=True) def wait_unit(state, previous_pid=None): for _ in range(100): current = json.loads(request("GET", "/api/pipe")[1]) ingress = current["units"][0] assert current["config"] == config, "service control changed saved routes" if ingress["activeState"] == state and (state != "active" or ingress["mainPid"] > 0) and (previous_pid is None or ingress["mainPid"] != previous_pid): return ingress["mainPid"] time.sleep(.1) raise AssertionError(f"ingress did not become {state}") def wait_traffic(): for _ in range(100): try: if request("GET", "/", ingress=True) == (200, b"ready"): return except OSError: pass time.sleep(.1) raise AssertionError("ingress traffic did not resume") for action in ["stop", "start", "restart"]: previous_pid = json.loads(request("GET", "/api/pipe")[1])["units"][0]["mainPid"] status, body = request("POST", "/api/pipe/service", {"service": "ingress", "action": action}) assert status == 200, body wait_unit("inactive" if action == "stop" else "active", previous_pid if action == "restart" else None) if action == "stop": try: request("GET", "/", ingress=True) except OSError: pass else: raise AssertionError("stopped ingress still accepts requests") else: wait_traffic() print("PASS API starts, stops and restarts ingress without changing saved routes", flush=True) status, body = request("POST", "/api/pipe/disable") assert status == 200 and json.loads(body)["units"][0]["mainPid"] == 0, body print("PASS API disable controls the independent unit", flush=True) except BaseException: for log in [helper_log, api_log]: log.seek(0); print(log.read()) subprocess.run(["journalctl", "-u", helper_unit, "-n", "35", "--no-pager"]) raise finally: for process in [api, helper]: if process and process.poll() is None: process.terminate(); process.wait(timeout=10) subprocess.run(["systemctl", "stop", helper_unit], capture_output=True) helper_path.unlink(missing_ok=True) subprocess.run(["systemctl", "disable", "--now", unit], capture_output=True) Path("/run/systemd/system", unit).unlink(missing_ok=True) subprocess.run(["systemctl", "daemon-reload"], check=True) subprocess.run(["systemctl", "reset-failed", unit], capture_output=True) backend.shutdown(); backend.server_close() helper_log.close(); api_log.close() if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--api", required=True) parser.add_argument("--helper", required=True) parser.add_argument("--ingress", required=True) main(parser.parse_args())