+161

tests/test_pairing_devices.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + from __future__ import annotations 5 5 + 6 6 + import json 7 7 + 8 8 + from think.pairing import devices 9 9 + 10 10 + 11 11 + def _devices_path(journal_copy): 12 12 + return journal_copy / "config" / "paired_devices.json" 13 13 + 14 14 + 15 15 + def test_load_devices_returns_empty_for_missing_store(journal_copy): 16 16 + assert devices.load_devices() == [] 17 17 + 18 18 + 19 19 + def test_register_load_and_remove_round_trip(journal_copy): 20 20 + device = devices.register_device( 21 21 + name="Phone", 22 22 + platform="ios", 23 23 + public_key="ssh-ed25519 AAAAphone", 24 24 + session_key_hash="sha256:abc", 25 25 + bundle_id="org.solpbc.solstone-swift", 26 26 + app_version="0.1.0", 27 27 + paired_at="2026-04-20T15:31:02Z", 28 28 + ) 29 29 + 30 30 + loaded = devices.load_devices() 31 31 + 32 32 + assert loaded == [ 33 33 + { 34 34 + "id": device["id"], 35 35 + "name": "Phone", 36 36 + "platform": "ios", 37 37 + "public_key": "ssh-ed25519 AAAAphone", 38 38 + "session_key_hash": "sha256:abc", 39 39 + "bundle_id": "org.solpbc.solstone-swift", 40 40 + "app_version": "0.1.0", 41 41 + "paired_at": "2026-04-20T15:31:02Z", 42 42 + "last_seen_at": None, 43 43 + } 44 44 + ] 45 45 + assert devices.find_device_by_id(device["id"]) == loaded[0] 46 46 + assert devices.find_device_by_session_key_hash("sha256:abc") == loaded[0] 47 47 + assert devices.remove_device(device["id"]) is True 48 48 + assert devices.load_devices() == [] 49 49 + 50 50 + 51 51 + def test_register_device_upserts_by_public_key(journal_copy): 52 52 + first = devices.register_device( 53 53 + name="Phone", 54 54 + platform="ios", 55 55 + public_key="ssh-ed25519 AAAAsame", 56 56 + session_key_hash="sha256:first", 57 57 + bundle_id="org.solpbc.solstone-swift", 58 58 + app_version="0.1.0", 59 59 + paired_at="2026-04-20T15:31:02Z", 60 60 + ) 61 61 + second = devices.register_device( 62 62 + name="Phone 2", 63 63 + platform="ios", 64 64 + public_key="ssh-ed25519 AAAAsame", 65 65 + session_key_hash="sha256:second", 66 66 + bundle_id="org.solpbc.solstone-swift", 67 67 + app_version="0.2.0", 68 68 + paired_at="2026-04-20T16:00:00Z", 69 69 + ) 70 70 + 71 71 + assert first["id"] == second["id"] 72 72 + assert devices.load_devices() == [ 73 73 + { 74 74 + "id": first["id"], 75 75 + "name": "Phone 2", 76 76 + "platform": "ios", 77 77 + "public_key": "ssh-ed25519 AAAAsame", 78 78 + "session_key_hash": "sha256:second", 79 79 + "bundle_id": "org.solpbc.solstone-swift", 80 80 + "app_version": "0.2.0", 81 81 + "paired_at": "2026-04-20T16:00:00Z", 82 82 + "last_seen_at": None, 83 83 + } 84 84 + ] 85 85 + 86 86 + 87 87 + def test_touch_last_seen_updates_existing_device(journal_copy): 88 88 + device = devices.register_device( 89 89 + name="Phone", 90 90 + platform="ios", 91 91 + public_key="ssh-ed25519 AAAAseen", 92 92 + session_key_hash="sha256:seen", 93 93 + bundle_id="org.solpbc.solstone-swift", 94 94 + app_version="0.1.0", 95 95 + paired_at="2026-04-20T15:31:02Z", 96 96 + ) 97 97 + 98 98 + touched = devices.touch_last_seen(device["id"], last_seen_at="2026-04-20T16:01:00Z") 99 99 + 100 100 + assert touched is True 101 101 + assert devices.find_device_by_id(device["id"]) is not None 102 102 + assert ( 103 103 + devices.find_device_by_id(device["id"])["last_seen_at"] 104 104 + == "2026-04-20T16:01:00Z" 105 105 + ) 106 106 + 107 107 + 108 108 + def test_load_devices_recovers_from_malformed_store(journal_copy, caplog): 109 109 + path = _devices_path(journal_copy) 110 110 + path.parent.mkdir(parents=True, exist_ok=True) 111 111 + path.write_text('{"devices": "bad"}', encoding="utf-8") 112 112 + 113 113 + assert devices.load_devices() == [] 114 114 + assert "paired device store unreadable" in caplog.text 115 115 + 116 116 + healed = devices.register_device( 117 117 + name="Phone", 118 118 + platform="ios", 119 119 + public_key="ssh-ed25519 AAAAheal", 120 120 + session_key_hash="sha256:heal", 121 121 + bundle_id="org.solpbc.solstone-swift", 122 122 + app_version="0.1.0", 123 123 + paired_at="2026-04-20T15:31:02Z", 124 124 + ) 125 125 + payload = json.loads(path.read_text(encoding="utf-8")) 126 126 + 127 127 + assert payload == { 128 128 + "devices": [ 129 129 + { 130 130 + "id": healed["id"], 131 131 + "name": "Phone", 132 132 + "platform": "ios", 133 133 + "public_key": "ssh-ed25519 AAAAheal", 134 134 + "session_key_hash": "sha256:heal", 135 135 + "bundle_id": "org.solpbc.solstone-swift", 136 136 + "app_version": "0.1.0", 137 137 + "paired_at": "2026-04-20T15:31:02Z", 138 138 + "last_seen_at": None, 139 139 + } 140 140 + ] 141 141 + } 142 142 + 143 143 + 144 144 + def test_status_view_redacts_secret_fields(journal_copy): 145 145 + device = devices.register_device( 146 146 + name="Phone", 147 147 + platform="ios", 148 148 + public_key="ssh-ed25519 AAAAredact", 149 149 + session_key_hash="sha256:redact", 150 150 + bundle_id="org.solpbc.solstone-swift", 151 151 + app_version="0.1.0", 152 152 + paired_at="2026-04-20T15:31:02Z", 153 153 + ) 154 154 + 155 155 + assert devices.status_view(device) == { 156 156 + "id": device["id"], 157 157 + "name": "Phone", 158 158 + "platform": "ios", 159 159 + "paired_at": "2026-04-20T15:31:02Z", 160 160 + "last_seen_at": None, 161 161 + }

