Invario/Inventarsystem
3
1
Fork 0
Code Issues 29 Pull Requests Actions Packages Projects Releases 216 Wiki Activity

changes to the encryption to the users and Items

Browse Source
This commit is contained in:
AIIrondev
2026-07-31 15:15:09 +02:00
parent 237788af96
commit 9b12d9a3d0
2 changed files with 593 additions and 1026 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Web/modules/database/items.py
+465 -775
View File
File diff suppressed because it is too large Load Diff
Web/modules/database/user.py
+128 -251
View File
@@ -110,13 +110,6 @@ 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()
@@ -325,7 +318,6 @@ def get_effective_permissions(username):
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'])
@@ -353,10 +345,10 @@ def update_user_permissions(username, preset_key, action_permissions=None, page_
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})
result = users.update_one({'Username': dp.encrypt_text(username)}, {'$set': update_data})
if result.matched_count == 0:
result = users.update_one({'username': username}, {'$set': update_data})
result = users.update_one({'username': dp.encrypt_text(username)}, {'$set': update_data})
client.close()
return result.matched_count > 0
@@ -368,7 +360,7 @@ def get_favorites(username):
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})
user = users.find_one({'Username': dp.encrypt_text(username)}) or users.find_one({'username': dp.encrypt_text(username)})
client.close()
if not user:
return []
@@ -383,7 +375,7 @@ def add_favorite(username, item_id):
db = _get_tenant_db(client)
users = db['users']
users.update_one(
{'$or': [{'Username': username}, {'username': username}]},
{'$or': [{'Username': dp.encrypt_text(username)}, {'username': dp.encrypt_text(username)}]},
{'$addToSet': {'favorites': ObjectId(item_id)}}
)
client.close()
@@ -398,7 +390,7 @@ def remove_favorite(username, item_id):
db = _get_tenant_db(client)
users = db['users']
users.update_one(
{'$or': [{'Username': username}, {'username': username}]},
{'$or': [{'Username': dp.encrypt_text(username)}, {'username': dp.encrypt_text(username)}]},
{'$pull': {'favorites': ObjectId(item_id)}}
)
client.close()
@@ -407,16 +399,9 @@ def remove_favorite(username, item_id):
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
@@ -436,19 +421,15 @@ def check_password_strength(password):
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>
Hasht ein Passwort mit scrypt.
"""
password_bytes = password.encode('utf-8') # Explizit UTF-8 für Plattformunabhängigkeit
password_bytes = password.encode('utf-8')
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()
@@ -456,25 +437,20 @@ def hashing(password, salt=None):
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)
@@ -495,22 +471,21 @@ def check_nm_pwd(username, password):
try:
db = client[db_name]
users = db['users']
query = {'$or': [{'Username': username}, {'username': username}]}
query = {'$or': [{'Username': dp.encrypt_text(username)}, {'username': dp.encrypt_text(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)
logger.warning("Kein Benutzer für %r in DB %r gefunden.", dp.encrypt_text(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)
logger.warning("Falsches Passwort für Benutzer %r in DB %r.", dp.encrypt_text(username), db_name)
return None
# Automatische Migration alter Hashes auf das neue Format
if not stored_password.startswith("v1$"):
if stored_password and not stored_password.startswith("v1$"):
users.update_one({'_id': user_record['_id']}, {'$set': {'Password': hashing(password)}})
return user_record
@@ -532,40 +507,35 @@ def add_user(
):
"""
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
"""
if not check_password_strength(password):
return False
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
safe_name = name.strip() if name else ''
safe_last_name = last_name.strip() if last_name else ''
user_doc = {
'Username': username,
'Username': dp.encrypt_text(username),
'Password': hashing(password),
'Admin': can_admin_preset_based,
'Admin': (permission_preset == "full_access"),
'active_ausleihung': None,
'name': name.strip() if name else '',
'last_name': last_name.strip() if last_name else '',
'name': dp.encrypt_text(safe_name) if safe_name else '',
'last_name': dp.encrypt_text(safe_last_name) if safe_last_name else '',
'IsStudent': bool(is_student),
'PermissionPreset': permission_defaults['preset'],
'ActionPermissions': permission_defaults['actions'],
@@ -602,7 +572,7 @@ def student_card_exists(student_card_id):
def get_user_by_student_card(student_card_id):
"""Return user by student card id or None."""
"""Return user dict by student card id or None."""
normalized = normalize_student_card_id(student_card_id)
if not normalized:
return None
@@ -611,65 +581,44 @@ def get_user_by_student_card(student_card_id):
users = db['student_cards']
found_user = users.find_one({'SchülerName': normalized})
client.close()
# Do not call dp.decrypt_text() here because found_user is a MongoDB dictionary.
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
"""
"""Grant administrator privileges to a user."""
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}})
result = users.update_one({'Username': dp.encrypt_text(username)}, {'$set': {'Admin': True}})
if result.matched_count == 0:
result = users.update_one({'username': username}, {'$set': {'Admin': True}})
result = users.update_one({'username': dp.encrypt_text(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
"""
"""Remove administrator privileges from a user."""
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}})
result = users.update_one({'Username': dp.encrypt_text(username)}, {'$set': {'Admin': False}})
if result.matched_count == 0:
result = users.update_one({'username': username}, {'$set': {'Admin': False}})
result = users.update_one({'username': dp.encrypt_text(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
"""
"""Retrieve a specific user by username."""
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})
return users.find_one({'Username': dp.encrypt_text(username)}) or users.find_one({'username': dp.encrypt_text(username)})
# Try current tenant first when available
tenant_db, tenant_id = _resolve_request_tenant_db()
if tenant_db:
user = find_in_db(tenant_db)
@@ -682,7 +631,6 @@ def get_user(username):
)
return None
# Fallback to default configured database
user = find_in_db(cfg.MONGODB_DB)
if user:
return user
@@ -693,147 +641,89 @@ def get_user(username):
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
"""
"""Check if a user has administrator privileges."""
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
"""
"""Update a user's active borrowing record."""
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}}})
result = users.update_one(
{'Username': dp.encrypt_text(username)},
{'$set': {'active_ausleihung': {'Item': id_item, 'Ausleihung': ausleihung}}}
)
if result.matched_count == 0:
users.update_one(
{'username': dp.encrypt_text(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
"""
"""Get a user's active borrowing record."""
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']
user = users.find_one({'Username': dp.encrypt_text(username)}) or users.find_one({'username': dp.encrypt_text(username)})
client.close()
if not user:
return None
return user.get('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
"""
"""Check if a user currently has an active borrowing."""
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)
user = users.find_one({'username': dp.encrypt_text(username)}) or users.find_one({'Username': dp.encrypt_text(username)})
client.close()
return has_active
if not user:
return False
return user.get('active_borrowing', False)
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
"""
"""Delete a user from the database."""
client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT)
db = _get_tenant_db(client)
users = db['users']
result = users.delete_one({'username': username})
client.close()
result = users.delete_one({'username': dp.encrypt_text(username)})
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()
result = users.delete_one({'Username': dp.encrypt_text(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
"""
"""Update a user's active borrowing status."""
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
}}
)
update_data = {'$set': {'active_borrowing': status, 'borrowed_item': item_id if status else None}}
result = users.update_one({'username': dp.encrypt_text(username)}, update_data)
if result.matched_count == 0:
result = users.update_one(
{'Username': username},
{'$set': {
'active_borrowing': status,
'borrowed_item': item_id if status else None
}}
)
result = users.update_one({'Username': dp.encrypt_text(username)}, update_data)
client.close()
return result.modified_count > 0
except Exception as e:
@@ -841,43 +731,37 @@ def update_active_borrowing(username, item_id, status):
def get_name(username):
"""
Retrieve the name that is assosiated with the username.
Returns:
str: String of name
"""
"""Retrieve the name that is associated with the username."""
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
user = users.find_one({'Username': dp.encrypt_text(username)}) or users.find_one({'username': dp.encrypt_text(username)})
client.close()
if not user or not user.get("name"):
return ""
return dp.decrypt_text(user.get("name"))
def get_last_name(username):
"""
Retrieve the last_name that is assosiated with the username.
Returns:
str: String of last_name
"""
"""Retrieve the last_name that is associated with the username."""
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
user = users.find_one({'Username': dp.encrypt_text(username)}) or users.find_one({'username': dp.encrypt_text(username)})
client.close()
if not user or not user.get("last_name"):
return ""
return dp.decrypt_text(user.get("last_name"))
def get_all_users():
"""
Retrieve all users from the database.
Administrative function for user management.
Returns:
list: List of all user documents
"""
"""Retrieve all users from the database."""
try:
client = MongoClient(cfg.MONGODB_HOST, cfg.MONGODB_PORT)
db = _get_tenant_db(client)
@@ -888,65 +772,58 @@ def get_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
"""
"""Update a user's password with a new one."""
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},
{'Username': dp.encrypt_text(username)},
{'$set': {'Password': hashed_password}}
)
if result.matched_count == 0:
result = users.update_one(
{'username': dp.encrypt_text(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
"""
"""Update a user's name and last name."""
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']
safe_name = dp.encrypt_text(name.strip()) if name else ''
safe_last_name = dp.encrypt_text(last_name.strip()) if last_name else ''
result = users.update_one(
{'Username': username},
{'$set': {'name': name_alias, 'last_name': ''}}
{'Username': dp.encrypt_text(username)},
{'$set': {'name': safe_name, 'last_name': safe_last_name}}
)
if result.matched_count == 0:
result = users.update_one(
{'username': dp.encrypt_text(username)},
{'$set': {'name': safe_name, 'last_name': safe_last_name}}
)
client.close()
return True
except Exception as e:
print(f"Error updating user name: {e}")
return False
return False