"""Pentest tool catalog: the boundary that turns a declared intent into an argv. The model never supplies an argument vector. It declares an *intent* (``{tool, target, args, level}``) and this module reconstructs the command line from the catalog, refusing anything the catalog does not already permit. That inversion is the whole point: * an allowlist of arguments the tool may carry, checked token by token; * a forbidden list and a forbidden-prefix list, checked **before** the allowlist, so no argument can be both; * a check on every invocation that the target lies inside the engagement's :class:`~cyberbrain26.agents.engagement.TargetScope`; * a check that the engagement artifact permits this tool at this level. Fail-closed everywhere. There is no fallback to a raw shell, no "unknown tool means run it anyway", and no path where a missing catalog, a missing scope or a missing engagement artifact produces a run instead of a refusal. Every refusal is written to the audit log. What the catalog deliberately does not model -------------------------------------------- * **No NSE scripts** for nmap (``--script``), no ``-e``, no ``-oN``/``-oG``: scripts execute code, output files escape the sandbox, and ``-e`` publishes the operator's MAC on the wire. * **No shells anywhere.** sqlmap's ``--os-shell``/``--os-cmd``/``--sql-shell`` and ``--exec`` are permanently forbidden, at every level, including L3: they hand a shell to a remote host rather than testing it. * **No output-to-file and no redirect options.** Anything that writes outside the one writable workspace bind is refused; ``-o``-family options are prefixes in the forbidden list for that reason. * **No redirect following** (curl ``-L``/``--location``, httpx ``-fr``): a redirect can leave the authorised target without any further check. * **No second target.** Every option that can itself carry a target (``-u``, ``-h``, ``-iL``, ``-t``, ``-x``, ``-l``…) is in the forbidden list, so the only target in the process argument vector is the one that was checked against the scope. * **No credentials on the command line** (hydra ``-p``, curl ``-u user:pw``). Password material must come from a file inside the workspace, which keeps it out of the audit trail. * Only unprivileged scan modes are permitted (nmap ``-sT``/``-sn``, not ``-sS``/``-sU``): a raw-socket scan needs privileges this system does not grant, and asking for it would mean asking for a weaker sandbox. Argv values are passed to the sandbox as a vector. They are still validated: a value is only accepted when it matches the narrow rule declared for its option, because the same token is what a reviewer reads in the audit trail. """ from __future__ import annotations import re import shlex from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from pathlib import Path from typing import Any, Final from cyberbrain26.agents.engagement import ( TOOL_CLASS_BY_LEVEL, EngagementAuthorization, TargetScope, ) from cyberbrain26.security_plane.audit import AuditEventType, AuditLog from cyberbrain26.security_plane.permission import SecurityError __all__ = [ "ACTION_BY_LEVEL", "CATALOG", "MAX_ARGUMENTS", "AuthorizedToolCall", "PentestRefusal", "PentestTool", "ValueRule", "action_for_level", "authorize_tool_call", "catalog_entry", "describe_catalog", "tool_class_for_level", ] # Engagement level -> the security-plane action a tool at that level performs. # The action, not the level, is what the permission gateway, policy engine and # approval flow understand, so the level is translated here and nowhere else. ACTION_BY_LEVEL: Final[dict[int, str]] = { 0: "read_file", 1: "scan_network", 2: "vuln_scan", 3: "exploit_target", } MAX_ARGUMENTS: Final[int] = 24 MAX_ARGUMENT_LENGTH: Final[int] = 200 _TARGET_KINDS: Final[frozenset[str]] = frozenset({"network", "workspace"}) class PentestRefusal(SecurityError): """A tool call was refused by the catalog. Attributes: reason_code: Stable machine-readable label recorded in the audit log, so a refusal can be counted and alerted on without parsing prose. """ def __init__(self, reason_code: str, message: str) -> None: super().__init__(message) self.reason_code = reason_code @dataclass(frozen=True) class ValueRule: """The narrow rule one option's value has to satisfy. Options are always written in attached form (``-p80``, ``--level3``): a separate value word would make "which token is the value" ambiguous, and an ambiguous boundary is an argument-injection boundary. Attributes: kind: Which validator applies (see :func:`_validate_value`). choices: Permitted values for the ``enum``/``enum_csv`` kinds. minimum: Lower bound for the numeric kinds. maximum: Upper bound for the numeric kinds. """ kind: str choices: tuple[str, ...] = () minimum: int = 0 maximum: int = 0 _PORT_SPEC_RE: Final[re.Pattern[str]] = re.compile(r"^[0-9]{1,5}(-[0-9]{1,5})?(,[0-9]{1,5}(-[0-9]{1,5})?){0,31}$") _IDENTIFIER_RE: Final[re.Pattern[str]] = re.compile(r"^[A-Za-z0-9._@-]{1,64}$") _TEXT_RE: Final[re.Pattern[str]] = re.compile(r"^[A-Za-z0-9_%=&+.:,/~@ -]{1,200}$") _WRITE_OUT_RE: Final[re.Pattern[str]] = re.compile(r'^[A-Za-z0-9_.\-%{}":, ]{1,120}$') _HEX_FLAGS_RE: Final[re.Pattern[str]] = re.compile(r"^[0-9a-fx]{1,16}$") _INT_RE: Final[re.Pattern[str]] = re.compile(r"^[0-9]{1,5}$") _FLOAT_RE: Final[re.Pattern[str]] = re.compile(r"^[0-9]{1,3}(\.[0-9]{1,2})?$") def _validate_value(rule: ValueRule, value: str, *, workspace: Path | None) -> str | None: """Return the in-sandbox path for a workspace option, or ``None``. Raises: PentestRefusal: If the value does not satisfy ``rule``. """ kind = rule.kind if kind == "enum": if value not in rule.choices: raise PentestRefusal( "value_not_allowed", f"value {value!r} is not one of {list(rule.choices)}", ) elif kind == "enum_csv": parts = value.split(",") if not parts or any(part not in rule.choices for part in parts): raise PentestRefusal( "value_not_allowed", f"value {value!r} is not a comma-separated list of {list(rule.choices)}", ) elif kind == "int_range": if not _INT_RE.match(value): raise PentestRefusal("value_malformed", f"value {value!r} is not an integer") # Two separate locals, one per branch: the int rule compares an integer # against the range and the float rule compares a float against it. They # share nothing but the range check itself, and merging them would let # the int rule accept "5.0" - a token the reviewer reads as different # from "5" and a tool that may not treat them the same way. int_number = int(value) if not rule.minimum <= int_number <= rule.maximum: raise PentestRefusal( "value_out_of_range", f"value {int_number} is outside the permitted range " f"{rule.minimum}-{rule.maximum}", ) elif kind == "float_range": if not _FLOAT_RE.match(value): raise PentestRefusal("value_malformed", f"value {value!r} is not a number") float_number = float(value) if not rule.minimum <= float_number <= rule.maximum: raise PentestRefusal( "value_out_of_range", f"value {float_number} is outside the permitted range " f"{rule.minimum}-{rule.maximum}", ) elif kind == "port_spec": if not _PORT_SPEC_RE.match(value): raise PentestRefusal("value_malformed", f"{value!r} is not a port specification") for part in value.split(","): bounds = part.split("-") for bound in bounds: if not 0 < int(bound) < 65536: raise PentestRefusal( "value_out_of_range", f"port {bound} is outside 1-65535" ) if len(bounds) == 2 and int(bounds[0]) > int(bounds[1]): raise PentestRefusal("value_malformed", f"port range {part!r} is inverted") elif kind == "identifier": if not _IDENTIFIER_RE.match(value): raise PentestRefusal( "value_malformed", f"{value!r} is not a plain identifier" ) elif kind == "text": if not _TEXT_RE.match(value): raise PentestRefusal( "value_malformed", f"{value!r} contains characters this option does not accept" ) elif kind == "write_out": if not _WRITE_OUT_RE.match(value): raise PentestRefusal( "value_malformed", f"{value!r} is not a permitted curl write-out format" ) elif kind == "hex_flags": if not _HEX_FLAGS_RE.match(value): raise PentestRefusal( "value_malformed", f"{value!r} is not a set of hex flag characters" ) elif kind == "workspace_path": if workspace is None: raise PentestRefusal( "no_workspace", "a workspace is required for this option but none was configured", ) return _in_sandbox_path(value, workspace) else: # An unknown rule kind would silently validate everything, so it denies. raise PentestRefusal("unknown_value_rule", f"no validator is defined for {kind!r}") return None def _in_sandbox_path(value: str, workspace: Path) -> str: """Validate a workspace-relative path and return its in-sandbox form. The sandbox bind-mounts the workspace at ``/workspace``, so the path that has to appear in the argv is the in-sandbox one. The host path is still resolved for the containment check, which is what refuses a symlink that points out of the workspace. Raises: PentestRefusal: If the value is absolute, escapes the workspace, or is not a plain relative path. """ if not value: raise PentestRefusal("path_not_allowed", "empty path") if value.startswith(("/", "~")): raise PentestRefusal( "absolute_path_not_allowed", f"absolute path {value!r} is not allowed; use a path inside the workspace", ) if "\\" in value or "\x00" in value: raise PentestRefusal("path_not_allowed", f"path {value!r} is not a POSIX relative path") parts = value.split("/") if any(part in {"", ".", ".."} for part in parts): raise PentestRefusal( "path_escape_not_allowed", f"path {value!r} contains an empty or traversing segment", ) relative = Path(*parts) root = workspace.resolve() resolved = (root / relative).resolve() if resolved != root and not resolved.is_relative_to(root): raise PentestRefusal( "path_escape_not_allowed", f"path {value!r} resolves to {resolved} which is outside the workspace {root}", ) return f"/workspace/{relative.as_posix()}" @dataclass(frozen=True) class PentestTool: """One catalog entry: what a tool is, and exactly how it may be invoked. Attributes: name: Catalog key a declared intent refers to. binary: Absolute path of the executable inside the sandbox. level: The engagement level at which this tool becomes usable. timeout_seconds: Wall-clock ceiling for one invocation. max_output_bytes: Captured-output ceiling the caller should apply. target_kind: ``"network"`` targets are scope-checked against the engagement; ``"workspace"`` targets must resolve inside the workspace and reach no network at all. target_option: Option the target is attached to, or ``None`` to place it positionally. The builder emits it; a caller may not. trailing_tokens: Literal tokens appended after a positional target. fixed_flags: Flags the builder always emits, declared in code. flags: Value-less tokens a caller may pass. value_options: Option prefix -> value rule, in attached form. forbidden: Tokens a caller may never pass. forbidden_prefixes: Token prefixes a caller may never start a token with. Checked before the allowlist, so the two can never both win. level_gated_flags: Level -> extra value-less flags permitted at or above that level. This is how one tool serves two levels: ``sqlmap`` is restricted at L2 and gains ``--dump`` only at L3. """ name: str binary: str level: int timeout_seconds: int max_output_bytes: int target_kind: str target_option: str | None = None trailing_tokens: tuple[str, ...] = () fixed_flags: tuple[str, ...] = () flags: frozenset[str] = frozenset() value_options: Mapping[str, ValueRule] = field(default_factory=dict) forbidden: frozenset[str] = frozenset() forbidden_prefixes: tuple[str, ...] = () level_gated_flags: Mapping[int, frozenset[str]] = field(default_factory=dict) level_gated_options: Mapping[int, Mapping[str, ValueRule]] = field(default_factory=dict) def __post_init__(self) -> None: """Refuse a catalog entry whose own rules contradict each other. An inconsistency here is not cosmetic: a flag that is also a forbidden prefix, or an option that a forbidden prefix swallows, means the allowlist and the deny list disagree and one of them silently loses. That has to be a loud failure at import time, not a surprising permission at run time. """ if self.target_kind not in _TARGET_KINDS: raise ValueError(f"{self.name}: target_kind must be one of {sorted(_TARGET_KINDS)}") if self.timeout_seconds <= 0 or self.max_output_bytes <= 0: raise ValueError(f"{self.name}: timeout and output ceiling must be positive") if not 0 <= self.level <= 3: raise ValueError(f"{self.name}: level must be between 0 and 3") if "" in self.forbidden_prefixes: raise ValueError(f"{self.name}: an empty forbidden prefix would deny everything") if self.flags & self.forbidden: raise ValueError(f"{self.name}: flags and forbidden tokens overlap") if set(self.fixed_flags) & self.forbidden: raise ValueError(f"{self.name}: fixed flags and forbidden tokens overlap") if set(self.fixed_flags) & self.flags: raise ValueError(f"{self.name}: fixed flags and caller flags overlap") for gated_level, gated_flags in self.level_gated_flags.items(): if gated_flags & self.forbidden: raise ValueError(f"{self.name}: level-gated flags overlap the forbidden set") if gated_flags & self.flags: raise ValueError(f"{self.name}: level-gated flags overlap the plain flags") if gated_flags & set(self.fixed_flags): raise ValueError(f"{self.name}: level-gated flags overlap the fixed flags") if gated_level <= self.level: raise ValueError( f"{self.name}: level-gated flags at level {gated_level} would be " f"available at the tool's base level {self.level}" ) permitted = set(self.value_options) | set(self.flags) | set(self.fixed_flags) for gated_options in self.level_gated_options.values(): permitted |= set(gated_options) exact = set(self.forbidden) overlap = permitted & exact if overlap: raise ValueError(f"{self.name}: permitted and forbidden tokens overlap: {sorted(overlap)}") for prefix in self.forbidden_prefixes: if prefix in exact or prefix in permitted: raise ValueError( f"{self.name}: forbidden prefix {prefix!r} is also a permitted token" ) shadowed = sorted(token for token in permitted if token.startswith(prefix)) if shadowed: raise ValueError( f"{self.name}: forbidden prefix {prefix!r} would swallow the permitted " f"token(s) {shadowed}" ) # --------------------------------------------------------------------------- # The catalog # --------------------------------------------------------------------------- # Per-tool notes are in the module docstring; what matters here is that every # target-bearing, output-bearing, redirect-following and shell-bearing option is # absent from the allowlists below. _WORKSPACE_PATH = ValueRule(kind="workspace_path") _PORT_SPEC = ValueRule(kind="port_spec") CATALOG: Final[dict[str, PentestTool]] = { # L0 — passive. No network namespace is involved at all: the only tool is a # file-type probe over a path inside the one writable workspace. "file": PentestTool( name="file", binary="/usr/bin/file", level=0, timeout_seconds=15, max_output_bytes=65_536, target_kind="workspace", flags=frozenset({"-b", "--brief"}), forbidden_prefixes=( "--compile", "--uncompress", "-f", "-P", "--preserve-date", "-L", "--dereference", "-w", ), ), # L1 — recon. "nmap": PentestTool( name="nmap", binary="/usr/bin/nmap", level=1, timeout_seconds=180, max_output_bytes=1_000_000, target_kind="network", flags=frozenset({"-sn", "-sT", "-Pn", "-n", "-sV", "--open", "-v"}), value_options={ "-p": _PORT_SPEC, "-T": ValueRule(kind="enum", choices=("0", "1", "2", "3", "4")), }, forbidden_prefixes=( "-oN", "-oG", "-oA", "-oX", "-oS", "-iL", "-iR", "--excludefile", "--iflist", "--resume", "--datadir", "--script", "-sS", "-sU", "-sA", "-sN", "-sF", "-sX", "-sM", "-sI", "-sC", "-e", "--source-route", "-S", "-D", "-6", "--send-eth", "--spoof-mac", "--traceroute", "--packet-trace", ), ), "curl": PentestTool( name="curl", binary="/usr/bin/curl", level=1, timeout_seconds=60, max_output_bytes=262_144, target_kind="network", target_option="--url", flags=frozenset({"--silent", "--show-error", "--head", "--compressed", "--ipv4", "--fail"}), value_options={ "--max-time": ValueRule(kind="int_range", minimum=1, maximum=120), "--connect-timeout": ValueRule(kind="int_range", minimum=1, maximum=30), "--retry": ValueRule(kind="int_range", minimum=0, maximum=3), "-w": ValueRule(kind="write_out"), # Only read methods. Anything that writes to the target is refused # by the value rule even though the option itself is permitted. "-X": ValueRule(kind="enum", choices=("GET", "HEAD", "OPTIONS")), }, forbidden_prefixes=( "-o", "--output", "-d", "--data", "-F", "--form", "-K", "--config", "--proxy", "-x", "--upload-file", "-T", "--cookie-jar", "-c", "--cookie", "--trace", "--dump-ca", "--cacert", "--capath", "--resolve", "--connect-to", "--interface", "--local-port", "-E", "--cert", "--key", "-L", "--location", "--max-redirs", "--path-as-is", "--proto", "--unix-socket", "--url", "-u", ), ), "ffuf": PentestTool( name="ffuf", binary="/usr/bin/ffuf", level=1, timeout_seconds=180, max_output_bytes=262_144, target_kind="network", target_option="-u", flags=frozenset({"-noninteractive", "-se"}), value_options={ "-w": _WORKSPACE_PATH, "-mc": ValueRule(kind="int_range", minimum=100, maximum=599), "-fs": ValueRule(kind="int_range", minimum=1, maximum=100_000), "-t": ValueRule(kind="int_range", minimum=1, maximum=50), "-timeout": ValueRule(kind="int_range", minimum=1, maximum=30), "-maxtime": ValueRule(kind="int_range", minimum=1, maximum=3600), "-rate": ValueRule(kind="int_range", minimum=1, maximum=1000), }, forbidden_prefixes=( "-o", "-ac", "-recursion", "-proxy", "-c", "-H", "-b", "-config", "-request", "-replay-proxy", "-x", "-fr", "-e", "-S", "-u", ), ), "httpx": PentestTool( name="httpx", binary="/usr/bin/httpx", level=1, timeout_seconds=120, max_output_bytes=262_144, target_kind="network", target_option="-u", flags=frozenset({"-silent", "-no-color", "-nc"}), value_options={ "-t": ValueRule(kind="int_range", minimum=1, maximum=50), "-timeout": ValueRule(kind="int_range", minimum=1, maximum=30), "-rl": ValueRule(kind="int_range", minimum=1, maximum=1000), "-sc": ValueRule(kind="int_range", minimum=100, maximum=599), "-retries": ValueRule(kind="int_range", minimum=0, maximum=3), }, forbidden_prefixes=( "-l", "-L", "-o", "-c", "-config", "-e", "-H", "-b", "-proxy", "-http-proxy", "-socks5", "-fr", "-follow-redirects", "-request", "-x", "-u", ), ), "nikto": PentestTool( name="nikto", binary="/usr/bin/nikto", level=1, timeout_seconds=300, max_output_bytes=1_000_000, target_kind="network", target_option="-h", flags=frozenset({"-nointeractive"}), value_options={ "-p": _PORT_SPEC, "-timeout": ValueRule(kind="int_range", minimum=1, maximum=60), "-maxtime": ValueRule(kind="int_range", minimum=1, maximum=3600), }, # Tuned scanning is a level-2 behaviour: at L1 nikto runs with its # default tuning and nothing else. level_gated_options={ 2: {"-Tuning": ValueRule(kind="hex_flags")}, }, forbidden_prefixes=( "-Format", "-output", "-o", "-h", "-host", "-evasion", "-Rules", "-db", "-C", "-config", "-ssl", "-NoUpdate", "-ask", "-Plugins", ), ), # L2 — vulnerability testing. "nuclei": PentestTool( name="nuclei", binary="/usr/bin/nuclei", level=2, timeout_seconds=300, max_output_bytes=1_000_000, target_kind="network", target_option="-u", flags=frozenset({"-silent", "-nc"}), value_options={ "-rl": ValueRule(kind="int_range", minimum=1, maximum=1000), "-retries": ValueRule(kind="int_range", minimum=0, maximum=3), "-severity": ValueRule( kind="enum_csv", choices=("info", "low", "medium", "high", "critical", "unknown"), ), }, forbidden_prefixes=( # Templates and payloads are code the sandbox would execute; the # whole template vocabulary is therefore unreachable from an intent. "-t", "-templates", "-tpl", "-w", "-wizard", "-c", "-config", "-proxy", "-http-proxy", "-socks5", "-payload", "-p", "-headless", "-interactsh", "-o", "-output", "-me", "-markdown-export", "-l", "-u", "-x", ), ), "sqlmap": PentestTool( name="sqlmap", binary="/usr/bin/sqlmap", level=2, timeout_seconds=600, max_output_bytes=1_000_000, target_kind="network", target_option="-u", flags=frozenset({"--batch", "--disable-coloring", "--forms"}), value_options={ "--level": ValueRule(kind="int_range", minimum=1, maximum=5), "--risk": ValueRule(kind="int_range", minimum=1, maximum=3), "--timeout": ValueRule(kind="int_range", minimum=1, maximum=30), "--retries": ValueRule(kind="int_range", minimum=0, maximum=3), "--threads": ValueRule(kind="int_range", minimum=1, maximum=10), "--dbms": ValueRule( kind="enum", choices=("mysql", "postgresql", "sqlite", "mssql", "oracle", "mariadb"), ), "--technique": ValueRule(kind="enum", choices=("BEUSTQ", "BEU", "UST", "BE", "S")), "--data": ValueRule(kind="text"), }, # Dumping the database is the level-3 behaviour this tool is gated for. level_gated_flags={3: frozenset({"--dump"})}, forbidden_prefixes=( "--os-shell", "--os-cmd", "--sql-shell", "--shell", "--exec", "--file", "--code", "--plugin", "--sql-queries", "--tmpdb", "--dbms-cred", "--output", "--request-file", "-r", "--flush-session", "--session-file", "--save", "--wizard", "--crawl", "--tamper", "--mobile", "--interactiveshell", "-c", "--conf", "--config", "-u", "-D", ), ), # L3 — exploitation. "hydra": PentestTool( name="hydra", binary="/usr/bin/hydra", level=3, timeout_seconds=300, max_output_bytes=262_144, target_kind="network", trailing_tokens=("ssh",), flags=frozenset({"-f"}), value_options={ "-l": ValueRule(kind="identifier"), # Passwords come from a file inside the workspace, never from the # argument vector: argv is recorded in approvals and in the audit # log, and a password there would outlive the engagement. "-P": _WORKSPACE_PATH, "-T": ValueRule(kind="float_range", minimum=0, maximum=60), }, forbidden_prefixes=( "-t", "-x", "-C", "-m", "-W", "-U", "-L", "-p", "-e", "-I", "-F", "-O", "-A", "-S", "-b", "-k", "-w", ), ), "msfpython": PentestTool( name="msfpython", binary="/usr/bin/msfpython", level=3, timeout_seconds=600, max_output_bytes=1_000_000, target_kind="workspace", flags=frozenset({"-q"}), # ``-q`` (quiet) is the only value-less flag; every other Metasploit # resource-selection or session option is refused. A prefix such as # ``--q-`` is not listed because no Metasploit option has that shape: # a deny rule that can never match is a claim of protection it does not # provide, so only prefixes a real argument could begin with are named. forbidden_prefixes=("-x", "-P", "-p", "-e", "-r", "-i", "-m"), ), "custom_script": PentestTool( name="custom_script", binary="/usr/bin/python3", level=3, timeout_seconds=300, max_output_bytes=1_000_000, target_kind="workspace", fixed_flags=("-I", "-B"), forbidden_prefixes=("-c", "-m", "-e", "-u"), ), } def catalog_entry(name: object) -> PentestTool: """Return the catalog entry for ``name``. Raises: PentestRefusal: If the name is unknown. An unknown tool is never a request to run something else. """ if not isinstance(name, str): raise PentestRefusal("unknown_tool", f"tool name must be a string, got {type(name).__name__}") entry = CATALOG.get(name.strip().lower()) if entry is None: raise PentestRefusal("unknown_tool", f"tool {name!r} is not in the pentest catalog") return entry def action_for_level(level: int) -> str: """Return the security-plane action an invocation at ``level`` performs. Raises: PentestRefusal: If the level has no registered action. """ action = ACTION_BY_LEVEL.get(level) if action is None: raise PentestRefusal("unknown_level", f"engagement level {level!r} has no registered action") return action def tool_class_for_level(level: int) -> str: """Return the tool class belonging to ``level``. Raises: PentestRefusal: If the level is outside the known range. """ tool_class = TOOL_CLASS_BY_LEVEL.get(level) if tool_class is None: raise PentestRefusal("unknown_level", f"engagement level {level!r} is not a known level") return tool_class def _authorize_token( tool: PentestTool, token: str, *, level: int, workspace: Path | None, ) -> str | None: """Validate one caller-supplied argument token. Order is the security property, not a style choice: an exact forbidden token is refused first, then an allowed value-less flag is accepted, then the forbidden prefixes are applied, and only then are value options considered. Anything unrecognised is refused. Returns: The token unchanged, or the in-sandbox path when the token is a workspace-path option. Raises: PentestRefusal: If the token is not permitted at ``level``. """ if token in tool.forbidden: raise PentestRefusal( "forbidden_argument", f"argument {token!r} is forbidden for tool {tool.name!r}" ) if token in tool.flags: return token for prefix in tool.forbidden_prefixes: if token.startswith(prefix): raise PentestRefusal( "forbidden_argument", f"argument {token!r} starts with the forbidden prefix {prefix!r} " f"for tool {tool.name!r}", ) for option, rule in tool.value_options.items(): if token.startswith(option) and len(token) > len(option): return _validate_value(rule, token[len(option) :], workspace=workspace) for gated_level, gated_flags in tool.level_gated_flags.items(): if level >= gated_level and token in gated_flags: return token for gated_level, gated_options in tool.level_gated_options.items(): if level < gated_level: continue for option, rule in gated_options.items(): if token.startswith(option) and len(token) > len(option): return _validate_value(rule, token[len(option) :], workspace=workspace) raise PentestRefusal( "argument_not_allowed", f"argument {token!r} is not permitted for tool {tool.name!r} at level {level}", ) def _resolve_target( tool: PentestTool, target: object, *, scope: TargetScope, workspace: Path | None, ) -> str: """Return the validated target token to place in the argv. ``workspace`` is optional in the type because a network tool never needs one; a workspace tool that is handed none is refused rather than resolved against an invented root. Raises: PentestRefusal: If the target is unusable, out of scope, or, for a workspace tool, outside the workspace or without a workspace. """ if tool.target_kind == "workspace": if workspace is None: raise PentestRefusal( "no_workspace", f"tool {tool.name!r} operates on a workspace path but no workspace was provided", ) if not isinstance(target, str): raise PentestRefusal("target_malformed", f"target must be a string, got {type(target).__name__}") return _in_sandbox_path(target.strip(), workspace) if not scope: raise PentestRefusal( "no_scope", f"tool {tool.name!r} addresses the network but the engagement authorises no target", ) if not isinstance(target, str): raise PentestRefusal("target_malformed", f"target must be a string, got {type(target).__name__}") text = target.strip() if not scope.allows(text): # The scope check is per invocation, not per session: a target that was # authorised a moment ago is re-checked here, every time. raise PentestRefusal( "target_out_of_scope", f"target {text!r} is not inside the authorised engagement scope", ) return text @dataclass(frozen=True) class AuthorizedToolCall: """A tool invocation the catalog has authorized, ready to be executed. This object — not the model, and not the caller — is what decides what runs. ``argv`` is built here; ``argv_text`` is a human-readable rendering of the very same vector, used as the approval record's resource so a reviewer reads the command that will actually run. Attributes: tool: Catalog entry that was used. target: The validated target, as it appears in ``argv``. level: Engagement level the invocation was authorised at. action: Security-plane action for that level. argv: The reconstructed argument vector. argv_text: ``shlex.join`` rendering of ``argv``, for display only. timeout_seconds: Wall-clock ceiling for this invocation. max_output_bytes: Captured-output ceiling for this invocation. engagement_digest: Digest of the engagement the call was authorised under, recorded so an audit reader can tie the run to the terms. """ tool: PentestTool target: str level: int action: str argv: tuple[str, ...] argv_text: str timeout_seconds: int max_output_bytes: int engagement_digest: str def authorize_tool_call( intent: Mapping[str, object], *, engagement: EngagementAuthorization | None, scope: TargetScope | None = None, workspace: str | Path | None = None, audit_log: AuditLog | None = None, now: float | None = None, ) -> AuthorizedToolCall: """Reconstruct an argv from a declared intent, or refuse. Args: intent: The declared intent. Only ``tool``, ``target``, ``args`` and ``level`` are read; any other key is ignored rather than trusted, so a payload cannot smuggle a pre-built command line past the catalog. engagement: The signed artifact. ``None`` is a refusal, never a default engagement. scope: Whitelist to check the target against; defaults to the scope derived from ``engagement.targets``. workspace: Host path of the writable workspace this call may use. audit_log: Log that receives the authorization decision and, for every refusal, a record naming the reason. now: Override for the engagement's time window, for tests. Returns: The authorized call. The caller executes ``result.argv`` and nothing else. Raises: PentestRefusal: For every refusal, with a ``reason_code`` suitable for the audit log. Each one is recorded there before it is raised. """ try: return _authorize_tool_call( intent, engagement=engagement, scope=scope, workspace=workspace, now=now, ) except PentestRefusal as refusal: if audit_log is not None: # The refusal for a non-mapping intent is raised from inside this # function, so the reader below has to tolerate the very value that # caused it: an audit writer that raised would lose the refusal it # exists to record. readable: Mapping[str, object] = intent if isinstance(intent, Mapping) else {} audit_log.record( AuditEventType.PERMISSION_CHECK, actor="pentester", agent_id="pentester", task_id=str(readable.get("task_id") or ""), action=ACTION_BY_LEVEL.get(_safe_level(readable), "pentest_tool"), result="refused", error=refusal.reason_code, metadata={ "detail": str(refusal), "tool": _safe_text(readable.get("tool")), "target": _safe_text(readable.get("target")), "engagement_digest": engagement.digest if engagement is not None else "", }, ) raise def _safe_level(intent: Mapping[str, object]) -> int: """Return the declared level as an int, or -1 when it is not one.""" level = intent.get("level") if isinstance(level, bool) or not isinstance(level, int): return -1 return level def _safe_text(value: object) -> str: return value if isinstance(value, str) else repr(value) def _authorize_tool_call( intent: Mapping[str, object], *, engagement: EngagementAuthorization | None, scope: TargetScope | None, workspace: str | Path | None, now: float | None, ) -> AuthorizedToolCall: """Build the authorized call. See :func:`authorize_tool_call` for the contract.""" if engagement is None: raise PentestRefusal( "no_engagement", "an engagement authorization is required; without one the tool call is refused", ) if not isinstance(intent, Mapping): raise PentestRefusal("intent_malformed", f"intent must be a mapping, got {type(intent).__name__}") if not engagement.is_active(now): raise PentestRefusal( "engagement_window_closed", f"engagement {engagement.digest[:12]} is outside its authorised time window", ) tool = catalog_entry(intent.get("tool")) declared_level = intent.get("level") if declared_level is None: level = engagement.max_level elif isinstance(declared_level, bool) or not isinstance(declared_level, int): raise PentestRefusal("level_malformed", f"level must be an integer, got {declared_level!r}") else: level = declared_level if level < 0 or level > 3: raise PentestRefusal("level_malformed", f"level {level!r} is outside 0-3") if level < tool.level: raise PentestRefusal( "tool_above_level", f"tool {tool.name!r} requires level {tool.level}, the call declared level {level}", ) if not engagement.engages_level(level): raise PentestRefusal( "level_not_engaged", f"engagement {engagement.digest[:12]} permits at most level " f"{engagement.max_level} and tool class {sorted(engagement.allowed_tool_classes)}, " f"which does not cover level {level} ({tool_class_for_level(level)})", ) resolved_scope = scope if scope is not None else engagement.scope workspace_path = Path(workspace).expanduser().resolve() if workspace is not None else None if tool.target_kind == "workspace" and workspace_path is None: raise PentestRefusal( "no_workspace", f"tool {tool.name!r} operates on a workspace path but no workspace was provided", ) target_token = _resolve_target(tool, intent.get("target"), scope=resolved_scope, workspace=workspace_path) raw_args = intent.get("args") if raw_args is None: raw_args = () if isinstance(raw_args, (str, bytes)) or not isinstance(raw_args, Sequence): raise PentestRefusal( "args_malformed", f"args must be a sequence of argument tokens, got {type(raw_args).__name__}", ) if len(raw_args) > MAX_ARGUMENTS: raise PentestRefusal( "too_many_arguments", f"at most {MAX_ARGUMENTS} arguments are accepted" ) tokens: list[str] = [] for raw in raw_args: if not isinstance(raw, str): raise PentestRefusal( "argument_malformed", f"every argument must be a string, got {type(raw).__name__}", ) token = raw.strip() if not token: raise PentestRefusal("argument_malformed", "an argument must not be empty") if len(token) > MAX_ARGUMENT_LENGTH: raise PentestRefusal( "argument_too_long", f"an argument must not exceed {MAX_ARGUMENT_LENGTH} characters", ) validated = _authorize_token(tool, token, level=level, workspace=workspace_path) tokens.append(validated if validated is not None else token) argv = [tool.binary, *tool.fixed_flags, *tokens] if tool.target_option is not None: argv.extend((tool.target_option, target_token)) else: argv.append(target_token) argv.extend(tool.trailing_tokens) return AuthorizedToolCall( tool=tool, target=target_token, level=level, action=action_for_level(level), argv=tuple(argv), argv_text=shlex.join(argv), timeout_seconds=tool.timeout_seconds, max_output_bytes=tool.max_output_bytes, engagement_digest=engagement.digest, ) def describe_catalog() -> list[dict[str, Any]]: """Return a JSON-compatible view of the catalog, for ``cyberbrain pentest --list``.""" return [ { "tool": entry.name, "level": entry.level, "tool_class": TOOL_CLASS_BY_LEVEL.get(entry.level, ""), "binary": entry.binary, "target_kind": entry.target_kind, "timeout_seconds": entry.timeout_seconds, "max_output_bytes": entry.max_output_bytes, "allowed_arguments": sorted(entry.flags) + sorted(entry.value_options) + sorted(entry.fixed_flags), "level_gated_arguments": sorted( flag for flags in entry.level_gated_flags.values() for flag in flags ) + sorted( option for options in entry.level_gated_options.values() for option in options ), } for entry in sorted(CATALOG.values(), key=lambda item: (item.level, item.name)) ]