우리집 비서 만들기 프로젝트 #8 - 서버편: 명령과 승인 대기열

1
"""승인 대기열.
2

3
모듈이 manifest에 "approval": true 로 적은 액션은 바로 실행하지 않고 여기 올린다.
4
대기열은 모듈 내용을 모른다. action("mail.send")과 payload(모듈, 액션, 입력값)를 보관했다가
5
승인되면 그대로 모듈에 넘길 뿐이다. 누가 승인할 수 있는지는 요청할 때 정해진다.
6

7
  same_user : 요청한 사람의 "다른" 기기만 승인할 수 있다 (관리자, 어른 가족)
8
  admin     : 관리자의 기기만 승인할 수 있다 (아이의 요청)
9

10
어느 경우든 요청한 기기 자신은 승인할 수 없고, 다른 가족의 요청은 보이지도 않는다.
11
"""
12
import json
13
import secrets
14
import time
15
from dataclasses import dataclass
16

17
from .auth.scopes import ADMIN, CHILD
18
from .db import Database
19

20
PENDING, APPROVED, REJECTED, EXPIRED, FAILED = "pending", "approved", "rejected", "expired", "failed"
21
RULE_SAME_USER, RULE_ADMIN = "same_user", "admin"
22

23

24
class ApprovalError(Exception):
25
    """승인 처리 중 규칙 위반. code는 HTTP 상태로 바꾸는 데 쓰고, message는 사용자에게 보여 준다."""
26

27
    def __init__(self, code: str, message: str):
28
        """code(예: not_found, same_device)와 사용자에게 보여 줄 message를 함께 담는다."""
29
        super().__init__(message)
30
        self.code = code
31
        self.message = message
32

33

34
@dataclass
35
class Approval:
36
    """승인 대기열의 요청 하나 (DB approvals 테이블의 한 줄)."""
37

38
    id: str
39
    action: str
40
    payload: dict
41
    summary: str
42
    requested_user: str
43
    requested_by: str
44
    rule: str
45
    status: str
46
    created_at: int
47
    expires_at: int
48
    decided_by: str | None
49
    decided_at: int | None
50
    result: str | None
51

52
    def public(self) -> dict:
53
        """API 응답용 사전(dict). payload(메일 본문 등)는 빼고 요약만 담는다."""
54
        # 대기열 조회에는 본문 전체 대신 요약만 내보낸다. 워치나 안경 화면에는 이걸로 충분하다.
55
        return {
56
            "id": self.id,
57
            "action": self.action,
58
            "summary": self.summary,
59
            "requested_user": self.requested_user,
60
            "requested_by": self.requested_by,
61
            "rule": self.rule,
62
            "status": self.status,
63
            "created_at": self.created_at,
64
            "expires_at": self.expires_at,
65
            "decided_by": self.decided_by,
66
            "decided_at": self.decided_at,
67
            "result": self.result,
68
        }
69

70

71
def _row(r) -> Approval:
72
    """DB에서 꺼낸 한 줄을 Approval 객체로 바꾼다. payload는 JSON 문자열이라 다시 dict로 풀어 준다."""
73
    return Approval(
74
        id=r["id"], action=r["action"], payload=json.loads(r["payload"]), summary=r["summary"],
75
        requested_user=r["requested_user"], requested_by=r["requested_by"], rule=r["rule"], status=r["status"],
76
        created_at=r["created_at"], expires_at=r["expires_at"], decided_by=r["decided_by"],
77
        decided_at=r["decided_at"], result=r["result"],
78
    )
79

80

81
def rule_for(role: str) -> str:
82
    """요청한 사람의 역할로 승인 규칙을 정한다. 아이는 관리자 승인(admin), 나머지는 본인 다른 기기 승인(same_user)."""
83
    return RULE_ADMIN if role == CHILD else RULE_SAME_USER
84

85

86
def can_decide(approval: Approval, user_id: str, role: str) -> bool:
87
    """이 사람이 이 요청을 승인/거절할 수 있는지.
88

89
    admin 규칙    : 관리자이면서 요청한 본인이 아닐 때
90
    same_user 규칙: 요청한 본인일 때 (같은 기기인지는 claim()에서 따로 막는다)
91
    """
