| """Posture audit: what the host exposes that the policy did not ask for - and the fix, taken or proposed. |
| |
| findings = audit(position, policy) # list of findings with severity, evidence, fix |
| taken = remediate(findings, policy, enforcer) # closes what policy.posture.close_unexpected allows (else reports) |
| |
| Checks (each a finding with an id, so an LLM or a human can refer to it): |
| P1 listener on a public address whose port is not in policy.posture.allowed_ports |
| P2 a service that must never face the public (redis, mongodb, elasticsearch, memcached, docker, db, rdp, smb, vnc) |
| P3 plaintext management services (telnet, ftp) listening |
| P4 no host firewall backend available while the policy is in enforce mode |
| P5 chain node RPC (8545/26657/8332) listening on a public address |
| """ |
| from __future__ import annotations |
|
|
| from dibaguard.shield.policy import KNOWN_SERVICES |
|
|
| SEVERITY = {"P1": 0.5, "P2": 0.9, "P3": 0.8, "P4": 0.6, "P5": 0.85} |
|
|
|
|
| def audit(position: dict, policy) -> list[dict]: |
| allowed = set(int(p) for p in policy.d["posture"]["allowed_ports"]) |
| deny = set(policy.d["posture"]["deny_public_services"]) |
| out = [] |
| for l in position.get("listeners", []): |
| svc = l.get("service") or KNOWN_SERVICES.get(l["port"], "") |
| if l["public"] and l["port"] not in allowed: |
| out.append({"id": "P1", "severity": SEVERITY["P1"], "port": l["port"], "proto": l["proto"], "service": svc, |
| "evidence": f"{l['proto']}/{l['port']} listening on {l['addr']} (public), not in allowed_ports", |
| "fix": f"close {l['proto']}/{l['port']} or add it to policy.posture.allowed_ports"}) |
| if l["public"] and svc in deny: |
| out.append({"id": "P2", "severity": SEVERITY["P2"], "port": l["port"], "proto": l["proto"], "service": svc, |
| "evidence": f"{svc} ({l['proto']}/{l['port']}) is reachable from outside", |
| "fix": f"bind {svc} to localhost or a private interface; firewall {l['proto']}/{l['port']}"}) |
| if svc in ("telnet", "ftp"): |
| out.append({"id": "P3", "severity": SEVERITY["P3"], "port": l["port"], "proto": l["proto"], "service": svc, |
| "evidence": f"plaintext {svc} on {l['proto']}/{l['port']}", "fix": f"disable {svc}; use ssh / sftp"}) |
| if l["public"] and svc in ("eth-rpc", "eth-ws", "cosmos-rpc", "bitcoin-rpc"): |
| out.append({"id": "P5", "severity": SEVERITY["P5"], "port": l["port"], "proto": l["proto"], "service": svc, |
| "evidence": f"chain RPC {svc} on {l['proto']}/{l['port']} is public", |
| "fix": "put the RPC behind the proxy with method allowlist + per-client rate limits; never expose admin/debug namespaces"}) |
| if policy.enforcing and position.get("firewall") == "none": |
| out.append({"id": "P4", "severity": SEVERITY["P4"], "evidence": "policy.mode = enforce but no firewall backend (nft/iptables/netsh) is available", |
| "fix": "install nftables or run the layer where it can reach the firewall; until then the layer reports only"}) |
| out.sort(key=lambda f: -f["severity"]) |
| return out |
|
|
|
|
| def remediate(findings: list[dict], policy, enforcer=None) -> list[dict]: |
| """close what the policy allows closing (posture.close_unexpected), report the rest; returns what was done/proposed""" |
| done = [] |
| close = bool(policy.d["posture"].get("close_unexpected")) and policy.enforcing and enforcer is not None |
| for f in findings: |
| if f["id"] in ("P1", "P2", "P5") and "port" in f: |
| if close: |
| ok = enforcer.backend.run(["nft", "add", "rule", "inet", "dibaguard", "input", f["proto"], "dport", str(f["port"]), "drop"]) \ |
| if enforcer.backend.name == "nftables" else enforcer.backend.run(["dryrun", "close", f"{f['proto']}/{f['port']}"]) |
| done.append(dict(f, taken=ok, how="firewall drop on the port (not the service itself)")) |
| else: |
| done.append(dict(f, taken=False, how="reported; set posture.close_unexpected=true and mode=enforce to close automatically")) |
| else: |
| done.append(dict(f, taken=False, how="reported")) |
| return done |
|
|
