우리집 비서 만들기 프로젝트 #5 - 서버편: 서명과 문지기

"""HMAC 요청 서명. 서버와 클라이언트 예제가 같은 함수를 쓴다.

서명할 문자열 = METHOD \\n PATH[?QUERY] \\n TIMESTAMP \\n NONCE \\n SHA256(body) 16진수
서명 = base64url(HMAC-SHA256(secret, 서명할 문자열))
"""
import base64
import hashlib
import hmac
import secrets
import time

HEADER_KEY = "X-Access-Key"
HEADER_TIMESTAMP = "X-Timestamp"
HEADER_NONCE = "X-Nonce"
HEADER_SIGNATURE = "X-Signature"


def canonical_string(method: str, path: str, query: str, timestamp: str, nonce: str, body: bytes) -> str:
    """서명할 문자열을 만든다. 서버와 클라이언트가 글자 하나까지 똑같이 만들어야 서명이 맞는다.

    예) GET /v1/me, 본문 없음 →
        GET
        /v1/me
        1762300000
        3f9c...(nonce)
        e3b0c442...(빈 본문의 SHA-256)
    본문은 통째로 넣지 않고 해시만 넣는다. 본문이 한 글자라도 바뀌면 해시가 달라진다.
    """
    target = f"{path}?{query}" if query else path
    body_hash = hashlib.sha256(body).hexdigest()
    return "\n".join([method.upper(), target, timestamp, nonce, body_hash])


def sign(secret: str, canonical: str) -> str:
    """서명할 문자열을 Secret으로 HMAC-SHA256 서명해 base64url 문자열로 돌려준다.

    base64url은 +, / 대신 -, _ 를 써서 HTTP 헤더에 그대로 넣을 수 있고, 끝의 = 채움 글자는 뗀다.
    """
    digest = hmac.new(secret.encode(), canonical.encode(), hashlib.sha256).digest()
    return base64.urlsafe_b64encode(digest).decode().rstrip("=")


def build_headers(access_key: str, secret: str, method: str, path: str, query: str = "", body: bytes = b"") -> dict:
    """요청 하나에 붙일 인증 헤더 4개를 만든다 (테스트와 예제 클라이언트용).

    타임스탬프는 지금 시각(초), Nonce는 매번 새 무작위 값이라 같은 요청도 매번 서명이 달라진다.
    """
    timestamp = str(int(time.time()))
    nonce = secrets.token_hex(16)
    canonical = canonical_string(method, path, query, timestamp, nonce, body)
    return {
        HEADER_KEY: access_key,
        HEADER_TIMESTAMP: timestamp,
        HEADER_NONCE: nonce,
        HEADER_SIGNATURE: sign(secret, canonical),
    }

"""모든 /v1 요청 앞에서 도는 인증 의존성.

확인 순서:
  1. Cloudflare Access를 거쳐 왔는지 (cloudflared가 붙이는 JWT 헤더)
  2. 헤더 형식, 본문 크기
  3. 키 잠금 여부 (실패가 몰린 키는 잠시 차단)
  4. 기기 존재 여부와 폐기 여부, 기기 주인(사용자)이 활성 상태인지
  5. 타임스탬프 허용 범위
  6. 서명 일치 (상수 시간 비교)
  7. Nonce 재사용 여부
  8. scope (기기 scope AND 사용자 역할이 허용하는 scope)

실패 원인은 감사 로그에만 남기고, 응답은 항상 같은 401 한 가지로 돌려준다.
어디서 막혔는지 알려 주면 공격자에게 힌트가 되기 때문이다.
"""
import hmac
import re
import time
from dataclasses import dataclass

from fastapi import HTTPException, Request, status

from .registry import Device, User
from .scopes import allows, effective_scopes
from .signing import (
    HEADER_KEY,
    HEADER_NONCE,
    HEADER_SIGNATURE,
    HEADER_TIMESTAMP,
    canonical_string,
    sign,
)

_KEY_RE = re.compile(r"^[A-Za-z0-9_-]{3,64}$")
_NONCE_RE = re.compile(r"^[A-Za-z0-9_-]{16,64}$")
_TS_RE = re.compile(r"^\d{9,11}$")

UNAUTHORIZED = HTTPException(status.HTTP_401_UNAUTHORIZED, detail="인증에 실패했습니다.")
FORBIDDEN = HTTPException(status.HTTP_403_FORBIDDEN, detail="권한이 없습니다.")


