Building a Home Secretary #8 - Server: Commands and the Approval Queue

1
"""Approval queue.
2

3
Actions a module marks with "approval": true in its manifest aren't executed right away but go up here.
4
The queue doesn't know module content. It keeps the action ("mail.send") and payload (module, action, input)
5
and, when approved, passes them to the module as is. Who can approve is decided at request time.
6

7
  same_user : Only the requester's "other" devices can approve (admins, adult family members)
8
  admin     : Only admin devices can approve (children's requests)
9

10
Either way, the requesting device itself can't approve, and other family members' requests aren't even visible.
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
    """A rule violation during approval handling. code is used to map to an HTTP status, message is shown to the user."""
26

27
    def __init__(self, code: str, message: str):
28
        """Holds the code (e.g. not_found, same_device) together with the message to show the user."""
29
        super().__init__(message)
30
        self.code = code
31
        self.message = message
32

33

34
@dataclass
35
class Approval:
36
    """One request in the approval queue (one row of the DB approvals table)."""
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
        """A dict for API responses. Leaves out the payload (email body, etc.) and includes only the summary."""
54
        # Queue listings expose only the summary instead of the full body. That's enough for a watch or glasses screen.
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
    """Turn one row from the DB into an Approval object. payload is a JSON string, so unpack it back into a 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
    """Decide the approval rule from the requester's role. Children need admin approval (admin); others need their own other device (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
    """Whether this person can approve/reject this request.
88

89
    admin rule    : when they're an admin and not the requester
90
    same_user rule: when they're the requester (whether it's the same device is blocked separately in 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
    """Approval queue. Puts up requests (create), finds them (get, list_for), decides them (claim), and records results (set_result)."""
99

100
    def __init__(self, db: Database, ttl_seconds: int):
101
        """Takes the DB and how long a request stays valid (seconds). If nobody decides within this time, it expires."""
102
        self.db = db
103
        self.ttl = ttl_seconds
104

105
    def _expire(self, c, now: int) -> None:
106
        """Change pending requests past their validity to expired.
107

108
        There's no separate timer; this is called first to clean up every time the queue is read or written.
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
        """Put a request on the queue in pending status and return the saved Approval.
114

115
        The ID is a hard-to-guess random string. The approval rule (rule_for) is fixed at this point and never changes later.
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
        """Find one request by ID. None if not found. Handles expiry first before reading."""
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
        """Return only the requests this person can decide. Other family members' requests aren't visible."""
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
        """Change pending → approved/rejected exactly once. Even if approved twice at the same time, it runs once."""
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
            # Someone else's request gets the exact same answer as a "nonexistent request." We don't even reveal it exists.
161
            if r is None or not can_decide(_row(r), user_id, role):
162
                raise ApprovalError("not_found", "Request not found.")
163
            if r["requested_by"] == device_key:
164
                raise ApprovalError("same_device", "You can't approve from the device that made the request. Please approve from another device.")
165
            if r["status"] != PENDING:
166
                raise ApprovalError("not_pending", "This request has already been handled or has expired.")
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
        """Whether this person has made even one request (regardless of status)."""
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
        """Record the final status (approved/failed) and result text on a request whose execution has finished."""
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
"""A single entry point for calling every module: POST /v1/commands, plus the list of usable modules, GET /v1/modules.
2

3
This file doesn't change as modules grow. Clients just send {"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
    """Command request body. The inner shape of params is checked by the module, not the hub."""
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
    """Handle one command.
28

29
    1. Permission: whether the device and role allow the "<module>:<action>" scope (and the manifest's roles)
30
    2. Validation: have the module check the input and get a summary
31
    3. If the action needs approval, put it on the queue and return 202; otherwise execute right away and return 200
32
    """
33
    state = request.app.state
34
    user, key = caller.user, caller.device.access_key
35
    scope = f"{req.module}:{req.action}"
36
    # Check permission first. Devices without permission aren't even told whether such a module exists.
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 = "an admin device" if approval.rule == RULE_ADMIN else "another one of your devices"
56
            return JSONResponse({"approval": approval.public(), "message": f"It will run once approved from {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
    """The modules and actions usable with this device. Used by the app or glasses to draw menus.
70

71
    Modules that are down are shown as available: false. Actions without permission are left out of the list.
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
"""Approval queue API: list, approve (and execute), reject."""
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
    """Turn an ApprovalError into the matching HTTP error and raise it (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
    """Only requests I can decide are shown. Other family members' requests aren't visible (children's requests are visible to admins)."""
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
    """The common part of approve/reject. Changes the status on the queue exactly once (claim) and writes the audit log.
39

40
    If it breaks the rules, record the reason in the log and end with an HTTP error.
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 : Approve the request and immediately hand execution to the module.
59

60
    If execution fails, the status is recorded as failed. If the module says "the input is wrong" (422), return that message;
61
    for other failures, return 502. Detailed causes go only to the server log.
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, "Execution failed.")
70
    except Exception:
71
        logger.exception("approval %s execution failed", approval.id)
72
        code, message = 502, "Execution failed."
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 : Reject the request. It isn't executed."""
88
    approval = await _decide(approval_id, request, caller, REJECTED)
89
    return {"approval": approval.public()}
90

91

92
async def _execute(state, approval) -> str:
93
    """Pass an approved request to the module for execution and return the result text.
94

95
    The hub doesn't know what the params in the payload mean. It passes them to the module exactly as received at request time.
96
    The requester may have been disabled in the meantime, so check once more right before executing.
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="Leaving work"))

try:
    MailRequest(to=["this-is-not-an-email-address"], subject="")
except ValidationError as e:
    for err in e.errors():
        print(err["loc"], err["msg"])