109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
"""
|
|
Authentication module — zero-dependency (stdlib only).
|
|
Password hashing via hashlib.scrypt, stateless tokens via HMAC-SHA256.
|
|
"""
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from . import database as db
|
|
from .config import ADMIN_PASSWORD, ADMIN_USERNAME, JWT_SECRET
|
|
|
|
JWT_ALGORITHM = "HS256"
|
|
JWT_EXPIRY = 86400 * 7 # 7 days
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
salt = os.urandom(16)
|
|
key = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1, dklen=32)
|
|
return base64.b64encode(salt + key).decode()
|
|
|
|
|
|
def verify_password(password: str, stored: str) -> bool:
|
|
try:
|
|
data = base64.b64decode(stored)
|
|
salt, key = data[:16], data[16:]
|
|
key2 = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1, dklen=32)
|
|
return hmac.compare_digest(key, key2)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def create_token(user_id: str, username: str, is_admin: bool) -> str:
|
|
header = {"alg": "HS256", "typ": "JWT"}
|
|
payload = {
|
|
"sub": user_id,
|
|
"username": username,
|
|
"admin": is_admin,
|
|
"iat": time.time(),
|
|
"exp": time.time() + JWT_EXPIRY,
|
|
}
|
|
h = base64.urlsafe_b64encode(json.dumps(header, separators=(",", ":")).encode()).rstrip(b"=").decode()
|
|
p = base64.urlsafe_b64encode(json.dumps(payload, separators=(",", ":")).encode()).rstrip(b"=").decode()
|
|
sig = hmac.new(JWT_SECRET.encode(), f"{h}.{p}".encode(), hashlib.sha256).digest()
|
|
s = base64.urlsafe_b64encode(sig).rstrip(b"=").decode()
|
|
return f"{h}.{p}.{s}"
|
|
|
|
|
|
def decode_token(token: str) -> Optional[dict]:
|
|
try:
|
|
parts = token.split(".")
|
|
if len(parts) != 3:
|
|
return None
|
|
h, p, s = parts
|
|
sig = base64.urlsafe_b64decode(s + "==")
|
|
expected = hmac.new(JWT_SECRET.encode(), f"{h}.{p}".encode(), hashlib.sha256).digest()
|
|
if not hmac.compare_digest(sig, expected):
|
|
return None
|
|
payload = json.loads(base64.urlsafe_b64decode(p + "=="))
|
|
if payload.get("exp", 0) < time.time():
|
|
return None
|
|
return payload
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def authenticate(username: str, password: str) -> Optional[dict]:
|
|
user = db.get_user_by_username(username)
|
|
if not user or not verify_password(password, user["password_hash"]):
|
|
return None
|
|
token = create_token(user["id"], user["username"], bool(user["is_admin"]))
|
|
return {
|
|
"token": token,
|
|
"user": {"id": user["id"], "username": user["username"], "is_admin": bool(user["is_admin"])},
|
|
}
|
|
|
|
|
|
def get_user_from_token(token: str) -> Optional[dict]:
|
|
payload = decode_token(token)
|
|
if not payload:
|
|
return None
|
|
user = db.get_user(payload["sub"])
|
|
if not user:
|
|
return None
|
|
return {"id": user["id"], "username": user["username"], "is_admin": bool(user["is_admin"])}
|
|
|
|
|
|
def ensure_admin_exists():
|
|
"""Called on startup — creates the admin user if no users exist,
|
|
and ensures the 'default' campaign is owned by the admin."""
|
|
conn = db.get_conn()
|
|
row = conn.execute("SELECT COUNT(*) as cnt FROM users").fetchone()
|
|
if row["cnt"] == 0:
|
|
pw = ADMIN_PASSWORD or os.urandom(8).hex()
|
|
db.create_user(ADMIN_USERNAME, hash_password(pw), is_admin=True)
|
|
print(f"[auth] Created admin user '{ADMIN_USERNAME}' (password: {pw})")
|
|
|
|
# Claim the "default" campaign for the admin so setup wizard works
|
|
admin = db.get_user_by_username(ADMIN_USERNAME)
|
|
if admin:
|
|
conn.execute(
|
|
"UPDATE campaigns SET created_by = ? WHERE id = 'default' AND created_by IS NULL",
|
|
(admin["id"],),
|
|
)
|