fix(harness): normalize local HTTP host case in skillscan (#6280)

* fix(harness): normalize local HTTP host case in skillscan

_LOCAL_HTTP_HOSTS holds lowercase names and every membership test
compared the extracted host verbatim, so an endpoint written with any
uppercase letter (e.g. http://LOCALHOST:8080) was classified as an
external endpoint: non-root deployments failed with EACCES while root
deployments silently wrote plans outside the workspace.

Normalize the host to lowercase in _http_host and in the
network-cleartext/local scan branch so all three call sites agree.

Fixes #6278

* test(skillscan): cover local host case and external endpoint controls

* fix(skillscan): parse IPv6 hosts and detect metadata case variants

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
jay 2026-10-05 15:17:03 +08:00 committed by GitHub
parent 57a2d7fa70
commit fff9d185c5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 134 additions and 9 deletions

View File

@ -1293,6 +1293,10 @@ If a trusted operator manages the configured skills directory through an externa
Skill installs and agent-managed skill edits run through **SkillScan**, a native deterministic safety scanner before the LLM-based skill scanner. Phase 1 runs offline with no Semgrep/OpenGrep dependency, blocks high-confidence `CRITICAL` findings such as private keys or shell execution, and passes warning findings to the LLM scanner for contextual review. Code files (anything under `scripts/`, a script suffix such as `.py`, `.sh`, or `.js`, or an extensionless file starting with `#!`) that are not NUL-free UTF-8 text raise a warning and are still analyzed over a lossy decode, so a single stray byte cannot hide them from `CRITICAL` checks. The moderation adapter normalizes both plain-text model responses and LangChain Responses API text blocks before parsing the required JSON decision. Python instance-client exfiltration checks follow a minimal same-scope evidence chain: a simple name bound to a known client constructor, optional name-to-name aliases, and an actual outbound method or context-manager use supported by that constructor. Constructor roots must be proven imports; bare canonical-looking names are not inferred as modules. Nested scopes do not inherit client handles and inherit only constructor import aliases that are never rebound in the enclosing scope. Comprehensions, walrus-bearing statements, annotations, complex binding targets, unsupported operations, and ambiguous branch flows produce no finding from this signal; skipped constructs conservatively invalidate every name they may bind so stale client state cannot create a finding. A deterministic work budget or recursion limit reached by this best-effort analysis does not discard findings already collected for the file. Set `skill_scan.enabled: false` in `config.yaml` to disable only the deterministic analyzers; safe archive extraction and the LLM scanner still run.
SkillScan treats HTTP hostnames case-insensitively and recognizes bracketed IPv6
loopback (`[::1]`) URLs as local. External IPv6 endpoints still trigger network findings,
and cloud-metadata hostname detection is case-insensitive.
For Python credential mappings, SkillScan checks literal values while treating ordinary dictionary keys as labels. A mapping such as `tokens = {"access_token": os.getenv("ACCESS_TOKEN")}` does not report a hardcoded credential. Keys matching a recognized cloud or API token format are still checked as embedded credentials.
DeerFlow also ships with **skill-reviewer**, a public skill for read-only skill quality review. It uses the built-in `review_skill_package` tool to inspect installed skills, local packages, archives, or pasted `SKILL.md` content without activating the target skill, binding its secrets, executing its scripts, or installing it. The tool returns a compact, tag-neutralized JSON payload to the model context and keeps the full raw review payload in the tool artifact for programmatic consumers. The deterministic review core reuses DeerFlow parsing and SkillScan facts, emits versioned JSON contracts under `contracts/skill_review/`, and can be run from the backend CLI:

File diff suppressed because one or more lines are too long

View File

@ -21,6 +21,7 @@ from collections.abc import Iterable
from dataclasses import dataclass
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import Any
from urllib.parse import urlsplit
from deerflow.skills.package_files import is_code_file, is_executable_binary_prefix
from deerflow.skills.package_paths import is_eval_fixture_skill_md
@ -134,7 +135,7 @@ _SECRET_TOKEN_PATTERNS = tuple(
)
)
_SENSITIVE_PATH_RE = re.compile(r"(~/.ssh|/etc/passwd|/etc/shadow|/var/run/docker\.sock|docker\.sock|169\.254\.169\.254)")
_EXTERNAL_HTTP_RE = re.compile(r"http://([A-Za-z0-9.-]+)(?::\d+)?(?:/|\b)")
_EXTERNAL_HTTP_RE = re.compile(r"http://(?:[^/?#\s)'\"<>]*@)?(\[[0-9A-Fa-f:.]+\]|[A-Za-z0-9.-]+)(?::\d+)?(?:/|\b|(?=$|[\s)'\"<>?#]))")
_URL_RE = re.compile(r"https?://[^\s)'\"<>]+")
_LOCAL_HTTP_HOSTS = {"localhost", "127.0.0.1", "0.0.0.0", "::1"}
# `rm` with a recursive flag (any order/combination, optional --no-preserve-root)
@ -800,16 +801,20 @@ def _scan_shell(rel_path: str, text: str) -> list[SecurityFinding]:
def _scan_network_and_resource(rel_path: str, text: str) -> list[SecurityFinding]:
findings: list[SecurityFinding] = []
if match := re.search(r"(169\.254\.169\.254|metadata\.google\.internal)", text):
if match := re.search(r"(169\.254\.169\.254|metadata\.google\.internal)", text, re.IGNORECASE):
findings.append(_finding_from_match("network-cloud-metadata", rel_path, text, match))
if match := re.search(r":\(\)\{\s*:\|:&\s*\};:", text):
findings.append(_finding_from_match("resource-fork-bomb", rel_path, text, match))
for match in _EXTERNAL_HTTP_RE.finditer(text):
host = match.group(1)
host = _http_host(match.group(0)) or ""
if host in _LOCAL_HTTP_HOSTS or host.startswith("10.") or host.startswith("192.168.") or re.match(r"172\.(1[6-9]|2\d|3[01])\.", host):
findings.append(_finding_from_match("network-local-http", rel_path, text, match))
rule_id = "network-local-http"
else:
findings.append(_finding_from_match("network-cleartext-http", rel_path, text, match))
rule_id = "network-cleartext-http"
finding = _finding_from_match(rule_id, rel_path, text, match)
if "@" in match.group(0):
finding["evidence"] = "http://" + match.group(0).rsplit("@", 1)[1]
findings.append(finding)
break
return findings
@ -1003,8 +1008,15 @@ def _looks_like_placeholder(value: str) -> bool:
def _http_host(url: str) -> str | None:
match = re.match(r"https?://\[?([^]/:]+)", url)
return match.group(1) if match else None
if not url.startswith(("http://", "https://")):
return None
try:
# Parse the authority so IPv6 brackets and userinfo cannot be mistaken
# for the host. Malformed URLs remain outbound in _is_outbound_url.
host = urlsplit(url).hostname
except ValueError:
return None
return host.lower() if host else None
def _is_outbound_url(value: str) -> bool:

View File

@ -344,12 +344,13 @@ def test_python_subprocess_kwargs_unpacking_without_shell_key_still_blocks(tmp_p
assert not [item for item in findings if item["rule_id"] == "python-subprocess"]
def test_cloud_metadata_access_is_reported_by_one_rule(tmp_path: Path) -> None:
@pytest.mark.parametrize("host", ["169.254.169.254", "metadata.google.internal", "METADATA.GOOGLE.INTERNAL", "Metadata.Google.Internal"])
def test_cloud_metadata_access_is_reported_by_one_rule(tmp_path: Path, host: str) -> None:
skill_dir = tmp_path / "demo-skill"
_write_skill(skill_dir)
scripts_dir = skill_dir / "scripts"
scripts_dir.mkdir()
(scripts_dir / "run.py").write_text('import urllib.request\nurllib.request.urlopen("http://169.254.169.254/latest/meta-data/")\n', encoding="utf-8")
(scripts_dir / "run.py").write_text(f'import urllib.request\nurllib.request.urlopen("http://{host}/latest/meta-data/")\n', encoding="utf-8")
findings = scan_skill_dir(skill_dir)["findings"]
@ -1824,3 +1825,110 @@ def test_bundled_public_skill_scripts_report_no_secret_assignment() -> None:
offenders[skill_dir.name] = [(finding["file"], finding["line"]) for finding in hits]
assert offenders == {}
@pytest.mark.parametrize(
"url, host",
[
("http://LOCALHOST:8080/api", "localhost"),
("http://[::1]/api", "::1"),
("https://[::1]:8443/api", "::1"),
("http://[2001:DB8::1]:8080/api", "2001:db8::1"),
("http://user@[::1]:8080/api", "::1"),
("http://[::1]@Example.COM/api", None),
("http://localhost@Example.COM/api", "example.com"),
("http://[::1/api", None),
("ftp://LOCALHOST/api", None),
],
)
def test_http_host_normalizes_case_and_ipv6(url: str, host: str | None) -> None:
from deerflow.skills.skillscan.orchestrator import _http_host
assert _http_host(url) == host
def test_uppercase_local_host_is_classified_local(tmp_path: Path) -> None:
skill_dir = tmp_path / "skill"
_write_skill(skill_dir)
(skill_dir / "run.py").write_text(
'import urllib.request\nurllib.request.urlopen("http://LOCALHOST:8080/config")\n',
encoding="utf-8",
)
findings = scan_skill_dir(skill_dir)["findings"]
assert _finding_by_rule(findings, "network-local-http")
assert not [finding for finding in findings if finding["rule_id"] == "network-cleartext-http"]
@pytest.mark.parametrize("host, external", [("localhost", False), ("LOCALHOST", False), ("LocalHost", False), ("Example.COM", True), ("[::1]", False), ("[2001:DB8::1]", True), ("localhost@Example.COM", True)])
def test_declared_http_host_case_classification(tmp_path: Path, host: str, external: bool) -> None:
skill_dir = tmp_path / "skill"
_write_skill(skill_dir, f"Endpoint: http://{host}:8080/api\n")
findings = scan_skill_dir(skill_dir)["findings"]
assert bool([finding for finding in findings if finding["rule_id"] == "declaration-external-endpoint"]) is external
assert bool([finding for finding in findings if finding["rule_id"] == "network-local-http"]) is (not external)
@pytest.mark.parametrize("host, external", [("localhost", False), ("LOCALHOST", False), ("LocalHost", False), ("Example.COM", True), ("[::1]", False), ("[2001:DB8::1]", True), ("[::1]@Example.COM", True)])
def test_sensitive_path_http_host_case_classification(tmp_path: Path, host: str, external: bool) -> None:
skill_dir = tmp_path / "skill"
_write_skill(skill_dir)
(skill_dir / "run.py").write_text(f'ENDPOINT = "http://{host}:8080/api"\nopen("/etc/passwd").read()\n', encoding="utf-8")
result = scan_skill_dir(skill_dir)
findings = result["findings"]
if external:
assert _finding_by_rule(findings, "python-sensitive-exfil")["severity"] == "CRITICAL"
else:
assert _finding_by_rule(findings, "python-sensitive-path-read")["severity"] == "HIGH"
assert not [finding for finding in findings if finding["rule_id"] == "python-sensitive-exfil"]
assert result["blocked"] is external
@pytest.mark.parametrize(
"url, local",
[
("http://[::1]:8080/api", True),
("http://[::1]", True),
("http://[::1]?mode=local", True),
("http://[2001:DB8::1]:8080/api", False),
("http://[2001:db8::1]", False),
],
)
def test_ipv6_cleartext_http_classification(tmp_path: Path, url: str, local: bool) -> None:
skill_dir = tmp_path / "skill"
_write_skill(skill_dir, f"# Demo\nEndpoint: {url}\n")
findings = scan_skill_dir(skill_dir)["findings"]
rule_id = "network-local-http" if local else "network-cleartext-http"
finding = _finding_by_rule(findings, rule_id)
assert finding["file"] == "SKILL.md"
assert finding["line"] == 7
other_rule = "network-cleartext-http" if local else "network-local-http"
assert not [item for item in findings if item["rule_id"] == other_rule]
def test_malformed_ipv6_url_remains_outbound(tmp_path: Path) -> None:
skill_dir = tmp_path / "skill"
_write_skill(skill_dir)
(skill_dir / "run.py").write_text('ENDPOINT = "http://[::1/api"\nopen("/etc/passwd").read()\n', encoding="utf-8")
result = scan_skill_dir(skill_dir)
assert _finding_by_rule(result["findings"], "python-sensitive-exfil")["severity"] == "CRITICAL"
assert result["blocked"]
assert result["scanner_errors"] == []
def test_cleartext_http_uses_host_after_userinfo_without_exposing_credentials(tmp_path: Path) -> None:
skill_dir = tmp_path / "skill"
_write_skill(skill_dir, "Endpoint: http://localhost:private-value@Example.COM:8080/api\n")
findings = scan_skill_dir(skill_dir)["findings"]
finding = _finding_by_rule(findings, "network-cleartext-http")
assert finding["evidence"] == "http://Example.COM:8080/"
assert not [item for item in findings if item["rule_id"] == "network-local-http"]