From 1803b4e0b8b4f2efbe8067d3f974f4a7bba2ef6b Mon Sep 17 00:00:00 2001 From: changukyu Date: Sat, 11 Jul 2026 16:06:55 +0900 Subject: [PATCH] fix: harden outbound integrations (errors, quota, SSRF, size cap) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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. --- src/rights_filter/analysis/preprocessing.py | 11 ++ .../cloud_vision_web_detection.py | 18 ++- src/rights_filter/integrations/env_clients.py | 36 ++++- .../integrations/google_custom_search.py | 16 ++- src/rights_filter/integrations/http_json.py | 22 ++- .../integrations/search_policy.py | 42 ++++-- .../server/store_remote_fetch.py | 12 +- .../test_cloud_vision_web_detection.py | 36 +++++ .../integrations/test_env_clients.py | 4 +- tests/rights_filter/test_review_fixes.py | 136 +++++++++++++++++- 10 files changed, 292 insertions(+), 41 deletions(-) diff --git a/src/rights_filter/analysis/preprocessing.py b/src/rights_filter/analysis/preprocessing.py index 3ae361a..1dddccf 100644 --- a/src/rights_filter/analysis/preprocessing.py +++ b/src/rights_filter/analysis/preprocessing.py @@ -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( diff --git a/src/rights_filter/integrations/cloud_vision_web_detection.py b/src/rights_filter/integrations/cloud_vision_web_detection.py index 588bc41..ed74a2e 100644 --- a/src/rights_filter/integrations/cloud_vision_web_detection.py +++ b/src/rights_filter/integrations/cloud_vision_web_detection.py @@ -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": [ { diff --git a/src/rights_filter/integrations/env_clients.py b/src/rights_filter/integrations/env_clients.py index 95973ea..422a969 100644 --- a/src/rights_filter/integrations/env_clients.py +++ b/src/rights_filter/integrations/env_clients.py @@ -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 diff --git a/src/rights_filter/integrations/google_custom_search.py b/src/rights_filter/integrations/google_custom_search.py index 8bae027..4e823dd 100644 --- a/src/rights_filter/integrations/google_custom_search.py +++ b/src/rights_filter/integrations/google_custom_search.py @@ -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] diff --git a/src/rights_filter/integrations/http_json.py b/src/rights_filter/integrations/http_json.py index e3e95ac..c81b98e 100644 --- a/src/rights_filter/integrations/http_json.py +++ b/src/rights_filter/integrations/http_json.py @@ -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: - detail = exc.read().decode("utf-8", errors="replace") + 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 {} diff --git a/src/rights_filter/integrations/search_policy.py b/src/rights_filter/integrations/search_policy.py index f332a39..2e09082 100644 --- a/src/rights_filter/integrations/search_policy.py +++ b/src/rights_filter/integrations/search_policy.py @@ -1,6 +1,8 @@ from __future__ import annotations +import threading from dataclasses import dataclass, field +from datetime import date @dataclass @@ -10,18 +12,36 @@ 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]: - if self.disabled: - return False, "search API disabled" - if not self.compliance_approved: - return False, "search API compliance not approved" - if provider not in self.allowed_providers: - return False, "search provider not allowed" - calls = max(1, int(requested_calls or 1)) - if self.daily_limit is not None and self.calls_made + calls > self.daily_limit: - return False, "search API usage limit reached" - return True, None + with self._lock: + self._roll_window_locked() + if self.disabled: + return False, "search API disabled" + if not self.compliance_approved: + return False, "search API compliance not approved" + if provider not in self.allowed_providers: + return False, "search provider not allowed" + calls = max(1, int(requested_calls or 1)) + if self.daily_limit is not None and self.calls_made + calls > self.daily_limit: + return False, "search API usage limit reached" + return True, None def record_call(self, count: int = 1) -> None: - self.calls_made += max(1, int(count or 1)) + with self._lock: + self._roll_window_locked() + self.calls_made += max(1, int(count or 1)) diff --git a/src/rights_filter/server/store_remote_fetch.py b/src/rights_filter/server/store_remote_fetch.py index 30a3627..2cdf268 100644 --- a/src/rights_filter/server/store_remote_fetch.py +++ b/src/rights_filter/server/store_remote_fetch.py @@ -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") diff --git a/tests/rights_filter/integrations/test_cloud_vision_web_detection.py b/tests/rights_filter/integrations/test_cloud_vision_web_detection.py index f9b6217..36d1e5f 100644 --- a/tests/rights_filter/integrations/test_cloud_vision_web_detection.py +++ b/tests/rights_filter/integrations/test_cloud_vision_web_detection.py @@ -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 diff --git a/tests/rights_filter/integrations/test_env_clients.py b/tests/rights_filter/integrations/test_env_clients.py index a14b57a..34e0b44 100644 --- a/tests/rights_filter/integrations/test_env_clients.py +++ b/tests/rights_filter/integrations/test_env_clients.py @@ -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" diff --git a/tests/rights_filter/test_review_fixes.py b/tests/rights_filter/test_review_fixes.py index e30a73d..9abb5af 100644 --- a/tests/rights_filter/test_review_fixes.py +++ b/tests/rights_filter/test_review_fixes.py @@ -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"not json" 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