mirror of
https://github.com/odysseus-dev/odysseus.git
synced 2026-08-30 02:41:47 +00:00
49e4e55d2c
* fix(skill-importer): validate URL scheme and improve skills.sh handling * fix(skill-importer): enhance DNS resolution and SSRF protection in fetch URL handling * fix(url-safety): add allowed_dist parameter to check_outbound_url for flexible private blocking * test(skill-importer): add comprehensive tests for URL parsing and outbound checks * ensure newline at end of file in test_check_outbound_url_allows_public_ip * fix(skill-importer): improve TLS certificate handling in _get_checked function * fix(skill-importer): enhance _check_fetch_url to handle both hostnames and full URLs * fix(skill-importer): enhance parse_skill_source to support skills.sh URLs in path and netloc * fix(skill-importer): simplify skills.sh hostname check in parse_skill_source * fix(skill-importer): enhance parse_skill_source to identify skills.sh URLs in path and handle localhost/IP addresses * fix(skill-importer): enhance _resolve_and_check_url to validate all resolved IP addresses and prevent TOCTOU vulnerabilities * fix(skill-importer): enhance parse_skill_source to support schemeless GitHub and skills.sh URLs * fix(memory): resolve CodeQL URL sanitization warning and restore _check_fetch_url test alias * fix(memory): pin skill fetch sockets without rewriting URLs * fix(memory): reject unsupported skill wrapper hosts * refactor(url-safety): remove unused importer exception * test(memory): keep redirect regression hermetic * test(dns-rebinding): add test for _PinnedTransport to ensure connection to pinned IP * fix(skill-importer): enhance skills.sh support to extract GitHub links from page content * fix(skill-importer): improve URL scheme validation for GitHub and skills.sh links * fix(skills): reject unusable skill URLs instead of guessing Resolving a skills.sh link by scraping the first github.com URL out of the page body cannot work. Skill pages only ever link the repository root, never the skill's subdirectory, so every skill in a repo resolved to the same bundle: importing skills.sh/anthropics/skills/pdf walked the whole monorepo, saturated the 64-file cap, and installed algorithmic-art behind an ok:true response. Restore the redirect-target unwrap and fail with a message that says what to do instead. Also report the real reason a URL is rejected. The scheme check keyed off "://" appearing anywhere in the string, so a supplied-but-unusable URL came back as "URL is required", and a schemeless URL carrying "://" in its query was reported as an unsupported scheme. Key off the parsed scheme and let opaque schemes (mailto:, javascript:) and a schemeless host:port fall through to the host check. * test(skills): tighten the real-socket pinning regression The handler swallowed its own exceptions, so a failure inside it surfaced as a confusing assertion on the captured client address. Record the exception and assert on it, run the thread as a daemon, and close the listening socket from the test so a hang cannot outlive the run. Also drop the duplicate ipaddress import and the missing newline. * fix(skills): require exact GitHub skill URLs * test(skills): read complete pinned request headers --------- Co-authored-by: RaresKeY <158580472+RaresKeY@users.noreply.github.com> Co-authored-by: Léo <leograndcontact@gmail.com>
155 lines
6 KiB
Python
155 lines
6 KiB
Python
import re
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from services.memory.skill_importer import (
|
|
ResolvedSource,
|
|
SkillImportError,
|
|
check_outbound_url,
|
|
parse_skill_source,
|
|
)
|
|
|
|
## 1. Tests for Hostname Dispatch & Substring Spoofing
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
[
|
|
"https://skills.sh.attacker.com/owner/repo",
|
|
"https://evilskills.sh/owner/repo",
|
|
"https://notskills.sh/owner/repo",
|
|
"https://api.skills.sh/owner/repo",
|
|
"https://1.1.1.1/skills.sh/owner/repo",
|
|
"http://localhost/skills.sh/owner/repo",
|
|
],
|
|
)
|
|
def test_parse_skill_source_rejects_unsupported_host_before_fetch(url):
|
|
"""Unsupported authorities must never reach the network unwrap path."""
|
|
with patch("services.memory.skill_importer._get_checked") as mock_get:
|
|
with pytest.raises(SkillImportError):
|
|
parse_skill_source(url)
|
|
mock_get.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("entry", ["https://skills.sh/my-skill", "https://www.skills.sh/my-skill"])
|
|
def test_parse_skill_source_unwraps_skills_sh_redirect_to_github(entry):
|
|
"""Both skills.sh spellings unwrap when the fetch lands on a GitHub host."""
|
|
with patch("services.memory.skill_importer._get_checked") as mock_get:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.url = "https://github.com/test-owner/test-repo"
|
|
mock_get.return_value = mock_response
|
|
|
|
source = parse_skill_source(entry)
|
|
assert source.owner == "test-owner"
|
|
assert source.repo == "test-repo"
|
|
|
|
|
|
def test_parse_skill_source_rejects_skills_sh_page_that_never_reaches_github():
|
|
"""A skills.sh page that does not redirect to GitHub must fail loudly.
|
|
|
|
The live site serves the skill page from ``www.skills.sh`` and only ever
|
|
links the repository root, never the skill's subdirectory. Scraping a
|
|
``github.com`` link out of the body therefore resolves every skill in a
|
|
repo to the same bundle, so the importer must refuse rather than guess.
|
|
"""
|
|
body = '<html><body><a href="https://github.com/test-owner/test-repo">Repository</a></body></html>'
|
|
with patch("services.memory.skill_importer._get_checked") as mock_get:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.url = "https://www.skills.sh/anthropics/skills/pdf"
|
|
mock_response.text = body
|
|
mock_get.return_value = mock_response
|
|
|
|
with pytest.raises(
|
|
SkillImportError, match="did not redirect to GitHub"
|
|
) as exc_info:
|
|
parse_skill_source("https://skills.sh/anthropics/skills/pdf")
|
|
|
|
message = str(exc_info.value)
|
|
assert "exact skill folder or SKILL.md file" in message
|
|
assert "repository-root link alone is not sufficient" in message
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("url", "expected"),
|
|
[
|
|
("ftp://github.com/o/r", "unsupported URL scheme: ftp"),
|
|
("file:///etc/passwd", "unsupported URL scheme: file"),
|
|
("gopher://github.com/o/r", "unsupported URL scheme: gopher"),
|
|
("javascript:alert(1)", "Only GitHub or skills.sh URLs are supported"),
|
|
("mailto:x@y.z", "Only GitHub or skills.sh URLs are supported"),
|
|
("data:text/html,<b>x</b>", "Only GitHub or skills.sh URLs are supported"),
|
|
("https://evil.example/o/r", "Only GitHub or skills.sh URLs are supported"),
|
|
],
|
|
)
|
|
def test_parse_skill_source_reports_the_real_reason_for_rejection(url, expected):
|
|
"""A supplied-but-unusable URL must not be reported as a missing URL."""
|
|
with patch("services.memory.skill_importer._get_checked") as mock_get:
|
|
with pytest.raises(SkillImportError, match=re.escape(expected)):
|
|
parse_skill_source(url)
|
|
mock_get.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("url", "owner", "repo"),
|
|
[
|
|
("github.com/octocat/Hello-World", "octocat", "Hello-World"),
|
|
("HTTPS://github.com/octocat/Hello-World", "octocat", "Hello-World"),
|
|
("github.com:443/octocat/Hello-World", "octocat", "Hello-World"),
|
|
("github.com/octocat/Hello-World?q=a://b", "octocat", "Hello-World"),
|
|
],
|
|
)
|
|
def test_parse_skill_source_accepts_schemeless_and_uppercase_github(url, owner, repo):
|
|
"""A schemeless host, an uppercase scheme, and a ``://`` in the query all parse."""
|
|
source = parse_skill_source(url)
|
|
assert (source.owner, source.repo) == (owner, repo)
|
|
|
|
|
|
def test_parse_skill_source_valid_github():
|
|
"""Ensure standard GitHub URLs parse into the correct ResolvedSource fields."""
|
|
source = parse_skill_source("https://github.com/octocat/Hello-World/tree/main/docs")
|
|
assert isinstance(source, ResolvedSource)
|
|
assert source.owner == "octocat"
|
|
assert source.repo == "Hello-World"
|
|
assert source.ref == "main"
|
|
assert source.path == "docs"
|
|
|
|
|
|
## 2. Tests for SSRF Guard & CGNAT
|
|
|
|
def test_check_outbound_url_blocks_cgnat():
|
|
"""Ensure Carrier-Grade NAT (RFC 6598) block 100.64.0.0/10 is blocked."""
|
|
def mock_resolver(host):
|
|
return ["100.64.5.10"]
|
|
|
|
ok, reason = check_outbound_url("http://example.com", block_private=True, resolver=mock_resolver)
|
|
assert not ok
|
|
assert "private/shared/loopback" in reason # Updated to match your codebase's error string
|
|
|
|
def test_check_outbound_url_blocks_loopback():
|
|
"""Ensure loopback IPs (127.0.0.1) are blocked by default."""
|
|
def mock_resolver(host):
|
|
return ["127.0.0.1"]
|
|
|
|
ok, reason = check_outbound_url("http://localhost", block_private=True, resolver=mock_resolver)
|
|
assert not ok
|
|
|
|
|
|
def test_check_outbound_url_blocks_metadata():
|
|
"""Ensure cloud metadata endpoints (169.254.169.254) are blocked."""
|
|
def mock_resolver(host):
|
|
return ["169.254.169.254"]
|
|
|
|
ok, reason = check_outbound_url("http://metadata.google.internal", block_private=True, resolver=mock_resolver)
|
|
assert not ok
|
|
|
|
|
|
def test_check_outbound_url_allows_public_ip():
|
|
"""Ensure public routable IPs pass successfully."""
|
|
def mock_resolver(host):
|
|
return ["93.184.216.34"]
|
|
|
|
ok, reason = check_outbound_url("http://example.com", block_private=True, resolver=mock_resolver)
|
|
assert ok
|
|
assert reason == "ok"
|