946 lines
28 KiB
Python
Executable File
946 lines
28 KiB
Python
Executable File
"""
|
|
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$<salt_hex>$<hash_hex>
|
|
"""
|
|
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
|