68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import secrets
|
|
from dataclasses import dataclass
|
|
|
|
|
|
def make_password_hash(password: str, iterations: int = 310_000) -> str:
|
|
salt = secrets.token_hex(16)
|
|
digest = hashlib.pbkdf2_hmac("sha256", password.encode(), bytes.fromhex(salt), iterations).hex()
|
|
return f"pbkdf2_sha256${iterations}${salt}${digest}"
|
|
|
|
|
|
def verify_password(password: str, encoded: str) -> bool:
|
|
try:
|
|
algorithm, iterations_text, salt, expected = encoded.split("$", 3)
|
|
if algorithm != "pbkdf2_sha256":
|
|
return False
|
|
digest = hashlib.pbkdf2_hmac(
|
|
"sha256", password.encode(), bytes.fromhex(salt), int(iterations_text)
|
|
).hex()
|
|
return hmac.compare_digest(digest, expected)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
@dataclass
|
|
class Session:
|
|
csrf: str
|
|
|
|
|
|
class SessionManager:
|
|
def __init__(self, password_hash: str | None):
|
|
self.password_hash = password_hash
|
|
self.sessions: dict[str, Session] = {}
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return bool(self.password_hash)
|
|
|
|
@classmethod
|
|
def from_environment(cls) -> "SessionManager":
|
|
encoded = os.getenv("PPM_PASSWORD_HASH", "").strip()
|
|
plain = os.getenv("PPM_PASSWORD", "")
|
|
if not encoded and plain:
|
|
encoded = make_password_hash(plain)
|
|
return cls(encoded or None)
|
|
|
|
def login(self, password: str) -> tuple[str, Session] | None:
|
|
if not self.password_hash or not verify_password(password, self.password_hash):
|
|
return None
|
|
session_id = secrets.token_urlsafe(32)
|
|
session = Session(csrf=secrets.token_urlsafe(32))
|
|
self.sessions[session_id] = session
|
|
return session_id, session
|
|
|
|
def get(self, session_id: str | None) -> Session | None:
|
|
if not session_id:
|
|
return None
|
|
return self.sessions.get(session_id)
|
|
|
|
def logout(self, session_id: str | None) -> None:
|
|
if session_id:
|
|
self.sessions.pop(session_id, None)
|
|
|