rscc-common

RSCC 공통 Python 라이브러리. 응답 봉투, 에러 코드, X-Trace-Id 전파, ASGI 미들웨어, 인증 모드(bearer / jwt-cookie / session + CSRF), 회복탄력성(재시도·서킷 브레이커·격벽·TTL 캐시), 보안(JWT·AES-GCM·웹훅 서명·로그 리댁션), 한국 도메인 유틸을 제공한다.

• 기본 설치는 의존성 0 (표준 라이브러리만). FastAPI/pydantic 통합과 암호화는 extra 로 opt-in.
• Java(com.rscc:rscc-common-*)·JS(@rscc/common-*) 구현과 와이어 계약(응답 봉투·에러 코드· 헤더 규약 등)과 공유 테스트 벡터를 맞춘 폴리글랏 라이브러리의 Python 구현이다. 계약의 핵심 규칙은 각 모듈 docstring 에 요약돼 있어 이 패키지만으로 사용할 수 있다.
• Python >= 3.11 (상한 없음 — 3.11 ~ 3.14 에서 CI 검증), 타입 힌트 포함(py.typed), 라이선스 MIT.
• 시간 단위는 초 (Java/JS 는 ms — 초 × 1000 == ms).

설치

pip install rscc-common                    # 기본 — 의존성 0
pip install "rscc-common[fastapi]"         # + fastapi>=0.110, pydantic>=2, httpx>=0.27
pip install "rscc-common[crypto]"          # + cryptography>=42 (AES-256/GCM)
pip install "rscc-common[fastapi,crypto]"  # 둘 다

extra 없이 해당 모듈을 임포트하면 그 모듈만 ImportError/ModuleNotFoundError 이고, 패키지 자체(import rscc_common)와 나머지 모듈은 계속 동작한다. 순수 ASGI 미들웨어 (install_exception_handlers 를 제외한 install_*)와 httpx 훅은 fastapi/httpx 를 임포트하지 않으므로 기본 설치로도 임포트된다.

PyPI 대신 git 태그로 직설치할 수도 있다 (X.Y.Z 는 설치할 릴리스 버전):

pip install "rscc-common[fastapi] @ git+https://github.com/Jeonghyeon-Ryu/r-common.git@py-vX.Y.Z#subdirectory=python/rscc-common"

모듈 개요

* = extra 필요. 무의존 모듈은 패키지 최상위에서 서브모듈 네임스페이스로도 쓸 수 있고 (from rscc_common import jwt), 자주 쓰는 심볼은 최상위에 플랫 재노출된다 (configure_logging, TraceIdFilter, install_* 계열, get_trace_id/set_trace_id/ reset_trace_id/new_trace_id, trace_request_hook, UserContext/get_user_context, get_principal/get_session, ResultCode, CommonException, ErrorCode/ErrorCodeEnum, install_locale_middleware 등).

웹·와이어 계약 (ASGI / FastAPI)

회복탄력성

로깅·보안

한국 도메인 유틸

빠른 시작

import logging

import httpx
from fastapi import FastAPI

from rscc_common import configure_logging, install_trace_middleware, trace_request_hook
from rscc_common.errors import CommonException, ResultCode
from rscc_common.handlers import install_exception_handlers   # [fastapi]
from rscc_common.response import CommonResponse               # [fastapi]

configure_logging(log_dir="logs", level=logging.INFO, app_name="order-api", quiet_uvicorn=True)
# → [2026-09-26 12:00:00] INFO  [3f9a1c2e] order_api - 주문 조회: 7
#   (traceId 는 요청 컨텍스트 밖이면 "-". uvicorn --workers > 1 이면 per_pid_file=True 필수)

app = FastAPI()
install_trace_middleware(app)      # X-Trace-Id 수신 재사용/생성 → contextvar → 응답 에코
install_exception_handlers(app)    # 예외 → 실패 봉투 (CommonException·검증 400·HTTPException·500)

client = httpx.Client(event_hooks={"request": [trace_request_hook]})  # 다운스트림 전파
log = logging.getLogger("order_api")

@app.get("/orders/{order_id}")
def get_order(order_id: int):
    log.info("주문 조회: %s", order_id)
    if order_id > 1000:
        raise CommonException(ResultCode.NOT_FOUND)             # → 404 + {"success": false, "code": "404", ...}
    return CommonResponse.ok({"id": order_id}).to_wire()        # 반드시 to_wire() 반환

회복탄력성 — 재시도 안에서 서킷 브레이커로 게이트하고, OPEN 차단은 재시도하지 않는다:

from rscc_common.circuit_breaker import CircuitBreaker, CircuitOpenError
from rscc_common.retry import retry_call

cb = CircuitBreaker(failure_threshold=5, open_duration=30.0)

data = retry_call(
    lambda attempt: cb.call(fetch_inventory, item_id),   # 비동기는 await cb.call_async(afetch, item_id)
    should_retry=lambda exc, attempt: not isinstance(exc, CircuitOpenError),
)

call/call_async 는 정상 반환 → 성공, Exception → 실패(같은 예외 재전파), 취소·인터럽트 등 그 밖의 BaseException → 판정 없는 반납(on_ignore)으로 정확히 1회 보고한다.

