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
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

### Bug Fixes

- Avoid false-positive `urllib` network findings for model cards whose only `urlopen` use is the documented `Image.open(urlopen(...))` sample-image example.
- Avoid network false positives for bounded README examples that download sample images over HTTPS from Hugging Face.
- Prevent Windows cache identity probes from creating locked temporary files inside scanned directories.
- Preserve locked Windows cache probes reached through directory aliases while clearing stale scan results.
Expand Down
224 changes: 208 additions & 16 deletions modelaudit/detectors/network_comm.py
Original file line number Diff line number Diff line change
Expand Up @@ -2338,24 +2338,216 @@ def _is_valid_official_readme_sample_image_example(example: bytes) -> bool:
url = bindings[0][1]
else:
return False
try:
parsed = urlsplit(url)
port = parsed.port
except ValueError:
if not _is_official_huggingface_documented_image_url(url):
return False
return True


def _is_official_huggingface_documented_image_url(url: str) -> bool:
"""Return whether a URL is a bounded HTTPS fetch of a documented huggingface.co image."""
try:
parsed = urlsplit(url)
port = parsed.port
except ValueError:
return False
segments = parsed.path.split("/")
return not (
parsed.scheme != "https"
or parsed.hostname != "huggingface.co"
or parsed.username is not None
or parsed.password is not None
or port is not None
or parsed.netloc.lower() != parsed.hostname
or parsed.fragment
or parsed.query not in {"", "download=true"}
or "resolve" not in segments
or any(segment in {".", ".."} for segment in segments)
or not parsed.path.lower().endswith(_DOCUMENTED_IMAGE_SUFFIXES)
)


@lru_cache(maxsize=1)
def official_readme_urlopen_image_example_spans(data: bytes) -> tuple[tuple[int, int], ...]:
"""Return byte spans of Python fences whose only ``urlopen`` use fetches a documented image.

Model-card generators (notably ``timm``) emit a fixed ``Image.open(urlopen(<literal URL>))``
snippet, so the documented shape is proven structurally instead of by pinning whole-file
digests. Callers use the returned spans to decide whether an individual ``urllib``/``urlopen``
finding sits inside a proven-inert example.

Results are memoised for the payload currently being classified, because every candidate
finding in one file re-checks the same spans.
"""
if b"urlopen" not in data:
return ()

spans: list[tuple[int, int]] = []
cursor = 0
while opening := _PYTHON_README_FENCE_PATTERN.search(data, cursor):
if len(spans) >= _MAX_README_IMAGE_EXAMPLE_FENCES:
return ()
Comment on lines +2386 to +2388

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Bound all inspected fences, not only accepted ones

When a Markdown file contains many invalid Python fences mentioning urlopen, len(spans) remains zero, so this loop can parse and walk every fence despite _MAX_README_IMAGE_EXAMPLE_FENCES. Since TextScanner accepts documentation payloads up to 100 MiB, an attacker-controlled model card can contain hundreds of thousands of small near-match fences and impose excessive CPU work during a scan. Track the number of encountered fence openings, as _index_official_readme_sample_image_fences already does, and fail closed after the configured limit.

Useful? React with 👍 / 👎.

closing = _matching_readme_image_fence_end(
data,
opening,
opening.end(),
min(len(data), opening.end() + _MAX_README_IMAGE_EXAMPLE_BYTES),
)
if closing is None:
next_cursor = min(len(data), opening.end() + _MAX_README_IMAGE_EXAMPLE_BYTES)
if next_cursor <= cursor:
break
cursor = next_cursor
continue
if _is_official_readme_urlopen_image_example(data[opening.end() : closing.start()]):
spans.append((opening.end(), closing.start()))
cursor = closing.end()

# A proven fence only speaks for itself. If any urllib reference in the file sits outside a
# proven fence - a second fence, prose, or trailing payload - the whole file stays actionable,
# mirroring how the `requests` example tracks unvalidated references.
#
# Guard on the bare substrings `urllib` and `urlopen` rather than on the specific tokens the
# findings report. The detector emits a single `network_library: urllib` finding per file and
# then retargets it to the earliest urllib token, so a narrower guard lets an unrelated
# `import urllib.request` + `urllib.request.build_opener()` fence inherit the benign example's
# position and be downgraded with it.
if not spans or _tokens_appear_outside_spans(data, (b"urllib", b"urlopen"), spans):
return ()
return tuple(spans)


def _tokens_appear_outside_spans(
data: bytes,
tokens: tuple[bytes, ...],
spans: list[tuple[int, int]],
) -> bool:
"""Return whether any token occurrence falls outside every proven span."""
for token in tokens:
position = data.find(token)
while position >= 0:
end = position + len(token)
if not any(start <= position and end <= stop for start, stop in spans):
return True
position = data.find(token, end)
return False


# Bare-name execution primitives. Attribute access is checked separately and far more narrowly:
# documented model cards legitimately call `model.eval()` (PyTorch eval mode), which has nothing to
# do with the `eval` builtin.
_DOCUMENTED_EXAMPLE_FORBIDDEN_NAMES = frozenset(
{
"__import__",
"compile",
"delattr",
"eval",
"exec",
"getattr",
"globals",
"locals",
"setattr",
"vars",
}
)
_DOCUMENTED_EXAMPLE_FORBIDDEN_ATTRIBUTES = frozenset({"popen", "system"})


