Jun 28 - Code optimize - Fix Medium/Low issues
This commit is contained in:
@@ -576,6 +576,7 @@ def impersonate_entry():
|
||||
|
||||
|
||||
@bp.route('/impersonate/end')
|
||||
@login_required
|
||||
def impersonate_end():
|
||||
"""Clear impersonation session keys and redirect back to the control panel."""
|
||||
from flask import session as flask_session
|
||||
|
||||
@@ -75,15 +75,13 @@ def _delete_logo(logo_url):
|
||||
"""Remove a logo file from disk. Silently ignores missing files.
|
||||
Safety guard: only deletes files inside the uploads/logos/ subfolder."""
|
||||
try:
|
||||
# Reconstruct the absolute path from the relative URL stored in the DB
|
||||
# logo_url is like "uploads/logos/<filename>"
|
||||
rel = logo_url.replace('uploads/', '', 1) # → "logos/<filename>"
|
||||
full_path = os.path.join(current_app.config['UPLOAD_FOLDER'], rel)
|
||||
# Use only the basename to avoid any path-traversal via the stored URL.
|
||||
# All logos are written into logos_dir by _save_logo(), so joining the
|
||||
# basename back to that directory is always the correct path.
|
||||
logos_dir = os.path.join(current_app.config['UPLOAD_FOLDER'], 'logos')
|
||||
abs_path = os.path.abspath(full_path)
|
||||
abs_logos = os.path.abspath(logos_dir)
|
||||
# Path traversal guard — only remove files inside logos/
|
||||
if abs_path.startswith(abs_logos + os.sep) and os.path.isfile(abs_path):
|
||||
abs_path = os.path.join(abs_logos, os.path.basename(logo_url))
|
||||
if os.path.isfile(abs_path):
|
||||
os.remove(abs_path)
|
||||
logger.info('SETTINGS | logo_deleted | path=%s', abs_path)
|
||||
except Exception as exc:
|
||||
@@ -153,23 +151,27 @@ def branding():
|
||||
row = TenantSettings()
|
||||
db.session.add(row)
|
||||
|
||||
# Capture old logo path before any modification so we can delete it
|
||||
# from disk only after the DB commit succeeds (avoids orphaned references
|
||||
# if the commit fails after the file has already been removed).
|
||||
old_logo_url = row.logo_url
|
||||
|
||||
row.company_name = company_name
|
||||
row.primary_color = primary_color or '#1a56db'
|
||||
row.accent_color = accent_color or '#16a34a'
|
||||
row.support_email = support_email
|
||||
if new_logo_url:
|
||||
# Delete old logo file from disk before replacing
|
||||
if row.logo_url:
|
||||
_delete_logo(row.logo_url)
|
||||
row.logo_url = new_logo_url
|
||||
elif request.form.get('clear_logo'):
|
||||
if row.logo_url:
|
||||
_delete_logo(row.logo_url)
|
||||
row.logo_url = None
|
||||
row.updated_at = now_eastern()
|
||||
row.updated_by = current_user.id
|
||||
|
||||
db.session.commit()
|
||||
|
||||
# Delete the old logo file only after the commit succeeds.
|
||||
if old_logo_url and (new_logo_url or request.form.get('clear_logo')):
|
||||
_delete_logo(old_logo_url)
|
||||
log_action(ACTION_UPDATE, 'TenantSettings', row.id, 'branding',
|
||||
f'company_name={company_name}')
|
||||
flash('Branding settings saved.', 'success')
|
||||
@@ -193,23 +195,21 @@ def plan():
|
||||
if _mt_enabled():
|
||||
tenant = getattr(g, 'tenant', None)
|
||||
if tenant is not None:
|
||||
from control.base import control_session
|
||||
from control.models import Plan
|
||||
with control_session() as s:
|
||||
p = s.get(Plan, tenant.plan_id)
|
||||
if p:
|
||||
plan_info = {
|
||||
'name': p.name,
|
||||
'code': p.code,
|
||||
'max_users': p.max_users,
|
||||
'max_facilities': p.max_facilities,
|
||||
'max_inspections_month': p.max_inspections_month,
|
||||
'max_issues_month': p.max_issues_month,
|
||||
'allow_mobile_api': p.allow_mobile_api,
|
||||
'allow_scheduled_reports': p.allow_scheduled_reports,
|
||||
'allow_branding': p.allow_branding,
|
||||
'allow_custom_domain': p.allow_custom_domain,
|
||||
}
|
||||
# All plan fields are already materialised on g.tenant by the
|
||||
# before_request resolver — no second control-DB query needed.
|
||||
# plan_code is the slug (e.g. "starter"); .title() gives "Starter".
|
||||
plan_info = {
|
||||
'name': tenant.plan_code.title() if tenant.plan_code else '—',
|
||||
'code': tenant.plan_code,
|
||||
'max_users': tenant.max_users,
|
||||
'max_facilities': tenant.max_facilities,
|
||||
'max_inspections_month': tenant.max_inspections_month,
|
||||
'max_issues_month': tenant.max_issues_month,
|
||||
'allow_mobile_api': tenant.allow_mobile_api,
|
||||
'allow_scheduled_reports': tenant.allow_scheduled_reports,
|
||||
'allow_branding': tenant.allow_branding,
|
||||
'allow_custom_domain': tenant.allow_custom_domain,
|
||||
}
|
||||
|
||||
# Live counts (tenant DB)
|
||||
from app.tenancy.quota import (
|
||||
|
||||
+34
-27
@@ -68,33 +68,40 @@ def init_tenancy(app):
|
||||
# tenant's DB. The session is server-signed so this is safe.
|
||||
imp_id = session.get('impersonating_tenant_id')
|
||||
if imp_id is not None:
|
||||
from control.base import control_session
|
||||
from control.models import Tenant
|
||||
from app.tenancy.context import TenantContext
|
||||
with control_session() as s:
|
||||
t = s.get(Tenant, imp_id)
|
||||
if t and t.status == 'active':
|
||||
plan = t.plan
|
||||
ctx = TenantContext(
|
||||
id=t.id,
|
||||
slug=t.slug,
|
||||
name=t.name,
|
||||
plan_id=t.plan_id,
|
||||
db_uri=t.db_uri,
|
||||
plan_code=plan.code if plan else None,
|
||||
max_users=plan.max_users if plan else None,
|
||||
max_facilities=plan.max_facilities if plan else None,
|
||||
max_inspections_month=plan.max_inspections_month if plan else None,
|
||||
max_issues_month=plan.max_issues_month if plan else None,
|
||||
allow_mobile_api=plan.allow_mobile_api if plan else True,
|
||||
allow_scheduled_reports=plan.allow_scheduled_reports if plan else True,
|
||||
allow_branding=plan.allow_branding if plan else True,
|
||||
allow_custom_domain=plan.allow_custom_domain if plan else True,
|
||||
)
|
||||
g.tenant = ctx
|
||||
g.tenant_engine = get_tenant_engine(ctx)
|
||||
return # skip normal Host resolution
|
||||
# Impersonation target invalid or suspended — clear and fall through
|
||||
try:
|
||||
from control.base import control_session
|
||||
from control.models import Tenant
|
||||
from app.tenancy.context import TenantContext
|
||||
with control_session() as s:
|
||||
t = s.get(Tenant, imp_id)
|
||||
if t and t.status == 'active':
|
||||
plan = t.plan
|
||||
ctx = TenantContext(
|
||||
id=t.id,
|
||||
slug=t.slug,
|
||||
name=t.name,
|
||||
plan_id=t.plan_id,
|
||||
db_uri=t.db_uri,
|
||||
plan_code=plan.code if plan else None,
|
||||
max_users=plan.max_users if plan else None,
|
||||
max_facilities=plan.max_facilities if plan else None,
|
||||
max_inspections_month=plan.max_inspections_month if plan else None,
|
||||
max_issues_month=plan.max_issues_month if plan else None,
|
||||
allow_mobile_api=plan.allow_mobile_api if plan else True,
|
||||
allow_scheduled_reports=plan.allow_scheduled_reports if plan else True,
|
||||
allow_branding=plan.allow_branding if plan else True,
|
||||
allow_custom_domain=plan.allow_custom_domain if plan else True,
|
||||
)
|
||||
g.tenant = ctx
|
||||
g.tenant_engine = get_tenant_engine(ctx)
|
||||
return # skip normal Host resolution
|
||||
except Exception:
|
||||
import logging as _logging
|
||||
_logging.getLogger(__name__).warning(
|
||||
'TENANCY | impersonation_failed | tenant_id=%s', imp_id
|
||||
)
|
||||
# Tenant not found, suspended, or engine error — clear stale session
|
||||
# keys so the next request doesn't retry a permanently failing lookup.
|
||||
session.pop('impersonating_tenant_id', None)
|
||||
session.pop('impersonating_superadmin_id', None)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user