433 lines
18 KiB
Python
433 lines
18 KiB
Python
"""
|
|
app/routes/tenant_settings.py
|
|
------------------------------
|
|
Tenant self-service settings (MT-7). Admin-only blueprint at /settings/.
|
|
|
|
Routes
|
|
------
|
|
GET /settings/ → redirect to /settings/branding
|
|
GET /settings/branding → branding form (company name, logo, colours)
|
|
POST /settings/branding → save branding
|
|
GET /settings/plan → read-only plan + live quota usage
|
|
GET /settings/domains → domain list + add-custom-domain form
|
|
POST /settings/domains/request → add an unverified custom domain request
|
|
POST /settings/domains/<id>/delete → remove a non-primary custom domain request
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
|
|
from flask import (
|
|
Blueprint, render_template, redirect, url_for,
|
|
request, flash, current_app,
|
|
)
|
|
from flask_login import login_required, current_user
|
|
|
|
from app import db
|
|
from app.models.tenant_settings import TenantSettings
|
|
from app.utils.decorators import admin_required
|
|
from app.utils.audit import log_action, ACTION_UPDATE, ACTION_CREATE, ACTION_DELETE
|
|
from app.utils.time_utils import now_eastern
|
|
from app.utils import storage
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
bp = Blueprint('tenant_settings', __name__, url_prefix='/settings')
|
|
|
|
_HEX_RE = re.compile(r'^#[0-9a-fA-F]{6}$')
|
|
_DOMAIN_RE = re.compile(
|
|
r'^(?:[a-z0-9](?:[a-z0-9\-]{0,61}[a-z0-9])?\.)+[a-z]{2,}$'
|
|
)
|
|
|
|
|
|
# ── helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
def _save_logo(file_obj):
|
|
"""Save an uploaded logo via the storage seam; return its key or None.
|
|
|
|
MT-22: this previously wrote straight to ``UPLOAD_FOLDER/logos/`` with
|
|
``file_obj.save()`` — the last direct-to-disk writer in the app. On an
|
|
R2-backed tenant the logo never reached the bucket, and on the local
|
|
backend every tenant's logo landed in one shared directory.
|
|
|
|
Validation stays here by design: ``storage.py`` only moves bytes, and
|
|
callers own extension / magic-byte checks (see its module docstring).
|
|
|
|
The returned key keeps the exact ``uploads/logos/<file>`` shape already
|
|
stored in ``TenantSettings.logo_url``, so no migration and no data rewrite.
|
|
"""
|
|
if not file_obj or not file_obj.filename:
|
|
return None
|
|
allowed = {'png', 'jpg', 'jpeg', 'gif', 'svg', 'webp'}
|
|
ext = file_obj.filename.rsplit('.', 1)[-1].lower()
|
|
if ext not in allowed:
|
|
return None
|
|
# Read magic bytes — reject non-image files (SVG is text, skip magic check)
|
|
if ext != 'svg':
|
|
header = file_obj.read(8)
|
|
file_obj.seek(0)
|
|
magic = {
|
|
b'\xff\xd8\xff': 'jpg',
|
|
b'\x89PNG': 'png',
|
|
b'GIF87a': 'gif',
|
|
b'GIF89a': 'gif',
|
|
b'RIFF': 'webp',
|
|
}
|
|
if not any(header.startswith(m) for m in magic):
|
|
return None
|
|
file_obj.seek(0)
|
|
return storage.save(file_obj, 'logos')
|
|
|
|
|
|
def _delete_logo(logo_url):
|
|
"""Remove a stored logo. Silently ignores missing objects.
|
|
Safety guard: only deletes keys inside the uploads/logos/ subfolder."""
|
|
if not logo_url:
|
|
return # nothing stored — never issue a delete for 'uploads/logos/'
|
|
try:
|
|
# Use only the basename to avoid any path-traversal via the stored URL.
|
|
# All logos are written to 'uploads/logos/' by _save_logo(), so
|
|
# rebuilding the key from the basename is always the correct target.
|
|
key = f'uploads/logos/{os.path.basename(str(logo_url or ""))}'
|
|
storage.delete(key)
|
|
logger.info('SETTINGS | logo_deleted | key=%s', key)
|
|
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)
|
|
|
|
|
|
# ── root redirect ─────────────────────────────────────────────────────────────
|
|
|
|
@bp.route('/')
|
|
@login_required
|
|
@admin_required
|
|
def index():
|
|
return redirect(url_for('tenant_settings.branding'))
|
|
|
|
|
|
# ── branding ──────────────────────────────────────────────────────────────────
|
|
|
|
@bp.route('/branding', methods=['GET', 'POST'])
|
|
@login_required
|
|
@admin_required
|
|
def branding():
|
|
from flask import g
|
|
# Feature gate: branding is only available on Pro/Enterprise.
|
|
# Soft check — show the form but warn if not on a branding plan.
|
|
branding_allowed = True
|
|
if _mt_enabled():
|
|
tenant = getattr(g, 'tenant', None)
|
|
if tenant is not None:
|
|
branding_allowed = tenant.allow_branding
|
|
|
|
settings = TenantSettings.get_or_default()
|
|
if settings is None:
|
|
flash('Branding settings are not available yet — the phase33 migration has not been applied to this tenant DB. '
|
|
'Run: python -m control.tenant_migrate upgrade --tenant <slug>', 'warning')
|
|
return redirect(url_for('dashboard.index'))
|
|
|
|
if request.method == 'POST':
|
|
if not branding_allowed:
|
|
flash('Branding customisation requires a Pro or Enterprise plan.', 'warning')
|
|
return redirect(url_for('tenant_settings.branding'))
|
|
|
|
company_name = request.form.get('company_name', '').strip()[:150] or None
|
|
primary_color = request.form.get('primary_color', '').strip()
|
|
accent_color = request.form.get('accent_color', '').strip()
|
|
support_email = request.form.get('support_email', '').strip()[:255] or None
|
|
|
|
if primary_color and not _HEX_RE.match(primary_color):
|
|
flash('Primary colour must be a valid hex code (e.g. #1a56db).', 'danger')
|
|
return render_template('tenant_settings/branding.html',
|
|
settings=settings, branding_allowed=branding_allowed)
|
|
if accent_color and not _HEX_RE.match(accent_color):
|
|
flash('Accent colour must be a valid hex code (e.g. #16a34a).', 'danger')
|
|
return render_template('tenant_settings/branding.html',
|
|
settings=settings, branding_allowed=branding_allowed)
|
|
|
|
# Logo upload (optional)
|
|
logo_file = request.files.get('logo')
|
|
new_logo_url = _save_logo(logo_file)
|
|
|
|
# Upsert: one row only
|
|
row = TenantSettings.query.first()
|
|
if row is None:
|
|
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:
|
|
row.logo_url = new_logo_url
|
|
elif request.form.get('clear_logo'):
|
|
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')
|
|
return redirect(url_for('tenant_settings.branding'))
|
|
|
|
return render_template('tenant_settings/branding.html',
|
|
settings=settings, branding_allowed=branding_allowed)
|
|
|
|
|
|
# ── plan + usage ──────────────────────────────────────────────────────────────
|
|
|
|
@bp.route('/plan')
|
|
@login_required
|
|
@admin_required
|
|
def plan():
|
|
from flask import g
|
|
|
|
plan_info = None
|
|
quota_usage = {}
|
|
|
|
if _mt_enabled():
|
|
tenant = getattr(g, 'tenant', None)
|
|
if tenant is not None:
|
|
# 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), one axis at a time.
|
|
#
|
|
# These used to be four calls inside a single dict literal in one
|
|
# try/except. Python evaluates every value before assigning, so ONE
|
|
# failing counter discarded the whole dict and the page rendered 0
|
|
# for all four axes — including users and facilities, which were
|
|
# fine. That is exactly how a wrong column name in the issues
|
|
# counter presented as "no usage number ever updates".
|
|
from app.tenancy.quota import (
|
|
count_active_users, count_active_facilities,
|
|
count_inspections_this_month, count_issues_this_month,
|
|
)
|
|
for axis, counter in (
|
|
('users', count_active_users),
|
|
('facilities', count_active_facilities),
|
|
('inspections', count_inspections_this_month),
|
|
('issues', count_issues_this_month),
|
|
):
|
|
try:
|
|
quota_usage[axis] = counter()
|
|
except Exception as exc:
|
|
# None (not 0) so the page shows "—": an unknown count and
|
|
# a genuine zero must not look the same.
|
|
quota_usage[axis] = None
|
|
logger.error('tenant_settings.plan: %s count failed: %s',
|
|
axis, exc)
|
|
|
|
# MT-8: billing fields are already on g.tenant — no extra DB query needed.
|
|
billing_enabled = current_app.config.get('BILLING_ENABLED', False)
|
|
subscription_status = None
|
|
trial_ends_at = None
|
|
has_stripe_customer = False
|
|
invoices = []
|
|
if _mt_enabled():
|
|
tenant_ctx = getattr(g, 'tenant', None)
|
|
if tenant_ctx is not None:
|
|
subscription_status = tenant_ctx.subscription_status
|
|
trial_ends_at = tenant_ctx.trial_ends_at
|
|
if billing_enabled:
|
|
try:
|
|
from control.base import control_session
|
|
from control.models import Tenant as ControlTenant
|
|
with control_session() as s:
|
|
t = s.get(ControlTenant, tenant_ctx.id)
|
|
has_stripe_customer = bool(t and t.stripe_customer_id)
|
|
stripe_customer_id = t.stripe_customer_id if t else None
|
|
except Exception:
|
|
stripe_customer_id = None
|
|
|
|
# Fetch last 10 invoices from Stripe.
|
|
if stripe_customer_id:
|
|
try:
|
|
import stripe as _stripe
|
|
_stripe.api_key = current_app.config.get('STRIPE_SECRET_KEY', '')
|
|
raw = _stripe.Invoice.list(customer=stripe_customer_id, limit=10)
|
|
for inv in raw.auto_paging_iter():
|
|
import datetime as _dt
|
|
invoices.append({
|
|
'id': inv.id,
|
|
'number': inv.number or inv.id,
|
|
'date': _dt.datetime.fromtimestamp(inv.created).strftime('%b %d, %Y'),
|
|
'amount': '${:,.2f}'.format(inv.amount_paid / 100),
|
|
'currency': (inv.currency or 'usd').upper(),
|
|
'status': inv.status,
|
|
'pdf_url': inv.invoice_pdf,
|
|
'hosted_url': inv.hosted_invoice_url,
|
|
})
|
|
if len(invoices) >= 10:
|
|
break
|
|
except Exception as exc:
|
|
logger.error('tenant_settings.plan: stripe invoice fetch failed: %s', exc)
|
|
|
|
return render_template(
|
|
'tenant_settings/plan.html',
|
|
plan_info=plan_info,
|
|
quota_usage=quota_usage,
|
|
billing_enabled=billing_enabled,
|
|
subscription_status=subscription_status,
|
|
trial_ends_at=trial_ends_at,
|
|
has_stripe_customer=has_stripe_customer,
|
|
invoices=invoices,
|
|
)
|
|
|
|
|
|
# ── custom domains ────────────────────────────────────────────────────────────
|
|
|
|
@bp.route('/domains')
|
|
@login_required
|
|
@admin_required
|
|
def domains():
|
|
from flask import g
|
|
domain_allowed = False
|
|
tenant_domains = []
|
|
tenant_slug = None
|
|
|
|
if _mt_enabled():
|
|
tenant = getattr(g, 'tenant', None)
|
|
if tenant is not None:
|
|
domain_allowed = tenant.allow_custom_domain
|
|
tenant_slug = tenant.slug
|
|
from control.base import control_session
|
|
from control.models import TenantDomain
|
|
with control_session() as s:
|
|
rows = (s.query(TenantDomain)
|
|
.filter_by(tenant_id=tenant.id)
|
|
.order_by(TenantDomain.id)
|
|
.all())
|
|
tenant_domains = [
|
|
{
|
|
'id': d.id,
|
|
'domain': d.domain,
|
|
'kind': d.kind,
|
|
'is_primary': d.is_primary,
|
|
'verified': d.verified,
|
|
'tls_status': d.tls_status,
|
|
'verification_token': d.verification_token,
|
|
}
|
|
for d in rows
|
|
]
|
|
|
|
return render_template('tenant_settings/domains.html',
|
|
domain_allowed=domain_allowed,
|
|
tenant_domains=tenant_domains,
|
|
tenant_slug=tenant_slug,
|
|
base_domain=os.environ.get('TENANT_BASE_DOMAIN', 'jqc.app'))
|
|
|
|
|
|
@bp.route('/domains/request', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def request_domain():
|
|
from flask import g
|
|
if not _mt_enabled():
|
|
flash('Custom domains require multi-tenancy to be enabled.', 'warning')
|
|
return redirect(url_for('tenant_settings.domains'))
|
|
|
|
tenant = getattr(g, 'tenant', None)
|
|
if tenant is None or not tenant.allow_custom_domain:
|
|
flash('Custom domains are not available on your current plan.', 'warning')
|
|
return redirect(url_for('tenant_settings.domains'))
|
|
|
|
raw = (request.form.get('domain') or '').strip().lower()
|
|
if not raw or not _DOMAIN_RE.match(raw):
|
|
flash('Enter a valid domain name (e.g. jqc.acme.com).', 'danger')
|
|
return redirect(url_for('tenant_settings.domains'))
|
|
|
|
import secrets as _secrets
|
|
from control.base import control_session
|
|
from control.models import TenantDomain
|
|
|
|
with control_session() as s:
|
|
existing = s.query(TenantDomain).filter_by(domain=raw).first()
|
|
if existing:
|
|
flash(f'{raw} is already registered.', 'warning')
|
|
return redirect(url_for('tenant_settings.domains'))
|
|
|
|
token = _secrets.token_hex(24)
|
|
s.add(TenantDomain(
|
|
tenant_id=tenant.id,
|
|
domain=raw,
|
|
kind='custom',
|
|
is_primary=False,
|
|
verified=False,
|
|
verification_token=token,
|
|
tls_status='pending',
|
|
created_at=now_eastern(),
|
|
))
|
|
|
|
log_action(ACTION_CREATE, 'TenantDomain', None, raw,
|
|
f'tenant_id={tenant.id} requested custom domain')
|
|
logger.info('SETTINGS | domain_request | tenant=%s domain=%s', tenant.slug, raw)
|
|
flash(
|
|
f'Domain {raw} added. Add the DNS TXT record shown below to verify ownership.',
|
|
'success',
|
|
)
|
|
return redirect(url_for('tenant_settings.domains'))
|
|
|
|
|
|
@bp.route('/domains/<int:domain_id>/delete', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def delete_domain(domain_id):
|
|
from flask import g
|
|
if not _mt_enabled():
|
|
flash('Multi-tenancy not enabled.', 'warning')
|
|
return redirect(url_for('tenant_settings.domains'))
|
|
|
|
tenant = getattr(g, 'tenant', None)
|
|
if tenant is None:
|
|
flash('Tenant not resolved.', 'danger')
|
|
return redirect(url_for('tenant_settings.domains'))
|
|
|
|
from control.base import control_session
|
|
from control.models import TenantDomain
|
|
|
|
with control_session() as s:
|
|
d = s.get(TenantDomain, domain_id)
|
|
if d is None or d.tenant_id != tenant.id:
|
|
flash('Domain not found.', 'danger')
|
|
return redirect(url_for('tenant_settings.domains'))
|
|
if d.is_primary:
|
|
flash('Cannot remove the primary subdomain.', 'danger')
|
|
return redirect(url_for('tenant_settings.domains'))
|
|
if d.kind != 'custom':
|
|
flash('Only custom domain requests can be removed here.', 'danger')
|
|
return redirect(url_for('tenant_settings.domains'))
|
|
domain_str = d.domain
|
|
s.delete(d)
|
|
|
|
log_action(ACTION_DELETE, 'TenantDomain', domain_id, domain_str)
|
|
flash(f'Domain {domain_str} removed.', 'success')
|
|
return redirect(url_for('tenant_settings.domains')) |