cesnimda
0954876295
Implement B01-B06 with regression coverage and retained validation evidence. Preserve the production ingress preflight; F04 runtime verification remains open.
82 lines
4.3 KiB
Plaintext
82 lines
4.3 KiB
Plaintext
"""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))
|