122 lines
3.6 KiB
Python
122 lines
3.6 KiB
Python
"""Benutzer, Sessions, API-Tokens und CSRF."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import secrets
|
||
import time
|
||
from collections import defaultdict
|
||
|
||
from werkzeug.security import check_password_hash, generate_password_hash
|
||
|
||
from grotti.errors import GrottiError
|
||
|
||
ROLES = ("admin", "operator")
|
||
ROLE_RANK = {"operator": 1, "admin": 2}
|
||
TOKEN_PREFIX = "grt_"
|
||
LOGIN_WINDOW_S = 300
|
||
LOGIN_MAX_ATTEMPTS = 12
|
||
|
||
_login_hits: dict[str, list[float]] = defaultdict(list)
|
||
|
||
|
||
def hash_password(password: str) -> str:
|
||
return generate_password_hash(password, method="scrypt")
|
||
|
||
|
||
def verify_password(password_hash: str, password: str) -> bool:
|
||
if not password_hash or not password:
|
||
return False
|
||
return check_password_hash(password_hash, password)
|
||
|
||
|
||
def validate_username(username: str) -> str:
|
||
name = (username or "").strip()
|
||
if len(name) < 2 or len(name) > 64:
|
||
raise GrottiError("Benutzername: 2–64 Zeichen.", "user")
|
||
if not all(ch.isalnum() or ch in "._-" for ch in name):
|
||
raise GrottiError("Benutzername: nur Buchstaben, Ziffern, . _ -", "user")
|
||
return name
|
||
|
||
|
||
def validate_password(password: str) -> str:
|
||
if not password or len(password) < 8:
|
||
raise GrottiError("Passwort mindestens 8 Zeichen.", "password")
|
||
if len(password) > 200:
|
||
raise GrottiError("Passwort zu lang.", "password")
|
||
return password
|
||
|
||
|
||
def normalize_role(role: str, *, default: str = "operator") -> str:
|
||
role = (role or default).strip().lower()
|
||
if role not in ROLES:
|
||
raise GrottiError("Rolle muss admin oder operator sein.", "role")
|
||
return role
|
||
|
||
|
||
def role_at_least(have: str, need: str) -> bool:
|
||
return ROLE_RANK.get(have or "", 0) >= ROLE_RANK.get(need or "", 99)
|
||
|
||
|
||
def clamp_token_role(user_role: str, token_role: str | None) -> str:
|
||
wanted = normalize_role(token_role or user_role, default=user_role)
|
||
if not role_at_least(user_role, wanted):
|
||
raise GrottiError("Token-Rolle darf nicht höher als die des Benutzers sein.", "role")
|
||
return wanted
|
||
|
||
|
||
def new_api_token() -> str:
|
||
return TOKEN_PREFIX + secrets.token_urlsafe(32)
|
||
|
||
|
||
def hash_token(raw: str) -> str:
|
||
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def token_prefix(raw: str) -> str:
|
||
return (raw or "")[:12]
|
||
|
||
|
||
def new_csrf() -> str:
|
||
return secrets.token_urlsafe(24)
|
||
|
||
|
||
def login_rate_ok(key: str) -> bool:
|
||
now = time.time()
|
||
hits = _login_hits[key]
|
||
_login_hits[key] = [t for t in hits if now - t < LOGIN_WINDOW_S]
|
||
return len(_login_hits[key]) < LOGIN_MAX_ATTEMPTS
|
||
|
||
|
||
def login_rate_hit(key: str) -> None:
|
||
_login_hits[key].append(time.time())
|
||
|
||
|
||
def public_user(row) -> dict:
|
||
if not row:
|
||
return {}
|
||
created = row.get("created_at")
|
||
last = row.get("last_login")
|
||
return {
|
||
"id": row.get("id"),
|
||
"username": row.get("username"),
|
||
"role": row.get("role"),
|
||
"active": bool(row["active"]) if "active" in row else True,
|
||
"created_at": created.isoformat() if created and hasattr(created, "isoformat") else created,
|
||
"last_login": last.isoformat() if last and hasattr(last, "isoformat") else last,
|
||
}
|
||
|
||
|
||
def public_token(row) -> dict:
|
||
return {
|
||
"id": row["id"],
|
||
"user_id": row["user_id"],
|
||
"name": row["name"],
|
||
"prefix": row["token_prefix"],
|
||
"role": row["role"],
|
||
"created_at": row["created_at"].isoformat() if row.get("created_at") else None,
|
||
"last_used": row["last_used"].isoformat() if row.get("last_used") else None,
|
||
"expires_at": row["expires_at"].isoformat() if row.get("expires_at") else None,
|
||
"revoked": bool(row.get("revoked_at")),
|
||
}
|