fix: harden outbound integrations (errors, quota, SSRF, size cap)

- cloud_vision: raise on a per-image error inside a 200 response so the adapter
  records a FAILURE instead of a silent "no results" that also debited quota;
  send the key via X-goog-api-key header (verified for Vision) not the URL.
- search_policy: reset the daily_limit counter on a calendar-day rollover (it
  never re-windowed, so a hit limit disabled the provider until restart) and
  guard calls_made with a lock.
- google_custom_search: map the API response inside the adapter's try so a
  malformed payload becomes ENRICHMENT_FAILURE, not an uncaught crash. (Key
  stays a query param — the header is unverified for this endpoint.)
- http_json: cap the response body (10 MiB) and convert decode/read/OS errors
  into HttpJsonError so nothing escapes the transport contract.
- store_remote_fetch: SSRF guard rejects any non-global address (`not
  is_global`), covering CGNAT 100.64/10 the flag list missed.
- env_clients: a malformed daily-limit env falls back to a conservative cap
  (not fail-open unlimited); clamp NAVER display sizes.
- preprocessing: warn when external-derivative re-encoding falls back to the
  original bytes (EXIF/GPS may not have been stripped).

Tests: Vision per-image error raises / converts to FAILURE; daily-limit
malformed falls back conservatively.
This commit is contained in:
changukyu 2026-07-11 16:06:55 +09:00
parent 6193ed242d
commit 1803b4e0b8
10 changed files with 292 additions and 41 deletions

View file

