diff --git a/app.py b/app.py index 00c017e..187d584 100644 --- a/app.py +++ b/app.py @@ -1258,62 +1258,6 @@ async def system_sysctl_audit(host: str, _=Depends(_verify)): raise HTTPException(502, "sysctl audit returned invalid JSON") from exc return {"host": host, **result} - -def _host_forensics_command(since_hours: int) -> str: - script = f'''import glob, json, os, subprocess - -def run(command): - proc = subprocess.run(command, shell=True, text=True, capture_output=True, timeout=30) - return {{"rc": proc.returncode, "stdout": proc.stdout.strip()[-12000:], "stderr": proc.stderr.strip()[-1000:]}} - -checks = {{ - "hostname": run("hostnamectl --static 2>/dev/null || hostname"), - "uptime": run("uptime"), - "disk": run("df -hT / /var/lib/docker 2>/dev/null || df -hT /"), - "failed_units": run("systemctl --failed --no-legend --no-pager"), - "docker_binary": run("command -v docker || true"), - "docker_packages": run("dpkg-query -W -f='${{Package}}|${{Status}}|${{Version}}\\n' 'docker*' 'containerd*' 2>/dev/null || true"), - "docker_units": run("systemctl is-active docker containerd 2>/dev/null; systemctl is-enabled docker containerd 2>/dev/null"), - "docker_containers": run("docker ps -a --format '{{{{.Names}}}}|{{{{.Image}}}}|{{{{.Status}}}}' 2>/dev/null || true"), - "docker_images": run("docker image ls --format '{{{{.Repository}}}}:{{{{.Tag}}}}|{{{{.ID}}}}|{{{{.Size}}}}' 2>/dev/null || true"), - "docker_volumes": run("docker volume ls --format '{{{{.Name}}}}' 2>/dev/null || true"), - "docker_disk_usage": run("docker system df 2>/dev/null || true"), - "iptables_docker_refs": run("iptables-save 2>/dev/null | grep -ci docker || true"), - "nft_docker_refs": run("nft list ruleset 2>/dev/null | grep -ci docker || true"), - "forward_policy": run("iptables -S FORWARD 2>/dev/null | head -40"), - "lvm": run("lvs -o lv_name,lv_size,data_percent,metadata_percent --units g --noheadings 2>/dev/null || true"), - "qemu_configs": run("ls -l /etc/pve/nodes/$(hostname)/qemu-server 2>/dev/null || true"), - "recent_system_files": run("find /etc/systemd/system /etc/docker /etc/network -type f -mmin -{since_hours * 60} -printf '%TY-%Tm-%Td %TH:%TM:%TS %p\\n' 2>/dev/null | sort"), - "recent_iso_builder_files": run("find /app-config/ansible/iso-builder -type f -mmin -{since_hours * 60} -printf '%TY-%Tm-%Td %TH:%TM:%TS %p\\n' 2>/dev/null | sort"), - "ansible_git_status": run("git -C /app-config/ansible status --short 2>/dev/null || true"), - "iso_builder_hashes": run("sha256sum /app-config/ansible/iso-builder/* 2>/dev/null || true"), -}} -print(json.dumps(checks)) -''' - encoded = base64.b64encode(script.encode()).decode() - return f'python3 -c "import base64;exec(base64.b64decode(\'{encoded}\'))"' - - -@app.get("/system/forensics/{host}") -async def system_forensics(host: str, since_hours: int = Query(48, ge=1, le=168), _=Depends(_verify)): - """Read-only host residue audit for failed deployments and package/network drift.""" - if not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,62}", host): - raise HTTPException(400, "Invalid host name") - inventory = await asyncio.to_thread(_find_inventory_host, host) - if not inventory: - raise HTTPException(404, f"Host {host} not found") - target = f'{inventory["user"]}@{inventory["ip"]}' - rc, out, err = await asyncio.to_thread(_ssh, target, _host_forensics_command(since_hours), 60) - if rc != 0: - raise HTTPException(502, (err or out).strip()[-500:] or "host forensics failed") - try: - result = json.loads(out) - except json.JSONDecodeError as exc: - raise HTTPException(502, "host forensics returned invalid JSON") from exc - _audit(f"/system/forensics/{host}", "GET", 200, f"since_hours={since_hours}") - return {"host": host, "since_hours": since_hours, "checks": result} - - def _pve_auth(): pv = _parse_kv("proxmox") return f"PVEAPIToken={pv.get('tokenid','')}={pv.get('secret','')}" diff --git a/tests/test_app.py b/tests/test_app.py index 8c32eb7..34932ce 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -174,33 +174,6 @@ def test_sysctl_audit_rejects_unknown_host_without_ssh(monkeypatch): assert response.status_code == 404 -def test_host_forensics_is_read_only_and_uses_inventory(monkeypatch): - payload = {"docker_binary": {"rc": 0, "stdout": "/usr/bin/docker", "stderr": ""}} - calls = [] - monkeypatch.setattr(app, "_find_inventory_host", lambda name: {"name": name, "user": "root", "ip": "10.5.85.13"}) - monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=600: (calls.append((host, command, timeout)) or (0, __import__("json").dumps(payload), ""))) - with TestClient(app.app) as client: - response = client.get("/system/forensics/node3?since_hours=24", headers={"Authorization": "Bearer test-token"}) - assert response.status_code == 200 - assert response.json()["checks"] == payload - assert calls[0][0] == "root@10.5.85.13" - assert calls[0][2] == 60 - assert "base64.b64decode" in calls[0][1] - command = app._host_forensics_command(24) - for destructive in ("systemctl restart", "systemctl stop", "systemctl disable", "docker rm", "docker system prune", "iptables -F", "rm -rf"): - assert destructive not in command - - -def test_host_forensics_rejects_unknown_host_and_invalid_window(monkeypatch): - monkeypatch.setattr(app, "_find_inventory_host", lambda _name: None) - monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not SSH"))) - with TestClient(app.app) as client: - missing = client.get("/system/forensics/not-there", headers={"Authorization": "Bearer test-token"}) - bad_window = client.get("/system/forensics/node3?since_hours=999", headers={"Authorization": "Bearer test-token"}) - assert missing.status_code == 404 - assert bad_window.status_code == 422 - - def test_invalid_log_target_is_rejected_before_ssh(): with TestClient(app.app) as client: response = client.get(