File size: 4,634 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 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """Position awareness: what this instance can see decides what it is.
roles = detect(inputs) # inputs: what the operator wired up (flows? http? host listeners? chain rpc?)
Roles (several at once are normal):
edge-waf HTTP requests arrive (reverse proxy / log tail / inline) -> HTTP tiers, challenge, rate limit
host-firewall runs on the host that is protected (can list listeners, can call the firewall) -> posture + block
net-sensor sees NetFlow / IPFIX / sFlow or VPC flow logs of a network -> entity stream for every source
chain-node sees JSON-RPC and P2P of a blockchain node -> RPC method abuse, peer floods, mempool spam
lb-embedded runs inside a Go load balancer (the Go runtime) -> fast path only, decisions from the sidecar
The result also says what the layer can *do* here (actions available) so the policy is interpreted correctly.
"""
from __future__ import annotations
import os
import shutil
import socket
from dibaguard.shield.policy import KNOWN_SERVICES
CAPABILITIES = {
"edge-waf": ["score_http", "challenge", "rate_limit", "block_at_proxy"],
"host-firewall": ["posture_audit", "block_at_host", "close_port"],
"net-sensor": ["score_flows", "entity_risk", "site_pressure", "report_upstream"],
"chain-node": ["score_rpc", "peer_flood", "mempool_spam", "rate_limit_rpc"],
"lb-embedded": ["score_http_fast", "block_at_proxy", "rate_limit"],
}
def _listening_ports() -> list[dict]:
"""TCP/UDP listeners on this host without psutil: /proc/net on Linux, `netstat -ano` on Windows, `ss` if present."""
out = []
if os.path.exists("/proc/net/tcp"):
for proto, path in (("tcp", "/proc/net/tcp"), ("tcp6", "/proc/net/tcp6"), ("udp", "/proc/net/udp")):
try:
with open(path) as f:
next(f)
for line in f:
parts = line.split()
local, state = parts[1], parts[3]
if proto.startswith("tcp") and state != "0A": # LISTEN
continue
addr, port = local.rsplit(":", 1)
out.append({"proto": proto[:3], "port": int(port, 16), "addr": addr, "public": addr not in ("0100007F", "00000000000000000000000001000000")})
except OSError:
continue
elif os.name == "nt" and shutil.which("netstat"):
import subprocess
try:
txt = subprocess.run(["netstat", "-ano"], capture_output=True, text=True, timeout=15).stdout
except Exception: # noqa: BLE001
txt = ""
for line in txt.splitlines():
p = line.split()
if len(p) >= 4 and p[0] in ("TCP", "UDP") and (p[0] == "UDP" or p[3] == "LISTENING"):
addr, _, port = p[1].rpartition(":")
if port.isdigit():
out.append({"proto": p[0].lower(), "port": int(port), "addr": addr, "public": addr not in ("127.0.0.1", "[::1]")})
seen, uniq = set(), []
for l in out:
k = (l["proto"], l["port"], l["public"])
if k not in seen:
seen.add(k)
uniq.append(dict(l, service=KNOWN_SERVICES.get(l["port"], "")))
return uniq
def detect(inputs: dict | None = None) -> dict:
"""inputs: {"flows": bool, "http": bool, "host": bool, "rpc": bool, "lb": bool} (None = probe what is possible here)"""
inputs = dict(inputs or {})
host_ok = inputs.get("host")
if host_ok is None:
host_ok = os.path.exists("/proc/net/tcp") or (os.name == "nt")
roles = []
if inputs.get("http"):
roles.append("edge-waf")
if host_ok:
roles.append("host-firewall")
if inputs.get("flows"):
roles.append("net-sensor")
if inputs.get("rpc"):
roles.append("chain-node")
if inputs.get("lb"):
roles.append("lb-embedded")
listeners = _listening_ports() if host_ok else []
if not inputs.get("rpc") and any(l["service"] in ("eth-rpc", "eth-ws", "cosmos-rpc", "bitcoin-rpc") for l in listeners):
roles.append("chain-node") # a chain node is running here: its RPC is worth guarding
fw = "nftables" if shutil.which("nft") else "iptables" if shutil.which("iptables") else "windows" if os.name == "nt" else "none"
caps = sorted({c for r in roles for c in CAPABILITIES[r]})
if fw == "none":
caps = [c for c in caps if c not in ("block_at_host", "close_port")]
return {"hostname": socket.gethostname(), "roles": roles, "capabilities": caps, "firewall": fw,
"listeners": listeners, "inputs": inputs}
|