{_SPAN} 전후면부 로고를 바꾸었다
" def _claim(**over): return {"statement_native": "로고가 DAEWOO 로 바뀌었다", "sources": [_OK], "evidence_span": _SPAN, **over} class TestTheSsrfBoundary: """★★모델이 낸 주소를 그대로 열면 **내부망·메타데이터를 우리 손으로** 긁어 온다 (Codex BLOCK). """ @pytest.mark.parametrize("url,why", [ ("http://127.0.0.1/x", "loopback"), ("http://169.254.169.254/latest/meta-data/", "link-local 메타데이터"), ("http://10.0.0.5/", "사설망"), ("http://192.168.1.1/", "사설망"), ("https://localhost/", "이름이 loopback 으로 풀린다"), ("file:///etc/passwd", "scheme"), ("ftp://a.org/", "scheme"), ("http://user:pw@ex.org/", "userinfo"), ("http:///nohost", "host 없음"), ]) def test_it_is_refused(self, url, why): with pytest.raises(UnsafeUrl): _assert_safe(url) def test_a_public_address_passes(self): """★positive control — 막기만 하면 아무것도 못 연다.""" assert _assert_safe("https://example.org/a") def test_the_dns_result_is_what_is_checked(self): """★★이름만 보면 못 막는다 — **풀어서** 본다.""" import socket real = socket.getaddrinfo try: socket.getaddrinfo = lambda *a, **k: [ (2, 1, 6, "", ("127.0.0.1", 80))] with pytest.raises(UnsafeUrl, match="내부/예약"): _assert_safe("https://looks-ordinary.example/ # ★host 는 ASCII — IDNA 를 재는 자리가 아니다") finally: socket.getaddrinfo = real class TestOnlyOfficialUrlsAreOpened: """★★모델이 **문장에 적어 넣은** 주소를 그대로 열면, 모델이 우리 서버에게 아무 데나 요청을 시킬 수 있다 (Codex). """ def test_a_url_outside_the_official_list_is_not_fetched(self): def _boom(u): raise AssertionError(f"열면 안 된다: {u}") got = check_claim(_claim(sources=["https://딴곳.org"]), allowed_urls=[_OK], fetch=_boom) assert got["verdict"] == S_UNREACHABLE assert "공식 출처 목록에 없는" in got["why"] def test_no_fallback_to_another_url(self): """★★임의 fallback 없다 — 다른 주소로 대신 확인해 주지 않는다.""" got = check_claim(_claim(sources=["https://딴곳.org", _OK]), allowed_urls=[_OK], fetch=lambda u: _PAGE) assert got["verdict"] == S_UNREACHABLE def test_the_official_url_is_fetched(self): """★positive control.""" got = check_claim(_claim(), allowed_urls=[_OK], fetch=lambda u: _PAGE) assert got["verdict"] == S_FOUND def test_the_batch_builds_the_allowed_list_from_official_paths(self): """★`action.sources` 와 `url_citation` 둘 다에서 모은다.""" b = {"sources": [{"url": _OK, "snippet": ""}], "citations": ["https://cited.org"], "parsed": {"results": [{"research_subject_id": "rs_1", "claims": [_claim( sources=["https://cited.org"])]}]}} got = check_batch(b, fetch=lambda u: _PAGE) assert got["counts"][S_FOUND] == 1 class TestFabricatedCitationsAreCaught: """★★이 축이 있는 이유 — 지어낸 인용이 정본에 들어가는 길을 막는다.""" def test_a_span_not_on_the_page_is_absent(self): got = check_claim(_claim(evidence_span="페이지에 없는 문장"), allowed_urls=[_OK], fetch=lambda u: _PAGE) assert got["verdict"] == S_ABSENT assert "지어낸 인용" in got["why"] def test_an_unreachable_page_is_not_absent(self): """★★**차단된 페이지에 있었을 수 있다.** 「없다」로 못 잡는다.""" got = check_claim(_claim(), allowed_urls=[_OK], fetch=lambda u: None) assert got["verdict"] == S_UNREACHABLE def test_an_empty_body_is_not_absent(self): got = check_claim(_claim(), allowed_urls=[_OK], fetch=lambda u: " ") assert got["verdict"] == S_UNREACHABLE def test_a_shortened_quote_matches_by_its_head(self): """★모델이 「…」로 줄여 쓴 판이 실제로 있었다.""" got = check_claim(_claim(evidence_span=f"{_SPAN} … 뒷부분은 다름"), allowed_urls=[_OK], fetch=lambda u: _PAGE) assert got["verdict"] == S_FOUND and got["matched"] == "head" def test_html_tags_do_not_hide_the_quote(self): page = f"{_SPAN[:10]}{_SPAN[10:]}
" got = check_claim(_claim(), allowed_urls=[_OK], fetch=lambda u: page) assert got["verdict"] == S_FOUND def test_script_content_is_stripped(self): """★`딴 글
" got = check_claim(_claim(), allowed_urls=[_OK], fetch=lambda u: page) assert got["verdict"] == S_ABSENT class TestTheRunReportsCountsNotRatios: """★개수 그대로. 그리고 **지어낸 것이 하나라도 있으면** 표를 남긴다.""" def _batch(self, claims): return {"sources": [{"url": _OK, "snippet": ""}], "citations": [], "parsed": {"results": [{"research_subject_id": "rs_1", "claims": claims}]}} def test_one_fabricated_flags_the_run(self): got = check_run([self._batch([ _claim(), _claim(evidence_span="없는 문장")])], fetch=lambda u: _PAGE) assert got["has_unsupported"] is True assert got["counts"] == {S_FOUND: 1, S_ABSENT: 1, S_UNREACHABLE: 0} def test_a_clean_run_is_not_flagged(self): got = check_run([self._batch([_claim()])], fetch=lambda u: _PAGE) assert got["has_unsupported"] is False def test_a_page_is_fetched_once(self): """★같은 주소를 여러 claim 이 쓴다 — 두 번 받지 않는다.""" n = {"i": 0} def _f(u): n["i"] += 1 return _PAGE check_run([self._batch([_claim(), _claim(), _claim()])], fetch=_f) assert n["i"] == 1 def test_it_says_unverified_is_not_fine(self): got = check_run([self._batch([_claim()])], fetch=lambda u: _PAGE) assert "미확정" in got["note"] and "눈으로" in got["note"] class TestNormalizeDoesNotJudgeMeaning: """★뜻으로 묶지 않는다 — **글자가 그대로 있는지**만 본다.""" def test_whitespace_and_case_are_folded(self): assert normalize(" A B\n C ") == normalize("a b c") def test_different_words_stay_different(self): assert normalize("요금통") != normalize("회수권") class TestDnsRebindingIsClosed: """★★★검사할 때 푼 주소와 **연결할 때 푸는 주소는 다를 수 있다**. 공격자 DNS 는 첫 조회에 공개 주소를, 두 번째에 내부 주소를 준다 (TOCTOU / DNS rebinding). 그래서 **붙은 뒤에 상대를 다시 본다** (Codex). """ class _Sock: def __init__(self, ip): self.ip, self.closed = ip, False def getpeername(self): return (self.ip, 443) def close(self): self.closed = True @pytest.mark.parametrize("ip", ["127.0.0.1", "10.1.2.3", "192.168.0.9", "169.254.169.254", "::1"]) def test_a_private_peer_is_refused_after_connect(self, ip): from app.modules.pipeline.grounding_claims_support import _check_peer s = self._Sock(ip) with pytest.raises(UnsafeUrl): _check_peer(s) assert s.closed, "막았는데 소켓을 안 닫았다" def test_a_public_peer_passes(self): """★positive control — 막기만 하면 아무것도 못 연다.""" from app.modules.pipeline.grounding_claims_support import _check_peer _check_peer(self._Sock("93.184.216.34")) def test_the_connection_classes_check_the_peer(self): """★★handler 를 안 끼우면 **검사가 아무 데도 안 닿는다**.""" import inspect from app.modules.pipeline.grounding_claims_support import ( _guarded_opener) src = inspect.getsource(_guarded_opener) assert src.count("_check_peer(self.sock)") == 2, "http/https 둘 다여야" assert "_HTTPHandler" in src and "_HTTPSHandler" in src class TestTheProxyPathIsFailClosed: """★★★proxy 를 거치면 `getpeername()` 은 **proxy** 를 가리킨다 — 최종 목적지는 검증되지 않는다. 그러면 이 경계 전체가 무의미하다 (Codex). """ def test_a_configured_proxy_stops_everything(self, monkeypatch): from app.modules.pipeline.grounding_claims_support import ( _guarded_opener, http_fetch) monkeypatch.setenv("HTTP_PROXY", "http://proxy.local:3128") with pytest.raises(UnsafeUrl, match="proxy"): _guarded_opener() # ★그리고 **한 페이지도 안 연다** assert http_fetch("https://example.org/") is None def test_without_a_proxy_the_opener_is_built(self, monkeypatch): """★positive control.""" from app.modules.pipeline.grounding_claims_support import ( _guarded_opener) for k in ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy", "ALL_PROXY", "all_proxy"): monkeypatch.delenv(k, raising=False) assert _guarded_opener() is not None def test_the_opener_disables_proxies_explicitly(self): """★환경에 없어도 **끼워 둔다** — 나중에 생겨도 안 거친다.""" import inspect from app.modules.pipeline.grounding_claims_support import ( _guarded_opener) assert "ProxyHandler({})" in inspect.getsource(_guarded_opener) class TestTruncationAndEncoding: """★★잘린 본문으로 「없다」를 말하지 않는다. 인코딩도 응답이 말한 것으로.""" def test_the_source_reads_one_more_than_the_cap(self): import inspect from app.modules.pipeline import grounding_claims_support as m src = inspect.getsource(m.http_fetch) assert "MAX_PAGE_BYTES + 1" in src assert "get_content_charset()" in src def test_a_truncated_page_is_unreachable_not_absent(self): """★뒤쪽에 인용이 있었을 수 있다 — 미확정이다.""" from app.modules.pipeline.grounding_claims_support import check_claim got = check_claim(_claim(), allowed_urls=[_OK], fetch=lambda u: None) assert got["verdict"] == S_UNREACHABLE class TestHttpFetchStopsBeforeReadingABody: """★★★**산 경로 끝점** — 소스 문자열 검사가 아니라, `http_fetch` 를 실제로 태워 private peer 에서 **본문을 읽기 전에** `None` 이 되는지 본다 (Codex). ★바깥 네트워크는 안 쓴다. 연결만 가짜로 만든다. """ BODY = "HTTP/1.1 200 OK\r\n\r\n비밀".encode("utf-8") def _run_with_peer(self, monkeypatch, peer_ip, *, body=None): body = self.BODY if body is None else body """★`_check_peer` 가 볼 소켓만 갈아 끼운다 — 그 위는 진짜 코드다.""" import http.client import urllib.request from app.modules.pipeline import grounding_claims_support as m read = {"n": 0} class _Sock: def __init__(self, ip): self.ip = ip def getpeername(self): return (self.ip, 443) def close(self): pass def makefile(self, *a, **k): read["n"] += 1 import io return io.BufferedReader(io.BytesIO(body)) def sendall(self, *a, **k): pass def settimeout(self, *a, **k): pass def _connect(self): self.sock = _Sock(peer_ip) monkeypatch.setattr(http.client.HTTPSConnection, "connect", _connect) monkeypatch.setattr(http.client.HTTPConnection, "connect", _connect) # ★DNS precheck 는 통과시킨다 — 여기서 재는 것은 **붙은 뒤** 검사다 monkeypatch.setattr(m, "_assert_safe", lambda u: u) # ★★돈 가드는 opener 층에서 막는다. 여기서는 **소켓이 가짜**라 한 # 바이트도 안 나간다 — 진짜 코드 경로를 태우려고 그 층만 되돌린다. real_open = getattr(urllib.request.OpenerDirector.open, "_netprobe_real", None) if real_open is not None: monkeypatch.setattr(urllib.request.OpenerDirector, "open", real_open) for k in ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy", "ALL_PROXY", "all_proxy"): monkeypatch.delenv(k, raising=False) # ★host 는 ASCII 로 둔다 — 여기서 재는 것은 IDNA 가 아니라 peer 검사다 return m.http_fetch("https://looks-ordinary.example/"), read @pytest.mark.parametrize("ip", ["127.0.0.1", "10.0.0.7", "169.254.169.254"]) def test_a_private_peer_returns_none_without_reading(self, monkeypatch, ip): got, read = self._run_with_peer(monkeypatch, ip) assert got is None, "내부 주소인데 본문을 돌려줬다" assert read["n"] == 0, "막았다면서 본문을 읽었다" def test_a_public_peer_reads_the_body(self, monkeypatch): """★positive control — 막기만 하면 아무것도 못 연다.""" got, read = self._run_with_peer(monkeypatch, "93.184.216.34") assert got is not None and "비밀" in got assert read["n"] >= 1 class TestWeDoNotCallItFabricated: """★★★「없다」를 **「지어냈다」로 단정하지 않는다**. 실측(2026-08-30, 유료 산출): 「없다」가 난 주소를 직접 열어 보니 그 기사가 아니라 **잡지 첫 화면**이었다 — 「회수권」·「버스」·「1983」 어느 것도 없다. 링크가 죽어 포털 화면이 온 것일 수 있다. 우리가 아는 것은 「그 주소에서 그 문장을 못 찾았다」까지이고, 지어낸 것인지는 **사람이 본다**. """ def test_the_field_name_does_not_claim_fabrication(self): from app.modules.pipeline import grounding_claims_support as m got = m.check_run([], fetch=lambda _u: None) assert "has_unsupported" in got assert "has_fabricated" not in got, "칸 이름이 단정한다" def test_the_source_does_not_assert_fabrication_anywhere(self): import pathlib as _pl from app.modules.pipeline import grounding_claims_support as m src = _pl.Path(m.__file__).read_text(encoding="utf-8") assert "has_fabricated" not in src