139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
"""Password hashing, API/session tokens, and FastAPI auth dependencies.
|
|
|
|
Stdlib-only crypto: PBKDF2-SHA256 for passwords, SHA-256 hashed bearer tokens
|
|
for both the desktop client (kind="api") and the WebUI session (kind="session").
|
|
"""
|
|
import hashlib
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import Depends, Header, HTTPException, Request, status
|
|
|
|
from app.db import get_connection
|
|
|
|
_PBKDF2_ITERATIONS = 600_000
|
|
|
|
ADMIN_ROLE = "admin"
|
|
EDITOR_ROLE = "editor"
|
|
USER_ROLE = "user"
|
|
ROLES = (ADMIN_ROLE, EDITOR_ROLE, USER_ROLE)
|
|
|
|
|
|
def now_iso():
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
salt = secrets.token_bytes(16)
|
|
digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, _PBKDF2_ITERATIONS)
|
|
return f"pbkdf2_sha256${_PBKDF2_ITERATIONS}${salt.hex()}${digest.hex()}"
|
|
|
|
|
|
def verify_password(password: str, stored: str) -> bool:
|
|
try:
|
|
scheme, iterations, salt_hex, digest_hex = stored.split("$")
|
|
if scheme != "pbkdf2_sha256":
|
|
return False
|
|
salt = bytes.fromhex(salt_hex)
|
|
expected = bytes.fromhex(digest_hex)
|
|
actual = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, int(iterations))
|
|
return secrets.compare_digest(actual, expected)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def _hash_token(raw: str) -> str:
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def issue_token(user_id: int, kind: str = "api", label: str = "", expires_at: str = "") -> str:
|
|
"""Create a token row and return the raw token (only shown once)."""
|
|
raw = secrets.token_urlsafe(32)
|
|
conn = get_connection()
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO api_tokens (user_id, token_hash, label, kind, created_at, last_used_at, expires_at, revoked_at) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?, NULL)",
|
|
(user_id, _hash_token(raw), label, kind, now_iso(), now_iso(), expires_at or None),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
return raw
|
|
|
|
|
|
def revoke_token(raw: str) -> None:
|
|
conn = get_connection()
|
|
try:
|
|
conn.execute(
|
|
"UPDATE api_tokens SET revoked_at = ? WHERE token_hash = ?",
|
|
(now_iso(), _hash_token(raw)),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_user_by_token(raw: str):
|
|
if not raw:
|
|
return None
|
|
conn = get_connection()
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT u.id, u.username, u.role, u.is_active FROM api_tokens t "
|
|
"JOIN users u ON u.id = t.user_id "
|
|
"WHERE t.token_hash = ? AND t.revoked_at IS NULL "
|
|
"AND (t.expires_at IS NULL OR t.expires_at = '' OR t.expires_at > ?)",
|
|
(_hash_token(raw), now_iso()),
|
|
).fetchone()
|
|
if row is None or not row["is_active"]:
|
|
return None
|
|
conn.execute(
|
|
"UPDATE api_tokens SET last_used_at = ? WHERE token_hash = ?",
|
|
(now_iso(), _hash_token(raw)),
|
|
)
|
|
conn.commit()
|
|
return dict(row)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_current_user(request: Request, authorization: str = Header(default="")):
|
|
"""Authenticate via ``Authorization: Bearer`` (client) or session cookie (WebUI)."""
|
|
token = ""
|
|
if authorization.startswith("Bearer "):
|
|
token = authorization[7:].strip()
|
|
else:
|
|
token = request.cookies.get("psppt_session", "")
|
|
user = get_user_by_token(token)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
|
|
return user
|
|
|
|
|
|
def require_role(*roles):
|
|
"""Dependency factory: only users with one of ``roles`` may pass."""
|
|
def dependency(user=Depends(get_current_user)):
|
|
if user["role"] not in roles:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions")
|
|
return user
|
|
return dependency
|
|
|
|
|
|
def ensure_admin(username: str, password: str) -> bool:
|
|
"""Create the initial admin account when the users table is empty."""
|
|
conn = get_connection()
|
|
try:
|
|
count = conn.execute("SELECT COUNT(*) AS n FROM users").fetchone()["n"]
|
|
if count == 0:
|
|
conn.execute(
|
|
"INSERT INTO users (username, password_hash, role, is_active, created_at) "
|
|
"VALUES (?, ?, 'admin', 1, ?)",
|
|
(username, hash_password(password), now_iso()),
|
|
)
|
|
conn.commit()
|
|
return True
|
|
finally:
|
|
conn.close()
|
|
return False
|