@ -1,10 +1,13 @@
from __future__ import annotations
import logging
from io import BytesIO
from dataclasses import dataclass, field
from rights_filter.domain.records import DataClass
logger = logging.getLogger(__name__)
FaceBox = tuple[int, int, int, int]
@ -46,6 +49,14 @@ def build_external_derivative(
# GPS coordinates inside the bytes that get shipped to external APIs.
content = _reencode_without_metadata(original.content, width, height)
if content is None:
# Re-encoding failed (Pillow missing or undecodable bytes): fall back to
# the original bytes, but warn — those bytes may still carry EXIF/GPS
# metadata that is about to be sent to an external API. On an air-gapped
# target without Pillow this fires on every submission.
logger.warning(
"external derivative fell back to original bytes; embedded EXIF/GPS "
"metadata may not have been stripped before external submission"
)
content = original.content
return ImagePayload(

View file

@ -2,12 +2,11 @@ from __future__ import annotations
import base64
from typing import Any
from urllib.parse import quote_plus
from rights_filter.analysis.preprocessing import ImagePayload
from rights_filter.domain.records import Evidence, EvidenceSource
from rights_filter.integrations.external_policy import ExternalApiPolicy
from rights_filter.integrations.http_json import UrllibJsonTransport
from rights_filter.integrations.http_json import HttpJsonError, UrllibJsonTransport
class FakeWebDetectionClient:
@ -54,11 +53,22 @@ class GoogleVisionRestClient:
payload["parent"] = self.parent
response = self.transport.request_json(
"POST",
f"{self.endpoint}?key={quote_plus(self.api_key)}",
self.endpoint,
# Pass the key as a header so it never appears in the request URL
# (which would surface in tracebacks / logs on error).
headers={"X-goog-api-key": self.api_key},
payload=payload,
timeout=self.timeout,
)
web_detection = (response.get("responses") or [{}])[0].get("webDetection", {})
first = (response.get("responses") or [{}])[0]
error = first.get("error")
if error:
# Vision returns per-image errors inside a 200 response. Surfacing it
# as an exception lets the adapter record a FAILURE instead of a
# silent "no results" (which also wrongly debited quota).
message = error.get("message") if isinstance(error, dict) else str(error)
raise HttpJsonError(f"Vision per-image error: {message or 'unknown error'}")
web_detection = first.get("webDetection", {})
return {
"web_entities": [
{

View file

@ -1,9 +1,12 @@
from __future__ import annotations
import logging
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Mapping
logger = logging.getLogger(__name__)
from rights_filter.analysis.llm_assistance import InternalLlmAssistant, OllamaGenerateLlmClient
from rights_filter.integrations.cloud_vision_web_detection import (
CloudVisionWebDetectionAdapter,
@ -46,10 +49,12 @@ def build_provider_runtime(
google_adapter = _google_adapter(env, transport)
google_custom_search_adapter = _google_custom_search_adapter(env, transport)
llm_assistant = _llm_assistant(env, transport)
search_limit = _optional_int(env.get("COPYRIGHTER_NAVER_DAILY_LIMIT"))
google_limit = _optional_int(env.get("COPYRIGHTER_GOOGLE_DAILY_LIMIT"))
google_custom_search_limit = _optional_int(env.get("COPYRIGHTER_GOOGLE_CUSTOM_SEARCH_DAILY_LIMIT"))
llm_limit = _optional_int(env.get("COPYRIGHTER_LLM_DAILY_LIMIT"))
search_limit = _daily_limit(env.get("COPYRIGHTER_NAVER_DAILY_LIMIT"), "COPYRIGHTER_NAVER_DAILY_LIMIT")
google_limit = _daily_limit(env.get("COPYRIGHTER_GOOGLE_DAILY_LIMIT"), "COPYRIGHTER_GOOGLE_DAILY_LIMIT")
google_custom_search_limit = _daily_limit(
env.get("COPYRIGHTER_GOOGLE_CUSTOM_SEARCH_DAILY_LIMIT"), "COPYRIGHTER_GOOGLE_CUSTOM_SEARCH_DAILY_LIMIT"
)
llm_limit = _daily_limit(env.get("COPYRIGHTER_LLM_DAILY_LIMIT"), "COPYRIGHTER_LLM_DAILY_LIMIT")
face_crop_search_enabled = _truthy(env.get("COPYRIGHTER_GOOGLE_FACE_CROP_SEARCH"))
auto_naver_query_limit = _bounded_int(
env.get("COPYRIGHTER_AUTO_NAVER_QUERY_LIMIT"),
@ -193,13 +198,13 @@ def _naver_adapter(env: Mapping[str, str], transport: Any | None) -> NaverSearch
client_id=client_id,
client_secret=client_secret,
transport=transport,
display=_optional_int(env.get("NAVER_SEARCH_DISPLAY")) or 10,
display=_bounded_int(env.get("NAVER_SEARCH_DISPLAY"), default=10, minimum=1, maximum=100),
sort=env.get("NAVER_SEARCH_SORT", "sim"),
image_pages=_bounded_int(env.get("NAVER_SEARCH_PAGES"), default=1, minimum=1, maximum=10),
blog_display=_optional_int(env.get("NAVER_BLOG_SEARCH_DISPLAY")) or 3,
blog_display=_bounded_int(env.get("NAVER_BLOG_SEARCH_DISPLAY"), default=3, minimum=1, maximum=100),
blog_sort=env.get("NAVER_BLOG_SEARCH_SORT", "sim"),
blog_pages=_bounded_int(env.get("NAVER_BLOG_SEARCH_PAGES"), default=1, minimum=1, maximum=10),
web_display=_optional_int(env.get("NAVER_WEB_SEARCH_DISPLAY")) or 3,
web_display=_bounded_int(env.get("NAVER_WEB_SEARCH_DISPLAY"), default=3, minimum=1, maximum=100),
web_pages=_bounded_int(env.get("NAVER_WEB_SEARCH_PAGES"), default=1, minimum=1, maximum=10),
)
return NaverSearchAdapter(client)
@ -283,6 +288,23 @@ def _optional_int(value: str | None) -> int | None:
return None
def _daily_limit(value: str | None, env_name: str) -> int | None:
# Unset means "no daily limit" (intentional). A malformed non-empty value
# must NOT silently mean unlimited (fail-open) — fall back to a conservative
# cap and warn so the misconfiguration is visible.
if not value:
return None
try:
return max(0, int(value))
except (ValueError, TypeError):
logger.warning(
"%s is not an integer (%r); applying a conservative daily limit of 100",
env_name,
value,
)
return 100
def _bounded_int(value: str | None, default: int, minimum: int, maximum: int) -> int:
if not value:
return default

View file

@ -63,6 +63,10 @@ class GoogleCustomSearchClient:
items: list[Any] = []
for page_index in range(max(1, pages)):
params = {
# The Custom Search JSON API's documented auth is the `key` query
# parameter; keep it here rather than an X-goog-api-key header
# (unverified for this endpoint — a wrong choice silently breaks
# search). Vision, which is verified, uses the header.
"key": self.api_key,
"cx": self.cx,
"q": query,
@ -117,6 +121,10 @@ class GoogleCustomSearchAdapter:
try:
response = self.client.search_image(query)
policy.record_call(requested_calls)
# Map inside the try so a malformed payload becomes ENRICHMENT_FAILURE
# evidence instead of an uncaught exception that aborts the request.
return _map_image_response(submission_id, query, response)
except Exception as exc: # pragma: no cover - defensive adapter boundary
return [
Evidence(
@ -133,9 +141,6 @@ class GoogleCustomSearchAdapter:
)
]
policy.record_call(requested_calls)
return _map_image_response(submission_id, query, response)
def search_web_pages(
self, submission_id: str, query: str, policy: SearchApiPolicy
) -> list[Evidence]:
@ -162,6 +167,8 @@ class GoogleCustomSearchAdapter:
try:
response = self.client.search_web(query)
policy.record_call(requested_calls)
return _map_web_response(submission_id, query, response)
except Exception as exc: # pragma: no cover - defensive adapter boundary
return [
Evidence(
@ -178,9 +185,6 @@ class GoogleCustomSearchAdapter:
)
]
policy.record_call(requested_calls)
return _map_web_response(submission_id, query, response)
def _map_image_response(
submission_id: str, query: str, response: dict[str, Any]

View file

@ -12,6 +12,11 @@ class HttpJsonError(RuntimeError):
pass
# Cap the response body so a runaway/hostile API (or a misrouted large payload)
# cannot exhaust memory. JSON API responses are small; 10 MiB is generous.
_MAX_RESPONSE_BYTES = 10 * 1024 * 1024
class UrllibJsonTransport:
def __init__(self, connect_timeout: float | None = None) -> None:
# When set, probe TCP connectivity with this (short) timeout before the
@ -36,12 +41,25 @@ class UrllibJsonTransport:
request.add_header("Content-Type", "application/json")
try:
with urlopen(request, timeout=timeout) as response:
raw = response.read().decode("utf-8")
raw_bytes = response.read(_MAX_RESPONSE_BYTES + 1)
if len(raw_bytes) > _MAX_RESPONSE_BYTES:
raise HttpJsonError(
f"response body exceeded {_MAX_RESPONSE_BYTES} bytes"
)
# JSON must be UTF-8; decode leniently so a malformed byte raises a
# controlled HttpJsonError at json.loads rather than a bare
# UnicodeDecodeError escaping the transport contract.
raw = raw_bytes.decode("utf-8", errors="replace")
except HTTPError as exc:
try:
detail = exc.read().decode("utf-8", errors="replace")
except Exception:
detail = ""
raise HttpJsonError(f"HTTP {exc.code}: {detail}") from exc
except URLError as exc:
raise HttpJsonError(str(exc.reason)) from exc
except (OSError, socket.timeout) as exc:
raise HttpJsonError(f"request failed: {exc}") from exc
if not raw:
return {}

View file

@ -1,6 +1,8 @@
from __future__ import annotations
import threading
from dataclasses import dataclass, field
from datetime import date
@dataclass
@ -10,8 +12,24 @@ class SearchApiPolicy:
allowed_providers: set[str] = field(default_factory=lambda: {"naver"})
daily_limit: int | None = None
calls_made: int = 0
window_day: str = ""
_lock: threading.Lock = field(
default_factory=threading.Lock, repr=False, compare=False
)
def _roll_window_locked(self) -> None:
# daily_limit is a per-day budget. Reset the counter when the calendar
# day changes; otherwise the count only ever grows and, once the limit is
# hit, the provider stays disabled until the process restarts (it never
# re-windows for the next day).
today = date.today().isoformat()
if self.window_day != today:
self.window_day = today
self.calls_made = 0
def can_call(self, provider: str, requested_calls: int = 1) -> tuple[bool, str | None]:
with self._lock:
self._roll_window_locked()
if self.disabled:
return False, "search API disabled"
if not self.compliance_approved:
@ -24,4 +42,6 @@ class SearchApiPolicy:
return True, None
def record_call(self, count: int = 1) -> None:
with self._lock:
self._roll_window_locked()
self.calls_made += max(1, int(count or 1))

View file

@ -78,14 +78,10 @@ def _assert_public_http_url(url: str) -> None:
raise ValueError("could not resolve fetch host") from exc
for *_, sockaddr in infos:
ip = ipaddress.ip_address(sockaddr[0])
if (
ip.is_private
or ip.is_loopback
or ip.is_link_local
or ip.is_reserved
or ip.is_multicast
or ip.is_unspecified
):
# `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 ip.is_global:
raise ValueError("refusing to fetch internal address")

View file

@ -1,3 +1,5 @@
import pytest
from rights_filter.analysis.preprocessing import ImagePayload, build_external_derivative
from rights_filter.domain.records import EvidenceSource
from rights_filter.integrations.cloud_vision_web_detection import (
@ -6,6 +8,7 @@ from rights_filter.integrations.cloud_vision_web_detection import (
GoogleVisionRestClient,
)
from rights_filter.integrations.external_policy import ExternalApiPolicy
from rights_filter.integrations.http_json import HttpJsonError
def test_disabled_or_unapproved_policy_skips_outbound_call():
@ -269,3 +272,36 @@ def test_quota_exhaustion_records_failure_reason():
assert evidence[0].source == EvidenceSource.EXTERNAL_SKIPPED
assert "usage limit" in evidence[0].reason
assert client.calls == []
def test_vision_per_image_error_raises_instead_of_reporting_no_results():
# 2026-07-11 HIGH: Vision returns per-image errors inside a 200 response.
# detect_web must raise so the adapter records a FAILURE, not a silent
# "no results" that also wrongly debited quota.
class _ErrorTransport:
def request_json(self, method, url, headers=None, payload=None, timeout=10):
return {"responses": [{"error": {"code": 400, "message": "image too large"}}]}
client = GoogleVisionRestClient(api_key="google-key", transport=_ErrorTransport())
with pytest.raises(HttpJsonError):
client.detect_web(ImagePayload(b"image-bytes", 100, 100, {}))
def test_vision_adapter_converts_per_image_error_to_failure_evidence():
class _ErrorClient:
def detect_web(self, image):
raise HttpJsonError("Vision per-image error: image too large")
adapter = CloudVisionWebDetectionAdapter(_ErrorClient())
policy = ExternalApiPolicy(
compliance_approved=True,
metadata_logging_accepted=True,
allow_online_sync=True,
daily_limit=10,
)
derivative = build_external_derivative(ImagePayload(b"image", 100, 100, {}))
evidence = adapter.detect("submission-1", derivative, policy)
assert evidence[0].source == EvidenceSource.FAILURE
assert "image too large" in evidence[0].reason

View file

@ -67,7 +67,9 @@ def test_google_vision_rest_client_sends_web_detection_request_with_api_key():
call = transport.calls[0]
assert call["method"] == "POST"
assert call["url"] == "https://vision.googleapis.com/v1/images:annotate?key=google-key"
# Key travels as a header, not in the URL.
assert call["url"] == "https://vision.googleapis.com/v1/images:annotate"
assert call["headers"]["X-goog-api-key"] == "google-key"
assert call["payload"]["requests"][0]["features"][0]["type"] == "WEB_DETECTION"
assert call["payload"]["requests"][0]["image"]["content"] == base64.b64encode(b"image-bytes").decode("ascii")
assert response["web_entities"][0]["description"] == "IU"

View file

@ -317,7 +317,9 @@ def test_provider_runtime_falls_back_on_malformed_numeric_env():
)
assert runtime.auto_naver_query_limit == 3
assert runtime.search_result_similarity_threshold == 0.9
assert runtime.search_policy.daily_limit is None
# A malformed daily limit must not fail open to unlimited (None); it falls
# back to a conservative cap.
assert runtime.search_policy.daily_limit == 100
# --- #29: a single bad face box does not discard all crops --------------------
@ -486,7 +488,7 @@ def test_invalid_json_response_raises_http_json_error(monkeypatch):
def __exit__(self, *exc):
return False
def read(self):
def read(self, size=-1):
return b"<html>not json</html>"
monkeypatch.setattr(http_json, "urlopen", lambda request, timeout=None: _Resp())
@ -515,3 +517,133 @@ def test_unique_texts_accepts_generator():
from rights_filter.integrations.google_custom_search import _unique_texts
assert _unique_texts(value for value in [" a ", "a", "", "b"]) == ["a", "b"]
# --- 2026-07-11 H6/H7/D1: scoring-relevant fields survive the DB round-trip ----
#
# Production scores come from _rescore_submission, which re-reads evidence via
# _domain_evidence_from_ui and re-scores. If serialization drops promoted /
# entity / the precise source, the re-scored value silently drops to 0 (or gains
# phantom failure points). These pin the full round-trip through the scorer.
def _roundtrip(submission_id: str, evidence: Evidence) -> Evidence:
from rights_filter.server.store_serialization import (
_domain_evidence_from_ui,
_evidence_payload,
)
return _domain_evidence_from_ui(_evidence_payload(submission_id, evidence))
def test_promoted_naver_evidence_scores_after_serialization_roundtrip():
from rights_filter.analysis.risk_scoring import RiskScorer
evidence = Evidence(
source=EvidenceSource.NAVER_SEARCH,
reason="Naver search result found",
confidence=0.8,
data={
"promoted": True,
"promotion_reason": "named person/work search evidence",
"query": "IU 공식 앨범",
"query_signature": "naver:iu 공식 앨범",
"title": "IU official album",
"result_url": "https://example.test/iu",
},
)
restored = _roundtrip("SUB-NAVER", evidence)
assert restored.data["promoted"] is True
# Would be 0 if the promoted flag were dropped in serialization.
assert RiskScorer().score([restored]).score > 0
def test_web_entity_evidence_scores_after_serialization_roundtrip():
from rights_filter.analysis.risk_scoring import RiskScorer
evidence = Evidence(
source=EvidenceSource.WEB_DETECTION,
reason="Google web entity: BTS",
confidence=1.0,
data={"entity": "BTS", "query": "reverse image"},
)
restored = _roundtrip("SUB-WEB", evidence)
assert restored.data["entity"] == "BTS"
assert RiskScorer().score([restored]).score > 0
def test_two_distinct_web_entities_score_after_roundtrip():
from rights_filter.analysis.risk_scoring import RiskScorer
first = _roundtrip(
"SUB-2E",
Evidence(EvidenceSource.WEB_DETECTION, "entity BTS", 1.0, {"entity": "BTS"}),
)
second = _roundtrip(
"SUB-2E",
Evidence(EvidenceSource.WEB_DETECTION, "entity BLACKPINK", 1.0, {"entity": "BLACKPINK"}),
)
one = RiskScorer().score([first]).score
both = RiskScorer().score([first, second]).score
# The entity-name bucket must survive the round-trip so a second distinct
# entity still adds points instead of being deduped away.
assert both > one
def test_search_skipped_roundtrip_adds_no_phantom_failure_points():
from rights_filter.analysis.risk_scoring import RiskScorer
from rights_filter.server.store_serialization import _evidence_payload
evidence = Evidence(
source=EvidenceSource.SEARCH_SKIPPED,
reason="Naver 검색 생략 (일일 한도)",
confidence=0.0,
data={"provider": "naver"},
)
payload = _evidence_payload("SUB-SKIP", evidence)
assert payload["domainSource"] == "search_skipped"
restored = _roundtrip("SUB-SKIP", evidence)
assert restored.source == EvidenceSource.SEARCH_SKIPPED
# Must not be reclassified as FAILURE (+30).
assert RiskScorer().score([restored]).score == 0
def test_promoted_naver_hits_from_same_query_score_once():
from rights_filter.analysis.risk_scoring import RiskScorer
def hit(reason: str) -> Evidence:
return Evidence(
source=EvidenceSource.NAVER_SEARCH,
reason=reason,
confidence=0.8,
data={"promoted": True, "query_signature": "naver:iu"},
)
one = RiskScorer().score([hit("hit A")]).score
two = RiskScorer().score([hit("hit A"), hit("hit B")]).score
assert two == one and one > 0
def test_promoter_does_not_promote_zero_result_blog_evidence():
promoter = SearchResultPromoter()
# No result content, but the query text contains a marker word.
zero_result = Evidence(
source=EvidenceSource.NAVER_SEARCH,
reason="Naver blog search returned no results",
confidence=0.0,
data={"provider": "naver_blog", "query": "official album", "query_signature": "naver-blog:official album"},
)
[promoted] = promoter.promote([zero_result])
assert promoted.data["promoted"] is False
def test_promoter_detects_korean_rights_marker():
promoter = SearchResultPromoter()
hit = Evidence(
source=EvidenceSource.NAVER_SEARCH,
reason="Naver search result found",
confidence=0.5,
data={"title": "아이유 공식 화보", "result_url": "https://example.test/x", "query": "무제"},
)
[promoted] = promoter.promote([hit])
assert promoted.data["promoted"] is True