def _is_official_readme_urlopen_image_example(example: bytes) -> bool:
"""Return whether every ``urlopen`` use in one fence is a documented sample-image fetch."""
if b"urlopen" not in example or len(example) > _MAX_README_IMAGE_EXAMPLE_BYTES:
return False
try:
tree = ast.parse(example.decode("utf-8"))
except (SyntaxError, UnicodeDecodeError, RecursionError, ValueError):
return False

nodes: list[ast.AST] = []
for node in ast.walk(tree):
if len(nodes) >= _MAX_README_IMAGE_EXAMPLE_AST_NODES:
return False
nodes.append(node)
parents = {id(child): parent for parent in nodes for child in ast.iter_child_nodes(parent)}

imports = [
node
for node in nodes
if isinstance(node, ast.ImportFrom) and node.level == 0 and node.module == "urllib.request"
]
if len(imports) != 1 or len(imports[0].names) != 1:
return False
alias = imports[0].names[0]
if alias.name != "urlopen" or alias.asname is not None:
return False

# `Image` must provably be PIL's. Without this the response sink is only checked by name, so a
# fence can define its own `class Image` whose `open` executes the downloaded bytes and still
# be treated as the documented example.
pil_image_imports = [
node
for node in nodes
if isinstance(node, ast.ImportFrom)
and node.level == 0
and node.module == "PIL"
and any(entry.name == "Image" for entry in node.names)
]
if len(pil_image_imports) != 1 or any(
entry.asname is not None for entry in pil_image_imports[0].names if entry.name == "Image"
):
return False

urlopen_calls: list[ast.Call] = []
for node in nodes:
# Neither `urlopen` nor `Image` may be rebound, shadowed, aliased, or reached via attribute.
if isinstance(node, ast.Attribute) and node.attr == "urlopen":
return False
if isinstance(node, ast.arg) and node.arg in {"urlopen", "Image"}:
return False
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node.name in {
"urlopen",
"Image",
}:
return False
if isinstance(node, ast.alias) and node.asname == "urlopen" and node is not alias:
return False
if isinstance(node, ast.alias) and node.asname == "Image":
return False
if isinstance(node, ast.Name) and node.id == "Image" and not isinstance(node.ctx, ast.Load):
return False
Comment on lines +2514 to +2515

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Reject mutations of PIL's Image.open sink

When a model-card fence imports PIL.Image but then assigns Image.open = pickle.load, this check misses the rebinding because the assignment target is an ast.Attribute, not an ast.Name with store context. The validator therefore accepts a subsequent Image.open(urlopen("https://huggingface.co/attacker/repo/resolve/main/payload.png")); the remote response is deserialized rather than decoded as an image, yet both urllib findings are downgraded to INFO and TextScanner reports success. Reject writes/deletes to Image.open (and other ways of replacing the imported Image binding) before treating the fence as inert.

AGENTS.md reference: AGENTS.md:L25-L25

Useful? React with 👍 / 👎.

# Execution primitives inside a fence we are about to call inert.
if isinstance(node, ast.Name) and node.id in _DOCUMENTED_EXAMPLE_FORBIDDEN_NAMES:
return False
if isinstance(node, ast.Attribute) and (
node.attr in _DOCUMENTED_EXAMPLE_FORBIDDEN_ATTRIBUTES or node.attr.startswith("__")
):
return False
if not isinstance(node, ast.Name) or node.id != "urlopen":
continue
parent = parents.get(id(node))
if not isinstance(node.ctx, ast.Load) or not isinstance(parent, ast.Call) or parent.func is not node:
return False
urlopen_calls.append(parent)
if not urlopen_calls:
return False

for call in urlopen_calls:
if call.keywords or len(call.args) != 1:
return False
argument = call.args[0]
if not isinstance(argument, ast.Constant) or not isinstance(argument.value, str):
return False
if not _is_official_huggingface_documented_image_url(argument.value):
return False
segments = parsed.path.split("/")
# The response must flow straight into `Image.open(...)` and nowhere else.
parent = parents.get(id(call))
if (
parsed.scheme != "https"
or parsed.hostname != "huggingface.co"
or parsed.username is not None
or parsed.password is not None
or port is not None
or parsed.netloc.lower() != parsed.hostname
or parsed.fragment
or parsed.query not in {"", "download=true"}
or "resolve" not in segments
or any(segment in {".", ".."} for segment in segments)
or not parsed.path.lower().endswith(_DOCUMENTED_IMAGE_SUFFIXES)
not isinstance(parent, ast.Call)
or parent.keywords
or len(parent.args) != 1
or parent.args[0] is not call
or not isinstance(parent.func, ast.Attribute)
or parent.func.attr != "open"
or not isinstance(parent.func.value, ast.Name)
or parent.func.value.id != "Image"
):
return False
return True
Expand Down
71 changes: 69 additions & 2 deletions modelaudit/scanners/text_scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,10 @@
from urllib.parse import parse_qsl, urlsplit, urlunsplit