+63

tests/test_pairing_keys.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + from __future__ import annotations 5 5 + 6 6 + from cryptography.hazmat.primitives import serialization 7 7 + from cryptography.hazmat.primitives.asymmetric import ec, ed25519, rsa 8 8 + 9 9 + from think.pairing import keys 10 10 + 11 11 + 12 12 + def _openssh_public_key(public_key) -> str: 13 13 + return public_key.public_bytes( 14 14 + encoding=serialization.Encoding.OpenSSH, 15 15 + format=serialization.PublicFormat.OpenSSH, 16 16 + ).decode("utf-8") 17 17 + 18 18 + 19 19 + def test_validate_public_key_accepts_ssh_ed25519(): 20 20 + key = ed25519.Ed25519PrivateKey.generate().public_key() 21 21 + encoded = _openssh_public_key(key) 22 22 + 23 23 + assert keys.validate_public_key(encoded) == encoded 24 24 + 25 25 + 26 26 + def test_validate_public_key_rejects_non_ed25519_algorithms(): 27 27 + rsa_key = rsa.generate_private_key( 28 28 + public_exponent=65537, key_size=2048 29 29 + ).public_key() 30 30 + ecdsa_key = ec.generate_private_key(ec.SECP256R1()).public_key() 31 31 + 32 32 + for candidate in (_openssh_public_key(rsa_key), _openssh_public_key(ecdsa_key)): 33 33 + try: 34 34 + keys.validate_public_key(candidate) 35 35 + except ValueError as exc: 36 36 + assert str(exc) == "public key must be ssh-ed25519" 37 37 + else: 38 38 + raise AssertionError("expected non-ed25519 key to be rejected") 39 39 + 40 40 + 41 41 + def test_validate_public_key_rejects_malformed_and_oversized_values(): 42 42 + for candidate, expected in ( 43 43 + ("ssh-ed25519 AAAA-not-valid", "public key is invalid"), 44 44 + ("ssh-ed25519 " + ("A" * 2049), "public key is too long"), 45 45 + ): 46 46 + try: 47 47 + keys.validate_public_key(candidate) 48 48 + except ValueError as exc: 49 49 + assert str(exc) == expected 50 50 + else: 51 51 + raise AssertionError("expected invalid public key to be rejected") 52 52 + 53 53 + 54 54 + def test_session_key_helpers(): 55 55 + session_key = keys.generate_session_key() 56 56 + session_hash = keys.hash_session_key(session_key) 57 57 + 58 58 + assert session_key.startswith("dsk_") 59 59 + assert session_hash.startswith("sha256:") 60 60 + assert len(session_hash) == len("sha256:") + 64 61 61 + assert keys.mask_session_key(session_key) == ( 62 62 + f"...{session_key[-4:]} (len={len(session_key)})" 63 63 + )

