Skip to main content

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 설치되는 것 이 extra 가 있어야 임포트되는 모듈
(없음) — 아래 표에서 * 표시가 없는 모든 모듈
fastapi fastapi>=0.110, pydantic>=2 (v2 전용 API 사용), httpx>=0.27 response, handlers, health_fastapi
crypto cryptography>=42 (네이티브 휠) 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 등).

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

모듈 주요 API
trace X-Trace-Id 규약 — install_trace_middleware(app)(수신 값 [A-Za-z0-9._-]{1,64} 이면 재사용, 아니면 8자리 hex 생성 → contextvar 바인딩 → 응답 에코), get_trace_id/set_trace_id/reset_trace_id/new_trace_id, httpx 발신 훅 trace_request_hook/async_trace_request_hook
user_context 내부 X-User-* 헤더 → UserContext contextvar — install_user_context_middleware(app), get_user_context(), httpx 훅 user_context_request_hook
auth 인증 모드 bearer/jwt-cookie — install_jwt_auth(app, secret=None, *, keys=None, mode="bearer", cookie=DEFAULT_TOKEN_COOKIE, accept_header=True, skip_paths=(), bind_user_context=True, allow_weak_secret=False, csrf=CsrfSettings(), revocation_checker=None)(순수 ASGI, 거절 없음), get_principal(), require_principal(*roles)(Depends — 401/403), AuthPrincipal, resolve_token, token_cookie(token, *, max_age)/expired_token_cookie(), TokenRevocationChecker — 아래 "인증 모드"
session 인증 모드 session(서버 세션) — install_session_auth(app, store=None, *, cookie=DEFAULT_SESSION_COOKIE, idle_timeout_seconds=1800.0, absolute_timeout_seconds=None, csrf=CsrfSettings(), bind_user_context=True, max_sessions=10000)(순수 ASGI + CSRF 자동), get_session() → Session(.id/.data/.principal, login(subject, role=None)/logout()/regenerate()), SessionStore/AsyncSessionStore Protocol, InMemorySessionStore, store_key(session_id) — 아래 "세션 모드"
csrf CSRF double-submit — install_csrf(app, *, mode, settings=CsrfSettings()), CsrfSettings(cookie, header_name, exempt_paths), csrf_required(mode, method, source, path), request_csrf_rotation(request)
cookies CookieSettings(name, path="/", domain=None, secure=True, same_site="Lax", http_only=True), format_set_cookie(settings, value, *, max_age)/format_delete_cookie(settings), parse_cookie_header/get_cookie(첫 값 우선)
errors ResultCode enum(.code 문자열 / .http_status / .message), 비즈니스 예외 CommonException(result_code, custom_message=None)
response * CommonResponse 봉투(success/code/message/data) — ok(data)/fail(result_code, message=None, data=None), to_wire()(data 없으면 키 생략), WireDateTime, BulkResultModel
handlers * install_exception_handlers(app, *, validation_to_400=True) — CommonException·검증 오류(기본 400)·HTTPException·catch-all 500 을 실패 봉투로 변환
validation flatten_loc(loc) — pydantic loc → 필드 경로(items[0].name)
access_log install_access_log(app, *, headers=(), include_query_string=False, slow_threshold_ms=None) — 요청당 1라인, 민감 헤더 마스킹
security_headers install_security_headers(app, *, hsts=False) — 보안 응답 헤더 프리셋(앱이 실은 헤더 우선)
rate_limit install_rate_limit(app, *, capacity=20, refill_per_second=10.0) — IP 키별 429 게이트, TokenBucket
client_ip resolve_client_ip/resolve_from_scope(scope, trusted_hops=…/trusted_proxies=…) — 안전 기본값 0홉(전달 헤더 무시)
idempotency install_idempotency(app, store=None, …) — Idempotency-Key 중복 시 409, IdempotencyStore Protocol / InMemoryIdempotencyStore
bulk 부분 실패 봉투 — BulkResultBuilder, BulkResult/BulkItem, to_wire/from_wire
query parse_sort(허용 목록 밖 → 400) / parse_page(관용 클램프)
upload validate_upload(filename, size, head, *, max_size_bytes, allowed_extensions) — 크기→확장자→매직바이트 3중 검증, sniff
health / health_fastapi * HealthRegistry, CheckResult / create_health_router(registry) — liveness·readiness(DOWN → 503)
sse SSE 프레임 빌더 — frame_delta/frame_sources/frame_error/frame_done 등
datetimes 오프셋 없는 와이어 datetime — to_wire_datetime/parse_wire_datetime(Z·오프셋은 변환 없이 버림)/strip_zone
feature_flags FeatureFlags(source) + is_enabled(key) — 참 집합 {"true","1","on","yes"}, RSCC_FLAG_* 환경변수 폴백
metrics 표준 메트릭·라벨 이름 문자열 상수 (계측 바인딩 없음)

