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..d5f694b 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). @@ -105,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 ------------------------------------------------------- @@ -409,21 +411,578 @@ 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", + 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. ``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: + 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") + 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) + + +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, need_path=mode == "dir") + 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 +# 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): + 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})" + 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: 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|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) + 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: + 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) + + +# 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", "--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", "--phone-out", +} # 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 == "--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": + 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 +# 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.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..df14abf --- /dev/null +++ b/tests/test_scope_web.py @@ -0,0 +1,370 @@ +""" +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 + + +# --- 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"] + + +@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")) + + +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"]