@dataclass
class Caller:
    """인증을 통과한 호출자. 라우터 함수는 이걸 받아 누가(user), 어떤 기기로(device), 무엇을 할 수 있는지(scopes) 안다."""

    user: User
    device: Device
    scopes: set[str]
    ip: str | None

    def can(self, scope: str) -> bool:
        """이 호출자가 scope 하나를 쓸 수 있는지. 와일드카드(mail:*, modules:*)도 따진다."""
        return allows(self.scopes, scope)

    def forbid(self, request: Request, scope: str) -> HTTPException:
        """권한 없음을 감사 로그에 남기고, 돌려줄 403 예외를 만든다. 부르는 쪽에서 raise 한다."""
        request.app.state.audit.write(
            "auth.forbidden", user=self.user.id, device=self.device.access_key, ip=self.ip,
            scope=scope, path=request.url.path,
        )
        return FORBIDDEN


def _client_ip(request: Request) -> str | None:
    """요청한 쪽의 실제 IP를 찾는다. 감사 로그에 남기는 용도."""
    # Cloudflare를 거친 요청은 이 헤더에 실제 클라이언트 IP가 있다.
    return request.headers.get("Cf-Connecting-Ip") or (request.client.host if request.client else None)


async def authenticate(request: Request) -> Caller:
    """요청 헤더의 서명을 검사해 호출자(Caller)를 돌려준다. 하나라도 틀리면 401.

    검사 순서는 이 파일 맨 위 설명 참고. 비용이 싼 검사(헤더 형식)를 먼저,
    비싼 검사(DB 조회, 서명 계산)를 나중에 한다.
    """
    state = request.app.state
    settings = state.settings
    audit = state.audit
    ip = _client_ip(request)

    def fail(reason: str, key: str | None = None) -> HTTPException:
        """인증 실패를 감사 로그와 실패 기록 테이블에 남기고, 돌려줄 401 예외를 만든다.

        부르는 쪽에서 raise fail("이유") 로 쓴다. 이유는 로그에만 남고 응답에는 나가지 않는다.
        키 형식이 정상일 때만 실패 기록을 쌓는다(이상한 값으로 테이블을 채우지 못하게).
        """
        audit.write("auth.fail", device=key, ip=ip, reason=reason, path=request.url.path)
        if key and _KEY_RE.match(key):
            with state.db.tx() as c:
                c.execute("INSERT INTO auth_failures (access_key, at) VALUES (?, ?)", (key, int(time.time())))
        return UNAUTHORIZED

    if settings.require_cf_access and not request.headers.get("Cf-Access-Jwt-Assertion"):
        raise fail("no_cf_access")

    key = request.headers.get(HEADER_KEY, "")
    ts = request.headers.get(HEADER_TIMESTAMP, "")
    nonce = request.headers.get(HEADER_NONCE, "")
    signature = request.headers.get(HEADER_SIGNATURE, "")
    if not (_KEY_RE.match(key) and _TS_RE.match(ts) and _NONCE_RE.match(nonce) and signature):
        raise fail("bad_headers", key or None)

    body = await request.body()
    if len(body) > settings.max_body_bytes:
        raise fail("body_too_large", key)

    now = int(time.time())
    with state.db.tx() as c:
        c.execute("DELETE FROM auth_failures WHERE at < ?", (now - settings.lockout_window_seconds,))
        failures = c.execute(
            "SELECT COUNT(*) FROM auth_failures WHERE access_key = ? AND at > ?",
            (key, now - settings.lockout_window_seconds),
        ).fetchone()[0]
    if failures >= settings.lockout_threshold:
        audit.write("auth.locked", device=key, ip=ip, path=request.url.path)
        raise UNAUTHORIZED

    device = state.registry.device(key)
    if device is None:
        raise fail("unknown_key", key)
    if device.revoked:
        raise fail("revoked_key", key)
    user = state.registry.user(device.user)
    if user is None or user.disabled:
        raise fail("user_disabled", key)

    if abs(now - int(ts)) > settings.clock_skew_seconds:
        raise fail("timestamp_skew", key)

    canonical = canonical_string(request.method, request.url.path, request.url.query, ts, nonce, body)
    expected = sign(device.secret, canonical)
    if not hmac.compare_digest(expected, signature):
        raise fail("bad_signature", key)

    # Nonce는 타임스탬프 허용 범위의 두 배 동안 기억한다. 그 뒤에는 타임스탬프 검사에서 걸린다.
    with state.db.tx() as c:
        c.execute("DELETE FROM nonces WHERE expires_at < ?", (now,))
        exists = c.execute(
            "SELECT 1 FROM nonces WHERE access_key = ? AND nonce = ?", (key, nonce)
        ).fetchone()
        if not exists:
            c.execute(
                "INSERT INTO nonces (access_key, nonce, expires_at) VALUES (?, ?, ?)",
                (key, nonce, now + settings.clock_skew_seconds * 2),
            )
    if exists:
        raise fail("nonce_replay", key)

    return Caller(user=user, device=device, scopes=effective_scopes(device.scopes, user.role), ip=ip)