요청 밖(백그라운드 태스크 등)에서 traceId 잇기:

from rscc_common import new_trace_id, reset_trace_id, set_trace_id

token = set_trace_id(captured_trace_id or new_trace_id())
try:
    do_background_work()     # 이 안의 로그 라인·httpx 발신에 traceId 가 실린다
finally:
    reset_trace_id(token)

자체 로깅 구성(dictConfig 등)을 쓰는 앱은 configure_logging 대신 TraceIdFilter 를 핸들러에 붙이고 포맷에 %(trace_id)s 를 넣는다:

from rscc_common.logging import TraceIdFilter

handler = logging.StreamHandler()
handler.addFilter(TraceIdFilter())    # dictConfig: {"()": "rscc_common.logging.TraceIdFilter"}
handler.setFormatter(logging.Formatter("[%(asctime)s] %(levelname)-5s [%(trace_id)s] %(name)s - %(message)s"))

에러 코드 확장 — 서비스 도메인 코드

ResultCode 10종("200"·"400"·"401"·"403"·"404"·"405"·"409"·"415"·"429"·"500")은 공통 카탈로그이고 라이브러리는 공통 코드만 방출한다. 서비스는 여기에 자기 도메인의 실패 코드를 얹어 쓸 수 있다.

from rscc_common import CommonException, ErrorCodeEnum

class OrderErrorCode(ErrorCodeEnum):          # 멤버 값 = (code, http_status, message) — ResultCode 와 같은 모양
    ORDER_OUT_OF_STOCK = ("ORDER_OUT_OF_STOCK", 409, "재고가 부족합니다.")
    PAYMENT_DECLINED = ("PAYMENT_DECLINED", 402, "결제가 거절되었습니다.")

@app.post("/orders")
def create_order(req: OrderRequest):
    if not stock.available(req.item_id):
        raise CommonException(OrderErrorCode.ORDER_OUT_OF_STOCK)          # 기본 메시지
    ...
    raise CommonException(OrderErrorCode.PAYMENT_DECLINED, "카드 한도를 초과했습니다.")  # 커스텀 메시지

install_exception_handlers 가 도메인 코드의 상태·code 를 그대로 쓴다 — HTTP 409 + {"success": false, "code": "ORDER_OUT_OF_STOCK", "message": "재고가 부족합니다."}. CommonResponse.fail(OrderErrorCode.ORDER_OUT_OF_STOCK)·BulkResultBuilder().failure(i, OrderErrorCode.…) 도 같은 코드를 받는다. 직접 만든 타입도 code/http_status/message 속성만 있으면 ErrorCode 로 쓸 수 있다.

검증은 런타임 fail-fast — 형식 위반 코드로 CommonException·CommonResponse.fail 을 만들면 즉시 ValueError 다(ResultCode 는 항상 통과). enum 정의 시점에는 검증하지 않으므로 서비스 테스트에서 전수 검증한다:

from rscc_common.errors import validate_error_codes

def test_error_codes_are_valid():
    validate_error_codes(OrderErrorCode)                           # 형식·상태 범위·code 중복
    validate_error_codes([*OrderErrorCode, *PaymentErrorCode])     # 서비스 전체 enum 간 중복까지

폴백 규칙 (소비자 필수) — 모르는 code 를 받은 클라이언트는 오류를 내지 말고 HTTP 상태로 공통 코드를 유도한다: ① 상태와 정확히 일치하는 공통 코드(SUCCESS 제외) → 그 코드, ② 그 외 4xx → "400", ③ 그 외 → "500". 원래의 도메인 코드는 그대로 보존해 도메인별 처리에 쓴다. 이 규칙 덕분에 도메인 코드 추가는 기존 소비자에게 non-breaking 이다.

from rscc_common.errors import is_common_code, result_code_for_status

body = resp.json()                                   # 예: HTTP 402 + {"code": "PAYMENT_DECLINED", ...}
common = body["code"] if is_common_code(body["code"]) else result_code_for_status(resp.status_code).code
# common == "400" — 공통 처리는 common 으로, 도메인별 처리는 body["code"] 로

같은 규칙이 서버의 "HTTP 상태 → 코드" 매핑(install_exception_handlers 의 HTTPException 처리)에도 쓰인다.

메시지 다국어 — 기본 ko + en

응답 봉투의 message 를 로케일별로 내보낸다. code 는 로케일과 무관하게 불변이고 message 만 바뀐다. 아무것도 설정하지 않으면(기본 로케일 ko, 협상 미들웨어 없음) 출력은 이전과 바이트 동일한 한국어다. 문구는 라이브러리에 내장돼 있다(ko·en — 공통 코드 메시지, 멱등성·업로드·정렬·CSRF 문구 22개). 그 외 언어는 리졸버(교체 훅)로 공급한다.

from rscc_common import install_idempotency, install_locale_middleware, install_trace_middleware
from rscc_common.handlers import install_exception_handlers
from rscc_common.messages import configure_messages

