diff --git a/backend/supabase/scripts/hosted_data_plane.py b/backend/supabase/scripts/hosted_data_plane.py index 0b16086d2d..bd7f979713 100644 --- a/backend/supabase/scripts/hosted_data_plane.py +++ b/backend/supabase/scripts/hosted_data_plane.py @@ -121,11 +121,42 @@ def classify_blob(path: str, size_bytes: int) -> str: return "postgres_ok" +APPLY_CANDIDATE_LABELS = frozenset( + {"apply_candidate_pending_geocoded", "apply_candidate_pending_review"} +) + + def preview_hash(payload: Mapping[str, Any]) -> str: - canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode( + "utf-8" + ) return hashlib.sha256(canonical).hexdigest() +def _preview_body(preview: Mapping[str, Any]) -> dict[str, Any]: + return {key: value for key, value in preview.items() if key != "previewSha256"} + + +def apply_payload_sha256( + rows: Iterable[Mapping[str, Any]], + hosted_youtube_ids: Iterable[str], +) -> str: + """Hash only the rows that preview would insert, in a stable order.""" + payloads = [ + pending_insert_payload(row) + for row in rows + if classify_evaluation_row(row, hosted_youtube_ids) in APPLY_CANDIDATE_LABELS + ] + payloads.sort( + key=lambda item: ( + str(item.get("youtube_link") or ""), + str(item.get("trace_id") or ""), + str(item.get("origin_name") or ""), + ) + ) + return preview_hash({"payloads": payloads}) + + def build_apply_preview( *, local_restaurant_ids: Iterable[str], @@ -136,12 +167,13 @@ def build_apply_preview( docker_class = classify_local_docker_restaurants( local_restaurant_ids, hosted_restaurant_ids ) + rows_list = list(evaluation_rows) candidates: list[str] = [] classes: dict[str, int] = {} - for row in evaluation_rows: + for row in rows_list: label = classify_evaluation_row(row, hosted_youtube_ids) classes[label] = classes.get(label, 0) + 1 - if label in {"apply_candidate_pending_geocoded", "apply_candidate_pending_review"}: + if label in APPLY_CANDIDATE_LABELS: video_id = row_youtube_id(row) if video_id: candidates.append(video_id) @@ -154,6 +186,7 @@ def build_apply_preview( "evaluationClasses": classes, "applyCandidateVideoIds": candidates, "applyCandidateCount": len(candidates), + "applyPayloadSha256": apply_payload_sha256(rows_list, hosted_youtube_ids), "insertStatus": "pending", "overwriteApprovedForbidden": True, } @@ -181,7 +214,12 @@ def assert_apply_authorized( if preview.get("dockerRestaurantApply"): _deny("forbidden_local_docker_apply") expected = preview.get("previewSha256") - if not isinstance(expected, str) or expected != presented_preview_sha256: + recomputed = preview_hash(_preview_body(preview)) + if ( + not isinstance(expected, str) + or expected != recomputed + or expected != presented_preview_sha256 + ): _deny("preview_hash_mismatch") if environment.get(APPROVAL_ENV) != "1": _deny("approval_missing") @@ -215,6 +253,9 @@ def pending_insert_payload(row: Mapping[str, Any]) -> dict[str, Any]: "is_missing": bool(row.get("is_missing")), "is_not_selected": bool(row.get("is_notSelected") or row.get("is_not_selected")), "youtube_meta": row.get("youtube_meta"), + "evaluation_results": row.get("evaluation_results") + if isinstance(row.get("evaluation_results"), Mapping) + else None, "source_type": row.get("source_type"), "review_count": 0, } @@ -238,7 +279,7 @@ def _json_request( extra_headers: Mapping[str, str] | None = None, ) -> tuple[int, Any]: from urllib.request import Request, urlopen - from urllib.error import HTTPError + from urllib.error import HTTPError, URLError headers = { "apikey": key, @@ -255,15 +296,22 @@ def _json_request( try: with urlopen(request, timeout=30) as response: raw = response.read() - parsed = json.loads(raw.decode("utf-8")) if raw else None + try: + parsed = json.loads(raw.decode("utf-8")) if raw else None + except json.JSONDecodeError: + _deny("hosted_response_invalid") return int(response.status), parsed + except HostedDataPlaneError: + raise except HTTPError as exc: raw = exc.read() try: parsed = json.loads(raw.decode("utf-8")) if raw else None - except ValueError: + except json.JSONDecodeError: parsed = None return int(exc.code), parsed + except (URLError, TimeoutError, OSError): + _deny("hosted_request_failed") def fetch_hosted_restaurant_snapshot( @@ -307,7 +355,10 @@ def load_evaluation_rows(path: str) -> list[dict[str, Any]]: for line in source.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue - parsed = json.loads(line) + try: + parsed = json.loads(line) + except json.JSONDecodeError: + _deny("evaluation_row_invalid") if isinstance(parsed, dict): rows.append(parsed) return rows @@ -335,6 +386,14 @@ def apply_pending_candidates( _deny("candidate_count_mismatch") if preview.get("dockerRestaurantApply"): _deny("forbidden_local_docker_apply") + rows_list = list(evaluation_rows) + hosted_for_rows = [ + video_id + for row in rows_list + if (video_id := row_youtube_id(row)) and video_id not in allowed + ] + if apply_payload_sha256(rows_list, hosted_for_rows) != preview.get("applyPayloadSha256"): + _deny("apply_payload_mismatch") inserted: list[str] = [] skipped: list[str] = [] unresolved: list[str] = [] @@ -345,10 +404,12 @@ def apply_pending_candidates( _APPLIED, _PRESENT, _UNRESOLVED = 3, 2, 1 outcome: dict[str, int] = {} requester = fetch or _json_request - for row in evaluation_rows: + for row in rows_list: video_id = row_youtube_id(row) if video_id is None or video_id not in allowed: continue + if classify_evaluation_row(row, hosted_for_rows) not in APPLY_CANDIDATE_LABELS: + continue payload = pending_insert_payload(row) if payload["status"] != "pending": _deny("approved_status_forbidden") diff --git a/backend/supabase/scripts/test_hosted_data_plane.py b/backend/supabase/scripts/test_hosted_data_plane.py index 6f17299aae..1567966157 100644 --- a/backend/supabase/scripts/test_hosted_data_plane.py +++ b/backend/supabase/scripts/test_hosted_data_plane.py @@ -4,6 +4,7 @@ from __future__ import annotations import unittest +from unittest.mock import patch from backend.supabase.scripts.hosted_data_plane import ( APPROVAL_ENV, @@ -18,6 +19,8 @@ classify_local_docker_restaurants, pending_insert_payload, r2_public_object_url, + load_evaluation_rows, + _json_request, ) @@ -489,6 +492,121 @@ def test_r2_url_uses_issued_r2_dev_not_invented_dns(self) -> None: with self.assertRaises(HostedDataPlaneError): r2_public_object_url("NOPE", "x") + def test_changing_candidate_ids_without_rehash_is_rejected(self) -> None: + preview = build_apply_preview( + local_restaurant_ids=[], + hosted_restaurant_ids=[], + hosted_youtube_ids=[], + evaluation_rows=[ + { + "youtube_link": "https://www.youtube.com/watch?v=newvideo111", + "trace_id": "trace-new", + "geocoding_success": True, + "lat": 1, + "lng": 2, + "origin_name": "신규집", + } + ], + ) + preview["applyCandidateVideoIds"] = ["othervideo1"] + with self.assertRaises(HostedDataPlaneError) as raised: + assert_apply_authorized( + preview, + environment={APPROVAL_ENV: "1"}, + presented_preview_sha256=preview["previewSha256"], + ) + self.assertEqual(str(raised.exception), "preview_hash_mismatch") + + def test_skip_row_sharing_a_candidate_video_is_not_posted(self) -> None: + candidate = { + "youtube_link": "https://www.youtube.com/watch?v=newvideo111", + "trace_id": "trace-new", + "geocoding_success": True, + "lat": 1, + "lng": 2, + "origin_name": "신규집", + "evaluation_results": {"visit_authenticity": {"name": "신규집", "eval_value": 1}}, + } + skipped = { + "youtube_link": "https://www.youtube.com/watch?v=newvideo111", + "trace_id": "trace-skip", + "is_missing": True, + "origin_name": "빠진집", + } + preview = build_apply_preview( + local_restaurant_ids=[], + hosted_restaurant_ids=[], + hosted_youtube_ids=[], + evaluation_rows=[candidate, skipped], + ) + calls: list[dict] = [] + + def fake_fetch(url, *, key, method="GET", payload=None, extra_headers=None): + calls.append(payload) + return 201, None + + apply_pending_candidates( + preview=preview, + evaluation_rows=[candidate, skipped], + url=HOSTED_URL, + service_role_key="service-role", + environment={APPROVAL_ENV: "1"}, + presented_preview_sha256=preview["previewSha256"], + fetch=fake_fetch, + ) + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0]["origin_name"], "신규집") + self.assertEqual(calls[0]["status"], "pending") + self.assertEqual(calls[0]["evaluation_results"]["visit_authenticity"]["eval_value"], 1) + + def test_changed_row_content_is_rejected_before_post(self) -> None: + row = { + "youtube_link": "https://www.youtube.com/watch?v=newvideo111", + "trace_id": "trace-new", + "geocoding_success": True, + "lat": 1, + "lng": 2, + "origin_name": "신규집", + } + preview = build_apply_preview( + local_restaurant_ids=[], + hosted_restaurant_ids=[], + hosted_youtube_ids=[], + evaluation_rows=[row], + ) + changed = dict(row) + changed["origin_name"] = "다른집" + with self.assertRaises(HostedDataPlaneError) as raised: + apply_pending_candidates( + preview=preview, + evaluation_rows=[changed], + url=HOSTED_URL, + service_role_key="service-role", + environment={APPROVAL_ENV: "1"}, + presented_preview_sha256=preview["previewSha256"], + fetch=lambda *args, **kwargs: (201, None), + ) + self.assertEqual(str(raised.exception), "apply_payload_mismatch") + + def test_invalid_evaluation_json_fails_closed(self) -> None: + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "rows.jsonl" + path.write_text("{not json}\n", encoding="utf-8") + with self.assertRaises(HostedDataPlaneError) as raised: + load_evaluation_rows(str(path)) + self.assertEqual(str(raised.exception), "evaluation_row_invalid") + + def test_network_failure_is_a_hosted_data_plane_error(self) -> None: + from urllib.error import URLError + + with patch("urllib.request.urlopen", side_effect=URLError("down")): + with self.assertRaises(HostedDataPlaneError) as raised: + _json_request(HOSTED_URL + "/rest/v1/restaurants", key="service-role") + self.assertEqual(str(raised.exception), "hosted_request_failed") + if __name__ == "__main__": unittest.main()