def require(scope: str):
    """scope 하나를 받아 "이 권한이 있어야 통과"하는 의존성 함수를 만들어 돌려준다.

        @router.get("/me")
        async def me(caller: Caller = Depends(require(SYSTEM_READ))):

    인증(401)을 먼저 하고, scope가 없으면 403. scope마다 검사 함수를 찍어 내는 함수다.
    """
    async def dependency(request: Request) -> Caller:
        """FastAPI가 요청마다 실제로 부르는 함수. 인증 후 scope를 확인한다."""
        caller = await authenticate(request)
        if not caller.can(scope):
            raise caller.forbid(request, scope)
        return caller

    return dependency

1
import asyncio
2
import time
3

4

5
async def work(name: str, seconds: float) -> str:
6
    await asyncio.sleep(seconds)  # 기다리는 동안 다른 일을 하러 간다
7
    return f"{name} 완료"
8

9

10
async def main():
11
    start = time.perf_counter()
12
    results = await asyncio.gather(work("메일", 1), work("노트", 1), work("전등", 1))
13
    print(results, f"{time.perf_counter() - start:.1f}초")
14

15

16
asyncio.run(main())

1
from fastapi import APIRouter, Depends
2

3
from ..auth.guard import Caller, require
4
from ..auth.scopes import SYSTEM_READ
5

6
router = APIRouter(prefix="/v1", tags=["system"])
7

8

9
@router.get("/me")
10
async def me(caller: Caller = Depends(require(SYSTEM_READ))):
11
    """GET /v1/me : 지금 이 키가 누구의 어떤 기기이고, 무엇을 할 수 있는지 돌려준다.
12

13
    새 기기를 설정하고 처음 호출해 보는 용도. 200이 오면 서명과 Cloudflare Access 설정이 모두 맞은 것이다.
14
    """
15
    d = caller.device
16
    return {
17
        "user": caller.user.public(),
18
        "device": {"access_key": d.access_key, "name": d.name},
19
        "scopes": sorted(caller.scopes),
20
    }

1
from fastapi import Depends, FastAPI, Header, HTTPException
2
from fastapi.testclient import TestClient
3

4
app = FastAPI()
5

6

7
def who_are_you(x_name: str = Header("")) -> str:
8
    if not x_name:
9
        raise HTTPException(401, "누구세요?")
10
    return x_name
11

12

13
@app.get("/hello")
14
def hello(name: str = Depends(who_are_you)):
15
    return {"hello": name}
16

17

18
client = TestClient(app)
19
print(client.get("/hello").status_code)
20
print(client.get("/hello", headers={"X-Name": "dad"}).json())

1
import logging
2
import uuid
3
from contextlib import asynccontextmanager
4

5
from fastapi import FastAPI, Request
6
from fastapi.exceptions import RequestValidationError
7
from fastapi.responses import JSONResponse
8
from starlette.exceptions import HTTPException
9

10
from .audit import AuditLog
11
from .auth.registry import Registry
12
from .config import Settings, get_settings
13
from .db import Database
14
from .approvals import ApprovalQueue
15
from .catalog import ModuleCatalog
16
from .events import EventStore
17
from .routers import approvals, commands, events, system
18

19
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
20
logger = logging.getLogger("secretary")
21

22

23
def create_app(settings: Settings | None = None, module_transport=None) -> FastAPI:
24
    """FastAPI 앱을 만든다. 테스트에서는 settings와 가짜 모듈(module_transport)을 넣어 따로 만든다.
25

26
    운영 서버는 맨 아래 app = create_app() 을 uvicorn이 불러 쓴다.
27
    """
