fix(security): enforce audited account, network and resource boundaries

Implement B01-B06 with regression coverage and retained validation evidence. Preserve the production ingress preflight; F04 runtime verification remains open.
2026-09-09 18:21:47 +02:00
parent 7d641edc7b
commit 0954876295
116 changed files with 10101 additions and 122 deletions
@@ -0,0 +1,81 @@
+"""Bounded, read-only B06 observations; no credentials, cookies or response bodies retained."""
+import concurrent.futures
+import datetime
+import json
+from pathlib import Path
+import socket
+import ssl
+import subprocess
+
+OUT = Path(__file__).resolve().parent
+HOST = "jobs.cesnimda.uk"  # .env.example canonical production hostname
+SAFE_HEADERS = {"server", "location", "content-type", "content-length", "strict-transport-security",
+                "x-content-type-options", "x-frame-options", "referrer-policy", "permissions-policy"}
+
+
+def request(name, url, headers=()):
+    command = ["curl.exe", "--noproxy", "*", "--connect-timeout", "4", "--max-time", "8",
+               "--silent", "--show-error", "--dump-header", "-", "--output", "NUL",
+               "--write-out", "\nB06_META %{json}\n", url]
+    for header in headers:
+        command.extend(["--header", header])
+    result = subprocess.run(command, capture_output=True, text=True, timeout=12)
+    raw_headers, _, metadata = result.stdout.partition("B06_META ")
+    safe = []
+    for line in raw_headers.splitlines():
+        if line.startswith("HTTP/") or line.partition(":")[0].lower() in SAFE_HEADERS:
+            safe.append(line)
+    try:
+        info = json.loads(metadata)
+        meta = {key: info.get(key) for key in ("http_code", "remote_ip", "remote_port", "local_ip",
+                                              "ssl_verify_result", "time_total", "exitcode")}
+    except ValueError:
+        meta = {}
+    return {"name": name, "command": command, "exit": result.returncode, "headers": safe,
+            "metadata": meta, "stderr": result.stderr.strip()}
+
+
+def certificate():
+    try:
+        context = ssl.create_default_context()
+        with socket.create_connection((HOST, 443), timeout=8) as tcp:
+            with context.wrap_socket(tcp, server_hostname=HOST) as tls:
+                return {"verified": True, "peer": tls.getpeername(), "tls": tls.version(),
+                        "certificate": tls.getpeercert()}
+    except Exception as error:
+        return {"verified": False, "error": str(error)}
+
+
+if __name__ == "__main__":
+    addresses = sorted({item[4][0] for item in socket.getaddrinfo(HOST, None, type=socket.SOCK_STREAM)})
+    cases = [
+        ("public_https", f"https://{HOST}/api/auth/config", []),
+        ("public_http", f"http://{HOST}/api/auth/config", []),
+        ("public_https_forged_forwarding", f"https://{HOST}/api/auth/config",
+         ["X-Forwarded-For: 198.51.100.10", "X-Forwarded-Proto: http", "X-Forwarded-Host: attacker.example.test"]),
+    ]
+    for address in addresses:
+        literal = f"[{address}]" if ":" in address else address
+        for name, headers in [
+            ("none", []),
+            ("xff", ["X-Forwarded-For: 198.51.100.10"]),
+            ("proto", ["X-Forwarded-Proto: https"]),
+            ("host", ["Host: attacker.example.test"]),
+            ("combined", ["X-Forwarded-For: 198.51.100.10", "X-Forwarded-Proto: https",
+                          "X-Forwarded-Host: attacker.example.test"]),
+            ("malformed_chain", ["X-Forwarded-For: not-an-ip, 198.51.100.10", "X-Forwarded-Proto: https"]),
+        ]:
+            actual = headers if name == "host" else [f"Host: {HOST}"] + headers
+            cases.append((f"direct_3000_{address}_{name}", f"http://{literal}:3000/api/auth/config", actual))
+        for port in (5202, 8080):
+            cases.append((f"direct_api_{address}_{port}", f"http://{literal}:{port}/health", [f"Host: {HOST}"]))
+    # Three simultaneous harmless GETs at most; no login, mutations or quota exhaustion.
+    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
+        observations = list(pool.map(lambda args: request(*args), cases))
+    report = {"utc": datetime.datetime.now(datetime.timezone.utc).isoformat(), "hostname": HOST,
+              "dns_addresses": addresses, "certificate": certificate(), "requests": observations,
+              "vantage": "Connor Windows workstation; external independence/NAT hairpin not established",
+              "limitations": ["No assertion of Traefik identity or upstream header values from public status alone",
+                              "Connection failure does not identify which firewall/NAT/listener rejected it"]}
+    (OUT / "public-observations.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
+    print(json.dumps(report, indent=2))