feat(lps): API guard — LPS_API_KEY 설정 시에만 /v1 에 X-API-Key 검증

외부에서 API 를 함부로 호출(비용 발생 enqueue 등)하지 못하도록 정적 키
guard 를 추가한다. '키의 존재'가 토글 — 개발(local/dev)은 env 를 비워
개방 모드(기동 시 WARN), prod 만 키를 주입한다(협의 결정).

- router/v1/validator/auth.py: X-API-Key 의존성 — secrets.compare_digest
  상수시간 비교, 콤마 구분 복수 키(무중단 키 교체), 매 요청 env 조회
  (재기동 없이 테스트 가능). /v1 라우터 전체에 적용.
- /healthz·/readyz 는 라우터 밖이라 항상 개방(LB 프로브).
- negodata lps_sync_service: LPS_API_KEY env 있으면 헤더 자동 첨부(한 곳).
- compose(lps-api·negodata-backend) LPS_API_KEY 패스스루 + .env.example.
- prod 체크리스트(operations.md): 키 주입 + lps-api 포트 비공개 + 기동
  로그 'API guard ON' 확인. api.md 인증 섹션 추가.
- 라이브 스모크: 무헤더/오키 401 · 정키 2종 200 · healthz 200 확인.
- 테스트 6건 추가, 전체 141 passed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.env.example

@@ -6,6 +6,9 @@
 LPS_DB_USER=postgres
 LPS_DB_PASSWORD=postgres
 
+# ── LPS API guard (개발은 빈값=개방 모드, prod 만 키 주입 — lps-api 검증·negodata 헤더 첨부 공용) ──
+LPS_API_KEY=                   # 콤마 구분 복수 허용(무중단 키 교체). 생성 예: openssl rand -hex 32
+
 # ── LPS API 스케일 (미설정 시 1 / 40 — 단일 프로세스로도 ~1,100 RPS) ──
 LPS_API_PROCESS_COUNT=1        # uvicorn 프로세스 수(=사용 코어 수). 커넥션 풀은 예산에서 자동 역산
 LPS_DB_CONNECTION_BUDGET=40    # lps API 커넥션 총예산. 공유 PG=40, 전용 PG(max_conn≈100)=90

docker-compose.yml

@@ -41,6 +41,7 @@ services:
       # ── LPS(인터넷 최저가) 연동 — 미설정이면 연동 비활성으로 조용히 동작 ──
       LPS_DB_HOST: host.docker.internal          # lps_db 읽기전용(수집 배치·조회 API)
       LPS_BASE_URL: http://host.docker.internal:9600   # 검색요청 enqueue. lps-api 컨테이너 사용 시 http://lps-api:9600
+      LPS_API_KEY: ${LPS_API_KEY:-}              # LPS API guard 키 — 개발은 빈값(개방), prod 만 주입
     volumes:
       - ./negodata/backend:/app       # 호스트 소스 = 컨테이너 코드. 이게 있어야 수정이 즉시 반영됨
     ports:
@@ -138,6 +139,7 @@ services:
       # 부하테스트/대량 폴링 대비 시에만 코어 수만큼 상향(예: 4).
       PROCESS_COUNT: ${LPS_API_PROCESS_COUNT:-1}
       DB_CONNECTION_BUDGET: ${LPS_DB_CONNECTION_BUDGET:-40}   # 전용 PG(max_connections≈100)면 90 근처로 상향
+      LPS_API_KEY: ${LPS_API_KEY:-}   # API guard — 빈값=개방 모드(개발), prod 는 키 주입 + 포트 비공개(operations.md)
     ports:
       - "9600:9600"
     extra_hosts:

lps/docs/api.md

