80 lines
3.3 KiB
Python
80 lines
3.3 KiB
Python
"""Separate human authentication backed by hashed passwords and sessions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi import HTTPException, Request
|
|
from sqlalchemy import select
|
|
|
|
from .config import settings
|
|
from .models import AdminUser, WebSession
|
|
|
|
|
|
def _digest(value: str) -> str:
|
|
return hashlib.sha256(value.encode()).hexdigest()
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
salt = os.urandom(16)
|
|
encoded = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 240_000)
|
|
return "pbkdf2_sha256$240000$%s$%s" % (base64.urlsafe_b64encode(salt).decode(), base64.urlsafe_b64encode(encoded).decode())
|
|
|
|
|
|
def verify_password(password: str, stored: str) -> bool:
|
|
try:
|
|
algorithm, rounds, salt, digest = stored.split("$", 3)
|
|
candidate = hashlib.pbkdf2_hmac(algorithm.removeprefix("pbkdf2_"), password.encode(), base64.urlsafe_b64decode(salt), int(rounds))
|
|
return hmac.compare_digest(base64.urlsafe_b64encode(candidate).decode(), digest)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def create_admin(session, username: str, password: str):
|
|
username = username.strip().lower()
|
|
if len(username) < 3:
|
|
raise ValueError("username must have 3 characters")
|
|
environment = os.getenv("APP_ENV") or settings.environment
|
|
weak_passwords_allowed = os.getenv("ALLOW_WEAK_DEV_PASSWORD", "0") == "1"
|
|
if len(password) < 12 and (environment.strip().lower() == "production" or not weak_passwords_allowed):
|
|
raise ValueError("password must have 12 characters")
|
|
if session.scalar(select(AdminUser).where(AdminUser.username == username)):
|
|
raise ValueError("username already exists")
|
|
admin = AdminUser(username=username, password_hash=hash_password(password))
|
|
session.add(admin); session.commit(); return admin
|
|
|
|
|
|
def create_session(session, admin: AdminUser, request: Request):
|
|
raw_token = secrets.token_urlsafe(32)
|
|
raw_csrf = secrets.token_urlsafe(32)
|
|
now = datetime.now(timezone.utc)
|
|
row = WebSession(admin_user_id=admin.id, token_hash=_digest(raw_token), csrf_hash=_digest(raw_csrf), expires_at=now + timedelta(seconds=settings.web_session_seconds), user_agent=request.headers.get("user-agent", "")[:512], remote_addr=request.client.host if request.client else None)
|
|
session.add(row); admin.last_login_at = now; session.commit()
|
|
return raw_token, raw_csrf
|
|
|
|
|
|
def current_session(request: Request, session):
|
|
token = request.cookies.get(settings.web_cookie_name)
|
|
if not token: return None, None
|
|
row = session.scalar(select(WebSession).where(WebSession.token_hash == _digest(token)))
|
|
if not row or row.expires_at.replace(tzinfo=timezone.utc) < datetime.now(timezone.utc): return None, None
|
|
admin = session.get(AdminUser, row.admin_user_id)
|
|
if not admin or not admin.is_active: return None, None
|
|
return row, admin
|
|
|
|
|
|
def require_human(request: Request, session):
|
|
web_session, admin = current_session(request, session)
|
|
if not admin: raise HTTPException(401, "human login required")
|
|
return web_session, admin
|
|
|
|
|
|
def require_csrf(web_session: WebSession, csrf_token: str | None):
|
|
if not csrf_token or not hmac.compare_digest(web_session.csrf_hash, _digest(csrf_token)):
|
|
raise HTTPException(403, "invalid CSRF token")
|