add auth system and split backend into two-image CI/CD build
Some checks failed
Build and Push / build (push) Failing after 17s
Some checks failed
Build and Push / build (push) Failing after 17s
Backend:
- password hashing via hashlib.scrypt
- stateless HMAC-SHA256 tokens (7-day expiry)
- POST /api/auth/login, /api/auth/register (admin), /api/auth/reset-password
- admin user created from NAT20_ADMIN_USERNAME/PASSWORD on first startup
- users table, campaign_shares table, created_by on campaigns
- require_user dependency on all routes except auth
- campaign sharing: GET/POST/DELETE /api/campaigns/{id}/shares
Frontend:
- AuthContext: user/token state, login/logout, global fetch Auth header
- Login page, Users page (admin user management)
- route protection, sidebar user info/sign out
Docker/CI:
- split backend/Dockerfile into thin app-only image
- backend/Dockerfile.deps builds the heavy WhisperX/PyTorch base
- CI builds deps only when requirements.txt changes
- docker compose pull now fetches ~100KB app layer instead of 3.5GB
This commit is contained in:
100
backend/app/auth.py
Normal file
100
backend/app/auth.py
Normal file
@@ -0,0 +1,100 @@
|
||||
"""
|
||||
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."""
|
||||
conn = db.get_conn()
|
||||
row = conn.execute("SELECT COUNT(*) as cnt FROM users").fetchone()
|
||||
if row["cnt"] > 0:
|
||||
return
|
||||
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})")
|
||||
Reference in New Issue
Block a user