회복탄력성

모듈 주요 API
retry retry_call(fn, *, retries=3, should_retry=…, override_delay=…, max_elapsed=…) / retry_call_async — 지수 백오프 + full jitter, parse_retry_after, is_retryable_status
circuit_breaker CircuitBreaker(failure_threshold=, open_duration=) — call(fn, *args, **kwargs) / await call_async(fn, …) 권장(게이트 + 보고 1회 보장, 차단 시 CircuitOpenError), 수동 allow_request/on_success/on_failure/on_ignore, CircuitState
bulkhead Bulkhead(max_concurrent=, max_queue=0)(with) / AsyncBulkhead(async with) — 만석 즉시 BulkheadFullError
cache TtlCache(ttl=, max_entries=) / AsyncTtlCache — get(key, loader) single-flight(동시 호출은 로더 1회 공유, 실패·None 비캐시)
rate_limit TokenBucket(capacity=, refill_per_second=) — try_acquire(n=1) 부족 시 대기 없이 False

로깅·보안

모듈 주요 API
logging configure_logging(log_dir, level, app_name, *, include_trace_id=True, clear_root_handlers=False, per_pid_file=False, quiet_uvicorn=False, noisy_loggers=()) — 콘솔 + 자정 롤링 파일(gzip, 30일), 로그 라인에 traceId. TraceIdFilter — 자체 로깅 구성용
log_redact install_secret_redaction() / redact_secrets(s) — Bearer/JWT/API 키/주민번호/카드번호 → [REDACTED:<type>]
log_sanitize sanitize_log_value(value, max_length=…) — 제어 문자 치환 + 절단(로그 인젝션 방지)
jwt HS256 — generate_token(user_id, secret, role=None, *, kid=None), decode_token(token, secret=None, *, keys=None)(kid 로테이션), extract_subject/extract_role, JwtError
crypto * encrypt(plain, key) / decrypt(cipher, key) — AES-256/GCM, Base64(IV‖암호문‖태그), 실패 시 CryptoError
webhook sign(secret, payload) → t=…,v1=… / verify(secrets, header, payload) — HMAC-SHA256, 다중 시크릿 로테이션, 스큐 300초
api_key issue("live")(또는 "test"), is_well_formed, checksum_ok, sha256_hex — rscc_live_… 48자 키
constant_time compare_digest — 시크릿·해시 상수시간 비교
masking mask_name/mask_phone/mask_email/mask_card_number/mask_secret

한국 도메인 유틸

모듈 주요 API
rrn is_valid_rrn(형식·성별코드·생년월일, 체크섬 미포함), is_foreigner_rrn, birth_date_of, checksum_ok_legacy
bizno is_valid_business_number(사업자 10자리) / is_valid_corporate_number(법인 13자리)
phone normalize_phone, classify_phone(mobile/landline/…), format_phone, to_e164
age age_man(만 나이) / age_by_year(연 나이) / age_insurance(보험 나이)
money to_korean_words, to_formal_notation, abbreviate_amount(예: 1.2억)
business_days BusinessDays(holidays=…) — 영업일 가감·판정(공휴일 주입형, 내장 테이블 없음)
text.chosung / text.jamo / text.josa 초성 변환 of·is_chosung_query / 자모 분해·결합·혼합 질의 matches / 조사 선택 pick·attach

