Building a Home Secretary #4 - Server: Family and Permissions

1
"""Permission scopes, roles, and device presets.
2

3
There are two kinds of scopes.
4
  Hub scopes      system:read, approvals:decide      Things the hub does itself
5
  Module scopes   <module>:<action>  e.g. mail:send   Created automatically for each action a module lists in its manifest
6

7
Module scopes can be granted in bundles with wildcards.
8
  mail:*      Every action of the mail module
9
  modules:*   Every action of every module (including modules added later)
10

11
Permissions you can actually use = scopes given to the device AND scopes the user's role allows.
12
So even if a child's device is accidentally given a broad preset, it can't exceed the role's ceiling.
13
"""
14
import re
15

16
SYSTEM_READ = "system:read"            # View my info, module list, events
17
APPROVALS_DECIDE = "approvals:decide"  # View, approve, and reject requests I can decide on
18
ALL_MODULES = "modules:*"              # Every action of every module
19

20
HUB_SCOPES = {SYSTEM_READ, APPROVALS_DECIDE}
21

22
# Module scope shape: module name ':' action name (or *)
23
_MODULE_SCOPE_RE = re.compile(r"([a-z0-9][a-z0-9-]{1,23}):([a-z0-9][a-z0-9_-]{0,31}|\*)")
24

25
# Roles
26
ADMIN = "admin"    # Head of the house. All permissions, approves children's requests, manages keys via the NAS CLI
27
MEMBER = "member"  # Adult family member. Approves their own requests with their own devices
28
CHILD = "child"    # Child. No approval permission. Requests needing approval are approved by an admin
29

30
ROLES = (ADMIN, MEMBER, CHILD)
31

32
# Role ceilings. To restrict further per module, use roles in the module manifest.
33
ROLE_SCOPES: dict[str, set[str]] = {
34
    ADMIN: {SYSTEM_READ, APPROVALS_DECIDE, ALL_MODULES},
35
    MEMBER: {SYSTEM_READ, APPROVALS_DECIDE, ALL_MODULES},
36
    CHILD: {SYSTEM_READ, ALL_MODULES},
37
}
38

39
# Recommended defaults per device type (used with --preset in the CLI)
40
PRESETS: dict[str, set[str]] = {
41
    "phone": {SYSTEM_READ, APPROVALS_DECIDE, ALL_MODULES},
42
    "watch": {SYSTEM_READ, APPROVALS_DECIDE},
43
    "glasses": {SYSTEM_READ, ALL_MODULES},
44
    "mcp": {SYSTEM_READ},
45
    "test": {SYSTEM_READ},
46
}
47

48

49
def module_of(scope: str) -> str | None:
50
    """Return the module name for a module scope, or None for a hub scope. e.g. "mail:send" → "mail", "modules:*" → "*" """
51
    if scope == ALL_MODULES:
52
        return "*"
53
    if scope in HUB_SCOPES:
54
        return None
55
    m = _MODULE_SCOPE_RE.fullmatch(scope)
56
    return m.group(1) if m else None
57

58

59
def is_valid_scope(scope: str) -> bool:
60
    """Whether it has a shape that can be given to a device. Scopes for modules that don't exist yet can be given in advance."""
61
    return scope in HUB_SCOPES or module_of(scope) is not None
62

63

64
def allows(granted, required: str) -> bool:
65
    """Whether granted (the scopes held) allows required (a single needed scope).
66

67
    Allowed if the exact same scope is present, or, for a module scope, if "<module>:*" or "modules:*" is present.
68
    e.g. {"mail:*"} allows "mail:send" but not "tapo:on".
69
    """
70
    granted = set(granted)
71
    if required in granted:
72
        return True
73
    module = module_of(required)
74
    if module is None or module == "*":
75
        return False
76
    return ALL_MODULES in granted or f"{module}:*" in granted
77

78

79
def effective_scopes(device_scopes, role: str) -> set[str]:
80
    """Permissions actually usable with this device = scopes given to the device AND scopes the role allows.
81

82
    Each device scope is checked against the role ceiling, and only those the ceiling allows are kept.
83
    e.g. Even if a child's device is given approvals:decide, it's dropped because it's not in the child ceiling.
84
    "modules:*" itself is kept only if it's in the ceiling as is. For an unknown role, an empty set (no permissions).
85
    """
