fix(skills): harden skill import against DNS rebinding and SSRF TOCTOU

services/memory/skill_importer.py

@@ -1,16 +1,18 @@
 """Import SKILL.md bundles from public GitHub (or skills.sh → GitHub) URLs."""
 from __future__ import annotations
 
+import ipaddress
 import logging
 import os
-import re
+import time
 from dataclasses import dataclass
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, Iterable, List, Optional, Tuple, cast
 from urllib.parse import quote, urljoin, urlparse
 
+import httpcore
 import httpx
 
-from src.url_safety import check_outbound_url
+from src.url_safety import _default_resolver, check_outbound_url
 
 logger = logging.getLogger(__name__)
 
@@ -25,6 +27,7 @@
 _GITHUB_HOSTS = frozenset({
     "github.com", "www.github.com", "api.github.com", "raw.githubusercontent.com",
 })
+_SKILLS_SH_HOSTS = frozenset({"skills.sh"})
 
 
 def _github_host(url: str) -> str:
@@ -72,18 +75,158 @@ def _is_text_file(name: str) -> bool:
 _MAX_FETCH_REDIRECTS = 5
 
 
-def _check_fetch_url(url: str) -> None:
-    """SSRF guard for skill-import fetches (defense-in-depth).
+def _validated_ips(raw_ips: List[str]) -> List[ipaddress._BaseAddress]:
+    """Parse and de-duplicate one resolver snapshot in resolver order."""
+    ips: List[ipaddress._BaseAddress] = []
+    seen = set()
+    for raw in raw_ips:
+        if not isinstance(raw, str):
+            continue
+        try:
+            ip = ipaddress.ip_address(raw.split("%", 1)[0])
+        except ValueError:
+            continue
+        if ip in seen:
+            continue
+        seen.add(ip)
+        ips.append(ip)
+    return ips
 
-    Skill bundles only ever come from public GitHub, never an internal
-    address, so block private/loopback/link-local targets on every hop —
-    matching the hardened web-fetch path in
-    ``services/search/content.py:_get_public_url`` rather than the lenient
-    default used for admin-configured model endpoints.
-    """
-    ok, reason = check_outbound_url(url, block_private=True)
+
+def _resolve_and_check_url(url: str) -> List[ipaddress._BaseAddress]:
+    """Return the exact address snapshot approved for one fetch hop."""
+    resolved_ips: List[str] = []
+
+    def _recording_resolver(host: str) -> List[str]:
+        answers = list(_default_resolver(host))
+        resolved_ips[:] = answers
+        return answers
+
+    ok, reason = check_outbound_url(
+        url,
+        block_private=True,
+        resolver=_recording_resolver,
+    )
     if not ok:
-        raise SkillImportError(reason)
+        raise SkillImportError(f"outbound URL blocked: {reason}")
+
+    pinned_ips = _validated_ips(resolved_ips)
+    if not pinned_ips:
+        raise SkillImportError("outbound URL blocked: host did not resolve to a usable address")
+    return pinned_ips
+
+
+# Backward compatibility alias for tests importing _check_fetch_url directly
+_check_fetch_url = _resolve_and_check_url
+
+
+class _PinnedBackend(httpcore.NetworkBackend):
+    """Connect only to addresses from one validated DNS snapshot."""
+
+    def __init__(self, ips: List[ipaddress._BaseAddress]):
+        self._ips = [str(ip) for ip in ips]
+        self._real = httpcore.SyncBackend()
+
+    def connect_tcp(
+        self,
+        host: str,
+        port: int,
+        timeout: float | None = None,
+        local_address: str | None = None,
+        socket_options=None,
+    ):
+        deadline = None if timeout is None else time.monotonic() + timeout
+        last_exc: Optional[Exception] = None
+        for ip in self._ips:
+            remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
+            try:
+                return self._real.connect_tcp(
+                    ip,
+                    port,
+                    remaining,
+                    local_address,
+                    socket_options,
+                )
+            except (httpcore.ConnectError, httpcore.ConnectTimeout) as exc:
+                last_exc = exc
+                if deadline is not None and time.monotonic() >= deadline:
+                    break
+        if last_exc is not None:
+            raise last_exc
+        raise httpcore.ConnectError("no validated address available")
+
+    def connect_unix_socket(self, path, timeout=None, socket_options=None):
+        return self._real.connect_unix_socket(path, timeout, socket_options)
+
+    def sleep(self, seconds: float) -> None:
+        return self._real.sleep(seconds)
+
+
+_HTTPCORE_TO_HTTPX_EXC = {
+    httpcore.ConnectError: httpx.ConnectError,
+    httpcore.ConnectTimeout: httpx.ConnectTimeout,
+    httpcore.LocalProtocolError: httpx.LocalProtocolError,
+    httpcore.NetworkError: httpx.NetworkError,
+    httpcore.PoolTimeout: httpx.PoolTimeout,
+    httpcore.ProtocolError: httpx.ProtocolError,
+    httpcore.ProxyError: httpx.ProxyError,
+    httpcore.ReadError: httpx.ReadError,
+    httpcore.ReadTimeout: httpx.ReadTimeout,
+    httpcore.RemoteProtocolError: httpx.RemoteProtocolError,
+    httpcore.TimeoutException: httpx.TimeoutException,
+    httpcore.UnsupportedProtocol: httpx.UnsupportedProtocol,
+    httpcore.WriteError: httpx.WriteError,
+    httpcore.WriteTimeout: httpx.WriteTimeout,
+}
+
+
+class _PinnedTransport(httpx.BaseTransport):
+    """Pin socket connects while preserving URL authority, Host, and TLS SNI."""
+
+    def __init__(self, ips: List[ipaddress._BaseAddress]):
+        self._pinned_ips = list(ips)
+        self._pool = httpcore.ConnectionPool(
+            ssl_context=httpx.create_ssl_context(),
+            http1=True,
+            http2=False,
+            network_backend=_PinnedBackend(ips),
+        )
+
+    def handle_request(self, request: httpx.Request) -> httpx.Response:
+        core_request = httpcore.Request(
+            method=request.method,
+            url=httpcore.URL(
+                scheme=request.url.raw_scheme,
+                host=request.url.raw_host,
+                port=request.url.port,
+                target=request.url.raw_path,
+            ),
+            headers=request.headers.raw,
+            content=request.stream,
+            extensions=request.extensions,
+        )
+        core_response = None
+        try:
+            core_response = self._pool.handle_request(core_request)
+            content = b"".join(cast(Iterable[bytes], core_response.stream))
+        except Exception as exc:
+            mapped = _HTTPCORE_TO_HTTPX_EXC.get(type(exc))
+            if mapped is not None:
+                raise mapped(str(exc)) from exc
+            raise
+        finally:
+            if core_response is not None:
+                core_response.close()
+
+        return httpx.Response(
+            status_code=core_response.status,
+            headers=core_response.headers,
+            content=content,
+            extensions=core_response.extensions,
+        )
+
+    def close(self) -> None:
+        self._pool.close()
 
 
 def _get_checked(
@@ -100,49 +243,70 @@ def _get_checked(
     hand lets us re-validate every hop, closing that blind-SSRF gap.
     """
     current = url
-    with httpx.Client(follow_redirects=False, timeout=timeout) as client:
-        for _ in range(_MAX_FETCH_REDIRECTS + 1):
-            _check_fetch_url(current)
+    for _ in range(_MAX_FETCH_REDIRECTS + 1):
+        pinned_ips = _resolve_and_check_url(current)
+        with httpx.Client(
+            transport=_PinnedTransport(pinned_ips),
+            follow_redirects=False,
+            timeout=timeout,
+        ) as client:
             r = client.get(current, headers=headers)
-            if r.status_code in (301, 302, 303, 307, 308):
-                location = r.headers.get("location")
-                if not location:
-                    return r
-                current = urljoin(str(r.url), location)
-                continue
-            return r
+
+        if r.status_code in (301, 302, 303, 307, 308):
+            location = r.headers.get("location")
+            if not location:
+                return r
+            current = urljoin(str(r.url), location)
+            continue
+        return r
     raise SkillImportError("too many redirects while fetching skill bundle")
 
 
 def parse_skill_source(url: str) -> ResolvedSource:
     """Normalize skills.sh / GitHub web URLs into owner/repo/ref/path."""
-    raw = (url or "").strip()
-    if not raw:
+    url = (url or "").strip()
+    if not url:
         raise SkillImportError("URL is required")
 
-    # skills.sh often links to GitHub; try to unwrap ?url= or redirect target later.
-    if "skills.sh" in raw and "github.com" not in raw:
-        r = _get_checked(raw, timeout=20.0)
+    # Support backwards compatibility for schemeless GitHub or skills.sh paths.
+    if not url.startswith(("http://", "https://")):
+        parsed_rough = urlparse("//" + url)
+        hostname = (parsed_rough.hostname or "").lower()
+
+        if hostname in _GITHUB_HOSTS or hostname in _SKILLS_SH_HOSTS:
+            url = "https://" + url
+        else:
+            if parsed_rough.scheme:
+                raise SkillImportError(f"unsupported URL scheme: {parsed_rough.scheme}")
+            else:
+                raise SkillImportError("URL is required")
+
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise SkillImportError(f"unsupported URL scheme: {parsed.scheme}")
+
+    hostname = (parsed.hostname or "").lower()
+    if hostname not in _GITHUB_HOSTS and hostname not in _SKILLS_SH_HOSTS:
+        raise SkillImportError(
+            "Only GitHub or skills.sh URLs are supported"
+        )
+
+    # skills.sh links must resolve to an exact supported GitHub host.
+    if hostname in _SKILLS_SH_HOSTS:
+        r = _get_checked(url, timeout=20.0)
         if r.status_code >= 400:
             raise _github_response_error(r)
         final = str(r.url)
         _assert_github_url(final, context="redirect target")
-        # Page may embed a github link; prefer final URL if redirected.
-        if "github.com" in final:
-            raw = final
-        else:
-            m = re.search(r"https?://github\.com/[^\s\"')]+", r.text or "")
-            if m:
-                raw = m.group(0).rstrip(".,)")
+        url = final
 
-    parsed = urlparse(raw)
-    host = _github_host(raw)
-    if host not in _GITHUB_HOSTS:
-        raise SkillImportError(
-            "Only GitHub URLs are supported (https://github.com/... or raw.githubusercontent.com/...)"
-        )
+    # Update parsed and hostname to reflect the new GitHub URL
+    parsed = urlparse(url)
+    hostname = (parsed.hostname or "").lower()
+
+    _assert_github_url(url)
 
-    if host == "raw.githubusercontent.com":
+    if hostname == "raw.githubusercontent.com":
         # /owner/repo/ref/path/to/file
         bits = [p for p in parsed.path.split("/") if p]
         if len(bits) < 4:

tests/test_skill_importer.py

@@ -1,4 +1,6 @@
 """Skill URL importer — GitHub path parsing."""
+import ipaddress
+
 import pytest
 
 from services.memory.skill_importer import (
@@ -11,6 +13,13 @@
 )
 
 
+def _allow_fetch(monkeypatch):
+    monkeypatch.setattr(
+        "services.memory.skill_importer._resolve_and_check_url",
+        lambda url: [ipaddress.ip_address("93.184.216.34")],
+    )
+
+
 def test_parse_github_blob_skill_md():
     src = parse_skill_source(
         "https://github.com/anthropics/skills/blob/main/skills/pdf/SKILL.md"
@@ -69,10 +78,7 @@ def get(self, url, headers=None):
             return _Resp()
 
     monkeypatch.setattr("services.memory.skill_importer.httpx.Client", _Client)
-    monkeypatch.setattr(
-        "services.memory.skill_importer.check_outbound_url",
-        lambda url, **kwargs: (True, ""),
-    )
+    _allow_fetch(monkeypatch)
     with pytest.raises(SkillImportError, match="redirect target"):
         _fetch_bytes("https://raw.githubusercontent.com/o/r/main/SKILL.md")
 
@@ -89,10 +95,7 @@ def test_list_github_dir_accepts_api_github_response(monkeypatch):
         "services.memory.skill_importer._fetch_text",
         lambda url: "# skill\n",
     )
-    monkeypatch.setattr(
-        "services.memory.skill_importer.check_outbound_url",
-        lambda url, **kwargs: (True, ""),
-    )
+    _allow_fetch(monkeypatch)
 
     class _Resp:
         url = "https://api.github.com/repos/o/r/contents?ref=main"
@@ -144,10 +147,7 @@ def get(self, url, headers=None):
             return response
 
     monkeypatch.setattr("services.memory.skill_importer.httpx.Client", _Client)
-    monkeypatch.setattr(
-        "services.memory.skill_importer.check_outbound_url",
-        lambda url, **kwargs: (True, ""),
-    )
+    _allow_fetch(monkeypatch)
 
 
 def test_list_github_dir_surfaces_rate_limit(monkeypatch):