# (a) 서비스 전체를 영어로 — 협상 없이 모든 봉투 message 가 en
configure_messages(default_locale="en")

# (b) 요청별 Accept-Language 협상 (opt-in) — 응답에 Vary: Accept-Language
install_exception_handlers(app, localize_default_detail=True)
install_idempotency(app)
install_locale_middleware(app)     # 봉투를 즉시 보내는 rscc 미들웨어보다 **나중에(바깥에)** 등록
install_trace_middleware(app)      # trace 와의 상대 순서는 무관

GET /orders/999   Accept-Language: en-US,en;q=0.9
→ 404 {"success": false, "code": "404", "message": "Resource not found."}   Vary: Accept-Language
GET /orders/999   (헤더 없음 · ko)
→ 404 {"success": false, "code": "404", "message": "리소스를 찾을 수 없습니다."}

• 협상 — , 로 나눈 항목을 q 내림차순 안정 정렬해 주 서브태그(en-US → en, 대소문자 무시)가 지원 언어에 있으면 그 언어, * 면 기본 언어. q=0 은 거부, 형식 오류 항목은 버림, 아무것도 없으면 기본 언어. negotiate_language(header, supported=("ko", "en"), default=None) 로 직접 쓸 수도 있다.
• 등록 순서 — install_locale_middleware 를 가장 나중에(바깥에) 등록한다. 멱등성 409·레이트리밋 429· CSRF 403 처럼 미들웨어가 즉시 보내는 봉투는 로케일 미들웨어 안쪽에 있어야 협상된 언어로 나간다(바깥에 있으면 기본 언어). 예외 핸들러는 순서와 무관하다. catch-all 500 은 contextvar 가 정리된 뒤 실행되므로 미들웨어가 남긴 request.state.rscc_locale 로 언어를 잇는다.
• localize_default_detail=True (install_exception_handlers, 기본 False) — Starlette HTTPException 의 기본 detail(라우트 없음 "Not Found", "Method Not Allowed" 등 상태 문구)을 그 code 의 로케일 문구로 바꾼다. 앱이 준 detail 은 그대로다. 기본 False 는 이전 동작(영문 상태 문구) 유지.
• 메시지 키로 예외 만들기 — CommonException.with_message_key(ec, key, *args) 는 생성 시점의 현재 로케일로 즉시 해석한다. {0}, {1} … 은 단일 패스 문자 치환(str.format 아님 — 인자가 없으면 {N} 유지).

from rscc_common import CommonException, ResultCode

raise CommonException.with_message_key(ResultCode.BAD_REQUEST, "rscc.upload.extension", "exe")
# ko: "허용되지 않는 파일 형식입니다: exe" / en: "File type is not allowed: exe"

• 다른 언어·문구 교체 — 리졸버 — (key, language) → 템플릿 | None. language 는 주 서브태그 소문자다. None·빈 문자열·키와 같은 문자열은 미해결로 보고 내장 표(요청 언어 → 기본 언어 → ko)로 폴백한다.

JA = {"rscc.result.NOT_FOUND": "リソースが見つかりません。", "rscc.idempotency.conflict": "処理済みです"}
configure_messages(resolver=lambda key, lang: JA.get(key) if lang == "ja" else None)
install_locale_middleware(app, supported_languages=("ko", "en", "ja"))

• fixed 키 규칙 — FIXED_KEYS(멱등성 3종·정렬 2종·CSRF — 클라이언트가 문구로 분기할 수 있는 규약 고정 문구)는 ko·en 이면 내장 값이 항상 이긴다(리졸버 무시). 그 외 언어(ja 등)는 리졸버 허용. 고정 문구는 로케일별로 고정이므로, 영어 협상을 켜면 문구로 분기하던 클라이언트는 en 문구도 함께 인식해야 한다.
• 도메인 코드 — ErrorCodeEnum 의 message_key 는 기본 None(항상 message). 도메인 enum 이 message_key 프로퍼티를 오버라이드해 키를 돌려주면 리졸버로 로케일별 문구를 공급할 수 있다(미해결이면 message).
• 미들웨어 밖 — set_locale("en") → 작업 → reset_locale(token) 으로 수동 바인딩(백그라운드 태스크 등). get_message(key, *args, locale=None) 는 미해결 키에 키 문자열을, find_message 는 None 을 돌려준다.
• pydantic 검증 사유(data.errors[].reason)는 프레임워크 몫 — 이 규약의 범위 밖이다.

SSE — 채팅 프레임 + 범용 이벤트 스트림

rscc_common.sse 는 두 층이다(표준 라이브러리만).

• 채팅 프로필 — frame_delta/frame_sources/frame_error/frame_conversation_id/DONE: 단일 data: 라인 + 빈 줄, 페이로드는 단일 키 JSON([DONE] 제외). 채팅·검색요약 스트림 전용.
• 범용 계층 — 알림·진행률·로그 등 표준 필드(event/id/retry)를 쓰는 SSE. 빌더는 줄 끝 LF, 필드 순서 id → event → retry → data, 멀티라인 data 는 줄마다 data:(CR·LF·CRLF 분할). id·event 의 CR·LF, id 의 NUL, 음수 retry 는 ValueError. id="" 는 id: 를 써서 수신측 마지막 이벤트 ID 를 초기화한다.

