fix: enforce status transport boundaries

Author: Nickfost

docs/STATUS-REPORTING.md

@@ -33,7 +33,7 @@ All times are Unix seconds. Commit values are empty when unavailable. Receiver v
 
 ## Authentication
 
-Each controller receives a unique random HMAC key. A key must contain 32-128 bytes and be owned by root with mode `0600` on the controller. The receiver keeps a separate copy owned by its service account with mode `0600`.
+Each controller receives a unique random HMAC key. A key must contain 32-128 bytes and be owned by root with mode `0600` on the controller. The receiver keeps a separate copy owned by its service account with mode `0600`. Key files are read byte-for-byte, including leading or trailing whitespace; copy the same raw bytes to both sides.
 
 Every report includes:
 

host/systemd/ci-fleet-health.service

@@ -8,5 +8,6 @@ Type=oneshot
 User=root
 WorkingDirectory=/opt/ci-fleet/manager/current
 EnvironmentFile=/etc/ci-fleet/ci-fleet.env
+Environment=CI_FLEET_HEALTH_DELIVER_STATUS=1
 ExecStart=/opt/ci-fleet/manager/current/scripts/healthcheck.sh
 SuccessExitStatus=1

scripts/health.py

@@ -579,6 +579,7 @@ def _send_status(
         return 1 if values.get("CI_FLEET_HEALTH_HEARTBEAT_URL") else 0
     try:
         parsed = urllib.parse.urlsplit(url)
+        parsed.port
     except ValueError:
         return 1
     if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.path != "/v1/status" or parsed.query or parsed.fragment:
@@ -592,7 +593,7 @@ def _send_status(
         expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0
         if info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) & 0o077:
             return 1
-        key = path.read_bytes().strip()
+        key = path.read_bytes()
     except OSError:
         return 1
     if not 32 <= len(key) <= 128:
@@ -611,7 +612,7 @@ def _send_status(
         transport = opener or urllib.request.build_opener(_NoRedirect).open
         with transport(request, timeout=10) as response:
             return 0 if 200 <= response.status < 300 else 1
-    except OSError:
+    except (OSError, ValueError):
         return 1
 
 
@@ -622,7 +623,7 @@ def _local(args: argparse.Namespace) -> int:
     report = evaluate(snapshot, thresholds_from(values))
     now = int(time.time())
     report["timestamp"] = now
-    delivery = _send_status(values, build_status_report(snapshot, report, generated_at=now), now=now)
+    delivery = _send_status(values, build_status_report(snapshot, report, generated_at=now), now=now) if values.get("CI_FLEET_HEALTH_DELIVER_STATUS") == "1" else 0
     if delivery:
         report["checks"].append({"id": "status_delivery", "status": "warning"})
         if report["exit_code"] == 0:

scripts/status_receiver.py

@@ -8,6 +8,7 @@
 import json
 import os
 import re
+import socket
 import sqlite3
 import threading
 import time
@@ -136,7 +137,7 @@ def integer(value: Any, minimum: int = 0) -> bool:
         def number(value: Any, minimum: float = 0) -> bool:
             return isinstance(value, (int, float)) and not isinstance(value, bool) and value >= minimum and value < float("inf")
 
-        if not isinstance(report, dict) or report.get("schema_version") != 1:
+        if not isinstance(report, dict) or type(report.get("schema_version")) is not int or report["schema_version"] != 1:
             raise StatusError(400, "unsupported_schema")
         root_keys = {"schema_version", "controller", "configuration", "reconciliation", "drift", "process", "timers", "runners", "metrics", "docker", "error", "generated_at"}
         if set(report) != root_keys:
@@ -247,6 +248,7 @@ def __init__(self, *args: Any, max_requests: int = 32, request_timeout: int = 15
         self.max_requests = max_requests
         self.request_timeout = request_timeout
         self._slots = threading.BoundedSemaphore(max_requests)
+        self._deadlines: dict[Any, threading.Timer] = {}
         super().__init__(*args, **kwargs)
 
     def process_request(self, request: Any, client_address: Any) -> None:
@@ -255,15 +257,32 @@ def process_request(self, request: Any, client_address: Any) -> None:
             return
         try:
             request.settimeout(self.request_timeout)
+            timer = threading.Timer(self.request_timeout, self._expire_request, (request,))
+            timer.daemon = True
+            self._deadlines[request] = timer
+            timer.start()
             super().process_request(request, client_address)
         except Exception:
+            timer = self._deadlines.pop(request, None)
+            if timer:
+                timer.cancel()
             self._slots.release()
             raise
 
+    @staticmethod
+    def _expire_request(request: Any) -> None:
+        try:
+            request.shutdown(socket.SHUT_RDWR)
+        except OSError:
+            pass
+
     def process_request_thread(self, request: Any, client_address: Any) -> None:
         try:
             super().process_request_thread(request, client_address)
         finally:
+            timer = self._deadlines.pop(request, None)
+            if timer:
+                timer.cancel()
             self._slots.release()
 
 
@@ -324,11 +343,13 @@ def log_message(self, format: str, *args: Any) -> None:
     return _BoundedHTTPServer((bind, port), Handler)
 
 
-def _read_secret(path: Path) -> bytes:
+def _read_secret(path: Path, *, textual: bool = False) -> bytes:
     info = path.stat()
     if info.st_uid != os.getuid() or info.st_mode & 0o077:
         raise ValueError(f"secret must be owned by the receiver user with mode 0600: {path}")
-    value = path.read_bytes().strip()
+    value = path.read_bytes()
+    if textual:
+        value = value.strip()
     if not 32 <= len(value) <= 128:
         raise ValueError(f"secret must contain 32-128 bytes: {path}")
     return value
@@ -349,7 +370,7 @@ def load_auth_config(path: Path) -> tuple[dict[str, bytes], str]:
         keys[controller] = _read_secret(resolve(key_file))
     if not keys or not isinstance(value["read_token_file"], str):
         raise ValueError("auth config requires at least one controller and a read token")
-    return keys, _read_secret(resolve(value["read_token_file"])).decode()
+    return keys, _read_secret(resolve(value["read_token_file"]), textual=True).decode()
 
 
 def main() -> int:

scripts/test-install-worker-controller.sh

@@ -3,8 +3,10 @@ set -Eeuo pipefail
 
 repo_root=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)
 health_timer=$repo_root/host/systemd/ci-fleet-health.timer
+health_service=$repo_root/host/systemd/ci-fleet-health.service
 grep -Fqx 'OnActiveSec=2min' "$health_timer" || { printf 'FAIL: health timer lacks activation-relative initial trigger\n' >&2; exit 1; }
 ! grep -Fq 'OnBootSec=' "$health_timer" || { printf 'FAIL: health timer initial trigger is boot-relative\n' >&2; exit 1; }
+grep -Fqx 'Environment=CI_FLEET_HEALTH_DELIVER_STATUS=1' "$health_service" || { printf 'FAIL: scheduled health does not enable status delivery\n' >&2; exit 1; }
 tmp=$(mktemp -d)
 trap 'rm -rf "$tmp"' EXIT
 fake_bin=$tmp/bin
@@ -496,7 +498,7 @@ warning_output=$tmp/warning-upgrade.out
 "$installer" --upgrade "${base_args[@]}" --ref "$warning_ref" >"$warning_output" 2>&1
 warning_upgrade=$(<"$warning_output")
 grep -Fq 'WARNING disk_root' <<<"$warning_upgrade" || fail 'warning health fixture did not produce a warning result'
-grep -Fq 'WARNING status_delivery' <<<"$warning_upgrade" || fail 'status reporting outage was not observable as a warning'
+if grep -Fq 'WARNING status_delivery' <<<"$warning_upgrade"; then fail 'ad-hoc reconciliation health check submitted duplicate status'; fi
 grep -Fq 'CONVERGED mode=upgrade' <<<"$warning_upgrade" || fail 'warning health result did not report convergence'
 grep -Fq "CI_FLEET_CONFIG_REF=$warning_ref" "$root/etc/ci-fleet/ci-fleet.env" || fail 'warning health result rolled back an otherwise healthy activation'
 rm -f "$root/etc/ci-fleet/monitoring.env"

scripts/test_health.py

@@ -345,11 +345,14 @@ def opener(request, timeout):
             try:
                 self.assertEqual(health._send_status(values, report, now=1_000, nonce="a" * 32, opener=opener), 0)
                 self.assertIn("CI-Fleet-HMAC-SHA256", captured["headers"]["Authorization"])
+                expected = health.sign_headers("example-ci-01", captured["body"], key.read_bytes(), timestamp=1_000, nonce="a" * 32)
+                self.assertEqual(captured["headers"]["Authorization"], expected["Authorization"])
                 original = copy.deepcopy(report)
                 self.assertEqual(health._send_status(values, report, now=1_000, nonce="b" * 32, opener=lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError())), 1)
                 self.assertEqual(report, original)
                 self.assertEqual(health._send_status({"CI_FLEET_HEALTH_STATUS_URL": "http://unsafe.invalid"}, report), 1)
                 self.assertEqual(health._send_status({"CI_FLEET_HEALTH_STATUS_URL": "https://[bad/v1/status"}, report), 1)
+                self.assertEqual(health._send_status({"CI_FLEET_HEALTH_STATUS_URL": "https://status.example.invalid:bad/v1/status"}, report), 1)
                 self.assertEqual(health._send_status({"CI_FLEET_HEALTH_HEARTBEAT_URL": "https://legacy.invalid"}, report), 1)
                 self.assertIsNone(health._NoRedirect().redirect_request(None, None, 302, None, {}, None))
             finally:

scripts/test_status_receiver.py

@@ -177,6 +177,11 @@ def test_schema_compatibility_and_malformed_metrics(self) -> None:
         body, headers = self.signed(future)
         self.assert_status_error(400, "unsupported_schema", lambda: self.receiver.submit(body, headers, now=1_000))
 
+        boolean = valid_report()
+        boolean["schema_version"] = True
+        body, headers = self.signed(boolean, nonce="f" * 32)
+        self.assert_status_error(400, "unsupported_schema", lambda: self.receiver.submit(body, headers, now=1_000))
+
         malformed = valid_report()
         malformed["metrics"]["memory"]["available_bytes"] = -1
         body, headers = self.signed(malformed, nonce="b" * 32)
@@ -209,15 +214,16 @@ def test_time_retention_is_enforced_on_read(self) -> None:
     def test_receiver_restart_reloads_rotated_key(self) -> None:
         directory = Path(self.temporary.name)
         key_file, token_file, config_file = directory / "key", directory / "token", directory / "auth.json"
-        key_file.write_bytes(b"a" * 32)
+        raw_key = b"\n" + b"a" * 30 + b" "
+        key_file.write_bytes(raw_key)
         token_file.write_bytes(b"r" * 32)
         config_file.write_text(json.dumps({"controllers": {"example-ci-01": "key"}, "read_token_file": "token"}))
         for path in (key_file, token_file, config_file):
             path.chmod(0o600)
         first, _ = status_receiver.load_auth_config(config_file)
         key_file.write_bytes(b"b" * 32)
         second, _ = status_receiver.load_auth_config(config_file)
-        self.assertEqual((first["example-ci-01"], second["example-ci-01"]), (b"a" * 32, b"b" * 32))
+        self.assertEqual((first["example-ci-01"], second["example-ci-01"]), (raw_key, b"b" * 32))
 
     def test_http_post_and_read_only_api(self) -> None:
         server = status_receiver.create_server("127.0.0.1", 0, self.receiver)
@@ -261,6 +267,17 @@ def test_http_server_bounds_slow_clients(self) -> None:
         for _ in range(server.max_requests):
             server._slots.release()
 
+        expired = threading.Event()
+        class SlowRequest:
+            def shutdown(self, how: int) -> None:
+                self.how = how
+                expired.set()
+        request = SlowRequest()
+        timer = threading.Timer(0.01, server._expire_request, (request,))
+        timer.start()
+        self.assertTrue(expired.wait(1))
+        self.assertEqual(request.how, status_receiver.socket.SHUT_RDWR)
+
 
 if __name__ == "__main__":
     unittest.main()