rscc-common

RSCC 공통 Python 라이브러리. 응답 봉투, 에러 코드, X-Trace-Id 전파, ASGI 미들웨어, 인증 모드(bearer / jwt-cookie / session + CSRF·출처 검사·WebSocket 인증), 회복탄력성(재시도·서킷 브레이커·격벽·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, JsonFormatter, 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_exception_handlers(app)    # 예외 → 실패 봉투 (CommonException·검증 400·HTTPException·500) — 미들웨어 아님, 순서 무관
# (다른 install_* 미들웨어는 여기 — add_middleware 는 나중 등록이 바깥이므로 trace 보다 먼저)
install_trace_middleware(app)      # 마지막 = 가장 바깥 — X-Trace-Id 수신 재사용/생성 → contextvar → 응답 에코

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() 반환

미들웨어 등록 순서 — Starlette add_middleware 는 나중에 등록한 것이 바깥이다. install_trace_middleware 를 마지막(가장 바깥) 에 등록해야 레이트리밋 429·멱등성 409·CSRF 403 처럼 미들웨어가 즉시 보내는 응답에도 X-Trace-Id 가 에코된다. install_access_log 는 순서와 무관하게 동작하지만(바깥이면 수신 X-Trace-Id 로 폴백) 새로 생성된 traceId 까지 남기려면 trace 보다 먼저(안쪽에) 등록한다. install_exception_handlers 는 미들웨어가 아니라 순서와 무관하다:

install_exception_handlers(app)
install_user_context_middleware(app)
install_rate_limit(app)
install_security_headers(app)
install_access_log(app, headers=["Authorization"])
install_trace_middleware(app)      # 마지막 = 가장 바깥

(인증·멱등성·다국어까지 쓰는 전체 순서는 아래 "인증 모드"의 등록 순서와 메시지 다국어 참고.)

회복탄력성 — 재시도 안에서 서킷 브레이커로 게이트하고, 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 11종("200"·"400"·"401"·"403"·"404"·"405"·"409"·"415"·"429"·"500"·"503")은 공통 카탈로그이고 라이브러리는 공통 코드만 방출한다. SERVICE_UNAVAILABLE("503")은 2026-09 추가 — 인증 저장소 (세션 저장소·토큰 폐기 확인) 장애의 fail-closed 응답에 쓴다(이전엔 상태 503 이 폴백 규칙으로 "500" — EC-07). 서비스는 여기에 자기 도메인의 실패 코드를 얹어 쓸 수 있다.

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 처리)에도 쓰인다.

예외 응답 헤더 · 본문 없는 상태

CommonException 에 응답 헤더를 실을 수 있다(contracts/error-code-extension.md §5 — 공통 코드·도메인 코드 모두). install_exception_handlers 가 봉투 응답에 그대로 싣는다. 주지 않으면 현행과 같다.

from rscc_common import CommonException, ResultCode

raise CommonException(ResultCode.SERVICE_UNAVAILABLE, headers={"Retry-After": "30"})
# → HTTP 503 + {"success": false, "code": "503", ...} + Retry-After: 30
raise CommonException(ResultCode.UNAUTHORIZED, headers={"WWW-Authenticate": "Bearer", "X-Reason": "expired"})

• exc.headers 는 읽기 전용 매핑(삽입 순서, 없으면 빈 매핑) — 생성 시점에 검증한 사본이다. CommonException.with_message_key(ec, key, *args, headers={...}) 도 같다(키워드로만).
• 생성 시점 검증 (응답 헤더 주입 방지): 이름은 RFC 9110 token(공백·:·비ASCII 불가), 값은 CR·LF·NUL 금지, 봉투 형식을 정하는 Content-Type·Content-Length·Transfer-Encoding(대소문자 무시)은 지정 불가 → ValueError. 값은 latin-1 로 인코딩할 수 있어야 한다(ASGI 서버가 헤더를 latin-1 로 보낸다 — 응답 시점 500 대신 생성 시점 ValueError, Python 한정). 매핑이 아니거나 이름·값이 문자열이 아니면({"Retry-After": 30}) TypeError.

본문을 실을 수 없는 상태 (§6) — HTTP 1xx·204·205·304 는 본문을 가질 수 없다(RFC 9110). install_exception_handlers 는 이 상태의 HTTPException 을 봉투 없이 상태 + 예외 헤더만으로 응답한다(Starlette 기본 핸들러의 204·304 처리를 1xx·205 까지 넓힘). 조건부 GET 에 쓴다:

from fastapi import HTTPException, Request

@app.get("/items/{item_id}")
def get_item(item_id: int, request: Request):
    etag = f'"{version_of(item_id)}"'
    if request.headers.get("if-none-match") == etag:
        raise HTTPException(status_code=304, headers={"ETag": etag})   # 304, 본문 없음, ETag 보존
    ...

이전 버전은 304 에도 JSON 봉투를 실어 uvicorn(h11)이 LocalProtocolError 로 연결을 끊었다(2026-09 수정). 이 경로는 오류가 아니므로 warn 이 아니라 debug 로그만 남긴다. CommonException 의 상태는 검증상 200·4xx·5xx(ResultCode)· 400~599(도메인 코드)라 이 규칙에 닿지 않는다 — CommonException(ResultCode.SUCCESS) 도 200 + 봉투다.

검증 오류 — rejectedValue 제어

검증 오류 봉투의 data.errors[].rejectedValue 는 기본으로 스칼라 입력값을 싣는다(현행 — contracts/validation-errors.md). 비밀번호류 입력이 응답에 반사될 수 있으므로 서비스가 조절한다. 라이브러리는 민감 필드를 미리 정하지 않는다(이름 추측은 오탐·누락을 낳으므로 판단은 서비스 몫). 둘 다 주지 않으면 이전과 같다.

from rscc_common.handlers import install_exception_handlers

install_exception_handlers(app, sensitive_fields=("password", "pin", "user.email"))   # 필드별 제외
install_exception_handlers(app, include_rejected_value=False)                          # 전체 끄기

• 필드별 제외 — 오류의 와이어 field 가 목록의 이름과 대소문자 무시·정확 일치하면 그 오류의 rejectedValue 를 생략한다. 비교 대상은 ① field 전체(user.email) ② field 의 마지막 프로퍼티 이름(뒤쪽 인덱스 [N] 제거 — items[0].pin → pin, profile.tags[2] → tags). [0] 처럼 이름 없는 경로는 ①만.
• 부분 일치는 없다 — password 는 passwordHint 를 가리지 않는다(필요하면 둘 다 적는다). 전체 경로 이름 (user.email)은 그 경로만 가린다(admin.email 은 그대로).
• field·reason 은 항상 그대로다. reason 문구에 입력값을 넣는 사용자 정의 메시지는 이 제어 밖이다.
• 설치 시점 오류 — 빈·공백 이름, 문자열 하나를 그대로 준 sensitive_fields="password"(→ ("password",)) 는 ValueError, 문자열이 아닌 항목·bool 이 아닌 include_rejected_value 는 TypeError. 앞뒤 공백은 무시한다.