92
    if approval.rule == RULE_ADMIN:
93
        return role == ADMIN and user_id != approval.requested_user
94
    return user_id == approval.requested_user
95

96

97
class ApprovalQueue:
98
    """승인 대기열. 요청을 올리고(create), 찾고(get, list_for), 결정하고(claim), 결과를 남긴다(set_result)."""
99

100
    def __init__(self, db: Database, ttl_seconds: int):
101
        """DB와 요청 유효 시간(초)을 받는다. 이 시간이 지나도록 아무도 결정하지 않으면 만료된다."""
102
        self.db = db
103
        self.ttl = ttl_seconds
104

105
    def _expire(self, c, now: int) -> None:
106
        """유효 시간이 지난 pending 요청을 expired로 바꾼다.
107

108
        따로 도는 타이머 없이, 대기열을 읽거나 쓸 때마다 먼저 이걸 불러 정리한다.
109
        """
110
        c.execute("UPDATE approvals SET status = ? WHERE status = ? AND expires_at < ?", (EXPIRED, PENDING, now))
111

112
    def create(self, action: str, payload: dict, summary: str, user_id: str, role: str, device_key: str) -> Approval:
113
        """요청을 pending 상태로 대기열에 올리고 저장된 Approval을 돌려준다.
114

115
        ID는 추측하기 어려운 무작위 문자열. 승인 규칙(rule_for)은 이때 정해져 나중에 바뀌지 않는다.
116
        """
117
        now = int(time.time())
118
        approval_id = secrets.token_urlsafe(9)
119
        with self.db.tx() as c:
120
            c.execute(
121
                "INSERT INTO approvals (id, action, payload, summary, requested_user, requested_by, rule, status,"
122
                " created_at, expires_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
123
                (approval_id, action, json.dumps(payload, ensure_ascii=False), summary, user_id, device_key,
124
                 rule_for(role), PENDING, now, now + self.ttl),
125
            )
126
        return self.get(approval_id)
127

128
    def get(self, approval_id: str) -> Approval | None:
129
        """ID로 요청 하나를 찾는다. 없으면 None. 읽기 전에 만료 처리를 먼저 한다."""
130
        with self.db.tx() as c:
131
            self._expire(c, int(time.time()))
132
            r = c.execute("SELECT * FROM approvals WHERE id = ?", (approval_id,)).fetchone()
133
        return _row(r) if r else None
134

135
    def list_for(self, user_id: str, role: str, status: str | None = PENDING, limit: int = 20) -> list[Approval]:
136
        """이 사람이 결정할 수 있는 요청만 돌려준다. 다른 가족의 요청은 보이지 않는다."""
137
        where = ["((rule = ? AND requested_user = ?)"]
138
        args: list = [RULE_SAME_USER, user_id]
139
        if role == ADMIN:
140
            where.append(" OR (rule = ? AND requested_user != ?)")
141
            args += [RULE_ADMIN, user_id]
142
        where.append(")")
143
        sql = "SELECT * FROM approvals WHERE " + "".join(where)
144
        if status:
145
            sql += " AND status = ?"
146
            args.append(status)
147
        sql += " ORDER BY created_at DESC LIMIT ?"
148
        args.append(limit)
149
        with self.db.tx() as c:
150
            self._expire(c, int(time.time()))
151
            rows = c.execute(sql, args).fetchall()
152
        return [_row(r) for r in rows]
153

154
    def claim(self, approval_id: str, user_id: str, role: str, device_key: str, decision: str) -> Approval:
155
        """pending → approved/rejected 로 한 번만 바꾼다. 동시에 두 번 승인돼도 실행은 한 번."""
156
        now = int(time.time())
157
        with self.db.tx() as c:
158
            self._expire(c, now)
159
            r = c.execute("SELECT * FROM approvals WHERE id = ?", (approval_id,)).fetchone()
