fix: SSRF DNS-rebinding TOCTOU 차단 + 실패 인증 스로틀

7/11 체크리스트에서 미착수로 남아 있던 보안 2건. 나머지 미체크 항목은 코드와
대조한 결과 이미 수정돼 있었거나(6건) 근거가 기재된 명시적 연기다.

SSRF DNS-rebinding (store_remote_fetch.py)
  _assert_public_http_url 이 호스트를 해석해 통과 여부를 정하고 나면,
  http.client 가 소켓을 열 때 같은 이름을 한 번 더 해석했다. 리바인딩
  호스트는 첫 조회에 공인 주소를, 두 번째에 127.0.0.1 을 답하면 그만이었다.
  _guarded_create_connection 이 getaddrinfo 를 1회만 수행하고, 그 답변의
  주소를 전부 검증한 뒤 그중 하나로만 연결한다 — 검사한 주소와 실제 다이얼
  하는 주소가 갈라지지 않는다. 답변에 내부 주소가 하나라도 섞이면 거부한다
  (공인+내부 혼합이 리바인딩 패턴 자체이고, connect 가 어느 항목에 닿을지는
  예측 대상이 아니다). HTTPConnection._create_connection 을 갈아끼우는
  방식이라 connect() 의 나머지(TCP_NODELAY, 터널링)와 HTTPS 의 TLS 검증
  (self.host 기준)은 그대로 유지된다.

실패 인증 스로틀 (auth_rate_limit.py)
  COPYRIGHTER_AUTH_TOKEN 이 데이터 라우트의 유일한 게이트인데 틀린 토큰의
  비용이 요청 1회였다. 클라이언트 주소별 고정 윈도우 10회/60초, 초과 시
  429 + Retry-After. 성공하면 카운터를 비운다.
  자격증명을 실제로 제시하고 틀린 경우만 센다 — 콘솔은 토큰 입력 전에도 API
  를 여러 건 호출하므로, 그 401 들을 세면 정직한 페이지 로드 두 번으로
  운영자가 잠긴다. Origin 거부는 유효 세션의 CSRF 검사 실패지 추측이 아니라
  세지 않는다. 추적 테이블은 1024 클라이언트로 상한(만료분 우선 회수).

테스트 26건 추가. 리바인딩 회귀 재현(사전검사는 통과시키고 connect 에서 거부),
스로틀의 429/Retry-After, 무자격증명 요청이 스로틀에 안 잡히는 것, 성공 시
카운터 초기화 포함.

src/rights_filter/server/auth_rate_limit.py

