Coverage for dispatch/runner.py: 99%
106 statements
« prev ^ index » next coverage.py v7.9.2, created at 2026-07-24 16:42 +0000
1"""Runner identity: registration, token verification, and lazy GC (spec §7.1, §8).
3Identity is soft state that can be rebuilt from scratch (ADR-0004): the
4``runners:registered`` ZSET tracks last-seen, while meta/token_hash carry a 7d
5TTL. TTL is the *only* thing that invalidates a live identity. GC never kills an
6identity — it lazily sweeps the corpses (ZSET members have no TTL of their own)
7left behind once ``token_hash`` has already evaporated, so a heartbeat that
8renews an identity between GC's scan and sweep can never lose its just-renewed
9token (the TOCTOU is structurally impossible, no atomicity machinery needed).
11Security notes:
12- The runner token is returned exactly once. Only its SHA-256 hex is stored, so
13 a Redis dump never reveals a usable credential.
14- All token comparisons are constant-time (``hmac.compare_digest``).
15- A missing ``token_hash`` key means the identity was revoked (or expired) →
16 verification fails. This is the single revocation mechanism (ADR-0004).
17- Registration verification fails closed: if the shared secret is unset/empty in
18 the deployment settings (startup snapshot, ADR-0005), every candidate is
19 rejected rather than accepted or crashing.
20- Verification never raises on hostile input: non-str values and lone UTF-16
21 surrogate strings (which json.loads accepts but .encode() rejects) fail closed
22 to False → 401, never a 500.
23"""
25import hashlib
26import hmac
27import secrets
28import time
29from dataclasses import dataclass
30from typing import Dict, List, Optional
32from ulid import ULID
34from config import settings
35from mongo.utils import RedisCache
36from . import params
37from . import redis_keys
39RUNNER_ID_PREFIX = 'rn_'
40RUNNER_TOKEN_PREFIX = 'rk_'
42# One shared RedisCache instance. In production every RedisCache() shares the
43# pooled real Redis, but under fakeredis each instance gets an isolated dataset;
44# caching one instance keeps register/verify/list coherent in both worlds.
45_cache: Optional[RedisCache] = None
48@dataclass(frozen=True)
49class Registration:
50 runner_id: str
51 token: str
54def _redis():
55 global _cache
56 if _cache is None:
57 _cache = RedisCache()
58 return _cache.client
61def _now() -> float:
62 return time.time()
65def _token_hash(token: str) -> str:
66 return hashlib.sha256(token.encode()).hexdigest()
69def _utf8(s: str) -> Optional[bytes]:
70 """UTF-8 encode ``s``, or None if it cannot be encoded (fail closed).
72 json.loads happily yields str values holding a lone UTF-16 surrogate (the
73 JSON literal "\\ud800" decodes to such a str), and .encode() on those raises
74 UnicodeEncodeError. Callers at the trust boundary use None to fail closed
75 instead of letting a hostile body turn into a 500.
76 """
77 try:
78 return s.encode()
79 except UnicodeEncodeError:
80 return None
83def verify_registration_token(candidate: Optional[str]) -> bool:
84 """Constant-time check of a register request's shared secret. Fails closed.
86 The secret is a startup-snapshot deployment setting (ADR-0005): rotating
87 it takes a Back-End restart; unset/empty ⇒ registration is disabled.
88 """
89 expected = settings.RUNNER_REGISTRATION_TOKEN
90 # Fail closed: no configured secret ⇒ registration is disabled, not open.
91 # Reject non-str candidates too — a JSON body can carry ints/lists/bytes,
92 # and .encode() below would otherwise raise instead of returning False.
93 if not expected or not isinstance(candidate, str) or not candidate:
94 return False
95 # Compare UTF-8 bytes: compare_digest accepts str only when both sides are
96 # ASCII ("str (ASCII only)", hmac docs) and raises TypeError otherwise;
97 # `candidate` is attacker-controlled, so str comparison could crash (500)
98 # instead of failing closed (401). _utf8 also fails closed on a lone-
99 # surrogate candidate whose .encode() would raise UnicodeEncodeError.
100 expected_bytes = _utf8(expected)
101 candidate_bytes = _utf8(candidate)
102 if expected_bytes is None or candidate_bytes is None:
103 return False
104 return hmac.compare_digest(expected_bytes, candidate_bytes)
107def register(name: str, ip: str) -> Registration:
108 """Mint a fresh runner identity and persist its soft state (spec §7.1).
110 Returns the runner_id and the plaintext token (shown once). Only the token's
111 SHA-256 hex is stored. Also sweeps expired identities before registering.
112 """
113 now = _now()
114 _gc(now)
116 runner_id = RUNNER_ID_PREFIX + str(ULID())
117 token = RUNNER_TOKEN_PREFIX + secrets.token_urlsafe(32)
119 client = _redis()
120 ttl = params.IDENTITY_TTL_SEC
121 meta_key = redis_keys.runner_meta(runner_id)
123 # Identity creation is deliberately all-or-nothing (MULTI/EXEC), unlike the
124 # non-transactional pipelines in _gc/list_runners.
125 pipe = client.pipeline()
126 pipe.zadd(redis_keys.RUNNERS_REGISTERED, {runner_id: now})
127 pipe.hset(
128 meta_key,
129 mapping={
130 'name': name,
131 'registered_at': repr(now),
132 'registration_ip': ip,
133 },
134 )
135 pipe.expire(meta_key, ttl)
136 pipe.set(redis_keys.runner_token_hash(runner_id),
137 _token_hash(token),
138 ex=ttl)
139 pipe.execute()
141 return Registration(runner_id=runner_id, token=token)
144def verify_token(runner_id: Optional[str], token: Optional[str]) -> bool:
145 """Constant-time check that ``token`` matches the stored hash for ``runner_id``.
147 A missing token_hash key (revoked or expired) yields ``False`` (→ 401).
148 """
149 # Reject non-str inputs from the trust boundary: runner_id feeds a Redis key
150 # and token feeds _token_hash().encode() — both would otherwise raise.
151 if not isinstance(runner_id, str) or not isinstance(token, str):
152 return False
153 if not runner_id or not token:
154 return False
155 # Fail closed before touching Redis on a lone-surrogate runner_id/token whose
156 # UTF-8 encode raises (redis-py encodes keys to UTF-8; _token_hash encodes
157 # the token) — a hostile body must 401, not 500.
158 if _utf8(runner_id) is None:
159 return False
160 token_bytes = _utf8(token)
161 if token_bytes is None:
162 return False
163 stored = _redis().get(redis_keys.runner_token_hash(runner_id))
164 if stored is None:
165 return False
166 # Compare raw bytes: no assumption the stored value decodes as UTF-8.
167 return hmac.compare_digest(
168 stored,
169 hashlib.sha256(token_bytes).hexdigest().encode())
172def list_runners() -> List[Dict]:
173 """Return identity-layer facts for all registered identities (spec §7.6 subset).
175 Not a liveness view: a revoked or dead runner stays listed (frozen
176 last_seen) until identity GC sweeps it — a deliberate observability
177 window; revocation only guarantees immediate auth failure (ADR-0004).
178 Sweeps expired identities first. Fields: runner_id, name, last_seen,
179 registered_at. Liveness/held-jobs are added by the admin-API slice.
180 """
181 now = _now()
182 _gc(now)
184 client = _redis()
185 members = client.zrange(redis_keys.RUNNERS_REGISTERED,
186 0,
187 -1,
188 withscores=True)
190 runner_ids = [
191 member.decode() if isinstance(member, bytes) else member
192 for member, _ in members
193 ]
195 meta_pipe = client.pipeline(transaction=False)
196 for runner_id in runner_ids:
197 meta_pipe.hgetall(redis_keys.runner_meta(runner_id))
198 raw_metas = meta_pipe.execute()
200 runners: List[Dict] = []
201 for (_, score), runner_id, raw_meta in zip(members, runner_ids, raw_metas):
202 meta = {
203 (k.decode() if isinstance(k, bytes) else k):
204 (v.decode() if isinstance(v, bytes) else v)
205 for k, v in raw_meta.items()
206 }
207 runners.append({
208 'runner_id': runner_id,
209 'name': meta.get('name'),
210 'last_seen': score,
211 'registered_at': meta.get('registered_at'),
212 })
213 return runners
216def _gc(now: Optional[float] = None) -> None:
217 """Sweep ZSET corpses whose identity keys have already evaporated via TTL.
219 TTL is the sole invalidator: GC only removes a member once its
220 ``token_hash`` is already gone. The score prefilter (>7d stale) just narrows
221 the candidate set; any candidate whose ``token_hash`` still exists is skipped
222 untouched — its TTL has not fired, so touching it is exactly what caused the
223 TOCTOU (a heartbeat renewing the key between scan and delete).
225 Race-freedom without atomicity: once a ``token_hash`` is gone it can never
226 reappear for the same ``rn_id`` — register always mints a fresh ULID, and
227 heartbeat (future) requires token auth that fails without the key. So an
228 ``EXISTS == 0`` observation stays true, making the sweep safe.
229 """
230 if now is None:
231 now = _now()
232 cutoff = now - params.IDENTITY_TTL_SEC
234 client = _redis()
235 # Strictly older than the cutoff: '(' makes the max bound exclusive.
236 candidates = client.zrangebyscore(
237 redis_keys.RUNNERS_REGISTERED,
238 '-inf',
239 f'({cutoff!r}',
240 )
241 if not candidates:
242 return
244 runner_ids = [
245 member.decode() if isinstance(member, bytes) else member
246 for member in candidates
247 ]
249 # Only sweep members whose token_hash has already expired (TTL fired).
250 exists_pipe = client.pipeline(transaction=False)
251 for runner_id in runner_ids:
252 exists_pipe.exists(redis_keys.runner_token_hash(runner_id))
253 token_hash_exists = exists_pipe.execute()
255 sweep = [
256 runner_id
257 for runner_id, has_token in zip(runner_ids, token_hash_exists)
258 if not has_token
259 ]
260 if not sweep:
261 return
263 pipe = client.pipeline(transaction=False)
264 for runner_id in sweep:
265 pipe.zrem(redis_keys.RUNNERS_REGISTERED, runner_id)
266 pipe.delete(redis_keys.runner_meta(runner_id))
267 pipe.delete(redis_keys.runner_alive(runner_id))
268 pipe.execute()