Building a Home Secretary #6 - Server: Family Registration and Tests

1
"""User/device management CLI. Run only inside the NAS (keys can't be issued through the external API).
2

3
  # Users (up to SECRETARY_MAX_USERS, 4 by default. The first user is admin)
4
  docker exec -it secretary python -m app.cli user add dad "Dad" --role admin --email dad@example.com
5
  docker exec -it secretary python -m app.cli user add kid1 "Eldest" --role child
6
  docker exec -it secretary python -m app.cli user list
7
  docker exec -it secretary python -m app.cli user disable kid1        # Also revokes all of that person's device keys
8
  docker exec -it secretary python -m app.cli user delete typo-id      # Fully deletes only users with no device/approval records
9

10
  # Devices
11
  docker exec -it secretary python -m app.cli device add dad phone --preset phone
12
  docker exec -it secretary python -m app.cli device list
13
  docker exec -it secretary python -m app.cli device revoke <access_key>
14
  docker exec -it secretary python -m app.cli device revoke-all       # Emergency stop
15

16
  # Modules
17
  docker exec -it secretary python -m app.cli module list             # Connection status and action (= scope) list
18
"""
19
import argparse
20
import sys
21
from datetime import datetime
22

23
import httpx
24

25
from .audit import AuditLog
26
from .auth.registry import Registry, RegistryError
27
from .catalog import Manifest
28
from .auth.scopes import HUB_SCOPES, PRESETS, ROLES, effective_scopes
29
from .config import get_settings
30
from .db import Database
31
from .approvals import ApprovalQueue
32

33

34
def _fmt(ts: int | None) -> str:
35
    """Turn a Unix time (seconds) into text like '2026-11-04 21:30'. '-' if there's no value."""
36
    return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M") if ts else "-"
37

38

39
def build_parser() -> argparse.ArgumentParser:
40
    """Define the command structure. Subcommands like add, list ... hang under user/device.
41

42
    argparse filters out invalid input and generates -h help automatically.
43
    """
44
    parser = argparse.ArgumentParser(prog="secretary")
45
    sub = parser.add_subparsers(dest="cmd", required=True)
46

47
    usr = sub.add_parser("user", help="Manage users").add_subparsers(dest="action", required=True)
48
    ua = usr.add_parser("add", help="Add a user")
49
    ua.add_argument("user_id", help="Lowercase ID. e.g. dad, mom, kid1")
50
    ua.add_argument("name", help="Display name. Also used as the sender name on emails")
51
    ua.add_argument("--role", choices=ROLES, required=True)
52
    ua.add_argument("--email", help="Your own email. Passed along to modules (the mail module uses it for Reply-To and to allow sending to yourself)")
53
    usr.add_parser("list", help="List users")
54
    ud = usr.add_parser("disable", help="Disable a user (also revokes all their device keys)")
55
    ud.add_argument("user_id")
56
    udel = usr.add_parser("delete", help="Fully delete a user (only if they have no device/approval records)")
57
    udel.add_argument("user_id")
58

59
    dev = sub.add_parser("device", help="Manage devices").add_subparsers(dest="action", required=True)
60
    da = dev.add_parser("add", help="Issue a device key")
61
    da.add_argument("user_id")
62
    da.add_argument("name", help="Lowercase device name. e.g. phone, watch, glasses")
63
    da.add_argument("--preset", choices=sorted(PRESETS))
64
    da.add_argument("--scope", action="append", default=[], help=f"Specify directly (repeatable): {', '.join(sorted(HUB_SCOPES))}, <module>:<action>, <module>:*, modules:*")
65
    dev.add_parser("list", help="List devices")
66
    dr = dev.add_parser("revoke", help="Revoke a device key")
67
    dr.add_argument("access_key")
68
    dev.add_parser("revoke-all", help="Revoke all device keys (emergency stop)")
69

70
    mod = sub.add_parser("module", help="Check modules").add_subparsers(dest="action", required=True)
71
    mod.add_parser("list", help="Connection status and action list of registered modules")
72
    return parser
73

74

75
def main(argv: list[str] | None = None) -> int:
76
    """CLI entry point. Parses the command and hands it to _user, _device, or _module.
77

78
    The return value is the exit code: 0 success, 1 target not found, 2 input error (rule violation).
79
    """
