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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
|
"""
How long a session lasts, and how it ends (design §7.7).
Three admin settings in hours — browser idle sign-out, refresh idle window,
maximum — plus a sign-out the hub honours and "sign out everywhere". The
browser half is `test_browser_idle_signout.py`; this is the hub's.
"""
from datetime import UTC, datetime, timedelta
import pytest
from meshbay_hub.api.deps import set_admin_usernames
from meshbay_hub.db.models import RefreshToken, User
from sqlalchemy import select, update
AUTH_KEY = "s" * 44
async def _register(client, username):
r = await client.post("/v1/users/register", json={
"username": username, "email": f"{username}@example.com", "auth_key": AUTH_KEY})
assert r.status_code == 201, r.text
async def _login(client, username):
r = await client.post("/v1/users/login", json={"username": username, "auth_key": AUTH_KEY})
assert r.status_code == 200, r.text
return r.json()
async def _admin(client):
await _register(client, "session_admin")
set_admin_usernames(["session_admin"])
return {"Authorization": f"Bearer {(await _login(client, 'session_admin'))['access_token']}"}
async def _refresh(client, token):
return await client.post("/v1/users/token/refresh", json={"refresh_token": token})
def _aware(dt):
return dt if dt.tzinfo else dt.replace(tzinfo=UTC)
async def _expiry(db_session, username):
uid = (await db_session.execute(
select(User.id).where(User.username == username))).scalar_one()
rows = (await db_session.execute(
select(RefreshToken.expires_at).where(RefreshToken.user_id == uid))).scalars().all()
return _aware(max(rows))
@pytest.mark.asyncio
async def test_the_defaults_are_in_the_panel_and_the_browser_delay_is_public(client):
admin = await _admin(client)
r = await client.get("/v1/admin/settings", headers=admin)
assert r.json()["session"] == {
"browser_idle_hours": 1, "refresh_idle_hours": 24, "max_hours": 720}
info = (await client.get("/v1/hub/info")).json()
assert info["browser_idle_hours"] == 1
@pytest.mark.asyncio
async def test_a_refresh_token_lasts_the_idle_window(client, db_session):
await _register(client, "idle_window")
await _login(client, "idle_window")
left = await _expiry(db_session, "idle_window") - datetime.now(UTC)
assert timedelta(hours=23, minutes=58) < left <= timedelta(hours=24)
@pytest.mark.asyncio
async def test_the_idle_window_never_undercuts_the_access_token(client, db_session):
"""One hour idle with a one-hour access token would lapse between renewals."""
admin = await _admin(client)
await client.patch("/v1/admin/settings", headers=admin,
json={"session": {"refresh_idle_hours": 1}})
await _register(client, "short_idle")
await _login(client, "short_idle")
left = await _expiry(db_session, "short_idle") - datetime.now(UTC)
# The test hub's access token lives 3600 s; the floor is that plus an hour.
assert left > timedelta(hours=1, minutes=58)
@pytest.mark.asyncio
async def test_no_session_renews_past_its_maximum(client, db_session):
await _register(client, "long_session")
first = await _login(client, "long_session")
# Renewal while young works, and slides the window.
r = await _refresh(client, first["refresh_token"])
assert r.status_code == 200, r.text
current = r.json()["refresh_token"]
# The family's sign-in is now older than the 720 h maximum.
uid = (await db_session.execute(
select(User.id).where(User.username == "long_session"))).scalar_one()
await db_session.execute(
update(RefreshToken).where(RefreshToken.user_id == uid)
.values(created_at=datetime.now(UTC) - timedelta(hours=721)))
await db_session.commit()
r = await _refresh(client, current)
assert r.status_code == 401
assert r.json()["detail"] == "Session expired"
@pytest.mark.asyncio
async def test_signing_out_ends_the_session_on_the_hub(client):
await _register(client, "signs_out")
tokens = await _login(client, "signs_out")
r = await client.post("/v1/users/logout", json={"refresh_token": tokens["refresh_token"]})
assert r.status_code == 200
assert (await _refresh(client, tokens["refresh_token"])).status_code == 401
@pytest.mark.asyncio
async def test_an_unknown_token_gets_the_same_answer(client):
r = await client.post("/v1/users/logout", json={"refresh_token": "not-a-token"})
assert r.status_code == 200
assert r.json() == {"status": "signed_out"}
@pytest.mark.asyncio
async def test_signing_out_ends_only_that_session(client):
await _register(client, "two_browsers")
here = await _login(client, "two_browsers")
there = await _login(client, "two_browsers")
await client.post("/v1/users/logout", json={"refresh_token": here["refresh_token"]})
assert (await _refresh(client, there["refresh_token"])).status_code == 200
@pytest.mark.asyncio
async def test_sign_out_everywhere_ends_every_session(client):
await _register(client, "everywhere")
here = await _login(client, "everywhere")
there = await _login(client, "everywhere")
r = await client.post("/v1/users/me/sessions/revoke",
headers={"Authorization": f"Bearer {here['access_token']}"})
assert r.status_code == 200
assert r.json()["revoked"] == 2
assert (await _refresh(client, here["refresh_token"])).status_code == 401
assert (await _refresh(client, there["refresh_token"])).status_code == 401
@pytest.mark.asyncio
async def test_sign_out_everywhere_needs_the_account(client):
# No header at all is refused before the route runs, as a missing field.
r = await client.post("/v1/users/me/sessions/revoke")
assert r.status_code in (401, 403, 422)
r = await client.post("/v1/users/me/sessions/revoke",
headers={"Authorization": "Bearer not-a-real-token"})
assert r.status_code == 401
@pytest.mark.asyncio
async def test_values_are_clamped_and_unknown_keys_refused(client):
admin = await _admin(client)
r = await client.patch("/v1/admin/settings", headers=admin,
json={"session": {"browser_idle_hours": 0, "max_hours": 10**9}})
assert r.status_code == 200
bounds = r.json()["session_bounds"]
assert r.json()["session"]["browser_idle_hours"] == bounds["browser_idle_hours"][0]
assert r.json()["session"]["max_hours"] == bounds["max_hours"][1]
r = await client.patch("/v1/admin/settings", headers=admin,
json={"session": {"idle_minutes": 5}})
assert r.status_code == 422
@pytest.mark.asyncio
async def test_only_an_admin_changes_them(client):
await _admin(client)
await _register(client, "not_an_admin")
token = (await _login(client, "not_an_admin"))["access_token"]
r = await client.patch("/v1/admin/settings",
headers={"Authorization": f"Bearer {token}"},
json={"session": {"browser_idle_hours": 168}})
assert r.status_code == 403
|