Skip to content
Open
Show file tree
Hide file tree
Changes from 11 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
168 changes: 152 additions & 16 deletions modelaudit/detectors/network_comm.py
Original file line number Diff line number Diff line change
Expand Up @@ -2338,24 +2338,160 @@ 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 `urlopen`/`urllib` token 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.
if not spans or _tokens_appear_outside_spans(data, (b"urlopen", b"from urllib"), 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


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

urlopen_calls: list[ast.Call] = []
for node in nodes:
# `urlopen` must never be rebound, shadowed, aliased, or reached through an attribute.
if isinstance(node, ast.Attribute) and node.attr == "urlopen":
return False
if isinstance(node, ast.arg) and node.arg == "urlopen":
return False
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node.name == "urlopen":
return False
if isinstance(node, ast.alias) and node.asname == "urlopen" and node is not alias:
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:
Comment on lines 2809 to 2812
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