from modelaudit.core_results import mark_operational_scan_error
from modelaudit.detectors.network_comm import redact_url_for_finding
from modelaudit.detectors.network_comm import (
official_readme_urlopen_image_example_spans,
redact_url_for_finding,
)
from modelaudit.detectors.secrets import SecretsDetector
from modelaudit.scanner_registry_metadata import TOKENIZER_VOCABULARY_CONTENT_FILENAMES
from modelaudit.scanner_results import INCONCLUSIVE_SCAN_OUTCOME, mark_inconclusive_scan_result
Expand Down Expand Up @@ -44,6 +47,10 @@
"readme.txt",
}
)
DOCUMENTATION_IMAGE_EXAMPLE_TOKENS: dict[tuple[str, str], bytes] = {
("network_function", "urlopen"): b"urlopen",
("network_library", "urllib"): b"from urllib",
}
PASSIVE_NETWORK_FINDING_TYPES = frozenset(
{
"cloud_storage_url",
Expand Down Expand Up @@ -2754,12 +2761,54 @@ def _passive_network_reporting_limit(
and cls._all_network_candidate_lines_are_bare(payload)
)

@classmethod
def _documentation_image_example_finding(
cls,
path: str,
payload: bytes,
finding: dict[str, Any],
) -> bool:
"""Return whether one ``urllib``/``urlopen`` finding sits inside a documented image example.

Model-card generators emit a fixed ``Image.open(urlopen(<literal huggingface.co URL>))``
snippet, so the example is proven structurally by
:func:`official_readme_urlopen_image_example_spans`. Findings outside a proven fence, and
every other finding type, stay actionable.
"""
filename = os.path.basename(path).lower()
if os.path.splitext(filename)[1] not in {".md", ".markdown"} or not (
cls._is_readme_documentation_filename(filename) or cls._is_model_card_documentation_filename(filename)
):
return False

finding_type = finding.get("type")
if finding_type == "network_function":
token = DOCUMENTATION_IMAGE_EXAMPLE_TOKENS.get((finding_type, str(finding.get("function"))))
elif finding_type == "network_library" and finding.get("pattern") == "from urllib":
token = DOCUMENTATION_IMAGE_EXAMPLE_TOKENS.get((finding_type, str(finding.get("library"))))
else:
return False
if token is None:
return False

position = finding.get("position")
if not isinstance(position, int) or position < 0:
return False
end = position + len(token)
if payload[position:end] != token:
return False
return any(
start <= position and end <= stop for start, stop in official_readme_urlopen_image_example_spans(payload)
)

@classmethod
def _sidecar_network_finding_is_informational(
cls,
path: str,
payload: bytes,
finding: dict[str, Any],
*,
allow_documentation_image_examples: bool = False,
) -> bool:
if cls._is_documentation_sidecar(path):
finding_type = finding.get("type")
Expand All @@ -2782,6 +2831,10 @@ def _sidecar_network_finding_is_informational(
)
or (finding_type == "network_library" and cls._documentation_network_library_is_prose(payload, finding))
or (finding_type == "cc_pattern" and cls._documentation_cc_finding_is_benign_prose(payload, finding))
or (
allow_documentation_image_examples
and cls._documentation_image_example_finding(path, payload, finding)
)
or (
finding_type == "suspicious_port" and not cls._documentation_finding_is_actionable(payload, finding)
)
Expand Down Expand Up @@ -3261,6 +3314,8 @@ def _downgrade_sidecar_network_findings(
path: str,
payload: bytes,
findings: list[dict[str, Any]],
*,
allow_documentation_image_examples: bool = False,
) -> tuple[list[dict[str, Any]], bool, set[str]]:
classified_findings: list[dict[str, Any]] = []
classification_incomplete = False
Expand Down Expand Up @@ -3297,7 +3352,12 @@ def _downgrade_sidecar_network_findings(
if retargeted_finding is None:
continue
finding = retargeted_finding
if not retarget_incomplete and cls._sidecar_network_finding_is_informational(path, payload, finding):
if not retarget_incomplete and cls._sidecar_network_finding_is_informational(
path,
payload,
finding,
allow_documentation_image_examples=allow_documentation_image_examples,
):
finding = {**finding, "severity": "INFO"}
classified_findings.append(finding)
return classified_findings, classification_incomplete, classification_limit_sources
Expand Down Expand Up @@ -3464,11 +3524,18 @@ def _run_content_security_checks(self, path: str, result: ScanResult, file_size:
max_findings=max_findings,
)
network_findings, finding_limit = self._split_detector_finding_limit(network_findings)
allow_documentation_image_examples = (
self._get_bool_config("use_hf_whitelist", default=True)
and not detector_incomplete
and not truncated
and finding_limit is None
)
network_findings, classification_incomplete, classification_limit_sources = (
self._downgrade_sidecar_network_findings(
path,
inspected_payload,
network_findings,
allow_documentation_image_examples=allow_documentation_image_examples,
)
)
network_findings = self._deduplicate_documentation_network_findings(
Expand Down
Loading
Loading