86
    cap = ROLE_SCOPES.get(role, set())
87
    return {s for s in device_scopes if s in cap or allows(cap, s)}

1
"""User and device store (data/registry.json).
2

3
There are people (users), and devices belong to people. Since it's a service for one
4
household, there's a cap on the number of active users (4 by default).
5

6
With HMAC, the server also has to recompute the signature with the same secret, so device
7
Secrets are stored in plain form. That's why the file is created with permissions 600 from
8
the start, and this file never leaves the data volume.
9
"""
10
import json
11
import os
12
import re
13
import secrets
14
import threading
15
import time
16
from dataclasses import asdict, dataclass
17
from pathlib import Path
18

19
from .scopes import ADMIN, ROLES, is_valid_scope
20

21
# Rules for user IDs and device names. Used with fullmatch(), so the "entire" string must have this shape.
22
# (It becomes the front part of the Access Key and travels in HTTP headers, so only lowercase letters, digits, and hyphens)
23
#   [a-z0-9]         First character: a lowercase letter or digit (can't start with a hyphen)
24
#   [a-z0-9-]{1,23}  The remaining 1–23 characters: lowercase letters, digits, hyphens
25
#   → 2–24 characters in total
26
#   Pass: dad, kid1, phone-flutter
27
#   Reject: -dad (starts with hyphen), Dad (uppercase), 내폰 (Korean, non-ASCII), a (1 character), 25+ characters
28
_ID_RE = re.compile(r"[a-z0-9][a-z0-9-]{1,23}")
29

30
# Email format check. It only lightly checks whether it "looks like an email," not whether the address actually exists.
31
#   [^@\s]+  1+ characters that are neither '@' nor whitespace → the local part (dad)
32
#   @        exactly one '@'
33
#   [^@\s]+  1+ characters that are neither '@' nor whitespace → the domain name (example)
34
#   \.       one dot (.). In regex, '.' means "any character," so add \ to use a real dot
35
#   [^@\s]+  the rest after the dot → (com, co.kr)
36
#   ([^...] means "not the characters in the brackets," \s is whitespace, + means "1 or more")
37
#   Pass: dad@example.com, mom@mail.co.kr
38
#   Reject: dad@example (no dot), dad@@example.com (two @), "a b@c.d" (whitespace)
39
_EMAIL_RE = re.compile(r"[^@\s]+@[^@\s]+\.[^@\s]+")
40

41

42
class RegistryError(ValueError):
43
    """Error raised when user/device registration rules are broken. The message is shown on the CLI screen as is."""
44

45

46
@dataclass
47
class User:
48
    """One family member. role is one of admin/member/child; email is used for the mail Reply-To."""
49

50
    id: str
51
    name: str
52
    role: str
53
    created_at: int
54
    email: str | None = None
55
    disabled: bool = False
56

57
    def public(self) -> dict:
58
        """Pick only the info to include in API responses (ID, name, and role, without email)."""
59
        return {"id": self.id, "name": self.name, "role": self.role}
60

61

62
@dataclass
63
class Device:
64
    """One device belonging to a person (phone, watch, glasses...). access_key is the device ID and secret is the signing secret.
65

66
    Revoking doesn't delete it; it's kept with revoked=True, so the keys the audit log points to can still be identified later.
67
    """
68

69
    access_key: str
70
    user: str
71
    name: str
72
    secret: str
73
    scopes: list[str]
74
    created_at: int
75
    revoked: bool = False
76
    revoked_at: int | None = None
77

78

79
class Registry:
80
    """A store that saves users and devices to a single registry.json file and reads them back."""
81

82
    def __init__(self, path: Path, max_users: int = 4):
83
        """Take the file path and max number of users, and read the file right away if it exists."""
84
        self.path = path
85
        self.max_users = max_users
86
        self._lock = threading.Lock()
87
        self._users: dict[str, User] = {}
88
        self._devices: dict[str, Device] = {}
89
        self._mtime: float | None = None
90
        self._load()
91

92
    def _load(self) -> None:
93
        """Read registry.json into memory. Reread only when the file's modification time changes.
94

95
        The CLI (docker exec) is a different process from the server, so adding a device via the CLI only changes the file.
96
        So it checks the modification time every time, and if it changed, reads the new content without restarting the server.
97
        """
