Skip to content

lablink_allocator_service.secret_hash

Argon2 hashing for at-rest deployment/client secrets (SR-F14).

Thread-safe TTL + LRU cache with an optional per-key version guard.

Both caches in this module are instances of this: the client_secret_hash cache (keyed by hostname, distinct positive and negative TTLs, version guard in use — see VmDatabase.get_client_secret_hash) and the verify-result cache (keyed by (subject, token-fingerprint), successes only, no version guard).

negative_ttl applies to a stored None and defaults to ttl. The secret-hash cache sets it shorter so a freshly registered host becomes auth-able within seconds without a register-time invalidate, and so a repeatedly probed unknown hostname doesn't linger.

Race-against-rotation: :meth:get returns a per-key version token that :meth:put re-checks under the lock. If :meth:invalidate ran in between — a concurrent register_client rotating the secret while a stale SELECT was in flight — the mismatch rejects the stale write, so the next call re-queries and picks up the new value. Pass expected_version=None to skip the check; the verify cache has nothing to race with, since it stores a result it derived itself rather than a value it fetched.

Bounded size: an OrderedDict with LRU eviction on insert beyond max_size. Both get and put move the entry to the most-recently-used end. The bound is a DoS guard against unique-key floods, not a working-set sizing knob.

Source code in packages/allocator/src/lablink_allocator_service/secret_hash.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def __init__(
    self,
    *,
    ttl: float,
    max_size: int,
    negative_ttl: float | None = None,
):
    if max_size < 1:
        raise ValueError(f"Invalid cache max_size: {max_size}")
    self._ttl = ttl
    self._negative_ttl = ttl if negative_ttl is None else negative_ttl
    self._max_size = max_size
    self._lock = threading.RLock()
    # key -> (value, expires_at) on the monotonic clock
    self._entries: OrderedDict = OrderedDict()
    # key -> int. Bumped only by invalidate(); never reset by TTL
    # expiry or LRU eviction (those aren't "the value changed"
    # events). put() rejects stores whose observed version is stale.
    self._versions: dict = {}
Source code in packages/allocator/src/lablink_allocator_service/secret_hash.py 
182
183
184
185
def clear(self) -> None:
    with self._lock:
        self._entries.clear()
        self._versions.clear()

Return (hit, value, version).

hit=False means the caller must re-fetch; pass version back to :meth:put so a concurrent invalidate between get and put rejects the stale write.

Source code in packages/allocator/src/lablink_allocator_service/secret_hash.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def get(self, key):
    """Return ``(hit, value, version)``.

    ``hit=False`` means the caller must re-fetch; pass ``version``
    back to :meth:`put` so a concurrent invalidate between get and
    put rejects the stale write.
    """
    with self._lock:
        version = self._versions.get(key, 0)
        entry = self._entries.get(key)
        if entry is None:
            return False, None, version
        value, expires_at = entry
        if time.monotonic() >= expires_at:
            del self._entries[key]
            return False, None, version
        # LRU touch
        self._entries.move_to_end(key)
        return True, value, version
Source code in packages/allocator/src/lablink_allocator_service/secret_hash.py 
166
167
168
169
def invalidate(self, key) -> None:
    with self._lock:
        self._versions[key] = self._versions.get(key, 0) + 1
        self._entries.pop(key, None)

Drop every entry whose key satisfies predicate.

Unlike :meth:invalidate this bumps no version — it exists for the verify cache, whose compound key means one subject owns an unbounded set of keys and which runs no version guard anyway.

Source code in packages/allocator/src/lablink_allocator_service/secret_hash.py 
171
172
173
174
175
176
177
178
179
180
def invalidate_where(self, predicate) -> None:
    """Drop every entry whose key satisfies ``predicate``.

    Unlike :meth:`invalidate` this bumps no version — it exists for
    the verify cache, whose compound key means one subject owns an
    unbounded set of keys and which runs no version guard anyway.
    """
    with self._lock:
        for key in [k for k in self._entries if predicate(k)]:
            del self._entries[key]

Store value for key if no invalidate raced.

Returns True if the entry was written, False if a concurrent :meth:invalidate bumped the version between the caller's :meth:get and this put.

