File size: 4,256 Bytes
5d97275 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | """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
|