""" 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//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 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 uploaded logo to static/uploads/logos/; return relative URL or None.""" 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 import secrets logos_dir = os.path.join(current_app.config['UPLOAD_FOLDER'], 'logos') os.makedirs(logos_dir, exist_ok=True) filename = f'{secrets.token_hex(12)}.{ext}' file_obj.save(os.path.join(logos_dir, filename)) return f'uploads/logos/{filename}' 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 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) 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() 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: 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, } # Live counts (tenant DB) from app.tenancy.quota import ( count_active_users, count_active_facilities, count_inspections_this_month, count_issues_this_month, ) try: quota_usage = { 'users': count_active_users(), 'facilities': count_active_facilities(), 'inspections': count_inspections_this_month(), 'issues': count_issues_this_month(), } except Exception as exc: logger.error('tenant_settings.plan: quota count failed: %s', exc) return render_template('tenant_settings/plan.html', plan_info=plan_info, quota_usage=quota_usage) # ── 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//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'))