로깅 — 콘솔 전용 · JSON(ECS)

configure_logging 의 기본(콘솔 + 자정 롤링 파일, 텍스트 포맷)은 그대로이고, 두 키워드로 바꾼다.

import logging

from rscc_common import JsonFormatter, configure_logging

# 컨테이너 — stdout 만 (디렉토리·파일을 만들지 않는다. app_name·per_pid_file 은 쓰이지 않는다)
configure_logging(log_dir=None, level=logging.INFO)

# 한 줄 JSON — ECS 필드. 파일 모드에서도 formatter= 는 만든 핸들러(콘솔·파일) 전부에 쓰인다
configure_logging(log_dir=None, level=logging.INFO, formatter=JsonFormatter(service_name="order-api"))

log = logging.getLogger("order_api")
log.info("주문 조회: %s", 7, extra={"order_id": 7})

{"@timestamp":"2026-09-28T03:04:05.007Z","log":{"level":"INFO","logger":"order_api"},"process":{"pid":4242,"thread":{"name":"MainThread"}},"service":{"name":"order-api"},"message":"주문 조회: 7","trace":{"id":"3f9a1c2e"},"order_id":7,"ecs":{"version":"8.11"}}

• 필드 — @timestamp(UTC ISO-8601 밀리초 + Z), log.level(파이썬 레벨 이름 — INFO·WARNING)·log.logger, process.pid·process.thread.name, service.name(생성자 인자 — 없으면 생략), message(%s 인자 포맷 완료), trace.id(TraceIdFilter 값 — 요청 밖 - 이면 생략), extra={...} 속성(최상위 키로), 예외가 있으면 error.type(모듈.클래스 — 내장 예외는 이름만)·error.message·error.stack_trace, ecs.version("8.11").
• 중첩 객체 — Elasticsearch 는 {"log":{"level":…}} 와 {"log.level":…} 를 같은 필드로 다루지만, Java 쪽 Spring Boot 4 의 logging.structured.format.console=ecs 출력이 중첩 모양이라 그대로 맞췄다(같은 인덱스·대시보드). extra 의 점 표기 키("user.id")는 그대로 둔다.
• 한 줄 보장 — 여러 줄 메시지·스택트레이스도 JSON 이스케이프로 한 줄이다. 한글 등 비ASCII 는 이스케이프하지 않는다 (ensure_ascii=False — 파일은 UTF-8).
• 포맷 중 예외 없음 — JSON 으로 못 쓰는 extra 값(객체·datetime·Decimal·NaN·순환 참조 등)은 str(value) 로, % 인자 불일치는 원문 메시지로 남긴다. 포매터의 최상위 키(log·service·trace·error·ecs 등)와 같은 이름의 extra, _ 로 시작하는 속성, uvicorn 이 붙이는 color_message(ANSI 색 사본)는 싣지 않는다.
• formatter= 를 주어도 include_trace_id=True(기본)면 TraceIdFilter 는 그대로 붙는다. logging.Formatter 가 아니면 TypeError(기존 핸들러를 건드리기 전에 검사). 자체 구성(dictConfig)에서는 {"()": "rscc_common.logging.JsonFormatter", "service_name": "order-api"}.
• 리댁션 — install_secret_redaction()(핸들러 필터)은 포맷 전에 메시지를 바꾸므로 JSON message 도 리댁션된다. extra 값과 error.*(예외 메시지·스택)는 그 필터의 대상이 아니다 — 시크릿을 넣지 않는 것이 1차 방어다.

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

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

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)는 프레임워크 몫 — 이 규약의 범위 밖이다.

날짜·시각 — 기준 시간대 + offset 프로필

와이어 datetime 은 두 프로필이 있다. 기본은 local 프로필 — 오프셋 없는 벽시계 yyyy-MM-ddTHH:mm:ss(서버 기본 시간대, 운영 전제 KST). 키워드 인자를 주지 않으면 모든 함수·WireDateTime 이 이전과 바이트 동일하게 동작한다. 기준 시간대(wire zone), offset 프로필, 수신 오프셋 정책은 서비스 단위 opt-in 이다(Java rscc.time.*·JS timeZone/ offset/inboundOffset 과 같은 규약 — 골든 벡터 DT-01..14).

from datetime import datetime, timedelta, timezone
from rscc_common import datetimes

instant = datetime(2026, 7, 10, 4, 0, tzinfo=timezone.utc)
datetimes.to_wire_datetime(instant, zone="Asia/Seoul")                       # "2026-07-10T13:00:00"       (local)
datetimes.to_wire_datetime(instant, zone="Asia/Seoul", with_offset=True)     # "2026-07-10T13:00:00+09:00" (offset)
datetimes.to_wire_datetime(instant, zone="UTC", with_offset=True)            # "2026-07-10T04:00:00Z"      (0 오프셋은 Z)
datetimes.to_wire_datetime(instant, zone=timezone(timedelta(hours=9)))       # tzinfo 도 가능 — 시간대 DB 불필요

datetimes.parse_wire_datetime("2026-07-10T04:00:00Z")                        # 13:00 이 아니라 04:00 (drop — 현행 스큐)
datetimes.parse_wire_datetime("2026-07-10T04:00:00Z", zone="Asia/Seoul", inbound_offset="convert")  # 13:00 (naive)
datetimes.parse_wire_datetime("2026-07-10T04:00:00Z", inbound_offset="reject")                      # ValueError

datetimes.to_instant(datetime(2026, 7, 10, 13, 0), "Asia/Seoul")             # 2026-07-10 04:00 UTC (aware)

• zone — IANA 이름 str("Asia/Seoul", "America/New_York" — zoneinfo) 또는 tzinfo. aware 값은 그 시간대 벽시계로 변환하고, naive 값은 이미 그 시간대 벽시계로 간주한다. "UTC" 는 시간대 DB 없이 해석된다. 알 수 없는 이름은 ZoneInfoNotFoundError(KeyError 하위).