98
        if not self.path.exists():
99
            self._users, self._devices, self._mtime = {}, {}, None
100
            return
101
        mtime = self.path.stat().st_mtime
102
        if mtime == self._mtime:
103
            return
104
        raw = json.loads(self.path.read_text(encoding="utf-8"))
105
        self._users = {k: User(**v) for k, v in raw.get("users", {}).items()}
106
        self._devices = {k: Device(**v) for k, v in raw.get("devices", {}).items()}
107
        self._mtime = mtime
108

109
    def _save(self) -> None:
110
        """Write the in-memory contents to registry.json.
111

112
        1. Create a temporary file (.tmp) with permissions 600 (only I can read and write) from the start and write to it,
113
        2. then swap it with the original file in one go using os.replace.
114
        Even if the server dies while writing, the original file isn't left half-written.
115
        """
116
        self.path.parent.mkdir(parents=True, exist_ok=True)
117
        tmp = self.path.with_suffix(".tmp")
118
        payload = json.dumps(
119
            {
120
                "users": {k: asdict(v) for k, v in self._users.items()},
121
                "devices": {k: asdict(v) for k, v in self._devices.items()},
122
            },
123
            ensure_ascii=False,
124
            indent=2,
125
        )
126
        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
127
        with os.fdopen(fd, "w", encoding="utf-8") as f:
128
            f.write(payload)
129
        os.replace(tmp, self.path)
130
        self._mtime = self.path.stat().st_mtime
131

132
    # ── Users ──────────────────────────────────────────────
133

134
    def user(self, user_id: str) -> User | None:
135
        """Find one user by ID. None if not found."""
136
        with self._lock:
137
            self._load()
138
            return self._users.get(user_id)
139

140
    def users(self) -> list[User]:
141
        """List of all users (including disabled ones)."""
142
        with self._lock:
143
            self._load()
144
            return list(self._users.values())
145

146
    def add_user(self, user_id: str, name: str, role: str, email: str | None = None) -> User:
147
        """Add a user. RegistryError if it breaks the rules.
148

149
        - ID is 2–24 lowercase letters, digits, and hyphens; role is one of admin/member/child
150
        - Up to max_users active users
151
        - The first user (when there's no active admin) must be an admin
152
        """
153
        if not _ID_RE.fullmatch(user_id):
154
            raise RegistryError("User ID must be 2-24 lowercase letters, digits, and hyphens. e.g. dad, mom, kid1")
155
        if role not in ROLES:
156
            raise RegistryError(f"Role must be one of {', '.join(ROLES)}.")
157
        if email and not _EMAIL_RE.fullmatch(email):
158
            raise RegistryError("Invalid email format.")
159
        with self._lock:
160
            self._load()
161
            if user_id in self._users:
162
                raise RegistryError("That user ID already exists.")
163
            active = [u for u in self._users.values() if not u.disabled]
164
            if len(active) >= self.max_users:
165
                raise RegistryError(f"You can register up to {self.max_users} users.")
166
            if not any(u.role == ADMIN for u in active) and role != ADMIN:
167
                raise RegistryError("The first user must be an admin.")
168
            user = User(id=user_id, name=name, role=role, email=email, created_at=int(time.time()))
169
            self._users[user_id] = user
170
            self._save()
171
            return user
172

173
    def disable_user(self, user_id: str) -> int:
174
        """Disable a user and revoke all of that person's device keys. Returns the number of keys revoked."""
175
        with self._lock:
176
            self._load()
177
            user = self._users.get(user_id)
178
            if not user or user.disabled:
179
                raise RegistryError("That user doesn't exist or is already disabled.")
180
            others = [u for u in self._users.values() if u.id != user_id and not u.disabled and u.role == ADMIN]
181
            if user.role == ADMIN and not others:
182
                raise RegistryError("The last admin can't be disabled.")
183
            user.disabled = True
184
            count = self._revoke_where(lambda d: d.user == user_id)
185
            self._save()
186
            return count
187

188
    def delete_user(self, user_id: str, has_history: bool) -> User:
189
        """Delete a user completely. Allowed only when there's no trace, like a user created by a typo.
190

191
        If they have devices (including revoked ones) or approval records, don't delete. The audit log
192
        and approval records point to this ID, and deleting it would make it impossible to tell who did what later.
193
        Disable such users with disable_user() instead.
194
        """