Source code in packages/allocator/src/lablink_allocator_service/secret_hash.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def put(self, key, value, expected_version: int | None = None) -> bool:
    """Store ``value`` for ``key`` if no invalidate raced.

    Returns ``True`` if the entry was written, ``False`` if a
    concurrent :meth:`invalidate` bumped the version between the
    caller's :meth:`get` and this ``put``.
    """
    ttl = self._ttl if value is not None else self._negative_ttl
    with self._lock:
        if (
            expected_version is not None
            and self._versions.get(key, 0) != expected_version
        ):
            return False
        self._entries[key] = (value, time.monotonic() + ttl)
        self._entries.move_to_end(key)
        while len(self._entries) > self._max_size:
            self._entries.popitem(last=False)
        return True

Test helper: drop every cached entry.

Source code in packages/allocator/src/lablink_allocator_service/secret_hash.py 
234
235
236
def clear_verify_cache() -> None:
    """Test helper: drop every cached entry."""
    _verify_cache.clear()

Return an argon2 hash string for plaintext (salted, unique per call).

Source code in packages/allocator/src/lablink_allocator_service/secret_hash.py 
15
16
17
def hash_secret(plaintext: str) -> str:
    """Return an argon2 hash string for `plaintext` (salted, unique per call)."""
    return _ph.hash(plaintext)

Drop every cached verify result for subject.

Call whenever the stored hash for a subject changes — e.g., on register_client (rotation) or unregister_client (removal), the same hook the secret-hash cache uses, so a rotated secret does not keep accepting old tokens up to the TTL.

Source code in packages/allocator/src/lablink_allocator_service/secret_hash.py 
223
224
225
226
227
228
229
230
231
def invalidate_verify(subject: str) -> None:
    """Drop every cached verify result for ``subject``.

    Call whenever the stored hash for a subject changes — e.g., on
    ``register_client`` (rotation) or ``unregister_client`` (removal),
    the same hook the secret-hash cache uses, so a rotated secret does
    not keep accepting old tokens up to the TTL.
    """
    _verify_cache.invalidate_where(lambda k: k[0] == subject)

True iff plaintext matches hashed. False on any mismatch or malformed hash (never raises).

Source code in packages/allocator/src/lablink_allocator_service/secret_hash.py 
20
21
22
23
24
25
26
def verify_secret(plaintext: str, hashed: str) -> bool:
    """True iff `plaintext` matches `hashed`. False on any mismatch or
    malformed hash (never raises)."""
    try:
        return _ph.verify(hashed, plaintext)
    except (VerifyMismatchError, VerificationError, InvalidHashError, TypeError):
        return False

Same contract as :func:verify_secret but memoizes successes by (subject, sha256(plaintext)) for ~60 s.

The first call pays the argon2 verify cost; subsequent calls with the same (subject, plaintext) within the TTL return True without running argon2. Failures are never cached, so a wrong token always pays the verify cost — there's no fast path for attackers, and a wrong token cannot poison an entry either: it has a different fingerprint, so it lands on a different key.

Use a subject the caller has already identified out-of-band (e.g., hostname from the request body, client_id from the URL), so a cache hit doesn't silently accept a token across identities.

Source code in packages/allocator/src/lablink_allocator_service/secret_hash.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def verify_secret_cached(subject: str, plaintext: str, hashed: str) -> bool:
    """Same contract as :func:`verify_secret` but memoizes successes by
    ``(subject, sha256(plaintext))`` for ~60 s.

    The first call pays the argon2 verify cost; subsequent calls with
    the same ``(subject, plaintext)`` within the TTL return True without
    running argon2. Failures are never cached, so a wrong token always
    pays the verify cost — there's no fast path for attackers, and a
    wrong token cannot poison an entry either: it has a different
    fingerprint, so it lands on a different key.

    Use a ``subject`` the caller has already identified out-of-band
    (e.g., hostname from the request body, client_id from the URL), so
    a cache hit doesn't silently accept a token across identities.
    """
    key = (subject, _token_fingerprint(plaintext))
    hit, _, _ = _verify_cache.get(key)
    if hit:
        return True
    if verify_secret(plaintext, hashed):
        _verify_cache.put(key, True)
        return True
    return False