""" Module for managing user accounts and authentication. Provides methods for creating, validating, and retrieving user information. """ ''' Copyright 2025-2026 AIIrondev Licensed under the Inventarsystem EULA (Endbenutzer-Lizenzvertrag). See Legal/LICENSE for the full license text. Unauthorized commercial use, SaaS hosting, or removal of branding is prohibited. For commercial licensing inquiries: https://github.com/AIIrondev ''' import hashlib import copy import importlib import logging import re import secrets import string from bson.objectid import ObjectId import Web.modules.database.settings as cfg from Web.modules.database.settings import MongoClient import hmac import os logger = logging.getLogger('app') logger.setLevel(logging.DEBUG) logger.propagate = True def normalize_student_card_id(card_id): """Normalize student card IDs for reliable lookup.""" if card_id is None: return '' return str(card_id).strip().upper() def _clean_name_fragment(value): cleaned = re.sub(r'[^A-Za-zÄÖÜäöüß]', '', str(value or '').strip()) if not cleaned: return '' replacements = { 'ä': 'ae', 'ö': 'oe', 'ü': 'ue', 'ß': 'ss', 'Ä': 'Ae', 'Ö': 'Oe', 'Ü': 'Ue', } for old_char, new_char in replacements.items(): cleaned = cleaned.replace(old_char, new_char) return cleaned def _get_tenant_db(client): """Return the current tenant database for the request, or fall back to default.""" try: from tenant import get_tenant_db return get_tenant_db(client) except Exception: return client[cfg.MONGODB_DB] def _has_tenant_configs(): for module_name in ('tenant', 'Web.tenant'): try: tenant_module = importlib.import_module(module_name) tenant_registry = getattr(tenant_module, 'TENANT_REGISTRY', None) if isinstance(tenant_registry, dict) and tenant_registry: return True except Exception: continue return isinstance(getattr(cfg, 'TENANT_CONFIGS', None), dict) and bool(cfg.TENANT_CONFIGS) def _resolve_request_tenant_db(): for module_name in ('tenant', 'Web.tenant'): try: tenant_module = importlib.import_module(module_name) get_tenant_context = getattr(tenant_module, 'get_tenant_context', None) if not callable(get_tenant_context): continue ctx = get_tenant_context() if ctx and ctx.tenant_id: return ctx.db_name or ctx.resolve_tenant(), ctx.tenant_id if ctx and ctx.db_name and not _has_tenant_configs(): return ctx.db_name, None except Exception as exc: logger.debug("Tenant context import %s failed: %s", module_name, exc) return None, None def build_name_synonym(first_name, last_name=''): """Build a deterministic, non-personalized short alias from 2 letters each.""" first = _clean_name_fragment(first_name) last = _clean_name_fragment(last_name) if first and last: return (first[:2] + last[:2]).title() combined = (first + last) if not combined: return 'User' return combined[:4].title() def build_username_from_name(first_name, last_name=''): """ Build a deterministic username abbreviation from first and last name. Uses 2 letters from each name and stores it lowercase. Args: first_name (str): First name last_name (str): Last name (optional) Returns: str: Generated username """ alias = build_name_synonym(first_name, last_name) return alias.lower() def build_unique_username_from_name(first_name, last_name=''): """ Build a unique username from the first 2 letters of the first name and the first 2 letters of the last name. """ first = _clean_name_fragment(first_name) last = _clean_name_fragment(last_name) base_username = (first[:2] + last[:2]).lower() if not base_username: base_username = 'user' if not get_user(base_username): return base_username suffix = 2 while get_user(f"{base_username}{suffix}"): suffix += 1 return f"{base_username}{suffix}" ACTION_PERMISSION_KEYS = ( 'can_borrow', 'can_insert', 'can_edit', 'can_delete', 'can_manage_users', 'can_manage_settings', 'can_view_logs', ) DEFAULT_ACTION_PERMISSIONS = { 'can_borrow': True, 'can_insert': False, 'can_edit': False, 'can_delete': False, 'can_manage_users': False, 'can_manage_settings': False, 'can_view_logs': False, } DEFAULT_PAGE_PERMISSIONS = { 'home': True, 'tutorial_page': True, 'my_borrowed_items': True, 'notifications_view': True, 'impressum': True, 'license': True, 'library_view': True, 'terminplan': True, 'home_admin': False, 'upload_admin': False, 'library_admin': False, 'admin_borrowings': False, 'library_loans_admin': False, 'admin_damaged_items': False, 'admin_audit_dashboard': False, 'logs': False, 'user_del': False, 'register': False, 'manage_filters': False, 'manage_locations': False, } PERMISSION_PRESETS = { 'standard_user': { 'label': 'Standard (Ausleihe)', 'actions': { 'can_borrow': True, }, 'pages': { 'home': True, 'tutorial_page': True, 'my_borrowed_items': True, 'notifications_view': True, 'impressum': True, 'license': True, 'library_view': True, 'terminplan': True, }, }, 'editor': { 'label': 'Editor (Einfügen/Bearbeiten)', 'actions': { 'can_borrow': True, 'can_insert': True, 'can_edit': True, }, 'pages': { 'home': True, 'tutorial_page': True, 'my_borrowed_items': True, 'notifications_view': True, 'impressum': True, 'license': True, 'library_view': True, 'terminplan': True, 'upload_admin': True, 'library_admin': True, }, }, 'manager': { 'label': 'Manager (inkl. Löschen)', 'actions': { 'can_borrow': True, 'can_insert': True, 'can_edit': True, 'can_delete': True, 'can_manage_settings': True, 'can_view_logs': True, }, 'pages': { 'home': True, 'tutorial_page': True, 'my_borrowed_items': True, 'notifications_view': True, 'impressum': True, 'license': True, 'library_view': True, 'terminplan': True, 'home_admin': True, 'upload_admin': True, 'library_admin': True, 'admin_borrowings': True, 'library_loans_admin': True, 'admin_damaged_items': True, 'admin_audit_dashboard': True, 'logs': True, 'manage_filters': True, 'manage_locations': True, }, }, 'full_access': { 'label': 'Vollzugriff', 'actions': { 'can_borrow': True, 'can_insert': True, 'can_edit': True, 'can_delete': True, 'can_manage_users': True, 'can_manage_settings': True, 'can_view_logs': True, }, 'pages': { 'home': True, 'tutorial_page': True, 'my_borrowed_items': True, 'notifications_view': True, 'impressum': True, 'license': True, 'library_view': True, 'terminplan': True, 'home_admin': True, 'upload_admin': True, 'library_admin': True, 'admin_borrowings': True, 'library_loans_admin': True, 'admin_damaged_items': True, 'admin_audit_dashboard': True, 'logs': True, 'user_del': True, 'register': True, 'manage_filters': True, 'manage_locations': True, }, }, } def _normalize_bool_map(source, defaults): result = dict(defaults) if isinstance(source, dict): for key, value in source.items(): result[str(key)] = bool(value) return result def get_permission_preset_definitions(): return copy.deepcopy(PERMISSION_PRESETS) def build_default_permission_payload(preset_key='standard_user'): selected_key = preset_key if preset_key in PERMISSION_PRESETS else 'standard_user' preset = PERMISSION_PRESETS.get(selected_key, {}) action_defaults = _normalize_bool_map(preset.get('actions', {}), DEFAULT_ACTION_PERMISSIONS) page_defaults = _normalize_bool_map(preset.get('pages', {}), DEFAULT_PAGE_PERMISSIONS) return { 'preset': selected_key, 'actions': action_defaults, 'pages': page_defaults, } def get_effective_permissions(username): user = get_user(username) if not user: return build_default_permission_payload('standard_user') # Admin users always have full access, independent of custom presets. if bool(user.get('Admin', False)): return build_default_permission_payload('full_access') preset_key = user.get('PermissionPreset') or 'standard_user' payload = build_default_permission_payload(preset_key) payload['actions'] = _normalize_bool_map(user.get('ActionPermissions', {}), payload['actions']) payload['pages'] = _normalize_bool_map(user.get('PagePermissions', {}), payload['pages']) return payload def update_user_permissions(username, preset_key, action_permissions=None, page_permissions=None): selected_key = preset_key if preset_key in PERMISSION_PRESETS else 'standard_user' payload = build_default_permission_payload(selected_key) if isinstance(action_permissions, dict): for key, value in action_permissions.items(): payload['actions'][str(key)] = bool(value) if isinstance(page_permissions, dict): for key, value in page_permissions.items(): payload['pages'][str(key)] = bool(value) update_data = { 'PermissionPreset': payload['preset'], 'ActionPermissions': payload['actions'], 'PagePermissions': payload['pages'], } client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] result = users.update_one({'Username': username}, {'$set': update_data}) if result.matched_count == 0: result = users.update_one({'username': username}, {'$set': update_data}) client.close() return result.matched_count > 0 # === FAVORITES MANAGEMENT === def get_favorites(username): """Return a list of favorite item ObjectId strings for the user.""" client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] user = users.find_one({'Username': username}) or users.find_one({'username': username}) client.close() if not user: return [] favs = user.get('favorites', []) # Normalize to strings return [str(f) for f in favs if f] def add_favorite(username, item_id): """Add an item to user's favorites (idempotent).""" try: client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] users.update_one( {'$or': [{'Username': username}, {'username': username}]}, {'$addToSet': {'favorites': ObjectId(item_id)}} ) client.close() return True except Exception: return False def remove_favorite(username, item_id): """Remove an item from user's favorites.""" try: client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] users.update_one( {'$or': [{'Username': username}, {'username': username}]}, {'$pull': {'favorites': ObjectId(item_id)}} ) client.close() return True except Exception: return False def check_password_strength(password): """ Check if a password meets minimum security requirements. Args: password (str): Password to check Returns: bool: True if password is strong enough, False otherwise """ if password is None: return False if len(password) < 12: return False has_lower = any(char.islower() for char in password) has_upper = any(char.isupper() for char in password) has_digit = any(char.isdigit() for char in password) has_symbol = any(not char.isalnum() for char in password) if not (has_lower and has_upper and has_digit and has_symbol): return False return True def hashing(password, salt=None): """ Hasht ein Passwort mit scrypt. - Wenn kein Salt übergeben wird, wird ein sicherer, zufälliger Salt generiert (für neue Passwörter). - Format für neue Hashes: v1$$ """ password_bytes = password.encode('utf-8') # Explizit UTF-8 für Plattformunabhängigkeit if salt is None: # Neuer Benutzer / Passwortänderung -> Dynamischer Salt random_salt = os.urandom(16) hashed = hashlib.scrypt(password_bytes, salt=random_salt, n=16384, r=8, p=1) return f"v1${random_salt.hex()}${hashed.hex()}" else: # Bestehender Benutzer (wird zur Verifizierung aufgerufen) hashed = hashlib.scrypt(password_bytes, salt=salt, n=16384, r=8, p=1) return hashed.hex() def verify_password(provided_password, stored_password_string): """ Verifiziert ein Passwort gegen einen gespeicherten Hash-String. Unterstützt das alte Format (statischer Salt) und das neue Format (v1$...). """ if not stored_password_string: return False # Überprüfung für das neue, sichere Format if stored_password_string.startswith("v1$"): try: _, salt_hex, hash_hex = stored_password_string.split("$") salt_bytes = bytes.fromhex(salt_hex) # Berechne den Hash des eingegebenen Passworts mit dem extrahierten Salt calculated_hash = hashing(provided_password, salt=salt_bytes) # Timing-Attack-sicherer Vergleich return hmac.compare_digest(calculated_hash, hash_hex) except (ValueError, TypeError): logger.error("Ungültiges Hash-Format in der Datenbank entdeckt.") return False else: # Abwärtskompatibilität: Altes Format mit statischem Salt b'some_salt' old_static_salt = b'some_salt' calculated_hash = hashing(provided_password, salt=old_static_salt) return hmac.compare_digest(calculated_hash, stored_password_string) def check_nm_pwd(username, password): """ Überprüft die Kombination aus Benutzername und Passwort (optimiert). """ db_name, tenant_id = _resolve_request_tenant_db() if not db_name: if _has_tenant_configs(): logger.warning("Default DB fallback verweigert, da Tenant-Konfigurationen existieren.") return None db_name = cfg.MONGODB_DB client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) try: db = client[db_name] users = db['users'] query = {'$or': [{'Username': username}, {'username': username}]} user_record = users.find_one(query) if user_record is None: logger.warning("Kein Benutzer für %r in DB %r gefunden.", username, db_name) return None stored_password = user_record.get('Password') or user_record.get('password') if not verify_password(password, stored_password): logger.warning("Falsches Passwort für Benutzer %r in DB %r.", username, db_name) return None # Automatische Migration alter Hashes auf das neue Format if not stored_password.startswith("v1$"): users.update_one({'_id': user_record['_id']}, {'$set': {'Password': hashing(password)}}) return user_record finally: client.close() def add_user( username, password, name='', last_name='', is_student=False, student_card_id=None, max_borrow_days=None, permission_preset='standard_user', action_permissions=None, page_permissions=None, ): """ Add a new user to the database. Args: username (str): Username for the new user password (str): Password for the new user Returns: bool: True if user was added successfully, False if password was too weak """ client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) try: db = _get_tenant_db(client) users = db['users'] if not check_password_strength(password): return False permission_defaults = build_default_permission_payload(permission_preset) if isinstance(action_permissions, dict): for key, value in action_permissions.items(): permission_defaults['actions'][str(key)] = bool(value) if isinstance(page_permissions, dict): for key, value in page_permissions.items(): permission_defaults['pages'][str(key)] = bool(value) user_doc = { 'Username': username, 'Password': hashing(password), 'Admin': False, 'active_ausleihung': None, 'name': name.strip() if name else '', 'last_name': last_name.strip() if last_name else '', 'IsStudent': bool(is_student), 'PermissionPreset': permission_defaults['preset'], 'ActionPermissions': permission_defaults['actions'], 'PagePermissions': permission_defaults['pages'], } normalized_card = normalize_student_card_id(student_card_id) if bool(is_student): if normalized_card: user_doc['StudentCardId'] = normalized_card if max_borrow_days is not None: try: user_doc['MaxBorrowDays'] = int(max_borrow_days) except (TypeError, ValueError): pass users.insert_one(user_doc) return True finally: client.close() def student_card_exists(student_card_id): """Return True if a student card id is already assigned to a user.""" normalized = normalize_student_card_id(student_card_id) if not normalized: return False client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] exists = users.find_one({'StudentCardId': normalized}) is not None client.close() return exists def get_user_by_student_card(student_card_id): """Return user by student card id or None.""" normalized = normalize_student_card_id(student_card_id) if not normalized: return None client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] found_user = users.find_one({'StudentCardId': normalized}) client.close() return found_user def make_admin(username): """ Grant administrator privileges to a user. Args: username (str): Username of the user to promote Returns: bool: True if user was promoted successfully """ client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] result = users.update_one({'Username': username}, {'$set': {'Admin': True}}) if result.matched_count == 0: result = users.update_one({'username': username}, {'$set': {'Admin': True}}) client.close() return result.matched_count > 0 def remove_admin(username): """ Remove administrator privileges from a user. Args: username (str): Username of the user to demote Returns: bool: True if user was demoted successfully """ client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] result = users.update_one({'Username': username}, {'$set': {'Admin': False}}) if result.matched_count == 0: result = users.update_one({'username': username}, {'$set': {'Admin': False}}) client.close() return result.matched_count > 0 def get_user(username): """ Retrieve a specific user by username. Args: username (str): Username to search for Returns: dict: User document or None if not found """ client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) try: def find_in_db(database_name): db = client[database_name] users = db['users'] return users.find_one({'Username': username}) or users.find_one({'username': username}) # Try current tenant first when available tenant_db, tenant_id = _resolve_request_tenant_db() if tenant_db: user = find_in_db(tenant_db) if user: return user if _has_tenant_configs() and tenant_db is None: logger.warning( "Refusing default DB fallback for user lookup because tenant configs exist and no tenant was resolved." ) return None # Fallback to default configured database user = find_in_db(cfg.MONGODB_DB) if user: return user return None finally: client.close() def check_admin(username): """ Check if a user has administrator privileges. Args: username (str): Username to check Returns: bool: True if user is an administrator, False otherwise """ user = get_user(username) return bool(user and user.get('Admin', False)) def update_active_ausleihung(username, id_item, ausleihung): """ Update a user's active borrowing record. Args: username (str): Username of the user id_item (str): ID of the borrowed item ausleihung (str): ID of the borrowing record Returns: bool: True if successful """ client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] users.update_one({'Username': username}, {'$set': {'active_ausleihung': {'Item': id_item, 'Ausleihung': ausleihung}}}) client.close() return True def get_active_ausleihung(username): """ Get a user's active borrowing record. Args: username (str): Username of the user Returns: dict: Active borrowing information or None """ client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] user = users.find_one({'Username': username}) return user['active_ausleihung'] def has_active_borrowing(username): """ Check if a user currently has an active borrowing. Args: username (str): Username to check Returns: bool: True if user has an active borrowing, False otherwise """ try: client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] user = users.find_one({'username': username}) if not user: user = users.find_one({'Username': username}) if not user: client.close() return False has_active = user.get('active_borrowing', False) client.close() return has_active except Exception as e: return False def delete_user(username): """ Delete a user from the database. Administrative function for removing user accounts. Args: username (str): Username of the account to delete Returns: bool: True if user was deleted successfully, False otherwise """ client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] result = users.delete_one({'username': username}) client.close() if result.deleted_count == 0: # Try with different field name client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] result = users.delete_one({'Username': username}) client.close() return result.deleted_count > 0 def update_active_borrowing(username, item_id, status): """ Update a user's active borrowing status. Args: username (str): Username of the user item_id (str): ID of the borrowed item or None if returning status (bool): True if borrowing, False if returning Returns: bool: True if successful, False on error """ try: client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] result = users.update_one( {'username': username}, {'$set': { 'active_borrowing': status, 'borrowed_item': item_id if status else None }} ) if result.matched_count == 0: result = users.update_one( {'Username': username}, {'$set': { 'active_borrowing': status, 'borrowed_item': item_id if status else None }} ) client.close() return result.modified_count > 0 except Exception as e: return False def get_name(username): """ Retrieve the name that is assosiated with the username. Returns: str: String of name """ client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] user = users.find_one({'Username': username}) name = user.get("name") return name def get_last_name(username): """ Retrieve the last_name that is assosiated with the username. Returns: str: String of last_name """ client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] user = users.find_one({'Username': username}) name = user.get("last_name") return name def get_all_users(): """ Retrieve all users from the database. Administrative function for user management. Returns: list: List of all user documents """ try: client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] all_users = list(users.find()) client.close() return all_users except Exception as e: return [] def update_password(username, new_password): """ Update a user's password with a new one. Args: username (str): Username of the user new_password (str): New password to set Returns: bool: True if password was updated successfully, False otherwise """ try: if not check_password_strength(new_password): return False client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] # Hash the new password hashed_password = hashing(new_password) # Update the user's password result = users.update_one( {'Username': username}, {'$set': {'Password': hashed_password}} ) client.close() return result.modified_count > 0 except Exception as e: print(f"Error updating password: {e}") return False def update_user_name(username, name, last_name): """ Update a user's name and last name. Args: username (str): Username of the user name (str): New first name last_name (str): New last name Returns: bool: True if updated successfully, False otherwise """ try: name_alias = build_name_synonym(name, last_name) client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT) db = _get_tenant_db(client) users = db['users'] result = users.update_one( {'Username': username}, {'$set': {'name': name_alias, 'last_name': ''}} ) client.close() return True except Exception as e: print(f"Error updating user name: {e}") return False