• with_offset=True (offset 프로필) — …+09:00·…-04:00, 0 오프셋은 Z. 오프셋은 그 순간의 기준 시간대 오프셋이다(뉴욕 여름 -04:00, 겨울 -05:00). zone 이 없으면 aware 값의 자체 오프셋을 쓰고, naive 값 + zone 없음은 ValueError. 직렬화는 local·offset 모두 초 단위(마이크로초 절단).

• DST — 벽시계 → 순간(to_instant, offset 프로필의 naive 값): 갭(없는 시각 — 뉴욕 2026-03-08 02:30)은 갭 길이만큼 뒤로(03:30-04:00), 겹침(두 번 있는 시각 — 2026-11-01 01:30)은 이른 오프셋(-04:00, fold=0). 입력 fold 는 무시한다(Java LocalDateTime.atZone 과 동일).

• 수신 오프셋 정책 inbound_offset — 결과는 항상 naive 벽시계이고, 오프셋 없는 입력은 모든 정책에서 그대로다. 소수부는 0~9 자리 허용(마이크로초까지 보존).

  정책	오프셋이 붙은 입력(…Z, …+09:00)
  drop (parse_wire_datetime·WireDateTime 기본 — 현행)	오프셋을 변환 없이 버림(±HHMM 포함) — 9시간 스큐가 조용히 생길 수 있다
  convert (make_wire_datetime 기본)	그 순간을 zone 의 벽시계로 변환 — zone 필수
  reject	ValueError (pydantic 필드면 검증 오류 → install_exception_handlers 기본 400)

  convert·reject 는 RFC 3339 오프셋(Z/±HH:MM)만 인식한다 — +0900 같은 표기는 형식 오류(ValueError).

pydantic 모델([fastapi] extra)은 make_wire_datetime 으로 필드 타입을 만든다 — 기준 시간대는 정의 시점에 해석되므로 오타·시간대 DB 부재가 첫 요청이 아니라 임포트에서 드러난다:

from pydantic import BaseModel
from rscc_common.response import make_wire_datetime

KstDateTime = make_wire_datetime(zone="Asia/Seoul")                              # local 출력 + convert 수신
KstOffsetDateTime = make_wire_datetime(zone="Asia/Seoul", with_offset=True)      # offset 출력 + convert 수신
StrictLocal = make_wire_datetime(zone=None, inbound_offset="reject")             # local 출력 + 오프셋 거부

class Event(BaseModel):
    at: KstDateTime

Event.model_validate_json('{"at": "2026-07-10T04:00:00Z"}').at    # datetime(2026, 7, 10, 13, 0) — naive KST
Event(at=datetime(2026, 7, 10, 13, 0)).model_dump_json()          # {"at":"2026-07-10T13:00:00"}

zone=None 은 with_offset=False 이고 inbound_offset 이 drop/reject 일 때만 허용된다. 정적 타입 검사기가 런타임 별칭을 거부하면 if TYPE_CHECKING: KstDateTime = datetime 으로 분기한다. 기존 WireDateTime 은 무변경이다.

전환 순서 (offset 프로필은 그 API 에 breaking) — 현행 수신 측(Java Jackson LocalDateTime)은 +09:00 을 거부하고, drop 수신은 오프셋을 버린다. 그래서:

1. 클라이언트가 먼저 수신 convert 를 켠다(inbound_offset="convert" + zone). 오프셋 없는 현행 응답도 그대로 읽힌다.
2. 모든 소비자가 배포된 뒤 서버가 with_offset=True(offset 프로필)로 바꾼다.
3. 요청 본문 쪽은 서버가 먼저 convert(또는 reject)를 켜 두면 클라이언트의 toISOString() 같은 Z 전송이 더는 스큐를 만들지 않는다.

시간대 DB (tzdata) — IANA 이름은 표준 라이브러리 zoneinfo 가 시스템 시간대 DB(/usr/share/zoneinfo 등)에서 읽는다. Windows 와 시스템 DB 가 없는 최소 컨테이너 이미지(이미지마다 다르다 — 배포 이미지에서 확인)에서는 ZoneInfoNotFoundError 가 난다 — 소비 앱에 pip install tzdata 를 추가한다(PyPI 의 IANA DB 패키지 — zoneinfo 가 자동으로 폴백). 라이브러리 기본 설치는 계속 의존성 0 이다. tzinfo 를 직접 주거나(timezone(timedelta(hours=9)) — DST 없는 고정 오프셋) "UTC" 만 쓰면 DB 가 필요 없다.

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). 예외는 폐기 확인 장애 하나 — 미들웨어가 바로 503 봉투로 응답한다(아래 revocation_checker).
• 역할 비교는 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 프리플라이트는 토큰을 파싱하지 않는다. 웹소켓은 기본적으로 손대지 않고 통과한다(주체 None — 현행). WebSocket 인증은 origin=OriginSettings(...) 로 출처 검사와 묶어서만 켠다(아래 "출처 검사·WebSocket").
• 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 로그(토큰 원문 미기록). 예외 → 503 + 공통 봉투 {"success":false,"code":"503","message":"일시적으로 서비스를 사용할 수 없습니다."}, 핸들러 미호출 + ERROR 로그 (fail-closed — 2026-09 변경, 이전엔 미인증 401. 저장소 장애를 "로그인 필요"로 오인해 클라이언트가 로그인 화면으로 보내지 않게. 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 에서 저장되지 않음).

하드닝 — __Host- 접두 (공유 도메인 배포에서는 필수, session-auth.md §2.4)

쿠키 이름에 __Host- 를 붙이면 브라우저는 Secure; Path=/ 이고 Domain 이 없는 쿠키만 그 이름으로 받아들인다 — 형제 서브도메인이 같은 이름의 쿠키를 심을 수(cookie tossing) 없다. Domain 속성과 양립하지 않아(같은 사이트 서브도메인 배포) 기본값으로 두지 않는다. CookieSettings 는 접두 요건(__Host- = secure=True·path="/"· domain=None, __Secure- = secure=True)을 어기면 생성 시점에 ValueError 를 던진다.

• [필수] 등록 도메인을 남과 공유하는 배포 — DDNS(*.iptime.org 등)처럼 공개 접미사 목록(PSL)에 없는 공유 상위 도메인 아래 호스트는 서로 같은 사이트다(SameSite 무력, 남이 Domain=<상위 도메인> 쿠키를 심을 수 있음). 자격·CSRF 쿠키 모두 __Host- 이름을 쓰고, 아래 중복 거부와 출처 검사를 켠다.
• 중복 자격 쿠키 거부 (reject_duplicate_cookies=True, 기본 꺼짐 — §2.3) — 토큰 쿠키 이름이 2번 이상 나오면(여러 Cookie 헤더는 이어서 본다) "첫 값" 대신 자격 없음 + 보안 이벤트 로그 duplicate_credential_cookie (쿠키 값 미기록). 형제 호스트가 심은 쿠키로 사용자가 공격자 세션에 로그인된 상태가 되는 것을 막는다(브라우저는 경로가 긴 쿠키를 먼저 보내므로 "첫 값" 규칙으로는 못 막는다). 헤더 토큰이 있으면 헤더 우선 그대로. 자동 설치되는 CSRF 의 출처 판정도 같은 규칙을 쓴다. 대가: 심어진 쿠키가 남아 있는 동안 그 사용자는 로그인되지 않는다.