생산 — FastAPI StreamingResponse + 권장 헤더(Content-Type: text/event-stream, Cache-Control: no-cache, no-transform, X-Accel-Buffering: no — 프록시 버퍼링·변환 차단):

import asyncio
import json

from fastapi.responses import StreamingResponse
from rscc_common.sse import SSE_RESPONSE_HEADERS, format_comment, format_event

@app.get("/jobs/{job_id}/events")
async def job_events(job_id: str):
    async def stream():
        yield format_comment("ping")                         # ": ping\n\n" — 하트비트(이벤트 없음)
        async for pct in watch_progress(job_id):
            yield format_event(json.dumps({"pct": pct}), event="progress", id=str(pct), retry=3000)
            # "id: 50\nevent: progress\nretry: 3000\ndata: {\"pct\": 50}\n\n"
        yield format_event("done", event="complete")
    return StreamingResponse(stream(), headers=SSE_RESPONSE_HEADERS)   # 헤더를 더하려면 {**SSE_RESPONSE_HEADERS, ...}

소비 — 업스트림 SSE 를 httpx 로 읽기. SseEventParser 는 WHATWG HTML 표준의 이벤트 스트림 해석을 그대로 따른다: 스트림 첫머리 BOM 1개 제거, 줄 끝 CRLF·LF·CR(청크 경계의 CR + LF 는 줄 끝 1개), 값 앞 공백 1개만 제거, event 없으면 "message", NUL 포함 id 무시, retry 는 ASCII 숫자만, 빈 줄로 끝나지 않은 마지막 이벤트는 폐기. bytes 청크는 증분 UTF-8 디코더로 풀어 한글·이모지가 청크 경계에서 잘려도 안전하다.

import httpx
from rscc_common.sse import aiter_sse_events

async with httpx.AsyncClient(timeout=None) as client:
    async with client.stream("GET", url, headers={"Accept": "text/event-stream"}) as resp:
        async for ev in aiter_sse_events(resp.aiter_bytes()):   # 동기는 iter_sse_events(resp.iter_bytes())
            if ev.event == "progress":
                update(json.loads(ev.data), last_id=ev.id)

재연결을 직접 구현한다면 SseEventParser 를 쓰고 parser.last_event_id 를 Last-Event-ID 헤더로, parser.retry_ms 를 대기 시간으로 쓴다(finish() 뒤의 feed() 는 새 스트림으로 취급되며 두 값은 유지된다). 레거시 채팅 리더처럼 EOF 의 미완성 이벤트까지 받으려면 finish(dispatch_incomplete=True) (헬퍼는 iter_sse_events(..., dispatch_incomplete=True)). 채팅 빌더가 만든 프레임도 이 파서로 읽으면 프레임마다 이벤트 1개, data = 페이로드다.

인증 모드 — bearer / jwt-cookie

JWT 가 요청에 실려 오는 방식을 서비스가 고른다. 기본은 bearer(현행 동작 그대로) 이고 jwt-cookie 는 opt-in 이다. 토큰 자체(HS256·클레임·kid 로테이션)는 rscc_common.jwt 그대로다.

import os

from fastapi import Depends, FastAPI, Request, Response
from pydantic import BaseModel

from rscc_common import install_jwt_auth, install_trace_middleware, jwt
from rscc_common.auth import AuthPrincipal, expired_token_cookie, require_principal, token_cookie
from rscc_common.csrf import request_csrf_rotation
from rscc_common.errors import CommonException, ResultCode
from rscc_common.handlers import install_exception_handlers   # [fastapi] — 401/403 을 봉투로 변환

SECRET = os.environ["JWT_SECRET"]          # 32바이트 이상 — 미만이면 설치 시점 ValueError (fail-fast)
EXPIRATION_MS = 3_600_000

app = FastAPI()
install_exception_handlers(app)
install_jwt_auth(app, SECRET, mode="jwt-cookie", skip_paths=["/openapi/**"])  # bearer 는 mode 생략
install_trace_middleware(app)              # 마지막 등록 = 가장 바깥 — CSRF 403 봉투에도 X-Trace-Id

class LoginBody(BaseModel):
    username: str
    password: str

