cesnimda
0954876295
Implement B01-B06 with regression coverage and retained validation evidence. Preserve the production ingress preflight; F04 runtime verification remains open.
127 lines
7.7 KiB
Python
127 lines
7.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Bounded, synthetic nginx topology; uses an installed image, never pulls/deploys."""
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def docker(*args):
|
|
result = subprocess.run(["docker", *args], capture_output=True, text=True, timeout=30)
|
|
if result.returncode:
|
|
raise RuntimeError(result.stderr.strip())
|
|
return result.stdout.strip()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--image", default="nginx:1.29.8-alpine", help="Already installed nginx image with realip and wget")
|
|
parser.add_argument("--baseline", action="store_true", help="Use HEAD's pre-B06 nginx sources and demonstrate the old bypass")
|
|
args = parser.parse_args()
|
|
docker("info", "--format", "{{.ServerVersion}}")
|
|
docker("image", "inspect", args.image) # Fail before resource creation if unavailable.
|
|
prefix = "b06-" + uuid.uuid4().hex[:10]
|
|
containers, networks = [], []
|
|
artifact_root = ROOT / ".artifacts" / "ingress-topology"
|
|
artifact_root.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.TemporaryDirectory(prefix=prefix, dir=artifact_root) as directory:
|
|
directory = Path(directory).resolve()
|
|
assert directory.is_relative_to(artifact_root.resolve())
|
|
def create(name, *options):
|
|
name = prefix + "-" + name
|
|
docker("create", "--pull=never", "--name", name, *options)
|
|
containers.append(name)
|
|
return name
|
|
try:
|
|
edge = prefix + "-edge"
|
|
private = prefix + "-private"
|
|
docker("network", "create", edge)
|
|
networks.append(edge)
|
|
docker("network", "create", "--internal", private)
|
|
networks.append(private)
|
|
peer = create("peer", "--network", edge, "--entrypoint", "sleep", args.image, "300")
|
|
docker("start", peer)
|
|
peer_state = json.loads(docker("inspect", peer))[0]
|
|
peer_ip = peer_state["NetworkSettings"]["Networks"][edge]["IPAddress"]
|
|
for source, target in [("nginx.conf.template", "template"), ("configure-nginx-origin.sh", "configure.sh")]:
|
|
content = (subprocess.check_output(["git", "show", f"HEAD:job-tracker-ui/{source}"], cwd=ROOT).decode()
|
|
if args.baseline else (ROOT / "job-tracker-ui" / source).read_text())
|
|
(directory / target).write_text(content, newline="\n")
|
|
(directory / "backend.conf").write_text('''server {
|
|
listen 8080;
|
|
location / { default_type application/json;
|
|
return 200 '{"ip":"$http_x_forwarded_for","scheme":"$http_x_forwarded_proto","host":"$http_host","forwardedHost":"$http_x_forwarded_host","forwarded":"$http_forwarded","session":"$http_x_session_context"}'; }
|
|
}
|
|
''', newline="\n")
|
|
backend = create("backend", "--network", private, "--network-alias", "backend-web",
|
|
"-v", f"{directory / 'backend.conf'}:/etc/nginx/conf.d/default.conf:ro", args.image)
|
|
docker("start", backend)
|
|
frontend = create("frontend", "--network", edge, "--network-alias", "frontend",
|
|
"-p", "127.0.0.1::80", "-e", "APP_PUBLIC_BASE_URL=https://jobs.example.test",
|
|
"-e", "INGRESS_MODE=proxy", "-e", f"INGRESS_PROXY_CIDRS={peer_ip}/32",
|
|
"-v", f"{directory / 'template'}:/etc/nginx/jobtracker.conf.template:ro",
|
|
"-v", f"{directory / 'configure.sh'}:/docker-entrypoint.d/15-jobtracker-origin.sh:ro",
|
|
# Invoke through sh so mounted Windows execute bits do not determine the result.
|
|
"--entrypoint", "sh", args.image, "-c",
|
|
"sh /docker-entrypoint.d/15-jobtracker-origin.sh && exec nginx -g 'daemon off;'")
|
|
docker("network", "connect", private, frontend)
|
|
docker("start", frontend)
|
|
docker("exec", frontend, "sh", "-c", "for i in 1 2 3 4 5; do wget -q -O /dev/null http://127.0.0.1/health && exit 0; sleep 1; done; exit 1")
|
|
headers = {"Host": "jobs.example.test", "X-Forwarded-For": "198.51.100.10", "X-Forwarded-Proto": "https",
|
|
"X-Forwarded-Host": "attacker.example.test", "Forwarded": "host=attacker.example.test", "X-Session-Context": "synthetic-session"}
|
|
def proxied(extra=None):
|
|
values = {**headers, **(extra or {})}
|
|
command = ["docker", "exec", peer, "wget", "-S", "-O", "-"]
|
|
for name, value in values.items():
|
|
command.append(f"--header={name}: {value}")
|
|
command.append("http://frontend/api/probe")
|
|
return subprocess.run(command, capture_output=True, text=True, timeout=10)
|
|
binding = docker("port", frontend, "80/tcp")
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
|
request = urllib.request.Request(f"http://{binding}/api/probe", headers=headers)
|
|
try:
|
|
with opener.open(request, timeout=10) as response:
|
|
direct_status, direct_body = response.status, response.read().decode()
|
|
except urllib.error.HTTPError as error:
|
|
direct_status, direct_body = error.code, error.read().decode()
|
|
if args.baseline:
|
|
assert direct_status == 200 and json.loads(direct_body)["ip"] == "198.51.100.10"
|
|
print(json.dumps({"baselineDirect": direct_status, "forgedIpAccepted": True}))
|
|
return
|
|
assert direct_status == 403, (direct_status, direct_body)
|
|
successful = []
|
|
for client in ["198.51.100.10", "198.51.100.11", "2001:db8::10"]:
|
|
result = proxied({"X-Forwarded-For": client})
|
|
assert result.returncode == 0, result.stderr
|
|
body = json.loads(result.stdout)
|
|
assert body == {"ip": client, "scheme": "https", "host": "jobs.example.test", "forwardedHost": "", "forwarded": "", "session": "synthetic-session"}, body
|
|
successful.append(client)
|
|
for extra in [{"X-Forwarded-For": "203.0.113.9, 198.51.100.10"}, {"X-Forwarded-Proto": "http"}, {"X-Forwarded-Proto": "https,http"}]:
|
|
result = proxied(extra)
|
|
assert result.returncode != 0 and "400" in result.stderr, result.stderr
|
|
malformed = json.loads(proxied({"X-Forwarded-For": "invalid-address"}).stdout)
|
|
assert malformed["ip"] == peer_ip # realip fails closed to the real peer, not forged data.
|
|
backend_state = json.loads(docker("inspect", backend))[0]
|
|
assert not any(backend_state["NetworkSettings"]["Ports"].values())
|
|
backend_ip = backend_state["NetworkSettings"]["Networks"][private]["IPAddress"]
|
|
# A peer attached only to the edge network cannot route directly to the internal backend.
|
|
denied = subprocess.run(["docker", "exec", peer, "wget", "-T", "2", "-qO-", f"http://{backend_ip}:8080/api/probe"], capture_output=True, timeout=5)
|
|
assert denied.returncode != 0
|
|
print(json.dumps({"directNginx": direct_status, "trustedClientIps": successful, "chainsAndBadScheme": 400,
|
|
"malformedIp": "socket-peer fallback", "directBackend": "unreachable", "backendPublishedPorts": 0}))
|
|
finally:
|
|
for name in reversed(containers):
|
|
subprocess.run(["docker", "rm", "-f", name], capture_output=True, timeout=30)
|
|
for name in reversed(networks):
|
|
subprocess.run(["docker", "network", "rm", name], capture_output=True, timeout=30)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|