Building a Home Secretary #9 - Server: Events and Fake Module Tests

1
"""Events that modules report to the hub (a Kakao notification came in, a summary finished ...).
2

3
The hub only keeps events briefly. Clients pick them up with GET /v1/events.
4
Only titles are kept; sensitive content like message bodies isn't accepted.
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
    """One event. If user is None, it's visible to the whole family."""
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
        """A dict for API responses."""
27
        return asdict(self)
28

29

30
class EventStore:
31
    """Event store. Every time something is added, events older than 7 days are deleted."""
32

33
    def __init__(self, db: Database):
34
        """Takes the DB. The table is created by SCHEMA in db.py."""
35
        self.db = db
36

37
    def add(self, module: str, type_: str, user: str | None, title: str) -> int:
38
        """Save an event and return its number."""
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
        """Return the events visible to this person (their own + those visible to everyone), starting after the given number.
50

51
        Clients remember the last id they received and pass it as after, so they get only new events.
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
"""Events: module → hub (POST /internal/events), hub → client (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
    """An event sent by a module. Only the title is accepted (up to 200 characters)."""
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
    """Figure out which module sent the request from the internal token.
26

27
    Requests that came through Cloudflare (with Cf- headers attached) came from outside, so 404 as if the path doesn't exist.
28
    401 if the token matches no module. Tokens are compared in constant time (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="Authentication failed.")
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
    """Save an event sent by a module. The title isn't written to the audit log (it may be a personal message)."""
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="No such user.")
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
    """Events for me and events for the whole family. Pass the last id you received as after to get only new ones."""
56
    items = request.app.state.events.list_for(caller.user.id, after)
57
    return {"items": [e.public() for e in items]}

1
"""Fake module for tests. The hub doesn't know module code, so it just has to answer per the contract (docs/module-contract.md).
2

3
  demo module
4
    send    Needs approval. 422 if to isn't @family.example
5
    ping    Runs immediately without approval
6
    secret  Admins only
7
  down module: always 500 (imitates a dead module)
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": "Demo",
17
    "actions": {
18
        "send": {"title": "Send", "approval": True},
19
        "ping": {"title": "Ping"},
20
        "secret": {"title": "Secret", "roles": ["admin"]},
21
    },
22
}
23

24

25
def build():
26
    """Return the fake module app and a list that collects executed requests."""
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": "Some recipients aren't allowed."}, 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} done"}
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