ske087/quality_app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix: Resolve newly created users unable to login - Add modules column support to user creation and login flow

Browse Source 
Changes:
1. Fixed create_user_handler to properly initialize modules JSON for new users
2. Fixed edit_user_handler to manage module assignments instead of non-existent email field
3. Updated settings_handler to select modules column instead of email from users table
4. Added validate_and_repair_user_modules function in setup_complete_database.py to ensure all users have correct module assignments
5. Added create_app_license function to create development license file during database setup
6. Added ensure_app_license function to docker-entrypoint.sh for license creation on container startup
7. Added user modules validation on Flask app startup to repair any malformed modules
8. License file is automatically created with 1-year validity on deployment

This ensures:
- New users created via UI get proper module assignments
- Existing users are validated/repaired on app startup
- Non-superadmin users can login after license check passes
- All deployments have a valid development license by default
This commit is contained in:
Quality App Developer
2026-01-09 13:45:08 +02:00
parent 8faf5cd9fe
commit 07614cf0bb
4 changed files with 281 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
py_app/app/__init__.py
View File

@@ -76,6 +76,88 @@ def create_app():
return None
# Initialize user modules validation and repair on app startup
def validate_user_modules_on_startup():
"""Validate and repair user modules during app startup"""
try:
import mariadb
import json
import os
# Get database config from instance folder
instance_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '../instance'))
config_path = os.path.join(instance_path, 'external_server.conf')
if not os.path.exists(config_path):
print("⚠️ Database config not found, skipping user modules validation")
return
# Parse config
db_config = {}
try:
with open(config_path, 'r') as f:
for line in f:
if '=' in line:
key, value = line.strip().split('=', 1)
db_config[key] = value
db_config = {
'user': db_config.get('username', 'trasabilitate'),
'password': db_config.get('password', 'Initial01!'),
'host': db_config.get('server_domain', 'localhost'),
'port': int(db_config.get('port', 3306)),
'database': db_config.get('database_name', 'trasabilitate')
}
except Exception as e:
print(f"⚠️ Could not parse database config: {e}")
return
# Connect and validate users
conn = mariadb.connect(**db_config)
cursor = conn.cursor()
# Check if users table exists
cursor.execute("SHOW TABLES LIKE 'users'")
if not cursor.fetchone():
print("⚠️ Users table not found, skipping validation")
conn.close()
return
# Get all users and validate/repair modules
cursor.execute("SELECT id, username, role, modules FROM users")
users = cursor.fetchall()
users_repaired = 0
for user_id, username, role, modules in users:
# Determine correct modules
if role == 'superadmin':
correct_modules = None
elif role == 'admin':
correct_modules = json.dumps(['quality', 'warehouse', 'labels', 'daily_mirror'])
elif role in ['manager', 'quality_manager', 'warehouse_manager']:
correct_modules = json.dumps(['quality', 'warehouse'])
else:
correct_modules = json.dumps([])
# Repair if needed
if modules != correct_modules:
cursor.execute("UPDATE users SET modules = %s WHERE id = %s", (correct_modules, user_id))
users_repaired += 1
if users_repaired > 0:
conn.commit()
print(f"✅ User modules validation complete: {users_repaired} users repaired")
cursor.close()
conn.close()
except Exception as e:
print(f"⚠️ Error during user modules validation: {e}")
# Run validation on startup
with app.app_context():
validate_user_modules_on_startup()
# Initialize automatic backup scheduler
from app.backup_scheduler import init_backup_scheduler
init_backup_scheduler(app)

130
py_app/app/db_create_scripts/setup_complete_database.py
View File

