feat(v2): add secure control plane
This commit is contained in:
@@ -0,0 +1,72 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import secrets
|
||||
from datetime import UTC, datetime
|
||||
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) -> str:
|
||||
payload = json.dumps(
|
||||
{"sub": user_id, "csrf": csrf, "iat": datetime.now(UTC).isoformat()},
|
||||
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):
|
||||
return None
|
||||
return decoded
|
||||
except (ValueError, json.JSONDecodeError, UnicodeDecodeError):
|
||||
return None
|
||||
Reference in New Issue
Block a user