195
        with self._lock:
196
            self._load()
197
            user = self._users.get(user_id)
198
            if not user:
199
                raise RegistryError("That user doesn't exist.")
200
            if any(d.user == user_id for d in self._devices.values()):
201
                raise RegistryError("A user who has ever had a device registered can't be deleted. Use user disable instead.")
202
            if has_history:
203
                raise RegistryError("A user with approval records can't be deleted. Use user disable instead.")
204
            if user.role == ADMIN and not user.disabled:
205
                others = [u for u in self._users.values() if u.id != user_id and not u.disabled and u.role == ADMIN]
206
                if not others and len(self._users) > 1:
207
                    raise RegistryError("The last admin can't be deleted while other users exist.")
208
            del self._users[user_id]
209
            self._save()
210
            return user
211

212
    # ── Devices ────────────────────────────────────────────
213

214
    def device(self, access_key: str) -> Device | None:
215
        """Find one device by Access Key. None if not found. (Revoked devices are returned too, so check revoked separately)"""
216
        with self._lock:
217
            self._load()
218
            return self._devices.get(access_key)
219

220
    def devices(self) -> list[Device]:
221
        """List of all devices (including revoked ones)."""
222
        with self._lock:
223
            self._load()
224
            return list(self._devices.values())
225

226
    def add_device(self, user_id: str, name: str, scopes: set[str]) -> Device:
227
        """Issue a new device key to a user.
228

229
        Access Key = "userID-deviceName-8 random chars" (e.g. dad-phone-1a2b3c4d)
230
        Secret     = 32 random bytes (secrets.token_urlsafe: an unguessable random value)
231
        scopes are stored as is; the role ceiling trims them with effective_scopes() at request time.
232
        """
233
        # The name becomes the front part of the Access Key and travels in HTTP headers, so only lowercase letters, digits, and hyphens.
234
        if not _ID_RE.fullmatch(name):
235
            raise RegistryError("Device name must be 2-24 lowercase letters, digits, and hyphens. e.g. phone, glasses")
236
        bad = {s for s in scopes if not is_valid_scope(s)}
237
        if bad:
238
            raise RegistryError(f"Invalid scope shape: {', '.join(sorted(bad))} (e.g. system:read, mail:send, mail:*, modules:*)")
239
        with self._lock:
240
            self._load()
241
            user = self._users.get(user_id)
242
            if not user or user.disabled:
243
                raise RegistryError("That user doesn't exist or is disabled.")
244
            access_key = f"{user_id}-{name}-{secrets.token_hex(4)}"
245
            device = Device(
246
                access_key=access_key,
247
                user=user_id,
248
                name=name,
249
                secret=secrets.token_urlsafe(32),
250
                scopes=sorted(scopes),
251
                created_at=int(time.time()),
252
            )
253
            self._devices[access_key] = device
254
            self._save()
255
            return device
256

257
    def _revoke_where(self, predicate) -> int:
258
        """Mark every device matching the condition (predicate) as revoked, and return how many were revoked.
259

260
        predicate is a function that takes a device and returns True/False. e.g. lambda d: d.user == "kid1"
261
        Saving (_save) is done by the caller.
262
        """
263
        now = int(time.time())
264
        count = 0
265
        for device in self._devices.values():
266
            if not device.revoked and predicate(device):
267
                device.revoked = True
268
                device.revoked_at = now
269
                count += 1
270
        return count
271

272
    def revoke(self, access_key: str) -> bool:
273
        """Revoke one device key. True if revoked, False if it doesn't exist or is already revoked."""
274
        with self._lock:
275
            self._load()
276
            count = self._revoke_where(lambda d: d.access_key == access_key)
277
            if count:
278
                self._save()
279
            return bool(count)
280

281
    def revoke_all(self) -> int:
282
        """Emergency stop: invalidate every device key at once."""
283
        with self._lock:
284
            self._load()
285
            count = self._revoke_where(lambda d: True)
286
            if count:
287
                self._save()
288
            return count

from dataclasses import asdict, dataclass


@dataclass
class User:
    id: str
    name: str
    role: str = "member"


u = User("mom", "Mom")
print(u)
print(asdict(u))
print(u == User("mom", "Mom"))