"""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/<company_id>/...`` 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/<company_id>/connectors
# =============================================================================
@super_admin_proxy_bp.route('/company/<company_id>/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/<company_id>/connectors/<connector_id>', 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/<company_id>/connectors/<connector_id>/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/<company_id>/connectors/<connector_id>/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/<company_id>/connectors/<connector_id>/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/<company_id>/connectors/<connector_id>/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/<company_id>/connectors/<connector_id>/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/<company_id>/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/<company_id>/connectors/<connector_id>', 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/<company_id>/connectors/<connector_id>', 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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/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/<company_id>/billing/*
# =============================================================================
@super_admin_proxy_bp.route('/company/<company_id>/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/<company_id>/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/<company_id>/sms/*
# =============================================================================
@super_admin_proxy_bp.route('/company/<company_id>/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/<company_id>/overview
# =============================================================================
@super_admin_proxy_bp.route('/company/<company_id>/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],
}
})