"""Super-admin proxy routes — allow super admins to manage ANY customer's integrations. Super admins are not members of customer companies by default, so standard company-scoping checks block them. This blueprint provides proxy endpoints under ``/api/super-admin/company//...`` that: 1. Verify the caller is a super_admin 2. Verify the target company exists 3. Execute the same business logic as the customer-facing endpoint, but with the company_id injected explicitly via ``g.proxy_company_id`` All proxy routes log the action for audit trail via audit_log. """ from __future__ import annotations import logging from functools import wraps from typing import Any, Dict from flask import Blueprint, request, jsonify, g, current_app from flask_login import current_user from ..models import db, Company, UserCompany, User logger = logging.getLogger(__name__) super_admin_proxy_bp = Blueprint("super_admin_proxy", __name__) # ─── Audit helper ──────────────────────────────────────────────────────────── def _audit(company_id: str, action: str, detail: str = "") -> None: """Log super-admin action to the audit trail.""" try: from ..models import AuditLog company = db.session.get(Company, company_id) company_name = company.name if company else company_id entry = AuditLog( user_id=current_user.id, action=action, entity_type="company", entity_id=company_id, new_values={ "detail": f"[super-admin] {action} on {company_name}: {detail}".strip(), }, ip_address=(request.remote_addr or "")[:50], user_agent=(request.headers.get("User-Agent", "") or "")[:500], ) db.session.add(entry) db.session.commit() except Exception: db.session.rollback() logger.exception("Failed to audit-log super-admin action") def _serialize_connector(c) -> Dict[str, Any]: """Serialize a Connector row (model has no to_dict method). Matches the format used by connectors.list_connectors. """ from app.connectors import get_connector_class, OAuthConnector, get_connector_metadata return { "id": c.id, "service": c.service, "serviceDisplayName": (get_connector_metadata(c.service) or {}).get( "display_name", c.service.replace("_", " ").title() ), "status": c.status, "auth_type": "oauth2" if issubclass(get_connector_class(c.service) or object, OAuthConnector) else "manual", "sync_frequency": c.sync_frequency, "last_sync_at": c.last_sync_at.isoformat() if c.last_sync_at else None, "error_message": c.error_message, "created_at": c.created_at.isoformat() if c.created_at else None, } # ─── Route guard ───────────────────────────────────────────────────────────── def require_super_admin(): """Decorator: require super_admin role, return JSON 403.""" def decorator(f): @wraps(f) def decorated(*args, **kwargs): if not current_user.is_authenticated or current_user.role != 'super_admin': return jsonify({'error': 'Forbidden — super_admin access required'}), 403 return f(*args, **kwargs) return decorated return decorator def _require_company(company_id: str) -> Company | None: """Validate the target company exists. Return None if not found.""" company = db.session.get(Company, company_id) return company # ============================================================================= # Connector proxy — /api/super-admin/company//connectors # ============================================================================= @super_admin_proxy_bp.route('/company//connectors', methods=['GET']) @require_super_admin() def proxy_list_connectors(company_id): """List all connectors for a customer company.""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 from ..models import Connector connectors = Connector.query.filter_by(company_id=company_id).all() _audit(company_id, "list_connectors", f"returned {len(connectors)} connectors") return jsonify({ 'success': True, 'data': [_serialize_connector(c) for c in connectors], }) @super_admin_proxy_bp.route('/company//connectors/', methods=['GET']) @require_super_admin() def proxy_get_connector(company_id, connector_id): """Get a single connector config.""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 from ..models import Connector connector = db.session.get(Connector, connector_id) if not connector or connector.company_id != company_id: return jsonify({'error': 'Connector not found'}), 404 _audit(company_id, "get_connector", f"connector_id={connector_id}") return jsonify({'success': True, 'data': _serialize_connector(connector)}) @super_admin_proxy_bp.route('/company//connectors//status', methods=['GET']) @require_super_admin() def proxy_connector_status(company_id, connector_id): """Check connector status (OAuth validity, last sync, etc.).""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 from ..models import Connector connector = db.session.get(Connector, connector_id) if not connector or connector.company_id != company_id: return jsonify({'error': 'Connector not found'}), 404 status = connector.get_status() _audit(company_id, "check_connector_status", f"connector_id={connector_id} status={status.get('status')}") return jsonify({'success': True, 'data': status}) @super_admin_proxy_bp.route('/company//connectors//sync', methods=['POST']) @require_super_admin() def proxy_connector_sync(company_id, connector_id): """Manually trigger a sync for a customer connector.""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 from ..models import Connector connector = db.session.get(Connector, connector_id) if not connector or connector.company_id != company_id: return jsonify({'error': 'Connector not found'}), 404 from ..connectors import build_connector conn = build_connector(connector.service, company_id, connector.config, connector.id) result = conn.sync() _audit(company_id, "manual_sync", f"connector_id={connector_id}") return jsonify({'success': True, 'data': result}) @super_admin_proxy_bp.route('/company//connectors//refresh', methods=['POST']) @require_super_admin() def proxy_connector_refresh(company_id, connector_id): """Refresh OAuth tokens for a customer connector.""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 from ..models import Connector connector = db.session.get(Connector, connector_id) if not connector or connector.company_id != company_id: return jsonify({'error': 'Connector not found'}), 404 from ..connectors import build_connector conn = build_connector(connector.service, company_id, connector.config, connector.id) result = conn.refresh_tokens() _audit(company_id, "refresh_oauth", f"connector_id={connector_id}") return jsonify({'success': True, 'data': result}) @super_admin_proxy_bp.route('/company//connectors//logs', methods=['GET']) @require_super_admin() def proxy_connector_logs(company_id, connector_id): """View sync logs for a customer connector.""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 from ..models import Connector, ConnectorLog connector = db.session.get(Connector, connector_id) if not connector or connector.company_id != company_id: return jsonify({'error': 'Connector not found'}), 404 limit = int(request.args.get('limit', 50)) logs = ConnectorLog.query.filter_by(connector_id=connector_id)\ .order_by(ConnectorLog.created_at.desc()).limit(limit).all() _audit(company_id, "view_connector_logs", f"connector_id={connector_id} limit={limit}") return jsonify({ 'success': True, 'data': [l.to_dict() for l in logs], }) @super_admin_proxy_bp.route('/company//connectors//disconnect', methods=['POST']) @require_super_admin() def proxy_connector_disconnect(company_id, connector_id): """Disconnect a customer connector (clears OAuth tokens).""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 from ..models import Connector connector = db.session.get(Connector, connector_id) if not connector or connector.company_id != company_id: return jsonify({'error': 'Connector not found'}), 404 # Attempt a clean disconnect via the actual connector instance (best effort) try: from ..connectors import build_connector conn = build_connector(connector.service, company_id, connector.config, connector.id) conn.disconnect() except Exception: pass # Remove connector records from database from ..models import ConnectorLog ConnectorLog.query.filter_by(connector_id=connector.id).delete() db.session.delete(connector) db.session.commit() _audit(company_id, "disconnect", f"connector_id={connector_id}") return jsonify({'success': True, 'message': 'Connector disconnected'}) @super_admin_proxy_bp.route('/company//connectors', methods=['POST']) @require_super_admin() def proxy_create_connector(company_id): """Create a connector for a customer company.""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 g.proxy_company_id = company_id endpoint = current_app.view_functions.get('connectors.create_connector') if endpoint is None: return jsonify({'error': 'Connector creation endpoint not found'}), 500 result = endpoint() _audit(company_id, "create_connector", request.get_json(silent=True).get("service", "unknown") if request.is_json else "unknown") return result @super_admin_proxy_bp.route('/company//connectors/', methods=['PUT']) @require_super_admin() def proxy_update_connector(company_id, connector_id): """Update connector configuration for a customer company.""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 from ..models import Connector connector = db.session.get(Connector, connector_id) if not connector or connector.company_id != company_id: return jsonify({'error': 'Connector not found'}), 404 g.proxy_company_id = company_id endpoint = current_app.view_functions.get('connectors.update_connector') if endpoint is None: return jsonify({'error': 'Connector update endpoint not found'}), 500 result = endpoint() _audit(company_id, "update_connector", f"connector_id={connector_id}") return result @super_admin_proxy_bp.route('/company//connectors/', methods=['DELETE']) @require_super_admin() def proxy_delete_connector(company_id, connector_id): """Delete a connector for a customer company.""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 from ..models import Connector connector = db.session.get(Connector, connector_id) if not connector or connector.company_id != company_id: return jsonify({'error': 'Connector not found'}), 404 g.proxy_company_id = company_id endpoint = current_app.view_functions.get('connectors.delete_connector') if endpoint is None: return jsonify({'error': 'Connector delete endpoint not found'}), 500 result = endpoint() _audit(company_id, "delete_connector", f"connector_id={connector_id}") return result # ============================================================================= # Analytics proxy — /api/super-admin/company//analytics/* # # These routes use the g.proxy_company_id mechanism: the analytics module # checks g.proxy_company_id in _get_company_id() before falling back to # the UserCompany query. We set it on g and call the actual analytics # endpoint directly. # ============================================================================= def _call_analytics_endpoint(company_id: str, endpoint_name: str): """Set g.proxy_company_id and invoke the analytics view function by name.""" g.proxy_company_id = company_id endpoint = current_app.view_functions.get(endpoint_name) if endpoint is None: return jsonify({'error': f'Analytics endpoint {endpoint_name} not found'}), 500 return endpoint() def _analytics_endpoint_name(route_path: str) -> str: """Map an analytics route path to its Flask endpoint name. analytics_bp endpoints are registered as 'analytics.function_name'. """ # Route path -> function name mapping _MAP = { '/api/analytics/pipeline': 'analytics.pipeline_overview', '/api/analytics/contacts': 'analytics.contact_analytics', '/api/analytics/deals/velocity': 'analytics.deal_velocity', '/api/analytics/crm/summary': 'analytics.crm_summary', '/api/analytics/forecast/revenue': 'analytics.revenue_forecast', '/api/analytics/forecast/pipeline': 'analytics.pipeline_forecast', '/api/analytics/kpis': 'analytics.kpis', '/api/analytics/goals': 'analytics.get_goals', '/api/analytics/users': 'analytics.get_analytics_users', '/api/analytics/strategic-intelligence/overview': 'analytics.strategic_intelligence_overview', '/api/analytics/strategic-intelligence/waterfall': 'analytics.strategic_intelligence_waterfall', '/api/analytics/strategic-intelligence/root-cause': 'analytics.strategic_intelligence_root_cause', '/api/analytics/strategic-intelligence/contribution': 'analytics.strategic_intelligence_contribution', '/api/analytics/scale-optimization/overview': 'analytics.scale_optimization_overview', '/api/analytics/scale-optimization/run-rate': 'analytics.scale_optimization_run_rate', '/api/analytics/scale-optimization/channels': 'analytics.scale_optimization_channels', '/api/analytics/scale-optimization/matrix': 'analytics.scale_optimization_matrix', '/api/analytics/scale-optimization/recommendations': 'analytics.scale_optimization_recommendations', } return _MAP.get(route_path) @super_admin_proxy_bp.route('/company//analytics/pipeline', methods=['GET']) @require_super_admin() def proxy_analytics_pipeline(company_id): """GET /api/analytics/pipeline for a specific company.""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.pipeline_overview') _audit(company_id, "view_analytics", "pipeline") return result @super_admin_proxy_bp.route('/company//analytics/contacts', methods=['GET']) @require_super_admin() def proxy_analytics_contacts(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.contact_analytics') _audit(company_id, "view_analytics", "contacts") return result @super_admin_proxy_bp.route('/company//analytics/deals/velocity', methods=['GET']) @require_super_admin() def proxy_analytics_deal_velocity(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.deal_velocity') _audit(company_id, "view_analytics", "deal_velocity") return result @super_admin_proxy_bp.route('/company//analytics/crm/summary', methods=['GET']) @require_super_admin() def proxy_analytics_crm_summary(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.crm_summary') _audit(company_id, "view_analytics", "crm_summary") return result @super_admin_proxy_bp.route('/company//analytics/forecast/revenue', methods=['GET']) @require_super_admin() def proxy_analytics_forecast_revenue(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.revenue_forecast') _audit(company_id, "view_analytics", "revenue_forecast") return result @super_admin_proxy_bp.route('/company//analytics/forecast/pipeline', methods=['GET']) @require_super_admin() def proxy_analytics_forecast_pipeline(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.pipeline_forecast') _audit(company_id, "view_analytics", "pipeline_forecast") return result @super_admin_proxy_bp.route('/company//analytics/kpis', methods=['GET']) @require_super_admin() def proxy_analytics_kpis(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.kpis') _audit(company_id, "view_analytics", "kpis") return result @super_admin_proxy_bp.route('/company//analytics/goals', methods=['GET']) @require_super_admin() def proxy_analytics_goals_list(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.get_goals') _audit(company_id, "view_analytics", "goals_list") return result @super_admin_proxy_bp.route('/company//analytics/users', methods=['GET']) @require_super_admin() def proxy_analytics_users(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.get_analytics_users') _audit(company_id, "view_analytics", "users") return result # Strategic intelligence proxies @super_admin_proxy_bp.route('/company//analytics/strategic-intelligence/overview', methods=['GET']) @require_super_admin() def proxy_strategic_intelligence_overview(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.strategic_intelligence_overview') _audit(company_id, "view_analytics", "strategic_intelligence_overview") return result @super_admin_proxy_bp.route('/company//analytics/strategic-intelligence/waterfall', methods=['GET']) @require_super_admin() def proxy_strategic_intelligence_waterfall(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.strategic_intelligence_waterfall') _audit(company_id, "view_analytics", "strategic_intelligence_waterfall") return result @super_admin_proxy_bp.route('/company//analytics/strategic-intelligence/root-cause', methods=['GET']) @require_super_admin() def proxy_strategic_intelligence_root_cause(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.strategic_intelligence_root_cause') _audit(company_id, "view_analytics", "strategic_intelligence_root_cause") return result @super_admin_proxy_bp.route('/company//analytics/strategic-intelligence/contribution', methods=['GET']) @require_super_admin() def proxy_strategic_intelligence_contribution(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.strategic_intelligence_contribution') _audit(company_id, "view_analytics", "strategic_intelligence_contribution") return result # Scale optimization proxies @super_admin_proxy_bp.route('/company//analytics/scale-optimization/overview', methods=['GET']) @require_super_admin() def proxy_scale_optimization_overview(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.scale_optimization_overview') _audit(company_id, "view_analytics", "scale_optimization_overview") return result @super_admin_proxy_bp.route('/company//analytics/scale-optimization/run-rate', methods=['GET']) @require_super_admin() def proxy_scale_optimization_run_rate(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.scale_optimization_run_rate') _audit(company_id, "view_analytics", "scale_optimization_run_rate") return result @super_admin_proxy_bp.route('/company//analytics/scale-optimization/channels', methods=['GET']) @require_super_admin() def proxy_scale_optimization_channels(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.scale_optimization_channels') _audit(company_id, "view_analytics", "scale_optimization_channels") return result @super_admin_proxy_bp.route('/company//analytics/scale-optimization/matrix', methods=['GET']) @require_super_admin() def proxy_scale_optimization_matrix(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.scale_optimization_matrix') _audit(company_id, "view_analytics", "scale_optimization_matrix") return result @super_admin_proxy_bp.route('/company//analytics/scale-optimization/recommendations', methods=['GET']) @require_super_admin() def proxy_scale_optimization_recommendations(company_id): if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 result = _call_analytics_endpoint(company_id, 'analytics.scale_optimization_recommendations') _audit(company_id, "view_analytics", "scale_optimization_recommendations") return result # ============================================================================= # Billing proxy — /api/super-admin/company//billing/* # ============================================================================= @super_admin_proxy_bp.route('/company//billing/subscription', methods=['GET']) @require_super_admin() def proxy_billing_subscription(company_id): """Get subscription status for a customer company.""" company = _require_company(company_id) if company is None: return jsonify({'error': 'Company not found'}), 404 from ..models import Subscription subscription = Subscription.query.filter_by(company_id=company.id).first() if not subscription: _audit(company_id, "view_billing", "no_subscription") return jsonify({ 'subscription': { 'status': 'none', 'plan': 'Starter', 'is_active': False, 'current_period_end': None, 'cancel_at_period_end': False, 'trial_end': None, } }) _audit(company_id, "view_billing", f"status={subscription.stripe_status}") return jsonify({ 'subscription': { 'id': subscription.id, 'status': subscription.stripe_status, 'plan': subscription.plan_id, 'is_active': subscription.is_active, 'current_period_start': subscription.current_period_start.isoformat() if subscription.current_period_start else None, 'current_period_end': subscription.current_period_end.isoformat() if subscription.current_period_end else None, 'cancel_at_period_end': subscription.cancel_at_period_end, 'trial_end': subscription.trial_end.isoformat() if subscription.trial_end else None, } }) @super_admin_proxy_bp.route('/company//billing/portal', methods=['POST']) @require_super_admin() def proxy_billing_portal(company_id): """Open a Stripe Customer Portal session for a customer (super-admin assisted).""" import stripe company = _require_company(company_id) if company is None: return jsonify({'error': 'Company not found'}), 404 if not company.stripe_customer_id: return jsonify({'error': 'No Stripe customer found for this company'}), 404 FRONTEND_URL = current_app.config.get('FRONTEND_URL', 'http://localhost:5003') portal_session = stripe.billing_portal.Session.create( customer=company.stripe_customer_id, return_url=f'{FRONTEND_URL}/app/billing', ) _audit(company_id, "open_billing_portal", "created portal session") return jsonify({'url': portal_session.url}) # ============================================================================= # SMS proxy — /api/super-admin/company//sms/* # ============================================================================= @super_admin_proxy_bp.route('/company//sms/config', methods=['GET']) @require_super_admin() def proxy_sms_config(company_id): """View SMS configuration for a customer company.""" if _require_company(company_id) is None: return jsonify({'error': 'Company not found'}), 404 from ..models import SmsConfig config = SmsConfig.query.filter_by(company_id=company_id).first() _audit(company_id, "view_sms_config", "") if config: return jsonify({'success': True, 'data': config.to_dict()}) return jsonify({'success': True, 'data': None}) # ============================================================================= # Company overview — /api/super-admin/company//overview # ============================================================================= @super_admin_proxy_bp.route('/company//overview', methods=['GET']) @require_super_admin() def proxy_company_overview(company_id): """Get a summary of a customer company: integrations, subscription, users.""" company = _require_company(company_id) if company is None: return jsonify({'error': 'Company not found'}), 404 from ..models import Connector, Subscription, UserCompany, ConnectorLog connectors = Connector.query.filter_by(company_id=company_id).all() subscription = Subscription.query.filter_by(company_id=company_id).first() members = UserCompany.query.filter_by(company_id=company_id).all() # Recent sync activity (last 50 logs across all connectors) recent_syncs = ConnectorLog.query.filter( ConnectorLog.connector_id.in_([c.id for c in connectors]) ).order_by(ConnectorLog.created_at.desc()).limit(50).all() # Connector health summary connector_health = [] for c in connectors: status = c.get_status() connector_health.append({ 'id': c.id, 'type': c.connector_type, 'status': status.get('status', 'unknown'), 'last_sync': c.last_sync_at.isoformat() if c.last_sync_at else None, 'error': status.get('error'), }) _audit(company_id, "view_overview", "") return jsonify({ 'success': True, 'data': { 'company': company.to_dict() if hasattr(company, 'to_dict') else { 'id': company.id, 'name': company.name, }, 'subscription': { 'status': subscription.stripe_status if subscription else 'none', 'plan': subscription.plan_id if subscription else None, 'is_active': subscription.is_active if subscription else False, 'current_period_end': subscription.current_period_end.isoformat() if subscription and subscription.current_period_end else None, }, 'members': len(members), 'connectors': connector_health, 'recent_syncs': [s.to_dict() for s in recent_syncs], } })