80
    args = build_parser().parse_args(argv)
81
    settings = get_settings()
82
    if args.cmd == "module":
83
        return _module(settings)
84
    reg = Registry(settings.registry_path, settings.max_users)
85
    audit = AuditLog(settings.audit_path)
86

87
    try:
88
        if args.cmd == "user":
89
            return _user(args, reg, audit)
90
        return _device(args, reg, audit)
91
    except RegistryError as e:
92
        print(e, file=sys.stderr)
93
        return 2
94

95

96
def _user(args, reg: Registry, audit: AuditLog) -> int:
97
    """Handle user add / list / disable / delete. Changes are recorded in the audit log with via=cli."""
98
    if args.action == "add":
99
        user = reg.add_user(args.user_id, args.name, args.role, args.email)
100
        audit.write("user.add", user=user.id, via="cli", role=user.role)
101
        print(f"Added user: {user.id} ({user.name}, {user.role})")
102
        return 0
103
    if args.action == "list":
104
        users = reg.users()
105
        active = sum(1 for u in users if not u.disabled)
106
        print(f"Active users {active}/{reg.max_users}")
107
        for u in users:
108
            state = "disabled" if u.disabled else "active"
109
            print(f"  {u.id:<12} {u.name:<10} {u.role:<7} {state:<8} {u.email or '-'}")
110
        return 0
111
    if args.action == "disable":
112
        count = reg.disable_user(args.user_id)
113
        audit.write("user.disable", user=args.user_id, via="cli", revoked_devices=count)
114
        print(f"Disabled. Revoked {count} device key(s).")
115
        return 0
116
    if args.action == "delete":
117
        settings = get_settings()
118
        db = Database(settings.db_path)
119
        try:
120
            history = ApprovalQueue(db, settings.approval_ttl_seconds).has_history(args.user_id)
121
        finally:
122
            db.close()
123
        user = reg.delete_user(args.user_id, has_history=history)
124
        audit.write("user.delete", user=user.id, via="cli", role=user.role)
125
        print(f"Deleted: {user.id} ({user.name}, {user.role})")
126
        return 0
127
    return 1
128

129

130
def _device(args, reg: Registry, audit: AuditLog) -> int:
131
    """Handle device add / list / revoke / revoke-all.
132

133
    add shows the Secret on screen only this once. If any scopes are dropped because of the role ceiling, it says so too.
134
    """
135
    if args.action == "add":
136
        scopes = set(PRESETS.get(args.preset, set())) | set(args.scope)
137
        if not scopes:
138
            print("You must specify either --preset or --scope.", file=sys.stderr)
139
            return 2
140
        device = reg.add_device(args.user_id, args.name, scopes)
141
        user = reg.user(args.user_id)
142
        effective = effective_scopes(device.scopes, user.role)
143
        audit.write("device.add", user=user.id, device=device.access_key, via="cli", scopes=device.scopes)
144
        print("Issued a device key. The Secret is shown only this once.\n")
145
        print(f"  User       : {user.id} ({user.name}, {user.role})")
146
        print(f"  Access Key : {device.access_key}")
147
        print(f"  Secret     : {device.secret}")
148
        print(f"  Scopes     : {', '.join(sorted(effective))}")
149
        dropped = set(device.scopes) - effective
150
        if dropped:
151
            print(f"  (Scopes not actually usable because of the {user.role} role ceiling: {', '.join(sorted(dropped))})")
152
        return 0
153
    if args.action == "list":
154
        for d in reg.devices():
155
            state = "revoked" if d.revoked else "in use"
156
            print(f"{d.access_key:<32} {d.user:<10} {state:<7} issued {_fmt(d.created_at)}  {', '.join(d.scopes)}")
157
        return 0
158
    if args.action == "revoke":
159
        ok = reg.revoke(args.access_key)
160
        if ok:
161
            audit.write("device.revoke", device=args.access_key, via="cli")
162
        print("Revoked." if ok else "That key doesn't exist or is already revoked.")
163
        return 0 if ok else 1
164
    if args.action == "revoke-all":
