diff --git a/docs/chapters/architecture.texi b/docs/chapters/architecture.texi index 69d6614..5684108 100644 --- a/docs/chapters/architecture.texi +++ b/docs/chapters/architecture.texi @@ -569,7 +569,8 @@ src/codebud/ the package ├── executor.py Executor (step-by-step runner) ├── privacy.py privacy level: locked / private / trusted ├── profiles.py named profiles: mode + privacy (+ policy, prompt) -├── isolation.py bubblewrap / unshare wrapping for command steps +├── isolation.py bubblewrap / unshare wrapping for command steps; +│ the network fence (pasta + nftables) ├── confine.py where file tools may read and write ├── redact.py mask secrets before sessions/history are saved ├── safety.py SafetyEngine (policy validation) @@ -641,6 +642,8 @@ src/codebud_sec/ the security layer (authorized security work) │ command line and checked (DNS tools, whois, │ scanners, nc, web clients, fuzzers, TLS scanners); │ registered with the core's safety.add_step_check() +├── fence.py network fence policy: which addresses a step +│ may reach (its targets, names resolved per step) └── evidence.py hash-chained, append-only evidence log, encrypted at rest (appends need no passphrase) diff --git a/docs/chapters/configuration.texi b/docs/chapters/configuration.texi index 99037cb..b1e5105 100644 --- a/docs/chapters/configuration.texi +++ b/docs/chapters/configuration.texi @@ -502,7 +502,15 @@ proxies, DNS servers, addresses given to @command{curl --resolve} --- and refuses the step when one is out of scope, excluded, or the engagement window is closed. A command it cannot read (an unknown option, a target list outside the project, an ambiguous URL) is refused -too. +too. Scanning steps also run behind the network fence (@pxref{Isolated +Steps}), which lets through only the addresses the step names --- its +names resolved just before it runs --- and never an excluded one, so a +tool that reaches further than its command line says meets a closed +door. Behind the fence nmap needs @option{--unprivileged} (connect +scans), and SYN, UDP and OS scans and @command{masscan} are refused; an +engagement file with @code{fence_scanning = false} gives them back, with +the scope guard as their only check. Lookups (@command{dig}, +@command{host}, @command{whois}) are not fenced. Its allow-list: reading results runs once the plan is approved (@command{cat}, @command{grep}, @command{rg}, @command{jq}, @@ -697,6 +705,41 @@ Install bubblewrap with your package manager (@command{pacman -S bubblewrap}, @command{apt install bubblewrap}, @command{dnf install bubblewrap}). +@subheading The network fence +@cindex network fence +@cindex pasta +@cindex nftables + +A step that declares the network normally shares the machine's network. +An optional layer --- the security layer, while an engagement is open +--- can instead have a network step @dfn{fenced}: it then runs in a +network namespace of its own whose only way out is @command{pasta} +(from the passt package), and @command{nft} rules in that namespace let +through only the addresses the layer allows for that step, refusing +excluded ranges first. + +@itemize +@item Nothing of the machine itself is reachable: not its loopback, not +through the gateway address, and no port is forwarded in either +direction (pasta runs with @option{--no-map-gw -t none -u none -T none -U +none}; its defaults would reach the machine's own services). +@item The step runs without capabilities (@code{bwrap --cap-drop ALL}), +in a user namespace of its own, so it cannot change the rules --- and +has no raw sockets, whose packets would pass them by. Scans that need +raw sockets (nmap's SYN, UDP and OS scans, masscan) cannot run fenced; +nmap needs @option{--unprivileged}. +@item There is no DNS: the names the step uses are resolved before it +runs and written into an @file{/etc/hosts} of its own. Lookups through +a resolver would carry data to whoever serves a name. +@end itemize + +A step that must be fenced is refused when the fence cannot be built: +without bubblewrap, @command{pasta} or @command{nft}, or at privacy +@code{trusted}. @command{codebud integrity pin} pins @command{pasta}, +@command{nft} and @command{sh} too when the security layer is installed. +Install them with @command{pacman -S passt nftables}, @command{apt +install passt nftables} or @command{dnf install passt nftables}. + @node Feature Flags @section Feature Flags diff --git a/docs/chapters/development.texi b/docs/chapters/development.texi index 64ea61e..cc11514 100644 --- a/docs/chapters/development.texi +++ b/docs/chapters/development.texi @@ -72,6 +72,8 @@ tests/ ├── test_evidence.py evidence log: hash chain, tamper │ detection, artefact hashes, encrypted │ entries (append while locked) +├── test_fence.py the network fence: rules, pasta options, +│ no DNS, refusals (live where available) ├── test_forensic_rules.py dc3dd/ewfacquire on images in the project, │ hashed and logged; exiftool only reads ├── test_fsutil.py atomic_write_text() / _bytes() @@ -172,6 +174,9 @@ tests/ │ third parties, ranges, active refused, │ passive asked; nmap scripts judged by │ nmap's script.db +├── test_scope_fence.py engagement fence: a scan reaches exactly +│ its targets, names resolved per step, raw +│ scans refused, fence_scanning = false ├── test_scope_web.py engagement scope for web and TLS tools: │ URL hosts read strictly, proxies, redirects │ and fuzz words in the host refused diff --git a/src/Makefile.am b/src/Makefile.am index 3687ac6..f042f93 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -63,6 +63,7 @@ nobase_python_PYTHON = \ codebud_sec/__init__.py \ codebud_sec/engagement.py \ codebud_sec/evidence.py \ + codebud_sec/fence.py \ codebud_sec/scope.py nobase_dist_python_DATA = \ diff --git a/src/codebud/cli/integrity.py b/src/codebud/cli/integrity.py index da8bfc6..8d78ccd 100644 --- a/src/codebud/cli/integrity.py +++ b/src/codebud/cli/integrity.py @@ -76,13 +76,15 @@ def _shipped_programs() -> set[str]: def _programs(extra) -> set[str]: - from codebud import isolation, policy + from codebud import isolation, policy, profiles programs = integrity.allow_listed_programs(policy.effective()) | set(extra) programs |= _shipped_programs() tool = isolation.available() if tool: programs.add(tool) # it runs every isolated step + if profiles.shipped_profile_dir() is not None and isolation.fence_available(): + programs |= set(isolation.FENCE_PROGRAMS) # the security layer's network fence return programs diff --git a/src/codebud/isolation.py b/src/codebud/isolation.py index 52455a0..550cd2d 100644 --- a/src/codebud/isolation.py +++ b/src/codebud/isolation.py @@ -21,12 +21,33 @@ but nothing else. Where neither works (macOS, BSD, containers, blocked user namespaces), steps are not isolated: privacy level "private" warns once and runs them, "locked" refuses them. "trusted" never isolates. + +A network step can be fenced (phase 10, step 5 plan, PR 6): an optional +layer registers a function (add_network_fence) that says which addresses +the step may reach, and the step then runs in a network namespace of its +own whose only way out is pasta (passt), with nftables letting through +only those addresses: + +- pasta --no-map-gw -t none -u none -T none -U none: nothing of the host + is reachable -- not its loopback, not through the gateway address, no + port forwarded either way (pasta's defaults reach the host's loopback); +- nft loads the rules in that namespace (pasta's user namespace gives + CAP_NET_ADMIN there) before bwrap starts the step with --cap-drop ALL in + a user namespace of its own, so the step cannot change them -- and has + no raw sockets, whose packets would pass by the rules; +- no DNS: the names the step uses are resolved before, into an /etc/hosts + of its own, and resolv.conf is empty -- DNS through the host's resolver + would carry data to whoever serves a name. """ +import atexit +import ipaddress import logging import os import shutil import subprocess +import tempfile +from dataclasses import dataclass from functools import lru_cache from pathlib import Path @@ -138,12 +159,20 @@ def _contains(directory: Path, path: Path) -> bool: def wrap( - argv: list[str], project: Path, network: bool, kind: str, observer: bool = False + argv: list[str], + project: Path, + network: bool, + kind: str, + observer: bool = False, + files: tuple[tuple[str, str], ...] = (), + no_capabilities: bool = False, ) -> list[str]: """``argv`` wrapped for ``kind`` of isolation. ``observer`` (bwrap only): the host's /proc instead of a fresh one, so the step sees the host's processes and, through /proc/1/net, its sockets -- while its own process - namespace keeps it from signalling them and it still has no network.""" + namespace keeps it from signalling them and it still has no network. + ``files`` (bwrap only): (host file, path in the step) bound read-only + over the step's view; ``no_capabilities``: --cap-drop ALL.""" if kind == "unshare": if network: return argv # unshare only cuts the network @@ -153,6 +182,8 @@ def wrap( wrapped = ["bwrap", "--die-with-parent", "--new-session", "--unshare-all"] if network: wrapped.append("--share-net") + if no_capabilities: + wrapped += ["--cap-drop", "ALL"] proc = ["--ro-bind", "/proc", "/proc"] if observer else ["--proc", "/proc"] wrapped += ["--ro-bind", "/", "/", "--dev", "/dev", *proc, "--tmpfs", "/tmp"] # After the /tmp tmpfs, so a project under /tmp is still there. @@ -163,41 +194,179 @@ def wrap( for name in HIDDEN_FILES: if (home / name).is_file(): wrapped += ["--ro-bind", "/dev/null", str(home / name)] + for source, target in files: + wrapped += ["--ro-bind", source, target] wrapped += ["--chdir", str(cwd), "--", *argv] return wrapped +# --- the network fence --------------------------------------------------------------- + +FENCE_PROGRAMS = ("pasta", "nft", "sh") +_PASTA_OPTIONS = ["--config-net", "--quiet", "--no-map-gw", "-t", "none", "-u", "none", + "-T", "none", "-U", "none"] # fmt: skip +# Run inside pasta's namespace: load the rules, then the step. +_FENCE_SCRIPT = 'printf %s "$1" | "$2" -f - || exit 97; shift 2; exec "$@"' + + +@dataclass(frozen=True) +class Fence: + """What a fenced step may reach: ``allow`` networks, never ``deny`` + networks (they win), and ``hosts``: (name, address) for its + /etc/hosts.""" + + allow: tuple = () + deny: tuple = () + hosts: tuple[tuple[str, str], ...] = () + + +def nft_ruleset(fence: Fence) -> str: + """The nftables rules for ``fence``: everything out is dropped but the + namespace's loopback, replies, and the allowed networks; denied + networks are dropped first, so they win over an allowed one around + them.""" + + def rule(network, verdict): + network = ipaddress.ip_network(network, strict=False) + family = "ip" if network.version == 4 else "ip6" + return f" {family} daddr {network} {verdict}" + + lines = [ + "table inet codebud_fence {", + " chain out {", + " type filter hook output priority 0; policy drop;", + ' oif "lo" accept', + " ct state established,related accept", + *(rule(n, "drop") for n in fence.deny), + *(rule(n, "accept") for n in fence.allow), + " }", + "}", + ] + return "\n".join(lines) + "\n" + + +def fence_available() -> bool: + return all(shutil.which(program) for program in FENCE_PROGRAMS) + + +_FENCES: list = [] +_fence_dir: Path | None = None + + +def add_network_fence(fence) -> None: + """Register ``fence(words)``: a Fence for a network step that must be + fenced, None for one that need not be, or a str: why it must not run.""" + _FENCES.append(fence) + + +def remove_network_fence(fence) -> None: + if fence in _FENCES: + _FENCES.remove(fence) + + +def _fence_for(words: list[str]): + for hook in list(_FENCES): + try: + decision = hook(words) + except Exception as exc: # noqa: BLE001 - a broken fence never lets a step out + return f"the network fence failed ({type(exc).__name__}: {exc})" + if decision is not None: + return decision + return None + + +def _fence_files(fence: Fence) -> tuple[tuple[str, str], ...]: + """A hosts file and an empty resolv.conf for the step, in a private + directory removed when codebud exits.""" + global _fence_dir + if _fence_dir is None: + _fence_dir = Path(tempfile.mkdtemp(prefix="codebud-fence-")) + atexit.register(shutil.rmtree, _fence_dir, True) + handle, hosts = tempfile.mkstemp(dir=_fence_dir, prefix="hosts-") + with os.fdopen(handle, "w") as f: + f.write("127.0.0.1 localhost\n::1 localhost\n") + f.writelines(f"{address} {name}\n" for name, address in fence.hosts) + resolv = _fence_dir / "resolv.conf" + resolv.write_text("# no DNS in a fenced step\n") + return ((hosts, "/etc/hosts"), (str(resolv), "/etc/resolv.conf")) + + +def fenced(inner: list[str], fence: Fence) -> list[str]: + """``inner`` (a bwrap command line) run behind ``fence``.""" + return ["pasta", *_PASTA_OPTIONS, "--", "sh", "-c", _FENCE_SCRIPT, "codebud-fence", + nft_ruleset(fence), "nft", *inner] # fmt: skip + + def prepare(argv: list[str], network: bool = False) -> tuple[list[str] | None, str | None]: """(argv to run, None) -- or (None, why the step must not run). Follows the privacy level: trusted runs ``argv`` as it is; private and locked isolate it where possible; without isolation private warns once - and runs it, locked refuses. Network steps are refused when locked. + and runs it, locked refuses. Network steps are refused when locked, and + fenced when a registered fence says so (which needs bwrap and pasta). """ + argv, refusal, _ = prepare_with_programs(argv, network) + return argv, refusal + + +def prepare_with_programs( + argv: list[str], network: bool = False +) -> tuple[list[str] | None, str | None, list[int]]: + """prepare(), and the positions in the result of the programs it added + (the isolation tool, the fence's), for the integrity check.""" global _warned from codebud import privacy, profiles from codebud.rag.indexer import project_root level = privacy.level() + fence = _fence_for(argv) if network else None + if isinstance(fence, str): + return None, f"Refused: {fence}", [] + if fence is not None: + if level == "trusted" or available() != "bwrap" or not fence_available(): + return ( + None, + ( + "this step must run behind the network fence, which needs bubblewrap, " + "pasta (passt) and nft, at privacy private or locked; not running it." + ), + [], + ) + if level == "locked": + return None, privacy.locked_refusal("a step that needs the network"), [] + inner = wrap(argv, project_root(), True, "bwrap", files=_fence_files(fence), + no_capabilities=True) # fmt: skip + outer = fenced(inner, fence) + start = len(outer) - len(inner) + sh, nft = outer.index("sh"), start - 1 + return outer, None, [0, sh, nft, start] if level == "trusted": - return argv, None + return argv, None, [] if network and level == "locked": - return None, privacy.locked_refusal("a step that needs the network") + return None, privacy.locked_refusal("a step that needs the network"), [] kind = available() # Observer isolation is bubblewrap's or nothing: anything less would # share the host's process namespace and could signal its processes. observer = profiles.active().isolation == "observer" if observer and kind != "bwrap": - return None, ( - "this profile's observer isolation needs bubblewrap: without it a step " - "would share the host's process namespace and could signal (kill) its " - "processes; not running it." + return ( + None, + ( + "this profile's observer isolation needs bubblewrap: without it a step " + "would share the host's process namespace and could signal (kill) its " + "processes; not running it." + ), + [], ) if kind is None: if level == "locked": - return None, ( - "privacy is locked and steps cannot be isolated here (install bubblewrap, " - "or allow user namespaces); not running it." + return ( + None, + ( + "privacy is locked and steps cannot be isolated here (install bubblewrap, " + "or allow user namespaces); not running it." + ), + [], ) if not _warned: _warned = True @@ -205,7 +374,7 @@ def prepare(argv: list[str], network: bool = False) -> tuple[list[str] | None, s "Steps are not isolated: bubblewrap / user namespaces are unavailable. " "Install bubblewrap for network-free, project-only steps." ) - return argv, None + return argv, None, [] if kind == "unshare" and network: # unshare can only cut the network; a step that needs it gets no # isolation at all here, so say so, as above (security review, low). @@ -217,4 +386,4 @@ def prepare(argv: list[str], network: bool = False) -> tuple[list[str] | None, s "unshare is available, and it can only cut the network. Install " "bubblewrap for project-only writes and hidden secrets." ) - return wrap(argv, project_root(), network, kind, observer=observer), None + return wrap(argv, project_root(), network, kind, observer=observer), None, [0] diff --git a/src/codebud/sandbox.py b/src/codebud/sandbox.py index 77d5eb9..35b0a71 100644 --- a/src/codebud/sandbox.py +++ b/src/codebud/sandbox.py @@ -188,11 +188,11 @@ def run_command(self, cmd: str, network: bool = False) -> dict[str, Any]: words, refusal = _pin_programs(words, [0, len(words) - len(_unwrap(words))]) if refusal: return {"stdout": "", "stderr": refusal, "returncode": 1} - argv, refusal = isolation.prepare(words, network=network) + argv, refusal, programs = isolation.prepare_with_programs(words, network=network) if refusal: return {"stdout": "", "stderr": refusal, "returncode": 1} - if argv[0] != words[0]: # the isolation tool - argv, refusal = _pin_programs(argv, [0]) + if programs: # the isolation tool, and the fence's programs + argv, refusal = _pin_programs(argv, programs) if refusal: return {"stdout": "", "stderr": refusal, "returncode": 1} diff --git a/src/codebud_sec/data/profiles/security-pentest.toml b/src/codebud_sec/data/profiles/security-pentest.toml index a6a95ae..c9f165f 100644 --- a/src/codebud_sec/data/profiles/security-pentest.toml +++ b/src/codebud_sec/data/profiles/security-pentest.toml @@ -18,9 +18,11 @@ engagement's scope; a step naming any other host is refused, so never guess or widen targets. Declare the network ("network": true) on every step that contacts a host. Reconnaissance and scanning only: no exploitation, no password guessing, no -denial of service. For nmap use -sV and named scripts, or --script -"default and not external" (plain -sC is refused). masscan at --rate 1000 -or less. Never follow redirects: curl without -L, wget --max-redirect=0, +denial of service. Scanning steps run behind a network fence: they reach +only the hosts they name, have no DNS beyond those names and no raw +sockets. So run nmap with --unprivileged -n (connect scans; no -sS, -sU or +-O) and -sV and named scripts, or --script "default and not external" +(plain -sC is refused). masscan cannot run behind the fence. Never follow redirects: curl without -L, wget --max-redirect=0, whatweb --follow-redirect=never. Fuzz only the path (https://host/FUZZ). nikto always with -Tuning x6 (its denial-of-service tests left out). Never send files to a target. diff --git a/src/codebud_sec/engagement.py b/src/codebud_sec/engagement.py index 388396c..8bab395 100644 --- a/src/codebud_sec/engagement.py +++ b/src/codebud_sec/engagement.py @@ -12,12 +12,12 @@ in-scope and out-of-scope targets (CIDRs, hosts, domains), and the time window the work is authorized for. -Scope handling is **warn, do not block** (D14): a target outside the scope -or the window raises a clear, logged warning and asks for confirmation, -but is allowed. The engagement file and every action are recorded -regardless, so the record is complete even when the operator proceeds. -Nothing here grants access; it only states, and helps keep to, what was -already authorized. +Scope handling (D14, changed by the step 5 plan's D14i): a target of an +active tool outside the scope, an excluded one, or outside the window is +refused; a passive lookup out of scope asks for confirmation +(codebud_sec/scope.py). Scanning steps also run behind the network fence +unless fence_scanning = false (codebud_sec/fence.py). Nothing here grants +access; it only states, and helps keep to, what was already authorized. """ import ipaddress @@ -34,7 +34,8 @@ _STRING_KEYS = ("client", "ticket", "operator", "notes") _LIST_KEYS = ("in_scope", "out_of_scope") _TIME_KEYS = ("starts", "ends") -_KEYS = (*_STRING_KEYS, *_LIST_KEYS, *_TIME_KEYS) +_BOOL_KEYS = ("fence_scanning",) +_KEYS = (*_STRING_KEYS, *_LIST_KEYS, *_TIME_KEYS, *_BOOL_KEYS) class EngagementError(Exception): @@ -87,6 +88,10 @@ class Engagement: out_of_scope: tuple[Target, ...] = () starts: datetime | None = None ends: datetime | None = None + # Scanning steps run behind the network fence (codebud_sec/fence.py); + # false gives raw scans (nmap -sS, masscan) back, judged by the scope + # guard alone. + fence_scanning: bool = True def header(self) -> str: """A one-line summary for the evidence log header.""" @@ -176,6 +181,9 @@ def load(path: str | Path) -> Engagement: for key in _LIST_KEYS: if key in data and not isinstance(data[key], list): raise EngagementError(f"{path}: '{key}' must be a list of strings") + for key in _BOOL_KEYS: + if key in data and not isinstance(data[key], bool): + raise EngagementError(f"{path}: '{key}' must be true or false") if not data.get("in_scope"): raise EngagementError(f"{path}: 'in_scope' is required and must list at least one target") @@ -189,4 +197,5 @@ def load(path: str | Path) -> Engagement: out_of_scope=_targets(data.get("out_of_scope", []), "out_of_scope", path), starts=_time(data["starts"], "starts", path) if "starts" in data else None, ends=_time(data["ends"], "ends", path) if "ends" in data else None, + fence_scanning=data.get("fence_scanning", True), ) diff --git a/src/codebud_sec/fence.py b/src/codebud_sec/fence.py new file mode 100644 index 0000000..52c3da7 --- /dev/null +++ b/src/codebud_sec/fence.py @@ -0,0 +1,132 @@ +""" +codebud_sec/fence.py + +The engagement's network fence (phase 10, step 5 plan, PR 6): which +addresses a network step may reach, for the core's fence (codebud. +isolation: pasta + nftables, no DNS, no capabilities). The scope guard +(scope.py) reads a step's targets from its command line; the fence makes +the network itself hold to them, so a tool that reaches further than its +command line says -- a redirect, a script, a bug, a plan that slipped +past a parser -- meets a closed door. + +Per step (the operator's decision, 2026-11-12): +- passive lookups (dig, host, whois) are not fenced: they need DNS or a + registry, and the scope guard alone judges them; +- scanning steps are fenced unless the engagement sets + fence_scanning = false (raw scans need raw sockets, which the fence + removes); +- a fenced step may reach exactly its targets: addresses and networks as + written, names resolved now (outside the fence) and written into the + step's /etc/hosts; the engagement's excluded networks are dropped + first. The client's DNS decides what a name resolves to, and an address + shared with other sites (a CDN) is reachable while the step runs. + +Under the fence nmap must run --unprivileged (connect scans), and scans +that need raw sockets -- SYN, UDP, OS detection, masscan -- are refused +with that reason rather than left to fail. +""" + +import ipaddress +import socket + +from codebud import isolation + +from .scope import PASSIVE, TargetError, targets_of + +# nmap options that need raw sockets. +_NMAP_RAW = {"-sS", "-sU", "-sA", "-sW", "-sM", "-sN", "-sF", "-sX", "-sY", "-sZ", "-sO", + "-sI", "-O", "-A", "--traceroute", "-f", "-ff", "--mtu", "-D", "-S", "--spoof-mac", + "--badsum", "--send-eth", "--send-ip", "--privileged"} # fmt: skip + + +def _raw_socket_problem(program: str, words: list[str]) -> str | None: + if program == "masscan": + return ( + "masscan needs raw sockets, which the network fence removes; " + "set fence_scanning = false in the engagement to run it" + ) + if program == "nmap": + raw = sorted(set(words) & _NMAP_RAW) + if raw: + return ( + f"nmap {' '.join(raw)} needs raw sockets, which the network fence removes; " + "use connect scans (-sT), or set fence_scanning = false in the engagement" + ) + if "--unprivileged" not in words: + return ( + "behind the network fence nmap needs --unprivileged " + "(it has no raw sockets there and would stop)" + ) + return None + + +def _resolve(name: str) -> list[str]: + infos = socket.getaddrinfo(name, None, type=socket.SOCK_STREAM) + return sorted({info[4][0] for info in infos}) + + +def _network(address: str): + return ipaddress.ip_network(address) + + +class ScopeFence: + """The fence for one engagement, to register with the core + (codebud.isolation.add_network_fence).""" + + def __init__(self, engagement): + self.engagement = engagement + + def _fenced(self, found) -> bool: + if not found or all(targets.tier == PASSIVE for _, targets in found): + return False + return self.engagement.fence_scanning + + def problem(self, words: list[str]) -> str | None: + """Why ``words`` cannot run behind the fence (raw sockets), or None: + for the scope guard, so a plan hears it before it runs.""" + try: + found = targets_of(words) + except TargetError: + return None # the scope guard reports it + if not self._fenced(found): + return None + for program, _ in found: + problem = _raw_socket_problem(program, words) + if problem: + return problem + return None + + def decide(self, words: list[str]): + """A Fence for ``words``, None (not fenced), or why it must not run.""" + try: + found = targets_of(words) + except TargetError as exc: + return f"engagement scope: {exc}" + if not self._fenced(found): + return None + problem = self.problem(words) + if problem: + return f"engagement scope: {problem}" + allow, hosts = set(), [] + for _, targets in found: + allow |= set(targets.networks) + for host in targets.hosts: + try: + allow.add(_network(host)) + continue + except ValueError: + pass # a name + try: + addresses = _resolve(host) + except OSError as exc: + return f"engagement scope: cannot resolve {host} for the network fence ({exc})" + allow |= {_network(a) for a in addresses} + hosts += [(host, a) for a in addresses] + deny = tuple(t.network for t in self.engagement.out_of_scope if t.network is not None) + return isolation.Fence(allow=tuple(sorted(allow, key=str)), deny=deny, hosts=tuple(hosts)) + + def activate(self) -> None: + isolation.add_network_fence(self.decide) + + def deactivate(self) -> None: + isolation.remove_network_fence(self.decide) diff --git a/src/codebud_sec/scope.py b/src/codebud_sec/scope.py index 08a421b..9210973 100644 --- a/src/codebud_sec/scope.py +++ b/src/codebud_sec/scope.py @@ -1177,7 +1177,10 @@ class ScopeGuard: safety engine (activate) and remove again (deactivate).""" def __init__(self, engagement): + from .fence import ScopeFence + self.engagement = engagement + self.fence = ScopeFence(engagement) def _problems(self, targets: Targets) -> list[str]: e = self.engagement @@ -1225,6 +1228,9 @@ def _judge(self, step) -> tuple[str | None, str | None]: "but codebud cannot read its targets", None, ) + fence_problem = self.fence.problem(words) + if fence_problem: + return f"engagement scope: {fence_problem}", None name = self.engagement.path.name for program, targets in found: problems = self._problems(targets) @@ -1247,9 +1253,11 @@ def activate(self) -> None: safety.add_step_check(self.refusal, provides="engagement") safety.add_step_notice(self.notice) + self.fence.activate() def deactivate(self) -> None: from codebud import safety safety.remove_step_check(self.refusal) safety.remove_step_notice(self.notice) + self.fence.deactivate() diff --git a/tests/test_fence.py b/tests/test_fence.py new file mode 100644 index 0000000..eb991d8 --- /dev/null +++ b/tests/test_fence.py @@ -0,0 +1,182 @@ +""" +tests/test_fence.py + +The network fence's mechanism in the core (phase 10, step 5 plan, PR 6): +a registered fence says which addresses a network step may reach, and +the step runs behind pasta and nftables -- nothing of the host +reachable, no DNS, no capabilities to change the rules. +""" + +import ipaddress +import shutil +import socket +import subprocess +import threading + +import pytest + +from codebud import isolation + +NET = ipaddress.ip_network + + +@pytest.fixture(autouse=True) +def no_fences(monkeypatch): + monkeypatch.setattr(isolation, "_FENCES", []) + monkeypatch.delenv("CODEBUD_PRIVACY", raising=False) + + +@pytest.fixture +def tools(monkeypatch): + """Isolation and the fence's programs count as present.""" + monkeypatch.setattr(isolation, "available", lambda: "bwrap") + monkeypatch.setattr(isolation, "fence_available", lambda: True) + + +FENCE = isolation.Fence( + allow=(NET("10.0.0.0/16"), NET("2001:db8::/48")), + deny=(NET("10.0.5.0/24"),), + hosts=(("app.example.com", "10.0.0.7"),), +) + + +def test_denied_networks_come_before_allowed_ones(): + """10.0.5.0/24 lies inside the allowed 10.0.0.0/16: dropped first, it wins.""" + rules = isolation.nft_ruleset(FENCE).splitlines() + deny = rules.index(" ip daddr 10.0.5.0/24 drop") + assert deny < rules.index(" ip daddr 10.0.0.0/16 accept") + assert " ip6 daddr 2001:db8::/48 accept" in rules + assert any("policy drop" in line for line in rules) + + +def test_a_fenced_step_runs_behind_pasta_with_nothing_of_the_host(tools, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + isolation.add_network_fence(lambda words: FENCE) + argv, refusal, programs = isolation.prepare_with_programs(["curl", "x"], network=True) + assert refusal is None + assert argv[0] == "pasta" + head = argv[: argv.index("--")] + for option in ("--no-map-gw", "--config-net"): + assert option in head + for flag in ("-t", "-u", "-T", "-U"): + assert head[head.index(flag) + 1] == "none" + inner = argv[programs[-1] :] + assert inner[0] == "bwrap" and inner[-2:] == ["curl", "x"] + assert inner[inner.index("--cap-drop") + 1] == "ALL" + assert "--share-net" in inner + # The programs to pin: pasta, sh, nft, bwrap. + assert [argv[i] for i in programs] == ["pasta", "sh", "nft", "bwrap"] + + +def test_a_fenced_step_gets_its_own_hosts_and_no_dns(tools, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + isolation.add_network_fence(lambda words: FENCE) + argv, _, programs = isolation.prepare_with_programs(["curl", "x"], network=True) + inner = argv[programs[-1] :] + binds = {inner[i + 2]: inner[i + 1] for i, w in enumerate(inner) if w == "--ro-bind"} + with open(binds["/etc/hosts"]) as f: + assert "10.0.0.7 app.example.com" in f.read() + with open(binds["/etc/resolv.conf"]) as f: + assert "nameserver" not in f.read() + + +def test_without_a_fence_nothing_changes(tools, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + isolation.add_network_fence(lambda words: None) + argv, refusal, programs = isolation.prepare_with_programs(["curl", "x"], network=True) + assert refusal is None and argv[0] == "bwrap" and programs == [0] + # A step without the network is never asked about. + asked = [] + isolation.add_network_fence(lambda words: asked.append(words)) + isolation.prepare_with_programs(["ls"], network=False) + assert asked == [] + + +@pytest.mark.parametrize( + "hook", [lambda words: "out of scope", lambda words: 1 / 0], ids=["refusal", "crash"] +) +def test_a_fence_refusal_or_failure_stops_the_step(tools, hook): + isolation.add_network_fence(hook) + argv, refusal, _ = isolation.prepare_with_programs(["curl", "x"], network=True) + assert argv is None and refusal.startswith("Refused") + + +@pytest.mark.parametrize( + ("setting", "value"), + [ + ("available", lambda: "unshare"), + ("available", lambda: None), + ("fence_available", lambda: False), + ], +) +def test_a_fence_that_cannot_be_built_refuses(monkeypatch, setting, value): + monkeypatch.setattr(isolation, "available", lambda: "bwrap") + monkeypatch.setattr(isolation, "fence_available", lambda: True) + monkeypatch.setattr(isolation, setting, value) + isolation.add_network_fence(lambda words: FENCE) + argv, refusal, _ = isolation.prepare_with_programs(["curl", "x"], network=True) + assert argv is None and "fence" in refusal + + +def test_privacy_trusted_does_not_skip_the_fence(tools, monkeypatch): + monkeypatch.setenv("CODEBUD_PRIVACY", "trusted") + isolation.add_network_fence(lambda words: FENCE) + argv, refusal, _ = isolation.prepare_with_programs(["curl", "x"], network=True) + assert argv is None and "fence" in refusal + + +def test_the_fence_programs_are_pinned(monkeypatch, tmp_path): + """codebud integrity pin covers them when the security layer is there.""" + from codebud import profiles + from codebud.cli import integrity as cli + + monkeypatch.setattr(isolation, "fence_available", lambda: True) + monkeypatch.setattr(profiles, "shipped_profile_dir", lambda: tmp_path) + assert set(isolation.FENCE_PROGRAMS) <= cli._programs(set()) + monkeypatch.setattr(profiles, "shipped_profile_dir", lambda: None) + assert not set(isolation.FENCE_PROGRAMS) & (cli._programs(set()) - {"sh"}) + + +# --- live, where bubblewrap, pasta and nft work ----------------------------------------- + + +def _fence_works() -> bool: + if not (shutil.which("bwrap") and isolation.fence_available()): + return False + probe = subprocess.run( + ["pasta", "--config-net", "--quiet", "--", "nft", "list", "ruleset"], + capture_output=True, + timeout=30, + ) + return probe.returncode == 0 and isolation.available() == "bwrap" + + +live = pytest.mark.skipif(not _fence_works(), reason="needs bwrap, pasta and nft") + + +@live +def test_live_the_step_cannot_reach_the_host_or_change_the_rules(tmp_path, monkeypatch): + from codebud.sandbox import Sandbox + + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("CODEBUD_PRIVACY", "private") + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.listen() + port = server.getsockname()[1] + threading.Thread(target=lambda: server.accept(), daemon=True).start() + isolation.add_network_fence(lambda words: isolation.Fence()) + sandbox = Sandbox(timeout=30) + try: + reach = sandbox.run_command( + f"timeout 3 bash -c 'exec 3<>/dev/tcp/127.0.0.1/{port}'", network=True + ) + assert reach["returncode"] != 0 # the host's loopback is not the step's + flush = sandbox.run_command("nft flush ruleset", network=True) + assert flush["returncode"] != 0 and "not permitted" in flush["stderr"] + dns = sandbox.run_command("getent hosts example.com", network=True) + assert dns["returncode"] != 0 + wrote = sandbox.run_command("touch made-in-step", network=True) + assert wrote["returncode"] == 0 and (tmp_path / "made-in-step").exists() + finally: + server.close() diff --git a/tests/test_integrity_steps.py b/tests/test_integrity_steps.py index 27b0c93..d658f12 100644 --- a/tests/test_integrity_steps.py +++ b/tests/test_integrity_steps.py @@ -36,7 +36,9 @@ def bindir(tmp_path, monkeypatch): exe.chmod(0o755) monkeypatch.setenv("PATH", f"{d}:/usr/bin:/bin") # No real isolation here; a test can wrap with `wrap` instead. - monkeypatch.setattr("codebud.isolation.prepare", lambda argv, network=False: (argv, None)) + monkeypatch.setattr( + "codebud.isolation.prepare_with_programs", lambda argv, network=False: (argv, None, []) + ) return d, marks @@ -92,7 +94,8 @@ def test_the_program_behind_a_wrapper_is_checked(bindir): def test_the_isolation_tool_is_checked(bindir, monkeypatch): d, marks = bindir monkeypatch.setattr( - "codebud.isolation.prepare", lambda argv, network=False: (["wrap", *argv], None) + "codebud.isolation.prepare_with_programs", + lambda argv, network=False: (["wrap", *argv], None, [0]), ) _pin(d, ["grep"]) # the wrapper is not pinned result = Sandbox().run_command("grep x") @@ -175,7 +178,8 @@ def test_the_checked_file_is_the_one_that_runs(bindir, tmp_path, monkeypatch): shift.write_text(f'#!/bin/sh\nPATH="{other}"; export PATH\nexec "$@"\n') shift.chmod(0o755) monkeypatch.setattr( - "codebud.isolation.prepare", lambda argv, network=False: ([str(shift), *argv], None) + "codebud.isolation.prepare_with_programs", + lambda argv, network=False: ([str(shift), *argv], None, [0]), ) _pin(d, ["grep", str(shift)]) result = Sandbox().run_command("grep x") diff --git a/tests/test_scope.py b/tests/test_scope.py index 5f70f9a..423e7fd 100644 --- a/tests/test_scope.py +++ b/tests/test_scope.py @@ -21,6 +21,7 @@ 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"] +fence_scanning = false # the command-line guard alone (the fence: test_scope_fence.py) """ @@ -68,6 +69,7 @@ def engagement(tmp_path): def guard(engagement, monkeypatch): monkeypatch.setattr(safety, "_STEP_CHECKS", []) monkeypatch.setattr(safety, "_STEP_NOTICES", []) + monkeypatch.setattr("codebud.isolation._FENCES", []) 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. @@ -250,6 +252,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._FENCES", []) monkeypatch.setattr("codebud.isolation.available", lambda: "bwrap") ScopeGuard(load(path)).activate() reason = _engine().refusal_reason(_step("nmap 10.0.0.5")) diff --git a/tests/test_scope_fence.py b/tests/test_scope_fence.py new file mode 100644 index 0000000..f0e7aef --- /dev/null +++ b/tests/test_scope_fence.py @@ -0,0 +1,126 @@ +""" +tests/test_scope_fence.py + +The engagement's network fence policy (phase 10, step 5 plan, PR 6): +which addresses a network step may reach. Scanning steps are fenced to +exactly their targets (names resolved per step), excluded networks are +dropped first, passive lookups are not fenced, and scans that need raw +sockets are refused with that reason. +""" + +import ipaddress +import shlex + +import pytest + +from codebud import isolation, safety +from codebud_sec import fence as fence_module +from codebud_sec.engagement import load +from codebud_sec.fence import ScopeFence +from codebud_sec.scope import ScopeGuard + +SCOPE = """\ +client = "ACME" +in_scope = ["10.0.0.0/16", "app.example.com"] +out_of_scope = ["10.0.5.0/24", "10.0.9.9"] +""" +NET = ipaddress.ip_network + + +@pytest.fixture(autouse=True) +def resolver(monkeypatch): + names = {"app.example.com": ["10.0.0.7", "2001:db8::7"]} + + def resolve(name): + if name not in names: + raise OSError("no such name") + return names[name] + + monkeypatch.setattr(fence_module, "_resolve", resolve) + return names + + +def _fence(tmp_path, extra=""): + path = tmp_path / "acme.toml" + path.write_text(SCOPE + extra) + return ScopeFence(load(path)) + + +def _decide(fence, cmd): + return fence.decide(shlex.split(cmd)) + + +def test_a_scan_may_reach_exactly_its_targets(tmp_path): + decision = _decide(_fence(tmp_path), "nmap --unprivileged -Pn -n 10.0.0.5 10.0.1.0/24") + assert isinstance(decision, isolation.Fence) + assert set(decision.allow) == {NET("10.0.0.5/32"), NET("10.0.1.0/24")} + assert decision.hosts == () + + +def test_names_are_resolved_into_the_steps_hosts(tmp_path): + decision = _decide(_fence(tmp_path), "curl -sS https://app.example.com/") + assert set(decision.allow) == {NET("10.0.0.7/32"), NET("2001:db8::7/128")} + assert set(decision.hosts) == { + ("app.example.com", "10.0.0.7"), + ("app.example.com", "2001:db8::7"), + } + + +def test_excluded_networks_are_always_dropped(tmp_path): + decision = _decide(_fence(tmp_path), "nmap --unprivileged 10.0.0.0/16") + assert set(decision.deny) == {NET("10.0.5.0/24"), NET("10.0.9.9/32")} + rules = isolation.nft_ruleset(decision) + assert rules.index("10.0.5.0/24 drop") < rules.index("10.0.0.0/16 accept") + + +def test_a_name_that_does_not_resolve_is_refused(tmp_path): + decision = _decide(_fence(tmp_path), "curl https://gone.app.example.com/") + assert isinstance(decision, str) and "cannot resolve" in decision + + +@pytest.mark.parametrize("cmd", ["dig app.example.com", "host 10.0.0.5", "whois app.example.com"]) +def test_passive_lookups_are_not_fenced(tmp_path, cmd): + assert _decide(_fence(tmp_path), cmd) is None + + +@pytest.mark.parametrize( + ("cmd", "why"), + [ + ("nmap -Pn 10.0.0.5", "--unprivileged"), + ("nmap --unprivileged -sS 10.0.0.5", "raw sockets"), + ("nmap --unprivileged -sU 10.0.0.5", "raw sockets"), + ("nmap --unprivileged -O 10.0.0.5", "raw sockets"), + ("masscan -p80 10.0.1.0/24", "raw sockets"), + ], +) +def test_scans_that_need_raw_sockets_are_refused(tmp_path, cmd, why): + decision = _decide(_fence(tmp_path), cmd) + assert isinstance(decision, str) and why in decision, cmd + + +def test_fence_scanning_false_gives_raw_scans_back(tmp_path): + fence = _fence(tmp_path, "fence_scanning = false\n") + for cmd in ("nmap -sS 10.0.0.5", "masscan -p80 10.0.1.0/24", "curl https://app.example.com/"): + assert _decide(fence, cmd) is None, cmd + + +def test_fence_scanning_must_be_a_boolean(tmp_path): + from codebud_sec.engagement import EngagementError + + path = tmp_path / "bad.toml" + path.write_text(SCOPE + 'fence_scanning = "no"\n') + with pytest.raises(EngagementError): + load(path) + + +def test_the_guard_turns_the_fence_on_and_off(tmp_path, monkeypatch): + monkeypatch.setattr(safety, "_STEP_CHECKS", []) + monkeypatch.setattr(safety, "_STEP_NOTICES", []) + monkeypatch.setattr(isolation, "_FENCES", []) + path = tmp_path / "acme.toml" + path.write_text(SCOPE) + guard = ScopeGuard(load(path)) + guard.activate() + assert isolation._fence_for(["nmap", "--unprivileged", "10.0.0.5"]) is not None + guard.deactivate() + assert isolation._FENCES == [] diff --git a/tests/test_scope_web.py b/tests/test_scope_web.py index 621e886..5ad3440 100644 --- a/tests/test_scope_web.py +++ b/tests/test_scope_web.py @@ -31,6 +31,7 @@ def guard(tmp_path, monkeypatch): path.write_text(SCOPE) monkeypatch.setattr(safety, "_STEP_CHECKS", []) monkeypatch.setattr(safety, "_STEP_NOTICES", []) + monkeypatch.setattr("codebud.isolation._FENCES", []) monkeypatch.delenv("CODEBUD_PRIVACY", raising=False) monkeypatch.setattr("codebud.isolation.available", lambda: "bwrap") g = ScopeGuard(load(path)) diff --git a/tests/test_security_pentest.py b/tests/test_security_pentest.py index 60a120d..054ad91 100644 --- a/tests/test_security_pentest.py +++ b/tests/test_security_pentest.py @@ -30,6 +30,7 @@ def pentest(monkeypatch): monkeypatch.setattr(paths, "PROFILE_POLICY_DIR", profile.policy) monkeypatch.setattr(safety, "_STEP_CHECKS", []) monkeypatch.setattr(safety, "_STEP_NOTICES", []) + monkeypatch.setattr("codebud.isolation._FENCES", []) monkeypatch.delenv("CODEBUD_PRIVACY", raising=False) monkeypatch.setattr("codebud.isolation.available", lambda: "bwrap") return profile @@ -77,8 +78,7 @@ def test_without_an_engagement_nothing_runs(pentest): [ "dig app.example.com", "whois app.example.com", - "nmap -sV -p 22,80,443 -oA scans/web 10.0.0.5", - "masscan -p1-1000 --rate 500 10.0.1.0/24", + "nmap --unprivileged -n -sV -p 22,80,443 -oA scans/web 10.0.0.5", "curl -sS https://app.example.com/", "wget --max-redirect=0 -O scans/index.html https://app.example.com/", "whatweb --follow-redirect=never https://app.example.com", @@ -86,7 +86,7 @@ def test_without_an_engagement_nothing_runs(pentest): "ffuf -u https://app.example.com/FUZZ -w words.txt -o scans/ffuf.json", "testssl.sh --jsonfile scans/tls.json app.example.com", "nc -zv 10.0.0.5 22", - "timeout 600 nmap -sV 10.0.0.5", + "timeout 600 nmap --unprivileged -sV 10.0.0.5", ], ) def test_recon_and_scanning_in_scope_run(engagement, cmd): @@ -110,6 +110,9 @@ def test_reading_results_needs_no_network(engagement, cmd): ("ncat -e /bin/sh 10.0.0.5 4444", "a shell"), ("curl -T scans/web.xml https://app.example.com/", "sends a file"), ("masscan -p80 --rate 100000 10.0.1.0/24", "too fast"), + ("masscan -p80 10.0.1.0/24", "raw sockets: not behind the fence"), + ("nmap -sS --unprivileged 10.0.0.5", "a SYN scan behind the fence"), + ("nmap -sV 10.0.0.5", "nmap without --unprivileged"), ("hydra -l admin -P words.txt ssh://10.0.0.5", "password guessing: not allowed"), ("sqlmap -u https://app.example.com/?id=1", "exploitation: not allowed"), ("rm -rf scans", "not allowed"),