@app.post("/auth/login")                   # JSON 전용 로그인 (교차 출처 JSON POST 는 preflight 강제)
def login(body: LoginBody, request: Request, response: Response):
    user = authenticate(body.username, body.password)          # 서비스 구현
    if user is None:
        raise CommonException(ResultCode.UNAUTHORIZED)
    token = jwt.generate_token(user.id, SECRET, user.role, expiration_ms=EXPIRATION_MS)
    response.headers.append("set-cookie", token_cookie(token, max_age=EXPIRATION_MS // 1000))
    request_csrf_rotation(request)         # 주체 변경 → CSRF 토큰 회전 (권장)
    return {"id": user.id, "role": user.role}                  # 토큰은 본문에 싣지 않는다

@app.post("/auth/logout")                  # 쿠키로 인증된 POST — X-XSRF-TOKEN 헤더 필요
def logout(response: Response):
    response.headers.append("set-cookie", expired_token_cookie())   # 빈 값 + Max-Age=0
    return {"ok": True}

@app.get("/me")                            # 쿠키 모드의 로그인 상태 확인은 토큰 만료 계산 대신 이 엔드포인트로
def me(p: AuthPrincipal = Depends(require_principal())):         # 토큰 없음·무효·만료 → 401 봉투
    return {"id": p.subject, "role": p.role}

@app.delete("/admin/users/{uid}", dependencies=[Depends(require_principal("ADMIN"))])  # 역할 부족 → 403
def delete_user(uid: str): ...

동작:

• 미들웨어는 스스로 거절하지 않는다(Java 필터와 동일) — 토큰이 없거나 무효면 get_principal() 이 None 일 뿐이다(무효 토큰은 WARN 한 줄). 401/403 은 보호 엔드포인트의 require_principal(...) 이 CommonException 으로 내고 install_exception_handlers 가 봉투로 바꾼다(미설치면 500).
• 역할 비교는 role 클레임(ADMIN/EDITOR/VIEWER) — role 없는 구버전 토큰은 "USER" 로 취급된다 (Java ROLE_USER 폴백). 역할 계층은 없다.
• bind_user_context=True(기본)면 UserContext 를 인증 결과(user_id=sub, role) 또는 None 으로 덮어쓴다 — 브라우저가 직접 붙는 에지 서비스에서 X-User-* 헤더로 신원을 위조할 수 없다. install_user_context_middleware 와 함께 설치해도 등록 순서와 무관하게 인증 결과가 이긴다(경고 1회). httpx 훅 user_context_request_hook 이 이 컨텍스트를 하류로 X-User-* 전파한다.
• skip_paths(Ant 패턴 — **·*·?, 경로는 scope["path"])와 OPTIONS 프리플라이트는 토큰을 파싱하지 않는다. 웹소켓은 손대지 않고 통과한다(웹소켓 인증은 비범위 — Origin 검사로 별도 방어).
• jwt-cookie 에서 accept_header=False 면 쿠키만 본다. 토큰 쿠키는 HttpOnly 고정이며 cookie=CookieSettings("RSCC_AT", same_site="Strict", domain=...) 로 속성을 바꾼다 (SameSite=None + secure=False 는 ValueError).

토큰 폐기 연결점 (revocation_checker) — JWT 는 만료 전엔 서명만으로 유효하므로, 로그아웃·계정 정지· 비밀번호 변경을 즉시 반영하려면 폐기 확인 함수를 주입한다(구현·저장소는 서비스 몫 — bearer·jwt-cookie 공통):

import hashlib

async def revoked(claims, token) -> bool:          # 동기 함수도 허용 — 결과가 awaitable 이면 await
    # (1) 사용자별 "이 시각 이전 발급분 폐기" — 비밀번호 변경·계정 정지 시 revoke-before:{sub} = now
    cutoff = await redis.get(f"revoke-before:{claims['sub']}")
    if cutoff is not None and claims.get("iat", 0) < int(cutoff):
        return True
    # (2) 로그아웃 토큰 거부 목록 — 원문 대신 해시 키, TTL = 토큰 남은 만료
    return await redis.exists(f"denylist:{hashlib.sha256(token.encode()).hexdigest()}") == 1

install_jwt_auth(app, SECRET, mode="jwt-cookie", revocation_checker=revoked)

• 서명·만료·kid 검증에 성공한 토큰에만 호출한다 — 위조·만료 토큰에는 저장소 조회를 하지 않는다.
• True → 미인증(보호 자원 401) + WARN 로그(토큰 원문 미기록). 예외 → fail-closed(미인증) + ERROR 로그 (fail-open 이 필요하면 함수 안에서 예외를 삼킨다). bool 이 아닌 결과(return 누락 등)도 fail-closed.
• 미등록(None, 기본)이면 현행 동작 그대로. 요청마다 호출되므로 네트워크 저장소는 async 로 구현한다.

CSRF (jwt-cookie — 자동 설치)

• 서버는 CSRF 쿠키가 없는 요청의 응답에 XSRF-TOKEN(HttpOnly 아님, Max-Age 없음, secrets.token_urlsafe(32)) 을 발급한다. 쿠키로 인증된 비안전 요청(POST/PUT/PATCH/DELETE …)은 X-XSRF-TOKEN 헤더 값이 쿠키 값과 같아야 한다(상수시간 비교). 아니면 핸들러 호출 없이 403 {"success":false,"code":"403","message":"CSRF 토큰이 없거나 올바르지 않습니다."} — 익명이어도 403. 헤더 인증·미인증 요청과 안전 메서드는 검사하지 않는다. 폼 파라미터 _csrf 는 받지 않는다.
• 클라이언트 의무: fetch credentials(같은 출처 "same-origin" 기본값 / 교차 출처 "include"), 비안전 요청마다 XSRF-TOKEN 쿠키 값을 X-XSRF-TOKEN 헤더로(같은 출처·허용 오리진에만 — 교차 출처로 토큰을 흘리지 말 것), 매 요청 쿠키를 다시 읽기(회전 대응), 토큰을 localStorage 등에 저장하지 않기. JS 는 @rscc/common-core 의 apiClient credentials·csrf 옵션이 이를 처리한다.
• 웹훅 등 제외: install_jwt_auth(..., csrf=CsrfSettings(exempt_paths=("/webhooks/**",))). skip_paths 도 자동 제외. csrf=None 은 옵트아웃(경고 로그). 이미 로그인된 브라우저가 다시 로그인하면 (쿠키 보유) 그 POST 도 CSRF 대상이다 — 클라이언트가 헤더를 붙이거나 로그인 경로를 exempt_paths 에 넣는다.
• 인증을 자체 구현한 서비스는 install_csrf(app, mode="jwt-cookie" | "session") 로 단독 설치한다.

배포 토폴로지

로컬 http 개발은 CookieSettings(..., secure=False) 가 필요하다(Secure 쿠키는 http 에서 저장되지 않음). 단일 호스트 배포는 CookieSettings("__Host-RSCC_AT") 로 형제 서브도메인의 쿠키 주입을 막을 수 있다.

미들웨어 등록 순서 — Starlette add_middleware 는 나중에 등록한 것이 바깥이다:

install_idempotency(app)          # (쓴다면) 가장 안쪽 — CSRF 403 이 멱등성 키를 소비하지 않게
install_jwt_auth(app, SECRET, mode="jwt-cookie")   # 인증 + CSRF
install_rate_limit(app)           # 그 밖 미들웨어
install_security_headers(app)
install_access_log(app)
install_trace_middleware(app)     # 마지막 = 가장 바깥 — CSRF 403 에도 X-Trace-Id 에코

세션 모드 — session

서버가 세션 저장소에 인증 상태를 두고 브라우저에는 세션 ID 쿠키(RSCC_SID)만 준다 — 로그아웃·강제 차단이 즉시 반영된다(무상태 JWT 와의 차이). 세션은 이 서비스 안에서만 쓰고, 하류 서비스에는 X-User-* 로 전파한다 (다언어 세션 공유는 비범위).

from fastapi import Depends, FastAPI
from pydantic import BaseModel

from rscc_common import get_session, install_session_auth, install_trace_middleware
from rscc_common.auth import AuthPrincipal, require_principal
from rscc_common.errors import CommonException, ResultCode
from rscc_common.handlers import install_exception_handlers   # [fastapi] — 401/403 을 봉투로 변환

app = FastAPI()
install_exception_handlers(app)
install_session_auth(app)                  # 기본: 인메모리 저장소(프로세스 단위), 유휴 1800초, CSRF 자동 설치
install_trace_middleware(app)              # 마지막 등록 = 가장 바깥 — CSRF 403 봉투에도 X-Trace-Id

class LoginBody(BaseModel):
    username: str
    password: str

@app.post("/auth/login")                   # session 모드는 로그인 POST 도 CSRF 대상 — X-XSRF-TOKEN 필요
def login(body: LoginBody):
    user = authenticate(body.username, body.password)          # 서비스 구현
    if user is None:
        raise CommonException(ResultCode.UNAUTHORIZED)
    get_session().login(user.id, user.role)                    # 세션 ID 교체 + CSRF 토큰 회전
    return {"id": user.id, "role": user.role}

@app.post("/auth/logout")
def logout():
    get_session().logout()                                     # 서버측 삭제 + RSCC_SID Max-Age=0
    return {"ok": True}

@app.get("/me")                            # 로그인 상태 확인 — 첫 GET 이 XSRF-TOKEN 쿠키도 받아 온다
def me(p: AuthPrincipal = Depends(require_principal())):      # 세션 없음·만료 → 401 봉투
    return {"id": p.subject, "role": p.role}

@app.post("/cart")                         # 익명 데이터도 가능 — 무언가 쓸 때 비로소 세션이 생긴다
def add_to_cart(item: str):
    cart = get_session().data.setdefault("cart", [])
    cart.append(item)
    return {"cart": cart}

동작:

• 쿠키 RSCC_SID=<id>; Path=/; Secure; HttpOnly; SameSite=Lax — Max-Age 없음(브라우저 세션 쿠키). cookie=CookieSettings("__Host-SID", same_site="Strict") 처럼 바꿀 수 있고 HttpOnly 해제는 ValueError.
• 세션 ID 는 secrets.token_urlsafe(32)(256비트). 저장소 키는 SHA-256(ID) hex 라 저장소 덤프가 살아있는 세션 ID 를 드러내지 않는다. 형식 밖 쿠키 값(16~256자 [A-Za-z0-9_-] 아님)은 저장소 조회 없이 "세션 없음".
• 지연 생성: data 에 아무것도 쓰지 않는 익명 요청은 세션도 Set-Cookie 도 만들지 않는다.
• login(subject, role=None): 세션 ID 를 교체(옛 ID 는 저장소에서 삭제 — 세션 고정 공격 방지)하고 CSRF 토큰 회전을 요청한다. data 는 새 ID 로 이어지고, 같은 요청의 get_principal()·UserContext 도 즉시 새 주체를 본다. regenerate() 는 ID 만 교체(권한 상승 시점 등). logout() 은 서버측 삭제 + 쿠키 Max-Age=0 — 옛 ID 는 다시 쓸 수 없다.
• 유휴 만료 = 저장소 TTL(idle_timeout_seconds, 기본 1800 — 마지막 요청 후 30분). 절대 만료 (absolute_timeout_seconds, 기본 없음) = 로그인(또는 세션 생성) 시각부터의 최대 수명 — 벽시계로 레코드에 보관하며 저장 TTL 도 남은 시간으로 캡한다.
• 저장 정책: 새 세션·data 변경·로그인 → save. 변경 없는 요청 → touch(저장소에 있으면, TTL 만 연장) 또는 save — 즉 살아있는 세션은 요청마다 저장소 쓰기 1회로 유휴 TTL 을 민다. 변경 감지는 data 첫 접근 시 사본과의 비교(중첩 수정도 감지). 변경은 응답 시작 시점에 저장되므로 핸들러 본문에서 한다.
• 주체는 AuthPrincipal(source="session", claims={"sub", "role"}) — require_principal(...) 이 jwt 모드와 똑같이 401(세션 없음·만료)/403(역할 부족)을 낸다. bind_user_context=True(기본)면 UserContext 를 세션 주체로 덮어써 X-User-* 위조를 막고(등록 순서 무관) httpx 훅이 하류로 전파한다.
• CSRF — install_csrf(mode="session") 자동 설치: 모든 비안전 메서드가 대상(세션 유무 무관 — 로그인 POST 포함). 클라이언트는 먼저 아무 GET(예: /me)으로 XSRF-TOKEN 을 받아 X-XSRF-TOKEN 헤더로 되돌린다. 클라이언트 의무·배포 토폴로지는 jwt-cookie 와 같다. csrf=CsrfSettings(exempt_paths=("/webhooks/**",)) 로 제외, csrf=None 은 옵트아웃(경고 로그).
• 저장소 조회 실패 → 미인증(fail-closed, ERROR 로그). 쓰기 실패 → 예외 전파(500 — 로그인·로그아웃이 조용히 실패하지 않게). 웹소켓은 손대지 않고 통과한다.
• 등록 순서는 install_jwt_auth 와 같다(install_idempotency → install_session_auth → … → install_trace_middleware). 한 앱에는 인증 모드 하나만 설치한다.

저장소 주입 (다중 인스턴스·멀티 워커 필수) — 기본 InMemorySessionStore(max_sessions=10000) 는 프로세스 단위라 워커·인스턴스끼리 세션을 공유하지 못한다(상한 초과 시 가장 오래 활동 없는 세션부터 축출). 분산 저장소를 SessionStore(동기) 또는 AsyncSessionStore(비동기 — 미들웨어가 await, 이벤트 루프를 막지 않음) Protocol 로 구현해 주입한다. 값은 내부 키가 든 dict 이므로 JSON 으로 보관하면 된다(data 에는 JSON 호환 값만):

import json

from redis.asyncio import Redis

class RedisSessionStore:                           # AsyncSessionStore 구현 (+ 선택 touch)
    def __init__(self, redis: Redis, prefix: str = "sess:") -> None:
        self._r, self._prefix = redis, prefix

    async def load(self, key: str) -> dict | None:
        raw = await self._r.get(self._prefix + key)
        return None if raw is None else json.loads(raw)

    async def save(self, key: str, data: dict, ttl_seconds: float) -> None:
        await self._r.set(self._prefix + key, json.dumps(data), px=max(1, int(ttl_seconds * 1000)))

    async def touch(self, key: str, ttl_seconds: float) -> None:   # 선택 — 없는 키는 무시(로그아웃된 세션을 되살리지 않음)
        await self._r.pexpire(self._prefix + key, max(1, int(ttl_seconds * 1000)))

    async def delete(self, key: str) -> None:
        await self._r.delete(self._prefix + key)

install_session_auth(app, RedisSessionStore(Redis.from_url(REDIS_URL)), absolute_timeout_seconds=8 * 3600)

• 저장소는 ttl_seconds 를 반드시 지켜야 한다(유휴 만료의 유일한 근거). touch 는 선택이지만 분산 저장소엔 권장 — 없으면 변경 없는 요청도 레코드 전체를 다시 써서 동시 요청의 data 쓰기를 덮거나 같은 순간 로그아웃된 세션을 되살릴 수 있다.
• 강제 차단(특정 사용자 세션 즉시 삭제)은 사용자→세션 색인이 필요하다 — 라이브러리는 색인을 두지 않으므로 로그인 직후 store_key(get_session().id) 를 사용자별로 기록해 두었다가 차단 시 store.delete(key) 한다.

알려진 제약

• 미처리 예외의 500 응답과 X-Trace-Id — 트레이스 미들웨어는 Starlette ServerErrorMiddleware 안쪽에서 실행되므로 500 응답에 에코하지 못한다. install_exception_handlers 의 catch-all 이 수신 X-Trace-Id 가 유효하면 그 값을 500 응답 헤더에 직접 에코해 부분 보완한다. 수신 값이 없어 미들웨어가 새로 생성한 경우엔 500 응답에 헤더가 없고, catch-all 의 에러 로그 라인도 traceId 가 - 로 찍힌다(그 시점엔 contextvar 가 이미 정리됨). HTTPException·검증 오류 등 핸들러 처리 경로(4xx)는 정상 에코된다.
• install_locale_middleware 의 Vary: Accept-Language 는 catch-all 500 응답에는 붙지 않는다(위와 같은 ServerErrorMiddleware 한계 — 500 의 message 언어는 request.state.rscc_locale 로 이어진다). CommonException.with_message_key 는 생성 시점에 해석하므로, 요청 컨텍스트 밖에서 만든 예외는 기본 언어다.
• CommonResponse 모델을 FastAPI 핸들러에서 직접 반환하지 말고 .to_wire() 를 반환할 것 — 직접 반환하면 data: null 이 실려 "데이터 없으면 키 생략" 계약이 깨진다.
• ErrorCodeEnum 은 정의 시점에 검증하지 않는다 — 형식 위반은 CommonException/CommonResponse.fail 생성 시점의 ValueError 로 드러나므로 validate_error_codes 테스트로 미리 잡는다. CommonException.result_code (및 error_code)의 타입 힌트는 ErrorCode 다 — Enum 전용 속성(.name/.value)은 isinstance 로 좁힌 뒤 쓴다.
• SseEventParser 는 스트림(연결)마다 1개(스레드 안전하지 않음), 한 파서에 str·bytes 청크를 섞지 않는 것을 권장한다(섞으면 str 도착 시 미완성 바이트를 U+FFFD 로 비운다). SSE_RESPONSE_HEADERS 는 공유 dict — 수정하지 말 것.
• configure_logging
  • 기본 파일명({app_name}.log)은 프로세스 1개 전제 — uvicorn --workers > 1 등 멀티 프로세스면 per_pid_file=True ({app_name}.{pid}.log) 필수.
  • 재호출 시 자신이 부착한 핸들러만 교체하고 다른 루트 핸들러(OTel·Sentry 등)는 보존한다 (clear_root_handlers=True 면 전부 제거 — 이전 버전의 동작). 기본 포맷에 [traceId] 가 들어간다(include_trace_id=False 로 이전 포맷).
  • level 기본값은 logging.DEBUG — 운영은 logging.INFO 권장.
  • install_secret_redaction() 은 configure_logging 이후에 호출(재구성마다 다시 호출).
• 인메모리 상태(install_rate_limit, install_idempotency·install_session_auth 기본 스토어, TtlCache, CircuitBreaker)는 프로세스(워커) 단위다 — 인스턴스 간 공유가 필요하면 분산 스토어를 IdempotencyStore·SessionStore/AsyncSessionStore Protocol 로 구현해 store= 로 주입한다.
• CircuitBreaker 를 수동 게이트(allow_request + 보고)로 쓸 때 보고가 누락되면 HALF_OPEN 프로브가 풀리지 않는다 — call/call_async 를 쓰면 보고가 보장된다.
• jwt 는 HS256 시크릿이 32바이트(UTF-8) 미만이면 ValueError (fail-fast, allow_weak_secret=True 는 교체 전 임시 옵트아웃). install_jwt_auth 도 설치 시점에 같은 게이트를 건다.
• jwt-cookie 는 무상태다 — 로그아웃은 쿠키 삭제일 뿐 이미 유출된 토큰은 만료까지 유효하다 (즉시 무효화가 필요하면 revocation_checker 폐기 연결점 또는 session 모드). 리프레시 토큰 회전·웹소켓 인증은 비범위.
• session 모드의 변경(data·login·logout)은 응답 시작 시점에 저장된다 — 스트리밍 본문 생성기·백그라운드 태스크에서의 변경은 저장되지 않는다. 같은 세션의 동시 요청이 data 를 바꾸면 마지막 저장이 이긴다.
• CSRF 미들웨어는 CSRF 쿠키가 없는 모든 응답에 XSRF-TOKEN 을 발급한다(규약) — 헤더 인증 API 클라이언트 응답에도 Set-Cookie 가 실릴 수 있다(무해).
• log_redact 는 정규식 기반 best-effort — 미탐·오탐이 공존한다. log_sanitize 와 병용 시 반드시 redact → sanitize(절단) 순서.
• rrn.is_valid_rrn 은 체크섬을 검사하지 않는다 — 2020-10 이후 발급분은 뒷자리가 임의번호라 mod-11 이 성립하지 않는다(checksum_ok_legacy 는 그 이전 발급분 전용). 저장 시 암호화 의무.
• text.chosung.of() 의 공백 스킵은 스페이스/탭만(Java 파리티).

버전·릴리스

버전의 단일 소스는 rscc_common.__version__ 이다. py-vX.Y.Z 태그 push 시 CI 가 태그와 버전 정합을 검증한 뒤 PyPI 에 배포한다(Trusted Publishing). 변경 이력은 소스 저장소의 CHANGELOG.md 에 있다.