@@ -423,14 +423,40 @@ def create_users_table_mariadb():
# Insert superadmin user if not exists
cursor.execute("SELECT COUNT(*) FROM users WHERE username = %s", ('superadmin',))
if cursor.fetchone()[0] == 0:
# Superadmin doesn't need explicit modules (handled at login)
cursor.execute("""
INSERT INTO users (username, password, role)
VALUES (%s, %s, %s)
""", ('superadmin', 'superadmin123', 'superadmin'))
INSERT INTO users (username, password, role, modules)
VALUES (%s, %s, %s, %s)
""", ('superadmin', 'superadmin123', 'superadmin', None))
print_success("Superadmin user created (username: superadmin, password: superadmin123)")
else:
print_success("Superadmin user already exists")
# Create additional role examples (if they don't exist)
cursor.execute("SELECT COUNT(*) FROM roles WHERE name = %s", ('admin',))
if cursor.fetchone()[0] == 0:
cursor.execute("""
INSERT INTO roles (name, access_level, description)
VALUES (%s, %s, %s)
""", ('admin', 'high', 'Administrator with access to all modules'))
print_success("Admin role created")
cursor.execute("SELECT COUNT(*) FROM roles WHERE name = %s", ('manager',))
if cursor.fetchone()[0] == 0:
cursor.execute("""
INSERT INTO roles (name, access_level, description)
VALUES (%s, %s, %s)
""", ('manager', 'medium', 'Manager with access to assigned modules'))
print_success("Manager role created")
cursor.execute("SELECT COUNT(*) FROM roles WHERE name = %s", ('worker',))
if cursor.fetchone()[0] == 0:
cursor.execute("""
INSERT INTO roles (name, access_level, description)
VALUES (%s, %s, %s)
""", ('worker', 'low', 'Worker with limited module access'))
print_success("Worker role created")
conn.commit()
cursor.close()
conn.close()
@@ -740,9 +766,103 @@ password={db_password}
print_error(f"Failed to update external config: {e}")
return False
def validate_and_repair_user_modules():
"""Validate and repair user modules - ensure all users have proper module assignments"""
print_step(11, "Validating and Repairing User Module Assignments")
try:
conn = mariadb.connect(**DB_CONFIG)
cursor = conn.cursor()
import json
# Get all users
cursor.execute("SELECT id, username, role, modules FROM users")
users = cursor.fetchall()
users_updated = 0
users_checked = 0
for user_id, username, role, modules in users:
users_checked += 1
# Determine what modules should be assigned
if role == 'superadmin':
# Superadmin doesn't need explicit modules (set to NULL)
correct_modules = None
elif role == 'admin':
# Admin gets all modules
correct_modules = json.dumps(['quality', 'warehouse', 'labels', 'daily_mirror'])
elif role in ['manager', 'quality_manager', 'warehouse_manager']:
# These roles get quality and warehouse by default
correct_modules = json.dumps(['quality', 'warehouse'])
else:
# worker and others get empty array
correct_modules = json.dumps([])
# Check if modules need to be updated
current_modules = modules
if current_modules != correct_modules:
cursor.execute("""
UPDATE users SET modules = %s WHERE id = %s
""", (correct_modules, user_id))
users_updated += 1
action = "assigned" if correct_modules else "cleared"
print(f" ✓ User '{username}' ({role}): modules {action}")
conn.commit()
cursor.close()
conn.close()
print_success(f"User modules validation complete: {users_checked} users checked, {users_updated} updated")
return True
except Exception as e:
print_error(f"Failed to validate/repair user modules: {e}")
return False
def create_app_license():
"""Create a default app license file for the application"""
print_step(12, "Creating Application License File")
try:
import json
from datetime import datetime, timedelta
# Get instance path
instance_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '../../instance'))
os.makedirs(instance_path, exist_ok=True)
license_path = os.path.join(instance_path, 'app_license.json')
# Check if license already exists
if os.path.exists(license_path):
print_success("License file already exists")
return True
# Create a default license valid for 1 year from today
valid_until = (datetime.utcnow() + timedelta(days=365)).strftime('%Y-%m-%d')
license_data = {
"valid_until": valid_until,
"customer": "Development",
"license_type": "development",
"created_at": datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')
}
with open(license_path, 'w') as f:
json.dump(license_data, f, indent=2)
print_success(f"Application license created (valid until: {valid_until})")
return True
except Exception as e:
print_error(f"Failed to create application license: {e}")
return False
def verify_database_setup():
"""Verify that all tables were created successfully"""
print_step(11, "Verifying Database Setup")
print_step(13, "Verifying Database Setup")
try:
conn = mariadb.connect(**DB_CONFIG)
@@ -825,6 +945,8 @@ def main():
create_database_triggers,
populate_permissions_data,
update_external_config,
validate_and_repair_user_modules, # Validate/repair user modules after all setup
create_app_license, # Create app license file
verify_database_setup
]

56
py_app/app/settings.py
View File

@@ -188,7 +188,7 @@ def settings_handler():
''')
# Get all users from external database
cursor.execute("SELECT id, username, password, role, email FROM users")
cursor.execute("SELECT id, username, password, role, modules FROM users")
users_data = cursor.fetchall()
# Convert to list of dictionaries for template compatibility
@@ -199,7 +199,7 @@ def settings_handler():
'username': user_data[1],
'password': user_data[2],
'role': user_data[3],
'email': user_data[4] if len(user_data) > 4 else None
'modules': user_data[4] if len(user_data) > 4 else None
})
conn.close()
@@ -268,17 +268,24 @@ def create_user_handler():
conn = get_external_db_connection()
cursor = conn.cursor()
# Create users table if it doesn't exist
# Create users table if it doesn't exist - with modules column
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
username VARCHAR(100) UNIQUE NOT NULL,
password VARCHAR(255) NOT NULL,
role VARCHAR(50) NOT NULL,
email VARCHAR(255)
modules JSON DEFAULT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Ensure modules column exists (for backward compatibility)
try:
cursor.execute("SELECT modules FROM users LIMIT 1")
except mariadb.ProgrammingError:
cursor.execute("ALTER TABLE users ADD COLUMN modules JSON DEFAULT NULL")
# Check if the username already exists
cursor.execute("SELECT id FROM users WHERE username = %s", (username,))
if cursor.fetchone():
@@ -286,11 +293,23 @@ def create_user_handler():
conn.close()
return redirect(url_for('main.settings'))
# Create a new user in external MariaDB
# Prepare modules based on role
import json
if role == 'superadmin':
# Superadmin doesn't need explicit modules (handled at login)
user_modules = None
elif role == 'admin':
# Admin gets access to all available modules
user_modules = json.dumps(['quality', 'warehouse', 'labels', 'daily_mirror'])
else:
# Other roles (manager, worker) get no modules by default
user_modules = json.dumps([])
# Create a new user in external MariaDB with modules
cursor.execute("""
INSERT INTO users (username, password, role, email)
INSERT INTO users (username, password, role, modules)
VALUES (%s, %s, %s, %s)
""", (username, password, role, email))
""", (username, password, role, user_modules))
conn.commit()
conn.close()
@@ -310,7 +329,7 @@ def edit_user_handler():
user_id = request.form.get('user_id')
password = request.form.get('password', '').strip()
role = request.form.get('role')
email = request.form.get('email', '').strip() or None # Optional field
modules = request.form.getlist('modules') # Get selected modules
if not user_id or not role:
flash('Missing required fields.')
@@ -328,17 +347,24 @@ def edit_user_handler():
conn.close()
return redirect(url_for('main.settings'))
# Prepare modules JSON
import json
if role == 'superadmin':
user_modules = None # Superadmin doesn't need explicit modules
else:
user_modules = json.dumps(modules) if modules else json.dumps([])
# Update the user's details in external MariaDB
if password: # Only update password if provided
cursor.execute("""
UPDATE users SET password = %s, role = %s, email = %s WHERE id = %s
""", (password, role, email, user_id))
UPDATE users SET password = %s, role = %s, modules = %s WHERE id = %s
""", (password, role, user_modules, user_id))
flash('User updated successfully (including password).')
else: # Just update role and email if no password provided
else: # Just update role and modules if no password provided
cursor.execute("""
UPDATE users SET role = %s, email = %s WHERE id = %s
""", (role, email, user_id))
flash('User role updated successfully.')
UPDATE users SET role = %s, modules = %s WHERE id = %s
""", (role, user_modules, user_id))
flash('User role and modules updated successfully.')
conn.commit()
conn.close()