빠른 시작

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"))

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

모드 토큰 운반 CSRF 적합한 경우
bearer (기본) Authorization: Bearer <JWT> 헤더만 (쿠키 무시) 없음 서버 간 호출·모바일·MSA
jwt-cookie HttpOnly 쿠키 RSCC_AT (+ 헤더 병행 — 헤더 우선, 무효 헤더는 쿠키로 폴백하지 않음) 쿠키로 인증된 비안전 요청에 double-submit (자동 설치) 브라우저 SPA — 토큰이 JS 에 노출되면 안 될 때
session 서버 세션 — 세션 ID 쿠키 RSCC_SID 만 (아래 "세션 모드") 모든 비안전 메서드 (자동 설치) 즉시 로그아웃·강제 차단·권한 즉시 반영이 필요한 브라우저 서비스
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") 로 단독 설치한다.

배포 토폴로지

구성 필요한 설정
같은 출처 (리버스 프록시로 프론트·API 를 한 오리진에) — 권장 없음
같은 사이트 서브도메인 (app.example.com → api.example.com) csrf=CsrfSettings(cookie=CookieSettings("XSRF-TOKEN", http_only=False, domain="example.com")) (프론트 JS 가 읽도록), JS credentials: "include" + csrf.allowedOrigins, API CORS 가 자격(allow_credentials=True, 명시 오리진)과 X-XSRF-TOKEN 헤더 허용
교차 사이트 (myapp.com → myapi.net) — 비권장 SameSite=None; Secure 필수(CookieSettings(..., same_site="None")) — CSRF 방어가 전적으로 토큰에 의존

로컬 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)는 정상 에코된다.
  • CommonResponse 모델을 FastAPI 핸들러에서 직접 반환하지 말고 .to_wire() 를 반환할 것 — 직접 반환하면 data: null 이 실려 "데이터 없으면 키 생략" 계약이 깨진다.
  • 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 에 있다.

Metadata

Release files for rscc-common 0.5.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for rscc-common 0.5.0
File Size Uploaded Provenance
rscc_common-0.5.0.tar.gz 223.1 kB With provenance Details

Built distribution (wheel)

Table of built distributions (wheels) for rscc-common 0.5.0
File Interpreter ABI Platform
rscc_common-0.5.0-py3-none-any.whl Python 3 none any Details

Total release size: 383.5 kB

Release files / rscc_common-0.5.0.tar.gz

Download URL rscc_common-0.5.0.tar.gz
Size 223.1 kB
Tags Source
SHA-256 checksum
How to use checksums
d2b7e21e9e38a01e9b1270034713e1089a87415f68011435fecae0c20add9aa1
BLAKE2b-256 checksum
How to use checksums
71e6691c066c2b89b12ea0a91ca2beaefeb59d3551d686b69fa438746cc5e663
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
Yes
Uploaded via twine/7.0.0 CPython/3.13.14

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Sep 27, 2026.

Transparency log

Release files / rscc_common-0.5.0-py3-none-any.whl

Download URL rscc_common-0.5.0-py3-none-any.whl
Size 160.4 kB
Tags Python 3
SHA-256 checksum
How to use checksums
867d7977feb7bee1f48ed408cda1a8d5aeafc1bee2ee4d79a612001bf05168da
BLAKE2b-256 checksum
How to use checksums
2a374f70263b3d7b2ed6a7b1f9b3c97a57b778c8f5b62ee49b42eab6192fbad7
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
Yes
Uploaded via twine/7.0.0 CPython/3.13.14

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Sep 27, 2026.

Transparency log

Security for rscc-common 0.5.0

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

Full provenance

All 2 files in this release have been uploaded with provenance metadata.

Select from 2 files

Attestation for rscc_common-0.5.0.tar.gz

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Sep 27, 2026.

Transparency log

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.

To learn more about protecting yourself from malicious actors, visit our package security documentation.

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page