Optimizing admin interface for user domain, completing security views
This commit is contained in:
@@ -3,11 +3,13 @@ import uuid
|
||||
from datetime import datetime as dt, timezone as tz
|
||||
from flask import request, redirect, flash, render_template, Blueprint, session, current_app, jsonify
|
||||
from flask_security import hash_password, roles_required, roles_accepted, current_user
|
||||
from itsdangerous import URLSafeTimedSerializer
|
||||
from sqlalchemy.exc import SQLAlchemyError
|
||||
import ast
|
||||
|
||||
from common.models.user import User, Tenant, Role, TenantDomain
|
||||
from common.extensions import db, kms_client
|
||||
from common.extensions import db, kms_client, security
|
||||
from common.utils.security_utils import send_confirmation_email, send_reset_email
|
||||
from .user_forms import TenantForm, CreateUserForm, EditUserForm, TenantDomainForm
|
||||
from common.utils.database import Database
|
||||
from common.utils.view_assistants import prepare_table_for_macro
|
||||
@@ -20,11 +22,19 @@ user_bp = Blueprint('user_bp', __name__, url_prefix='/user')
|
||||
@user_bp.before_request
|
||||
def log_before_request():
|
||||
current_app.logger.debug(f"Before request (user_bp): {request.method} {request.url}")
|
||||
if current_user and current_user.is_authenticated:
|
||||
current_app.logger.debug(f"After request (user_bp): Current User: {current_user.email}")
|
||||
else:
|
||||
current_app.logger.debug(f"After request (user_bp): No user logged in")
|
||||
|
||||
|
||||
@user_bp.after_request
|
||||
def log_after_request(response):
|
||||
current_app.logger.debug(f"After request (user_bp): {request.method} {request.url} - Status: {response.status}")
|
||||
if current_user and current_user.is_authenticated:
|
||||
current_app.logger.debug(f"After request (user_bp): Current User: {current_user.email}")
|
||||
else:
|
||||
current_app.logger.debug(f"After request (user_bp): No user logged in")
|
||||
return response
|
||||
|
||||
|
||||
@@ -119,7 +129,7 @@ def edit_tenant(tenant_id):
|
||||
@roles_accepted('Super User', 'Tenant Admin')
|
||||
def user():
|
||||
form = CreateUserForm()
|
||||
form.tenant_id.data = session.get('tenant').get('id') # It is only possible to create users for the session tenant
|
||||
form.tenant_id.data = session.get('tenant').get('id') # It is only possible to create users for the session tenant
|
||||
if form.validate_on_submit():
|
||||
current_app.logger.info(f"Adding User for tenant {session['tenant']['id']} ")
|
||||
if form.password.data != form.confirm_password.data:
|
||||
@@ -133,12 +143,10 @@ def user():
|
||||
password=hashed_password,
|
||||
first_name=form.first_name.data,
|
||||
last_name=form.last_name.data,
|
||||
is_active=form.is_active.data,
|
||||
valid_to=form.valid_to.data,
|
||||
tenant_id=form.tenant_id.data
|
||||
)
|
||||
|
||||
new_user.fs_uniquifier = str(uuid.uuid4())
|
||||
timestamp = dt.now(tz.utc)
|
||||
new_user.created_at = timestamp
|
||||
new_user.updated_at = timestamp
|
||||
@@ -158,8 +166,11 @@ def user():
|
||||
try:
|
||||
db.session.add(new_user)
|
||||
db.session.commit()
|
||||
current_app.logger.debug(f'User {new_user.id} with name {new_user.user_name} added to database')
|
||||
flash('User added successfully.', 'success')
|
||||
security.datastore.set_uniquifier()
|
||||
send_confirmation_email(new_user)
|
||||
current_app.logger.debug(f'User {new_user.id} with name {new_user.user_name} added to database'
|
||||
f'Confirmation email sent to {new_user.email}')
|
||||
flash('User added successfully and confirmation email sent.', 'success')
|
||||
return redirect(prefixed_url_for('user_bp.view_users'))
|
||||
except Exception as e:
|
||||
current_app.logger.error(f'Failed to add user with name {new_user.user_name}. Error: {str(e)}')
|
||||
@@ -179,7 +190,6 @@ def edit_user(user_id):
|
||||
# Populate the user with form data
|
||||
user.first_name = form.first_name.data
|
||||
user.last_name = form.last_name.data
|
||||
user.is_active = form.is_active.data
|
||||
user.valid_to = form.valid_to.data
|
||||
user.updated_at = dt.now(tz.utc)
|
||||
|
||||
@@ -200,7 +210,8 @@ def edit_user(user_id):
|
||||
db.session.commit()
|
||||
flash('User updated successfully.', 'success')
|
||||
return redirect(
|
||||
prefixed_url_for('user_bp.edit_user', user_id=user.id)) # Assuming there's a user profile view to redirect to
|
||||
prefixed_url_for('user_bp.edit_user',
|
||||
user_id=user.id)) # Assuming there's a user profile view to redirect to
|
||||
|
||||
form.roles.data = [role.id for role in user.roles]
|
||||
return render_template('user/edit_user.html', form=form, user_id=user_id)
|
||||
@@ -209,11 +220,17 @@ def edit_user(user_id):
|
||||
@user_bp.route('/select_tenant')
|
||||
@roles_required('Super User')
|
||||
def select_tenant():
|
||||
tenants = Tenant.query.all() # Fetch all tenants from the database
|
||||
page = request.args.get('page', 1, type=int)
|
||||
per_page = request.args.get('per_page', 10, type=int)
|
||||
|
||||
prepared_data = prepare_table_for_macro(tenants, [('id', ''), ('name', ''), ('website', '')])
|
||||
query = Tenant.query.order_by(Tenant.name) # Fetch all tenants from the database
|
||||
|
||||
return render_template('user/select_tenant.html', tenants=prepared_data)
|
||||
pagination = query.paginate(page=page, per_page=per_page)
|
||||
tenants = pagination.items
|
||||
|
||||
rows = prepare_table_for_macro(tenants, [('id', ''), ('name', ''), ('website', '')])
|
||||
|
||||
return render_template('user/select_tenant.html', rows=rows, pagination=pagination)
|
||||
|
||||
|
||||
@user_bp.route('/handle_tenant_selection', methods=['POST'])
|
||||
@@ -239,17 +256,24 @@ def handle_tenant_selection():
|
||||
return redirect(prefixed_url_for('select_tenant'))
|
||||
|
||||
|
||||
@user_bp.route('/view_users/<int:tenant_id>')
|
||||
@user_bp.route('/view_users')
|
||||
@roles_accepted('Super User', 'Tenant Admin')
|
||||
def view_users(tenant_id):
|
||||
tenant_id = int(tenant_id)
|
||||
users = User.query.filter_by(tenant_id=tenant_id).all()
|
||||
def view_users():
|
||||
page = request.args.get('page', 1, type=int)
|
||||
per_page = request.args.get('per_page', 10, type=int)
|
||||
|
||||
prepared_data = prepare_table_for_macro(users, [('id', ''), ('user_name', ''), ('email', '')])
|
||||
current_app.logger.debug(f'prepared_data: {prepared_data}')
|
||||
tenant_id = session.get('tenant').get('id')
|
||||
query = User.query.filter_by(tenant_id=tenant_id).order_by(User.user_name)
|
||||
|
||||
pagination = query.paginate(page=page, per_page=per_page)
|
||||
users = pagination.items
|
||||
|
||||
# prepare table data
|
||||
|
||||
rows = prepare_table_for_macro(users, [('id', ''), ('user_name', ''), ('email', '')])
|
||||
|
||||
# Render the users in a template
|
||||
return render_template('user/view_users.html', users=prepared_data)
|
||||
return render_template('user/view_users.html', rows=rows, pagination=pagination)
|
||||
|
||||
|
||||
@user_bp.route('/handle_user_action', methods=['POST'])
|
||||
@@ -257,25 +281,38 @@ def view_users(tenant_id):
|
||||
def handle_user_action():
|
||||
user_identification = request.form['selected_row']
|
||||
user_id = ast.literal_eval(user_identification).get('value')
|
||||
user = User.query.get_or_404(user_id)
|
||||
action = request.form['action']
|
||||
|
||||
if action == 'edit_user':
|
||||
return redirect(prefixed_url_for('user_bp.edit_user', user_id=user_id))
|
||||
# Add more conditions for other actions
|
||||
return redirect(prefixed_url_for('view_users'))
|
||||
elif action == 'resend_confirmation_email':
|
||||
send_confirmation_email(user)
|
||||
flash(f'Confirmation email sent to {user.email}.', 'success')
|
||||
elif action == 'reset_uniquifier':
|
||||
reset_uniquifier(user)
|
||||
flash(f'Uniquifier reset for {user.user_name}.', 'success')
|
||||
|
||||
return redirect(prefixed_url_for('user_bp.view_users'))
|
||||
|
||||
|
||||
@user_bp.route('/view_tenant_domains/<int:tenant_id>')
|
||||
@user_bp.route('/view_tenant_domains')
|
||||
@roles_accepted('Super User', 'Tenant Admin')
|
||||
def view_tenant_domains(tenant_id):
|
||||
tenant_id = int(tenant_id)
|
||||
tenant_domains = TenantDomain.query.filter_by(tenant_id=tenant_id).all()
|
||||
def view_tenant_domains():
|
||||
page = request.args.get('page', 1, type=int)
|
||||
per_page = request.args.get('per_page', 10, type=int)
|
||||
|
||||
tenant_id = session.get('tenant').get('id')
|
||||
query = TenantDomain.query.filter_by(tenant_id=tenant_id).order_by(TenantDomain.domain)
|
||||
|
||||
pagination = query.paginate(page=page, per_page=per_page)
|
||||
tenant_domains = pagination.items
|
||||
|
||||
# prepare table data
|
||||
prepared_data = prepare_table_for_macro(tenant_domains, [('id', ''), ('domain', ''), ('valid_to', '')])
|
||||
rows = prepare_table_for_macro(tenant_domains, [('id', ''), ('domain', ''), ('valid_to', '')])
|
||||
|
||||
# Render the users in a template
|
||||
return render_template('user/view_tenant_domains.html', tenant_domains=prepared_data)
|
||||
return render_template('user/view_tenant_domains.html', rows=rows, pagination=pagination)
|
||||
|
||||
|
||||
@user_bp.route('/handle_tenant_domain_action', methods=['POST'])
|
||||
@@ -336,7 +373,8 @@ def edit_tenant_domain(tenant_domain_id):
|
||||
f'for tenant {session["tenant"]["id"]}'
|
||||
f'Error: {str(e)}')
|
||||
return redirect(
|
||||
prefixed_url_for('user_bp.view_tenant_domains', tenant_id=session['tenant']['id'])) # Assuming there's a user profile view to redirect to
|
||||
prefixed_url_for('user_bp.view_tenant_domains',
|
||||
tenant_id=session['tenant']['id'])) # Assuming there's a user profile view to redirect to
|
||||
|
||||
return render_template('user/edit_tenant_domain.html', form=form, tenant_domain_id=tenant_domain_id)
|
||||
|
||||
@@ -374,12 +412,23 @@ def generate_chat_api_key():
|
||||
@user_bp.route('/tenant_overview', methods=['GET'])
|
||||
@roles_accepted('Super User', 'Tenant Admin')
|
||||
def tenant_overview():
|
||||
current_app.logger.debug('Rendering tenant overview')
|
||||
current_app.logger.debug(f'current_user: {current_user}')
|
||||
current_app.logger.debug(f'Current user roles: {current_user.roles}')
|
||||
tenant_id = session['tenant']['id']
|
||||
current_app.logger.debug(f'Generating overview for tenant {tenant_id}')
|
||||
tenant = Tenant.query.get_or_404(tenant_id)
|
||||
form = TenantForm(obj=tenant)
|
||||
return render_template('user/tenant_overview.html', form=form)
|
||||
|
||||
|
||||
def reset_uniquifier(user):
|
||||
security.datastore.set_uniquifier(user)
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
send_reset_email(user)
|
||||
|
||||
|
||||
def set_logging_information(obj, timestamp):
|
||||
obj.created_at = timestamp
|
||||
obj.updated_at = timestamp
|
||||
|
||||
Reference in New Issue
Block a user