from __future__ import annotations import base64 import hashlib import hmac import json import secrets from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any from argon2 import PasswordHasher from argon2.exceptions import VerifyMismatchError _PASSWORDS = PasswordHasher() def hash_password(password: str) -> str: return _PASSWORDS.hash(password) def verify_password(password_hash: str, password: str) -> bool: try: return _PASSWORDS.verify(password_hash, password) except VerifyMismatchError: return False def hash_token(token: str) -> str: return hashlib.sha256(token.encode()).hexdigest() def new_token() -> str: return secrets.token_urlsafe(32) def new_csrf_token() -> str: return secrets.token_urlsafe(32) def _session_key(master_key_file: Path) -> bytes: return hashlib.sha256(b"backup-tool-session\0" + master_key_file.read_bytes()).digest() def sign_session( user_id: str, csrf: str, master_key_file: Path, *, expires_at: datetime | None = None, session_id: str | None = None, ) -> str: issued_at = datetime.now(UTC) expires_at = expires_at or issued_at.replace(microsecond=0) + timedelta(hours=8) payload = json.dumps( { "sub": user_id, "csrf": csrf, "iat": issued_at.isoformat(), "exp": expires_at.astimezone(UTC).isoformat(), "sid": session_id or secrets.token_urlsafe(24), }, separators=(",", ":"), sort_keys=True, ).encode() encoded = base64.urlsafe_b64encode(payload).decode().rstrip("=") signature = hmac.new( _session_key(master_key_file), encoded.encode(), hashlib.sha256 ).hexdigest() return f"{encoded}.{signature}" def verify_session(value: str, master_key_file: Path) -> dict[str, Any] | None: try: encoded, supplied = value.rsplit(".", 1) expected = hmac.new( _session_key(master_key_file), encoded.encode(), hashlib.sha256 ).hexdigest() if not hmac.compare_digest(supplied, expected): return None padded = encoded + "=" * (-len(encoded) % 4) decoded = json.loads(base64.urlsafe_b64decode(padded)) if ( not isinstance(decoded, dict) or not isinstance(decoded.get("sub"), str) or not isinstance(decoded.get("sid"), str) or not isinstance(decoded.get("exp"), str) or datetime.fromisoformat(decoded["exp"]).astimezone(UTC) <= datetime.now(UTC) ): return None return decoded except (ValueError, json.JSONDecodeError, UnicodeDecodeError): return None