우리집 비서 만들기 프로젝트 #9 - 서버편: 이벤트와 가짜 모듈 테스트

1
"""모듈이 허브에 알려 오는 이벤트 (카카오 알림이 왔다, 요약이 끝났다 ...).
2

3
허브는 이벤트를 짧게 보관만 한다. 클라이언트가 GET /v1/events 로 가져간다.
4
제목만 남기고 본문 같은 민감한 내용은 받지 않는다.
5
"""
6
import time
7
from dataclasses import asdict, dataclass
8

9
from .db import Database
10

11
KEEP_SECONDS = 7 * 24 * 3600
12

13

14
@dataclass
15
class Event:
16
    """이벤트 하나. user가 None이면 가족 모두에게 보인다."""
17

18
    id: int
19
    module: str
20
    type: str
21
    user: str | None
22
    title: str
23
    created_at: int
24

25
    def public(self) -> dict:
26
        """API 응답용 사전(dict)."""
27
        return asdict(self)
28

29

30
class EventStore:
31
    """이벤트 보관함. 넣을 때마다 7일 지난 것을 지운다."""
32

33
    def __init__(self, db: Database):
34
        """DB를 받는다. 테이블은 db.py의 SCHEMA가 만든다."""
35
        self.db = db
36

37
    def add(self, module: str, type_: str, user: str | None, title: str) -> int:
38
        """이벤트를 저장하고 번호를 돌려준다."""
39
        now = int(time.time())
40
        with self.db.tx() as c:
41
            c.execute("DELETE FROM events WHERE created_at < ?", (now - KEEP_SECONDS,))
42
            cur = c.execute(
43
                "INSERT INTO events (module, type, user, title, created_at) VALUES (?, ?, ?, ?, ?)",
44
                (module, type_, user, title, now),
45
            )
46
            return cur.lastrowid
47

48
    def list_for(self, user_id: str, after: int = 0, limit: int = 50) -> list[Event]:
49
        """이 사람에게 보이는 이벤트(본인 것 + 모두에게 보이는 것)를 after 번호 다음부터 돌려준다.
50

51
        클라이언트는 마지막으로 받은 id를 기억했다가 after로 넘기면 새 이벤트만 받는다.
52
        """
53
        with self.db.tx() as c:
54
            rows = c.execute(
55
                "SELECT * FROM events WHERE (user = ? OR user IS NULL) AND id > ? ORDER BY id LIMIT ?",
56
                (user_id, after, limit),
57
            ).fetchall()
58
        return [Event(**dict(r)) for r in rows]

1
"""이벤트: 모듈 → 허브 (POST /internal/events), 허브 → 클라이언트 (GET /v1/events)."""
2
import hmac
3

4
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
5
from pydantic import BaseModel, Field
6

7
from ..auth.guard import Caller, require
8
from ..auth.scopes import SYSTEM_READ
9

10
router = APIRouter(prefix="/v1", tags=["events"])
11
internal = APIRouter(prefix="/internal", include_in_schema=False)
12

13
NOT_FOUND = HTTPException(status.HTTP_404_NOT_FOUND, detail="Not Found")
14

15

16
class EventIn(BaseModel):
17
    """모듈이 보내는 이벤트. 제목만 받는다(200자까지)."""
18

19
    type: str = Field(pattern=r"^[a-z0-9][a-z0-9._-]{0,39}$")
20
    user: str | None = Field(default=None, pattern=r"^[a-z0-9][a-z0-9-]{1,23}$")
21
    title: str = Field(min_length=1, max_length=200)
22

23

24
def module_from_token(request: Request) -> str:
25
    """내부 토큰으로 어느 모듈이 보낸 요청인지 알아낸다.
26

27
    Cloudflare를 거쳐 온 요청(Cf- 헤더가 붙어 있음)은 바깥에서 온 것이라 없는 경로처럼 404.
28
    토큰이 어느 모듈과도 맞지 않으면 401. 토큰 비교는 상수 시간 비교(compare_digest).
29
    """
30
    if request.headers.get("Cf-Connecting-Ip") or request.headers.get("Cf-Access-Jwt-Assertion"):