28
    settings = settings or get_settings()
29

30
    @asynccontextmanager
31
    async def lifespan(app: FastAPI):
32
        """서버가 켜질 때(yield 전) 필요한 객체를 만들어 app.state에 두고, 꺼질 때(yield 뒤) 정리한다.
33

34
        라우터는 request.app.state.db 처럼 꺼내 쓴다.
35
        """
36
        app.state.settings = settings
37
        app.state.db = Database(settings.db_path)
38
        app.state.registry = Registry(settings.registry_path, settings.max_users)
39
        app.state.audit = AuditLog(settings.audit_path)
40
        app.state.approvals = ApprovalQueue(app.state.db, settings.approval_ttl_seconds)
41
        app.state.events = EventStore(app.state.db)
42
        app.state.catalog = ModuleCatalog(settings.module_endpoints, settings.module_timeout_seconds,
43
                                          transport=module_transport)
44
        app.state.audit.write("server.start", modules=sorted(settings.module_endpoints))
45
        yield
46
        await app.state.catalog.close()
47
        app.state.db.close()
48

49
    # 운영에서는 문서 페이지를 열지 않는다. 스펙이 필요하면 로컬에서 따로 뽑는다.
50
    app = FastAPI(title="Secretary", lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)
51

52
    @app.middleware("http")
53
    async def request_id(request: Request, call_next):
54
        """모든 응답에 X-Request-Id를 붙이는 미들웨어.
55

56
        예상 못 한 오류가 나면 자세한 내용은 서버 로그에만 남기고, 클라이언트에는 같은 ID가 붙은 500만 돌려준다.
57
        사용자가 ID를 알려 주면 로그에서 그 요청을 바로 찾을 수 있다.
58
        """
59
        rid = uuid.uuid4().hex[:12]
60
        try:
61
            response = await call_next(request)
62
        except Exception:
63
            logger.exception("unhandled error rid=%s path=%s", rid, request.url.path)
64
            response = JSONResponse({"code": "INTERNAL_ERROR", "message": "일시적인 오류가 발생했습니다."}, 500)
65
        response.headers["X-Request-Id"] = rid
66
        return response
67

68
    @app.exception_handler(HTTPException)
69
    async def http_error(request: Request, exc: HTTPException):
70
        """HTTP 오류(401, 403, 404 ...)를 {code, message} 모양의 JSON으로 통일한다."""
71
        return JSONResponse({"code": f"HTTP_{exc.status_code}", "message": exc.detail}, exc.status_code)
72

73
    @app.exception_handler(RequestValidationError)
74
    async def validation_error(request: Request, exc: RequestValidationError):
75
        """요청 형식 오류(422)를 {code, message, fields} 모양으로 바꾼다."""
76
        # 어떤 필드가 왜 틀렸는지만 알려 주고, 입력값 자체는 돌려주지 않는다.
77
        fields = [".".join(str(p) for p in e["loc"][1:]) for e in exc.errors()]
78
        return JSONResponse({"code": "INVALID_REQUEST", "message": "요청 형식이 올바르지 않습니다.", "fields": fields}, 422)
79

80
    @app.get("/healthz", include_in_schema=False)
81
    async def healthz():
82
        """GET /healthz : 서버가 살아 있는지. 인증 없이 열려 있고, 컨테이너 상태 확인(healthcheck)이 부른다."""
83
        return {"ok": True}
84

85
    app.include_router(system.router)
86
    app.include_router(commands.router)
87
    app.include_router(events.router)
88
    app.include_router(events.internal)
89
    app.include_router(approvals.router)
90
    return app
91

92

93
app = create_app()

1
from contextlib import asynccontextmanager
2

3
from fastapi import FastAPI
4
from fastapi.testclient import TestClient
5

6

7
@asynccontextmanager
8
async def lifespan(app: FastAPI):
9
    print("서버 시작: DB 연결")
10
    yield
11
    print("서버 종료: DB 닫기")
12

13

14
app = FastAPI(lifespan=lifespan)
15

16

17
@app.get("/")
18
def root():
19
    print("요청 처리")
20
    return {}
21

22

23
with TestClient(app) as client:
24
    client.get("/")