From 94679c0a63c87452b3a5c520f979ef556c0d6488 Mon Sep 17 00:00:00 2001 From: Mohammadreza Khellat Date: Thu, 12 Nov 2026 15:20:13 +0330 Subject: [PATCH 1/4] feat(scope): read the targets of web and TLS tools Phase 10, step 5 plan, PR 4b. The engagement scope guard now reads the targets of curl, wget, whatweb, nikto, gobuster, ffuf, feroxbuster, sslscan and testssl.sh, so these may run under an engagement instead of being refused as unparsed. A URL's host is read by a strict grammar of codebud's own; a URL two parsers could read differently (a second @, a backslash, a %-escape in the host) is refused. Proxies, curl --resolve and --connect-to hosts, DNS and DoH servers are targets; a proxy on this machine (Burp) is not, since it passes requests on to the URL's own host. Following redirects is refused (curl -L, -r of the fuzzers, nikto -followredirects); wget and whatweb, which follow them by default, need --max-redirect=0 and --follow-redirect=never. A fuzz keyword (FUZZ, or one named in -w file:KEYWORD) is allowed in the path, query and headers, never before the path: the wordlist is not read, and a word like evil.example/# would become the host. feroxbuster joins words to the URL, so its wordlist is read and a word that would leave the host (//x, https:x) refuses it. Config files a step could write in the project (ferox-config.toml, nikto.conf), option files (curl -K, wget -e/--config, nikto -config, ffuf -request), target files and mass-testing modes are refused, as are gobuster's fuzz, s3, gcs and tftp modes. gobuster dns is passive: out of scope it asks. * src/codebud_sec/scope.py (_host_port, _url, _proxy, _loopback) (_no_local_config, _need_targets, _wordlist_problem): New helpers. (_curl, _wget, _whatweb, _nikto, _gobuster, _ffuf, _feroxbuster) (_sslscan, _testssl): New parsers. (PARSERS): Add them. * tests/test_scope_web.py: New file. * tests/test_scope.py (test_an_unparsed_network_tool_is_refused): Use socat; curl has a parser now. * docs/chapters/architecture.texi, docs/chapters/development.texi: Describe them. --- docs/chapters/architecture.texi | 5 +- docs/chapters/development.texi | 3 + src/codebud_sec/scope.py | 536 +++++++++++++++++++++++++++++++- tests/test_scope.py | 4 +- tests/test_scope_web.py | 294 ++++++++++++++++++ 5 files changed, 834 insertions(+), 8 deletions(-) create mode 100644 tests/test_scope_web.py diff --git a/docs/chapters/architecture.texi b/docs/chapters/architecture.texi index 25eb40e..d50f343 100644 --- a/docs/chapters/architecture.texi +++ b/docs/chapters/architecture.texi @@ -638,8 +638,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() +│ command line and checked (DNS tools, scanners, +│ nc, web clients, fuzzers, TLS scanners); +│ 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 d44c752..b69765b 100644 --- a/docs/chapters/development.texi +++ b/docs/chapters/development.texi @@ -168,6 +168,9 @@ tests/ ├── test_scope.py engagement scope: targets read per tool, │ third parties, ranges, active refused, │ passive asked +├── test_scope_web.py engagement scope for web and TLS tools: +│ URL hosts read strictly, proxies, redirects +│ and fuzz words in the host refused ├── 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/codebud_sec/scope.py b/src/codebud_sec/scope.py index 7731dbd..a3ccf0b 100644 --- a/src/codebud_sec/scope.py +++ b/src/codebud_sec/scope.py @@ -5,10 +5,11 @@ 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); +- active tools (scanners, nc, web clients, fuzzers, TLS scanners) 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). @@ -409,9 +410,536 @@ def _ncat(args): return targets +# --- web and TLS tools (PR 4b) ----------------------------------------------------- +# +# Their targets are URLs and host[:port]s. The host of a URL is read by +# codebud's own strict grammar, not a URL library's: a URL that two +# parsers could read differently (a second @, a backslash, a %-escape in +# the host) is refused rather than judged. A redirect sends a tool where +# the server says, so following redirects is refused (and, for the tools +# that follow them by default, turning it off is required). A config file +# that the tool reads from the project directory -- which steps can +# write -- could add targets or a proxy unseen, so its presence refuses +# the tool. + +_AUTHORITY = re.compile( + r"^(?:[^@/\\\s%]*@)?(?P\[[0-9A-Fa-f:.]+\]|[A-Za-z0-9._-]+)(?::(?P\d*))?$" +) +_WEB_SCHEMES = ("http", "https") + + +def _host_port(text: str, what: str) -> str: + """The host of ``host``, ``host:port``, ``[v6]:port`` or a bare IPv6 + address.""" + if text.startswith("["): + host, _, rest = text[1:].partition("]") + if rest and not re.fullmatch(r":\d*", rest): + raise TargetError(f"{what} {text!r} is not host[:port]") + return host + if text.count(":") > 1: + return text # a bare IPv6 address + host, _, port = text.partition(":") + if port and not port.isdigit(): + raise TargetError(f"{what} {text!r} is not host[:port]") + return host + + +def _url(url: str, targets: Targets, what: str = "URL", schemes=_WEB_SCHEMES, + keywords=(), default_scheme: str | None = "http") -> None: # fmt: skip + """Add the host of ``url``. A URL without a scheme gets + ``default_scheme`` (None: the scheme is required). ``keywords`` are + words a fuzzer replaces (FUZZ): one before the path could become any + host, so it is refused.""" + scheme, sep, rest = url.partition("://") + if not sep: + if default_scheme is None: + raise TargetError(f"{what} {url!r} needs a scheme (https://...)") + scheme, rest = default_scheme, url + if scheme.lower() not in schemes: + raise TargetError(f"{what} {url!r}: only {', '.join(schemes)} URLs can be judged") + authority = re.split(r"[/?#]", rest, maxsplit=1)[0] + for keyword in keywords: + if keyword in scheme or keyword in authority: + raise TargetError( + f"{what} {url!r}: {keyword} in the host could become any host " + "(the wordlist is not read), so only the path can be fuzzed" + ) + match = _AUTHORITY.match(authority) + if not match: + raise TargetError(f"{what} {url!r}: its host cannot be read unambiguously") + targets.add(match["host"], what) + + +def _loopback(host: str) -> bool: + if host.lower() == "localhost": + return True + try: + return ipaddress.ip_address(host.strip("[]")).is_loopback + except ValueError: + return False + + +def _proxy(spec: str, targets: Targets) -> None: + """A proxy: [scheme://][user@]host[:port][/]. One on this machine (an + intercepting proxy such as Burp) passes requests on to their own + host, which is judged as the target; any other is a target itself.""" + rest = re.sub(r"^[A-Za-z][A-Za-z0-9+.-]*://", "", spec).rstrip("/") + match = _AUTHORITY.match(rest) + if not match: + raise TargetError(f"proxy {spec!r} cannot be read") + if not _loopback(match["host"]): + targets.add(match["host"], "proxy") + + +def _no_local_config(tool: str, *names: str) -> None: + for name in names: + if os.path.lexists(name): + raise TargetError( + f"{tool} reads ./{name}, which could add targets or a proxy codebud does not see" + ) + + +def _need_targets(tool: str, targets: Targets) -> Targets: + if not targets.hosts and not targets.networks: + raise TargetError(f"{tool} names no target") + return targets + + +# curl: no -L (redirects), no -K (a config file), URL globbing only with -g. +_CURL_VALUE = { + "-X", "--request", "-H", "--header", "-d", "--data", "--data-raw", "--data-binary", + "--data-urlencode", "--json", "-F", "--form", "-o", "--output", "-A", "--user-agent", + "-e", "--referer", "-b", "--cookie", "-c", "--cookie-jar", "-u", "--user", "-m", + "--max-time", "--connect-timeout", "-w", "--write-out", "-T", "--upload-file", "-r", + "--range", "--retry", "--retry-delay", "--retry-max-time", "--limit-rate", "--cacert", + "--capath", "-E", "--cert", "--key", "--cert-type", "--key-type", "--ciphers", + "--tls-max", "--tls13-ciphers", "--url", "-x", "--proxy", "-U", "--proxy-user", + "--preproxy", "--resolve", "--connect-to", "--dns-servers", "--doh-url", "--interface", + "-Y", "--speed-limit", "-y", "--speed-time", "--output-dir", "-D", "--dump-header", + "--trace", "--trace-ascii", "--stderr", "--max-filesize", + "--oauth2-bearer", "--aws-sigv4", "-z", "--time-cond", "-K", "--config", + "--unix-socket", "--abstract-unix-socket", +} # fmt: skip +_CURL_FLAGS = { + "-s", "--silent", "-S", "--show-error", "-v", "--verbose", "-i", "--include", "-I", + "--head", "-k", "--insecure", "-f", "--fail", "--fail-with-body", "-g", "--globoff", + "-G", "--get", "-4", "--ipv4", "-6", "--ipv6", "--compressed", "-O", "--remote-name", + "-J", "--remote-header-name", "-q", "--disable", "-:", "--next", "-#", + "--progress-bar", "-N", "--no-buffer", "--http1.0", "--http1.1", "--http2", "--http3", + "--tlsv1.2", "--tlsv1.3", "--ssl-reqd", "--path-as-is", "--raw", "--tcp-nodelay", + "--no-keepalive", "--create-dirs", "-n", "--netrc", "--anyauth", "--basic", "--digest", + "--ntlm", "--negotiate", "-l", "--list-only", "-h", "--help", "-V", "--version", + "-L", "--location", "--location-trusted", "--proxy-insecure", "--noproxy-all", + "--fail-early", "--no-progress-meter", "--http2-prior-knowledge", +} # fmt: skip +_CURL_PATTERNS = (r"^-X[A-Z]+$",) +_CONNECT_TO = re.compile(r"^(\[[^\]]*\]|[^:]*):([^:]*):(\[[^\]]*\]|[^:]*):(.*)$") + + +def _curl(args): + targets = Targets(ACTIVE) + urls, globoff = [], False + for option, word in _walk(args, _CURL_VALUE, _CURL_FLAGS, _CURL_PATTERNS, short_clusters=True): + if option in ("-L", "--location", "--location-trusted"): + raise TargetError("curl -L follows redirects to hosts the server chooses") + if option in ("-K", "--config"): + raise TargetError("curl -K reads options, URLs included, from a file") + if option in ("--unix-socket", "--abstract-unix-socket"): + raise TargetError(f"curl {option} talks to a local service, not a target") + if option in ("-g", "--globoff"): + globoff = True + elif option in ("-x", "--proxy", "--preproxy"): + _proxy(word, targets) + elif option == "--doh-url": + _url(word, targets, "DoH server", default_scheme=None) + elif option == "--dns-servers": + for server in word.split(","): + targets.add(_host_port(server, "DNS server"), "DNS server") + elif option == "--resolve": + # [+]host:port:addr[,addr]... ; -host:port removes an entry. + if not word.startswith("-"): + parts = word.lstrip("+").split(":", 2) + if len(parts) != 3: + raise TargetError(f"curl --resolve {word!r} cannot be read") + for addr in parts[2].split(","): + targets.add(addr, "--resolve address") + elif option == "--connect-to": + match = _CONNECT_TO.match(word) + if not match: + raise TargetError(f"curl --connect-to {word!r} cannot be read") + if match[3]: + targets.add(match[3], "--connect-to host") + elif option == "--url" or option is None: + urls.append(word) + for url in urls: + if not globoff and any(c in url for c in "[]{}"): + raise TargetError(f"curl {url!r}: URL globbing names hosts codebud cannot list; use -g") + _url(url, targets, schemes=("http", "https", "ftp", "ftps")) + return _need_targets("curl", targets) + + +# wget: redirects off (it follows them by default), no wgetrc commands or +# other config, no input files, no spanning hosts. +_WGET_VALUE = { + "-O", "--output-document", "-o", "--output-file", "-a", "--append-output", "-P", + "--directory-prefix", "-U", "--user-agent", "--header", "--method", "--body-data", + "--body-file", "--post-data", "--post-file", "-T", "--timeout", "-t", "--tries", "-w", + "--wait", "--user", "--password", "--http-user", "--http-password", "-l", "--level", + "-A", "--accept", "-R", "--reject", "--load-cookies", "--save-cookies", + "--ca-certificate", "--certificate", "--private-key", "--max-redirect", "-e", + "--execute", "-i", "--input-file", "--config", "-B", "--base", "-D", "--domains", + "--limit-rate", "-Q", "--quota", "--referer", "--secure-protocol", +} # fmt: skip +_WGET_FLAGS = { + "-q", "--quiet", "-v", "--verbose", "-nv", "--no-verbose", "-S", "--server-response", + "-c", "--continue", "-N", "--timestamping", "-nc", "--no-clobber", "-r", "--recursive", + "-np", "--no-parent", "-k", "--convert-links", "-p", "--page-requisites", "-m", + "--mirror", "-nd", "--no-directories", "-x", "--force-directories", "--spider", + "--no-check-certificate", "--no-config", "--no-proxy", "-4", "--inet4-only", "-6", + "--inet6-only", "--content-disposition", "--no-cookies", "--keep-session-cookies", + "-H", "--span-hosts", "-h", "--help", "-V", "--version", "-d", "--debug", +} # fmt: skip + + +def _wget(args): + targets = Targets(ACTIVE) + redirects = None + for option, word in _walk(args, _WGET_VALUE, _WGET_FLAGS, short_clusters=True): + if option in ("-e", "--execute") and word.replace(" ", "") in ("robots=off", "robots=on"): + continue + if option in ("-e", "--execute", "--config"): + raise TargetError( + f"wget {option} sets options, a proxy included, codebud does not read" + ) + if option in ("-i", "--input-file", "-B", "--base"): + raise TargetError(f"wget {option} takes URLs from a file") + if option in ("-H", "--span-hosts", "-D", "--domains"): + raise TargetError(f"wget {option} lets it go to other hosts") + if option == "--max-redirect": + redirects = word + elif option is None: + _url(word, targets, schemes=("http", "https", "ftp", "ftps")) + if redirects != "0": + raise TargetError( + "wget follows redirects to hosts the server chooses; give --max-redirect=0" + ) + return _need_targets("wget", targets) + + +# whatweb: operands are URLs, hosts, addresses, ranges -- or a file of them. +_WHATWEB_VALUE = { + "-a", "--aggression", "-U", "--user-agent", "-H", "--header", "--proxy", + "--proxy-user", "--follow-redirect", "--max-redirects", "-u", "--user", "-c", + "--cookie", "--log-brief", "--log-verbose", "--log-json", "--log-xml", "--log-sql", + "--open-timeout", "--read-timeout", "-t", "--max-threads", "--wait", "-p", "--plugins", +} # fmt: skip +_WHATWEB_FLAGS = { + "-v", "--verbose", "-q", "--quiet", "--no-errors", "--color=never", "--colour=never", + "-h", "--help", "--version", +} # fmt: skip + + +def _whatweb(args): + targets = Targets(ACTIVE) + redirects = None + for option, word in _walk(args, _WHATWEB_VALUE, _WHATWEB_FLAGS): + if option in ("-p", "--plugins") and ("/" in word or word.startswith(".")): + raise TargetError("whatweb --plugins from a path runs code codebud does not read") + if option == "--follow-redirect": + redirects = word + elif option == "--proxy": + _proxy(word, targets) + elif option is None: + if "://" in word: + _url(word, targets) + elif os.path.lexists(word): + raise TargetError(f"whatweb {word}: reads targets from a file") + else: + targets.add(_host_port(word, "target")) + if redirects != "never": + raise TargetError( + "whatweb follows redirects to hosts the server chooses; give --follow-redirect=never" + ) + return _need_targets("whatweb", targets) + + +# nikto: Perl's option parser takes -opt and --opt alike. +_NIKTO_VALUE = { + "h", "host", "p", "port", "Tuning", "Display", "o", "output", "Format", "maxtime", + "timeout", "Pause", "root", "vhost", "id", "Plugins", "useragent", "mutate", "evasion", + "useproxy", "config", +} # fmt: skip +_NIKTO_FLAGS = {"ssl", "nossl", "no404", "nointeractive", "ask=no", "noslash", "ipv4", "ipv6", + "Help", "Version", "followredirects", "nolookup", "list-plugins"} # fmt: skip + + +def _nikto(args): + targets = Targets(ACTIVE) + value = {d + o for o in _NIKTO_VALUE for d in ("-", "--")} + flags = {d + o for o in _NIKTO_FLAGS for d in ("-", "--")} + _no_local_config("nikto", "nikto.conf") + for option, word in _walk(args, value, flags): + name = (option or "").lstrip("-") + if name == "followredirects": + raise TargetError("nikto -followredirects goes to hosts the server chooses") + if name == "config": + raise TargetError("nikto -config reads settings, a proxy included, from a file") + if name == "useproxy": + _proxy(word, targets) + elif name in ("h", "host"): + if os.path.lexists(word): + raise TargetError(f"nikto -h {word}: reads targets from a file") + for host in word.split(","): + if "://" in host: + _url(host, targets) + else: + targets.add(_host_port(host, "target")) + elif option is None: + raise TargetError(f"nikto {word}: targets are given with -h") + return _need_targets("nikto", targets) + + +# gobuster: dir and vhost send requests to -u; dns looks names up (passive). +# fuzz (FUZZ anywhere), s3 and gcs (third-party clouds) and tftp are refused. +# Letters mean different things per mode (-r, -d, -c), so each has its own. +_GOBUSTER_COMMON_VALUE = {"-w", "--wordlist", "-t", "--threads", "-o", "--output", "--delay", + "--timeout"} # fmt: skip +_GOBUSTER_COMMON_FLAGS = {"-q", "--quiet", "-z", "--no-progress", "--no-error", "-v", + "--verbose", "-h", "--help"} # fmt: skip +_GOBUSTER_MODES = { + "dir": ( + {"-u", "--url", "-x", "--extensions", "-s", "--status-codes", "-b", + "--status-codes-blacklist", "-c", "--cookies", "-H", "--headers", "-a", + "--useragent", "-U", "--username", "-P", "--password", "--proxy", "-m", "--method", + "--exclude-length", "--client-cert-pem", "--client-cert-pem-key"}, + {"-k", "--no-tls-validation", "-e", "--expanded", "-n", "--no-status", "-f", + "--add-slash", "-d", "--discover-backup", "--hide-length", "--random-agent", + "-r", "--follow-redirect"}, + ), + "vhost": ( + {"-u", "--url", "-c", "--cookies", "-H", "--headers", "-a", "--useragent", "-U", + "--username", "-P", "--password", "--proxy", "-m", "--method", "--domain", + "--exclude-length", "--client-cert-pem", "--client-cert-pem-key"}, + {"-k", "--no-tls-validation", "--append-domain", "--random-agent", "-r", + "--follow-redirect"}, + ), + "dns": ( + {"-d", "--domain", "-r", "--resolver"}, + {"-c", "--show-cname", "-i", "--show-ips", "--wildcard"}, + ), +} # fmt: skip + + +def _gobuster(args): + if not args or args[0] not in _GOBUSTER_MODES: + mode = args[0] if args else "(none)" + raise TargetError(f"gobuster mode {mode}: only dir, vhost and dns can be judged") + mode, args = args[0], args[1:] + targets = Targets(PASSIVE if mode == "dns" else ACTIVE) + value, flags = _GOBUSTER_MODES[mode] + for option, word in _walk(args, value | _GOBUSTER_COMMON_VALUE, flags | _GOBUSTER_COMMON_FLAGS): + if mode != "dns" and option in ("-r", "--follow-redirect"): + raise TargetError("gobuster -r follows redirects to hosts the server chooses") + if mode != "dns" and option in ("-u", "--url"): + _url(word, targets, default_scheme=None) + elif mode == "dns" and option in ("-d", "--domain"): + targets.add(word, "domain") + elif mode == "dns" and option in ("-r", "--resolver"): + targets.add(_host_port(word, "DNS server"), "DNS server") + elif option == "--proxy": + _proxy(word, targets) + elif option is None: + raise TargetError(f"gobuster {word}: targets are given with -u or -d") + return _need_targets("gobuster", targets) + + +# ffuf: Go flags (-opt or --opt). FUZZ and the -w file:KEYWORD keywords may +# be in the path, query and headers, never in the scheme or host. +_FFUF_VALUE = { + "u", "w", "H", "X", "d", "b", "x", "replay-proxy", "t", "p", "rate", "maxtime", + "maxtime-job", "timeout", "e", "mc", "ml", "mr", "ms", "mt", "mw", "fc", "fl", "fr", + "fs", "ft", "fw", "o", "of", "od", "mode", "recursion-depth", "recursion-strategy", + "acc", "sni", "debug-log", "input-num", "cc", "ck", "request", "request-proto", + "input-cmd", "config", +} # fmt: skip +_FFUF_FLAGS = {"ac", "c", "v", "s", "sa", "se", "sf", "ic", "k", "recursion", "or", "noninteractive", + "json", "r", "h", "V", "http2", "raw", "ignore-body", "ach", "D"} # fmt: skip + + +def _ffuf(args): + targets = Targets(ACTIVE) + # Go's flag package takes "-opt", "--opt" and "-opt=value" alike. + value = {"-" + o for o in _FFUF_VALUE} + flags = {"-" + o for o in _FFUF_FLAGS} + words = [a[1:] if a.startswith("--") and len(a) > 2 else a for a in args] + keywords, urls = {"FUZZ"}, [] + for option, word in _walk(words, value, flags): + if option == "-input-cmd": + raise TargetError("ffuf -input-cmd runs a command to make its inputs") + if option in ("-request", "-request-proto", "-config"): + raise TargetError(f"ffuf {option} reads the request or settings from a file") + if option == "-r": + raise TargetError("ffuf -r follows redirects to hosts the server chooses") + if option == "-w": + _, sep, keyword = word.rpartition(":") + if sep and keyword and "/" not in keyword: + keywords.add(keyword) + elif option in ("-x", "-replay-proxy"): + _proxy(word, targets) + elif option == "-u": + urls.append(word) + elif option is None: + raise TargetError(f"ffuf {word}: the target is given with -u") + for url in urls: + _url(url, targets, keywords=sorted(keywords), default_scheme=None) + return _need_targets("ffuf", targets) + + +# feroxbuster: joins each word to the URL, and a word like //evil.example +# or https:evil.example would leave the host, so the wordlist is read. +_FEROX_VALUE = { + "-u", "--url", "-w", "--wordlist", "-p", "--proxy", "-P", "--replay-proxy", "-t", + "--threads", "-d", "--depth", "-x", "--extensions", "-H", "--headers", "-b", + "--cookies", "-m", "--methods", "--data", "-a", "--user-agent", "-o", "--output", + "-s", "--status-codes", "-C", "--filter-status", "-S", "--filter-size", "-W", + "--filter-words", "-N", "--filter-lines", "-X", "--filter-regex", "-T", "--timeout", + "--time-limit", "--rate-limit", "-L", "--scan-limit", "--dont-scan", "-R", + "--replay-codes", "--resume-from", "--debug-log", +} # fmt: skip +_FEROX_FLAGS = { + "-k", "--insecure", "-n", "--no-recursion", "-q", "--quiet", "--silent", "-v", + "--verbosity", "-e", "--extract-links", "--auto-tune", "--auto-bail", "-f", + "--add-slash", "-r", "--redirects", "--stdin", "--burp", "--burp-replay", "--json", + "--force-recursion", "--dont-extract-links", "-h", "--help", "-V", "--version", + "--no-state", +} # fmt: skip +_JOIN_ESCAPE = re.compile(r"^(//|\\\\|/\\|\\/|[A-Za-z][A-Za-z0-9+.-]*:)") + + +def _wordlist_problem(path: str) -> str | None: + try: + with open(path, encoding="utf-8", errors="replace") as f: + for number, line in enumerate(f, 1): + if _JOIN_ESCAPE.match(line.strip()): + return f"line {number} ({line.strip()!r}) would leave the target host" + except OSError as exc: + return f"cannot be read ({exc.strerror})" + return None + + +def _feroxbuster(args): + targets = Targets(ACTIVE) + _no_local_config("feroxbuster", "ferox-config.toml") + wordlist = None + for option, word in _walk(args, _FEROX_VALUE, _FEROX_FLAGS, (r"^-v+$",)): + if option in ("-r", "--redirects"): + raise TargetError("feroxbuster -r follows redirects to hosts the server chooses") + if option in ("--stdin", "--resume-from"): + raise TargetError(f"feroxbuster {option} takes targets from elsewhere") + if option in ("-w", "--wordlist"): + wordlist = word + elif option in ("-p", "--proxy", "-P", "--replay-proxy"): + _proxy(word, targets) + elif option in ("-u", "--url"): + _url(word, targets, default_scheme=None) + elif option is None: + raise TargetError(f"feroxbuster {word}: targets are given with -u") + if wordlist is None: + raise TargetError("feroxbuster needs -w: its wordlist is checked before it runs") + problem = _wordlist_problem(wordlist) + if problem: + raise TargetError(f"feroxbuster wordlist {wordlist}: {problem}") + return _need_targets("feroxbuster", targets) + + +# sslscan: --opt or --opt=value; operands host[:port]; --targets a list file. +_SSLSCAN_VALUE = {"--targets", "--sni-name", "--xmpp-server", "--xml", "--timeout", + "--connect-timeout", "--sleep", "--certs", "--pk", "--pkpass"} # fmt: skip +_SSLSCAN_PATTERNS = ( + r"^--(ipv4|ipv6|show-certificate|show-certs|show-client-cas|show-ciphers|show-cipher-ids" + r"|show-times|show-sigs|no-ciphersuites|no-compression|no-fallback|no-groups|no-heartbleed" + r"|no-renegotiation|no-check-certificate|no-cipher-details|no-colour|no-color|ocsp|tlsall" + r"|ssl2|ssl3|tls10|tls11|tls12|tls13|bugs|verbose|version|help|iana-names|rdp" + r"|starttls-(ftp|imap|irc|ldap|mysql|pop3|psql|smtp|xmpp))$", + r"^-[46]$", +) + + +def _sslscan(args): + targets = Targets(ACTIVE) + for option, word in _walk(args, _SSLSCAN_VALUE, set(), _SSLSCAN_PATTERNS): + if option == "--targets": + real = os.path.realpath(word) + project = os.path.realpath(os.getcwd()) + if os.path.commonpath([real, project]) != project or not os.path.isfile(real): + raise TargetError(f"target list {word}: 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(_host_port(spec, "target"), f"target in {word}") + elif option is None: + targets.add(_host_port(word, "target")) + return _need_targets("sslscan", targets) + + +# testssl.sh: one URI (host[:port] or https://...); mass testing and --mx +# (which tests the hosts a DNS answer names) are refused. +_TESTSSL_VALUE = { + "-t", "--starttls", "--xmpphost", "--ip", "--proxy", "--sneaky", "--jsonfile", + "--jsonfile-pretty", "--csvfile", "--htmlfile", "--logfile", "--outfile", "--severity", + "--connect-timeout", "--openssl-timeout", "--basicauth", "--reqheader", "--color", + "--add-ca", "--mapping", "--phone-out", "--file", "-iL", "--input-file", "--mx", + "--openssl", "--warnings", +} # fmt: skip +_TESTSSL_FLAGS = { + "-p", "--protocols", "-e", "--each-cipher", "-E", "--cipher-per-proto", "-s", + "--std", "--categories", "-f", "--fs", "--forward-secrecy", "-S", "--server-defaults", + "-P", "--server-preference", "-h", "--header", "--headers", "-U", "--vulnerable", + "-c", "--client-simulation", "-g", "--grease", "-q", "--quiet", "--wide", "--fast", + "--ids-friendly", "--hints", "-4", "-6", "--nodns", "--ssl-native", "--sneaky", + "-b", "--banner", "-v", "--version", "--help", "-9", "--full", "-H", "--heartbleed", + "-I", "--ccs", "-T", "--ticketbleed", "-R", "--renegotiation", "-C", "--compression", + "-B", "--breach", "-O", "--poodle", "-Z", "--tls-fallback", "-W", "--sweet32", + "-A", "--beast", "-L", "--lucky13", "-F", "--freak", "-J", "--logjam", "-D", "--drown", + "--rc4", "--appelbaum", +} # fmt: skip + + +def _testssl(args): + targets = Targets(ACTIVE) + operands = [] + for option, word in _walk(args, _TESTSSL_VALUE - {"--sneaky"}, _TESTSSL_FLAGS): + if option in ("--file", "-iL", "--input-file"): + raise TargetError("testssl.sh mass testing reads targets and options from a file") + if option == "--mx": + raise TargetError("testssl.sh --mx tests the hosts a DNS answer names") + if option == "--openssl": + raise TargetError("testssl.sh --openssl runs a program codebud has not checked") + if option == "--proxy": + if word != "auto": + _proxy(word, targets) + else: + raise TargetError("testssl.sh --proxy auto takes a proxy from the environment") + elif option == "--ip": + if word not in ("one", "proxy"): + targets.add(word, "--ip address") + elif option is None: + operands.append(word) + if len(operands) != 1: + raise TargetError("testssl.sh takes one target") + if "://" in operands[0]: + _url(operands[0], targets) + else: + targets.add(_host_port(operands[0], "target")) + return targets + + PARSERS = { "dig": _dig, "host": _host, "nmap": _nmap, "masscan": _masscan, "nc": _nc, "netcat": _nc, "ncat": _ncat, + "curl": _curl, "wget": _wget, "whatweb": _whatweb, "nikto": _nikto, "nikto.pl": _nikto, + "gobuster": _gobuster, "ffuf": _ffuf, "feroxbuster": _feroxbuster, + "sslscan": _sslscan, "testssl.sh": _testssl, "testssl": _testssl, } # fmt: skip diff --git a/tests/test_scope.py b/tests/test_scope.py index 7a778c8..e53f387 100644 --- a/tests/test_scope.py +++ b/tests/test_scope.py @@ -146,7 +146,7 @@ def test_tools_are_found_behind_wrappers_and_by_variant_name(): def _engine(): engine = SafetyEngine() engine.harmless_commands = ["dig", "host", "ls"] - engine.powerful_commands = ["nmap", "masscan", "nc", "ncat", "timeout", "curl"] + engine.powerful_commands = ["nmap", "masscan", "nc", "ncat", "timeout", "socat"] return engine @@ -204,7 +204,7 @@ def test_passive_lookups_out_of_scope_ask_instead(guard): def test_an_unparsed_network_tool_is_refused(guard): - reason = _engine().refusal_reason(_step("curl https://app.example.com/")) + reason = _engine().refusal_reason(_step("socat - TCP:app.example.com:80")) assert reason is not None and "cannot read its targets" in reason diff --git a/tests/test_scope_web.py b/tests/test_scope_web.py new file mode 100644 index 0000000..63b6fc2 --- /dev/null +++ b/tests/test_scope_web.py @@ -0,0 +1,294 @@ +""" +tests/test_scope_web.py + +The engagement scope guard for web and TLS tools (phase 10, step 5 +plan, PR 4b): curl, wget, whatweb, nikto, gobuster, ffuf, feroxbuster, +sslscan and testssl.sh. A URL's host is read by a strict grammar (an +ambiguous URL is refused), proxies, --resolve and --connect-to hosts +are targets, following redirects is refused, and a fuzz keyword may +only stand in the path. +""" + +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", "admin.app.example.com"] +""" + + +@pytest.fixture +def guard(tmp_path, monkeypatch): + path = tmp_path / "acme.toml" + path.write_text(SCOPE) + monkeypatch.setattr(safety, "_STEP_CHECKS", []) + monkeypatch.setattr(safety, "_STEP_NOTICES", []) + monkeypatch.delenv("CODEBUD_PRIVACY", raising=False) + monkeypatch.setattr("codebud.isolation.available", lambda: "bwrap") + g = ScopeGuard(load(path)) + g.activate() + return g + + +@pytest.fixture +def project(tmp_path, monkeypatch): + root = tmp_path / "case" + root.mkdir() + (root / "words.txt").write_text("admin\nlogin\n/backup\n.git/HEAD\n") + monkeypatch.chdir(root) + return root + + +def _hosts(cmd): + [(_, targets)] = targets_of(shlex.split(cmd)) + return sorted(targets.hosts) + + +NO_REDIRECTS = "--max-redirect=0" + + +# --- reading targets --------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("cmd", "hosts"), + [ + ("curl -sS https://app.example.com/login", ["app.example.com"]), + ("curl -sSo page.html http://10.0.0.5:8080/", ["10.0.0.5"]), + ("curl -XPOST -d @body.json https://user:pw@app.example.com/api", ["app.example.com"]), + ("curl -g 'http://[2001:db8::5]:80/'", ["2001:db8::5"]), + ("curl app.example.com", ["app.example.com"]), + ( + "curl -x http://proxy.example.org:3128 https://app.example.com/", + ["app.example.com", "proxy.example.org"], + ), + ("curl -x http://127.0.0.1:8080 https://app.example.com/", ["app.example.com"]), + ( + "curl --resolve app.example.com:443:10.0.0.9 https://app.example.com/", + ["10.0.0.9", "app.example.com"], + ), + ( + "curl --connect-to app.example.com:443:10.0.0.7:8443 https://app.example.com/", + ["10.0.0.7", "app.example.com"], + ), + ("curl --dns-servers 10.0.0.53 https://app.example.com/", ["10.0.0.53", "app.example.com"]), + ( + "curl --doh-url https://dns.example.net/dns-query https://app.example.com/", + ["app.example.com", "dns.example.net"], + ), + (f"wget {NO_REDIRECTS} -q -O out.html https://app.example.com/", ["app.example.com"]), + ( + f"wget {NO_REDIRECTS} -r -np -e robots=off https://app.example.com/docs/", + ["app.example.com"], + ), + ( + "whatweb --follow-redirect=never -a 3 https://app.example.com 10.0.0.5", + ["10.0.0.5", "app.example.com"], + ), + ("nikto -h https://app.example.com -Tuning 123 -nointeractive", ["app.example.com"]), + ("nikto -h 10.0.0.5 -p 8443 -ssl", ["10.0.0.5"]), + ("gobuster dir -u https://app.example.com/ -w words.txt -x php -t 20", ["app.example.com"]), + ( + "gobuster dir -u https://app.example.com/ -w words.txt -d -c session=1", + ["app.example.com"], + ), + ( + "gobuster vhost -u https://10.0.0.5 -w words.txt --append-domain --domain app.example.com", + ["10.0.0.5"], + ), + ( + "gobuster dns -d app.example.com -w words.txt -r 10.0.0.53 -c", + ["10.0.0.53", "app.example.com"], + ), + ("ffuf -u https://app.example.com/FUZZ -w words.txt -mc 200,301", ["app.example.com"]), + ( + "ffuf -u https://app.example.com/?id=FUZZ -w words.txt -H 'Host: FUZZ.app.example.com'", + ["app.example.com"], + ), + ( + "ffuf --u=https://app.example.com/W1 -w words.txt:W1 -x http://127.0.0.1:8080", + ["app.example.com"], + ), + ("feroxbuster -u https://app.example.com -w words.txt -n -q", ["app.example.com"]), + ("feroxbuster -u https://app.example.com -w words.txt --burp", ["app.example.com"]), + ("sslscan --no-colour app.example.com:443", ["app.example.com"]), + ("sslscan --starttls-smtp [2001:db8::25]:25", ["2001:db8::25"]), + ("testssl.sh --fast --jsonfile out.json https://app.example.com", ["app.example.com"]), + ("testssl.sh -t smtp --ip 10.0.0.25 app.example.com:25", ["10.0.0.25", "app.example.com"]), + ("testssl -U 10.0.0.5:443", ["10.0.0.5"]), + ], +) +def test_targets_are_read(project, cmd, hosts): + assert _hosts(cmd) == sorted(hosts), cmd + + +def test_tiers(project): + [(_, ffuf)] = targets_of(shlex.split("ffuf -u https://app.example.com/FUZZ -w words.txt")) + [(_, dns)] = targets_of(shlex.split("gobuster dns -d app.example.com -w words.txt")) + assert ffuf.tier == ACTIVE and dns.tier == PASSIVE + + +@pytest.mark.parametrize( + ("cmd", "why"), + [ + # redirects go where the server says + ("curl -L https://app.example.com/", "follows redirects"), + ("curl -sSL https://app.example.com/", "follows redirects, in a cluster"), + ("wget https://app.example.com/", "follows redirects by default"), + ("wget --max-redirect=5 https://app.example.com/", "follows redirects"), + ("whatweb https://app.example.com", "follows redirects by default"), + ("nikto -h app.example.com -followredirects", "follows redirects"), + ("gobuster dir -u https://app.example.com -w words.txt -r", "follows redirects"), + ("ffuf -u https://app.example.com/FUZZ -w words.txt -r", "follows redirects"), + ("feroxbuster -u https://app.example.com -w words.txt -r", "follows redirects"), + # targets or options from elsewhere + ("curl -K opts.txt", "config file"), + (f"wget {NO_REDIRECTS} -i urls.txt", "input file"), + (f"wget {NO_REDIRECTS} -e use_proxy=on https://app.example.com/", "wgetrc command"), + (f"wget {NO_REDIRECTS} -r -H https://app.example.com/", "spans hosts"), + ("ffuf -request req.txt -w words.txt", "raw request file"), + ("ffuf -input-cmd 'seq 100' -u https://app.example.com/FUZZ", "runs a command"), + ("feroxbuster --stdin -w words.txt", "targets from stdin"), + ("testssl.sh --file targets.txt", "mass testing"), + ("testssl.sh --mx example.com", "hosts a DNS answer names"), + ("testssl.sh --openssl /tmp/x app.example.com", "runs a program"), + ("nikto -config my.conf -h app.example.com", "settings file"), + ("gobuster s3 -w words.txt", "third-party cloud"), + ("gobuster fuzz -u https://FUZZ.example.com -w words.txt", "fuzz mode"), + ("curl --unix-socket /run/docker.sock http://x/", "a local service"), + # ambiguous or unjudgeable URLs + ("curl 'http://a@evil.example.org@app.example.com/'", "two @"), + ("curl 'http://app.example.com\\@evil.example.org/'", "a backslash"), + ("curl 'http://%61pp.example.com/'", "an escape in the host"), + ("curl 'http://10.0.0.[1-254]/'", "globbing"), + ("curl 'http://{app,evil}.example.com/'", "globbing"), + ("curl file:///etc/passwd", "not a web URL"), + ("curl gopher://app.example.com/", "not a web URL"), + ("gobuster dir -u app.example.com -w words.txt", "needs a scheme"), + # a keyword before the path could become any host + ("ffuf -u https://FUZZ.app.example.com/ -w words.txt", "keyword in the host"), + ("ffuf -u https://FUZZ@app.example.com/ -w words.txt", "keyword in the userinfo"), + ("ffuf -u https://app.example.comFUZZ -w words.txt", "keyword after the host"), + ("ffuf -u FUZZ://app.example.com/ -w words.txt", "keyword in the scheme"), + ("ffuf -u https://W1.app.example.com/ -w words.txt:W1", "custom keyword in the host"), + # unknown options and missing targets + ("curl --frobnicate https://app.example.com/", "unknown option"), + ("nikto -h app.example.com extra", "stray operand"), + ("ffuf -w words.txt", "no target"), + ("feroxbuster -u https://app.example.com", "no wordlist to check"), + ("testssl.sh a.example.com b.example.com", "two targets"), + ], +) +def test_what_cannot_be_judged_is_an_error(project, cmd, why): + with pytest.raises(TargetError): + targets_of(shlex.split(cmd)) + + +def test_target_files_are_refused(project): + (project / "hosts.txt").write_text("10.0.0.5\n") + for cmd in ( + "whatweb --follow-redirect=never hosts.txt", + "nikto -h hosts.txt", + ): + with pytest.raises(TargetError): + targets_of(shlex.split(cmd)) + + +def test_sslscan_reads_a_target_list_in_the_project(project): + (project / "tls.txt").write_text("# web\napp.example.com:443\n10.0.0.5:8443\n") + assert _hosts("sslscan --targets=tls.txt") == ["10.0.0.5", "app.example.com"] + with pytest.raises(TargetError): + targets_of(shlex.split("sslscan --targets=/etc/hosts")) + + +@pytest.mark.parametrize( + ("tool", "config"), + [("feroxbuster", "ferox-config.toml"), ("nikto", "nikto.conf")], +) +def test_a_config_file_in_the_project_refuses_the_tool(project, tool, config): + """A step could write it, adding a proxy or targets codebud never sees.""" + cmd = { + "feroxbuster": "feroxbuster -u https://app.example.com -w words.txt", + "nikto": "nikto -h app.example.com", + }[tool] + assert _hosts(cmd) == ["app.example.com"] + (project / config).write_text('proxy = "http://evil.example.org:8080"\n') + with pytest.raises(TargetError, match=config): + targets_of(shlex.split(cmd)) + + +@pytest.mark.parametrize( + "line", ["//evil.example.org", "\\\\evil.example.org", "https:evil.example.org", "http://x"] +) +def test_feroxbuster_wordlist_words_that_leave_the_host_are_refused(project, line): + """feroxbuster joins each word to the URL, and these would replace the host.""" + (project / "bad.txt").write_text(f"admin\n{line}\n") + with pytest.raises(TargetError, match="line 2"): + targets_of(shlex.split("feroxbuster -u https://app.example.com -w bad.txt")) + + +# --- the guard, through the safety engine ---------------------------------------- + + +def _engine(): + engine = SafetyEngine() + engine.powerful_commands = [ + "curl", "wget", "whatweb", "nikto", "gobuster", "ffuf", "feroxbuster", "sslscan", + "testssl.sh", + ] # fmt: skip + return engine + + +def _step(cmd): + return {"tool": "command", "args": {"cmd": cmd, "network": True}} + + +@pytest.mark.parametrize( + "cmd", + [ + "curl -sS https://app.example.com/", + f"wget {NO_REDIRECTS} -O out.html http://10.0.0.5/", + "ffuf -u https://app.example.com/FUZZ -w words.txt", + "testssl.sh --fast app.example.com", + ], +) +def test_in_scope_web_steps_run(guard, project, cmd): + assert _engine().refusal_reason(_step(cmd)) is None, cmd + + +@pytest.mark.parametrize( + ("cmd", "why"), + [ + ("curl https://evil.example.org/", "outside"), + ("curl https://admin.app.example.com/", "excluded subdomain"), + ("curl -x http://proxy.example.org:3128 https://app.example.com/", "a proxy outside"), + ( + "curl --resolve app.example.com:443:192.168.1.1 https://app.example.com/", + "resolved outside", + ), + ("curl --connect-to ::evil.example.org: https://app.example.com/", "connects outside"), + ("sslscan 10.0.5.20", "excluded network"), + ("curl -L https://app.example.com/", "redirects"), + ("ffuf -u https://FUZZ.app.example.com/ -w words.txt", "keyword in the host"), + ], +) +def test_out_of_scope_web_steps_are_refused(guard, project, cmd, why): + reason = _engine().refusal_reason(_step(cmd)) + assert reason is not None and "engagement scope" in reason, f"{cmd} ({why})" + + +def test_dns_brute_forcing_out_of_scope_asks(guard, project): + engine = SafetyEngine() + engine.harmless_commands = ["gobuster"] + step = _step("gobuster dns -d evil.example.org -w words.txt") + assert engine.refusal_reason(step) is None + notice = engine.confirmation_reason(step) + assert notice is not None and "evil.example.org" in notice From a510d52af9fd4f8b059cdf92263d3a6f23b99072 Mon Sep 17 00:00:00 2001 From: Mohammadreza Khellat Date: Thu, 12 Nov 2026 15:23:21 +0330 Subject: [PATCH 2/4] fix(scope): refuse environment assignments before a network tool Review of #20. env HTTPS_PROXY=... curl, env WGETRC=./x wget or env HOME=. curl give the tool a proxy, or a config file a step can write, that its words do not show, so the guard judged the wrong hosts. An assignment before a parsed tool is now refused, except for locale, time zone and terminal names. * src/codebud_sec/scope.py (_ASSIGNMENT, _HARMLESS_ENV): New. (targets_of): Refuse other assignments before a parsed tool. * tests/test_scope_web.py (test_environment_assignments_before_a_tool_are_refused) (test_harmless_environment_assignments_are_fine): New tests. --- src/codebud_sec/scope.py | 17 ++++++++++++++++- tests/test_scope_web.py | 25 +++++++++++++++++++++++++ 2 files changed, 41 insertions(+), 1 deletion(-) diff --git a/src/codebud_sec/scope.py b/src/codebud_sec/scope.py index a3ccf0b..6030b04 100644 --- a/src/codebud_sec/scope.py +++ b/src/codebud_sec/scope.py @@ -943,15 +943,30 @@ def _testssl(args): } # fmt: skip +# An environment assignment (env NAME=value tool ...) can give a tool a +# proxy (HTTPS_PROXY, ALL_PROXY) or a config it reads (WGETRC, CURL_HOME, +# HOME, XDG_CONFIG_HOME) that its words do not show; only these are fine. +_ASSIGNMENT = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*)=") +_HARMLESS_ENV = re.compile(r"^(LANG|LANGUAGE|LC_[A-Z]+|TZ|TERM|NO_COLOR|COLUMNS)$") + + 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.""" + cannot be read, or an environment assignment before one could change + where it connects.""" found = [] for i, word in enumerate(words): name = os.path.basename(word) name = "nc" if name.startswith("nc.") else name if name in PARSERS: + for before in words[:i]: + match = _ASSIGNMENT.match(before) + if match and not _HARMLESS_ENV.match(match[1]): + raise TargetError( + f"{before.split('=', 1)[0]}= before {name} could set a proxy or a " + "config file codebud does not read" + ) found.append((name, PARSERS[name](words[i + 1 :]))) return found diff --git a/tests/test_scope_web.py b/tests/test_scope_web.py index 63b6fc2..cd4bee6 100644 --- a/tests/test_scope_web.py +++ b/tests/test_scope_web.py @@ -292,3 +292,28 @@ def test_dns_brute_forcing_out_of_scope_asks(guard, project): assert engine.refusal_reason(step) is None notice = engine.confirmation_reason(step) assert notice is not None and "evil.example.org" in notice + + +# --- review of #20 ------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "cmd", + [ + "env HTTPS_PROXY=http://evil.example.org:8080 curl https://app.example.com/", + "env ALL_PROXY=socks5h://evil.example.org gobuster dir -u https://app.example.com/ -w w", + f"env WGETRC=./x wget {NO_REDIRECTS} https://app.example.com/", + "env HOME=. curl https://app.example.com/", + "env -u LANG XDG_CONFIG_HOME=. ffuf -u https://app.example.com/FUZZ -w words.txt", + "timeout 60 env CURL_HOME=. curl https://app.example.com/", + ], +) +def test_environment_assignments_before_a_tool_are_refused(project, cmd): + """Review of #20: a proxy or config set in the environment is not in + the tool's words.""" + with pytest.raises(TargetError, match="before"): + targets_of(shlex.split(cmd)) + + +def test_harmless_environment_assignments_are_fine(project): + assert _hosts("env LANG=C LC_ALL=C TZ=UTC curl https://app.example.com/") == ["app.example.com"] From e277f2494e5f3917d2fdccd9bed3baa634f62079 Mon Sep 17 00:00:00 2001 From: Mohammadreza Khellat Date: Thu, 12 Nov 2026 15:23:56 +0330 Subject: [PATCH 3/4] fix(scope): read sslscan as it reads itself; refuse testssl --phone-out Review of #20. sslscan takes values only as --opt=value, and --xmpp-server is a flag, so taking the next word as a value judged a host it does not scan (sslscan --certs evil.example.org app...). Values are now accepted only after =, --xmpp-server is a flag, and one host is taken unless --targets=file lists them. testssl.sh --phone-out contacts the CRL and OCSP servers of third parties; it was read as taking a value, and is now refused. * src/codebud_sec/scope.py (_list_file): Take a spec function. (_SSLSCAN_VALUE): Remove. (_SSLSCAN_PATTERNS): Values only after =; --xmpp-server a flag. (_sslscan): One host, or --targets=file. (_TESTSSL_VALUE, _TESTSSL_FLAGS): --phone-out is a flag. (_testssl): Refuse it. * tests/test_scope_web.py (test_sslscan_is_read_as_sslscan_reads_it) (test_sslscan_values_after_equals_are_fine) (test_testssl_phone_out_is_refused): New tests. --- src/codebud_sec/scope.py | 45 +++++++++++++++++++++------------------- tests/test_scope_web.py | 27 ++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 21 deletions(-) diff --git a/src/codebud_sec/scope.py b/src/codebud_sec/scope.py index 6030b04..7f56164 100644 --- a/src/codebud_sec/scope.py +++ b/src/codebud_sec/scope.py @@ -106,16 +106,17 @@ def _spec(spec: str, what: str): 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.""" +def _list_file(path: str, targets: Targets, spec=lambda text: text) -> None: + """Targets read from a file: only a file inside the project. ``spec`` + turns each word into a target specification (host:port -> host).""" 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}") + for word in line.split("#", 1)[0].split(): + targets.add(spec(word), f"target in {path}") # --- a small option walker ------------------------------------------------------- @@ -852,33 +853,31 @@ def _feroxbuster(args): return _need_targets("feroxbuster", targets) -# sslscan: --opt or --opt=value; operands host[:port]; --targets a list file. -_SSLSCAN_VALUE = {"--targets", "--sni-name", "--xmpp-server", "--xml", "--timeout", - "--connect-timeout", "--sleep", "--certs", "--pk", "--pkpass"} # fmt: skip +# sslscan: values only as --opt=value (sslscan does not take the next +# word); one host[:port], or a list in --targets=file. _SSLSCAN_PATTERNS = ( r"^--(ipv4|ipv6|show-certificate|show-certs|show-client-cas|show-ciphers|show-cipher-ids" r"|show-times|show-sigs|no-ciphersuites|no-compression|no-fallback|no-groups|no-heartbleed" r"|no-renegotiation|no-check-certificate|no-cipher-details|no-colour|no-color|ocsp|tlsall" - r"|ssl2|ssl3|tls10|tls11|tls12|tls13|bugs|verbose|version|help|iana-names|rdp" + r"|ssl2|ssl3|tls10|tls11|tls12|tls13|bugs|verbose|version|help|iana-names|rdp|xmpp-server" r"|starttls-(ftp|imap|irc|ldap|mysql|pop3|psql|smtp|xmpp))$", + r"^--(targets|sni-name|xml|timeout|connect-timeout|sleep|certs|pk|pkpass)=.+$", r"^-[46]$", ) def _sslscan(args): targets = Targets(ACTIVE) - for option, word in _walk(args, _SSLSCAN_VALUE, set(), _SSLSCAN_PATTERNS): - if option == "--targets": - real = os.path.realpath(word) - project = os.path.realpath(os.getcwd()) - if os.path.commonpath([real, project]) != project or not os.path.isfile(real): - raise TargetError(f"target list {word}: 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(_host_port(spec, "target"), f"target in {word}") + hosts = [] + for option, word in _walk(args, set(), set(), _SSLSCAN_PATTERNS): + if option and option.startswith("--targets="): + _list_file(option.partition("=")[2], targets, lambda w: _host_port(w, "target")) elif option is None: - targets.add(_host_port(word, "target")) + hosts.append(word) + if len(hosts) > 1: + raise TargetError("sslscan takes one host (or a list with --targets=file)") + for host in hosts: + targets.add(_host_port(host, "target")) return _need_targets("sslscan", targets) @@ -888,7 +887,7 @@ def _sslscan(args): "-t", "--starttls", "--xmpphost", "--ip", "--proxy", "--sneaky", "--jsonfile", "--jsonfile-pretty", "--csvfile", "--htmlfile", "--logfile", "--outfile", "--severity", "--connect-timeout", "--openssl-timeout", "--basicauth", "--reqheader", "--color", - "--add-ca", "--mapping", "--phone-out", "--file", "-iL", "--input-file", "--mx", + "--add-ca", "--mapping", "--file", "-iL", "--input-file", "--mx", "--openssl", "--warnings", } # fmt: skip _TESTSSL_FLAGS = { @@ -901,7 +900,7 @@ def _sslscan(args): "-I", "--ccs", "-T", "--ticketbleed", "-R", "--renegotiation", "-C", "--compression", "-B", "--breach", "-O", "--poodle", "-Z", "--tls-fallback", "-W", "--sweet32", "-A", "--beast", "-L", "--lucky13", "-F", "--freak", "-J", "--logjam", "-D", "--drown", - "--rc4", "--appelbaum", + "--rc4", "--appelbaum", "--phone-out", } # fmt: skip @@ -913,6 +912,10 @@ def _testssl(args): raise TargetError("testssl.sh mass testing reads targets and options from a file") if option == "--mx": raise TargetError("testssl.sh --mx tests the hosts a DNS answer names") + if option == "--phone-out": + raise TargetError( + "testssl.sh --phone-out contacts CRL and OCSP servers of third parties" + ) if option == "--openssl": raise TargetError("testssl.sh --openssl runs a program codebud has not checked") if option == "--proxy": diff --git a/tests/test_scope_web.py b/tests/test_scope_web.py index cd4bee6..bb89d0d 100644 --- a/tests/test_scope_web.py +++ b/tests/test_scope_web.py @@ -317,3 +317,30 @@ def test_environment_assignments_before_a_tool_are_refused(project, cmd): def test_harmless_environment_assignments_are_fine(project): assert _hosts("env LANG=C LC_ALL=C TZ=UTC curl https://app.example.com/") == ["app.example.com"] + + +@pytest.mark.parametrize( + "cmd", + [ + "sslscan --certs evil.example.org app.example.com", # --certs takes =value only + "sslscan --sni-name x app.example.com", + "sslscan --xmpp-server evil.example.org app.example.com", # a flag: two hosts + "sslscan app.example.com evil.example.org", + ], +) +def test_sslscan_is_read_as_sslscan_reads_it(project, cmd): + """Review of #20: sslscan takes values only after =, and --xmpp-server + is a flag, so reading them otherwise judged a host it does not scan.""" + with pytest.raises(TargetError): + targets_of(shlex.split(cmd)) + + +def test_sslscan_values_after_equals_are_fine(project): + cmd = "sslscan --sni-name=app.example.com --xml=out.xml --xmpp-server 10.0.0.5:5269" + assert _hosts(cmd) == ["10.0.0.5"] + + +def test_testssl_phone_out_is_refused(project): + """Review of #20: it contacts the CRL and OCSP servers of third parties.""" + with pytest.raises(TargetError, match="phone-out"): + targets_of(shlex.split("testssl.sh --phone-out x app.example.com")) From bfb06cb90b561d2851e1799a24d6f512952eeec1 Mon Sep 17 00:00:00 2001 From: Mohammadreza Khellat Date: Thu, 12 Nov 2026 15:24:29 +0330 Subject: [PATCH 4/4] fix(scope): keep appended words out of the host Review of #20. gobuster dir appends each word to its URL; without a path, a word like @evil.example.org lands in the authority, so dir mode now needs a URL with a path. feroxbuster joins words to its URL, and the wordlist check missed words the URL standard cleans up into //host (leading control characters, tabs and newlines anywhere, as in //evil); it now cleans up as WHATWG URL does first. It refused words with any scheme, though only special schemes reach a host, so javascript: and mailto: lines no longer refuse a wordlist. * src/codebud_sec/scope.py (_url): New argument need_path. (_gobuster): Use it in dir mode. (_JOIN_ESCAPE): Two slashes or backslashes, or a special scheme. (_C0_AND_SPACE): New. (_wordlist_problem): Clean up each word as the URL standard does. * tests/test_scope_web.py (test_gobuster_dir_needs_a_path_to_keep_words_out_of_the_host) (test_feroxbuster_wordlist_is_judged_after_the_url_cleanup) (test_feroxbuster_wordlist_words_that_reach_no_host_are_fine): New. --- src/codebud_sec/scope.py | 23 ++++++++++++++++++----- tests/test_scope_web.py | 24 ++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 5 deletions(-) diff --git a/src/codebud_sec/scope.py b/src/codebud_sec/scope.py index 7f56164..d5f694b 100644 --- a/src/codebud_sec/scope.py +++ b/src/codebud_sec/scope.py @@ -446,11 +446,13 @@ def _host_port(text: str, what: str) -> str: def _url(url: str, targets: Targets, what: str = "URL", schemes=_WEB_SCHEMES, - keywords=(), default_scheme: str | None = "http") -> None: # fmt: skip + keywords=(), default_scheme: str | None = "http", + need_path: bool = False) -> None: # fmt: skip """Add the host of ``url``. A URL without a scheme gets ``default_scheme`` (None: the scheme is required). ``keywords`` are words a fuzzer replaces (FUZZ): one before the path could become any - host, so it is refused.""" + host, so it is refused. ``need_path``: the tool appends words to the + URL, so a path must keep them out of the host.""" scheme, sep, rest = url.partition("://") if not sep: if default_scheme is None: @@ -468,6 +470,11 @@ def _url(url: str, targets: Targets, what: str = "URL", schemes=_WEB_SCHEMES, match = _AUTHORITY.match(authority) if not match: raise TargetError(f"{what} {url!r}: its host cannot be read unambiguously") + if need_path and not rest[len(authority) :].startswith("/"): + raise TargetError( + f"{what} {url!r}: words are appended to it, so it needs a path " + "(.../) to keep them out of the host" + ) targets.add(match["host"], what) @@ -742,7 +749,7 @@ def _gobuster(args): if mode != "dns" and option in ("-r", "--follow-redirect"): raise TargetError("gobuster -r follows redirects to hosts the server chooses") if mode != "dns" and option in ("-u", "--url"): - _url(word, targets, default_scheme=None) + _url(word, targets, default_scheme=None, need_path=mode == "dir") elif mode == "dns" and option in ("-d", "--domain"): targets.add(word, "domain") elif mode == "dns" and option in ("-r", "--resolver"): @@ -814,14 +821,20 @@ def _ffuf(args): "--force-recursion", "--dont-extract-links", "-h", "--help", "-V", "--version", "--no-state", } # fmt: skip -_JOIN_ESCAPE = re.compile(r"^(//|\\\\|/\\|\\/|[A-Za-z][A-Za-z0-9+.-]*:)") +# After the URL standard's clean-up (WHATWG URL: leading and trailing C0 +# controls and spaces stripped, tabs and newlines removed anywhere), a +# word starting // or with a backslash in their place, or with a special +# scheme that reaches a host, would replace the host when joined. +_JOIN_ESCAPE = re.compile(r"^([/\\]{2}|(https?|ftp|wss?):)", re.IGNORECASE) +_C0_AND_SPACE = "".join(map(chr, range(0x21))) def _wordlist_problem(path: str) -> str | None: try: with open(path, encoding="utf-8", errors="replace") as f: for number, line in enumerate(f, 1): - if _JOIN_ESCAPE.match(line.strip()): + word = re.sub(r"[\t\n\r]", "", line).strip(_C0_AND_SPACE) + if _JOIN_ESCAPE.match(word): return f"line {number} ({line.strip()!r}) would leave the target host" except OSError as exc: return f"cannot be read ({exc.strerror})" diff --git a/tests/test_scope_web.py b/tests/test_scope_web.py index bb89d0d..df14abf 100644 --- a/tests/test_scope_web.py +++ b/tests/test_scope_web.py @@ -344,3 +344,27 @@ def test_testssl_phone_out_is_refused(project): """Review of #20: it contacts the CRL and OCSP servers of third parties.""" with pytest.raises(TargetError, match="phone-out"): targets_of(shlex.split("testssl.sh --phone-out x app.example.com")) + + +def test_gobuster_dir_needs_a_path_to_keep_words_out_of_the_host(project): + """Review of #20: gobuster appends each word to the URL, and without a + path a word like @evil.example.org would land in the authority.""" + with pytest.raises(TargetError, match="path"): + targets_of(shlex.split("gobuster dir -u https://app.example.com -w words.txt")) + assert _hosts("gobuster dir -u https://app.example.com/ -w words.txt") == ["app.example.com"] + + +@pytest.mark.parametrize( + "line", ["\x01//evil.example.org", "/\t/evil.example.org", " \\/evil.example.org", "HTTPS:x"] +) +def test_feroxbuster_wordlist_is_judged_after_the_url_cleanup(project, line): + """Review of #20: the URL standard strips leading control characters + and removes tabs, so these become //evil.example.org.""" + (project / "bad.txt").write_text(f"admin\n{line}\n") + with pytest.raises(TargetError, match="line 2"): + targets_of(shlex.split("feroxbuster -u https://app.example.com -w bad.txt")) + + +def test_feroxbuster_wordlist_words_that_reach_no_host_are_fine(project): + (project / "ok.txt").write_text("javascript:void(0)\nmailto:x\n/admin\nadmin\n") + assert _hosts("feroxbuster -u https://app.example.com -w ok.txt") == ["app.example.com"]