165
        count = reg.revoke_all()
166
        audit.write("device.revoke_all", via="cli", count=count)
167
        print(f"Revoked {count} key(s).")
168
        return 0
169
    return 1
170

171

172
def _module(settings) -> int:
173
    """Handle module list. Fetch each module's manifest and show its actions and required scopes.
174

175
    Look at this list when you're not sure which --scope to give a device.
176
    """
177
    endpoints = settings.module_endpoints
178
    if not endpoints:
179
        print("No modules registered. Check SECRETARY_MODULES.")
180
        return 0
181
    for name, (url, token) in endpoints.items():
182
        try:
183
            res = httpx.get(f"{url}/manifest", headers={"Authorization": f"Bearer {token}"}, timeout=5)
184
            res.raise_for_status()
185
            manifest = Manifest.model_validate(res.json())
186
        except Exception as e:  # Whatever the reason, show it as "not connected" and move on to the next module
187
            print(f"{name:<12} not connected  {url}  ({type(e).__name__})")
188
            continue
189
        print(f"{name:<12} OK             {url}  ({manifest.title})")
190
        for action, spec in manifest.actions.items():
191
            notes = (["approval required"] if spec.approval else []) + ([f"roles: {', '.join(spec.roles)}"] if spec.roles else [])
192
            print(f"    {name}:{action:<20} {spec.title}{'  (' + ', '.join(notes) + ')' if notes else ''}")
193
    return 0
194

195

196
if __name__ == "__main__":
197
    sys.exit(main())

import json
import time

import httpx
import pytest
from fastapi.testclient import TestClient

from app.auth.registry import Registry
from app.auth.scopes import ADMIN, CHILD, MEMBER, PRESETS
from app.auth.signing import build_headers
from app.config import Settings
from app.main import create_app
from tests import fake_module

CF = {"Cf-Access-Jwt-Assertion": "test-jwt"}
DOWN_TOKEN = "x" * 40


@pytest.fixture
def env(tmp_path):
    settings = Settings(
        data_dir=tmp_path, require_cf_access=True, lockout_threshold=5,
        modules="demo=http://fake/demo,down=http://fake/down",
        module_tokens=f"demo={fake_module.TOKEN},down={DOWN_TOKEN}",
    )
    reg = Registry(settings.registry_path, settings.max_users)
    reg.add_user("dad", "Dad", ADMIN, "dad@example.com")
    reg.add_user("mom", "Mom", MEMBER, "mom@example.com")
    reg.add_user("kid", "Eldest", CHILD)
    d = {
        "dad_phone": reg.add_device("dad", "phone", PRESETS["phone"]),
        "dad_watch": reg.add_device("dad", "watch", PRESETS["watch"]),
        "dad_glasses": reg.add_device("dad", "glasses", PRESETS["glasses"]),
        "mom_phone": reg.add_device("mom", "phone", PRESETS["phone"]),
        "mom_watch": reg.add_device("mom", "watch", PRESETS["watch"]),
        "kid_phone": reg.add_device("kid", "phone", PRESETS["phone"]),  # Approval permission gets cut by the role ceiling
        "dad_ping": reg.add_device("dad", "ping-only", {"demo:ping"}),
    }
    module_app, executed = fake_module.build()
    app = create_app(settings, module_transport=httpx.ASGITransport(app=module_app))
    with TestClient(app) as client:
        yield client, settings, reg, d, executed


def call(client, device, method, path, body=None, query="", headers=None):
    raw = json.dumps(body).encode() if body is not None else b""
    h = {**CF, **build_headers(device.access_key, device.secret, method, path, query, raw)}
    if body is not None:
        h["Content-Type"] = "application/json"
    h.update(headers or {})
    url = f"{path}?{query}" if query else path
    return client.request(method, url, content=raw, headers=h)


def command(client, device, action="send", module="demo", **params):
    return call(client, device, "POST", "/v1/commands", {"module": module, "action": action, "params": params})


def request_send(client, device, to="grandma@family.example"):
    return command(client, device, "send", to=to)


def pending_ids(client, device):
    r = call(client, device, "GET", "/v1/approvals", query="status=pending")
    assert r.status_code == 200
    return [a["id"] for a in r.json()["items"]]