160
            # 남의 요청은 "없는 요청"과 똑같이 대답한다. 있다는 사실도 알려 주지 않는다.
161
            if r is None or not can_decide(_row(r), user_id, role):
162
                raise ApprovalError("not_found", "요청을 찾을 수 없습니다.")
163
            if r["requested_by"] == device_key:
164
                raise ApprovalError("same_device", "요청한 기기에서는 승인할 수 없습니다. 다른 기기에서 승인해 주세요.")
165
            if r["status"] != PENDING:
166
                raise ApprovalError("not_pending", "이미 처리됐거나 만료된 요청입니다.")
167
            c.execute(
168
                "UPDATE approvals SET status = ?, decided_by = ?, decided_at = ? WHERE id = ?",
169
                (decision, device_key, now, approval_id),
170
            )
171
        return self.get(approval_id)
172

173
    def has_history(self, user_id: str) -> bool:
174
        """이 사람이 요청한 기록이 하나라도 있는지 (상태 무관)."""
175
        with self.db.tx() as c:
176
            return c.execute("SELECT 1 FROM approvals WHERE requested_user = ? LIMIT 1", (user_id,)).fetchone() is not None
177

178
    def set_result(self, approval_id: str, status: str, result: str) -> Approval:
179
        """실행이 끝난 요청에 최종 상태(approved/failed)와 결과 문구를 기록한다."""
180
        with self.db.tx() as c:
181
            c.execute("UPDATE approvals SET status = ?, result = ? WHERE id = ?", (status, result, approval_id))
182
        return self.get(approval_id)

1
"""모든 모듈을 부르는 하나의 입구: POST /v1/commands, 그리고 쓸 수 있는 모듈 목록 GET /v1/modules.
2

3
모듈이 늘어나도 이 파일은 바뀌지 않는다. 클라이언트는 {"module", "action", "params"}만 보내면 된다.
4
"""
5
from fastapi import APIRouter, Depends, HTTPException, Request, status
6
from fastapi.responses import JSONResponse
7
from pydantic import BaseModel, Field
8

9
from ..approvals import RULE_ADMIN
10
from ..auth.guard import Caller, authenticate, require
11
from ..auth.scopes import SYSTEM_READ
12
from ..catalog import ModuleError, caller_context
13

14
router = APIRouter(prefix="/v1", tags=["commands"])
15

16

17
class CommandRequest(BaseModel):
18
    """명령 요청 본문. params 안쪽 모양은 허브가 아니라 모듈이 검사한다."""
19

20
    module: str = Field(pattern=r"^[a-z0-9][a-z0-9-]{1,23}$")
21
    action: str = Field(pattern=r"^[a-z0-9][a-z0-9_-]{0,31}$")
22
    params: dict = Field(default_factory=dict)
23

24

25
@router.post("/commands")
26
async def run_command(req: CommandRequest, request: Request, caller: Caller = Depends(authenticate)):
27
    """명령 하나를 처리한다.
28

29
    1. 권한: 기기와 역할이 "<모듈>:<액션>" scope를 허락하는지 (manifest의 roles도)
30
    2. 검사: 모듈에 입력을 검사받고 요약을 받는다
31
    3. 승인이 필요한 액션이면 대기열에 올리고 202, 아니면 바로 실행하고 200
32
    """
33
    state = request.app.state
34
    user, key = caller.user, caller.device.access_key
35
    scope = f"{req.module}:{req.action}"
36
    # 권한부터 본다. 권한이 없는 기기에는 그런 모듈이 있는지조차 알려 주지 않는다.
37
    if not caller.can(scope):
38
        raise caller.forbid(request, scope)
39
    try:
40
        spec = await state.catalog.action(req.module, req.action)
41
        if spec.roles is not None and user.role not in spec.roles:
42
            raise caller.forbid(request, scope)
43
        ctx = caller_context(user)
44
        summary = await state.catalog.validate(req.module, req.action, req.params, ctx)
45

46
        if spec.approval:
