From 3dcff8d872acdd3c7bfe167d8b4aa730b52e4505 Mon Sep 17 00:00:00 2001 From: Mohammadreza Khellat Date: Thu, 12 Nov 2026 09:28:52 +0330 Subject: [PATCH 1/5] feat(safety): let an optional layer register step checks Phase 10, step 5 plan, PR 4. The engagement scope check belongs to the security layer, but steps are judged in the core, which never imports that layer. The core now keeps two hook lists: - add_step_check(fn): fn(step) returns why the step is refused, or None; - add_step_notice(fn): fn(step) returns why it must be confirmed, or None. SafetyEngine asks the checks after its own rules (an allow-list refusal is reported as such first), for every step and again right before it runs (may_execute), and the notices in confirmation_reason. A hook that raises refuses the step: a broken check must never let one through. The engine does not know what the hooks are -- complete mediation without a dependency on the layer. Manual (Safety and Sandbox > The Safety Engine) and the test list updated. --- docs/chapters/development.texi | 1 + docs/chapters/safety-and-sandbox.texi | 9 ++++ src/codebud/safety.py | 52 +++++++++++++++++-- tests/test_step_checks.py | 73 +++++++++++++++++++++++++++ 4 files changed, 131 insertions(+), 4 deletions(-) create mode 100644 tests/test_step_checks.py diff --git a/docs/chapters/development.texi b/docs/chapters/development.texi index 1b8a9d8..f9f0e5d 100644 --- a/docs/chapters/development.texi +++ b/docs/chapters/development.texi @@ -168,6 +168,7 @@ tests/ ├── test_skill.py SKILL.md variants (default, Agensi), skill CLI ├── test_step_environment.py steps never see the passphrase or secret │ settings (any privacy level) +├── test_step_checks.py the core's hook for registered step checks ├── test_steps.py shared step loop: approval policies, the │ first failure stops the run ├── test_terminal_safety.py control characters in replies, output and diff --git a/docs/chapters/safety-and-sandbox.texi b/docs/chapters/safety-and-sandbox.texi index c393fe3..882a43e 100644 --- a/docs/chapters/safety-and-sandbox.texi +++ b/docs/chapters/safety-and-sandbox.texi @@ -219,6 +219,15 @@ command.}), and so is a policy rejection, naming the step and the reason allow-list}; @pxref{Safety Rejection}). The same reasons are given when a step is refused right before it runs. +@cindex step checks, registered +An optional layer can add checks of its own: @code{add_step_check()} +registers a function that may refuse a step, and +@code{add_step_notice()} one that may ask for a confirmation. The +engine asks them after its own rules, for every step and again right +before it runs, without knowing what they are; a check that fails with +an error refuses the step. The security layer's engagement scope guard +is registered this way, so the core never imports that layer. + @code{describe_rules()} returns a compact summary of the three policy lists, which is embedded in the planner prompt so the model can avoid proposing forbidden commands in the first place. diff --git a/src/codebud/safety.py b/src/codebud/safety.py index f097460..11ab634 100644 --- a/src/codebud/safety.py +++ b/src/codebud/safety.py @@ -472,6 +472,45 @@ def egress_reason(step: dict[str, Any]) -> str | None: return None +# Extra checks registered by an optional layer (the security layer's scope +# check, when an engagement is active): the core asks them without +# knowing what they are. A check returns why a step is refused (or None); +# a notice returns why it must be confirmed (or None). +_STEP_CHECKS: list = [] +_STEP_NOTICES: list = [] + + +def add_step_check(check) -> None: + _STEP_CHECKS.append(check) + + +def remove_step_check(check) -> None: + if check in _STEP_CHECKS: + _STEP_CHECKS.remove(check) + + +def add_step_notice(notice) -> None: + _STEP_NOTICES.append(notice) + + +def remove_step_notice(notice) -> None: + if notice in _STEP_NOTICES: + _STEP_NOTICES.remove(notice) + + +def _registered(hooks: list, step: dict[str, Any]) -> str | None: + """The first reason a registered hook gives for ``step``. A hook that + fails is a refusal: a broken check must never let a step through.""" + for hook in list(hooks): + try: + reason = hook(step) + except Exception as exc: # noqa: BLE001 - any failure refuses + return f"a registered step check failed ({type(exc).__name__}: {exc})" + if reason: + return reason + return None + + class SafetyEngine: """ Validates steps produced by the LLM planner. @@ -520,10 +559,12 @@ def refusal_reason(self, step: dict[str, Any]) -> str | None: if privacy.level() == "locked": return "steps that need the network are not allowed at privacy level locked" if tool == "command": - return self._command_refusal(step.get("args", {}).get("cmd", "")) - if tool in ("file_read", "file_write", "patch"): - return _file_refusal(tool, step.get("args", {})) - return None + reason = self._command_refusal(step.get("args", {}).get("cmd", "")) + elif tool in ("file_read", "file_write", "patch"): + reason = _file_refusal(tool, step.get("args", {})) + else: + reason = None + return reason or _registered(_STEP_CHECKS, step) def is_powerful(self, step: dict[str, Any]) -> bool: """Return whether ``step`` is a command on the powerful list. @@ -542,6 +583,9 @@ def confirmation_reason(self, step: dict[str, Any]) -> str | None: Powerful commands run code or change files; egress steps (web search) send data off the machine. """ + notice = _registered(_STEP_NOTICES, step) + if notice: + return notice if self.is_powerful(step): return POWERFUL_REASON # At privacy level "trusted" egress runs without asking; safety diff --git a/tests/test_step_checks.py b/tests/test_step_checks.py new file mode 100644 index 0000000..f1c4424 --- /dev/null +++ b/tests/test_step_checks.py @@ -0,0 +1,73 @@ +""" +tests/test_step_checks.py + +The core's hook for extra step checks (phase 10, step 5 plan, PR 4). +The security layer registers its scope check here when an engagement is +active; the core asks every registered check for every step without +knowing what they are, and never imports the layer. +""" + +import pytest + +from codebud import safety +from codebud.safety import SafetyEngine + + +@pytest.fixture(autouse=True) +def clean_hooks(monkeypatch): + monkeypatch.setattr(safety, "_STEP_CHECKS", []) + monkeypatch.setattr(safety, "_STEP_NOTICES", []) + + +def _engine(): + engine = SafetyEngine() + engine.harmless_commands = ["ls"] + engine.powerful_commands = ["nmap"] + return engine + + +def _step(cmd, **args): + return {"tool": "command", "args": {"cmd": cmd, **args}} + + +def test_a_registered_check_refuses_steps(): + safety.add_step_check(lambda step: "out of scope" if "10.9" in step["args"]["cmd"] else None) + engine = _engine() + assert engine.refusal_reason(_step("nmap 10.9.0.1")) == "out of scope" + assert engine.refusal_reason(_step("nmap 10.0.0.1")) is None + + +def test_checks_run_after_the_built_in_rules(): + """A command the allow-list refuses is refused for that reason first.""" + seen = [] + safety.add_step_check(lambda step: seen.append(step) or None) + reason = _engine().refusal_reason(_step("rm -rf x")) + assert "allow-list" in reason + assert seen == [] + + +def test_checks_apply_right_before_running_too(): + safety.add_step_check(lambda step: "window closed") + allowed, reason = _engine().may_execute(_step("ls")) + assert not allowed and "window closed" in reason + + +def test_a_notice_asks_for_confirmation(): + safety.add_step_notice(lambda step: "host is outside the scope") + assert _engine().confirmation_reason(_step("ls")) == "host is outside the scope" + + +def test_removing_a_check(): + check = lambda step: "no" # noqa: E731 + safety.add_step_check(check) + safety.remove_step_check(check) + assert _engine().refusal_reason(_step("ls")) is None + + +def test_a_crashing_check_refuses_rather_than_lets_through(): + def broken(step): + raise ValueError("bug") + + safety.add_step_check(broken) + reason = _engine().refusal_reason(_step("ls")) + assert reason is not None and "check failed" in reason From 35841ce6056c3f563fdb0a7c93f95cc51fc0a2d0 Mon Sep 17 00:00:00 2001 From: Mohammadreza Khellat Date: Thu, 12 Nov 2026 09:29:35 +0330 Subject: [PATCH 2/5] feat(security): the engagement scope guard (DNS, scanners, nc) Phase 10, step 5 plan, PR 4 (decision D14i). Until now no code read a network step's target: the engagement's scope was only recorded. codebud_sec/scope.py reads every target a network tool will contact from its command line and checks it against the engagement: - active tools (nmap, masscan, nc, ncat) are refused when a target is outside the scope, excluded, or the window is closed -- a hard block: a model's plan is untrusted, and the UK AISI incident report (4 Aug 2026) shows prompt-level scope is not enough; - passive lookups (dig, host) out of scope ask for confirmation, D14's warn-and-confirm kept for them. A target that cannot be read is refused. Each tool's parser knows which options take values; an unknown option is refused, since its value could be a target the guard cannot see. Hosts a tool contacts besides its target are targets too: DNS servers (dig @, nmap --dns-servers), proxies (nc -x, ncat --proxy, nmap --proxies), the idle-scan zombie (nmap -sI) and the FTP relay (nmap -b). nmap -iR (random hosts), --resume, masscan -c and dig -f are refused; target lists (-iL) are read only from a file inside the project. Ranges -- CIDRs, masscan a-b, nmap octet ranges by their bounding range -- must fit in the scope as a whole, and engagement.covers_network() checks that against the excluded networks too. A name never matches addresses: resolving it would itself be a network action. A step that declares the network with a tool the guard cannot read is refused, and an engagement at privacy trusted (no isolation) refuses every command. ScopeGuard.activate() registers the guard with the core's step-check hook; the codebud-sec entry point will open the engagement and activate it (PR 7). CORE_API widens by codebud.safety (the hook) and codebud.privacy, deliberately. HTTP tools, web fuzzers and TLS scanners follow in PR 4b. Architecture layout, build list and test list updated. --- docs/chapters/architecture.texi | 3 + docs/chapters/development.texi | 3 + src/Makefile.am | 3 +- src/codebud_sec/engagement.py | 18 ++ src/codebud_sec/scope.py | 469 ++++++++++++++++++++++++++++++++ tests/test_engagement.py | 18 ++ tests/test_layers.py | 2 + tests/test_scope.py | 230 ++++++++++++++++ 8 files changed, 745 insertions(+), 1 deletion(-) create mode 100644 src/codebud_sec/scope.py create mode 100644 tests/test_scope.py diff --git a/docs/chapters/architecture.texi b/docs/chapters/architecture.texi index 5b477e7..25eb40e 100644 --- a/docs/chapters/architecture.texi +++ b/docs/chapters/architecture.texi @@ -637,6 +637,9 @@ src/codebud/ the package src/codebud_sec/ the security layer (authorized security work) ├── engagement.py engagement file: authorized scope +├── scope.py scope guard: every network target read from the +│ command line and checked; registered with the +│ core's safety.add_step_check() └── evidence.py hash-chained, append-only evidence log, encrypted at rest (appends need no passphrase) diff --git a/docs/chapters/development.texi b/docs/chapters/development.texi index f9f0e5d..d44c752 100644 --- a/docs/chapters/development.texi +++ b/docs/chapters/development.texi @@ -165,6 +165,9 @@ tests/ │ what would change the host is refused ├── test_shipped_profiles.py profiles of codebud_sec: found by name │ beside codebud, not imported, not shadowed +├── test_scope.py engagement scope: targets read per tool, +│ third parties, ranges, active refused, +│ passive asked ├── test_skill.py SKILL.md variants (default, Agensi), skill CLI ├── test_step_environment.py steps never see the passphrase or secret │ settings (any privacy level) diff --git a/src/Makefile.am b/src/Makefile.am index 5d0a044..ef2acd7 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -62,7 +62,8 @@ nobase_python_PYTHON = \ codebud/tools/web_search.py \ codebud_sec/__init__.py \ codebud_sec/engagement.py \ - codebud_sec/evidence.py + codebud_sec/evidence.py \ + codebud_sec/scope.py nobase_dist_python_DATA = \ codebud/data/policy/harmless_commands.json \ diff --git a/src/codebud_sec/engagement.py b/src/codebud_sec/engagement.py index 8c4c272..388396c 100644 --- a/src/codebud_sec/engagement.py +++ b/src/codebud_sec/engagement.py @@ -108,6 +108,24 @@ def covers(self, target: str) -> bool: return False return any(t.matches(target) for t in self.in_scope) + def covers_network(self, network) -> bool: + """Whether every address of ``network`` (an ipaddress network) is in + scope: inside one in-scope network and overlapping no excluded + network or address. Names never cover addresses.""" + inside = any( + t.network is not None + and t.network.version == network.version + and network.subnet_of(t.network) + for t in self.in_scope + ) + excluded = any( + t.network is not None + and t.network.version == network.version + and network.overlaps(t.network) + for t in self.out_of_scope + ) + return inside and not excluded + def out_of_scope_reason(self, target: str, when: datetime | None = None) -> str | None: """Why acting on ``target`` is outside the engagement, or None. diff --git a/src/codebud_sec/scope.py b/src/codebud_sec/scope.py new file mode 100644 index 0000000..d7e1357 --- /dev/null +++ b/src/codebud_sec/scope.py @@ -0,0 +1,469 @@ +""" +codebud_sec/scope.py + +Keeping network steps inside the engagement (phase 10, step 5 plan, +PR 4; decision D14i). Every target a network tool will contact is read +from its command line and checked against the engagement file: + +- active tools (scanners, nc) are refused when a target is outside the + scope, excluded, or the engagement window is closed -- a hard block, + because a model's plan is untrusted input and a warning is clicked + through (UK AISI, 4 Aug 2026; Akhawe & Felt 2013); +- passive lookups (dig, host) out of scope ask for confirmation (D14's + warn-and-confirm, kept for them). + +A target that cannot be read is refused: each tool's parser knows which +options take values, and an option it does not know is refused, since +its value could be a target it cannot see. Hosts a tool contacts besides +its target -- a DNS server, a proxy, an idle-scan zombie, an FTP relay +-- are targets too. Target lists from a file are read only from a file +inside the project. A range must fit in the scope as a whole; a name is +never matched against addresses (that would mean resolving it, itself a +network action). + +A step that declares the network with a tool that has no parser here is +refused. The guard is registered with the core's safety engine +(codebud.safety.add_step_check / add_step_notice) by whoever opens the +engagement; the core never imports this layer. +""" + +import ipaddress +import os +import re +import shlex +from dataclasses import dataclass, field + +PASSIVE, ACTIVE = "passive", "active" + + +class TargetError(Exception): + """A command line whose targets cannot be read.""" + + +@dataclass +class Targets: + tier: str + hosts: list[str] = field(default_factory=list) # names and single addresses + networks: list = field(default_factory=list) # ipaddress networks (ranges, CIDRs) + + def add(self, spec: str, what: str = "target") -> None: + kind, value = _spec(spec, what) + (self.hosts if kind == "host" else self.networks).extend(value) + + +# --- target specifications ---------------------------------------------------- + +_NAME = re.compile(r"^(?=.{1,253}$)([A-Za-z0-9_]([A-Za-z0-9_-]{0,61}[A-Za-z0-9])?\.)*" + r"[A-Za-z0-9_]([A-Za-z0-9_-]{0,61}[A-Za-z0-9])?\.?$") # fmt: skip +_OCTET = re.compile(r"^(\*|\d{1,3}(-\d{1,3})?(,\d{1,3}(-\d{1,3})?)*)$") + + +def _spec(spec: str, what: str): + """("host", [str]) or ("networks", [network, ...]) for one target + specification, or TargetError if it is not one codebud can judge.""" + text = spec.strip().strip("[]") + if not text: + raise TargetError(f"empty {what}") + try: + return "host", [str(ipaddress.ip_address(text))] + except ValueError: + pass + if "/" in text: + try: + return "networks", [ipaddress.ip_network(text, strict=False)] + except ValueError as exc: + raise TargetError(f"{what} {spec!r}: {exc}") from exc + if text.count("-") == 1 and text.count(".") == 6: # masscan: a.b.c.d-e.f.g.h + first, last = text.split("-") + try: + lo, hi = ipaddress.ip_address(first), ipaddress.ip_address(last) + return "networks", list(ipaddress.summarize_address_range(lo, hi)) + except (ValueError, TypeError) as exc: + raise TargetError(f"{what} {spec!r}: {exc}") from exc + octets = text.split(".") + # Four dotted parts of digits, ranges and * (a plain address was handled + # above) are a range, though "10.0.0.1-5" would also pass as a name. + if len(octets) == 4 and all(_OCTET.match(o) for o in octets): + # nmap octet ranges (10.0.0-3.1-254, 10.0.0.*): judged by their + # bounding range, which holds every address they name. + lows, highs = [], [] + for octet in octets: + numbers = [0, 255] if octet == "*" else [int(n) for n in re.split(r"[-,]", octet)] + if any(n > 255 for n in numbers): + raise TargetError(f"{what} {spec!r}: not an address range") + lows.append(min(numbers)) + highs.append(max(numbers)) + lo = ipaddress.ip_address(".".join(map(str, lows))) + hi = ipaddress.ip_address(".".join(map(str, highs))) + return "networks", list(ipaddress.summarize_address_range(lo, hi)) + if _NAME.match(text) and not text.replace(".", "").isdigit(): + return "host", [text.lower().rstrip(".")] + raise TargetError(f"{what} {spec!r} is not an address, network, range or host name") + + +def _list_file(path: str, targets: Targets) -> None: + """Targets read from a file: only a file inside the project.""" + real = os.path.realpath(path) + project = os.path.realpath(os.getcwd()) + if os.path.commonpath([real, project]) != project or not os.path.isfile(real): + raise TargetError(f"target list {path}: only a file inside the project can be read") + with open(real, encoding="utf-8", errors="replace") as f: + for line in f: + for spec in line.split("#", 1)[0].split(): + targets.add(spec, f"target in {path}") + + +# --- a small option walker ------------------------------------------------------- + + +def _walk(args, value_opts, flag_opts, flag_patterns=(), short_clusters=False): + """Yield (option, value) for options and (None, word) for operands. + An option that is neither a known flag nor takes a value is a + TargetError: its value could be a target.""" + i = 0 + while i < len(args): + arg = args[i] + i += 1 + if arg == "--": + yield from ((None, a) for a in args[i:]) + return + if not arg.startswith("-") or arg == "-": + yield None, arg + continue + name, eq, attached = arg.partition("=") + if name in value_opts: + if eq: + yield name, attached + elif i < len(args): + yield name, args[i] + i += 1 + else: + raise TargetError(f"{name} needs a value") + continue + if arg in flag_opts or any(re.match(p, arg) for p in flag_patterns): + yield arg, None + continue + if short_clusters and not arg.startswith("--") and len(arg) > 2: + # -lvnp 4444: flags, the last one possibly taking the next word. + letters = [f"-{c}" for c in arg[1:]] + if all(c in flag_opts for c in letters[:-1]) and ( + letters[-1] in flag_opts or letters[-1] in value_opts + ): + for letter in letters[:-1]: + yield letter, None + if letters[-1] in value_opts: + if i >= len(args): + raise TargetError(f"{letters[-1]} needs a value") + yield letters[-1], args[i] + i += 1 + else: + yield letters[-1], None + continue + raise TargetError(f"{arg}: an option codebud does not know, so it cannot tell its targets") + + +# --- the tools ------------------------------------------------------------------- + +_DNS_TYPES = { + "A", "AAAA", "ANY", "AXFR", "CAA", "CNAME", "DNSKEY", "DS", "HINFO", "IXFR", "MX", + "NAPTR", "NS", "NSEC", "NSEC3", "PTR", "RRSIG", "SOA", "SPF", "SRV", "SSHFP", + "TLSA", "TXT", "IN", "CH", "HS", +} # fmt: skip + + +def _dig(args): + targets = Targets(PASSIVE) + value = {"-t", "-c", "-p", "-b", "-k", "-y", "-q", "-x", "-f"} + flags = {"-4", "-6", "-m", "-r", "-u", "-h", "-v", "-i"} + for option, word in _walk(args, value, flags): + if option == "-f": + raise TargetError("dig -f reads queries from a file, which cannot be judged") + if option in ("-x", "-q"): + targets.add(word, "lookup") + elif option is None: + if word.startswith("@"): + targets.add(word[1:], "DNS server") + elif word.startswith("+"): + continue + elif word.upper() not in _DNS_TYPES: + targets.add(word, "lookup") + return targets + + +def _host(args): + targets = Targets(PASSIVE) + value = {"-t", "-c", "-W", "-R", "-m", "-N"} + flags = { + "-a", + "-A", + "-C", + "-d", + "-i", + "-l", + "-r", + "-s", + "-T", + "-U", + "-v", + "-V", + "-w", + "-4", + "-6", + } + operands = [w for o, w in _walk(args, value, flags) if o is None] + if not operands or len(operands) > 2: + raise TargetError("host takes a name and optionally a DNS server") + targets.add(operands[0], "lookup") + if len(operands) == 2: + targets.add(operands[1], "DNS server") + return targets + + +_NMAP_VALUE = { + "-p", "--exclude-ports", "--top-ports", "--port-ratio", "-oN", "-oX", "-oS", "-oG", "-oA", + "-e", "-g", "--source-port", "--data-length", "--data", "--data-string", "--ttl", + "--spoof-mac", "--scan-delay", "--max-scan-delay", "--min-rate", "--max-rate", + "--max-retries", "--host-timeout", "--min-rtt-timeout", "--max-rtt-timeout", + "--initial-rtt-timeout", "--min-hostgroup", "--max-hostgroup", "--min-parallelism", + "--max-parallelism", "--version-intensity", "--script", "--script-args", "--mtu", + "--datadir", "--servicedb", "--versiondb", "--stylesheet", "--exclude", "--excludefile", + "--script-timeout", "--max-os-tries", "-iL", "-iR", "--dns-servers", "-sI", "-b", "-S", + "-D", "--proxies", "--resume", "--script-args-file", +} # fmt: skip +_NMAP_FLAGS = { + "-6", "-A", "-F", "-r", "-n", "-R", "-f", "-ff", "-O", "--open", "--reason", "--traceroute", + "--packet-trace", "--badsum", "--privileged", "--unprivileged", "--system-dns", + "--osscan-limit", "--osscan-guess", "--version-light", "--version-all", "--version-trace", + "--script-trace", "--defeat-rst-ratelimit", "--append-output", "--no-stylesheet", + "--webxml", "--noninteractive", "--send-eth", "--send-ip", "-h", "--help", "-V", "--version", +} # fmt: skip +_NMAP_PATTERNS = (r"^-s[A-HJ-Za-z]{1,2}$", r"^-P[A-Za-z][0-9,\-]*$", r"^-T[0-5]$", r"^-[vd]+$", + r"^-p[0-9TUS:,\-]+$") # fmt: skip + + +def _nmap(args): + targets = Targets(ACTIVE) + for option, word in _walk(args, _NMAP_VALUE, _NMAP_FLAGS, _NMAP_PATTERNS): + if option == "-iR": + raise TargetError("nmap -iR scans random hosts, outside any scope") + if option in ("--resume", "--script-args-file"): + raise TargetError(f"nmap {option} reads from a file that cannot be judged") + if option == "-iL": + _list_file(word, targets) + elif option == "--dns-servers": + for server in word.split(","): + targets.add(server, "DNS server") + elif option == "-sI": + targets.add(word.split(":")[0], "idle-scan zombie") + elif option == "-b": + relay = word.rsplit("@", 1)[-1].split(":")[0] + targets.add(relay, "FTP relay") + elif option == "--proxies": + for proxy in word.split(","): + targets.add(re.sub(r"^\w+://", "", proxy).split(":")[0].split("/")[0], "proxy") + elif option is None: + targets.add(word) + if not targets.hosts and not targets.networks: + raise TargetError("nmap names no target") + return targets + + +_MASSCAN_VALUE = { + "-p", "--ports", "--rate", "-oL", "-oX", "-oJ", "-oG", "-oB", "--exclude", "--excludefile", + "--adapter", "--adapter-ip", "--adapter-port", "--adapter-mac", "--router-mac", + "--source-ip", "--source-port", "--wait", "--retries", "--ttl", "--shard", "--seed", + "--connection-timeout", "--range", "-iL", "--includefile", "-c", "--conf", "--resume", + "--readscan", "--http-user-agent", +} # fmt: skip +_MASSCAN_FLAGS = { + "--banners", + "--open", + "--open-only", + "--ping", + "--offline", + "-v", + "-vv", + "--echo", +} + + +def _masscan(args): + targets = Targets(ACTIVE) + for option, word in _walk(args, _MASSCAN_VALUE, _MASSCAN_FLAGS, (r"^-p[0-9U:,\-]+$",)): + if option in ("-c", "--conf", "--resume"): + raise TargetError(f"masscan {option} reads settings, targets included, from a file") + if option in ("-iL", "--includefile"): + _list_file(word, targets) + elif option == "--range" or option is None: + for spec in word.split(","): + targets.add(spec) + if not targets.hosts and not targets.networks: + raise TargetError("masscan names no target") + return targets + + +_NC_VALUE = {"-p", "-s", "-w", "-i", "-q", "-x", "-X", "-T", "-I", "-O", "-M", "-m", "-P", "-V"} +_NC_FLAGS = { + "-l", + "-k", + "-n", + "-u", + "-v", + "-z", + "-4", + "-6", + "-N", + "-d", + "-D", + "-h", + "-C", + "-t", + "-b", + "-r", +} + + +def _nc(args): + targets = Targets(ACTIVE) + listen, operands = False, [] + for option, word in _walk(args, _NC_VALUE, _NC_FLAGS, (r"^-v+$",), short_clusters=True): + if option == "-l": + listen = True + elif option == "-x": + targets.add(word.rsplit(":", 1)[0] if word.count(":") == 1 else word, "proxy") + elif option is None: + operands.append(word) + if listen: + return targets # listening: nothing contacted, only proxies (if any) + if not operands: + raise TargetError("nc names no host") + targets.add(operands[0]) + return targets + + +_NCAT_VALUE = { + "-s", "--source", "-p", "--source-port", "-w", "--wait", "-i", "--idle-timeout", + "--proxy", "--proxy-type", "--proxy-auth", "-o", "--output", "-x", "--hex-dump", + "--allow", "--deny", "--allowfile", "--denyfile", "-m", "--max-conns", "--ssl-cert", + "--ssl-key", "--ssl-trustfile", "--ssl-ciphers", "--ssl-alpn", "-d", "--delay", +} # fmt: skip +_NCAT_FLAGS = { + "-l", "--listen", "-k", "--keep-open", "-u", "--udp", "--ssl", "--ssl-verify", "--send-only", + "--recv-only", "-n", "--nodns", "-v", "--verbose", "-C", "--crlf", "-4", "-6", "-z", + "-t", "--telnet", "--broker", "--chat", "-h", "--help", "--version", +} # fmt: skip + + +def _ncat(args): + targets = Targets(ACTIVE) + listen, operands = False, [] + for option, word in _walk(args, _NCAT_VALUE, _NCAT_FLAGS, (r"^-v+$",), short_clusters=True): + if option in ("-l", "--listen"): + listen = True + elif option == "--proxy": + targets.add(word.rsplit(":", 1)[0] if word.count(":") == 1 else word, "proxy") + elif option is None: + operands.append(word) + if listen: + return targets + if not operands: + raise TargetError("ncat names no host") + targets.add(operands[0]) + return targets + + +PARSERS = { + "dig": _dig, "host": _host, "nmap": _nmap, "masscan": _masscan, + "nc": _nc, "netcat": _nc, "ncat": _ncat, +} # fmt: skip + + +def targets_of(words: list[str]) -> list[tuple[str, Targets]]: + """(program, Targets) for every word of the command line that names a + tool with a parser -- behind wrappers too. TargetError if any of them + cannot be read.""" + found = [] + for i, word in enumerate(words): + name = os.path.basename(word) + name = "nc" if name.startswith("nc.") else name + if name in PARSERS: + found.append((name, PARSERS[name](words[i + 1 :]))) + return found + + +# --- the guard ------------------------------------------------------------------- + + +class ScopeGuard: + """The scope check for one engagement, to register with the core's + safety engine (activate) and remove again (deactivate).""" + + def __init__(self, engagement): + self.engagement = engagement + + def _problems(self, targets: Targets) -> list[str]: + e = self.engagement + problems = [f"{h} is not in the engagement scope" for h in targets.hosts if not e.covers(h)] + problems += [ + f"{n} is not wholly in the engagement scope" + for n in targets.networks + if not e.covers_network(n) + ] + if not e.in_window(): + problems.append("the engagement window is not open now") + return problems + + def _judge(self, step) -> tuple[str | None, str | None]: + """(refusal, notice) for ``step``.""" + if step.get("tool") != "command": + return None, None + from codebud import privacy + + if privacy.level() == "trusted": + return ( + "an engagement needs isolated steps: privacy level trusted runs them " + "unisolated, with the whole network", + None, + ) + args = step.get("args", {}) + try: + words = shlex.split(args.get("cmd", "")) + found = targets_of(words) + except (ValueError, TargetError) as exc: + return f"engagement scope: {exc}", None + if not found: + if args.get("network"): + return ( + f"engagement scope: {words[0] if words else 'this step'} needs the network, " + "but codebud cannot read its targets", + None, + ) + return None, None + name = self.engagement.path.name + for program, targets in found: + problems = self._problems(targets) + if not problems: + continue + reason = f"{program}: {'; '.join(problems)} ({name})" + if targets.tier == ACTIVE: + return f"engagement scope: {reason}", None + return None, f"Outside the engagement: {reason}" + return None, None + + def refusal(self, step): + return self._judge(step)[0] + + def notice(self, step): + return self._judge(step)[1] + + def activate(self) -> None: + from codebud import safety + + safety.add_step_check(self.refusal) + safety.add_step_notice(self.notice) + + def deactivate(self) -> None: + from codebud import safety + + safety.remove_step_check(self.refusal) + safety.remove_step_notice(self.notice) diff --git a/tests/test_engagement.py b/tests/test_engagement.py index db33945..4374c92 100644 --- a/tests/test_engagement.py +++ b/tests/test_engagement.py @@ -144,3 +144,21 @@ def test_a_fully_qualified_exclusion_still_excludes(tmp_path): assert not eng.covers("prod.example.com") assert not eng.covers("prod.example.com.") assert eng.out_of_scope_reason("prod.example.com") + + +def test_covers_network(tmp_path): + import ipaddress + + from codebud_sec.engagement import load + + path = tmp_path / "e.toml" + path.write_text( + 'in_scope = ["10.0.0.0/16", "app.example.com"]\nout_of_scope = ["10.0.5.0/24", "10.0.9.9"]\n' + ) + engagement = load(path) + net = ipaddress.ip_network + assert engagement.covers_network(net("10.0.1.0/24")) + assert not engagement.covers_network(net("10.0.0.0/8")) # wider than the scope + assert not engagement.covers_network(net("10.0.5.0/25")) # excluded + assert not engagement.covers_network(net("10.0.9.0/24")) # holds an excluded address + assert not engagement.covers_network(net("192.168.0.0/24")) diff --git a/tests/test_layers.py b/tests/test_layers.py index f9f5b3d..813ebb3 100644 --- a/tests/test_layers.py +++ b/tests/test_layers.py @@ -17,6 +17,8 @@ CORE_API = { "codebud.crypto", # encryption at rest: the evidence log (D14b) "codebud.fsutil", + "codebud.privacy", # the scope guard refuses an engagement at privacy trusted + "codebud.safety", # the scope guard registers its step check (D14i, PR 4) } diff --git a/tests/test_scope.py b/tests/test_scope.py new file mode 100644 index 0000000..1837810 --- /dev/null +++ b/tests/test_scope.py @@ -0,0 +1,230 @@ +""" +tests/test_scope.py + +The engagement scope guard (phase 10, step 5 plan, PR 4; D14i): every +target of a network tool is read from its command line and checked; +active tools out of scope are refused, passive lookups ask; a target +that cannot be read is refused. +""" + +import ipaddress +import shlex + +import pytest + +from codebud import safety +from codebud.safety import SafetyEngine +from codebud_sec.engagement import load +from codebud_sec.scope import ACTIVE, PASSIVE, ScopeGuard, TargetError, targets_of + +SCOPE = """\ +client = "ACME" +in_scope = ["10.0.0.0/16", "app.example.com", "2001:db8::/48"] +out_of_scope = ["10.0.5.0/24", "10.0.9.9"] +""" + + +@pytest.fixture +def engagement(tmp_path): + path = tmp_path / "acme.toml" + path.write_text(SCOPE) + return load(path) + + +@pytest.fixture +def guard(engagement, monkeypatch): + monkeypatch.setattr(safety, "_STEP_CHECKS", []) + monkeypatch.setattr(safety, "_STEP_NOTICES", []) + monkeypatch.delenv("CODEBUD_PRIVACY", raising=False) + g = ScopeGuard(engagement) + g.activate() + return g + + +def _targets(cmd): + [(program, targets)] = targets_of(shlex.split(cmd)) + return targets + + +def _hosts(cmd): + return sorted(_targets(cmd).hosts) + + +def _nets(cmd): + return [str(n) for n in _targets(cmd).networks] + + +# --- reading targets ------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("cmd", "hosts"), + [ + ("nmap -sV -p 22,80 -T4 10.0.0.5", ["10.0.0.5"]), + ("nmap -Pn -oN out.txt --top-ports 100 app.example.com", ["app.example.com"]), + ("nmap -p80 -sS --open 10.0.0.5 10.0.0.6", ["10.0.0.5", "10.0.0.6"]), + ("nmap --dns-servers 8.8.8.8 10.0.0.5", ["10.0.0.5", "8.8.8.8"]), + ("nmap -sI zombie.example.org:80 10.0.0.5", ["10.0.0.5", "zombie.example.org"]), + ("nmap -b anon@ftp.example.org:21 10.0.0.5", ["10.0.0.5", "ftp.example.org"]), + ("dig @1.1.1.1 app.example.com MX +short", ["1.1.1.1", "app.example.com"]), + ("dig -x 10.0.0.5", ["10.0.0.5"]), + ("host app.example.com 10.0.0.53", ["10.0.0.53", "app.example.com"]), + ("nc -zv 10.0.0.5 22", ["10.0.0.5"]), + ("nc -lvnp 4444", []), # listening contacts no one + ("nc -x proxy.example.org:1080 10.0.0.5 80", ["10.0.0.5", "proxy.example.org"]), + ("ncat --proxy 10.0.0.7:3128 10.0.0.5 80", ["10.0.0.5", "10.0.0.7"]), + ], +) +def test_targets_are_read_including_third_parties(cmd, hosts): + assert _hosts(cmd) == sorted(hosts) + + +@pytest.mark.parametrize( + ("cmd", "networks"), + [ + ("nmap 10.0.1.0/24", ["10.0.1.0/24"]), + ("nmap 10.0.0.1-5", ["10.0.0.1/32", "10.0.0.2/31", "10.0.0.4/31"]), + ("masscan -p80 --rate 1000 10.0.2.0-10.0.2.255", ["10.0.2.0/24"]), + ("masscan -p80 --range 10.0.3.0/24", ["10.0.3.0/24"]), + ], +) +def test_ranges_are_read_as_networks(cmd, networks): + assert _nets(cmd) == networks + + +def test_an_octet_range_is_judged_by_its_bounds(): + """10.0.0-3.1-254 names 4 x 254 addresses; their bounding range, + 10.0.0.1 - 10.0.3.254, holds them all, so judging it errs strict.""" + nets = _targets("nmap 10.0.0-3.1-254").networks + first, last = nets[0].network_address, nets[-1].broadcast_address + assert (str(first), str(last)) == ("10.0.0.1", "10.0.3.254") + + +@pytest.mark.parametrize( + ("cmd", "why"), + [ + ("nmap -iR 100", "random hosts"), + ("nmap --resume scan.log", "a file"), + ("nmap --frobnicate x 10.0.0.5", "unknown option"), + ("nmap -sV", "no target"), + ("nmap 'not a host!'", "not a target"), + ("masscan -c scan.conf", "settings file"), + ("dig -f queries.txt", "batch file"), + ("nc -Z 10.0.0.5 80", "unknown option"), + ("nmap -iL /etc/hosts", "a list outside the project"), + ], +) +def test_what_cannot_be_read_is_an_error(cmd, why): + with pytest.raises(TargetError): + targets_of(shlex.split(cmd)) + + +def test_a_target_list_in_the_project_is_read(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "hosts.txt").write_text("# web tier\n10.0.0.5\napp.example.com 10.0.1.0/24\n") + targets = _targets("nmap -iL hosts.txt") + assert sorted(targets.hosts) == ["10.0.0.5", "app.example.com"] + assert targets.networks == [ipaddress.ip_network("10.0.1.0/24")] + + +def test_tiers(): + assert _targets("dig app.example.com").tier == PASSIVE + assert _targets("nmap 10.0.0.5").tier == ACTIVE + + +def test_tools_are_found_behind_wrappers_and_by_variant_name(): + assert [p for p, _ in targets_of(shlex.split("timeout 60 nmap 10.0.0.5"))] == ["nmap"] + assert [p for p, _ in targets_of(shlex.split("nc.openbsd 10.0.0.5 80"))] == ["nc"] + + +# --- the guard, through the safety engine ---------------------------------------- + + +def _engine(): + engine = SafetyEngine() + engine.harmless_commands = ["dig", "host", "ls"] + engine.powerful_commands = ["nmap", "masscan", "nc", "ncat", "timeout", "curl"] + return engine + + +def _step(cmd, network=True): + return {"tool": "command", "args": {"cmd": cmd, "network": network}} + + +@pytest.mark.parametrize( + "cmd", + [ + "nmap -sV 10.0.0.5", + "nmap -sn 10.0.1.0/24", + "nmap app.example.com", + "nmap 2001:db8::5", + "masscan -p443 10.0.2.0/24", + "nc -zv 10.0.0.5 22", + "timeout 600 nmap 10.0.0.5", + ], +) +def test_in_scope_active_steps_run(guard, cmd): + assert _engine().refusal_reason(_step(cmd)) is None, cmd + + +@pytest.mark.parametrize( + ("cmd", "why"), + [ + ("nmap 192.168.1.1", "outside"), + ("nmap 10.0.5.20", "excluded network"), + ("nmap 10.0.9.9", "excluded address"), + ("nmap 10.0.0.0/8", "wider than the scope"), + ("nmap 10.0.9.0/24", "holds an excluded address"), + ("nmap evil.example.org", "a name not in scope"), + ("nmap --dns-servers 8.8.8.8 10.0.0.5", "a DNS server outside"), + ("nmap -sI zombie.example.org 10.0.0.5", "an idle-scan zombie outside"), + ("nc -x proxy.example.org:1080 10.0.0.5 80", "a proxy outside"), + ("timeout 60 nmap 192.168.1.1", "behind a wrapper"), + ("nmap -iR 10", "random targets"), + ("nmap --weird 10.0.0.5", "unreadable"), + ], +) +def test_out_of_scope_active_steps_are_refused(guard, cmd, why): + reason = _engine().refusal_reason(_step(cmd)) + assert reason is not None and "engagement scope" in reason, f"{cmd} ({why})" + + +def test_passive_lookups_out_of_scope_ask_instead(guard): + engine = _engine() + step = _step("dig evil.example.org") + assert engine.refusal_reason(step) is None + notice = engine.confirmation_reason(step) + assert notice is not None and "evil.example.org" in notice + # In scope: only the usual question for a network step, no scope warning. + in_scope = engine.confirmation_reason(_step("dig app.example.com")) + assert in_scope is None or "Outside the engagement" not in in_scope + + +def test_an_unparsed_network_tool_is_refused(guard): + reason = _engine().refusal_reason(_step("curl https://app.example.com/")) + assert reason is not None and "cannot read its targets" in reason + + +def test_steps_without_network_tools_are_untouched(guard): + assert _engine().refusal_reason(_step("ls -la", network=False)) is None + + +def test_a_closed_window_refuses_active_steps(tmp_path, monkeypatch): + path = tmp_path / "late.toml" + path.write_text(SCOPE + "starts = 2020-01-01T00:00:00Z\nends = 2020-01-02T00:00:00Z\n") + monkeypatch.setattr(safety, "_STEP_CHECKS", []) + monkeypatch.setattr(safety, "_STEP_NOTICES", []) + ScopeGuard(load(path)).activate() + reason = _engine().refusal_reason(_step("nmap 10.0.0.5")) + assert reason is not None and "window" in reason + + +def test_privacy_trusted_refuses_every_command(guard, monkeypatch): + monkeypatch.setenv("CODEBUD_PRIVACY", "trusted") + reason = _engine().refusal_reason(_step("ls", network=False)) + assert reason is not None and "trusted" in reason + + +def test_deactivate_removes_the_guard(guard): + guard.deactivate() + assert _engine().refusal_reason(_step("nmap 192.168.1.1")) is None From 57c4e7099df73796ea6a42595c8fba412b897f86 Mon Sep 17 00:00:00 2001 From: Mohammadreza Khellat Date: Thu, 12 Nov 2026 12:13:09 +0330 Subject: [PATCH 3/5] fix(scope): check only network steps, and require isolation Review of PR #18, findings 1 and 2. The guard read targets from every word naming a tool, in every command step, so data was taken for a command: `grep host /etc/hosts` and `grep -c nmap auth.log` were refused, "host" and "nmap" read as the tools. A step that does not declare the network contacts no one when it is isolated, so only network steps are checked now. That relies on isolation, which the guard did not require: at privacy private with no isolation tool, codebud warns and runs steps unisolated, and an undeclared step then has the network. An engagement now refuses every command step unless isolation is available (and the privacy level is not trusted). CORE_API widens by codebud.isolation, deliberately. The tests now decide isolation for themselves instead of inheriting the machine's (macOS CI has no bubblewrap); the suite was also run with isolation forced unavailable. --- src/codebud_sec/scope.py | 31 ++++++++++++++++++------------- tests/test_layers.py | 1 + tests/test_scope.py | 28 +++++++++++++++++++++++++++- 3 files changed, 46 insertions(+), 14 deletions(-) diff --git a/src/codebud_sec/scope.py b/src/codebud_sec/scope.py index d7e1357..fc4a30f 100644 --- a/src/codebud_sec/scope.py +++ b/src/codebud_sec/scope.py @@ -21,8 +21,11 @@ never matched against addresses (that would mean resolving it, itself a network action). -A step that declares the network with a tool that has no parser here is -refused. The guard is registered with the core's safety engine +Only steps that declare the network are checked: under isolation a step +without it contacts no one (and reading "host" in a log is not running +it). So an engagement needs isolation; without it every command step is +refused. A step that declares the network with a tool that has no +parser here is refused. The guard is registered with the core's safety engine (codebud.safety.add_step_check / add_step_notice) by whoever opens the engagement; the core never imports this layer. """ @@ -417,28 +420,30 @@ def _judge(self, step) -> tuple[str | None, str | None]: """(refusal, notice) for ``step``.""" if step.get("tool") != "command": return None, None - from codebud import privacy + from codebud import isolation, privacy - if privacy.level() == "trusted": + # A step without the network can reach no one -- only if steps really + # are isolated; otherwise an undeclared step has the network too. + if privacy.level() == "trusted" or isolation.available() is None: return ( - "an engagement needs isolated steps: privacy level trusted runs them " - "unisolated, with the whole network", + "an engagement needs isolated steps (bubblewrap or user namespaces, " + "at privacy private or locked); without them any step has the network", None, ) args = step.get("args", {}) + if not args.get("network"): + return None, None # isolated without the network: contacts no one try: words = shlex.split(args.get("cmd", "")) found = targets_of(words) except (ValueError, TargetError) as exc: return f"engagement scope: {exc}", None if not found: - if args.get("network"): - return ( - f"engagement scope: {words[0] if words else 'this step'} needs the network, " - "but codebud cannot read its targets", - None, - ) - return None, None + return ( + f"engagement scope: {words[0] if words else 'this step'} needs the network, " + "but codebud cannot read its targets", + None, + ) name = self.engagement.path.name for program, targets in found: problems = self._problems(targets) diff --git a/tests/test_layers.py b/tests/test_layers.py index 813ebb3..531f4d9 100644 --- a/tests/test_layers.py +++ b/tests/test_layers.py @@ -17,6 +17,7 @@ CORE_API = { "codebud.crypto", # encryption at rest: the evidence log (D14b) "codebud.fsutil", + "codebud.isolation", # the scope guard needs isolated steps (review of #18) "codebud.privacy", # the scope guard refuses an engagement at privacy trusted "codebud.safety", # the scope guard registers its step check (D14i, PR 4) } diff --git a/tests/test_scope.py b/tests/test_scope.py index 1837810..9e5a4e9 100644 --- a/tests/test_scope.py +++ b/tests/test_scope.py @@ -36,6 +36,9 @@ def guard(engagement, monkeypatch): monkeypatch.setattr(safety, "_STEP_CHECKS", []) monkeypatch.setattr(safety, "_STEP_NOTICES", []) monkeypatch.delenv("CODEBUD_PRIVACY", raising=False) + # Steps count as isolated here whatever the machine has (macOS CI has + # no bubblewrap); tests of the no-isolation case set it back. + monkeypatch.setattr("codebud.isolation.available", lambda: "bwrap") g = ScopeGuard(engagement) g.activate() return g @@ -214,6 +217,7 @@ def test_a_closed_window_refuses_active_steps(tmp_path, monkeypatch): path.write_text(SCOPE + "starts = 2020-01-01T00:00:00Z\nends = 2020-01-02T00:00:00Z\n") monkeypatch.setattr(safety, "_STEP_CHECKS", []) monkeypatch.setattr(safety, "_STEP_NOTICES", []) + monkeypatch.setattr("codebud.isolation.available", lambda: "bwrap") ScopeGuard(load(path)).activate() reason = _engine().refusal_reason(_step("nmap 10.0.0.5")) assert reason is not None and "window" in reason @@ -222,9 +226,31 @@ def test_a_closed_window_refuses_active_steps(tmp_path, monkeypatch): def test_privacy_trusted_refuses_every_command(guard, monkeypatch): monkeypatch.setenv("CODEBUD_PRIVACY", "trusted") reason = _engine().refusal_reason(_step("ls", network=False)) - assert reason is not None and "trusted" in reason + assert reason is not None and "isolated steps" in reason def test_deactivate_removes_the_guard(guard): guard.deactivate() assert _engine().refusal_reason(_step("nmap 192.168.1.1")) is None + + +# --- review of #18 ----------------------------------------------------------------- + + +@pytest.mark.parametrize("cmd", ["grep host /etc/hosts", "grep -c nmap auth.log", "cat nc.log"]) +def test_tool_names_as_data_in_steps_without_network_are_fine(guard, monkeypatch, cmd): + """An isolated step without the network contacts no one; reading "host" + or "nmap" in a log is not running them.""" + monkeypatch.setattr("codebud.isolation.available", lambda: "bwrap") + engine = _engine() + engine.harmless_commands += ["grep", "cat"] + assert engine.refusal_reason(_step(cmd, network=False)) is None, cmd + + +@pytest.mark.parametrize("kind", [None]) +def test_an_engagement_needs_isolation(guard, monkeypatch, kind): + """Without isolation an undeclared step has the network too, so the + guard could not trust the declaration: every command is refused.""" + monkeypatch.setattr("codebud.isolation.available", lambda: kind) + reason = _engine().refusal_reason(_step("ls", network=False)) + assert reason is not None and "isolat" in reason From f1fe0fd7445807040367fd882c11bcd7810d7e2c Mon Sep 17 00:00:00 2001 From: Mohammadreza Khellat Date: Thu, 12 Nov 2026 12:15:21 +0330 Subject: [PATCH 4/5] fix(scope): refuse nmap scripts that reach past the command line Review of PR #18, finding 3. nmap's scripts can contact hosts the command line never names, and the guard let them through: - broadcast-* scripts send to the whole local network; they are in the broadcast and discovery categories and partly in safe, so --script safe or --script discovery runs them too; - targets-* scripts with --script-args newtargets add the hosts they find as scan targets; - external scripts send data to third-party services (whois, ASN). --script is refused when any name, category or glob in it could select these -- broadcast, discovery, safe, external, all, broadcast*, targets-*, or a glob matching them -- in a list or a boolean expression (it errs strict: "not broadcast" is refused too, since expressions are not evaluated); --script-args with newtargets is refused. -sC and named scripts such as http-title stay allowed. Eleven cases, all allowed before. --- src/codebud_sec/scope.py | 34 ++++++++++++++++++++++++++++++++++ tests/test_scope.py | 36 ++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/src/codebud_sec/scope.py b/src/codebud_sec/scope.py index fc4a30f..11bfa38 100644 --- a/src/codebud_sec/scope.py +++ b/src/codebud_sec/scope.py @@ -30,6 +30,7 @@ engagement; the core never imports this layer. """ +import fnmatch import ipaddress import os import re @@ -244,6 +245,31 @@ def _host(args): r"^-p[0-9TUS:,\-]+$") # fmt: skip +# nmap scripts that reach past the command line: broadcast-* send to the +# whole local network (they are in the broadcast, discovery and partly +# safe categories), targets-* with newtargets add the hosts they find as +# targets, external ones contact third-party services. Selected by name, +# category, glob or boolean expression -- refused on any mention. +_NMAP_WIDE_CATEGORIES = {"broadcast", "discovery", "safe", "external", "all"} +_NMAP_WIDE_SAMPLES = ("broadcast-ping", "targets-sniffer", "whois-ip") + + +def _nmap_script_problem(expression: str) -> str | None: + for token in re.split(r"[\s,()]+", expression.lower()): + if not token or token in ("and", "or", "not", "default"): + continue + if token in _NMAP_WIDE_CATEGORIES: + return f"nmap --script {token} reaches hosts past the command line" + if token.startswith(("broadcast", "targets-")): + return f"nmap --script {token} reaches hosts past the command line" + if any(c in token for c in "*?[") and any( + fnmatch.fnmatch(sample, token) + for sample in (*_NMAP_WIDE_SAMPLES, *_NMAP_WIDE_CATEGORIES) + ): + return f"nmap --script {token} could select scripts that reach past the command line" + return None + + def _nmap(args): targets = Targets(ACTIVE) for option, word in _walk(args, _NMAP_VALUE, _NMAP_FLAGS, _NMAP_PATTERNS): @@ -251,6 +277,14 @@ def _nmap(args): raise TargetError("nmap -iR scans random hosts, outside any scope") if option in ("--resume", "--script-args-file"): raise TargetError(f"nmap {option} reads from a file that cannot be judged") + if option == "--script": + problem = _nmap_script_problem(word) + if problem: + raise TargetError(problem) + if option == "--script-args" and "newtargets" in word.lower(): + raise TargetError( + "nmap --script-args newtargets adds the hosts scripts find as targets" + ) if option == "-iL": _list_file(word, targets) elif option == "--dns-servers": diff --git a/tests/test_scope.py b/tests/test_scope.py index 9e5a4e9..feabd77 100644 --- a/tests/test_scope.py +++ b/tests/test_scope.py @@ -254,3 +254,39 @@ def test_an_engagement_needs_isolation(guard, monkeypatch, kind): monkeypatch.setattr("codebud.isolation.available", lambda: kind) reason = _engine().refusal_reason(_step("ls", network=False)) assert reason is not None and "isolat" in reason + + +@pytest.mark.parametrize( + "cmd", + [ + "nmap --script broadcast-dhcp-discover 10.0.0.5", + "nmap --script=broadcast 10.0.0.5", + "nmap --script discovery 10.0.0.5", # holds the broadcast-* scripts + "nmap --script safe 10.0.0.5", # so does safe + "nmap --script external 10.0.0.5", # contacts third-party services + "nmap --script all 10.0.0.5", + "nmap --script 'b*' 10.0.0.5", + "nmap --script 'default or safe' 10.0.0.5", + "nmap --script targets-sniffer 10.0.0.5", + "nmap --script http-title --script-args newtargets 10.0.0.5", + "nmap --script-args=newtargets,maxdepth=2 --script http-title 10.0.0.5", + ], +) +def test_nmap_scripts_that_reach_past_the_command_line_are_refused(guard, cmd): + """Review of #18: broadcast scripts send to the whole local network, + targets-* with newtargets add the hosts they discover, external ones + contact third-party services.""" + reason = _engine().refusal_reason(_step(cmd)) + assert reason is not None and "engagement scope" in reason, cmd + + +@pytest.mark.parametrize( + "cmd", + [ + "nmap -sC 10.0.0.5", + "nmap --script http-title,ssl-cert 10.0.0.5", + "nmap --script default 10.0.0.5", + ], +) +def test_ordinary_scripts_run(guard, cmd): + assert _engine().refusal_reason(_step(cmd)) is None, cmd From 098f3f8d1023c5dbc9bfc26cdfc61853c2566685 Mon Sep 17 00:00:00 2001 From: Mohammadreza Khellat Date: Thu, 12 Nov 2026 12:16:26 +0330 Subject: [PATCH 5/5] fix(scope): refuse web search under an engagement Re-review of PR #18. The guard judged only command steps, but web_search is a network step too: it sends the query to DuckDuckGo, a third party outside any engagement. Web search is off by default and the security profiles keep it off, but the guard should not depend on a setting; it now refuses web_search steps. File tools and host_network stay on this machine and are untouched (tested). --- src/codebud_sec/scope.py | 8 +++++++- tests/test_scope.py | 15 +++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/src/codebud_sec/scope.py b/src/codebud_sec/scope.py index 11bfa38..7731dbd 100644 --- a/src/codebud_sec/scope.py +++ b/src/codebud_sec/scope.py @@ -452,8 +452,14 @@ def _problems(self, targets: Targets) -> list[str]: def _judge(self, step) -> tuple[str | None, str | None]: """(refusal, notice) for ``step``.""" + if step.get("tool") == "web_search": + return ( + "engagement scope: web search sends the query to a third party " + "(DuckDuckGo), outside any engagement", + None, + ) if step.get("tool") != "command": - return None, None + return None, None # file tools and host_network stay on this machine from codebud import isolation, privacy # A step without the network can reach no one -- only if steps really diff --git a/tests/test_scope.py b/tests/test_scope.py index feabd77..7a778c8 100644 --- a/tests/test_scope.py +++ b/tests/test_scope.py @@ -290,3 +290,18 @@ def test_nmap_scripts_that_reach_past_the_command_line_are_refused(guard, cmd): ) def test_ordinary_scripts_run(guard, cmd): assert _engine().refusal_reason(_step(cmd)) is None, cmd + + +def test_web_search_is_refused_under_an_engagement(guard): + """Re-review of #18: it sends the query to a third party (DuckDuckGo), + outside any scope, whatever the web search setting.""" + reason = _engine().refusal_reason({"tool": "web_search", "args": {"query": "acme vpn"}}) + assert reason is not None and "engagement" in reason + + +def test_local_tools_are_untouched(guard): + for step in ( + {"tool": "file_read", "args": {"path": "notes.txt"}}, + {"tool": "host_network", "args": {}}, + ): + assert _engine().refusal_reason(step) is None, step