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
"""
Root auth provider: base settings and interface common to all auth methods.
"""
from __future__ import annotations
import hmac
import secrets
import time
from datetime import timedelta
class RootAuthProvider:
"""
Base auth provider with admin credentials, session timeout, and session create/validate.
Subclass or use as-is for simple admin user/password auth.
"""
def __init__(
self,
*,
admin_user: bytes | None = None,
admin_password: bytes | None = None,
session_timeout: timedelta | float | None = None, # duration; None = no expiry
) -> None:
"""
admin_user: optional admin username (bytes). If None, no admin login is accepted.
admin_password: optional admin password (bytes). If None, no admin login is accepted.
session_timeout: optional duration in seconds (float) or timedelta. If None, sessions do not expire.
"""
self.admin_user = admin_user
self.admin_password = admin_password
if session_timeout is not None and hasattr(session_timeout, "total_seconds"):
self._timeout_seconds = session_timeout.total_seconds()
else:
self._timeout_seconds = session_timeout # float or None
self._sessions: dict[str, float] = {} # token -> created_at (monotonic or epoch)
def create_session(
self,
*,
user: bytes | None = None,
password: bytes | None = None,
) -> str | None:
"""
Authenticate with user/password and create a session if valid.
Returns a session token or None if credentials are missing or invalid.
"""
if self.admin_user is None or self.admin_password is None:
return None
if user is None or password is None:
return None
if not hmac.compare_digest(user, self.admin_user) or not hmac.compare_digest(password, self.admin_password):
return None
token = secrets.token_urlsafe(32)
self._sessions[token] = time.monotonic()
return token
def validate_session(self, session_token: str) -> bool:
"""
Return True if the session token exists and (when session_timeout is set) is not expired.
"""
if not session_token:
return False
created = self._sessions.get(session_token)
if created is None:
return False
if self._timeout_seconds is not None and time.monotonic() - created > self._timeout_seconds:
del self._sessions[session_token]
return False
return True