Skip to content
Merged
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
23 changes: 23 additions & 0 deletions src/skillspector/input_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -211,6 +211,28 @@ def _is_private_ip(host: str) -> bool:
return False


def _raw_file_url(url: str) -> str:
"""Point a GitHub or GitLab ``/blob/`` file page at the file's raw bytes.

Those pages are HTML viewers, so downloading one scans the forge's page
markup instead of the file. Every other URL is returned unchanged.
"""
parsed = urlparse(url)
host = (parsed.hostname or "").lower()
segments = parsed.path.split("/")
# /<owner>/<repo>/blob/<ref>/<path> -> raw.githubusercontent.com/<owner>/<repo>/<ref>/<path>
if host == "github.com" and len(segments) > 5 and segments[3] == "blob":
raw_path = "/".join(segments[:3] + segments[4:])
return parsed._replace(netloc="raw.githubusercontent.com", path=raw_path).geturl()
# /<namespace>/<project>/-/blob/<ref>/<path> -> /<namespace>/<project>/-/raw/<ref>/<path>
if host == "gitlab.com" and "-" in segments[3:]:
marker = segments.index("-", 3)
if marker + 3 < len(segments) and segments[marker + 1] == "blob":
segments[marker + 1] = "raw"
return parsed._replace(path="/".join(segments)).geturl()
return url


def _root_owned_root_alias(path: Path) -> Path | None:
"""Return a root-owned symlink directly below ``/``, if *path* is one."""
absolute_path = Path(os.path.abspath(path))
Expand Down Expand Up @@ -1123,6 +1145,7 @@ def _download_file(self, url: str) -> Path:
partial file produced by a mid-stream breach is removed before
the exception propagates.
"""
url = _raw_file_url(url)
if self._transitive_budget is not None:
return self._download_transitive_file(url)
self._validate_url_host(url, ALLOWED_DOWNLOAD_HOSTS)
Expand Down
47 changes: 47 additions & 0 deletions tests/unit/test_input_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import httpx
import pytest

from skillspector.input_handler import (
Expand All @@ -32,6 +33,7 @@
_open_regular_file_from_windows_handle,
_open_regular_file_no_follow,
)
from skillspector.state import WorkflowResourceBudget


def _mock_windows_secure_open(
Expand Down Expand Up @@ -460,6 +462,51 @@ def test_http_urls_are_not_accepted_as_remote_inputs() -> None:
assert handler._is_file_url("http://raw.githubusercontent.com/org/repo/SKILL.md") is False


@pytest.mark.parametrize("budgeted", [False, True], ids=["direct", "workflow-budget"])
@pytest.mark.parametrize(
("page_url", "raw_url"),
[
(
"https://github.com/org/repo/blob/main/skills/demo/SKILL.md",
"https://raw.githubusercontent.com/org/repo/main/skills/demo/SKILL.md",
),
(
"https://gitlab.com/group/repo/-/blob/main/skills/demo/SKILL.md",
"https://gitlab.com/group/repo/-/raw/main/skills/demo/SKILL.md",
),
],
ids=["github", "gitlab"],
)
def test_file_page_url_downloads_the_raw_file(
monkeypatch: pytest.MonkeyPatch, page_url: str, raw_url: str, budgeted: bool
) -> None:
"""A forge's /blob/ file page resolves to the file itself, not its HTML viewer."""
skill = b"---\nname: demo\ndescription: demo\n---\n# Demo\n"
requested: list[str] = []

def serve(request: httpx.Request) -> httpx.Response:
requested.append(str(request.url))
if str(request.url) == raw_url:
return httpx.Response(200, content=skill, headers={"content-type": "text/plain"})
return httpx.Response(200, content=b"<!DOCTYPE html><html></html>")

real_client = httpx.Client
monkeypatch.setattr(
"skillspector.input_handler.httpx.Client",
lambda *args, **kwargs: real_client(*args, transport=httpx.MockTransport(serve), **kwargs),
)
monkeypatch.setattr("skillspector.input_handler._is_private_ip", lambda _host: False)
handler = InputHandler(transitive_budget=WorkflowResourceBudget() if budgeted else None)
try:
resolved, source_type = handler.resolve(page_url)

assert source_type == "url"
assert requested == [raw_url]
assert (resolved / "SKILL.md").read_bytes() == skill
finally:
handler.cleanup()


def test_validate_url_host_scp_extracts_github() -> None:
"""_validate_url_host extracts 'github.com' from an scp-style URL."""
with patch("skillspector.input_handler._is_private_ip", return_value=False):
Expand Down
Loading