31
        raise NOT_FOUND
32
    auth = request.headers.get("Authorization", "")
33
    token = auth[len("Bearer "):] if auth.startswith("Bearer ") else ""
34
    for name, (_, expected) in request.app.state.catalog.endpoints.items():
35
        if token and hmac.compare_digest(expected, token):
36
            return name
37
    request.app.state.audit.write("internal.auth_fail", ip=request.client.host if request.client else None)
38
    raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="인증에 실패했습니다.")
39

40

41
@internal.post("/events", status_code=status.HTTP_201_CREATED)
42
async def receive_event(event: EventIn, request: Request, module: str = Depends(module_from_token)):
43
    """모듈이 보낸 이벤트를 저장한다. 감사 로그에는 제목을 남기지 않는다(개인 메시지일 수 있다)."""
44
    state = request.app.state
45
    if event.user is not None and state.registry.user(event.user) is None:
46
        raise HTTPException(422, detail="없는 사용자입니다.")
47
    event_id = state.events.add(module, event.type, event.user, event.title)
48
    state.audit.write("event.receive", user=event.user, module=module, type=event.type, event_id=event_id)
49
    return {"id": event_id}
50

51

52
@router.get("/events")
53
async def list_events(request: Request, after: int = Query(0, ge=0),
54
                      caller: Caller = Depends(require(SYSTEM_READ))):
55
    """내게 온 이벤트와 가족 모두에게 온 이벤트. after에 마지막으로 받은 id를 넣으면 새것만 온다."""
56
    items = request.app.state.events.list_for(caller.user.id, after)
57
    return {"items": [e.public() for e in items]}

1
"""테스트용 가짜 모듈. 허브는 모듈 코드를 모르니, 계약(docs/module-contract.md)대로만 답하면 된다.
2

3
  demo 모듈
4
    send    승인 필요. to 가 @family.example 이 아니면 422
5
    ping    승인 없이 바로 실행
6
    secret  관리자(admin)만
7
  down 모듈: 항상 500 (꺼진 모듈 흉내)
8
"""
9
from fastapi import FastAPI, Request
10
from fastapi.responses import JSONResponse
11

12
TOKEN = "d" * 40
13

14
MANIFEST = {
15
    "name": "demo",
16
    "title": "데모",
17
    "actions": {
18
        "send": {"title": "보내기", "approval": True},
19
        "ping": {"title": "핑"},
20
        "secret": {"title": "비밀", "roles": ["admin"]},
21
    },
22
}
23

24

25
def build():
26
    """가짜 모듈 앱과, 실행된 요청을 모아 두는 리스트를 돌려준다."""
27
    app = FastAPI()
28
    executed: list[dict] = []
29

30
    def authorized(request: Request) -> bool:
31
        return request.headers.get("Authorization") == f"Bearer {TOKEN}"
32

33
    @app.get("/demo/manifest")
34
    async def manifest(request: Request):
35
        if not authorized(request):
36
            return JSONResponse({"message": "unauthorized"}, 401)
37
        return MANIFEST
38

39
    @app.post("/demo/actions/{action}/validate")
40
    async def validate(action: str, request: Request):
41
        if not authorized(request):
42
            return JSONResponse({"message": "unauthorized"}, 401)
43
        body = await request.json()
44
        to = body["params"].get("to", "")
45
        if action == "send" and not to.endswith("@family.example"):
46
            return JSONResponse({"message": "허용되지 않은 수신자가 있습니다."}, 422)
47
        return {"summary": f"{action} → {to}"}
48

49
    @app.post("/demo/actions/{action}")
50
    async def execute(action: str, request: Request):
51
        if not authorized(request):
52
            return JSONResponse({"message": "unauthorized"}, 401)
53
        body = await request.json()
54
        if body["params"].get("boom"):
55
            return JSONResponse({"message": "boom"}, 500)
56
        executed.append({"action": action, **body})
57
        return {"result": f"{action} 완료"}
58

59
    @app.api_route("/down/{path:path}", methods=["GET", "POST"])
60
    async def down(path: str):
61
        return JSONResponse({"message": "down"}, 500)
62

63
    return app, executed