+65

tests/test_pairing_tokens.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + from __future__ import annotations 5 5 + 6 6 + from think.pairing import tokens 7 7 + 8 8 + 9 9 + def setup_function() -> None: 10 10 + tokens._TOKENS.clear() 11 11 + 12 12 + 13 13 + def test_create_token_uses_expected_shape_and_metadata(): 14 14 + token = tokens.create_token(ttl_seconds=600, now=1000) 15 15 + 16 16 + assert token.token.startswith("ptk_") 17 17 + assert token.issued_at == 1000 18 18 + assert token.expires_at == 1600 19 19 + assert token.ttl_seconds == 600 20 20 + assert token.consumed_at is None 21 21 + 22 22 + 23 23 + def test_create_token_clamps_ttl(): 24 24 + low = tokens.create_token(ttl_seconds=1, now=1000) 25 25 + high = tokens.create_token(ttl_seconds=9999, now=1000) 26 26 + 27 27 + assert low.ttl_seconds == 60 28 28 + assert low.expires_at == 1060 29 29 + assert high.ttl_seconds == 3600 30 30 + assert high.expires_at == 4600 31 31 + 32 32 + 33 33 + def test_consume_token_marks_token_used_once(): 34 34 + created = tokens.create_token(ttl_seconds=600, now=1000) 35 35 + 36 36 + first = tokens.consume_token(created.token, now=1100) 37 37 + second = tokens.consume_token(created.token, now=1101) 38 38 + peeked = tokens.peek_token(created.token, now=1101) 39 39 + 40 40 + assert first is not None 41 41 + assert first.consumed_at == 1100 42 42 + assert second is None 43 43 + assert peeked is not None 44 44 + assert peeked.consumed_at == 1100 45 45 + 46 46 + 47 47 + def test_consume_token_rejects_expired_token(): 48 48 + created = tokens.create_token(ttl_seconds=60, now=1000) 49 49 + 50 50 + assert tokens.consume_token(created.token, now=1060) is None 51 51 + expired = tokens.peek_token(created.token, now=1060) 52 52 + assert expired is not None 53 53 + assert expired.expires_at == 1060 54 54 + assert expired.consumed_at is None 55 55 + 56 56 + 57 57 + def test_purge_expired_tokens_removes_only_expired_entries(): 58 58 + keep = tokens.create_token(ttl_seconds=600, now=1000) 59 59 + drop = tokens.create_token(ttl_seconds=60, now=1000) 60 60 + 61 61 + purged = tokens.purge_expired_tokens(now=1060) 62 62 + 63 63 + assert purged == 1 64 64 + assert tokens.peek_token(drop.token, now=1060) is None 65 65 + assert tokens.peek_token(keep.token, now=1060) is not None

+213