# ── Authentication ───────────────────────────────────────────

1
#!/usr/bin/env python3
2
"""Example client for calling Secretary. Uses only the standard library.
3

4
Other clients (Flutter, G2, Zepp) just need to sign the same way as sign_headers() in this file.
5

6
Environment variables:
7
  SECRETARY_URL          https://api.example.house
8
  CF_CLIENT_ID           Cloudflare Access service token ID
9
  CF_CLIENT_SECRET       Cloudflare Access service token Secret
10
  SECRETARY_ACCESS_KEY   Secretary device key
11
  SECRETARY_SECRET       Secretary device Secret
12

13
Usage:
14
  python3 secretary_client.py GET /v1/me
15
  python3 secretary_client.py GET /v1/modules
16
  python3 secretary_client.py POST /v1/commands '{"module":"mail","action":"send","params":{"to":["me@example.com"],"subject":"Test","body":"Hello"}}'
17
  python3 secretary_client.py GET "/v1/approvals?status=pending"
18
  python3 secretary_client.py POST /v1/approvals/<id>/approve
19
"""
20
import base64
21
import hashlib
22
import hmac
23
import json
24
import os
25
import secrets
26
import sys
27
import time
28
import urllib.error
29
import urllib.parse
30
import urllib.request
31

32
USER_AGENT = "secretary-client/1.0"
33

34

35
def sign_headers(access_key: str, secret: str, method: str, path: str, query: str, body: bytes) -> dict:
36
    """Build the 4 signature headers to attach to one request. Exactly the same method as the server's app/auth/signing.py.
37

38
    When building a client in another language, just port this one function.
39
    """
40
    timestamp = str(int(time.time()))
41
    nonce = secrets.token_hex(16)
42
    target = f"{path}?{query}" if query else path
43
    canonical = "\n".join([method.upper(), target, timestamp, nonce, hashlib.sha256(body).hexdigest()])
44
    digest = hmac.new(secret.encode(), canonical.encode(), hashlib.sha256).digest()
45
    return {
46
        "X-Access-Key": access_key,
47
        "X-Timestamp": timestamp,
48
        "X-Nonce": nonce,
49
        "X-Signature": base64.urlsafe_b64encode(digest).decode().rstrip("="),
50
    }
51

52

53
def main() -> int:
54
    """Send the request given as command-line arguments, and print the status code and response JSON nicely.
55

56
    Attaches both the Cloudflare Access service token headers and the Secretary signature headers. Exit code 1 for 400 and above.
57
    """
58
    if len(sys.argv) < 3:
59
        print(__doc__)
60
        return 2
61
    method, target = sys.argv[1].upper(), sys.argv[2]
62
    body = sys.argv[3].encode() if len(sys.argv) > 3 else b""
63
    parsed = urllib.parse.urlsplit(target)
64

65
    env = os.environ
66
    headers = {
67
        # The default urllib User-Agent (Python-urllib/3.x) gets blocked by Cloudflare Browser Integrity Check (error code: 1010)
68
        "User-Agent": USER_AGENT,
69
        "CF-Access-Client-Id": env["CF_CLIENT_ID"],
70
        "CF-Access-Client-Secret": env["CF_CLIENT_SECRET"],
71
        **sign_headers(env["SECRETARY_ACCESS_KEY"], env["SECRETARY_SECRET"], method, parsed.path, parsed.query, body),
72
    }
73
    if body:
74
        headers["Content-Type"] = "application/json"
75

76
    req = urllib.request.Request(env["SECRETARY_URL"].rstrip("/") + target, data=body or None, method=method,
77
                                 headers=headers)
78
    try:
79
        with urllib.request.urlopen(req, timeout=10) as res:
80
            status, payload = res.status, res.read()
81
    except urllib.error.HTTPError as e:
82
        status, payload = e.code, e.read()
83
    print(status)
84
    try:
85
        print(json.dumps(json.loads(payload), ensure_ascii=False, indent=2))
86
    except ValueError:
87
        print(payload.decode(errors="replace")[:500])
88
    return 0 if status < 400 else 1
89

90

91
if __name__ == "__main__":
92
    sys.exit(main())