feat: project scaffold, unified Pydantic schema, scope gating

recon-triage/src/recon_triage/scope.py

@@ -0,0 +1,129 @@
+"""Scope enforcement — mandatory gate. Anything not explicitly in scope is dropped.
+
+A ``scope.yaml`` declares in-scope domains and CIDRs plus optional out-of-scope
+exclusions. The tool refuses to touch anything not in scope: out-of-scope input is
+dropped with a logged warning, never scanned.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import yaml
+
+from .schema import ScopeUsed
+
+log = logging.getLogger("recon_triage.scope")
+
+
+def _norm_domain(d: str) -> str:
+    return d.strip().lower().rstrip(".")
+
+
+def _is_ip(value: str) -> bool:
+    try:
+        ipaddress.ip_address(value)
+        return True
+    except ValueError:
+        return False
+
+
+@dataclass
+class Scope:
+    """Loaded scope rules with in/out gating.
+
+    Domain matching includes subdomains (``example.com`` matches ``api.example.com``).
+    Out-of-scope rules take precedence over in-scope rules.
+    """
+
+    domains: list[str] = field(default_factory=list)
+    cidrs: list[str] = field(default_factory=list)
+    out_of_scope: list[str] = field(default_factory=list)
+
+    @classmethod
+    def load(cls, path: str | Path) -> Scope:
+        data = yaml.safe_load(Path(path).read_text()) or {}
+        return cls.from_dict(data)
+
+    @classmethod
+    def from_dict(cls, data: dict) -> Scope:
+        domains = [_norm_domain(d) for d in (data.get("in_scope_domains") or []) if d]
+        cidrs = [str(c).strip() for c in (data.get("in_scope_cidrs") or []) if c]
+        out = [str(o).strip().lower().rstrip(".") for o in (data.get("out_of_scope") or []) if o]
+        # Validate CIDRs early; drop invalid with a warning rather than crashing.
+        valid_cidrs = []
+        for c in cidrs:
+            try:
+                ipaddress.ip_network(c, strict=False)
+                valid_cidrs.append(c)
+            except ValueError:
+                log.warning("Ignoring invalid CIDR in scope: %s", c)
+        return cls(domains=domains, cidrs=valid_cidrs, out_of_scope=out)
+
+    # -- matching -----------------------------------------------------------
+
+    def _domain_in(self, host: str, rules: list[str]) -> bool:
+        host = _norm_domain(host)
+        for rule in rules:
+            if host == rule or host.endswith("." + rule):
+                return True
+        return False
+
+    def _ip_in_cidrs(self, ip: str) -> bool:
+        try:
+            addr = ipaddress.ip_address(ip)
+        except ValueError:
+            return False
+        for c in self.cidrs:
+            try:
+                if addr in ipaddress.ip_network(c, strict=False):
+                    return True
+            except ValueError:
+                continue
+        return False
+
+    def _ip_out(self, ip: str) -> bool:
+        for rule in self.out_of_scope:
+            if rule == ip:
+                return True
+            # out_of_scope may also carry CIDRs
+            try:
+                if ipaddress.ip_address(ip) in ipaddress.ip_network(rule, strict=False):
+                    return True
+            except ValueError:
+                continue
+        return False
+
+    def is_in_scope(self, value: str) -> bool:
+        """True iff ``value`` (a hostname or IP) is in scope and not excluded."""
+        value = value.strip()
+        if not value:
+            return False
+        if _is_ip(value):
+            if self._ip_out(value):
+                return False
+            return self._ip_in_cidrs(value)
+        # hostname
+        if self._domain_in(value, self.out_of_scope):
+            return False
+        return self._domain_in(value, self.domains)
+
+    def filter(self, values: list[str]) -> list[str]:
+        """Return only in-scope values; log a warning for each dropped one."""
+        kept: list[str] = []
+        for v in values:
+            if self.is_in_scope(v):
+                kept.append(v)
+            else:
+                log.warning("Dropping out-of-scope target: %s", v)
+        return kept
+
+    def as_used(self) -> ScopeUsed:
+        return ScopeUsed(
+            in_scope_domains=list(self.domains),
+            in_scope_cidrs=list(self.cidrs),
+            out_of_scope=list(self.out_of_scope),
+        )