1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
|
"""
A hub that keeps the rules the CLI depends on, in memory.
Its refresh tokens **rotate, and a spent one presented again revokes the
whole family**, as the real hub does. A lax stub would pass a client that
renews twice with the same token — which is exactly the failure two runs at
once produce.
"""
import asyncio
import base64
import secrets
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from meshbay_cli.client.hub import HubError
from meshbay_cli.client.keys import auth_key
USERNAME = "alice-test"
PASSPHRASE = "a passphrase for the fake hub"
USER_ID = "0b4f6f0e-5d7e-4e8a-9d2b-6a1c1b9e2f11"
PEPPER = base64.b64encode(b"\x07" * 32).decode()
class FakeHub:
def __init__(self, url: str = "https://hub.test"):
self.url = url
self.ak = auth_key(USERNAME, PASSPHRASE)
self.refresh_tokens: dict[str, dict] = {} # token → {family, spent}
self.revoked_families: set[str] = set()
self.access: set[str] = set()
self.devices: dict[str, str] = {} # id → pk
self.refresh_calls = 0
self.reuse_detected = False
self.refresh_delay = 0.0
self.groups = [{"id": "g1", "name": "Family", "owner_username": USERNAME,
"node_online": True, "visibility": "private", "is_admin": True}]
def _session(self, family: str | None = None) -> dict:
family = family or secrets.token_hex(4)
rt, at = secrets.token_hex(8), secrets.token_hex(8)
self.refresh_tokens[rt] = {"family": family, "spent": False}
self.access.add(at)
return {"access_token": at, "refresh_token": rt, "expires_in": 14400}
def _auth(self, token: str) -> None:
if token not in self.access:
raise HubError(401, "Invalid token")
async def aclose(self) -> None:
pass
async def check_version(self) -> None:
return None
async def login(self, username, ak, known_browser):
from meshbay_cli.client.hub import _sign_in_refusal
if username != USERNAME or ak != self.ak:
raise _sign_in_refusal(HubError(401, "Invalid credentials"))
return {**self._session(), "bundle_pepper": PEPPER, "bundle_pepper_version": 1,
"known_browser": "kb-1"}
async def me(self, token):
self._auth(token)
return {"user_id": USER_ID, "username": USERNAME}
async def register_device(self, token, pk, label, ak):
self._auth(token)
assert ak == self.ak, "registering a device takes the passphrase"
for did, known in self.devices.items():
if known == pk:
return {"id": did, "existing": True}
did = f"dev-{len(self.devices) + 1}"
self.devices[did] = pk
return {"id": did, "existing": False}
async def delete_device(self, token, device_id):
self._auth(token)
self.devices.pop(device_id)
async def device_auth(self, username, sk):
pk = base64.b64encode(sk.public_key().public_bytes_raw()).decode()
if pk not in self.devices.values():
from meshbay_cli.errors import AUTH, CliError
raise CliError(AUTH, "Sign-in refused")
Ed25519PublicKey.from_public_bytes(sk.public_key().public_bytes_raw())
return {**self._session(), "bundle_pepper": PEPPER}
async def refresh(self, rt):
self.refresh_calls += 1
row = self.refresh_tokens.get(rt)
if row is None or row["family"] in self.revoked_families:
raise HubError(401, "Invalid refresh token")
if row["spent"]:
self.reuse_detected = True
self.revoked_families.add(row["family"])
raise HubError(401, "Token reuse detected — family revoked")
row["spent"] = True
await asyncio.sleep(self.refresh_delay)
return self._session(row["family"])
async def logout(self, rt):
row = self.refresh_tokens.get(rt)
if row:
self.revoked_families.add(row["family"])
async def my_groups(self, token):
self._auth(token)
return self.groups
|