"""anu_v2.owner_trigger_http_post — production GitHub comment http_post 구현체 + token_provider. 회장 verbatim §1~§4 1:1 박제 (2026-05-27 KST, task-2699): §1 OWNER_GEMINI_TRIGGER_TOKEN 전용 token_provider: ``make_owner_trigger_token_provider(env)`` 는 env(또는 os.environ)에서 ``OWNER_GEMINI_TRIGGER_TOKEN`` 만 읽는다. 부재/빈문자/비문자열 → TokenBoundaryViolation raise. BOT_GITHUB_TOKEN/GH_TOKEN/GITHUB_TOKEN/OWNER_PAT 등 fallback 절대 0. §2 production http_post 구현체: ``make_production_http_post(*, api_base, dry_run, opener)`` 가 반환하는 callable은 urllib.request 로 POST 를 실행한다. dry_run=True 면 실제 네트워크 호출 0. 방어적 재검증 (assert_endpoint_allowed) 을 http_post 내부에서 이중 실행. §3 보안 원칙: token raw value / Authorization header 를 예외 메시지 / 로그 / 반환 dict 어디에도 미포함. 반환 dict 는 {"status": code, "endpoint": path, "method": method, "response": <_redact_tokens 적용 결과>} 형식. §4 one-way isolation: 본 모듈은 anu_v2 내부 + stdlib 만 import. 외부(utils/dispatch/scripts) 의존성 0. 테스트 시 make_production_http_post(opener=) 으로 네트워크 교체 가능. """ from __future__ import annotations import json import os import urllib.error import urllib.request from typing import Callable, Mapping from anu_v2.auto_gemini_triage import _redact_tokens from anu_v2.owner_trigger_only import ( TOKEN_ENV_NAME, TokenBoundaryViolation, assert_endpoint_allowed, ) # ─── 상수 ──────────────────────────────────────────────────────────────────── GITHUB_API_BASE: str = "https://api.github.com" _HTTP_TIMEOUT: int = 30 # seconds # F-1 defense-in-depth: _redact_tokens 가 커버하지 못하는 추가 prefix 목록 _GITHUB_TOKEN_PREFIXES: tuple[str, ...] = ( "ghp_", "ghs_", "gho_", "ghu_", "ghr_", "github_pat_", ) # ─── 예외 ──────────────────────────────────────────────────────────────────── class OwnerTriggerHttpError(RuntimeError): """GitHub API HTTP 실패 시 raise. 보안 원칙: 메시지에 token / Authorization / headers 절대 미포함. status code + endpoint path 만 포함. """ # ─── F-1 로컬 redaction wrapper ─────────────────────────────────────────────── def _redact_owner_response(value): """_redact_tokens + _GITHUB_TOKEN_PREFIXES 이중 redaction (defense-in-depth). 먼저 기존 _redact_tokens(value) 를 적용한 뒤, 결과를 재귀 순회하며 str 값이 _GITHUB_TOKEN_PREFIXES 에 속하는 prefix 로 시작하면 "***REDACTED***" 로 치환. dict key/value, list, tuple 모두 재귀 처리. Args: value: 임의 JSON-serializable 값. Returns: token prefix 로 시작하는 str 이 ***REDACTED*** 로 치환된 결과. """ def _walk(v): if isinstance(v, str): if v.startswith(_GITHUB_TOKEN_PREFIXES): return "***REDACTED***" return v if isinstance(v, dict): return {_walk(k): _walk(val) for k, val in v.items()} if isinstance(v, (list, tuple)): walked = [_walk(item) for item in v] return type(v)(walked) return v return _walk(_redact_tokens(value)) # ─── token_provider 팩토리 ──────────────────────────────────────────────────── def make_owner_trigger_token_provider( env: Mapping[str, str] | None = None, ) -> Callable[[], str]: """``OWNER_GEMINI_TRIGGER_TOKEN`` 전용 token_provider 팩토리. 반환 callable 은 ``env``(또는 os.environ)에서 ``OWNER_GEMINI_TRIGGER_TOKEN`` 만 읽는다. 부재 / 빈문자 / 비문자열 → TokenBoundaryViolation raise. BOT_GITHUB_TOKEN / GH_TOKEN / GITHUB_TOKEN / OWNER_PAT 등 fallback 절대 0. Args: env: 환경변수 dict. None 이면 os.environ 에서 읽음. Returns: Callable[[], str] — 호출 시마다 TOKEN_ENV_NAME 값 반환. """ def _provider() -> str: source: Mapping[str, str] = env if env is not None else os.environ token = source.get(TOKEN_ENV_NAME) if not isinstance(token, str) or not token: raise TokenBoundaryViolation( f"{TOKEN_ENV_NAME} not set or empty — owner trigger fail-closed. " f"BOT_GITHUB_TOKEN/GH_TOKEN/GITHUB_TOKEN/OWNER_PAT fallback 절대 0." ) return token return _provider # ─── production http_post 팩토리 ────────────────────────────────────────────── def make_production_http_post( *, api_base: str = GITHUB_API_BASE, dry_run: bool = False, opener=None, ) -> Callable[[str, str, dict, dict], dict]: """production GitHub comment http_post callable 팩토리. 반환 callable ``_http_post(method, path, body, headers) -> dict``: 1. assert_endpoint_allowed 방어적 재검증 (이중 안전). 2. dry_run=True 면 네트워크 호출 0 — {"dry_run": True, ...} 반환. 3. dry_run=False 면 urllib.request 로 POST 실행. 4. 응답 dict 는 _redact_tokens 적용 후 반환. 5. 예외 시 OwnerTriggerHttpError raise. headers/token 미포함. Args: api_base: GitHub API 베이스 URL. 기본 "https://api.github.com". dry_run: True 면 실제 네트워크 호출 없이 dry_run dict 반환. opener: 테스트 주입용 urllib opener. None 이면 urllib.request.urlopen 사용. Returns: Callable[[str, str, dict, dict], dict] """ def _http_post( method: str, path: str, body: dict, headers: dict, ) -> dict: # 1) 방어적 재검증 — 코어에서 한 번, 여기서 한 번 (이중 안전) assert_endpoint_allowed(method, path) # 2) URL 조합 url = api_base.rstrip("/") + path # 3) dry_run 경로 — 네트워크 호출 0 if dry_run: return { "dry_run": True, "status": 0, "endpoint": path, "method": method, } # 4) urllib.request 로 POST 실행 data = json.dumps(body).encode("utf-8") req = urllib.request.Request( url, data=data, headers=headers, method="POST", ) try: if opener is not None: response = opener.open(req, timeout=_HTTP_TIMEOUT) else: response = urllib.request.urlopen(req, timeout=_HTTP_TIMEOUT) # noqa: S310 status_code = response.status raw_body = response.read() try: parsed = json.loads(raw_body) except Exception: parsed = raw_body.decode("utf-8", errors="replace") # 5) 반환 dict — response 에 _redact_owner_response 적용 (F-1 이중 redaction) return { "status": status_code, "endpoint": path, "method": method, "response": _redact_owner_response(parsed), } except urllib.error.HTTPError as exc: # headers 절대 미포함 — status + path 만 status_code = exc.code # raw body 도 redact 후 버림 (로그/예외메시지 미포함) raise OwnerTriggerHttpError( f"http_post failed: status={status_code} endpoint={path}" ) from None except urllib.error.URLError as exc: # reason 에도 token 이 섞일 수 있으므로 _redact_owner_response 처리 후 버림 _redact_owner_response(str(exc.reason)) raise OwnerTriggerHttpError( f"http_post failed: status=0 endpoint={path}" ) from None except Exception as exc: # noqa: BLE001 — 예외 메시지 token 포함 방지 _redact_owner_response(str(exc)) raise OwnerTriggerHttpError( f"http_post failed: status=-1 endpoint={path}" ) from None return _http_post