think/pairing/devices.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + """Paired-device storage.""" 5 5 + 6 6 + from __future__ import annotations 7 7 + 8 8 + import json 9 9 + import logging 10 10 + import secrets 11 11 + from datetime import datetime, timezone 12 12 + from pathlib import Path 13 13 + from typing import Any, TypedDict 14 14 + 15 15 + from think.entities.core import atomic_write 16 16 + from think.utils import get_journal 17 17 + 18 18 + logger = logging.getLogger(__name__) 19 19 + 20 20 + 21 21 + class Device(TypedDict): 22 22 + id: str 23 23 + name: str 24 24 + platform: str 25 25 + public_key: str 26 26 + session_key_hash: str 27 27 + bundle_id: str 28 28 + app_version: str 29 29 + paired_at: str 30 30 + last_seen_at: str | None 31 31 + 32 32 + 33 33 + def _devices_path() -> Path: 34 34 + return Path(get_journal()) / "config" / "paired_devices.json" 35 35 + 36 36 + 37 37 + def _utc_now_iso() -> str: 38 38 + return ( 39 39 + datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") 40 40 + ) 41 41 + 42 42 + 43 43 + def _empty_store() -> dict[str, list[Device]]: 44 44 + return {"devices": []} 45 45 + 46 46 + 47 47 + def _clean_str(value: Any) -> str: 48 48 + return str(value or "").strip() 49 49 + 50 50 + 51 51 + def _validate_timestamp(value: Any) -> str | None: 52 52 + if value is None: 53 53 + return None 54 54 + if not isinstance(value, str) or not value.strip(): 55 55 + raise ValueError("device timestamp fields must be strings or null") 56 56 + return value 57 57 + 58 58 + 59 59 + def _validate_store(payload: Any) -> list[Device]: 60 60 + if not isinstance(payload, dict): 61 61 + raise ValueError("paired device store must be a JSON object") 62 62 + devices = payload.get("devices") 63 63 + if not isinstance(devices, list): 64 64 + raise ValueError("paired device store must contain a devices list") 65 65 + normalized: list[Device] = [] 66 66 + for device in devices: 67 67 + if not isinstance(device, dict): 68 68 + raise ValueError("paired device rows must be JSON objects") 69 69 + normalized.append( 70 70 + Device( 71 71 + id=_require_field(device, "id"), 72 72 + name=_require_field(device, "name"), 73 73 + platform=_require_field(device, "platform"), 74 74 + public_key=_require_field(device, "public_key"), 75 75 + session_key_hash=_require_field(device, "session_key_hash"), 76 76 + bundle_id=_require_field(device, "bundle_id"), 77 77 + app_version=_require_field(device, "app_version"), 78 78 + paired_at=_require_field(device, "paired_at"), 79 79 + last_seen_at=_validate_timestamp(device.get("last_seen_at")), 80 80 + ) 81 81 + ) 82 82 + return normalized 83 83 + 84 84 + 85 85 + def _require_field(device: dict[str, Any], field: str) -> str: 86 86 + value = _clean_str(device.get(field)) 87 87 + if not value: 88 88 + raise ValueError(f"paired device row missing required field: {field}") 89 89 + return value 90 90 + 91 91 + 92 92 + def _read_store() -> list[Device]: 93 93 + path = _devices_path() 94 94 + if not path.exists(): 95 95 + return [] 96 96 + try: 97 97 + payload = json.loads(path.read_text(encoding="utf-8")) 98 98 + return _validate_store(payload) 99 99 + except Exception as exc: 100 100 + logger.warning("paired device store unreadable path=%s error=%s", path, exc) 101 101 + return [] 102 102 + 103 103 + 104 104 + def _write_store(devices: list[Device]) -> None: 105 105 + payload = json.dumps({"devices": devices}, indent=2, ensure_ascii=False) + "\n" 106 106 + atomic_write(_devices_path(), payload, prefix=".paired_devices_") 107 107 + 108 108 + 109 109 + def load_devices() -> list[Device]: 110 110 + return _read_store() 111 111 + 112 112 + 113 113 + def find_device_by_id(device_id: str) -> Device | None: 114 114 + target = _clean_str(device_id) 115 115 + if not target: 116 116 + return None 117 117 + for device in load_devices(): 118 118 + if device["id"] == target: 119 119 + return device 120 120 + return None 121 121 + 122 122 + 123 123 + def find_device_by_session_key_hash(session_key_hash: str) -> Device | None: 124 124 + target = _clean_str(session_key_hash) 125 125 + if not target: 126 126 + return None 127 127 + for device in load_devices(): 128 128 + if device["session_key_hash"] == target: 129 129 + return device 130 130 + return None 131 131 + 132 132 + 133 133 + def register_device( 134 134 + *, 135 135 + name: str, 136 136 + platform: str, 137 137 + public_key: str, 138 138 + session_key_hash: str, 139 139 + bundle_id: str, 140 140 + app_version: str, 141 141 + paired_at: str | None = None, 142 142 + ) -> Device: 143 143 + devices = load_devices() 144 144 + row: Device = Device( 145 145 + id=f"dev_{secrets.token_urlsafe(16)}", 146 146 + name=_clean_str(name), 147 147 + platform=_clean_str(platform), 148 148 + public_key=_clean_str(public_key), 149 149 + session_key_hash=_clean_str(session_key_hash), 150 150 + bundle_id=_clean_str(bundle_id), 151 151 + app_version=_clean_str(app_version), 152 152 + paired_at=_clean_str(paired_at) or _utc_now_iso(), 153 153 + last_seen_at=None, 154 154 + ) 155 155 + for index, device in enumerate(devices): 156 156 + if device["public_key"] != row["public_key"]: 157 157 + continue 158 158 + row["id"] = device["id"] 159 159 + devices[index] = row 160 160 + _write_store(devices) 161 161 + return row 162 162 + devices.append(row) 163 163 + _write_store(devices) 164 164 + return row 165 165 + 166 166 + 167 167 + def touch_last_seen(device_id: str, *, last_seen_at: str | None = None) -> bool: 168 168 + target = _clean_str(device_id) 169 169 + if not target: 170 170 + return False 171 171 + devices = load_devices() 172 172 + timestamp = _clean_str(last_seen_at) or _utc_now_iso() 173 173 + for device in devices: 174 174 + if device["id"] != target: 175 175 + continue 176 176 + device["last_seen_at"] = timestamp 177 177 + _write_store(devices) 178 178 + return True 179 179 + return False 180 180 + 181 181 + 182 182 + def remove_device(device_id: str) -> bool: 183 183 + target = _clean_str(device_id) 184 184 + if not target: 185 185 + return False 186 186 + devices = load_devices() 187 187 + remaining = [device for device in devices if device["id"] != target] 188 188 + if len(remaining) == len(devices): 189 189 + return False 190 190 + _write_store(remaining) 191 191 + return True 192 192 + 193 193 + 194 194 + def status_view(device: Device) -> dict[str, Any]: 195 195 + return { 196 196 + "id": device["id"], 197 197 + "name": device["name"], 198 198 + "platform": device["platform"], 199 199 + "paired_at": device["paired_at"], 200 200 + "last_seen_at": device["last_seen_at"], 201 201 + } 202 202 + 203 203 + 204 204 + __all__ = [ 205 205 + "Device", 206 206 + "find_device_by_id", 207 207 + "find_device_by_session_key_hash", 208 208 + "load_devices", 209 209 + "register_device", 210 210 + "remove_device", 211 211 + "status_view", 212 212 + "touch_last_seen", 213 213 + ]

+55

think/pairing/keys.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + """Pairing key validation and session-key helpers.""" 5 5 + 6 6 + from __future__ import annotations 7 7 + 8 8 + import hashlib 9 9 + import secrets 10 10 + 11 11 + from cryptography.hazmat.primitives import serialization 12 12 + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey 13 13 + 14 14 + MAX_PUBLIC_KEY_LENGTH = 2048 15 15 + 16 16 + 17 17 + def validate_public_key(public_key: str) -> str: 18 18 + candidate = str(public_key or "").strip() 19 19 + if not candidate: 20 20 + raise ValueError("public key is required") 21 21 + if len(candidate) > MAX_PUBLIC_KEY_LENGTH: 22 22 + raise ValueError("public key is too long") 23 23 + try: 24 24 + parsed = serialization.load_ssh_public_key(candidate.encode("utf-8")) 25 25 + except Exception as exc: 26 26 + raise ValueError("public key is invalid") from exc 27 27 + if not isinstance(parsed, Ed25519PublicKey): 28 28 + raise ValueError("public key must be ssh-ed25519") 29 29 + return parsed.public_bytes( 30 30 + encoding=serialization.Encoding.OpenSSH, 31 31 + format=serialization.PublicFormat.OpenSSH, 32 32 + ).decode("utf-8") 33 33 + 34 34 + 35 35 + def generate_session_key() -> str: 36 36 + return f"dsk_{secrets.token_urlsafe(32)}" 37 37 + 38 38 + 39 39 + def hash_session_key(session_key: str) -> str: 40 40 + digest = hashlib.sha256(session_key.encode("utf-8")).hexdigest() 41 41 + return f"sha256:{digest}" 42 42 + 43 43 + 44 44 + def mask_session_key(session_key: str) -> str: 45 45 + value = str(session_key or "") 46 46 + return f"...{value[-4:]} (len={len(value)})" 47 47 + 48 48 + 49 49 + __all__ = [ 50 50 + "MAX_PUBLIC_KEY_LENGTH", 51 51 + "generate_session_key", 52 52 + "hash_session_key", 53 53 + "mask_session_key", 54 54 + "validate_public_key", 55 55 + ]

+105

think/pairing/tokens.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + """In-memory pairing token store.""" 5 5 + 6 6 + from __future__ import annotations 7 7 + 8 8 + import secrets 9 9 + import threading 10 10 + import time 11 11 + from dataclasses import dataclass, replace 12 12 + 13 13 + from think.pairing.config import ( 14 14 + MAX_TOKEN_TTL_SECONDS, 15 15 + MIN_TOKEN_TTL_SECONDS, 16 16 + get_token_ttl_seconds, 17 17 + ) 18 18 + 19 19 + 20 20 + @dataclass(frozen=True) 21 21 + class PairingToken: 22 22 + token: str 23 23 + issued_at: int 24 24 + expires_at: int 25 25 + ttl_seconds: int 26 26 + consumed_at: int | None 27 27 + 28 28 + 29 29 + _TOKENS: dict[str, PairingToken] = {} 30 30 + _TOKENS_LOCK = threading.Lock() 31 31 + 32 32 + 33 33 + def _now(now: int | None) -> int: 34 34 + return int(now if now is not None else time.time()) 35 35 + 36 36 + 37 37 + def _clamp_ttl(ttl_seconds: int) -> int: 38 38 + return max(MIN_TOKEN_TTL_SECONDS, min(MAX_TOKEN_TTL_SECONDS, ttl_seconds)) 39 39 + 40 40 + 41 41 + def _purge_expired_locked(now: int, *, exclude: str | None = None) -> int: 42 42 + expired = [ 43 43 + token 44 44 + for token, entry in _TOKENS.items() 45 45 + if token != exclude and entry.expires_at <= now 46 46 + ] 47 47 + for token in expired: 48 48 + del _TOKENS[token] 49 49 + return len(expired) 50 50 + 51 51 + 52 52 + def create_token( 53 53 + *, ttl_seconds: int | None = None, now: int | None = None 54 54 + ) -> PairingToken: 55 55 + ts = _now(now) 56 56 + effective_ttl = _clamp_ttl( 57 57 + get_token_ttl_seconds() if ttl_seconds is None else int(ttl_seconds) 58 58 + ) 59 59 + entry = PairingToken( 60 60 + token=f"ptk_{secrets.token_urlsafe(32)}", 61 61 + issued_at=ts, 62 62 + expires_at=ts + effective_ttl, 63 63 + ttl_seconds=effective_ttl, 64 64 + consumed_at=None, 65 65 + ) 66 66 + with _TOKENS_LOCK: 67 67 + _purge_expired_locked(ts) 68 68 + _TOKENS[entry.token] = entry 69 69 + return entry 70 70 + 71 71 + 72 72 + def consume_token(token: str, *, now: int | None = None) -> PairingToken | None: 73 73 + ts = _now(now) 74 74 + with _TOKENS_LOCK: 75 75 + _purge_expired_locked(ts, exclude=token) 76 76 + entry = _TOKENS.get(token) 77 77 + if entry is None: 78 78 + return None 79 79 + if entry.expires_at <= ts or entry.consumed_at is not None: 80 80 + return None 81 81 + consumed = replace(entry, consumed_at=ts) 82 82 + _TOKENS[token] = consumed 83 83 + return consumed 84 84 + 85 85 + 86 86 + def peek_token(token: str, *, now: int | None = None) -> PairingToken | None: 87 87 + ts = _now(now) 88 88 + with _TOKENS_LOCK: 89 89 + _purge_expired_locked(ts, exclude=token) 90 90 + return _TOKENS.get(token) 91 91 + 92 92 + 93 93 + def purge_expired_tokens(*, now: int | None = None) -> int: 94 94 + ts = _now(now) 95 95 + with _TOKENS_LOCK: 96 96 + return _purge_expired_locked(ts) 97 97 + 98 98 + 99 99 + __all__ = [ 100 100 + "PairingToken", 101 101 + "consume_token", 102 102 + "create_token", 103 103 + "peek_token", 104 104 + "purge_expired_tokens", 105 105 + ]