Blablablab/audio-classification
0
1"""2Authentication System Module3 4This module provides a comprehensive authentication system for the Potato annotation platform.5It supports multiple authentication backends including in-memory storage, database storage,6and third-party SSO providers like Clerk.7 8The system is designed to be extensible and supports both password-based and passwordless9authentication modes. It includes user management, session validation, and secure10password handling.11 12Key Features:13- Multiple authentication backends (in-memory, database, Clerk SSO)14- Password hashing with PBKDF2 and per-user salts15- Passwordless authentication support16- User registration and management17- Password reset with secure tokens18- Session-based authentication19- Configurable authentication requirements20"""21 22import os23import json24import logging25import hashlib26import hmac27import secrets28import sqlite329import requests30import threading31import time32from abc import ABC, abstractmethod33from typing import Optional, Dict, Any, List, Union34 35logger = logging.getLogger(__name__)36 37# Global singleton instance of the user authenticator with thread-safe lock38USER_AUTHENTICATOR_SINGLETON = None39_USER_AUTHENTICATOR_LOCK = threading.Lock()40 41# Format for per-user salt storage: "<32-char-hex-salt>$<hash-hex>"42_SALT_HASH_SEPARATOR = "$"43 44 45def _is_salted_hash(value: str) -> bool:46 """Check if a stored password value is in the per-user salt$hash format."""47 if not value or _SALT_HASH_SEPARATOR not in value:48 return False49 parts = value.split(_SALT_HASH_SEPARATOR, 1)50 # salt is 32 hex chars (16 bytes), hash is 64 hex chars (32 bytes sha256)51 return len(parts) == 2 and len(parts[0]) == 32 and len(parts[1]) == 6452 53 54def _hash_password_with_salt(password: str, salt: str = None) -> str:55 """Hash a password with a per-user salt using PBKDF2.56 57 Args:58 password: The plain text password to hash59 salt: Hex-encoded salt string. If None, generates a new random salt.60 61 Returns:62 str: The combined "salt$hash" string63 """64 if not password:65 return ""66 if salt is None:67 salt = secrets.token_hex(16)68 hash_value = hashlib.pbkdf2_hmac(69 'sha256',70 password.encode('utf-8'),71 salt.encode('utf-8'),72 10000073 ).hex()74 return f"{salt}{_SALT_HASH_SEPARATOR}{hash_value}"75 76 77def _verify_password(password: str, stored: str) -> bool:78 """Verify a password against a stored salt$hash value using constant-time comparison."""79 if not password or not stored:80 return False81 if not _is_salted_hash(stored):82 return False83 salt, expected_hash = stored.split(_SALT_HASH_SEPARATOR, 1)84 actual_hash = hashlib.pbkdf2_hmac(85 'sha256',86 password.encode('utf-8'),87 salt.encode('utf-8'),88 10000089 ).hex()90 return hmac.compare_digest(expected_hash, actual_hash)91 92 93class AuthBackend(ABC):94 """95 Abstract base class for authentication backends.96 97 This class defines the interface that all authentication backends must implement.98 It provides a consistent API for user authentication, registration, and validation99 regardless of the underlying storage mechanism.100 """101 @abstractmethod102 def authenticate(self, username: str, password: Optional[str]) -> bool:103 """Authenticate a user against this backend."""104 pass105 106 @abstractmethod107 def add_user(self, username: str, password: Optional[str], **kwargs) -> str:108 """Add a user to this backend. Returns status message."""109 pass110 111 @abstractmethod112 def is_valid_username(self, username: str) -> bool:113 """Check if a username exists in this backend."""114 pass115 116 @abstractmethod117 def update_password(self, username: str, new_password: str) -> bool:118 """Update a user's password. Returns True on success."""119 pass120 121 @abstractmethod122 def get_all_users(self) -> List[str]:123 """Return list of all usernames."""124 pass125 126 def add_user_prehashed(self, username: str, hashed_password: str, **kwargs) -> str:127 """Load user with already-hashed password (for file loading). Override in subclasses."""128 raise NotImplementedError("This backend does not support loading pre-hashed passwords")129 130 131class InMemoryAuthBackend(AuthBackend):132 """133 Authentication backend that stores users in memory with per-user salts.134 135 Password storage format: "salt$hash" where salt is 32 hex chars and hash is 64 hex chars.136 """137 def __init__(self):138 self.users = {} # username -> "salt$hash"139 self.user_data = {} # username -> additional data140 141 def authenticate(self, username: str, password: Optional[str]) -> bool:142 if username not in self.users:143 return False144 if password is None: # Passwordless login145 return True146 return _verify_password(password, self.users[username])147 148 def add_user(self, username: str, password: Optional[str], **kwargs) -> str:149 if username in self.users:150 return "Duplicate user"151 self.users[username] = _hash_password_with_salt(password) if password else ""152 self.user_data[username] = kwargs153 return "Success"154 155 def add_user_prehashed(self, username: str, hashed_password: str, **kwargs) -> str:156 """Store a user with an already-hashed password (salt$hash format)."""157 if username in self.users:158 return "Duplicate user"159 self.users[username] = hashed_password160 self.user_data[username] = kwargs161 return "Success"162 163 def is_valid_username(self, username: str) -> bool:164 return username in self.users165 166 def update_password(self, username: str, new_password: str) -> bool:167 if username not in self.users:168 return False169 self.users[username] = _hash_password_with_salt(new_password)170 return True171 172 def get_all_users(self) -> List[str]:173 return list(self.users.keys())174 175 176class DatabaseAuthBackend(AuthBackend):177 """178 Authentication backend using SQLite (stdlib) or PostgreSQL (psycopg2).179 180 Connection string formats:181 sqlite:///path/to/db.db (relative or absolute)182 postgresql://user:pass@host/dbname183 """184 def __init__(self, db_connection_string: str):185 self.db_connection_string = db_connection_string186 self._lock = threading.Lock()187 self._db_type = None # 'sqlite' or 'postgresql'188 self._connection = None189 190 if db_connection_string.startswith("sqlite:///"):191 self._db_type = "sqlite"192 self._init_sqlite(db_connection_string[len("sqlite:///"):])193 elif db_connection_string.startswith("postgresql://"):194 self._db_type = "postgresql"195 self._init_postgresql(db_connection_string)196 else:197 raise ValueError(198 f"Unsupported database URL: {db_connection_string}. "199 "Use sqlite:///path/to/db or postgresql://user:pass@host/dbname"200 )201 202 logger.info(f"Database auth backend initialized ({self._db_type})")203 204 def _init_sqlite(self, db_path: str):205 """Initialize SQLite database."""206 # Create parent directories if needed207 db_dir = os.path.dirname(db_path)208 if db_dir:209 os.makedirs(db_dir, exist_ok=True)210 211 self._connection = sqlite3.connect(db_path, check_same_thread=False)212 self._connection.execute("PRAGMA journal_mode=WAL")213 self._connection.execute("""214 CREATE TABLE IF NOT EXISTS users (215 username TEXT PRIMARY KEY,216 password_hash TEXT NOT NULL,217 email TEXT,218 created_at TEXT DEFAULT (datetime('now')),219 updated_at TEXT DEFAULT (datetime('now'))220 )221 """)222 self._connection.commit()223 224 def _init_postgresql(self, connection_string: str):225 """Initialize PostgreSQL database."""226 try:227 import psycopg2228 except ImportError:229 raise ImportError(230 "psycopg2 is required for PostgreSQL authentication backend. "231 "Install it with: pip install psycopg2-binary"232 )233 self._connection = psycopg2.connect(connection_string)234 self._connection.autocommit = True235 with self._connection.cursor() as cur:236 cur.execute("""237 CREATE TABLE IF NOT EXISTS users (238 username TEXT PRIMARY KEY,239 password_hash TEXT NOT NULL,240 email TEXT,241 created_at TIMESTAMP DEFAULT NOW(),242 updated_at TIMESTAMP DEFAULT NOW()243 )244 """)245 246 def _execute(self, query: str, params: tuple = (), fetch: str = None):247 """Thread-safe query execution.248 249 Args:250 query: SQL query with ? placeholders (auto-converted to %s for PostgreSQL)251 params: Query parameters252 fetch: None, 'one', or 'all'253 254 Returns:255 Query result based on fetch parameter256 """257 with self._lock:258 if self._db_type == "postgresql":259 query = query.replace("?", "%s")260 261 if self._db_type == "sqlite":262 cursor = self._connection.cursor()263 cursor.execute(query, params)264 if fetch == "one":265 result = cursor.fetchone()266 elif fetch == "all":267 result = cursor.fetchall()268 else:269 self._connection.commit()270 result = None271 cursor.close()272 return result273 else:274 with self._connection.cursor() as cur:275 cur.execute(query, params)276 if fetch == "one":277 return cur.fetchone()278 elif fetch == "all":279 return cur.fetchall()280 return None281 282 def authenticate(self, username: str, password: Optional[str]) -> bool:283 row = self._execute(284 "SELECT password_hash FROM users WHERE username = ?",285 (username,), fetch="one"286 )287 if not row:288 return False289 if password is None: # Passwordless login290 return True291 return _verify_password(password, row[0])292 293 def add_user(self, username: str, password: Optional[str], **kwargs) -> str:294 existing = self._execute(295 "SELECT 1 FROM users WHERE username = ?",296 (username,), fetch="one"297 )298 if existing:299 return "Duplicate user"300 301 hashed = _hash_password_with_salt(password) if password else ""302 email = kwargs.get("email", "")303 self._execute(304 "INSERT INTO users (username, password_hash, email) VALUES (?, ?, ?)",305 (username, hashed, email)306 )307 return "Success"308 309 def add_user_prehashed(self, username: str, hashed_password: str, **kwargs) -> str:310 """Store a user with an already-hashed password."""311 existing = self._execute(312 "SELECT 1 FROM users WHERE username = ?",313 (username,), fetch="one"314 )315 if existing:316 return "Duplicate user"317 318 email = kwargs.get("email", "")319 self._execute(320 "INSERT INTO users (username, password_hash, email) VALUES (?, ?, ?)",321 (username, hashed_password, email)322 )323 return "Success"324 325 def is_valid_username(self, username: str) -> bool:326 row = self._execute(327 "SELECT 1 FROM users WHERE username = ?",328 (username,), fetch="one"329 )330 return row is not None331 332 def update_password(self, username: str, new_password: str) -> bool:333 if not self.is_valid_username(username):334 return False335 hashed = _hash_password_with_salt(new_password)336 if self._db_type == "sqlite":337 self._execute(338 "UPDATE users SET password_hash = ?, updated_at = datetime('now') WHERE username = ?",339 (hashed, username)340 )341 else:342 self._execute(343 "UPDATE users SET password_hash = ?, updated_at = NOW() WHERE username = ?",344 (hashed, username)345 )346 return True347 348 def get_all_users(self) -> List[str]:349 rows = self._execute("SELECT username FROM users", fetch="all")350 return [r[0] for r in rows]351 352 def close(self):353 """Close the database connection."""354 if self._connection:355 self._connection.close()356 self._connection = None357 358 359class ClerkAuthBackend(AuthBackend):360 """361 Authentication backend that uses Clerk for SSO.362 """363 def __init__(self, api_key: str, frontend_api: str):364 self.api_key = api_key365 self.frontend_api = frontend_api366 self.users = {} # Cache of known users367 logger.info("Clerk SSO backend initialized")368 369 def authenticate(self, username: str, token: Optional[str]) -> bool:370 if not token:371 return False372 try:373 headers = {374 "Authorization": f"Bearer {self.api_key}",375 "Content-Type": "application/json"376 }377 response = requests.get(378 f"https://api.clerk.dev/v1/sessions/{token}",379 headers=headers380 )381 if response.status_code == 200:382 user_data = response.json()383 self.users[username] = user_data384 return True385 return False386 except Exception as e:387 logger.error(f"Error authenticating with Clerk: {str(e)}")388 return False389 390 def add_user(self, username: str, password: Optional[str], **kwargs) -> str:391 return "User management happens through Clerk dashboard"392 393 def is_valid_username(self, username: str) -> bool:394 return username in self.users395 396 def update_password(self, username: str, new_password: str) -> bool:397 raise NotImplementedError("Password management is handled by Clerk")398 399 def get_all_users(self) -> List[str]:400 return list(self.users.keys())401 402 403class UserAuthenticator:404 """405 A class for maintaining state on which users are allowed to use the system.406 407 This class provides a unified interface for user authentication and management408 regardless of the underlying backend. It supports multiple authentication methods409 and can be configured for passwordless operation.410 """411 412 def __init__(self, user_config_path, auth_method="in_memory", auth_config=None):413 self.allow_all_users = True414 self.user_config_path = user_config_path415 self.user_config_path_explicit = False # Set to True if path was explicitly configured416 self.authorized_users = []417 self.userlist = []418 self.usernames = set()419 self.users = {}420 self.required_user_info_keys = ["username", "password"]421 self.require_password = True422 self.auth_method = auth_method423 self.auth_config = auth_config or {}424 self.auth_backend = self._initialize_backend(auth_method, auth_config)425 426 # Token management for password reset427 self._reset_tokens = {} # token -> {username, expires}428 self._token_lock = threading.Lock()429 430 # Track load outcomes so init_from_config can warn on a silently empty431 # (e.g. wrong-format) user file. F-036.432 self.users_loaded_from_file = 0433 self.user_file_parse_errors = 0434 435 # Load users from config file if it exists436 if os.path.isfile(self.user_config_path):437 logger.info(f"Loading users from {self.user_config_path}")438 before = len(self.users)439 with open(self.user_config_path, "rt", encoding="utf-8") as f:440 for lineno, line in enumerate(f.readlines(), start=1):441 line = line.strip()442 if not line:443 continue444 # Tolerate a malformed line instead of aborting the whole445 # load (and crashing server boot) on one bad row.446 try:447 single_user = json.loads(line)448 except (ValueError, TypeError) as e:449 self.user_file_parse_errors += 1450 logger.error(451 f"User file {self.user_config_path} line {lineno}: "452 f"not valid JSON ({e}); skipping. Expected JSONL — "453 f'one object per line, e.g. {{"username": "alice", "password": "x"}}'454 )455 continue456 # Detect salt$hash format in password field457 password_val = single_user.get("password", "") if isinstance(single_user, dict) else ""458 if password_val and _is_salted_hash(password_val):459 self._add_user_prehashed(single_user)460 else:461 self.add_single_user(single_user)462 self.users_loaded_from_file = len(self.users) - before463 464 def _initialize_backend(self, auth_method: str, auth_config: dict = None) -> AuthBackend:465 if auth_method == "in_memory":466 return InMemoryAuthBackend()467 elif auth_method == "database":468 db_url = (auth_config or {}).get("database_url") or \469 os.environ.get("POTATO_DB_CONNECTION", "sqlite:///potato_users.db")470 return DatabaseAuthBackend(db_url)471 elif auth_method == "clerk":472 api_key = os.environ.get("CLERK_API_KEY", "")473 frontend_api = os.environ.get("CLERK_FRONTEND_API", "")474 if not api_key:475 logger.error("CLERK_API_KEY environment variable is not set")476 raise ValueError("CLERK_API_KEY must be set for Clerk authentication")477 return ClerkAuthBackend(api_key, frontend_api)478 elif auth_method == "oauth":479 from potato.auth_backends.oauth_backend import OAuthBackend480 if not auth_config:481 raise ValueError("OAuth authentication requires an 'authentication' config section with 'providers'")482 return OAuthBackend(auth_config)483 else:484 logger.error(f"Unknown authentication method: {auth_method}")485 raise ValueError(f"Unknown authentication method: {auth_method}")486 487 @staticmethod488 def init_from_config(config: dict) -> "UserAuthenticator":489 """Initialize the UserAuthenticator from a configuration dictionary (singleton)."""490 global USER_AUTHENTICATOR_SINGLETON491 492 if USER_AUTHENTICATOR_SINGLETON is None:493 with _USER_AUTHENTICATOR_LOCK:494 if USER_AUTHENTICATOR_SINGLETON is None:495 auth_method = config.get("authentication", {}).get("method", "in_memory")496 user_config_path = config.get("authentication", {}).get("user_config_path", None)497 require_password = config.get("require_password", True)498 499 path_explicit = user_config_path is not None500 501 if user_config_path is None:502 config_dir = os.path.dirname(config['output_annotation_dir'])503 user_config_path = os.path.join(config_dir, "user_config.json")504 else:505 # Don't raise if file doesn't exist — it will be created on first registration506 if not os.path.isfile(user_config_path):507 logger.info(f"user_config_path '{user_config_path}' does not exist yet; will be created on first registration")508 509 logger.debug(f"User config path: {user_config_path}")510 511 auth_config = config.get("authentication", {})512 513 USER_AUTHENTICATOR_SINGLETON = UserAuthenticator(user_config_path, auth_method, auth_config)514 USER_AUTHENTICATOR_SINGLETON.require_password = require_password515 USER_AUTHENTICATOR_SINGLETON.user_config_path_explicit = path_explicit516 517 # F-036: a user file was explicitly configured and exists, but518 # produced zero usable users (e.g. wrong format / all rows519 # invalid). With closed enrolment this is a silently broken520 # deployment — nobody can log in. Warn prominently.521 _auth = USER_AUTHENTICATOR_SINGLETON522 if (path_explicit and os.path.isfile(user_config_path)523 and _auth.users_loaded_from_file == 0):524 allow_all = config.get("user_config", {}).get("allow_all_users", False)525 logger.warning(526 "user_config_path '%s' was configured but loaded 0 users "527 "(%d malformed line(s)). Expected JSONL — one object per "528 'line, e.g. {"username": "alice", "password": "x"}. %s',529 user_config_path, _auth.user_file_parse_errors,530 ("Open registration is on, so new users can still self-register."531 if allow_all else532 "allow_all_users is false, so NO ONE will be able to log in."),533 )534 535 logger.info(f"Initialized UserAuthenticator with method: {auth_method}, require_password: {require_password}")536 537 return USER_AUTHENTICATOR_SINGLETON538 539 @staticmethod540 def get_instance():541 global USER_AUTHENTICATOR_SINGLETON542 if USER_AUTHENTICATOR_SINGLETON is None:543 raise ValueError("UserAuthenticator not initialized; call init_from_config first")544 return USER_AUTHENTICATOR_SINGLETON545 546 @staticmethod547 def authenticate(username: str, password: Optional[str]) -> bool:548 authenticator = UserAuthenticator.get_instance()549 550 if not authenticator.auth_backend.is_valid_username(username):551 logger.warning(f"Authentication failed: user '{username}' does not exist")552 return False553 554 if not authenticator.require_password:555 logger.debug(f"Passwordless authentication for user: {username}")556 return authenticator.auth_backend.authenticate(username, None)557 558 return authenticator.auth_backend.authenticate(username, password)559 560 def add_user(self, username, password: Optional[str], **kwargs):561 """Add a user to the authentication system."""562 if not self.require_password:563 logger.debug(f"Passwordless mode - allowing any user: {username}")564 elif self.allow_all_users == False and not self.is_authorized_user(username):565 return "Unauthorized user"566 567 result = self.auth_backend.add_user(username, password, **kwargs)568 if result == "Success":569 user_data = {"username": username}570 user_data.update(kwargs)571 self.users[username] = user_data572 self.userlist.append(username)573 return result574 575 def _add_user_prehashed(self, single_user):576 """Add a user with an already-hashed password (loaded from file)."""577 username = single_user["username"]578 hashed_password = single_user.get("password", "")579 580 result = self.auth_backend.add_user_prehashed(581 username,582 hashed_password,583 **{k: v for k, v in single_user.items() if k not in ["username", "password"]}584 )585 586 if result == "Success":587 self.users[username] = single_user588 self.userlist.append(username)589 590 return result591 592 def add_single_user(self, single_user):593 """Add a single user to the full user dict."""594 if not self.require_password:595 logger.debug(f"Passwordless mode - allowing any user: {single_user['username']}")596 elif self.allow_all_users == False and not self.is_authorized_user(single_user["username"]):597 return "Unauthorized user"598 599 if not self.require_password:600 required_keys = ["username"]601 else:602 required_keys = self.required_user_info_keys603 604 for key in required_keys:605 if key not in single_user:606 logger.error(f"Missing {key} in user info")607 return f"Missing {key} in user info"608 609 result = self.auth_backend.add_user(610 single_user["username"],611 single_user.get("password"),612 **{k: v for k, v in single_user.items() if k not in ["username", "password"]}613 )614 615 if result == "Success":616 self.users[single_user["username"]] = single_user617 self.userlist.append(single_user["username"])618 619 return result620 621 def update_password(self, username: str, new_password: str) -> bool:622 """Update a user's password via the backend."""623 result = self.auth_backend.update_password(username, new_password)624 if result and username in self.users:625 # Update the stored user dict with the new hash for save_user_config626 if isinstance(self.users[username], dict):627 self.users[username]["password"] = self.auth_backend.users[username] \628 if hasattr(self.auth_backend, 'users') else _hash_password_with_salt(new_password)629 return result630 631 def save_user_config(self):632 """Save user config to file.633 634 Saves when:635 - auth_method is in_memory AND user_config_path was explicitly configured636 - auth_method is not in_memory and not database (other file-based methods)637 638 Skips when:639 - auth_method is database (DB handles its own persistence)640 - auth_method is in_memory with auto-generated default path (preserve old behavior)641 """642 if self.auth_method == "database":643 logger.debug("User config not saved - using database authentication (DB handles persistence)")644 return645 646 if self.auth_method == "in_memory" and not self.user_config_path_explicit:647 logger.debug("User config not saved - using in_memory with default path")648 return649 650 if self.user_config_path:651 with open(self.user_config_path, "wt", encoding="utf-8") as f:652 for k in self.userlist:653 user_data = self.users.get(k, {})654 if isinstance(user_data, dict):655 # Ensure password field contains the hashed value656 output = dict(user_data)657 if hasattr(self.auth_backend, 'users') and k in self.auth_backend.users:658 output["password"] = self.auth_backend.users[k]659 f.write(json.dumps(output) + "\n")660 else:661 f.write(json.dumps({"username": k}) + "\n")662 logger.info(f"User info file saved at: {self.user_config_path}")663 else:664 logger.warning("WARNING: user_config_path not specified, user registration info are not saved")665 666 # --- Token-based password reset ---667 668 def create_reset_token(self, username: str, ttl_hours: int = 24) -> Optional[str]:669 """Create a password reset token for a user.670 671 Args:672 username: The username to create a token for673 ttl_hours: Token validity in hours (default 24)674 675 Returns:676 The token string, or None if user doesn't exist677 """678 if not self.auth_backend.is_valid_username(username):679 return None680 681 token = secrets.token_urlsafe(32)682 expires = time.time() + (ttl_hours * 3600)683 684 with self._token_lock:685 # Invalidate any existing tokens for this user686 self._reset_tokens = {687 t: v for t, v in self._reset_tokens.items()688 if v["username"] != username689 }690 self._reset_tokens[token] = {691 "username": username,692 "expires": expires693 }694 695 return token696 697 def validate_reset_token(self, token: str) -> Optional[str]:698 """Validate a reset token and return the username, or None if invalid/expired."""699 with self._token_lock:700 # Clean expired tokens701 now = time.time()702 self._reset_tokens = {703 t: v for t, v in self._reset_tokens.items()704 if v["expires"] > now705 }706 707 if token not in self._reset_tokens:708 return None709 return self._reset_tokens[token]["username"]710 711 def consume_reset_token(self, token: str) -> Optional[str]:712 """Validate, delete, and return the username for a reset token. Single-use."""713 with self._token_lock:714 now = time.time()715 self._reset_tokens = {716 t: v for t, v in self._reset_tokens.items()717 if v["expires"] > now718 }719 720 if token not in self._reset_tokens:721 return None722 username = self._reset_tokens[token]["username"]723 del self._reset_tokens[token]724 return username725 726 # --- End token management ---727 728 def is_authorized_user(self, username):729 return username in self.authorized_users730 731 def is_valid_username(self, username):732 return self.auth_backend.is_valid_username(username)733 734 def is_valid_password(self, username, password):735 return self.authenticate(username, password)736 737 def get_clerk_frontend_api(self) -> str:738 if self.auth_method == "clerk" and isinstance(self.auth_backend, ClerkAuthBackend):739 return self.auth_backend.frontend_api740 return ""741 742 def get_oauth_backend(self):743 if self.auth_method == "oauth":744 return self.auth_backend745 return None746 747 def get_login_providers(self) -> list:748 oauth_backend = self.get_oauth_backend()749 if oauth_backend:750 return oauth_backend.get_login_providers()751 return []752