Building a Home Secretary #7 - Server: The Hub–Module Contract

"""Audit log. One JSON object per line (JSONL).

Secrets and email bodies are never recorded. Only who (person, device), when, what, and how it turned out.
"""
import json
import logging
import threading
import time
from pathlib import Path

logger = logging.getLogger("secretary.audit")
_lock = threading.Lock()


class AuditLog:
    """Audit log writer. It only appends lines to the end of the file, never edits or deletes them."""

    def __init__(self, path: Path):
        """Remember the log file path, and create the folder if it doesn't exist."""
        self.path = path
        path.parent.mkdir(parents=True, exist_ok=True)

    def write(self, event: str, *, user: str | None = None, device: str | None = None, ip: str | None = None,
              **fields) -> None:
        """Record one event as one line of JSON. The same line also goes to the server log (docker logs).

        event : What happened. e.g. "auth.fail", "approval.approve"
        user, device, ip : Who, with which device, from where
        **fields : Any other info. e.g. reason="bad_signature". Never include secrets or email bodies.
        """
        record = {"ts": int(time.time()), "event": event, "user": user, "device": device, "ip": ip, **fields}
        line = json.dumps(record, ensure_ascii=False, default=str)
        with _lock:
            with self.path.open("a", encoding="utf-8") as f:
                f.write(line + "\n")
        logger.info(line)

1
def write(event, *, user=None, **fields):
2
    print(event, user, fields)
3

4

5
write("approval.create", user="dad", approval="nCCzRFogNQvd", rule="same_user")
6

7
info = {"approval": "nCCzRFogNQvd", "action": "mail.send"}
8
write("approval.approve", user="mom", **info)
9

10
print({**info, "rule": "admin"})

{
  "name": "mail",
  "title": "Mail",
  "actions": {
    "send": {
      "title": "Send email",
      "approval": true,
      "roles": null,
      "params": { "type": "object", "required": ["to", "subject", "body"] }
    }
  }
}

{
  "params": { "to": ["grandma@family.example"], "subject": "Leaving work", "body": "Heading out now" },
  "caller": { "user_id": "dad", "name": "Dad", "role": "admin", "email": "dad@example.com" }
}

1
"""Module catalog. The list of modules the hub knows, and the HTTP client for calling them.
2

3
The hub doesn't import module code. It fetches the manifest (self-introduction) from the URL in the settings,
4
and when a command comes in, it just sends an HTTP request to that URL. See docs/module-contract.md for the contract.
5
"""
6
import logging
7
import re
8
import time
9

10
import httpx
11
from pydantic import BaseModel, Field, ValidationError, field_validator
12

13
logger = logging.getLogger("secretary.catalog")
14

15
MANIFEST_TTL_SECONDS = 60
16
# Action names go into URL paths and scopes, so only lowercase letters, digits, _, and -
17
ACTION_RE = re.compile(r"[a-z0-9][a-z0-9_-]{0,31}")
18

19

20
class ModuleError(Exception):
21
    """Raised when calling a module fails. status is the HTTP status code to return to the client."""
22

23
    def __init__(self, status: int, message: str):
24
        super().__init__(message)
25
        self.status = status
26
        self.message = message
27

28

29
class ActionSpec(BaseModel):
30
    """One action in a manifest. The required scope isn't written separately; it's set as "<module>:<action>"."""
31

32
    title: str = Field(max_length=60)
33
    approval: bool = False
34
    roles: list[str] | None = None
35
    params: dict | None = None
36

37

38
class Manifest(BaseModel):
39
    """A module's self-introduction. The GET /manifest response is checked against this shape. Modules with the wrong shape aren't used."""
40

41
    name: str
42
    title: str = Field(max_length=60)
43
    actions: dict[str, ActionSpec]
44

45
    @field_validator("actions")
46
    @classmethod
47
    def _action_names(cls, actions: dict) -> dict:
48
        """If even one action breaks the action name rule (ACTION_RE), reject the whole manifest."""
49
        bad = [a for a in actions if not ACTION_RE.fullmatch(a)]
50
        if bad:
51
            raise ValueError(f"bad action names: {bad}")
52
        return actions
53

54

55
class ModuleCatalog:
56
    """Remembers the manifests of modules registered in the settings, and sends requests to modules."""
57

58
    def __init__(self, endpoints: dict[str, tuple[str, str]], timeout: float = 10.0, transport=None):
59
        """endpoints: module name → (URL, token). transport is passed only in tests, to attach a fake module."""
60
        self.endpoints = endpoints
61
        self._http = httpx.AsyncClient(timeout=timeout, transport=transport)
62
        self._cache: dict[str, tuple[float, Manifest]] = {}
63

64
    async def close(self) -> None:
65
        """Close the HTTP connections. Called when the server stops."""
66
        await self._http.aclose()
67

68
    def _headers(self, name: str) -> dict:
69
        """Headers with that module's internal token. The module only accepts requests when this token matches."""
70
        return {"Authorization": f"Bearer {self.endpoints[name][1]}"}
71

72
    async def manifest(self, name: str) -> Manifest:
73
        """Return the module's manifest. Use the remembered one for 1 minute, then fetch it again.
74

75
        503 if the module is down. Even if refetching fails, use the previously fetched manifest if there is one.
76
        """
77
        if name not in self.endpoints:
78
            raise ModuleError(404, "No such module.")
79
        cached = self._cache.get(name)
80
        if cached and time.monotonic() - cached[0] < MANIFEST_TTL_SECONDS:
81
            return cached[1]
82
        url = self.endpoints[name][0]
83
        try:
84
            res = await self._http.get(f"{url}/manifest", headers=self._headers(name))
85
            res.raise_for_status()
86
            manifest = Manifest.model_validate(res.json())
87
            if manifest.name != name:
88
                raise ValueError(f"manifest name mismatch: {manifest.name}")
89
        except (httpx.HTTPError, ValidationError, ValueError) as e:
90
            logger.warning("module %s manifest failed: %s", name, e)
91
            if cached:
92
                return cached[1]
93
            raise ModuleError(503, "Can't connect to the module.") from e
94
        self._cache[name] = (time.monotonic(), manifest)
95
        return manifest
96

97
    async def action(self, name: str, action: str) -> ActionSpec:
98
        """Find one of the module's actions. 404 if it doesn't exist."""
99
        spec = (await self.manifest(name)).actions.get(action)
100
        if spec is None:
101
            raise ModuleError(404, "No such action.")
102
        return spec
103

104
    async def _post(self, name: str, path: str, params: dict, caller: dict) -> dict:
105
        """Send a POST to the module and return the JSON response.
106

107
        If the module returns 422, show that message to the user as is (because the input was wrong).
108
        Other failures (not connected, 500, 401) record details only in the server log and become 502.
109
        """
110
        url = self.endpoints[name][0]
111
        try:
112
            res = await self._http.post(f"{url}{path}", json={"params": params, "caller": caller},
113
                                        headers=self._headers(name))
114
        except httpx.HTTPError as e:
115
            logger.warning("module %s %s failed: %s", name, path, e)
116
            raise ModuleError(502, "Can't connect to the module.") from e
117
        if res.status_code == 422:
118
            try:
119
                message = str(res.json().get("message", ""))[:200]
120
            except ValueError:
121
                message = ""
122
            raise ModuleError(422, message or "The request content is invalid.")
123
        if res.status_code != 200:
124
            logger.warning("module %s %s returned %s", name, path, res.status_code)
125
            raise ModuleError(502, "The module failed to execute.")
126
        return res.json()
127

128
    async def validate(self, name: str, action: str, params: dict, caller: dict) -> str:
129
        """Have the input checked before execution, and get a one-line summary to show on the approval screen."""
130
        data = await self._post(name, f"/actions/{action}/validate", params, caller)
131
        return str(data.get("summary", f"{name}.{action}"))[:200]
132

133
    async def execute(self, name: str, action: str, params: dict, caller: dict) -> str:
134
        """Hand the actual execution to the module and return the result text."""
135
        data = await self._post(name, f"/actions/{action}", params, caller)
136
        return str(data.get("result", "Done"))[:200]
137

138

139
def caller_context(user) -> dict:
140
    """The "who asked for it" info to pass to modules. Auth info like device keys or Secrets is never passed."""
141
    return {"user_id": user.id, "name": user.name, "role": user.role, "email": user.email}

1
import re
2

3
from pydantic import BaseModel, Field, ValidationError, field_validator
4

5
ACTION_RE = re.compile(r"[a-z0-9][a-z0-9_-]{0,31}")
6

7

8
class ActionSpec(BaseModel):
9
    title: str = Field(max_length=60)
10
    approval: bool = False
11

12

13
class Manifest(BaseModel):
14
    name: str
15
    actions: dict[str, ActionSpec]
16

17
    @field_validator("actions")
18
    @classmethod
19
    def _action_names(cls, actions: dict) -> dict:
20
        bad = [a for a in actions if not ACTION_RE.fullmatch(a)]
21
        if bad:
22
            raise ValueError(f"bad action names: {bad}")
23
        return actions
24

25

26
def check(data: dict) -> None:
27
    try:
28
        m = Manifest.model_validate(data)
29
        print("Passed:", m.actions["send"].title, m.actions["send"].approval)
30
    except ValidationError as e:
31
        for err in e.errors():
32
            print("Rejected:", err["loc"], err["msg"])
33

34

35
check({"name": "mail", "actions": {"send": {"title": "Send email", "approval": True}}})
36
check({"name": "mail", "actions": {"send": {"approval": "maybe"}}})
37
check({"name": "mail", "actions": {"../admin": {"title": "Admin"}}})