| """Policy: the operator's contract with the layer. JSON, versioned, validated, with safe defaults (report only). |
| |
| Example (policy.json): |
| { |
| "mode": "report", # report | enforce |
| "alpha": 0.001, # false-alarm budget per benign (source, minute) |
| "actions": {"portscan": "block", "bruteforce": "block", "dos": "rate_limit", "ddos": "rate_limit", |
| "webattack": "block", "lateral": "block", "bot": "challenge", "recon": "monitor", "*": "monitor"}, |
| "min_action_score": 60, # 0-100 event score below which only 'monitor' is taken |
| "block_ttl_s": 1800, "rate_limit_ttl_s": 600, "challenge_ttl_s": 900, |
| "allow": ["10.0.0.0/8", "192.168.0.0/16"], # never acted upon (still reported) |
| "posture": {"allowed_ports": [22, 80, 443], "close_unexpected": false, "deny_public_services": ["redis", "mongodb", "elasticsearch", "memcached", "docker"]}, |
| "report": {"every_s": 300, "llm": false} |
| } |
| """ |
| from __future__ import annotations |
|
|
| import ipaddress |
| import json |
| import os |
|
|
| from dibaguard.shield.database import DB_PORTS |
|
|
| ACTIONS = ("monitor", "rate_limit", "challenge", "block") |
| DEFAULT = { |
| "version": 1, "mode": "report", "alpha": 1e-3, |
| "actions": {"portscan": "block", "bruteforce": "block", "dos": "rate_limit", "ddos": "rate_limit", "webattack": "block", "lateral": "block", |
| "sqli": "block", "xss": "block", "cmdi": "block", "rce": "block", "traversal": "block", "lfi": "block", |
| "nosqli": "block", "ssti": "block", "ssrf": "block", "xxe": "block", |
| "bot": "challenge", "scan": "challenge", "recon": "monitor", "attack_generic": "monitor", |
| "db_rce": "block", "db_privilege": "block", "db_destructive": "block", "db_injection": "block", "db_exfil": "rate_limit", |
| "db_bruteforce": "block", "db_recon": "monitor", "db_new_client": "monitor", "db_unbounded": "monitor", "*": "monitor"}, |
| "min_action_score": 60, "block_ttl_s": 1800, "rate_limit_ttl_s": 600, "challenge_ttl_s": 900, |
| "allow": ["127.0.0.0/8", "::1/128"], |
| "posture": {"allowed_ports": [22, 80, 443], "close_unexpected": False, |
| "deny_public_services": ["redis", "mongodb", "elasticsearch", "memcached", "docker", "mysql", "postgres", "rdp", "smb", "vnc", |
| "mssql", "oracle", "cassandra", "cassandra-internode", "cassandra-jmx", "couchdb", "neo4j", "clickhouse", |
| "cockroach", "influxdb", "kafka", "zookeeper", "etcd", "rethinkdb", "arangodb"]}, |
| |
| |
| |
| "database": {"block_app_clients": False, "correlate_window_s": 10, "auth_fail_floor": 8, "exempt_shapes": [], "learn_new_clients": False}, |
| "report": {"every_s": 300, "llm": False}, |
| |
| |
| |
| "verified_crawlers": [{"name": "googlebot", "ua": "googlebot", "cidrs": ["66.249.64.0/19"]}, |
| {"name": "bingbot", "ua": "bingbot", "cidrs": ["157.55.39.0/24", "207.46.13.0/24", "40.77.167.0/24"]}], |
| |
| |
| "machine_challenge": "rate_limit", |
| } |
| KNOWN_SERVICES = {6379: "redis", 27017: "mongodb", 9200: "elasticsearch", 11211: "memcached", 2375: "docker", 2376: "docker", |
| 3306: "mysql", 5432: "postgres", 3389: "rdp", 445: "smb", 139: "smb", 5900: "vnc", 23: "telnet", 21: "ftp", |
| 25: "smtp", 53: "dns", 22: "ssh", 80: "http", 443: "https", 8080: "http-alt", 8443: "https-alt", 9090: "prometheus", |
| 5601: "kibana", 15672: "rabbitmq", 8545: "eth-rpc", 8546: "eth-ws", 30303: "eth-p2p", 26656: "cosmos-p2p", |
| 26657: "cosmos-rpc", 8333: "bitcoin-p2p", 8332: "bitcoin-rpc"} |
| KNOWN_SERVICES.update({p: eng for p, (eng, _) in DB_PORTS.items() if p not in KNOWN_SERVICES}) |
|
|
|
|
| class Policy: |
| def __init__(self, data: dict | None = None): |
| self.d = json.loads(json.dumps(DEFAULT)) |
| for k, v in (data or {}).items(): |
| if isinstance(v, dict) and isinstance(self.d.get(k), dict): |
| self.d[k].update(v) |
| else: |
| self.d[k] = v |
| self.validate() |
| self._allow = [ipaddress.ip_network(a, strict=False) for a in self.d["allow"]] |
| self._crawlers = [(str(c["ua"]).lower(), [ipaddress.ip_network(x, strict=False) for x in c.get("cidrs", [])]) |
| for c in self.d.get("verified_crawlers", [])] |
|
|
| @classmethod |
| def load(cls, path: str | None) -> "Policy": |
| if path and os.path.exists(path): |
| with open(path, encoding="utf-8") as f: |
| return cls(json.load(f)) |
| return cls() |
|
|
| def validate(self) -> None: |
| d = self.d |
| if d["mode"] not in ("report", "enforce"): |
| raise ValueError(f"policy.mode must be report|enforce, got {d['mode']!r}") |
| if not 0 < float(d["alpha"]) < 0.5: |
| raise ValueError("policy.alpha must be in (0, 0.5)") |
| for fam, act in d["actions"].items(): |
| if act not in ACTIONS: |
| raise ValueError(f"policy.actions[{fam!r}] must be one of {ACTIONS}, got {act!r}") |
| for k in ("block_ttl_s", "rate_limit_ttl_s", "challenge_ttl_s", "min_action_score"): |
| if float(d[k]) < 0: |
| raise ValueError(f"policy.{k} must be >= 0") |
| for a in d["allow"]: |
| ipaddress.ip_network(a, strict=False) |
| for c in d.get("verified_crawlers", []): |
| if not c.get("ua") or not c.get("cidrs"): |
| raise ValueError("policy.verified_crawlers entries need 'ua' and 'cidrs' (both must match)") |
| for x in c["cidrs"]: |
| ipaddress.ip_network(x, strict=False) |
| if d.get("machine_challenge", "rate_limit") not in ACTIONS: |
| raise ValueError(f"policy.machine_challenge must be one of {ACTIONS}") |
|
|
| |
| def allowed(self, ip: str) -> bool: |
| try: |
| addr = ipaddress.ip_address(ip) |
| except ValueError: |
| return False |
| return any(addr in net for net in self._allow) |
|
|
| def verified_crawler(self, ip: str, ua: str) -> bool: |
| """the user-agent names a listed crawler AND the address is in that crawler's published ranges""" |
| ua = ua.lower() |
| try: |
| addr = ipaddress.ip_address(str(ip).split("#", 1)[0]) |
| except ValueError: |
| return False |
| return any(tok in ua and any(addr in n for n in nets) for tok, nets in self._crawlers) |
|
|
| def action_for(self, family: str, score: int, ip: str | None = None) -> str: |
| """what the policy lets the layer do about one verdict (mode 'report' still returns the would-be action)""" |
| if ip is not None and self.allowed(ip): |
| return "monitor" |
| if score < int(self.d["min_action_score"]): |
| return "monitor" |
| return self.d["actions"].get(family, self.d["actions"].get("*", "monitor")) |
|
|
| def ttl(self, action: str) -> float: |
| return float(self.d.get(f"{action}_ttl_s", 0.0)) |
|
|
| @property |
| def enforcing(self) -> bool: |
| return self.d["mode"] == "enforce" |
|
|
| def dump(self) -> str: |
| return json.dumps(self.d, ensure_ascii=False, indent=1) |
|
|
