diff --git a/app/routes/auth.py b/app/routes/auth.py index ae7dd64..d83f461 100644 --- a/app/routes/auth.py +++ b/app/routes/auth.py @@ -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 diff --git a/app/routes/tenant_settings.py b/app/routes/tenant_settings.py index bc38634..374ff25 100644 --- a/app/routes/tenant_settings.py +++ b/app/routes/tenant_settings.py @@ -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/" - rel = logo_url.replace('uploads/', '', 1) # → "logos/" - 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 ( diff --git a/app/tenancy/middleware.py b/app/tenancy/middleware.py index d5d3bcc..30e0441 100644 --- a/app/tenancy/middleware.py +++ b/app/tenancy/middleware.py @@ -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)