@@ -9,6 +9,10 @@
   "result": { "success": true, "code": 0, "desc": "SUCCESS" }
   ```
   (실패 시 `success:false`, `code`/`desc`에 오류 코드)
+- **인증(guard)**: 서버에 `LPS_API_KEY` 가 설정된 환경(prod)에서는 모든 `/v1/*` 요청에
+  `X-API-Key` 헤더가 필요합니다(불일치 시 `401`). 개발(local/dev)은 env 를 비워 **개방 모드**로
+  동작합니다. `/healthz`·`/readyz` 는 항상 개방(LB 프로브). 키는 콤마 구분 복수 등록 가능
+  (무중단 키 교체). 호출 예: `curl -H "X-API-Key: <키>" http://.../v1/lps/queue/stats`
 
 ---
 

lps/docs/operations.md

@@ -208,5 +208,10 @@ docker ps                          # lps-worker "(healthy)" 확인
   `LPS_PORT_COOLDOWN_SEC`(기본 max(sticky, 1800)) — 차단 감지된 포트 격리 시간. 포트 수를 늘리면
   (DECODO_PORT_START/END) 자동 반영 — 코드에 포트 수 하드코딩 없음. 튜닝은 `ip_session` 분석 쿼리(database.md) 참고.
 
-**남은 배포 과제**: API 인증·레이트리밋(비용 남용 방지), 다중 레플리카 시 분산 레이트리밋/프록시 IP 조정.
+- **API guard**: `LPS_API_KEY` 설정 시 `/v1/*` 전체에 X-API-Key 검증(콤마 구분 복수 키 —
+  무중단 교체). 개발(local/dev)은 미설정=개방 모드. **prod 체크리스트**: ① `.env` 에
+  `LPS_API_KEY` 주입(negodata-backend 도 같은 키 — 헤더 자동 첨부) ② lps-api 포트 공개
+  제거(내부 네트워크만, `ports:` 삭제) ③ 기동 로그에서 `API guard ON` 확인.
+
+**남은 배포 과제**: 레이트리밋(키별 요청량 제한), 다중 레플리카 시 분산 레이트리밋/프록시 IP 조정.
 **비용**: 대역폭이 원가의 대부분(오픈마켓 크롤) — 같은 상품 재크롤을 줄이는 **TTL 캐시**가 다음 절감 후보.

lps/router/router.py

@@ -2,7 +2,7 @@ import asyncio
 import time
 from contextlib import asynccontextmanager
 
-from fastapi import FastAPI, Request
+from fastapi import Depends, FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.middleware.gzip import GZipMiddleware
 
@@ -11,6 +11,7 @@ from common.database.db_session_manager import DB_SESSION_MNG
 from common.logger import LOG
 from common.utils.gtime import GTime
 from config.server_configs import web_server_config
+from router.v1.validator.auth import configured_keys, require_api_key
 import router.v1.lps.search
 
 API_SERVER_START_TIME = GTime.UTCStr()
@@ -85,4 +86,10 @@ async def readyz():
 
 
 # 각 도메인 라우터를 등록한다. 새 기능 추가 시 router.v1.<domain>.<file> 를 import 후 include.
-app.include_router(router.v1.lps.search.router)
+# guard: LPS_API_KEY 설정 시 /v1 전체에 X-API-Key 검증(개발은 미설정=개방 — auth.py 참고).
+app.include_router(router.v1.lps.search.router, dependencies=[Depends(require_api_key)])
+
+if configured_keys():
+    LOG.i(f"API guard ON — X-API-Key 검증({len(configured_keys())}개 키)")
+else:
+    LOG.w("LPS_API_KEY 미설정 — API 개방 모드(개발용). prod 배포 시 키 주입 + 포트 비공개 필수")

lps/router/v1/validator/auth.py

@@ -0,0 +1,28 @@
+"""API 키 guard — LPS_API_KEY 가 설정된 경우에만 /v1 라우터 전체를 보호한다.
+
+개발(local/dev)은 env 를 비워 **개방 모드**로 쓰고, prod 에서만 키를 주입한다(협의 결정
+2026-07-13). '키의 존재'가 토글이라 APP_ENV=local 고정 운영 전제와 충돌하지 않는다.
+- 키는 콤마 구분 복수 허용 — 무중단 키 교체(새 키 추가 → 호출자 전환 → 옛 키 제거).
+- 비교는 secrets.compare_digest(상수시간) — 타이밍 공격 방지.
+- /healthz·/readyz 는 라우터 밖이라 guard 대상이 아니다(LB/오케스트레이터 프로브).
+- prod 는 여기에 더해 lps-api 포트 비공개(내부 네트워크만)를 권장 — docs/operations.md.
+"""
+
+import os
+import secrets
+
+from fastapi import Header, HTTPException
+
+
+def configured_keys() -> set[str]:
+    """유효 API 키 집합. 매 호출 env 를 읽는다 — 프로세스 재기동 없이 테스트 가능, 비용은 무시 수준."""
+    return {k.strip() for k in os.environ.get("LPS_API_KEY", "").split(",") if k.strip()}
+
+
+async def require_api_key(x_api_key: str | None = Header(None, alias="X-API-Key")):
+    """/v1 공통 의존성. 키 미설정=개방 모드(무검증), 설정 시 X-API-Key 불일치는 401."""
+    keys = configured_keys()
+    if not keys:
+        return
+    if not x_api_key or not any(secrets.compare_digest(x_api_key, k) for k in keys):
+        raise HTTPException(status_code=401, detail="invalid or missing X-API-Key")

lps/tests/test_api_guard.py

@@ -0,0 +1,43 @@
+"""API guard 테스트 — LPS_API_KEY 설정 시에만 /v1 에 X-API-Key 검증(개발=미설정=개방 모드).
+
+auth.configured_keys 가 매 요청 env 를 읽으므로 monkeypatch.setenv 만으로 on/off 를 전환한다
+(앱 재기동 불필요).
+"""
+
+import pytest
+
+
+@pytest.fixture
+def guarded(monkeypatch):
+    monkeypatch.setenv("LPS_API_KEY", "k1,k2")
+
+
+async def test_open_mode_without_key_env(client, monkeypatch):
+    monkeypatch.delenv("LPS_API_KEY", raising=False)
+    r = await client.get("/v1/lps/queue/stats")       # 개방 모드 — 헤더 없이 통과
+    assert r.status_code == 200
+
+
+async def test_guarded_rejects_missing_header(client, guarded):
+    r = await client.get("/v1/lps/queue/stats")
+    assert r.status_code == 401
+
+
+async def test_guarded_rejects_wrong_key(client, guarded):
+    r = await client.get("/v1/lps/queue/stats", headers={"X-API-Key": "nope"})
+    assert r.status_code == 401
+
+
+async def test_guarded_accepts_any_configured_key(client, guarded):
+    for key in ("k1", "k2"):                          # 복수 키 — 무중단 키 교체용
+        r = await client.get("/v1/lps/queue/stats", headers={"X-API-Key": key})
+        assert r.status_code == 200
+
+
+async def test_guard_covers_post_search(client, guarded):
+    r = await client.post("/v1/lps/search", json={"data": []})
+    assert r.status_code == 401                       # enqueue(비용 발생 경로)도 보호
+
+
+async def test_healthz_stays_open(client, guarded):
+    assert (await client.get("/healthz")).status_code == 200   # LB 프로브는 guard 밖

negodata/backend/services/lps_sync_service.py

@@ -17,6 +17,7 @@
 - 반영은 한 트랜잭션(execute_lambda_run) — 부분 반영으로 워터마크가 오염되지 않는다.
 """
 import asyncio
+import os
 import uuid
 from collections import Counter
 from datetime import timedelta, timezone
@@ -72,9 +73,11 @@ class LpsSyncService:
             "price": str(item.price) if item.price else "",
         }
         base = web_server_config.lps_base_url.rstrip("/")
+        # LPS API guard: prod 는 LPS_API_KEY 를 주입해 X-API-Key 로 인증(개발은 미설정=개방 모드).
+        headers = {"X-API-Key": os.environ["LPS_API_KEY"]} if os.environ.get("LPS_API_KEY") else None
         try:
             async with httpx.AsyncClient(timeout=10.0) as client:
-                r = await client.post(f"{base}/v1/lps/search", json={"data": [payload]})
+                r = await client.post(f"{base}/v1/lps/search", json={"data": [payload]}, headers=headers)
                 r.raise_for_status()
                 body = r.json()
             entry = (body.get("items") or [{}])[0]