@@ -0,0 +1,107 @@
+"""Failed-authentication throttling for the operator server.
+
+``COPYRIGHTER_AUTH_TOKEN`` is the only gate in front of the data routes, and
+nothing else slows down guessing it: every wrong bearer token costs an attacker
+one cheap request. This adds a fixed-window cap on *failed* attempts per client
+address, so an online brute force stops being viable while a working operator —
+whose attempts succeed, clearing the counter — never meets it.
+
+Deliberately only failures are counted. Throttling successful traffic would
+punish an operator working through a review queue for no security gain.
+
+Deployment note: the client identity is the peer address. On the intended
+single-host / closed-network deployment that is the real client. Behind a
+reverse proxy every request would share one address, and one bad actor could
+then throttle the operator too — put the proxy in front of a token it controls,
+or terminate closer to the console, if that ever becomes the topology.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from dataclasses import dataclass
+
+# A wrong token ten times in a minute is not an operator typo pattern; it is a
+# script. Generous enough that a person fumbling a paste never notices.
+MAX_FAILURES_PER_WINDOW = 10
+WINDOW_SECONDS = 60
+
+# Bound the table so a spoofed-source flood cannot grow it without limit. Far
+# above the number of clients a single-operator console legitimately sees.
+MAX_TRACKED_CLIENTS = 1024
+
+
+@dataclass
+class _Window:
+    failures: int
+    started_at: float
+
+
+class FailedAuthRateLimiter:
+    """Fixed-window cap on failed auth attempts, keyed by client address."""
+
+    def __init__(
+        self,
+        max_failures: int = MAX_FAILURES_PER_WINDOW,
+        window_seconds: int = WINDOW_SECONDS,
+        max_tracked_clients: int = MAX_TRACKED_CLIENTS,
+        clock=time.monotonic,
+    ) -> None:
+        self._max_failures = max(1, max_failures)
+        self._window_seconds = max(1, window_seconds)
+        self._max_tracked_clients = max(1, max_tracked_clients)
+        self._clock = clock
+        self._lock = threading.Lock()
+        self._windows: dict[str, _Window] = {}
+
+    def retry_after(self, client: str) -> int:
+        """Seconds the client must wait, or 0 when it may attempt now."""
+        with self._lock:
+            window = self._windows.get(client)
+            if window is None:
+                return 0
+            elapsed = self._clock() - window.started_at
+            if elapsed >= self._window_seconds:
+                # Window rolled; the client starts clean.
+                del self._windows[client]
+                return 0
+            if window.failures < self._max_failures:
+                return 0
+            # Round up so a caller that waits exactly this long is past the edge
+            # rather than one fractional second short of it.
+            return max(1, int(self._window_seconds - elapsed) + 1)
+
+    def record_failure(self, client: str) -> None:
+        with self._lock:
+            now = self._clock()
+            window = self._windows.get(client)
+            if window is None or now - window.started_at >= self._window_seconds:
+                self._evict_if_full_locked(now)
+                self._windows[client] = _Window(failures=1, started_at=now)
+                return
+            window.failures += 1
+
+    def record_success(self, client: str) -> None:
+        """A correct token clears the client's record."""
+        with self._lock:
+            self._windows.pop(client, None)
+
+    def _evict_if_full_locked(self, now: float) -> None:
+        if len(self._windows) < self._max_tracked_clients:
+            return
+        expired = [
+            client
+            for client, window in self._windows.items()
+            if now - window.started_at >= self._window_seconds
+        ]
+        for client in expired:
+            del self._windows[client]
+        if len(self._windows) < self._max_tracked_clients:
+            return
+        # Still full of live windows: drop the oldest so the table stays bounded.
+        oldest = min(self._windows, key=lambda client: self._windows[client].started_at)
+        del self._windows[oldest]
+
+
+__all__ = ["FailedAuthRateLimiter", "MAX_FAILURES_PER_WINDOW", "WINDOW_SECONDS"]

src/rights_filter/server/http_app.py

@@ -13,6 +13,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from urllib.parse import parse_qs, unquote, urlparse
 
+from rights_filter.server.auth_rate_limit import FailedAuthRateLimiter
 from rights_filter.server.image_store import LocalSubmissionImageStore, SUPPORTED_IMAGE_SUFFIXES
 from rights_filter.server.sqlite_store import CopyrighterStore
 
@@ -86,6 +87,9 @@ def build_server(
     auth_token: str | None = None,
 ) -> ThreadingHTTPServer:
     static_root = Path(static_dir).resolve()
+    # Shared across every worker thread: the token is the only gate in front of
+    # the data routes, so failed attempts have to be counted somewhere central.
+    auth_rate_limiter = FailedAuthRateLimiter()
 
     class CopyrighterHandler(BaseHTTPRequestHandler):
         server_version = "CopyrighterHTTP/0.1"
