Jun 28 Optimize code
This commit is contained in:
+58
-1
@@ -518,4 +518,61 @@ def reset_password(token):
|
||||
flash('Your password has been reset successfully. Please log in.', 'success')
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
return render_template('auth/reset_password.html', form=form, user=user)
|
||||
return render_template('auth/reset_password.html', form=form, user=user)
|
||||
|
||||
|
||||
# ── MT-4: Superadmin impersonation ────────────────────────────────────────────
|
||||
|
||||
@bp.route('/impersonate')
|
||||
def impersonate_entry():
|
||||
"""
|
||||
Validate a superadmin impersonation token and bind the session to a tenant.
|
||||
Called by the control panel redirect:
|
||||
GET /auth/impersonate?token=<signed_token>
|
||||
Sets session['impersonating_tenant_id'] which the tenancy middleware reads
|
||||
to short-circuit normal Host resolution for the duration of the session.
|
||||
This route is CSRF-exempt by nature — the HMAC token already provides auth.
|
||||
"""
|
||||
from flask import session as flask_session
|
||||
token = request.args.get('token', '')
|
||||
if not token:
|
||||
flash('Missing impersonation token.', 'danger')
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
try:
|
||||
from control.panel.impersonate import validate_token
|
||||
payload = validate_token(token)
|
||||
except ValueError as e:
|
||||
logger.warning('AUTH | impersonate_invalid | reason=%s', e)
|
||||
flash('Invalid or expired impersonation link.', 'danger')
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
tenant_id = payload.get('tid')
|
||||
superadmin_id = payload.get('said')
|
||||
|
||||
flask_session['impersonating_tenant_id'] = tenant_id
|
||||
flask_session['impersonating_superadmin_id'] = superadmin_id
|
||||
|
||||
logger.info('AUTH | impersonate_start | sa=%s tenant=%s', superadmin_id, tenant_id)
|
||||
|
||||
import os
|
||||
panel_url = f"https://admin.{os.environ.get('TENANT_BASE_DOMAIN', 'jqc.app')}"
|
||||
flash(
|
||||
f'Impersonating tenant #{tenant_id} as superadmin. '
|
||||
f'<a href="{url_for(\"auth.impersonate_end\")}" class="alert-link">'
|
||||
f'End impersonation</a>',
|
||||
'warning',
|
||||
)
|
||||
return redirect(url_for('dashboard.index'))
|
||||
|
||||
|
||||
@bp.route('/impersonate/end')
|
||||
def impersonate_end():
|
||||
"""Clear impersonation session keys and redirect back to the control panel."""
|
||||
from flask import session as flask_session
|
||||
import os
|
||||
flask_session.pop('impersonating_tenant_id', None)
|
||||
flask_session.pop('impersonating_superadmin_id', None)
|
||||
panel_url = f"https://admin.{os.environ.get('TENANT_BASE_DOMAIN', 'jqc.app')}"
|
||||
logger.info('AUTH | impersonate_end | redirecting to panel')
|
||||
return redirect(panel_url)
|
||||
@@ -71,6 +71,25 @@ def _save_logo(file_obj):
|
||||
return f'uploads/logos/{filename}'
|
||||
|
||||
|
||||
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)
|
||||
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):
|
||||
os.remove(abs_path)
|
||||
logger.info('SETTINGS | logo_deleted | path=%s', abs_path)
|
||||
except Exception as exc:
|
||||
logger.warning('SETTINGS | logo_delete_failed | url=%s err=%s', logo_url, exc)
|
||||
|
||||
|
||||
def _mt_enabled():
|
||||
return current_app.config.get('MULTI_TENANT_ENABLED', False)
|
||||
|
||||
@@ -135,8 +154,13 @@ def branding():
|
||||
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
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
{% block title %}{{ title }}{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
{% include '_quota_warning.html' %}
|
||||
<div class="row">
|
||||
<div class="col-md-8 offset-md-2">
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
{% block title %}{{ title }}{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
{% include '_quota_warning.html' %}
|
||||
<div class="row">
|
||||
<div class="col-md-8 offset-md-2">
|
||||
<div class="card shadow-sm">
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
{% extends "base.html" %}
|
||||
{% block title %}New Inspection{% endblock %}
|
||||
{% block content %}
|
||||
{% include '_quota_warning.html' %}
|
||||
<div class="row justify-content-center">
|
||||
<div class="col-lg-6">
|
||||
<div class="card shadow-sm">
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
{% extends "base.html" %}
|
||||
{% block title %}{{ title }}{% endblock %}
|
||||
{% block content %}
|
||||
{% include '_quota_warning.html' %}
|
||||
<div class="row justify-content-center">
|
||||
<div class="col-lg-6">
|
||||
<div class="card shadow-sm">
|
||||
|
||||
@@ -7,6 +7,8 @@ Wires tenant resolution into the Flask request lifecycle.
|
||||
* always clears g.tenant / g.tenant_engine (so downstream code can rely on them)
|
||||
* does NOTHING further when MULTI_TENANT_ENABLED is False → today's behaviour
|
||||
* bypasses static + configured exempt paths (health checks)
|
||||
* MT-4: checks session['impersonating_tenant_id'] and short-circuits Host
|
||||
resolution when a superadmin is impersonating a tenant
|
||||
* otherwise resolves the Host header to a tenant and selects its engine
|
||||
* returns a 404 page for an unknown / unverified / suspended host
|
||||
|
||||
@@ -15,7 +17,7 @@ so each request rebinds via RoutingSession.get_bind against the fresh
|
||||
g.tenant_engine — no teardown handler is needed here.
|
||||
"""
|
||||
|
||||
from flask import g, request, current_app, Response
|
||||
from flask import g, request, current_app, Response, session
|
||||
|
||||
from app.tenancy.resolver import resolve_tenant
|
||||
from app.tenancy.engine_cache import get_tenant_engine
|
||||
@@ -60,6 +62,43 @@ def init_tenancy(app):
|
||||
if _is_exempt(request.path):
|
||||
return
|
||||
|
||||
# ── MT-4: Superadmin impersonation override ───────────────────────
|
||||
# When the control panel places a signed token in the session via
|
||||
# /auth/impersonate, bypass Host resolution and bind directly to that
|
||||
# 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
|
||||
session.pop('impersonating_tenant_id', None)
|
||||
session.pop('impersonating_superadmin_id', None)
|
||||
|
||||
# ── Normal Host → tenant resolution ──────────────────────────────
|
||||
host = (request.host or '').split(':')[0].strip().lower()
|
||||
tenant = resolve_tenant(host)
|
||||
if tenant is None:
|
||||
|
||||
Reference in New Issue
Block a user