47
            approval = state.approvals.create(
48
                action=f"{req.module}.{req.action}",
49
                payload={"module": req.module, "action": req.action, "params": req.params},
50
                summary=f"{user.name}: {summary}",
51
                user_id=user.id, role=user.role, device_key=key,
52
            )
53
            state.audit.write("approval.create", user=user.id, device=key, ip=caller.ip,
54
                              approval=approval.id, action=approval.action, rule=approval.rule)
55
            who = "관리자 기기에서" if approval.rule == RULE_ADMIN else "다른 내 기기에서"
56
            return JSONResponse({"approval": approval.public(), "message": f"{who} 승인하면 실행됩니다."},
57
                                status.HTTP_202_ACCEPTED)
58

59
        result = await state.catalog.execute(req.module, req.action, req.params, ctx)
60
    except ModuleError as e:
61
        state.audit.write("command.fail", user=user.id, device=key, ip=caller.ip, action=scope, status=e.status)
62
        raise HTTPException(e.status, detail=e.message)
63
    state.audit.write("command.execute", user=user.id, device=key, ip=caller.ip, action=scope)
64
    return {"result": result}
65

66

67
@router.get("/modules")
68
async def list_modules(request: Request, caller: Caller = Depends(require(SYSTEM_READ))):
69
    """이 기기로 쓸 수 있는 모듈과 액션 목록. 앱이나 안경이 메뉴를 그릴 때 쓴다.
70

71
    꺼져 있는 모듈은 available: false 로 보여 준다. 권한이 없는 액션은 목록에서 뺀다.
72
    """
73
    catalog = request.app.state.catalog
74
    items = []
75
    for name in catalog.endpoints:
76
        try:
77
            manifest = await catalog.manifest(name)
78
        except ModuleError:
79
            items.append({"name": name, "available": False, "actions": []})
80
            continue
81
        actions = [
82
            {"name": a, "title": spec.title, "approval": spec.approval, "params": spec.params}
83
            for a, spec in manifest.actions.items()
84
            if caller.can(f"{name}:{a}") and (spec.roles is None or caller.user.role in spec.roles)
85
        ]
86
        if actions:
87
            items.append({"name": name, "title": manifest.title, "available": True, "actions": actions})
88
    return {"items": items}

1
"""승인 대기열 API: 목록, 승인(그리고 실행), 거절."""
2
import logging
3

4
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
5

6
from ..auth.guard import Caller, require
7
from ..auth.scopes import APPROVALS_DECIDE
8
from ..approvals import APPROVED, FAILED, REJECTED, ApprovalError
9
from ..catalog import ModuleError, caller_context
10

11
router = APIRouter(prefix="/v1/approvals", tags=["approvals"])
12
logger = logging.getLogger("secretary.approvals")
13

14
_ERROR_STATUS = {
15
    "not_found": status.HTTP_404_NOT_FOUND,
16
    "not_pending": status.HTTP_409_CONFLICT,
17
    "same_device": status.HTTP_403_FORBIDDEN,
18
}
19

20

21
def _raise(e: ApprovalError):
22
    """ApprovalError를 알맞은 HTTP 오류로 바꿔 던진다 (not_found → 404, not_pending → 409, same_device → 403)."""
23
    raise HTTPException(_ERROR_STATUS.get(e.code, 400), detail=e.message)
24

25

26
@router.get("")
27
async def list_approvals(
28
    request: Request,
29
    status_filter: str | None = Query("pending", alias="status", pattern="^(pending|approved|rejected|expired|failed)$"),
30
    caller: Caller = Depends(require(APPROVALS_DECIDE)),
31
):
32
    """내가 결정할 수 있는 요청만 보인다. 다른 가족의 요청은 보이지 않는다(아이 요청은 관리자에게 보인다)."""
33
    items = request.app.state.approvals.list_for(caller.user.id, caller.user.role, status_filter)
34
    return {"items": [a.public() for a in items]}
35

36

37
async def _decide(approval_id: str, request: Request, caller: Caller, decision: str):
38
    """승인/거절의 공통 부분. 대기열에서 상태를 한 번만 바꾸고(claim) 감사 로그를 남긴다.
39

40
    규칙에 어긋나면 이유를 로그에 남기고 HTTP 오류로 끝낸다.
41
    """