@@ -335,6 +339,15 @@ def build_server(
             # routes are protected.
             if not any(path.startswith(prefix) for prefix in _PROTECTED_PREFIXES):
                 return True
+            client = self.client_address[0] if self.client_address else "unknown"
+            retry_after = auth_rate_limiter.retry_after(client)
+            if retry_after:
+                self._json(
+                    {"error": "too many failed authentication attempts"},
+                    HTTPStatus.TOO_MANY_REQUESTS,
+                    extra_headers={"Retry-After": str(retry_after)},
+                )
+                return False
             header = self.headers.get("Authorization", "")
             # Compare as bytes: hmac.compare_digest raises TypeError on str inputs
             # containing non-ASCII, and that exception escapes before any response
@@ -347,12 +360,15 @@ def build_server(
                 # cookie so the browser can load <img>/media (which cannot send an
                 # Authorization header) without ever putting the token in a URL.
                 self._issue_session_cookie = True
+                auth_rate_limiter.record_success(client)
                 return True
             cookie_header = self.headers.get("Cookie", "")
+            presented_session_cookie = False
             if cookie_header:
                 jar = SimpleCookie()
                 jar.load(cookie_header)
                 morsel = jar.get(_SESSION_COOKIE)
+                presented_session_cookie = morsel is not None
                 if morsel is not None and hmac.compare_digest(
                     morsel.value.encode("utf-8", "replace"),
                     _session_tag(auth_token).encode("utf-8"),
@@ -362,7 +378,16 @@ def build_server(
                     if self.command in _STATE_CHANGING_METHODS and not self._origin_ok():
                         self._json({"error": "invalid origin"}, HTTPStatus.FORBIDDEN)
                         return False
+                    auth_rate_limiter.record_success(client)
                     return True
+            # Count only attempts that actually presented a credential and got it
+            # wrong. A request carrying neither header nor cookie is the console
+            # bootstrapping before the operator has typed the token — the GUI
+            # fires several of those per load, and counting them would let two
+            # honest page loads lock the operator out. A rejected origin above is
+            # likewise not a guess: that is a valid session failing a CSRF check.
+            if header or presented_session_cookie:
+                auth_rate_limiter.record_failure(client)
             self._json({"error": "unauthorized"}, HTTPStatus.UNAUTHORIZED)
             return False
 
@@ -416,11 +441,18 @@ def build_server(
                     f"{_SESSION_COOKIE}={_session_tag(auth_token)}; HttpOnly; SameSite=Strict; Path=/",
                 )
 
-        def _json(self, payload: object, status: HTTPStatus = HTTPStatus.OK) -> None:
+        def _json(
+            self,
+            payload: object,
+            status: HTTPStatus = HTTPStatus.OK,
+            extra_headers: dict[str, str] | None = None,
+        ) -> None:
             data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
             self.send_response(status)
             self.send_header("Content-Type", "application/json; charset=utf-8")
             self.send_header("Content-Length", str(len(data)))
+            for name, value in (extra_headers or {}).items():
+                self.send_header(name, value)
             self._maybe_set_session_cookie()
             self.end_headers()
             self.wfile.write(data)

src/rights_filter/server/store_remote_fetch.py

@@ -9,8 +9,9 @@ from __future__ import annotations
 
 import ipaddress
 import socket
+from http.client import HTTPConnection, HTTPSConnection
 from urllib.parse import urlparse
-from urllib.request import HTTPRedirectHandler, Request, build_opener
+from urllib.request import HTTPHandler, HTTPRedirectHandler, HTTPSHandler, Request, build_opener
 
 from rights_filter.server.store_url_utils import _is_http_url
 
@@ -63,11 +64,23 @@ def _fetch_stylesheet_url_bytes(url: str, timeout: int = 10, limit: int = CANDID
     return _fetch_url_bytes_with_headers(url, STYLESHEET_FETCH_HEADERS, timeout, limit)
 
 
+def _assert_public_ip(raw_ip: str) -> None:
+    # `not is_global` rejects every special-purpose range at once — including
+    # CGNAT 100.64.0.0/10, which an explicit is_private/is_loopback/... list
+    # missed — and only lets genuinely globally-routable addresses through.
+    if not ipaddress.ip_address(raw_ip).is_global:
+        raise ValueError("refusing to fetch internal address")
+
+
 def _assert_public_http_url(url: str) -> None:
     # SSRF guard: candidate image / page / stylesheet URLs come from external
     # search-result content (attacker-influenceable), so refuse to fetch
     # anything that resolves to a non-public address (loopback, RFC1918,
     # link-local cloud-metadata 169.254.169.254, etc.).
+    #
+    # This is the pre-flight rejection. It is NOT what makes the fetch safe on
+    # its own — see _guarded_create_connection for the check that binds to the
+    # address actually connected to.
     parsed = urlparse(url)
     if parsed.scheme not in {"http", "https"} or not parsed.hostname:
         raise ValueError("only public http(s) URLs may be fetched")
@@ -77,12 +90,76 @@ def _assert_public_http_url(url: str) -> None:
     except OSError as exc:
         raise ValueError("could not resolve fetch host") from exc
     for *_, sockaddr in infos:
-        ip = ipaddress.ip_address(sockaddr[0])
+        _assert_public_ip(sockaddr[0])
-        # `not is_global` rejects every special-purpose range at once — including
+
-        # CGNAT 100.64.0.0/10, which an explicit is_private/is_loopback/... list
+
-        # missed — and only lets genuinely globally-routable addresses through.
+def _guarded_create_connection(address, timeout=None, source_address=None):
-        if not ip.is_global:
+    """``socket.create_connection`` that vets the address it actually dials.
-            raise ValueError("refusing to fetch internal address")
+
+    Without this, the guard is a TOCTOU: ``_assert_public_http_url`` resolves the
+    host to decide whether the fetch is allowed, and then http.client resolves it
+    a *second* time to open the socket. A DNS-rebinding host answers the first
+    lookup with a public address and the second with 127.0.0.1 — the guard passes
+    and the connection still lands inside the network. Resolving once here and
+    dialing only an address from that same answer closes the window.
+
+    Every address in the answer must be public, not merely the one we dial: a
+    host resolving to both a routable and an internal address is the rebinding
+    pattern itself, and which entry ``connect`` reaches is not ours to predict.
+    """
+    host, port = address[0], address[1]
+    try:
+        infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
+    except OSError as exc:
+        raise ValueError("could not resolve fetch host") from exc
+    if not infos:
+        raise ValueError("could not resolve fetch host")
+
+    for *_, sockaddr in infos:
+        _assert_public_ip(sockaddr[0])
+
+    last_error: Exception | None = None
+    for family, socktype, proto, _canonname, sockaddr in infos:
+        sock = socket.socket(family, socktype, proto)
+        try:
+            # http.client passes socket._GLOBAL_DEFAULT_TIMEOUT (a sentinel
+            # object, not a number) when no timeout was set; only a real number
+            # is meaningful to settimeout.
+            if isinstance(timeout, (int, float)):
+                sock.settimeout(timeout)
+            if source_address:
+                sock.bind(source_address)
+            sock.connect(sockaddr)
+            return sock
+        except OSError as exc:
+            last_error = exc
+            sock.close()
+    raise last_error if last_error is not None else ValueError("could not connect to fetch host")
+
+
+class _GuardedHTTPConnection(HTTPConnection):
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        # `_create_connection` is the seam HTTPConnection.connect dials through,
+        # so overriding it keeps the rest of connect() (TCP_NODELAY, tunnelling)
+        # and, for HTTPS, the TLS wrap that still verifies against self.host.
+        self._create_connection = _guarded_create_connection
+
+
+class _GuardedHTTPSConnection(HTTPSConnection):
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self._create_connection = _guarded_create_connection
+
+
+class _GuardedHTTPHandler(HTTPHandler):
+    def http_open(self, req):
+        return self.do_open(_GuardedHTTPConnection, req)
+
+
+class _GuardedHTTPSHandler(HTTPSHandler):
+    def https_open(self, req):
+        return self.do_open(_GuardedHTTPSConnection, req, context=self._context)
 
 
 class _ValidatingRedirectHandler(HTTPRedirectHandler):
@@ -93,7 +170,11 @@ class _ValidatingRedirectHandler(HTTPRedirectHandler):
         return super().redirect_request(req, fp, code, msg, headers, newurl)
 
 
-_SSRF_SAFE_OPENER = build_opener(_ValidatingRedirectHandler())
+_SSRF_SAFE_OPENER = build_opener(
+    _GuardedHTTPHandler(),
+    _GuardedHTTPSHandler(),
+    _ValidatingRedirectHandler(),
+)
 
 
 def _open_url(request: Request, timeout: int):

tests/rights_filter/server/test_auth_rate_limit.py

@@ -0,0 +1,123 @@
+"""Specs for the failed-authentication throttle."""
+
+from __future__ import annotations
+
+from rights_filter.server.auth_rate_limit import FailedAuthRateLimiter
+
+
+class FakeClock:
+    def __init__(self) -> None:
+        self.now = 1000.0
+
+    def __call__(self) -> float:
+        return self.now
+
+    def advance(self, seconds: float) -> None:
+        self.now += seconds
+
+
+def _limiter(clock: FakeClock, **kwargs) -> FailedAuthRateLimiter:
+    return FailedAuthRateLimiter(max_failures=3, window_seconds=60, clock=clock, **kwargs)
+
+
+def test_client_may_attempt_before_any_failure():
+    limiter = _limiter(FakeClock())
+
+    assert limiter.retry_after("10.0.0.1") == 0
+
+
+def test_failures_below_the_cap_do_not_block():
+    clock = FakeClock()
+    limiter = _limiter(clock)
+
+    limiter.record_failure("10.0.0.1")
+    limiter.record_failure("10.0.0.1")
+
+    assert limiter.retry_after("10.0.0.1") == 0
+
+
+def test_reaching_the_cap_blocks_the_client():
+    clock = FakeClock()
+    limiter = _limiter(clock)
+
+    for _ in range(3):
+        limiter.record_failure("10.0.0.1")
+
+    retry_after = limiter.retry_after("10.0.0.1")
+    assert retry_after > 0
+    assert retry_after <= 61
+
+
+def test_block_applies_only_to_the_offending_client():
+    limiter = _limiter(FakeClock())
+
+    for _ in range(3):
+        limiter.record_failure("10.0.0.1")
+
+    assert limiter.retry_after("10.0.0.1") > 0
+    assert limiter.retry_after("10.0.0.2") == 0
+
+
+def test_window_rolls_after_it_expires():
+    clock = FakeClock()
+    limiter = _limiter(clock)
+
+    for _ in range(3):
+        limiter.record_failure("10.0.0.1")
+    assert limiter.retry_after("10.0.0.1") > 0
+
+    clock.advance(61)
+
+    assert limiter.retry_after("10.0.0.1") == 0
+
+
+def test_waiting_the_advertised_retry_after_is_enough():
+    clock = FakeClock()
+    limiter = _limiter(clock)
+
+    for _ in range(3):
+        limiter.record_failure("10.0.0.1")
+    clock.advance(30)
+
+    retry_after = limiter.retry_after("10.0.0.1")
+    assert retry_after > 0
+
+    clock.advance(retry_after)
+    assert limiter.retry_after("10.0.0.1") == 0
+
+
+def test_success_clears_the_record():
+    limiter = _limiter(FakeClock())
+
+    limiter.record_failure("10.0.0.1")
+    limiter.record_failure("10.0.0.1")
+    limiter.record_success("10.0.0.1")
+    limiter.record_failure("10.0.0.1")
+
+    # The counter restarted, so this client is nowhere near the cap.
+    assert limiter.retry_after("10.0.0.1") == 0
+
+
+def test_tracking_table_stays_bounded_under_many_distinct_clients():
+    clock = FakeClock()
+    limiter = _limiter(clock, max_tracked_clients=8)
+
+    for index in range(200):
+        limiter.record_failure(f"10.0.0.{index}")
+
+    assert len(limiter._windows) <= 8
+
+
+def test_expired_windows_are_reclaimed_before_live_ones_are_dropped():
+    clock = FakeClock()
+    limiter = _limiter(clock, max_tracked_clients=4)
+
+    for index in range(4):
+        limiter.record_failure(f"10.0.0.{index}")
+    clock.advance(61)  # every tracked window is now expired
+
+    limiter.record_failure("10.0.9.9")
+
+    assert limiter.retry_after("10.0.9.9") == 0
+    assert "10.0.9.9" in limiter._windows
+    assert len(limiter._windows) <= 4

tests/rights_filter/server/test_http_app.py

@@ -1181,8 +1181,7 @@ def test_http_server_runs_auto_search_for_submission(tmp_path: Path):
 def test_operator_gui_uses_api_bootstrap():
     # app.js was split into ordered app-*.js parts; this spec cares that the
     # console bootstraps off the API, not which part holds the call.
-    app_js = "
+    app_js = "\n".join(
-".join(
         path.read_text(encoding="utf-8") for path in sorted(Path("web/operator-gui").glob("app-*.js"))
     )
 
@@ -1408,6 +1407,98 @@ def test_protected_routes_require_token_when_configured(tmp_path: Path):
         server.shutdown()
 
 
+def test_repeated_wrong_tokens_are_throttled_with_retry_after(tmp_path: Path):
+    # 2026-07-11 #8: the shared token is the only gate on the data routes, so a
+    # wrong guess must not stay free forever.
+    from rights_filter.server.auth_rate_limit import MAX_FAILURES_PER_WINDOW
+
+    static_dir, image_store, store = _fixtures(tmp_path)
+    server = build_server(
+        host="127.0.0.1", port=0, store=store, image_store=image_store,
+        static_dir=static_dir, auth_token="secret",
+    )
+    _start(server)
+    base = f"http://127.0.0.1:{server.server_port}"
+    try:
+        for _ in range(MAX_FAILURES_PER_WINDOW):
+            guess = Request(base + "/api/bootstrap", headers={"Authorization": "Bearer wrong"})
+            with pytest.raises(HTTPError) as exc:
+                urlopen(guess)
+            assert exc.value.code == 401
+
+        # Cap reached: further attempts are refused outright, with a wait hint.
+        blocked = Request(base + "/api/bootstrap", headers={"Authorization": "Bearer wrong"})
+        with pytest.raises(HTTPError) as exc_blocked:
+            urlopen(blocked)
+        assert exc_blocked.value.code == 429
+        assert int(exc_blocked.value.headers.get("Retry-After")) > 0
+
+        # The throttle covers the correct token too while the window is open —
+        # otherwise it would leak which guess was right.
+        correct = Request(base + "/api/bootstrap", headers={"Authorization": "Bearer secret"})
+        with pytest.raises(HTTPError) as exc_correct:
+            urlopen(correct)
+        assert exc_correct.value.code == 429
+
+        # The unauthenticated shell still loads, so the operator is not locked
+        # out of the page itself.
+        assert urlopen(base + "/health").status == 200
+    finally:
+        server.shutdown()
+
+
+def test_credential_less_requests_do_not_count_toward_the_auth_throttle(tmp_path: Path):
+    # The console fires several API calls per load before the operator has typed
+    # the token. Those are 401s, not guesses — counting them would let two honest
+    # page loads lock the operator out.
+    from rights_filter.server.auth_rate_limit import MAX_FAILURES_PER_WINDOW
+
+    static_dir, image_store, store = _fixtures(tmp_path)
+    server = build_server(
+        host="127.0.0.1", port=0, store=store, image_store=image_store,
+        static_dir=static_dir, auth_token="secret",
+    )
+    _start(server)
+    base = f"http://127.0.0.1:{server.server_port}"
+    try:
+        for _ in range(MAX_FAILURES_PER_WINDOW * 2):
+            with pytest.raises(HTTPError) as exc:
+                urlopen(base + "/api/bootstrap")
+            assert exc.value.code == 401
+
+        # Never throttled, and the right token still works.
+        authed = Request(base + "/api/bootstrap", headers={"Authorization": "Bearer secret"})
+        assert urlopen(authed).status == 200
+    finally:
+        server.shutdown()
+
+
+def test_successful_auth_clears_earlier_failures(tmp_path: Path):
+    from rights_filter.server.auth_rate_limit import MAX_FAILURES_PER_WINDOW
+
+    static_dir, image_store, store = _fixtures(tmp_path)
+    server = build_server(
+        host="127.0.0.1", port=0, store=store, image_store=image_store,
+        static_dir=static_dir, auth_token="secret",
+    )
+    _start(server)
+    base = f"http://127.0.0.1:{server.server_port}"
+    try:
+        # Fumble almost to the cap, then get it right.
+        for _ in range(MAX_FAILURES_PER_WINDOW - 1):
+            with pytest.raises(HTTPError):
+                urlopen(Request(base + "/api/bootstrap", headers={"Authorization": "Bearer wrong"}))
+        assert urlopen(Request(base + "/api/bootstrap", headers={"Authorization": "Bearer secret"})).status == 200
+
+        # The counter restarted, so the operator gets a full allowance again.
+        for _ in range(MAX_FAILURES_PER_WINDOW - 1):
+            with pytest.raises(HTTPError) as exc:
+                urlopen(Request(base + "/api/bootstrap", headers={"Authorization": "Bearer wrong"}))
+            assert exc.value.code == 401
+    finally:
+        server.shutdown()
+
+
 def test_cookie_authed_state_change_requires_same_origin(tmp_path: Path):
     # 2026-07-11 #10: a cookie-authenticated POST from a foreign Origin (classic
     # CSRF) is rejected; header-authed requests are not forgeable.

tests/rights_filter/server/test_store_remote_fetch.py

@@ -0,0 +1,143 @@
+"""SSRF guard specs for the remote fetch boundary.
+
+Candidate image / page / stylesheet URLs come out of external search results, so
+the host on the other end is attacker-influenceable. These specs pin the two
+halves of the guard: the pre-flight URL check, and the check on the address the
+socket actually dials (which is what closes the DNS-rebinding window).
+"""
+
+from __future__ import annotations
+
+import socket
+from urllib.request import HTTPHandler, HTTPRedirectHandler, HTTPSHandler
+
+import pytest
+
+from rights_filter.server import store_remote_fetch as remote_fetch
+from rights_filter.server.store_remote_fetch import (
+    _assert_public_http_url,
+    _guarded_create_connection,
+    _GuardedHTTPConnection,
+    _GuardedHTTPSConnection,
+)
+
+
+def _addrinfo(*ips: str, port: int = 80):
+    return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (ip, port)) for ip in ips]
+
+
+PUBLIC_IP = "93.184.216.34"
+
+
+@pytest.mark.parametrize(
+    "internal_ip",
+    [
+        "127.0.0.1",  # loopback
+        "10.0.0.5",  # RFC1918
+        "192.168.1.10",  # RFC1918
+        "169.254.169.254",  # cloud metadata
+        "100.64.0.1",  # CGNAT
+    ],
+)
+def test_preflight_refuses_urls_resolving_to_internal_addresses(monkeypatch, internal_ip):
+    monkeypatch.setattr(socket, "getaddrinfo", lambda *a, **k: _addrinfo(internal_ip))
+
+    with pytest.raises(ValueError, match="refusing to fetch internal address"):
+        _assert_public_http_url("http://collector.example/image.png")
+
+
+def test_preflight_refuses_non_http_schemes():
+    for url in ["file:///etc/passwd", "ftp://example.com/x", "gopher://example.com"]:
+        with pytest.raises(ValueError, match="only public http"):
+            _assert_public_http_url(url)
+
+
+def test_preflight_allows_a_public_address(monkeypatch):
+    monkeypatch.setattr(socket, "getaddrinfo", lambda *a, **k: _addrinfo(PUBLIC_IP))
+
+    _assert_public_http_url("https://example.com/image.png")  # does not raise
+
+
+def test_connection_guard_refuses_an_internal_address(monkeypatch):
+    monkeypatch.setattr(socket, "getaddrinfo", lambda *a, **k: _addrinfo("127.0.0.1"))
+
+    with pytest.raises(ValueError, match="refusing to fetch internal address"):
+        _guarded_create_connection(("collector.example", 80), 5)
+
+
+def test_connection_guard_refuses_a_mixed_public_and_internal_answer(monkeypatch):
+    # A host resolving to both a routable and an internal address is the
+    # rebinding pattern itself; which entry connect() reaches is not predictable,
+    # so the whole answer is refused rather than just the internal entry.
+    monkeypatch.setattr(socket, "getaddrinfo", lambda *a, **k: _addrinfo(PUBLIC_IP, "10.0.0.5"))
+
+    with pytest.raises(ValueError, match="refusing to fetch internal address"):
+        _guarded_create_connection(("collector.example", 80), 5)
+
+
+def test_dns_rebinding_between_preflight_and_connect_is_refused(monkeypatch):
+    """The regression this guard exists for.
+
+    The pre-flight resolves the host and sees a public address. By the time the
+    socket is opened the same name resolves to loopback. Before the connection
+    was guarded, the fetch would have been allowed and would have hit the
+    internal address; now the second answer is re-checked and refused.
+    """
+    answers = [_addrinfo(PUBLIC_IP), _addrinfo("127.0.0.1")]
+
+    def rebinding_getaddrinfo(*_args, **_kwargs):
+        return answers.pop(0) if answers else _addrinfo("127.0.0.1")
+
+    monkeypatch.setattr(socket, "getaddrinfo", rebinding_getaddrinfo)
+
+    # First resolution: passes, exactly as the attacker intends.
+    _assert_public_http_url("http://rebind.example/image.png")
+
+    # Second resolution, at connect time: refused.
+    with pytest.raises(ValueError, match="refusing to fetch internal address"):
+        _guarded_create_connection(("rebind.example", 80), 5)
+
+
+def test_connection_guard_reports_unresolvable_hosts_as_value_error(monkeypatch):
+    def failing_getaddrinfo(*_args, **_kwargs):
+        raise socket.gaierror("no such host")
+
+    monkeypatch.setattr(socket, "getaddrinfo", failing_getaddrinfo)
+
+    with pytest.raises(ValueError, match="could not resolve fetch host"):
+        _guarded_create_connection(("nope.example", 80), 5)
+
+
+def test_guarded_connections_dial_through_the_validating_resolver():
+    # The guard only holds while http.client dials through it, so pin the seam
+    # rather than trusting that a later refactor keeps the override.
+    for connection_class in (_GuardedHTTPConnection, _GuardedHTTPSConnection):
+        connection = connection_class("example.com")
+        assert connection._create_connection is _guarded_create_connection
+
+
+def test_shared_opener_installs_the_guarded_handlers():
+    handlers = remote_fetch._SSRF_SAFE_OPENER.handlers
+
+    assert any(isinstance(handler, remote_fetch._GuardedHTTPHandler) for handler in handlers)
+    assert any(isinstance(handler, remote_fetch._GuardedHTTPSHandler) for handler in handlers)
+    assert any(isinstance(handler, remote_fetch._ValidatingRedirectHandler) for handler in handlers)
+
+
+def test_shared_opener_has_no_unguarded_http_handler_left():
+    """An unguarded sibling handler would quietly reopen the hole.
+
+    build_opener drops a default class only when a subclass instance is passed
+    in; if the guarded handlers were ever registered under a different base, the
+    stock HTTPHandler would come back and serve http:// with no address check.
+    """
+    handlers = remote_fetch._SSRF_SAFE_OPENER.handlers
+
+    http_handlers = [h for h in handlers if isinstance(h, HTTPHandler)]
+    https_handlers = [h for h in handlers if isinstance(h, HTTPSHandler)]
+    redirect_handlers = [h for h in handlers if isinstance(h, HTTPRedirectHandler)]
+
+    assert len(http_handlers) == 1 and type(http_handlers[0]) is remote_fetch._GuardedHTTPHandler
+    assert len(https_handlers) == 1 and type(https_handlers[0]) is remote_fetch._GuardedHTTPSHandler
+    assert len(redirect_handlers) == 1
+    assert type(redirect_handlers[0]) is remote_fetch._ValidatingRedirectHandler