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"))
인증 모드 — bearer / jwt-cookie
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"로 취급된다 (JavaROLE_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의apiClientcredentials·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/AsyncSessionStoreProtocol 로 구현해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)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| rscc_common-0.5.0.tar.gz | 223.1 kB | With provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| 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 logRelease 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 logSecurity 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.
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 logA 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.