install_jwt_auth(app, SECRET, mode="jwt-cookie",
                 cookie=CookieSettings(name="__Host-RSCC_AT"),
                 csrf=CsrfSettings(cookie=CookieSettings(name="__Host-XSRF-TOKEN", http_only=False)),
                 reject_duplicate_cookies=True,
                 origin=OriginSettings(trusted_origins=("https://app.example.com",)))

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

install_idempotency(app)          # (쓴다면) 가장 안쪽 — CSRF 403 이 멱등성 키를 소비하지 않게
install_api_key_auth(app, resolve)  # (쓴다면) API 키 영역 — JWT 보다 안쪽이어야 주체가 덮이지 않는다
install_jwt_auth(app, SECRET, mode="jwt-cookie")   # 인증 + CSRF (+ origin= 이면 출처 검사가 그 바깥)
install_rate_limit(app)           # 그 밖 미들웨어
install_security_headers(app)
install_access_log(app)
install_trace_middleware(app)     # 마지막 = 가장 바깥 — CSRF 403 에도 X-Trace-Id 에코

API 키 인증 — /openapi/**

외부 백엔드(파트너 서버 등)가 Authorization: Bearer rscc_live_… 로 호출하는 API 키 영역을 인증한다(선택, 기본 꺼짐 — contracts/api-key.md "인증 필터", 골든 벡터 AK-09~20). 키 문자열 규약은 rscc_common.api_key, 이 기능은 그 위의 순수 ASGI 미들웨어다. 키 저장소 조회는 앱이 리졸버로 주입한다 — 라이브러리는 키 원문 대신 SHA-256 hex 를 넘긴다:

from fastapi import Depends, FastAPI
from rscc_common import api_key, install_api_key_auth, install_jwt_auth, install_trace_middleware
from rscc_common.api_key_auth import ApiKeyPrincipal
from rscc_common.auth import AuthPrincipal, require_principal
from rscc_common.handlers import install_exception_handlers

async def resolve(key_hash: str) -> ApiKeyPrincipal | None:     # 동기 함수도 된다. 예외 = 저장소 장애(503)
    row = await db.fetch_one("select partner_id, role, id from api_keys where key_hash = :h and not revoked",
                             {"h": key_hash})
    return None if row is None else ApiKeyPrincipal(f"apikey:{row.partner_id}", row.role, str(row.id))

app = FastAPI()
install_exception_handlers(app)
install_api_key_auth(app, resolve, accepted_envs=("live",))   # 인증 미들웨어보다 먼저(안쪽)
install_jwt_auth(app, SECRET)                                  # /openapi/** 를 skip-paths 에 자동 합류
install_trace_middleware(app)

@app.post("/openapi/orders")
def create_order(p: AuthPrincipal = Depends(require_principal("PARTNER"))):   # p.source == "api_key"
    ...

# 발급 — 원문은 응답에서 1회만 보여주고 저장은 해시만
raw = api_key.issue("live")
await db.execute("insert into api_keys(partner_id, role, key_hash) values (:p, 'PARTNER', :h)",
                 {"p": partner_id, "h": api_key.sha256_hex(raw)})

• 검사 순서: 헤더 추출(기본 Authorization + scheme Bearer — 대소문자 무시, scheme=None 이면 헤더 값 전체) → 형식 → 체크섬 → 환경(accepted_envs) → resolver(sha256_hex(key)). 앞 단계는 저장소 조회 전 게이트다.
• 거절하지 않는다(JWT·세션과 같음) — 키 없음·불량·미등록은 주체 None, 401 은 require_principal 이 낸다. 리졸버 예외만 503 + 공통 봉투(핸들러 미호출). 리졸버가 ApiKeyPrincipal·None 아닌 값을 돌려주면 구현 버그로 보고 미인증 + ERROR.
• API 키 영역은 API 키로만 인증한다 — 영역 안에서는 세션·JWT 쿠키 주체를 API 키 결과로 덮는다. 그래서 먼저 설치해 두면 이후 설치하는 install_jwt_auth(skip-paths)·CSRF(install_*_auth 가 자동 설치하는 것 포함)·install_origin_check 가 이 경로를 자동 제외한다 — 서버 간 호출이 CSRF·출처 검사로 403 이 되지 않는다. JWT 미들웨어는 skip-path 에서도 주체를 None 으로 덮으므로 API 키가 안쪽이어야 한다 — 인증·CSRF·출처 검사 미들웨어가 이미 설치돼 있으면 ValueError.
• 주체: AuthPrincipal(source="api_key"), claims = sub·role·key_id(있으면)·env. UserContext 도 이 주체로 덮는다 (bind_user_context=True) — 하류로는 subject 가 X-User-Id 가 되므로 사용자와 구분하려면 apikey: 같은 이름공간을 붙인다.
• 보안 이벤트 로그 api_key_rejected(malformed·checksum·env·unknown) — 키 원문은 기록하지 않고 형식이 맞는 키만 env·key_hash(SHA-256 앞 12자)를 남긴다.
• OPTIONS 프리플라이트는 조회 없이 통과, WebSocket 은 손대지 않는다. 레이트리밋은 IP 기준 그대로(키별 한도는 리졸버·핸들러 몫).
• 테스트·소규모 파트너는 static_api_key_resolver({api_key.sha256_hex(raw): ApiKeyPrincipal("partner-1", "PARTNER")}) (상수시간 전수 대조)로 충분하다.

세션 모드 — 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.

• 영속 세션 쿠키(선택, 기본 꺼짐 — §2.2 CK-07): persistent_cookie=True 면 쿠키에 Max-Age = 절대 만료까지 남은 초(내림, 최소 1)를 실어 브라우저를 닫아도 로그인이 유지된다. absolute_timeout_seconds 필수(없으면 ValueError) — 쿠키가 서버 세션의 최대 수명보다 오래 남지 않게. 쿠키는 발급(로그인·regenerate()·지연 생성) 때만 내리므로 활동해도 늘지 않는다. 유휴 만료는 그대로라, 브라우저를 오래 닫아 둬도 유지하려면 유휴 만료도 늘린다:

  install_session_auth(app, RedisSessionStore(r),
                       idle_timeout_seconds=7 * 86400,        # 7일 쉬면 로그아웃
                       absolute_timeout_seconds=30 * 86400,   # 로그인 후 최대 30일 (쿠키 Max-Age 도 이 기준)
                       persistent_cookie=True)
  

• 세션 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 은 옵트아웃(경고 로그).

• 저장소 장애 — 세션 쿠키(형식 적합)가 있는 요청에서 저장소 조회가 예외를 던지면 503 + 공통 봉투 {"success":false,"code":"503","message":"일시적으로 서비스를 사용할 수 없습니다."}, 핸들러 미호출 + ERROR 로그 (fail-closed — 2026-09 변경, 이전엔 미인증 401. 클라이언트는 로그인 화면이 아니라 재시도·장애 안내를 한다). 세션 쿠키가 없거나 형식 밖인 요청은 저장소를 부르지 않으므로 영향이 없다. 응답 시점 쓰기 실패 → 예외 전파 (500 — 로그인·로그아웃이 조용히 실패하지 않게).

• 중복 세션 쿠키 거부 (reject_duplicate_cookies=True, 기본 꺼짐) — RSCC_SID 가 2번 이상이면 익명(저장소 미조회) + duplicate_credential_cookie 로그. 공유 도메인 배포는 cookie=CookieSettings(name="__Host-RSCC_SID") 와 함께 쓴다(위 "하드닝").

• 웹소켓은 기본적으로 손대지 않고 통과한다(주체 None). origin=OriginSettings(...) 를 주면 핸드셰이크를 읽기 전용으로 인증한다(아래 "출처 검사·WebSocket").

• 등록 순서는 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 쓰기를 덮거나 같은 순간 로그아웃된 세션을 되살릴 수 있다.

사용자별 세션 일괄 폐기 (강제 차단 — 선택, 기본 꺼짐) — install_session_auth(..., user_revocation=True) 가 돌려주는 SessionAuth 핸들로 한 사용자의 지금까지 로그인한 모든 세션을 무효화한다(비밀번호 변경·계정 정지 등):

auth = install_session_auth(app, RedisSessionStore(r), user_revocation=True)

@app.post("/admin/users/{uid}/block", dependencies=[Depends(require_principal("ADMIN"))])
async def block(uid: str):
    await auth.arevoke_user_sessions(uid)     # 비동기 저장소 — 동기 저장소는 auth.revoke_user_sessions(uid)
    return {"ok": True}

• 저장소에 사용자별 "이 시각 이전에 로그인한 세션은 무효" 표식(키 = SHA-256("rscc-user:" + subject), 값 = 폐기 벽시계, TTL = 유휴 만료 + 여유 60초)을 남긴다. 로그인된 세션의 요청마다 표식을 조회해 로그인 시각 ≤ 표식 시각이면 세션을 삭제하고 미인증으로 처리한다 — 재로그인한 새 세션은 유효, 다른 사용자는 무영향.
• 폐기 직후 재로그인(SS-13) — 비밀번호 변경처럼 한 요청에서 revoke_user_sessions(uid) 후 get_session().login(uid) 하면 다른 기기의 세션만 무효가 되고 새 세션은 유효하다. 표식·로그인 시각은 벽시계를 따르되 한 설치 안에서 엄격히 증가하도록 발급한다 — 벽시계 해상도가 거친 환경(Windows 의 Python 3.11·3.12 = 15.6ms)에서도 두 시각이 같아지지 않는다(py 0.9.0 까지는 이 환경에서 새 세션도 무효가 됐다).
• 대가: 로그인된 세션의 요청마다 저장소 조회 1회 추가. 워커·인스턴스 사이에는 시계 해상도·차이만큼 오차가 있다. 표식 조회 장애는 세션 조회 장애와 같이 503.
• user_revocation=False(기본)에서 호출하면 RuntimeError(조용한 무시 방지). 비동기 저장소에 동기 revoke_user_sessions 를 쓰면 TypeError(→ await arevoke_user_sessions). 호출마다 보안 이벤트 로그(INFO) user_sessions_revoked subject=<사용자> 를 남긴다(요청 안이면 method·path 포함).
• 특정 세션 하나만 골라 지우려면 로그인 직후 store_key(get_session().id) 를 기록해 두었다가 store.delete(key) 한다.

사용자당 세션 상한 (동시 세션 수 제한 — 선택, 기본 무제한) — max_sessions_per_user=N 을 주면 한 사용자가 로그인해 둘 수 있는 세션을 N 개로 제한한다. 로그인할 때 N 개를 넘으면 로그인 시각이 가장 오래된 세션부터 저장소에서 삭제한다 (방금 로그인한 세션은 항상 남는다). Spring maximumSessions(N) 기본값처럼 새 로그인을 받고 기존 세션을 끊지만, Spring 은 마지막 요청이 가장 오래된 세션부터 끊는다는 점이 다르다(session-auth.md §7):

install_session_auth(app, RedisSessionStore(r), max_sessions_per_user=10)   # 11번째 로그인 → 가장 오래된 세션 끊김

• 끊긴 세션의 다음 요청은 미인증(401) — 클라이언트는 평소처럼 로그인 화면으로 보낸다. 끊을 때마다 보안 이벤트 로그(INFO) user_sessions_evicted reason=limit subject=<사용자> evicted=<수> limit=<N> 를 남긴다(세션 ID·저장소 키는 기록 안 함).
• 세는 대상: 살아있는 세션만. 로그아웃·유휴 만료·절대 만료된 세션은 세지 않는다(SS-16). user_revocation=True 와 함께 켜면 사용자별 폐기된 세션도 세지 않는다(SS-25 — 등록 때 표식 조회 1 추가). 같은 기기의 재로그인은 옛 ID 를 교체할 뿐이라 수가 늘지 않고(SS-19), regenerate() 한 세션은 새 ID 로 계속 센다(SS-18). 다른 사용자는 무영향(SS-17).
• 추적 방식: 저장소에 사용자별 세션 색인(키 = SHA-256("rscc-sessions:" + subject), 값 = 세션 저장소 키 목록)을 두고 로그인·regenerate() 때 갱신한다 — 앱이 내부 레코드 키를 읽을 일이 없다. 로그인된 세션은 재등록 주기(min(유휴 만료, 1시간))마다 색인에 다시 등록되므로, 저장소 축출로 색인이 사라지거나 상한을 나중에 켜도 기존 세션이 다음 요청에서 등록된다(그 요청의 세션은 남고, 넘치면 나머지 중 로그인이 오래된 것이 끊긴다).
• 비용: 로그인·regenerate() 마다 저장소 조회 1 + 색인에 든 세션 수(최대 N) + 쓰기 1(+ 끊는 세션 삭제). 로그인된 세션은 재등록 주기마다 한 번 같은 작업을 하고 그 요청은 touch 대신 save 한다. 그 밖의 요청은 추가 비용 없음. 색인도 저장소 항목이라 InMemorySessionStore(max_sessions=...) 상한에 포함된다.
• 원자성: 기본 저장소 계약(load·save·delete)만으로는 색인을 읽고 다시 쓰므로, 같은 사용자가 여러 워커·인스턴스에서 동시에 로그인하면 한쪽 색인 갱신을 잃을 수 있다 — 상한을 잠시 넘을 수 있고 재등록 주기 안에 바로잡힌다. 경합을 없애려면 저장소에 아래 원자 색인 연산을 구현한다.
• 응답 시점의 색인 조회·저장 실패는 세션 저장 실패와 같이 예외 전파(500) — 상한을 조용히 건너뛰지 않는다. WebSocket 핸드셰이크는 읽기 전용이라 색인을 건드리지 않는다.

세션 수 조회 — 끊는 대신 새 로그인 거부 (상한을 켰을 때) — 라이브러리는 로그인을 거부하지 않는다(거부는 앱 흐름 결정). SessionAuth.acount_user_sessions(subject)(동기 저장소는 count_user_sessions)로 살아있는 세션 수를 세어 앱이 판단한다:

auth = install_session_auth(app, RedisSessionStore(r), max_sessions_per_user=3)

@app.post("/auth/login")
async def login(body: LoginBody):
    user = authenticate(body)
    # exclude_current=True — 이 기기의 기존 세션은 재로그인으로 교체되므로 빼고 센다(같은 기기 재로그인은 허용)
    if await auth.acount_user_sessions(user.id, exclude_current=True) >= auth.max_sessions_per_user:
        raise CommonException(ResultCode.CONFLICT, "다른 기기에서 로그인 중입니다.")
    get_session().login(user.id, user.role)
    return {"id": user.id}

• 읽기 전용(저장소 쓰기·보안 로그 없음)이고 원자적이지 않다 — 조회와 로그인 사이에 다른 기기가 로그인하면 상한 축출이 안전망이 된다. 세는 대상은 위 "세는 대상"과 같다(SS-20~22). 색인에 아직 등록되지 않은 세션(상한을 켜기 전의 세션 등)은 다음 재등록 전까지 세지 않는다.
• 로그인 전에 부른다 — 이번 요청의 로그인은 응답 시점에 저장된다. exclude_current=True 는 이번 요청이 불러온 세션을 뺀다(SS-23, 요청 밖이면 영향 없음).
• max_sessions_per_user 없이 부르면 RuntimeError(SS-24), 비동기 저장소에 동기 count_user_sessions 는 TypeError. 저장소 예외는 그대로 전파된다. 비용: 색인 조회 1 + 색인에 든 세션 수(+ 사용자별 폐기가 켜져 있으면 표식 1) — 요청마다 부르지 말고 로그인 때만 쓴다.

원자 색인 연산 (선택 — 동시 로그인 경합 제거) — 저장소가 index_members·index_admit·index_remove (AtomicSessionIndex/AsyncAtomicSessionIndex Protocol) 셋 다 구현하면 색인을 load/save 대신 이 연산으로만 다룬다 (일부만 있으면 설치 시 ValueError). index_admit 이 "자기 추가 + 로그인이 오래된 순으로 초과분 제거"를 한 번에 해 갱신 유실이 없고(SS-26), 세션은 저장한 뒤 색인에 넣으며, 로그아웃·다른 주체로 전환한 세션은 색인에서 즉시 빠진다. Redis 는 정렬 집합(점수 = 로그인 시각) + 단일 키 Lua 로 구현한다 — 위 RedisSessionStore 에 추가:

_ADMIT = """
redis.call('ZREM', KEYS[1], ARGV[1])
local excess = redis.call('ZCARD', KEYS[1]) - (tonumber(ARGV[3]) - 1)
local evicted = {}
if excess > 0 then
  evicted = redis.call('ZRANGE', KEYS[1], 0, excess - 1)
  redis.call('ZREMRANGEBYRANK', KEYS[1], 0, excess - 1)
end
redis.call('ZADD', KEYS[1], ARGV[2], ARGV[1])
redis.call('PEXPIRE', KEYS[1], ARGV[4])
return evicted
"""

class RedisSessionStore:                  # (계속) AsyncAtomicSessionIndex 구현
    def __init__(self, redis: Redis, prefix: str = "sess:", index_prefix: str = "sessidx:") -> None:
        self._r, self._prefix, self._index_prefix = redis, prefix, index_prefix
        self._admit = redis.register_script(_ADMIT)

    async def index_members(self, index_key: str) -> list:
        return await self._r.zrange(self._index_prefix + index_key, 0, -1)

    async def index_admit(self, index_key: str, member: str, score: float, limit: int, ttl_seconds: float) -> list:
        return await self._admit(
            keys=[self._index_prefix + index_key],
            args=[member, repr(score), limit, max(1, int(ttl_seconds * 1000))],
        )

    async def index_remove(self, index_key: str, members: list[str]) -> None:
        await self._r.zrem(self._index_prefix + index_key, *members)

• 정렬 집합은 레코드와 다른 접두사(sessidx:)에 둔다 — 기본(JSON) 색인이 같은 이름으로 남아 있으면 전환 시 타입 충돌 (WRONGTYPE)이 난다. 반환 원소는 str 또는 ASCII bytes(redis-py 기본 응답) 모두 받는다. 단일 키 스크립트라 Redis Cluster 에서도 동작한다.
• 한 배포의 모든 인스턴스가 같은 색인 표현을 써야 한다(섞으면 상한이 깨진다). 원자 경로로 바꾼 직후 옛 JSON 색인은 쓰이지 않고, 기존 세션은 재등록 주기 안에 정렬 집합에 다시 등록된다.
• 자기 세션 저장 후 index_admit 이 실패하면 로그인 응답은 500(쿠키 없음)이고 저장된 레코드는 쿠키 없이 TTL 로 사라진다.

출처 검사·WebSocket — Origin / Fetch Metadata

CSRF 토큰의 심층 방어(선택, 기본 꺼짐 — contracts/session-auth.md §4.3·§10). 브라우저가 붙이는 Sec-Fetch-Site· Origin 헤더로 "다른 출처에서 시작된 요청"을 거절한다. 토큰 검사가 닿지 않는 곳 — jwt-cookie 로그인 요청, 같은 사이트 형제 호스트, WebSocket 핸드셰이크(교차 사이트 WebSocket 하이재킹) — 을 메운다. 모드 무관하게 켤 수 있다:

from rscc_common import install_origin_check, install_session_auth
from rscc_common.origin import OriginSettings

origin = OriginSettings(
    trusted_origins=("https://app.example.com",),   # scheme://host[:port] 만 — 경로·쿼리·조각·* 는 ValueError
    allow_same_host=True,                            # Origin 의 host[:port] == 요청 Host 면 통과 (기본)
    exempt_paths=("/webhooks/**",),                  # Ant 패턴 — 검사 제외
)
install_session_auth(app, origin=origin)             # 또는 install_jwt_auth(app, SECRET, ..., origin=origin)
# install_origin_check(app, origin)                  # 인증 없이 출처 검사만 (단독 설치)

• 대상: 비안전 메서드(GET·HEAD·OPTIONS·TRACE 외) HTTP 요청 + WebSocket 핸드셰이크(메서드 무관).
• 판정 순서(처음 결론에서 멈춤): ① exempt_paths → 통과 ② Origin 이 신뢰 목록 → 통과 ③ Sec-Fetch-Site 가 있으면 same-origin·none → 통과, 그 외(same-site 포함) → 거절 cross_origin ④ Origin 없음 → 통과(서버 간 호출·CLI) ⑤ Origin: null → 거절 null_origin ⑥ allow_same_host 이고 Origin 의 host[:port] == Host (대소문자 무시) → 통과 ⑦ 그 외 → 거절 untrusted_origin. 신뢰 출처와 요청 Origin 은 소문자화·기본 포트(443/80) 제거·끝 / 하나 제거로 정규화해 정확 일치로 비교한다("https://App.Example.com:443/" == "https://app.example.com").
• 거절 — HTTP: 핸들러 호출 없이 403 {"success":false,"code":"403","message":"허용되지 않은 출처의 요청입니다."} (규약 고정 문구 — en "The request origin is not allowed."). WebSocket: accept 전 websocket.close code 1008 (서버는 HTTP 403). 둘 다 보안 이벤트 로그 origin_rejected.
• 설치 위치 — install_*_auth(origin=...) 는 출처 검사를 CSRF 보다 바깥에 설치한다(출처 거절 → CSRF → 인증 순). 단독 install_origin_check 는 인증·CSRF 보다 나중에 등록해야 그보다 먼저 거절한다.
• 리버스 프록시 — ⑥ 은 요청의 원본 Host 헤더를 쓴다(X-Forwarded-Host 미반영). 프록시가 Host 를 바꾸면 같은 출처 요청이 거절되므로 프론트 출처를 trusted_origins 에 명시한다(명시가 가장 안전하다).

WebSocket 인증 — origin 을 주면 인증 미들웨어가 WebSocket 핸드셰이크도 인증한다(origin=None 이면 현행대로 손대지 않음 — 출처 검사 없이 WebSocket 쿠키 인증을 켜는 구성은 제공하지 않는다):

1. 출처 검사(위 판정 순서, 메서드 무관) — 실패하면 close 1008, 앱 미호출.
2. 자격 해석 — session 은 세션 쿠키로 읽기 전용 조회(쿠키 발급·저장·TTL 연장 없음), bearer/jwt-cookie 는 모드 규칙대로. 중복 거부·사용자별 폐기·토큰 폐기 확인도 HTTP 와 같다.
3. 저장소·폐기 확인 장애 → HTTP 503 + 공통 봉투 거절 응답(ASGI websocket.http.response 확장 — uvicorn·hypercorn 지원), 확장이 없는 서버는 close 1011(→ HTTP 403). 앱 미호출. 브라우저는 어느 쪽이든 이유를 볼 수 없다(close 1006) — 클라이언트가 미인증을 구분해야 하면 앱이 accept() 한 뒤 close(4401) 처럼 닫는다(accept 후 close 코드는 브라우저에 보인다).
4. 인증되면 주체(get_principal(), UserContext)를 연결 수명 동안 바인딩한다. get_session() 은 WebSocket 에서 None 이다(세션 변경 불가). 미인증 연결을 받을지는 앱이 정한다:

from fastapi import WebSocket
from rscc_common import get_principal

@app.websocket("/ws/notifications")
async def notifications(websocket: WebSocket):
    principal = get_principal()                # 핸드셰이크 시점의 인증 주체 (연결 수명 동안 유지)
    if principal is None:
        await websocket.close(code=1008)       # accept 전 close — 서버는 HTTP 403 으로 거절
        return
    await websocket.accept()
    await websocket.send_json({"hello": principal.subject})

• 열린 연결 동안의 로그아웃·만료·폐기는 연결을 끊지 않는다 — 긴 연결은 앱이 주기적으로 재확인한다.

보안 이벤트 로그

거절·차단을 운영자가 추적하도록 전용 로거 rscc_common.security.audit 에 한 줄씩 남긴다(감사 시스템 연결은 이 로거의 핸들러로, 공격 중 대량 발생은 로거 레벨·필터로 조절). 레벨 WARN(폐기 작업 기록은 INFO), traceId 는 TraceIdFilter 가 붙인다. 토큰·쿠키 값·세션 ID·CSRF 토큰·API 키 원문은 절대 기록하지 않는다(Origin 은 256자로 자름, 제어문자는 공백 치환). 기존 로그(토큰 검증 실패 등 각 모듈 로거)는 그대로다.

[SECURITY] event=csrf_rejected reason=missing_header method=POST path=/orders
[SECURITY] event=origin_rejected reason=cross_origin method=POST path=/orders origin=https://evil.example
[SECURITY] event=duplicate_credential_cookie reason=duplicate method=GET path=/me cookie=RSCC_AT
[SECURITY] event=api_key_rejected reason=unknown method=GET path=/openapi/items env=live key_hash=d44eccd8f51c
[SECURITY] event=user_sessions_revoked reason=revoked method=POST path=/admin/users/42/block subject=42

레이트리밋 게이트 — 키·제외·거부

install_rate_limit 은 키별 토큰버킷으로 HTTP 요청을 세고, 한도를 넘으면 앱을 호출하지 않고 429 + 공통 봉투 ({"success": false, "code": "429", "message": "요청 한도를 초과했습니다."})를 즉시 응답한다 (contracts/rate-limit.md, Java RateLimitFilter 와 골든 벡터 RG-01~11 공유). 키·제외는 인자를 주지 않으면 이전 버전과 같다 — 모든 HTTP 요청을 피어 IP 키로 센다. websocket·lifespan 은 게이트 없이 통과한다. 429 에는 기본으로 Retry-After 헤더가 붙는다(아래 — 헤더 추가라 모르는 클라이언트는 무시한다).

from rscc_common import install_rate_limit

# (a) 리버스 프록시·로드밸런서 뒤 — 신뢰 설정 필수 (contracts/client-ip.md)
install_rate_limit(app, trusted_hops=1)                     # 프록시 1홉: XFF 오른쪽에서 1번째 = 클라이언트
install_rate_limit(app, trusted_proxies=["10.0.0.0/8"])     # 또는 신뢰 CIDR — rightmost-untrusted (hops 대신 동작)

# (b) 헬스체크 경로 제외 + CORS 사전 요청(OPTIONS) 제외 — 둘 다 토큰을 쓰지 않는다
install_rate_limit(
    app,
    exempt_paths=["/health", "/actuator/**"],               # Ant 패턴(* ** ?) — scope["path"](쿼리 제외), 대소문자 구분
    should_limit=lambda scope: scope["method"] != "OPTIONS",  # scope → bool: True 면 센다, False 면 제외
)

# (c) 키 직접 지정 — 사용자 ID·API 키 해시 등 (신뢰 설정과 함께 주면 설치 시점 ValueError)
install_rate_limit(app, key_func=lambda scope: tenant_of(scope))

# (d) 429 에 Retry-After 를 싣지 않으려면 (기본은 켜짐)
install_rate_limit(app, retry_after=False)

• 키 — key_func 가 있으면 그 값. 없으면 클라이언트 IP: 신뢰 설정이 없으면(기본) 피어 IP(scope["client"])이고 X-Forwarded-For/X-Real-IP 는 무시한다(스푸핑 차단 안전 기본값). 그래서 프록시 뒤에서 신뢰 설정을 빼먹으면 모든 사용자가 프록시 IP 하나의 한도를 나눠 쓴다 — 인프라에 맞는 trusted_hops 또는 trusted_proxies 를 줄 것. 키를 얻지 못하면(UDS 등) "-" 하나로 센다.
• 판단 순서 — ① exempt_paths → ② should_limit → ③ 키 → ④ 버킷. ①·② 에서 제외된 요청은 토큰을 쓰지 않고 그대로 통과한다(제외 경로는 should_limit·key_func 를 부르지 않는다).
• fail-closed — should_limit 이 예외를 던지거나 bool 이 아닌 값(return 누락 None, async 함수의 코루틴 등)을 돌려주면 그 요청은 센다 + 발생마다 경고 로그 1줄(rscc_common.rate_limit). 예외 내용은 응답에 싣지 않는다. should_limit 은 동기 함수만 지원한다.
• Retry-After (기본 켜짐) — 429 응답에 Retry-After: <초> 를 싣는다. 값 = 거부된 그 키의 버킷에 토큰 1개가 다시 찰 때까지 남은 시간(TokenBucket.seconds_until_available(1))을 올림한 정수, 최소 1·최대 86400(1일) — RFC 9110 delay-seconds. 예: 용량 1·리필 0.5/초로 막 소진 → 2, 리필 4/초(남은 0.25초) → 1, 리필이 사실상 0 → 86400. 버킷이 워커 단위라 이 값은 그 워커(인스턴스) 기준 힌트다 — 로드밸런서가 다른 워커로 보내면 더 빨리 통과할 수 있고, 같은 키로 다른 요청이 먼저 토큰을 쓰면 더 늦어질 수 있다. 끄려면 retry_after=False(200 등 통과 응답에는 원래 없다). 클라이언트 쪽은 rscc_common.retry.parse_retry_after 로 읽어 재시도 대기 하한으로 쓰면 된다.
• 설정 오류는 설치 시점 예외 — key_func + trusted_hops/trusted_proxies 동시 지정·음수 trusted_hops·해석 불가 CIDR·문자열 하나를 그대로 준 exempt_paths/trusted_proxies(("/health",) 처럼 시퀀스로) → ValueError, 정수가 아닌 trusted_hops·호출 불가 should_limit·bool 이 아닌 retry_after("false"·0 등) → TypeError.
• 한계 — 버킷은 인메모리·프로세스(워커) 단위다(다중 워커·인스턴스면 각자 독립 한도, 분산 한도 비범위). max_keys(기본 10000)를 넘으면 ① 토큰이 가득 찬 버킷(새 버킷과 같아 손실 없음)을 정리하고 ② 그래도 상한의 90% 보다 많으면 가장 오래 안 쓴 키부터 축출한다(RG-12·13 — 계속 요청하는 키는 새 키를 대량으로 만들어도 축출되지 않는다).
• 등록 순서 — 429 에도 X-Trace-Id 에코·협상된 언어를 원하면 install_trace_middleware·install_locale_middleware 를 이 게이트보다 나중에(바깥에) 등록한다.

알려진 제약

• 미처리 예외의 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 이후에 호출(재구성마다 다시 호출). JsonFormatter 와 함께 쓰면 message 만 리댁션되고 extra·error.* 는 대상이 아니다.
  • log_dir=None(콘솔 전용)에서는 app_name·per_pid_file 이 쓰이지 않는다. 파일 ↔ 콘솔 전용 전환도 재호출로 된다 (이전 파일 핸들러는 close 후 제거).
• 인메모리 상태(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 모드). 리프레시 토큰 회전은 비범위.
• WebSocket 인증은 origin=OriginSettings(...) 와 묶어서만 켜진다(기본은 WebSocket 을 손대지 않음 — 주체 None). 인증은 핸드셰이크 시점 1회라 열린 연결 중의 로그아웃·만료·폐기는 연결을 끊지 않는다(앱이 주기적으로 재확인).
• 인증 저장소 장애(세션 조회·사용자별 폐기 표식 조회·토큰 폐기 확인 예외)는 401 이 아니라 503(code "503")이다 — 클라이언트는 503 을 "로그인 필요"로 해석하지 말 것.
• 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 파리티).
• datetimes·make_wire_datetime 의 IANA 시간대 이름은 시스템 시간대 DB 가 필요하다 — Windows·DB 없는 컨테이너는 소비 앱에 pip install tzdata. parse_wire_datetime 의 기본 drop 은 Z·오프셋을 변환 없이 버린다(현행 스큐 — inbound_offset="convert"/"reject" 로 방어). 헬퍼 직렬화는 초 단위(마이크로초 절단)다.

버전·릴리스

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