thomas
1f110866f5
CI / backend (pull_request) Canceled after 0s
CI / shell (pull_request) Canceled after 0s
CI / frontend (pull_request) Canceled after 0s
CI / arm64-smoke (pull_request) Canceled after 0s
CI / backend (push) Canceled after 0s
CI / shell (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
CI / arm64-smoke (push) Canceled after 0s
58 lines
1.6 KiB
Python
58 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
import secrets
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from argon2 import PasswordHasher
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class SessionData:
|
|
session_id: str
|
|
csrf_token: str
|
|
expires_at: datetime
|
|
role: str
|
|
username: str
|
|
|
|
|
|
class AuthService:
|
|
def __init__(self) -> None:
|
|
self._hasher = PasswordHasher()
|
|
self._sessions: dict[str, SessionData] = {}
|
|
|
|
def hash_password(self, password: str) -> str:
|
|
return self._hasher.hash(password)
|
|
|
|
def verify_password(self, password: str, password_hash: str) -> bool:
|
|
try:
|
|
return self._hasher.verify(password_hash, password)
|
|
except Exception:
|
|
return False
|
|
|
|
def create_session(self, username: str, role: str) -> SessionData:
|
|
session_id = secrets.token_hex(24)
|
|
csrf_token = secrets.token_hex(16)
|
|
expires_at = datetime.now(UTC) + timedelta(hours=12)
|
|
session = SessionData(
|
|
session_id=session_id,
|
|
csrf_token=csrf_token,
|
|
expires_at=expires_at,
|
|
role=role,
|
|
username=username,
|
|
)
|
|
self._sessions[session_id] = session
|
|
return session
|
|
|
|
def get_session(self, session_id: str | None) -> SessionData | None:
|
|
if session_id is None:
|
|
return None
|
|
session = self._sessions.get(session_id)
|
|
if session is None:
|
|
return None
|
|
if session.expires_at < datetime.now(UTC):
|
|
self._sessions.pop(session_id, None)
|
|
return None
|
|
return session
|
|
|