""" 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 Web.modules.inventarsystem.data_protection as dp 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[:3] + last[:3]).title() combined = (first + last) if not combined: return 'User' return combined[:6].title() def build_username_from_name(first_name, last_name=''): """ Build a deterministic username abbreviation from first and last name. Uses 3 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 3 letters of the first name and the first 3 letters of the last name. """ first = _clean_name_fragment(first_name) last = _clean_name_fragment(last_name) base_username = (first[:3] + last[:3]).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') print(preset_key) 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) if permission_preset == "full_access": can_admin_preset_based = True else: can_admin_preset_based = False user_doc = { 'Username': username, 'Password': hashing(password), 'Admin': can_admin_preset_based, '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['student_cards'] exists = users.find_one({'SchülerName': 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['student_cards'] found_user = users.find_one({'SchülerName': 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