""" Authentication Service for Aumentum Application Handles user authentication, JWT access/refresh tokens, and role management. Industry standard: short-lived access token, longer-lived refresh token. """ import os import secrets from datetime import datetime, timedelta from decimal import Decimal from typing import Optional, Dict, List, Any from jose import JWTError, jwt from passlib.context import CryptContext try: from backend.app.config import ( AUTH_SECRET_KEY, REFRESH_SECRET_KEY, ACCESS_TOKEN_EXPIRE_MINUTES, REFRESH_TOKEN_EXPIRE_DAYS, ) except ImportError: AUTH_SECRET_KEY = os.environ.get("AUTH_SECRET_KEY", "").strip() or "dev-secret-key-change-in-production-min-32-chars" REFRESH_SECRET_KEY = os.environ.get("REFRESH_SECRET_KEY", "").strip() or AUTH_SECRET_KEY ACCESS_TOKEN_EXPIRE_MINUTES = int(os.environ.get("ACCESS_TOKEN_EXPIRE_MINUTES", "15")) REFRESH_TOKEN_EXPIRE_DAYS = int(os.environ.get("REFRESH_TOKEN_EXPIRE_DAYS", "7")) SECRET_KEY = AUTH_SECRET_KEY ALGORITHM = "HS256" # Conditional import for pyodbc (only needed for MSSQL) try: import pyodbc except ImportError: pyodbc = None # Optional: bcrypt for verification fallback when passlib's bcrypt backend fails (e.g. bcrypt 4.1+) try: import bcrypt as _bcrypt_mod except ImportError: _bcrypt_mod = None # Password hashing context: # - Use bcrypt_sha256 to safely support passwords of any length (bcrypt truncates at 72 bytes). # - Keep bcrypt for backwards compatibility with existing hashes. pwd_context = CryptContext(schemes=["bcrypt_sha256", "bcrypt"], deprecated="auto") # Active sessions tracking (in-memory, for simple deployment) # For production, use Redis or database active_sessions = set() blacklisted_tokens = set() class AuthService: """ Authentication service for Aumentum Land Registry System Integrates with Alfresco authority framework for user management """ def __init__(self, db_config: Dict): """ Initialize authentication service Args: db_config: Database configuration dictionary """ self.db_config = db_config def _get_db_connection(self): """Get database connection (MySQL or MSSQL). Uses fresh config from env so .env is always applied.""" try: from backend.app.config import get_db_config self.db_config = get_db_config() except Exception: pass db_type = self.db_config.get("type", "mssql").lower() if db_type == "mysql": # MySQL connection try: import pymysql mysql_driver = pymysql except ImportError: try: import mysql.connector mysql_driver = mysql.connector except ImportError: raise ImportError("MySQL driver not found. Install with: pip install pymysql") host = self.db_config.get("host") or self.db_config.get("server", "localhost") port = self.db_config.get("port", 3306) database = self.db_config.get("database") username = self.db_config.get("username") or self.db_config.get("user") password = self.db_config.get("password", "") charset = self.db_config.get("charset", "utf8mb4") if mysql_driver == pymysql: conn = mysql_driver.connect( host=host, port=port, user=username, password=password, database=database, charset=charset, connect_timeout=10 ) else: conn = mysql_driver.connect( host=host, port=port, user=username, password=password, database=database, charset=charset, connection_timeout=10 ) return conn else: # MSSQL connection using pyodbc (FreeTDS or ODBC driver) if pyodbc is None: raise ImportError("pyodbc not available. Install with: pip install pyodbc") server = self.db_config.get('server', 'MSSQL_LRS43_PROD') database = self.db_config.get('database', 'LRS43') username = self.db_config.get('username', 'sa') password = self.db_config.get('password', '') driver = self.db_config.get('driver', 'FreeTDS') if driver == 'FreeTDS': conn_str = ( f"DRIVER={{{driver}}};" f"SERVERNAME={server};" f"DATABASE={database};" f"UID={username};" f"PWD={password};" ) else: conn_str = ( f"DRIVER={{{driver}}};" f"SERVER={server},1433;" f"DATABASE={database};" f"UID={username};" f"PWD={password};" "Encrypt=no;" "TrustServerCertificate=yes;" ) conn = pyodbc.connect(conn_str, timeout=10) return conn def verify_password(self, plain_password: str, hashed_password: str) -> bool: """ Verify a password against its hash (bcrypt). Uses passlib first; falls back to bcrypt.checkpw when passlib's bcrypt backend fails (e.g. bcrypt 4.1+). """ # bcrypt truncates to 72 bytes. passlib/bcrypt may raise ValueError instead of truncating, # so we explicitly emulate bcrypt behavior for legacy $2... hashes. candidate_password = plain_password try: if hashed_password and hashed_password.startswith("$2") and len(plain_password.encode("utf-8")) > 72: candidate_password = plain_password.encode("utf-8")[:72].decode("utf-8", errors="ignore") except Exception: candidate_password = plain_password try: return pwd_context.verify(candidate_password, hashed_password) except Exception: if _bcrypt_mod and hashed_password and hashed_password.startswith("$2"): try: return _bcrypt_mod.checkpw( candidate_password.encode("utf-8"), hashed_password.encode("utf-8") if isinstance(hashed_password, str) else hashed_password, ) except Exception: pass return False def get_password_hash(self, password: str) -> str: """ Hash a password. Truncates to 72 UTF-8 bytes before hashing so bcrypt never receives >72 bytes (avoids ValueError from passlib/bcrypt on production). """ raw = password.encode("utf-8") if len(raw) > 72: password = raw[:72].decode("utf-8", errors="ignore") or password[:1] return pwd_context.hash(password) def _get_stored_password_hash(self, cursor, db_type: str, username: str) -> Optional[str]: """ Get stored password hash for a user from plg_user_credentials (if table exists and user has a row). Returns None if table does not exist or user has no stored password (legacy users). """ try: if db_type == "mysql": cursor.execute( "SELECT password_hash FROM plg_user_credentials WHERE username = %s", (username,), ) else: cursor.execute( "SELECT password_hash FROM LRSAdmin.plg_user_credentials WHERE username = ?", (username,), ) row = cursor.fetchone() return row[0] if row else None except Exception: # Table may not exist yet (no users created via UI) return None def authenticate_user(self, username: str, password: str, include_profile: bool = True) -> Optional[Dict]: """ Authenticate user with username and password Args: username: Username password: Plain text password Returns: User dictionary if authenticated, None otherwise """ try: conn = self._get_db_connection() cursor = conn.cursor() # Get user from alf_authority db_type = self.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT id, authority FROM alf_authority WHERE authority = %s AND authority NOT LIKE 'GROUP_%%' AND authority NOT LIKE 'ROLE_%%' """, (username,)) else: cursor.execute(""" SELECT id, authority FROM LRSAdmin.alf_authority WHERE authority = ? AND authority NOT LIKE 'GROUP_%' AND authority NOT LIKE 'ROLE_%' """, (username,)) user_row = cursor.fetchone() if not user_row: cursor.close() conn.close() return None user_id, authority = user_row # Only verify password when the caller provided one (e.g. login). For token-based # lookups (e.g. /auth/me) we pass empty password and should not require verification. if password: stored_hash = self._get_stored_password_hash(cursor, db_type, username) if stored_hash is not None and not self.verify_password(password, stored_hash): cursor.close() conn.close() return None # Always load roles so is_admin/is_super_admin and roles list are correct for /auth/me and login roles = self.get_user_roles(username) groups = self.get_user_groups(username) if include_profile else ["GROUP_EVERYONE"] profile_details: Dict[str, Any] = {} if include_profile: # Load user profile details from alf_node properties profile_details = self._fetch_user_profile(conn, authority) cursor.close() conn.close() is_admin = ("ROLE_ADMINISTRATOR" in roles) or (authority == "admin") return { "id": user_id, "username": authority, "roles": roles, "groups": groups, "is_admin": is_admin, "is_super_admin": is_admin, "primary_role": roles[0] if roles else None, **profile_details, } except Exception as e: err_msg = str(e) if "1045" in err_msg and "using password: NO" in err_msg: print( "❌ Authentication error: MySQL rejected login (no password sent). " "Set DB_PASSWORD in the project root .env file and restart the backend." ) else: print(f"❌ Authentication error: {e}") import traceback traceback.print_exc() return None def get_user_roles(self, username: str) -> List[str]: """ Get roles for a user via alf_authority_alias (auth_id = user id, alias_id = role id). """ try: conn = self._get_db_connection() cursor = conn.cursor() db_type = self.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT DISTINCT r.authority FROM alf_authority u INNER JOIN alf_authority_alias aa ON aa.auth_id = u.id INNER JOIN alf_authority r ON r.id = aa.alias_id WHERE u.authority = %s AND u.authority NOT LIKE 'ROLE_%%' AND u.authority NOT LIKE 'GROUP_%%' AND r.authority LIKE 'ROLE_%%' """, (username,)) else: cursor.execute(""" SELECT DISTINCT r.authority FROM LRSAdmin.alf_authority u INNER JOIN LRSAdmin.alf_authority_alias aa ON aa.auth_id = u.id INNER JOIN LRSAdmin.alf_authority r ON r.id = aa.alias_id WHERE u.authority = ? AND u.authority NOT LIKE 'ROLE_%' AND u.authority NOT LIKE 'GROUP_%' AND r.authority LIKE 'ROLE_%' """, (username,)) roles = [row[0] for row in cursor.fetchall() if row[0] and row[0].startswith('ROLE_')] cursor.close() conn.close() return roles except Exception as e: print(f"❌ Error getting user roles: {e}") return [] def get_user_groups(self, username: str) -> List[str]: """ Get groups for a user Args: username: Username Returns: List of group names """ try: conn = self._get_db_connection() cursor = conn.cursor() # For now, return GROUP_EVERYONE for all users # TODO: Implement proper group membership queries when structure is clarified groups = ["GROUP_EVERYONE"] cursor.close() conn.close() return groups except Exception as e: print(f"❌ Error getting user groups: {e}") return ["GROUP_EVERYONE"] def create_access_token(self, data: dict, expires_delta: Optional[timedelta] = None) -> str: """ Create short-lived JWT access token and track session. """ to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) session_id = secrets.token_urlsafe(16) to_encode.update({"exp": expire, "session_id": session_id}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) active_sessions.add(session_id) return encoded_jwt def create_refresh_token(self, data: dict) -> str: """ Create longer-lived refresh token (no session tracking; stateless). Use REFRESH_SECRET_KEY so access and refresh can be rotated independently. """ to_encode = data.copy() to_encode["type"] = "refresh" expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS) to_encode["exp"] = expire return jwt.encode(to_encode, REFRESH_SECRET_KEY, algorithm=ALGORITHM) def verify_refresh_token(self, token: str) -> Optional[Dict]: """Verify refresh token and return payload (sub = username).""" try: payload = jwt.decode(token, REFRESH_SECRET_KEY, algorithms=[ALGORITHM]) if payload.get("type") != "refresh": return None return payload except JWTError: return None def verify_token(self, token: str) -> Optional[Dict]: """ Verify JWT token and return payload Checks: - Token signature and expiration - Token not blacklisted - Session still active Args: token: JWT token string Returns: Token payload if valid, None otherwise """ try: # Check if token is blacklisted if token in blacklisted_tokens: return None payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) # Check if session is still active session_id = payload.get("session_id") if session_id and session_id not in active_sessions: return None return payload except JWTError: return None def logout(self, token: str) -> bool: """ Logout user by invalidating token and ending session Args: token: JWT token to invalidate Returns: True if logout successful, False otherwise """ try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) session_id = payload.get("session_id") # Remove session from active sessions if session_id and session_id in active_sessions: active_sessions.remove(session_id) # Add token to blacklist blacklisted_tokens.add(token) print(f"✅ User logged out: {payload.get('sub')}") return True except JWTError: return False def get_all_users(self) -> List[Dict]: """ Get all users from the system OPTIMIZED: Reuses single database connection instead of opening new connection per user Returns: List of user dictionaries """ try: conn = self._get_db_connection() cursor = conn.cursor() db_type = self.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT id, authority FROM alf_authority WHERE authority NOT LIKE 'GROUP_%%' AND authority NOT LIKE 'ROLE_%%' ORDER BY authority """) else: cursor.execute(""" SELECT id, authority FROM LRSAdmin.alf_authority WHERE authority NOT LIKE 'GROUP_%' AND authority NOT LIKE 'ROLE_%' ORDER BY authority """) user_rows = cursor.fetchall() if not user_rows: cursor.close() conn.close() return [] users = [] for user_id, username in user_rows: # OPTIMIZATION: Use existing connection instead of opening new one # Get roles using existing connection user_roles = self._get_user_roles_with_connection(cursor, username) user_groups = ["GROUP_EVERYONE"] # Default group is_admin = "ROLE_ADMINISTRATOR" in user_roles or username == "admin" users.append({ "id": user_id, "username": username, "roles": user_roles, "groups": user_groups, "is_admin": is_admin, "is_super_admin": is_admin, "primary_role": user_roles[0] if user_roles else None, }) cursor.close() conn.close() return users except Exception as e: print(f"❌ Error getting users: {e}") import traceback traceback.print_exc() return [] def _get_user_roles_with_connection(self, cursor, username: str) -> List[str]: """ Get roles for a user using an existing cursor, via alf_authority_alias. """ try: db_type = self.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT DISTINCT r.authority FROM alf_authority u INNER JOIN alf_authority_alias aa ON aa.auth_id = u.id INNER JOIN alf_authority r ON r.id = aa.alias_id WHERE u.authority = %s AND u.authority NOT LIKE 'ROLE_%%' AND u.authority NOT LIKE 'GROUP_%%' AND r.authority LIKE 'ROLE_%%' """, (username,)) else: cursor.execute(""" SELECT DISTINCT r.authority FROM LRSAdmin.alf_authority u INNER JOIN LRSAdmin.alf_authority_alias aa ON aa.auth_id = u.id INNER JOIN LRSAdmin.alf_authority r ON r.id = aa.alias_id WHERE u.authority = ? AND u.authority NOT LIKE 'ROLE_%' AND u.authority NOT LIKE 'GROUP_%' AND r.authority LIKE 'ROLE_%' """, (username,)) return [row[0] for row in cursor.fetchall() if row[0] and row[0].startswith('ROLE_')] except Exception as e: print(f"⚠️ Error getting roles for {username}: {e}") return [] def _normalize_property_value(self, *values: Any) -> Optional[Any]: for value in values: if value is None: continue if isinstance(value, Decimal): if value == value.to_integral_value(): return int(value) return float(value) if isinstance(value, bytes): try: return value.decode("utf-8", errors="ignore").strip() except Exception: return value.hex() if isinstance(value, str): clean = value.strip() if clean == "": continue return clean return value return None def _to_bool(self, value: Any) -> Optional[bool]: if value is None: return None if isinstance(value, bool): return value if isinstance(value, (int, float, Decimal)): return value != 0 if isinstance(value, str): return value.strip().lower() in ("true", "1", "yes", "y", "enabled") return None def _fetch_user_profile(self, conn, username: str) -> Dict[str, Any]: """ Fetch extended profile details for a user from Alfresco node properties. """ cursor = None try: cursor = conn.cursor() db_type = self.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT np.node_id FROM alf_node_properties np JOIN alf_qname q ON q.id = np.qname_id WHERE q.local_name IN ('userName', 'username') AND np.string_value = %s ORDER BY np.node_id DESC LIMIT 1 """, (username,)) else: cursor.execute(""" SELECT TOP 1 np.node_id FROM LRSAdmin.alf_node_properties np JOIN LRSAdmin.alf_qname q ON q.id = np.qname_id WHERE q.local_name IN ('userName', 'username') AND np.string_value = ? ORDER BY np.node_id DESC """, (username,)) node_row = cursor.fetchone() if not node_row: return {} node_id = int(node_row[0]) if db_type == "mysql": cursor.execute(""" SELECT q.local_name, np.string_value, np.long_value, np.double_value, np.boolean_value, np.float_value FROM alf_node_properties np JOIN alf_qname q ON q.id = np.qname_id WHERE np.node_id = %s """, (node_id,)) else: cursor.execute(""" SELECT q.local_name, np.string_value, np.long_value, np.double_value, np.boolean_value, np.float_value FROM LRSAdmin.alf_node_properties np JOIN LRSAdmin.alf_qname q ON q.id = np.qname_id WHERE np.node_id = ? """, (node_id,)) raw_properties: Dict[str, Any] = {} for name, string_val, long_val, double_val, bool_val, float_val in cursor.fetchall(): normalized = self._normalize_property_value( string_val, long_val, double_val, bool_val, float_val, ) if normalized is not None: raw_properties[name] = normalized if db_type == "mysql": cursor.execute(""" SELECT audit_creator, audit_created, audit_modifier, audit_modified FROM alf_node WHERE id = %s """, (node_id,)) else: cursor.execute(""" SELECT audit_creator, audit_created, audit_modifier, audit_modified FROM LRSAdmin.alf_node WHERE id = ? """, (node_id,)) audit_row = cursor.fetchone() first_name = raw_properties.get("firstName") or raw_properties.get("firstname") last_name = raw_properties.get("lastName") or raw_properties.get("lastname") full_name = " ".join(filter(None, [first_name, last_name])).strip() if (first_name or last_name) else None email = raw_properties.get("email") job_title = raw_properties.get("jobtitle") or raw_properties.get("jobTitle") organization = raw_properties.get("organizationId") or raw_properties.get("organization") phone = raw_properties.get("telephoneNumber") or raw_properties.get("phone") mobile_phone = raw_properties.get("mobilePhone") or raw_properties.get("mobilephone") enabled_flag = self._to_bool(raw_properties.get("enabled")) safe_properties: Dict[str, Any] = {} for key, value in raw_properties.items(): if isinstance(value, (str, int, float, bool)) or value is None: safe_properties[key] = value else: safe_properties[key] = str(value) # Ensure top-level scalar fields are JSON-serializable strings where appropriate def _as_optional_str(value: Any) -> Optional[str]: if value is None: return None if isinstance(value, str): clean = value.strip() return clean or None # Convert any non-string (e.g. 0, Decimal) to string to satisfy Pydantic's str type return str(value) profile = { "node_id": node_id, "first_name": _as_optional_str(first_name), "last_name": _as_optional_str(last_name), "full_name": _as_optional_str(full_name), "email": _as_optional_str(email), "job_title": _as_optional_str(job_title), "organization": _as_optional_str(organization), "phone": _as_optional_str(phone), "mobile_phone": _as_optional_str(mobile_phone), "enabled": enabled_flag, "properties": safe_properties, "created_at": audit_row[1] if audit_row else None, "updated_at": audit_row[3] if audit_row else None, "created_by": _as_optional_str(audit_row[0] if audit_row else None), "updated_by": _as_optional_str(audit_row[2] if audit_row else None), } return profile except Exception as e: print(f"⚠️ Unable to load profile for user {username}: {e}") return {} finally: if cursor: cursor.close() def get_all_roles(self) -> List[str]: """ Get all roles from the system Returns: List of role names """ try: conn = self._get_db_connection() cursor = conn.cursor() db_type = self.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT authority FROM alf_authority WHERE authority LIKE 'ROLE_%%' ORDER BY authority """) else: cursor.execute(""" SELECT authority FROM LRSAdmin.alf_authority WHERE authority LIKE 'ROLE_%' ORDER BY authority """) roles = [row[0] for row in cursor.fetchall()] # Add named functional roles if db_type == "mysql": cursor.execute(""" SELECT authority FROM alf_authority WHERE authority IN ( 'Surveyor General', 'Director of Town Planning', 'Records Clerk', 'Intake Clerk', 'Deeds Registrar', 'Correction Officer', 'Rent Officer' ) ORDER BY authority """) else: cursor.execute(""" SELECT authority FROM LRSAdmin.alf_authority WHERE authority IN ( 'Surveyor General', 'Director of Town Planning', 'Records Clerk', 'Intake Clerk', 'Deeds Registrar', 'Correction Officer', 'Rent Officer' ) ORDER BY authority """) roles.extend([row[0] for row in cursor.fetchall()]) cursor.close() conn.close() return roles except Exception as e: print(f"❌ Error getting roles: {e}") return [] def get_all_groups(self) -> List[str]: """ Get all groups from the system Returns: List of group names """ try: conn = self._get_db_connection() cursor = conn.cursor() db_type = self.db_config.get("type", "mssql").lower() if db_type == "mysql": cursor.execute(""" SELECT authority FROM alf_authority WHERE authority LIKE 'GROUP_%%' ORDER BY authority """) else: cursor.execute(""" SELECT authority FROM LRSAdmin.alf_authority WHERE authority LIKE 'GROUP_%' ORDER BY authority """) groups = [row[0] for row in cursor.fetchall()] cursor.close() conn.close() return groups except Exception as e: print(f"❌ Error getting groups: {e}") return []