42
    state = request.app.state
43
    user, key = caller.user, caller.device.access_key
44
    try:
45
        approval = state.approvals.claim(approval_id, user.id, user.role, key, decision)
46
    except ApprovalError as e:
47
        state.audit.write("approval.denied", user=user.id, device=key, ip=caller.ip, approval=approval_id,
48
                          reason=e.code)
49
        _raise(e)
50
    event = "approval.approve" if decision == APPROVED else "approval.reject"
51
    state.audit.write(event, user=user.id, device=key, ip=caller.ip, approval=approval.id, action=approval.action,
52
                      requested_user=approval.requested_user)
53
    return approval
54

55

56
@router.post("/{approval_id}/approve")
57
async def approve(approval_id: str, request: Request, caller: Caller = Depends(require(APPROVALS_DECIDE))):
58
    """POST /v1/approvals/{id}/approve : 요청을 승인하고 바로 모듈에 실행을 맡긴다.
59

60
    실행이 실패하면 상태를 failed로 남긴다. 모듈이 "입력이 틀렸다"(422)고 하면 그 메시지를,
61
    그 밖의 실패는 502를 돌려준다. 자세한 원인은 서버 로그에만 남긴다.
62
    """
63
    state = request.app.state
64
    approval = await _decide(approval_id, request, caller, APPROVED)
65
    try:
66
        result = await _execute(state, approval)
67
    except ModuleError as e:
68
        logger.warning("approval %s execution failed: %s %s", approval.id, e.status, e.message)
69
        code, message = (422, e.message) if e.status == 422 else (502, "실행에 실패했습니다.")
70
    except Exception:
71
        logger.exception("approval %s execution failed", approval.id)
72
        code, message = 502, "실행에 실패했습니다."
73
    else:
74
        approval = state.approvals.set_result(approval.id, APPROVED, result)
75
        state.audit.write("approval.executed", user=caller.user.id, device=caller.device.access_key, ip=caller.ip,
76
                          approval=approval.id, action=approval.action)
77
        return {"approval": approval.public()}
78

79
    state.approvals.set_result(approval.id, FAILED, message)
80
    state.audit.write("approval.failed", user=caller.user.id, device=caller.device.access_key, ip=caller.ip,
81
                      approval=approval.id, action=approval.action)
82
    raise HTTPException(code, detail=message)
83

84

85
@router.post("/{approval_id}/reject")
86
async def reject(approval_id: str, request: Request, caller: Caller = Depends(require(APPROVALS_DECIDE))):
87
    """POST /v1/approvals/{id}/reject : 요청을 거절한다. 실행하지 않는다."""
88
    approval = await _decide(approval_id, request, caller, REJECTED)
89
    return {"approval": approval.public()}
90

91

92
async def _execute(state, approval) -> str:
93
    """승인된 요청을 모듈에 넘겨 실행하고 결과 문구를 돌려준다.
94

95
    허브는 payload 안의 params가 무슨 뜻인지 모른다. 요청할 때 받아 둔 그대로 모듈에 넘긴다.
96
    요청한 사람이 그사이 비활성화됐을 수 있으니 실행 직전에 한 번 더 확인한다.
97
    """
98
    requester = state.registry.user(approval.requested_user)
99
    if requester is None or requester.disabled:
100
        raise RuntimeError("requester disabled")
101
    p = approval.payload
102
    return await state.catalog.execute(p["module"], p["action"], p["params"], caller_context(requester))

from pydantic import BaseModel, EmailStr, Field, ValidationError


class MailRequest(BaseModel):
    to: list[EmailStr] = Field(min_length=1, max_length=5)
    subject: str = Field(min_length=1, max_length=200)


print(MailRequest(to=["mom@example.com"], subject="퇴근"))

try:
    MailRequest(to=["이건-메일-주소가-아님"], subject="")
except ValidationError as e:
    for err in e.errors():
        print(err["loc"], err["msg"])