"""Engagement scope and owner authorization for the pentest agent. This module owns the two pieces of state a penetration test needs and the current policy language cannot express: * :class:`TargetScope` — the whitelist of hosts/CIDRs a test may touch, and * :class:`EngagementAuthorization` — the artifact a human signed off on. Why this is code and not a policy rule -------------------------------------- The policy DSL evaluates a boolean expression over an AST allowlist (``security_plane/policy/__init__.py``); a condition has to be a bool no longer than 256 characters and may only call ``is_sensitive``. "Is this address inside the approved range" is a function call returning a bool, so it *looks* expressible — but writing it would mean either giving up that allowlist or smuggling the range into a string. Both are worse than a typed object with its own tests, so the whitelist is a value with a real parser, and it is consulted on **every** tool invocation rather than once per session. Scope decisions that are security properties, not conveniences ------------------------------------------------------------- * An empty scope authorises nothing. ``TargetScope`` with no networks and no hosts refuses every target; there is no "no scope configured means allow". * A target may only be a single host, a ``host:port`` pair, or an ``http(s)`` URL. Credentials in the authority (``http://user:pw@host``), control characters, whitespace and shell metacharacters are refused outright, because the target text ends up in a process argument vector. * A CIDR mask **in a tool target** is accepted only when the resulting network is a subnet of an authorised network. Checking only the address would let ``10.0.0.0/8`` through a scope of ``10.0.0.0/24``, which is the obvious way to bypass a whitelist and the reason the mask is validated, not stripped. * Host *names* are accepted only when the name itself was listed in the engagement. Names are not resolved during authorization: resolving here would make the check depend on DNS the caller controls (rebinding between the check and the run), so a name is either explicitly authorised or refused. """ from __future__ import annotations import hashlib import ipaddress import json import re import time from collections.abc import Iterable from dataclasses import dataclass from typing import Any, Final from urllib.parse import urlsplit from cyberbrain26.security_plane.permission import SecurityError __all__ = [ "MAX_ENGAGEMENT_LEVEL", "TOOL_CLASSES", "TOOL_CLASS_BY_LEVEL", "EngagementAuthorization", "TargetScope", "TargetScopeError", ] # Highest engagement level the catalog knows about. Levels are integers on # purpose: they order, so "level <= max_level" is the whole gating rule. MAX_ENGAGEMENT_LEVEL: Final[int] = 3 # Tool classes an engagement may authorise, one per level. Named in code because # the value ends up in the signed artifact and in the audit trail; it is not a # free-form string the caller supplies. TOOL_CLASSES: Final[tuple[str, ...]] = ("passive", "recon", "vulnerability", "exploitation") # Engagement level -> the tool class that level belongs to. Declared here, next # to ``allowed_tool_classes``, so an engagement and the catalog that enforces it # cannot disagree about which class a number means. TOOL_CLASS_BY_LEVEL: Final[dict[int, str]] = { 0: "passive", 1: "recon", 2: "vulnerability", 3: "exploitation", } _ALLOWED_URL_SCHEMES: Final[frozenset[str]] = frozenset({"http", "https"}) _HOSTNAME_RE: Final[re.Pattern[str]] = re.compile(r"^[a-z0-9]([a-z0-9._-]{0,251}[a-z0-9])?$") _MAX_TARGET_LENGTH: Final[int] = 255 _HEX_ONLY_RE: Final[re.Pattern[str]] = re.compile(r"^0[xX][0-9a-fA-F]+$") class TargetScopeError(SecurityError): """A target is unusable or lies outside the authorised engagement scope. Inherits from ``SecurityError`` so a caller can refuse on the same signal the rest of the security plane uses; every instance is a refusal, never a recoverable error. """ def _is_ip_literal(host: str) -> bool: try: ipaddress.ip_address(host) except ValueError: return False return True def _host_from_netloc(netloc: str) -> str: """Return the bare host of an authority component, refusing anything odd.""" value = netloc.strip() if value.startswith("["): end = value.find("]") if end == -1: raise TargetScopeError(f"malformed IPv6 literal in target: {netloc!r}") host = value[1:end] remainder = value[end + 1 :] if remainder and not remainder.startswith(":"): raise TargetScopeError(f"malformed IPv6 literal in target: {netloc!r}") elif value.count(":") == 1: host, _, port = value.partition(":") if not port.isdigit() or not 0 < int(port) < 65536: raise TargetScopeError(f"target port is not a valid TCP/UDP port: {port!r}") else: # Either a bare host, or a bare IPv6 literal, or a malformed authority. host = value host = host.strip().lower() if not host: raise TargetScopeError("target carries no host part") if not _is_ip_literal(host) and not _HOSTNAME_RE.match(host): raise TargetScopeError(f"target host {host!r} is neither an IP literal nor a host name") return host def _prefix_of(target: str) -> int | None: """Return the CIDR prefix length a target declares, or ``None``. A ``/`` is only meaningful on the network part of a literal address. A URL path (``http://host/x``) is not a mask, so the split happens on the raw text only when the text is not a URL. A bracketed IPv6 authority (``[2001:db8::1]:443``) is not a mask either, and returns ``None``. Anything after the closing bracket other than a port is refused outright rather than ignored: dropping the tail would silently turn ``[2001:db8::1]/64`` into a bare address, and a single-address engagement would then authorise 2**64 addresses to a tool that reads the same text as ``addr/prefix``. Refusing keeps "one string, one parser". """ if "://" in target: return None if "[" in target: closing = target.find("]") if closing == -1: raise TargetScopeError(f"unbalanced bracket in target: {target!r}") tail = target[closing + 1 :] if not (tail == "" or (tail.startswith(":") and tail[1:].isdigit())): raise TargetScopeError( f"target must not carry anything after a bracketed IPv6 literal: {target!r}" ) return None if "/" not in target: return None address_part, _, prefix_text = target.partition("/") if not address_part or not prefix_text.isdigit(): raise TargetScopeError(f"malformed CIDR in target: {target!r}") prefix = int(prefix_text) if not 0 <= prefix <= 128: raise TargetScopeError(f"CIDR prefix out of range in target: {target!r}") return prefix def extract_target_host(target: object) -> str: """Return the host of ``target``, refusing anything that is not a plain host. Accepts ``host``, ``host:port``, ``[v6]:port``, ``10.0.0.0/24`` and ``http(s)://host[:port]/path``. Everything else — a non-string, an empty string, embedded whitespace or control characters, an unsupported URL scheme, credentials in the authority — is a :class:`TargetScopeError`, because each of those is a way to make the text in the argument vector mean something other than the address the reviewer approved. """ if not isinstance(target, str): raise TargetScopeError("target must be a string") text = target.strip() if not text: raise TargetScopeError("target must not be empty") if len(text) > _MAX_TARGET_LENGTH: raise TargetScopeError(f"target is longer than {_MAX_TARGET_LENGTH} characters") if any(character.isspace() or ord(character) < 0x20 for character in text): raise TargetScopeError("target must not contain whitespace or control characters") if "://" in text: # urlsplit raises a bare ValueError from CPython's _check_bracketed_host # on an unbalanced bracket ("http://[", "http://[::1]x"). One character # from the model would otherwise escape this module as an exception that # is NOT a TargetScopeError, and the caller audits only TargetScopeError # - so the refusal would never reach the audit trail. Convert it here, # at the boundary, so it can only ever arrive as a refusal. try: parts = urlsplit(text) except ValueError as exc: raise TargetScopeError(f"target is not a well-formed URL: {text!r}") from exc if parts.scheme.lower() not in _ALLOWED_URL_SCHEMES: raise TargetScopeError( f"target URL scheme {parts.scheme!r} is not allowed " f"(allowed: {sorted(_ALLOWED_URL_SCHEMES)})" ) if "@" in parts.netloc: raise TargetScopeError("target must not carry credentials in its authority") return _reject_numeric_host(_host_from_netloc(parts.netloc)) # Drop a CIDR mask before validating the host part, but keep the mask for the # containment check performed by TargetScope.allows. address_part = text.partition("/")[0] return _reject_numeric_host(_host_from_netloc(address_part)) def _reject_numeric_host(host: str) -> str: """Refuse a host part that is only digits or only hex digits. ``2130706433`` and ``0x7f000001`` are not host names and are not :func:`ipaddress.ip_address` literals either, so they survive every other check here and then reach the tool, which resolves them numerically - to loopback, in the case of the two examples above. Refusing them at the boundary keeps "the string we checked" and "the string the tool reads" the same string, instead of relying on the tool to agree with us about syntax. """ if host and (host.isdigit() or _HEX_ONLY_RE.fullmatch(host)): raise TargetScopeError( f"target host must not be a bare numeric address: {host!r}" ) return host @dataclass(frozen=True) class TargetScope: """Whitelist of networks and hosts a pentest tool call may address. Checked on every invocation from :func:`authorize_tool_call` in ``agents.pentest_catalog``; never once per session. Construct it from the signed engagement artifact with :meth:`from_targets` rather than by hand, so the whitelist and the artifact cannot disagree. Attributes: networks: Parsed CIDR networks. Empty means "no network target". hosts: Individually authorised hosts (literal IPs and names). An IP listed here does not imply its /24; only ``networks`` does. """ networks: tuple[Any, ...] = () hosts: frozenset[str] = frozenset() def __post_init__(self) -> None: parsed: list[Any] = [] for raw in self.networks: text = str(raw).strip() if not text: raise TargetScopeError("engagement scope contains an empty network entry") try: network = ipaddress.ip_network(text, strict=False) except ValueError as exc: raise TargetScopeError(f"invalid CIDR in engagement scope: {text!r}") from exc # A default route as an *authorised* scope is never a real engagement # - it is the entire internet, or the entire v6 internet - and it is # the one value a mistyped artifact would silently widen to. It is # refused at construction, before any prefix floor, so there is no # configuration of TargetScope that carries it. if network.prefixlen == 0: raise TargetScopeError( f"engagement scope must not be a default route: {text!r}" ) parsed.append(network) object.__setattr__(self, "networks", tuple(parsed)) object.__setattr__( self, "hosts", frozenset(str(host).strip().lower() for host in self.hosts) ) @classmethod def from_targets(cls, targets: Iterable[object]) -> TargetScope: """Build a scope from the target list an engagement was signed off on. A bare literal IP becomes a ``/32`` host entry rather than a network, so listing one address never widens into its surrounding prefix. """ networks: list[str] = [] hosts: list[str] = [] for raw in targets: text = str(raw).strip() if not text: raise TargetScopeError("engagement target must not be empty") host = extract_target_host(text) prefix = _prefix_of(text) if prefix is None: hosts.append(host) continue try: network = ipaddress.ip_network(f"{host}/{prefix}", strict=False) except ValueError as exc: raise TargetScopeError(f"invalid CIDR in engagement target: {text!r}") from exc if network.num_addresses == 1: hosts.append(host) else: networks.append(f"{network.network_address}/{network.prefixlen}") if not networks and not hosts: raise TargetScopeError("engagement scope is empty; it authorises no target") return cls(networks=tuple(networks), hosts=frozenset(hosts)) def __bool__(self) -> bool: """Report whether this scope authorises anything at all.""" return bool(self.networks or self.hosts) def allows_host(self, host: str) -> bool: """Report whether a bare host (no port, no mask) is authorised.""" candidate = host.strip().lower() if candidate in self.hosts: return True try: address = ipaddress.ip_address(candidate) except ValueError: # A name is only authorised when it was named in the engagement. return False return self._network_contains(address) def allows(self, target: object) -> bool: """Report whether ``target`` is inside this scope. Never raises: an unparseable target is simply not authorised, and the caller records the refusal. A target carrying a CIDR mask is authorised only when the whole network is a subnet of an authorised one. """ try: host = extract_target_host(target) prefix = _prefix_of(target) if isinstance(target, str) else None except TargetScopeError: return False if prefix is None: return self.allows_host(host) try: network = ipaddress.ip_network(f"{host}/{prefix}", strict=False) except ValueError: return False return self._network_contains(network) def _network_contains( self, candidate: ipaddress.IPv4Address | ipaddress.IPv6Address | ipaddress.IPv4Network | ipaddress.IPv6Network, ) -> bool: for network in self.networks: if candidate.version != network.version: continue if isinstance(candidate, (ipaddress.IPv4Address, ipaddress.IPv6Address)): if candidate in network: return True elif isinstance(candidate, (ipaddress.IPv4Network, ipaddress.IPv6Network)): if candidate == network or candidate.subnet_of(network): return True else: # Unexpected type: fail closed, do not raise. return False return False @dataclass(frozen=True) class EngagementAuthorization: """The artifact a human authorised a pentest engagement under. Everything a reviewer needs in order to say yes or no is a field here, and :attr:`digest` is a stable fingerprint of all of it. The digest is what the operator re-states on the command line and what the audit trail records, so a run cannot quietly happen under different terms than the ones that were shown. Attributes: owner: Human or system accountable for the engagement. targets: The authorised target list; the scope is derived from it. max_level: Highest engagement level permitted (0-3). allowed_tool_classes: Tool classes this engagement may use. Must include the class belonging to ``max_level``. window_start: POSIX timestamp the engagement opens, or ``None``. window_end: POSIX timestamp the engagement closes, or ``None``. intensity: Free-form operator label recorded in the artifact. rules_of_engagement: Human-readable constraints, recorded verbatim. """ owner: str targets: tuple[str, ...] max_level: int = 1 allowed_tool_classes: frozenset[str] = frozenset({"recon"}) window_start: float | None = None window_end: float | None = None intensity: str = "measured" rules_of_engagement: tuple[str, ...] = () def __post_init__(self) -> None: owner = str(self.owner).strip() if not owner: raise TargetScopeError("engagement owner must not be empty") object.__setattr__(self, "owner", owner) targets = tuple(str(target).strip() for target in self.targets) if not targets or any(not target for target in targets): raise TargetScopeError("engagement must name at least one non-empty target") for target in targets: # Parse now so a malformed target is refused when the artifact is # built, not on the first tool call half an engagement later. extract_target_host(target) object.__setattr__(self, "targets", targets) if isinstance(self.max_level, bool) or not isinstance(self.max_level, int): raise TargetScopeError(f"engagement max_level must be an integer, got {self.max_level!r}") if not 0 <= self.max_level <= MAX_ENGAGEMENT_LEVEL: raise TargetScopeError( f"engagement max_level must be between 0 and {MAX_ENGAGEMENT_LEVEL}, " f"got {self.max_level}" ) classes = frozenset(str(name).strip().lower() for name in self.allowed_tool_classes) unknown = classes - set(TOOL_CLASSES) if unknown: raise TargetScopeError( f"unknown engagement tool class(es): {sorted(unknown)} " f"(known: {list(TOOL_CLASSES)})" ) if not classes: raise TargetScopeError("engagement must allow at least one tool class") object.__setattr__(self, "allowed_tool_classes", classes) for name in ("window_start", "window_end"): value = getattr(self, name) if value is not None and (isinstance(value, bool) or not isinstance(value, (int, float))): raise TargetScopeError(f"engagement {name} must be a POSIX timestamp or None") if ( self.window_start is not None and self.window_end is not None and self.window_start >= self.window_end ): raise TargetScopeError("engagement window_start must be before window_end") object.__setattr__(self, "intensity", str(self.intensity).strip() or "measured") object.__setattr__( self, "rules_of_engagement", tuple(str(rule) for rule in self.rules_of_engagement), ) @property def scope(self) -> TargetScope: """Return the whitelist derived from :attr:`targets`.""" return TargetScope.from_targets(self.targets) def canonical(self) -> dict[str, Any]: """Return the artifact as a JSON-compatible mapping, order-independent.""" return { "owner": self.owner, "targets": list(self.targets), "max_level": self.max_level, "allowed_tool_classes": sorted(self.allowed_tool_classes), "window_start": self.window_start, "window_end": self.window_end, "intensity": self.intensity, "rules_of_engagement": list(self.rules_of_engagement), } @property def digest(self) -> str: """Return a stable SHA-256 fingerprint of the whole artifact. Stable means: two artifacts with the same terms produce the same digest, and any change to any term — including the rules of engagement text — produces a different one. It is a fingerprint, not a signature: it proves the terms did not change, not who agreed to them. """ material = json.dumps( self.canonical(), sort_keys=True, separators=(",", ":"), default=str ) return hashlib.sha256(material.encode("utf-8")).hexdigest() def is_active(self, now: float | None = None) -> bool: """Report whether the engagement's time window is open.""" moment = time.time() if now is None else now if self.window_start is not None and moment < self.window_start: return False if self.window_end is not None and moment > self.window_end: return False return True def allows_level(self, level: object) -> bool: """Report whether ``level`` is at or below the engagement's ceiling.""" if isinstance(level, bool) or not isinstance(level, int): return False return 0 <= level <= self.max_level def allows_tool_class(self, tool_class: str) -> bool: """Report whether the engagement permits a tool class.""" return tool_class.strip().lower() in self.allowed_tool_classes def engages_level(self, level: object) -> bool: """Report whether one tool invocation at ``level`` is permitted. Both halves have to hold: the level is at or below ``max_level`` *and* the tool class that level belongs to is in ``allowed_tool_classes``. Checking only the number would let an engagement that set ``max_level=3`` silently inherit exploitation even when it named only recon. ``level`` is narrowed rather than coerced. ``int("2")``, ``int(2.0)`` and ``int(True)`` all produce an int, so a blind conversion would turn a malformed declaration into an authorised one and would let anything with ``__int__`` decide its own level. A value that is not already a plain int - a bool included, because ``isinstance(True, int)`` is True - is refused here, which is the same rule :meth:`allows_level` applies. """ if isinstance(level, bool) or not isinstance(level, int): return False if not self.allows_level(level): return False tool_class = TOOL_CLASS_BY_LEVEL.get(int(level)) if tool_class is None: return False return self.allows_tool_class(tool_class)