#!/usr/bin/env python3 """task-2973 L1-B — 신규설계 비교 API 종단(HTTP) E2E 하드닝 게이트 실증. ★★★ 이 스크립트는 InsuRo 리포 밖(워크스페이스)에만 존재한다. InsuRo 리포의 어떤 파일도 이 스크립트로 수정하지 않는다 — 대상 server 디렉토리는 sys.path 에 추가해 import 하는 용도로만 사용한다(읽기 전용). 목적: `server/policy_grouping/comparison_gate.py` 봉인 하드닝이 내놓는 신규 사유 2종(`excluded_malformed_entry`/`excluded_axes_malformed`)이 `server/new_design_comparison/engine.py` 의 `_GATE_REASON_TO_EXCLUDED_REASON` 매핑에 없는 상태(안전 폴백 `.get(..., "NOT_COMPARABLE")`)로 방치된 것이, **FastAPI TestClient 를 통한 실제 라우터→엔진→게이트 종단 경로**에서 정말 크래시(500) 없이 처리되는지 실측한다. 요청 스키마는 추측하지 않는다 — `server/routes/new_design_comparison_v1.py` 의 `NewDesignComparisonRequest`/`server/tests/test_new_design_comparison.py` 의 `_http_payload()`/`_seed_confirmed()`/`test_route_smoke_200` 패턴을 그대로 따른다(동일 파일 664~731줄 참고). 두 공격 케이스: A. forged axes (6축 중 1축 value 타입 오염, axes_verified=True) — 지식DB (`InMemoryKnowledgeStore`) 에 `save_mapping()` 으로 **직접** 심는다(이건 `Axis` 데이터클래스가 런타임에 타입을 강제하지 않는다는 점을 이용한, 저장소 오염을 흉내낸 것이지 요청 바디 조작이 아니다 — 아래 "왜 이 방식인가" 참고). C. `map_state` 키 자체가 없는 entry — `new_design_comparison.engine._new_entries()` 를 몽키패치해 정상 엔트리 뒤에 손상된 entry 1건을 주입한다. ── 왜 이 방식인가(정직성 규칙 준수, 추측 금지) ── `policy_extract/grouping_bridge.py::lookup_group()` 를 정독한 결과, 이 엔진의 실제 배선에서는: - `axes_verified=True` 는 오직 `_axes_cross_verify()` 가 저장 axes와 요청 axes 의 `axis_signature()` 일치를 확인했을 때만 세팅된다. 요청 axes 는 `CoverageAxesIn.value: str`(순수 문자열, Literal 아님)이라 임의 문자열을 보낼 수 있지만, 그 자체로는 "축 자체가 빠진" 위조를 만들 수 없다 — `CoverageAxesIn` 이 6축 전부를 Pydantic 레벨에서 요구하기 때문이다. 즉 순수 HTTP 요청 바디만으로는 A 케이스(6축 스키마 손상 + axes_verified =True 조합)를 만들 수 없다 — 이건 추측이 아니라 모델 정의(`models.py` 136~150줄) 를 읽고 확인한 사실이다. - 하지만 `Axis` 는 frozen 이 아닌 평범한 dataclass 이고 `value: str` 타입 힌트는 런타임에 강제되지 않는다. `Axis(value=42)` 처럼 지식DB에 **저장** 시점에 타입이 깨진 축을 넣으면(예: 과거 스키마 버전의 마이그레이션 오류, 수동 DB 편집 등 실제 운영에서 있을 수 있는 데이터 오염), `_axes_cross_verify()` 의 `CoverageAxes.from_dict()` 는 `str(d["value"])` 로 관대하게 강제 변환해 통과시키고(KeyError 없음), `axis_signature()` 정규화 문자열이 요청측과 일치하면 axes_verified=True 로 승격된다 — 그런데 게이트에 전달되는 `entry["axes"]` 는 **원본 저장 dict**(타입 오염 그대로, 즉 `{"value": 42, ...}`)이므로 `axes_schema_valid()` 의 `isinstance(value, str)` 검사에서 걸린다. 이게 A 케이스가 실제로 뚫을 수 있는 유일한 경로다(지식DB 오염, 요청 바디 위조 아님). - C 케이스(`map_state` 키 자체 없음)는 `_new_entries()`/`_existing_entries()` 가 항상 리터럴 dict 로 `"map_state": _map_state_of(link)` 를 채우고, `_map_state_of()` 자체가 `link.map_state is not None` 이면 그대로, 아니면 `_STATUS_TO_MAP_STATE.get(..., "UNKNOWN")` 폴백을 쓰므로 **문자열 값이 항상 보장**된다 — 즉 현재 배선에서는 지식DB를 아무리 오염시켜도 entry dict 에 map_state 키가 아예 빠지는 상황을 만들 수 없다(정적 분석 결론, 아래 실행으로 재확인). 그래서 C 는 `_new_entries()` 몽키패치로 "만약 이 방어벽이 뚫린다면" 시나리오를 인위적으로 재현한다 — 이건 순수 HTTP 경로만으로는 도달 불가능하다는 사실 자체가 보고할 발견이다. """ from __future__ import annotations import json import os import sys import traceback from unittest.mock import patch if len(sys.argv) != 2: print("usage: python3 l1_e2e_api.py ", file=sys.stderr) raise SystemExit(2) SERVER_DIR = os.path.abspath(sys.argv[1]) if not os.path.isdir(SERVER_DIR): print(f"ERROR: server_dir not found: {SERVER_DIR}", file=sys.stderr) raise SystemExit(2) sys.path.insert(0, SERVER_DIR) RESULTS: dict[str, dict] = {} def _record(name: str, **fields) -> None: RESULTS[name] = fields print(f"[{name}] " + json.dumps(fields, ensure_ascii=False, default=str)) print("-" * 78) def main() -> int: # ------------------------------------------------------------------- # import (server_dir 를 sys.path 루트로 삼는 기존 관례를 그대로 따름) # ------------------------------------------------------------------- try: import main as main_mod # type: ignore[import-not-found] from fastapi.testclient import TestClient from new_design_comparison import engine from new_design_comparison.models import AxisIn, CoverageAxesIn from policy_grouping import Axis, CoverageAxes, InMemoryKnowledgeStore from policy_grouping.comparison_gate import ( REASON_EXCLUDED_AXES_MALFORMED, REASON_EXCLUDED_MALFORMED_ENTRY, ) from policy_grouping.taxonomy import COARSE_DIAGNOSIS from routes.new_design_comparison_v1 import ( NEW_DESIGN_COMPARISON_PATH, _fa_account_id_dependency, ) except Exception as e: # noqa: BLE001 print(f"IMPORT FAILED: {type(e).__name__}: {e}") traceback.print_exc() print("RESULT_JSON:" + json.dumps({"import_error": f"{type(e).__name__}: {e}"})) return 1 print(f"server_dir = {SERVER_DIR}") print(f"engine module file = {engine.__file__}") print(f"comparison_gate REASON_EXCLUDED_MALFORMED_ENTRY = {REASON_EXCLUDED_MALFORMED_ENTRY!r}") print(f"comparison_gate REASON_EXCLUDED_AXES_MALFORMED = {REASON_EXCLUDED_AXES_MALFORMED!r}") print(f"engine._GATE_REASON_TO_EXCLUDED_REASON has malformed_entry mapped: " f"{REASON_EXCLUDED_MALFORMED_ENTRY in engine._GATE_REASON_TO_EXCLUDED_REASON}") print(f"engine._GATE_REASON_TO_EXCLUDED_REASON has axes_malformed mapped: " f"{REASON_EXCLUDED_AXES_MALFORMED in engine._GATE_REASON_TO_EXCLUDED_REASON}") print("-" * 78) # ------------------------------------------------------------------- # 인증 우회 — dependency_overrides (지시서 명시 방식) # ------------------------------------------------------------------- main_mod.app.dependency_overrides[_fa_account_id_dependency] = lambda: "TEST-FA-DUMMY" client = TestClient(main_mod.app) # ------------------------------------------------------------------- # 0) 인증 없이 라우트가 살아있는지(계약과 무관 401/403 도 "라우트 생존" 증거) — # ★ 여기서는 dependency override 를 끄고 별도 raw client 로 검증한다. # ------------------------------------------------------------------- raw_client = TestClient(main_mod.app) # override 는 app 전역에 걸리므로 override 제거 후 재확인 del main_mod.app.dependency_overrides[_fa_account_id_dependency] try: resp0 = raw_client.post(NEW_DESIGN_COMPARISON_PATH, json={}) _record( "0_route_alive_no_auth", status_code=resp0.status_code, body=resp0.text[:500], ) except Exception as e: # noqa: BLE001 _record("0_route_alive_no_auth", exception=f"{type(e).__name__}: {e}") finally: # override 복구 — 이후 케이스는 인증 우회 상태로 진행 main_mod.app.dependency_overrides[_fa_account_id_dependency] = lambda: "TEST-FA-DUMMY" # ------------------------------------------------------------------- # 공통 페이로드 빌더 — server/tests/test_new_design_comparison.py::_http_payload() # 를 그대로 복제(추측 금지 — 원본 664~677줄 그대로 옮김). # ------------------------------------------------------------------- def _base_http_payload() -> dict: return { "existing_analysis": { "analysis_table": { "columns": [{"column_id": "col0", "insurer": "KB손해보험", "premium_won": 28900}], "sections": [ { "coarse": COARSE_DIAGNOSIS, "rows": [{"coarse": COARSE_DIAGNOSIS, "name": "암진단비", "cells": {"col0": "3000"}}], } ], }, "policy_meta": [], }, "new_design": { "coverages": [ { "name": "암진단비", "baseline_amount_manwon": 1000, "target_amount_manwon": 3000, "premiums_won": {"삼성화재": 12000}, "amount_form": "FIXED_BENEFIT", "payout_form": "LUMP_SUM", "has_waiting_or_reduction": False, "is_renewable": False, } ] }, "existing_premium_hints": [], } # ------------------------------------------------------------------- # 1) 베이스라인 정상 200 스모크 (하드닝 코드가 정상 케이스를 깨지 않는지 먼저 확인) # ------------------------------------------------------------------- store_baseline = InMemoryKnowledgeStore() with patch.object(engine, "_default_store", return_value=store_baseline): resp1 = client.post( NEW_DESIGN_COMPARISON_PATH, json=_base_http_payload(), headers={"Authorization": "Bearer fake"}, ) _record( "1_baseline_smoke_200", status_code=resp1.status_code, has_official_comparison="official_comparison" in resp1.json() if resp1.status_code == 200 else None, ) # ------------------------------------------------------------------- # 2) 공격 A — forged axes (지식DB에 타입 오염된 axes 를 CONFIRMED 로 저장) # ------------------------------------------------------------------- store_a = InMemoryKnowledgeStore() forged_axes = CoverageAxes( trigger=Axis(value=42), # ★ 타입 오염 — value: str 이지만 dataclass 는 런타임 강제 없음 benefit_scope=Axis(value="UNKNOWN"), independence=Axis(value="UNKNOWN"), payout_form=Axis(value="UNKNOWN"), target_scope=Axis(value="UNKNOWN"), amount_form=Axis(value="UNKNOWN"), ) store_a.save_mapping("ATTACK-A-CLAUSE", forged_axes, "DX_BASIC", confidence=1.0, status="confirmed") payload_a = _base_http_payload() payload_a["new_design"]["coverages"][0]["name"] = "공격A-forged-axes" payload_a["new_design"]["coverages"][0]["clause_id"] = "ATTACK-A-CLAUSE" payload_a["new_design"]["coverages"][0]["axes"] = { # trigger 는 저장측(42)과 axis_signature() 정규화 문자열이 일치해야 # axes_verified=True 로 승격된다 — CoverageAxesIn.value 는 str 이므로 # "42" 문자열을 그대로 보낸다(_norm_sig_token 은 str(42).strip()=="42"). "trigger": {"value": "42"}, "benefit_scope": {"value": "UNKNOWN"}, "independence": {"value": "UNKNOWN"}, "payout_form": {"value": "UNKNOWN"}, "target_scope": {"value": "UNKNOWN"}, "amount_form": {"value": "UNKNOWN"}, } with patch.object(engine, "_default_store", return_value=store_a): resp2 = client.post( NEW_DESIGN_COMPARISON_PATH, json=payload_a, headers={"Authorization": "Bearer fake"}, ) body2 = None attack_a_row = None try: body2 = resp2.json() for row in body2.get("unknown_report", {}).get("rows", []): if row.get("coverage_name") == "공격A-forged-axes": attack_a_row = row break official_names = [ (r.get("existing") or {}).get("coverage_name") if r.get("existing") else None for r in body2.get("official_comparison", {}).get("rows", []) ] in_official = "공격A-forged-axes" in official_names or any( any(n.get("company") for n in r.get("new", [])) and r.get("group_label_ko") is not None and False # placeholder, real check below via group_id presence for r in body2.get("official_comparison", {}).get("rows", []) ) except Exception: official_names = None in_official = None _record( "2_attack_A_forged_axes", status_code=resp2.status_code, is_500=resp2.status_code == 500, found_in_unknown_report=attack_a_row is not None, unknown_row=attack_a_row, official_comparison_row_count=len(body2.get("official_comparison", {}).get("rows", [])) if body2 else None, raw_body_snippet=(resp2.text[:800] if resp2.status_code != 200 else None), ) # ------------------------------------------------------------------- # 3) 공격 C — map_state 키 자체가 없는 entry (몽키패치 fault-injection) # ★ 위 docstring 에 밝혔듯, 현재 배선(엔진 _map_state_of() 의 무조건 # 폴백)상 이 상태는 정상 HTTP 요청만으로는 도달 불가능하다 — 그래서 # _new_entries() 자체를 감싸 손상된 entry 1건을 주입한다. 이건 # "만약 방어벽이 뚫리면" 시나리오의 인위적 재현이며, 정직성 규칙에 따라 # 이렇게 명시한다(추측/은폐 금지). # ------------------------------------------------------------------- _orig_new_entries = engine._new_entries def _poisoned_new_entries(new_design, store): entries = _orig_new_entries(new_design, store) # attack_repro.py 의 C_missing_map_state_key 케이스와 동일 shape — # map_state 키 자체가 없다(값이 None 인 것과도 다름 — 키 부재). entries.append({"group_id": "DX_BASIC"}) return entries store_c = InMemoryKnowledgeStore() payload_c = _base_http_payload() payload_c["new_design"]["coverages"][0]["name"] = "공격C-정상엔트리" with ( patch.object(engine, "_default_store", return_value=store_c), patch.object(engine, "_new_entries", side_effect=_poisoned_new_entries), ): try: resp3 = client.post( NEW_DESIGN_COMPARISON_PATH, json=payload_c, headers={"Authorization": "Bearer fake"}, ) status3 = resp3.status_code body3_text = resp3.text[:2000] except Exception as e: # noqa: BLE001 — TestClient 가 예외를 그대로 올릴 수도 있음 status3 = None body3_text = f"CLIENT-SIDE EXCEPTION: {type(e).__name__}: {e}\n" + traceback.format_exc() _record( "3_attack_C_missing_map_state_key", status_code=status3, is_500=(status3 == 500), note="몽키패치로 _new_entries() 에 map_state 키 없는 entry 1건 주입 — 정상 HTTP 경로로는 도달 불가(위 docstring 근거)", body_snippet=body3_text, ) # ------------------------------------------------------------------- # 4) 참고 — gate_for_official_comparison() 자체는 직접 호출 시 예외를 # 던지지 않음을 재확인(engine.py 를 우회한 순수 게이트 fail-closed 검증, # attack_repro.py 와 동일 목적이지만 여기서 한 번 더 재확인). # ------------------------------------------------------------------- from policy_grouping.comparison_gate import gate_for_official_comparison try: gate_result = gate_for_official_comparison([{"group_id": "DX_BASIC"}]) gate_exc = None gate_excluded_reason = gate_result.excluded[0][1] if gate_result.excluded else None except Exception as e: # noqa: BLE001 gate_result = None gate_exc = f"{type(e).__name__}: {e}" gate_excluded_reason = None _record( "4_gate_direct_call_sanity", gate_raised=gate_exc is not None, gate_exception=gate_exc, gate_excluded_reason=gate_excluded_reason, expected_reason=REASON_EXCLUDED_MALFORMED_ENTRY, ) print("RESULT_JSON:" + json.dumps(RESULTS, ensure_ascii=False, default=str)) return 0 if __name__ == "__main__": raise SystemExit(main())