Spaces:
Running
Running
| """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() | |