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 1b8a9d8..d44c752 100644 --- a/docs/chapters/development.texi +++ b/docs/chapters/development.texi @@ -165,9 +165,13 @@ 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) +├── 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/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/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/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..7731dbd --- /dev/null +++ b/src/codebud_sec/scope.py @@ -0,0 +1,514 @@ +""" +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). + +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. +""" + +import fnmatch +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 + + +# 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): + 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 == "--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": + 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") == "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 # 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 + # 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 (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: + 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) + 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..531f4d9 100644 --- a/tests/test_layers.py +++ b/tests/test_layers.py @@ -17,6 +17,9 @@ 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 new file mode 100644 index 0000000..7a778c8 --- /dev/null +++ b/tests/test_scope.py @@ -0,0 +1,307 @@ +""" +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) + # 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 + + +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", []) + 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 + + +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 "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 + + +@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 + + +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 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