Spaces:
Running
Running
File size: 3,956 Bytes
d660fa9 f4542ce d660fa9 f4542ce | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """manager — in-memory admin session authentication.
Loads email:password pairs from ADMIN_CREDENTIALS, hashes with salted
scrypt, issues 24-hour bearer tokens, and validates/revokes them.
Architecture: used by auth/routes.py to guard all sim-control and roster
endpoints in Odin.py.
Design: in-memory sessions are intentionally simple (lost on restart);
when no credentials are configured, login is disabled with a warning.
"""
import os
import hashlib
import hmac
import secrets
import logging
from datetime import datetime, timedelta
from typing import Optional
logger = logging.getLogger(__name__)
SESSION_EXPIRY_HOURS = 24
class AuthManager:
def __init__(self):
self._users: dict[str, dict] = {}
self._sessions: dict[str, dict] = {}
self._load_users()
def _load_users(self):
from src import config as _cfg
raw = _cfg.ADMIN_CREDENTIALS
if not raw:
logger.warning("[Auth] No ADMIN_CREDENTIALS configured — login disabled")
return
for pair in raw.split(";"):
pair = pair.strip()
if ":" not in pair:
continue
email, password = pair.split(":", 1)
email = email.strip().lower()
password = password.strip()
if email and password:
self._users[email] = {
"name": email.split("@")[0],
"password_hash": self._hash_password(password),
}
logger.info("[Auth] Loaded admin user: %s", email)
def _hash_password(self, password: str) -> str:
salt = os.urandom(32)
key = hashlib.scrypt(
password.encode("utf-8"), salt=salt, n=16384, r=8, p=1, dklen=64
)
return salt.hex() + ":" + key.hex()
def verify_password(self, password: str, stored_hash: str) -> bool:
try:
salt_hex, key_hex = stored_hash.split(":")
salt = bytes.fromhex(salt_hex)
stored_key = bytes.fromhex(key_hex)
computed = hashlib.scrypt(
password.encode("utf-8"), salt=salt, n=16384, r=8, p=1, dklen=64
)
return hmac.compare_digest(computed, stored_key)
except (ValueError, AttributeError):
return False
def authenticate(self, email: str, password: str) -> Optional[dict]:
"""Verify credentials and return user dict, or None."""
user = self._users.get(email.lower().strip())
if not user:
return None
if not self.verify_password(password, user["password_hash"]):
return None
return {"email": email.lower().strip(), "name": user["name"], "role": "admin"}
def create_session(self, email: str) -> str:
user = self._users.get(email.lower())
if not user:
return ""
token = secrets.token_urlsafe(32)
self._sessions[token] = {
"email": email.lower(),
"name": user["name"],
"role": "admin",
"created_at": datetime.utcnow().isoformat(),
"expires_at": (
datetime.utcnow() + timedelta(hours=SESSION_EXPIRY_HOURS)
).isoformat(),
}
return token
def validate_session(self, token: str) -> Optional[dict]:
if not token:
return None
session = self._sessions.get(token)
if not session:
return None
try:
expires = datetime.fromisoformat(session["expires_at"])
except (ValueError, TypeError):
return None
if datetime.utcnow() > expires:
self._sessions.pop(token, None)
return None
return {
"email": session["email"],
"name": session["name"],
"role": session["role"],
}
def revoke_session(self, token: str):
self._sessions.pop(token, None)
auth_manager = AuthManager()
|