fix(packs-lab): 일회성 토큰 jti 영속화 (SQLite) — 재시작 replay 방어 유지
인메모리 _used_jti set은 컨테이너 재시작 시 비워져 TTL 내 토큰 replay가 가능했음(webhook 배포가 잦아 실재 구멍). 영속 볼륨(PACK_BASE_DIR)의 jti_store.db에 사용 jti를 기록(PK 원자성), 만료 항목은 lazy 정리. verify_upload_token이 jti_store.consume 사용. TDD 3 + 기존 replay 테스트 보존. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
42
packs-lab/app/jti_store.py
Normal file
42
packs-lab/app/jti_store.py
Normal file
@@ -0,0 +1,42 @@
|
||||
"""일회성 업로드 토큰 jti의 영속 저장소 (SQLite).
|
||||
|
||||
인메모리 set은 컨테이너 재시작 시 비워져 replay 방어가 풀린다. 영속 볼륨(PACK_BASE_DIR)의
|
||||
SQLite 파일에 사용된 jti를 기록해 재시작에도 단발성을 유지한다. 단일 컨테이너 가정.
|
||||
|
||||
만료된 jti는 정리한다 — 만료 토큰은 auth의 TTL 검사가 먼저 거부하므로 기억할 필요가 없고,
|
||||
테이블 무한 증식을 막는다.
|
||||
"""
|
||||
import os
|
||||
import sqlite3
|
||||
import time
|
||||
|
||||
# 영속 볼륨 경로. 모듈 변수라 테스트에서 monkeypatch로 tmp 경로 주입 가능.
|
||||
JTI_DB_PATH = os.path.join(os.getenv("PACK_BASE_DIR", "/app/data/packs"), "jti_store.db")
|
||||
|
||||
|
||||
def _conn() -> sqlite3.Connection:
|
||||
os.makedirs(os.path.dirname(JTI_DB_PATH), exist_ok=True)
|
||||
conn = sqlite3.connect(JTI_DB_PATH, timeout=10)
|
||||
conn.execute(
|
||||
"CREATE TABLE IF NOT EXISTS used_jti("
|
||||
"jti TEXT PRIMARY KEY, expires_at INTEGER NOT NULL)"
|
||||
)
|
||||
return conn
|
||||
|
||||
|
||||
def consume(jti: str, expires_at: int) -> bool:
|
||||
"""jti를 사용 마킹. 처음이면 True, 이미 사용됐으면 False(replay 차단).
|
||||
|
||||
매 호출 새 연결을 열어 파일에서 읽으므로 재시작에도 단발성이 유지된다.
|
||||
PRIMARY KEY 제약으로 원자적(동일 jti 동시 INSERT 시 하나만 성공).
|
||||
"""
|
||||
now = int(time.time())
|
||||
with _conn() as conn:
|
||||
conn.execute("DELETE FROM used_jti WHERE expires_at < ?", (now,))
|
||||
try:
|
||||
conn.execute(
|
||||
"INSERT INTO used_jti(jti, expires_at) VALUES(?, ?)", (jti, expires_at)
|
||||
)
|
||||
return True
|
||||
except sqlite3.IntegrityError:
|
||||
return False
|
||||
Reference in New Issue
Block a user