# ─── Security Routes (insert after line 590 in api_proxy.py) ──────────
# These routes serve the /api/security/* endpoints consumed by
# frontend/src/api/securityApi.ts.
# NOTE: pyotp, qrcode, io, base64, uuid are imported inline inside each
# route handler to avoid polluting the module-level namespace and to match
# the existing api_proxy.py pattern.
# ═══════════════════════════════════════════════════════════════════════
# 2FA — TOTP
# ═══════════════════════════════════════════════════════════════════════
@api_proxy_bp.route('/api/security/2fa/status', methods=['GET'])
@require_auth_json()
def api_security_2fa_status():
"""Return current 2FA status for the authenticated user."""
return jsonify({
'enabled': bool(current_user.totp_enabled),
'totp_enabled': bool(current_user.totp_enabled),
})
@api_proxy_bp.route('/api/security/2fa/setup', methods=['POST'])
@require_auth_json()
def api_security_2fa_setup():
"""Generate a fresh TOTP secret and QR code URL."""
import pyotp
import qrcode
import io
import base64
from app.models import db
# Generate secret
secret = pyotp.random_base32()
# Build otpauth URI
issuer = 'Command Sovereignty'
otpauth_url = pyotp.totp.TOTP(secret).provisioning_uri(
name=current_user.email,
issuer_name=issuer,
)
# Generate QR code as base64 PNG
qr = qrcode.make(otpauth_url)
buf = io.BytesIO()
qr.save(buf, format='PNG')
qr_base64 = base64.b64encode(buf.getvalue()).decode('utf-8')
qr_url = f'data:image/png;base64,{qr_base64}'
# Temporarily store secret on user (not enabled yet)
current_user.totp_secret = secret
db.session.commit()
return jsonify({
'secret': secret,
'qr_url': qr_url,
'otpauth_url': otpauth_url,
'user_email': current_user.email,
})
@api_proxy_bp.route('/api/security/2fa/enable', methods=['POST'])
@require_auth_json()
def api_security_2fa_enable():
"""Verify TOTP code and enable 2FA."""
import pyotp
from app.models import db
data = request.get_json()
if not data:
return jsonify({'error': 'Missing request body'}), 400
code = data.get('code', '').replace(' ', '')
secret = data.get('secret', '')
if not code or not secret:
return jsonify({'error': 'code and secret are required'}), 400
if not current_user.totp_secret:
return jsonify({'error': 'No TOTP secret found — please set up 2FA first'}), 400
# Verify against stored secret (or the provided one)
totp = pyotp.TOTP(secret if secret else current_user.totp_secret)
if not totp.verify(code):
return jsonify({'error': 'Invalid TOTP code'}), 400
current_user.totp_secret = secret if secret else current_user.totp_secret
current_user.totp_enabled = True
db.session.commit()
return jsonify({'success': True, 'message': '2FA enabled successfully'})
@api_proxy_bp.route('/api/security/2fa/disable', methods=['POST'])
@require_auth_json()
def api_security_2fa_disable():
"""Disable 2FA after verifying the user's password."""
from app.models import db
data = request.get_json()
if not data:
return jsonify({'error': 'Missing request body'}), 400
current_password = data.get('currentPassword', '')
if not current_password:
return jsonify({'error': 'currentPassword is required'}), 400
if not current_user.check_password(current_password):
return jsonify({'error': 'Incorrect password'}), 403
current_user.totp_enabled = False
current_user.totp_secret = None
db.session.commit()
return jsonify({'success': True, 'message': '2FA disabled successfully'})
# ═══════════════════════════════════════════════════════════════════════
# API Keys
# ═══════════════════════════════════════════════════════════════════════
@api_proxy_bp.route('/api/security/api-keys', methods=['GET'])
@require_auth_json()
def api_security_api_keys_list():
"""List API keys for the current user (masked keys only)."""
from app.models import ApiKey
keys = ApiKey.query.filter_by(user_id=current_user.id).order_by(
ApiKey.created_at.desc()
).all()
result = []
for key in keys:
result.append({
'id': key.id,
'name': key.name,
'masked_key': (key.prefix or 'csk_') + '••••••••••••••••',
'last_used_at': key.last_used_at.isoformat() if key.last_used_at else None,
'created_at': key.created_at.isoformat() if key.created_at else None,
'expires_at': key.expires_at.isoformat() if key.expires_at else None,
'permissions': key.permissions or {},
})
return jsonify({'success': True, 'data': result})
@api_proxy_bp.route('/api/security/api-keys', methods=['POST'])
@require_auth_json()
def api_security_api_keys_create():
"""Create a new API key — returns the full key only once."""
import uuid
from app.models import db, ApiKey
data = request.get_json()
if not data:
return jsonify({'error': 'Missing request body'}), 400
name = data.get('name', '').strip()
if not name:
return jsonify({'error': 'name is required'}), 400
permissions = data.get('permissions', {})
prefix = data.get('prefix', 'csk_')
expires_at_raw = data.get('expires_at')
# Parse optional expires_at
expires_at = None
if expires_at_raw:
from datetime import timezone as tz
expires_at = datetime.fromisoformat(expires_at_raw.replace('Z', '+00:00'))
# Generate the actual API key
raw_key = prefix + secrets.token_hex(24) # 48 hex chars
hashed = generate_password_hash(raw_key)
api_key = ApiKey(
id=str(uuid.uuid4()),
user_id=current_user.id,
name=name,
prefix=prefix,
hashed_key=hashed,
permissions=permissions,
expires_at=expires_at,
)
db.session.add(api_key)
db.session.commit()
return jsonify({
'success': True,
'data': {
'id': api_key.id,
'name': api_key.name,
'key': raw_key, # Only returned once!
'masked_key': prefix + '••••••••••••••••',
'last_used_at': None,
'created_at': api_key.created_at.isoformat() if api_key.created_at else None,
'expires_at': api_key.expires_at.isoformat() if api_key.expires_at else None,
'permissions': api_key.permissions or {},
}
})
@api_proxy_bp.route('/api/security/api-keys/<key_id>', methods=['PUT'])
@require_auth_json()
def api_security_api_keys_update(key_id):
"""Update the name of an API key."""
from app.models import db, ApiKey
api_key = ApiKey.query.filter_by(id=key_id, user_id=current_user.id).first()
if not api_key:
return jsonify({'error': 'API key not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Missing request body'}), 400
name = data.get('name')
if name:
api_key.name = name.strip()
db.session.commit()
return jsonify({
'success': True,
'data': {
'id': api_key.id,
'name': api_key.name,
}
})
@api_proxy_bp.route('/api/security/api-keys/<key_id>', methods=['DELETE'])
@require_auth_json()
def api_security_api_keys_revoke(key_id):
"""Revoke (delete) an API key."""
from app.models import db, ApiKey
api_key = ApiKey.query.filter_by(id=key_id, user_id=current_user.id).first()
if not api_key:
return jsonify({'error': 'API key not found'}), 404
db.session.delete(api_key)
db.session.commit()
return jsonify({'success': True, 'message': 'API key revoked'})
# ═══════════════════════════════════════════════════════════════════════
# Connected Accounts (via Connector model)
# ═══════════════════════════════════════════════════════════════════════
@api_proxy_bp.route('/api/security/connected-accounts', methods=['GET'])
@require_auth_json()
def api_security_connected_accounts():
"""List connected integrations for the user's company."""
from app.models import Connector
company, _ = _get_user_company()
if not company:
return jsonify({'success': True, 'data': []})
connectors = Connector.query.filter_by(company_id=company.id).all()
# Service display map
service_info = {
'google_ads': {
'display_name': 'Google Ads',
'icon': 'google-ads',
'providerUrl': 'https://ads.google.com',
},
'facebook_ads': {
'display_name': 'Facebook Ads',
'icon': 'facebook-ads',
'providerUrl': 'https://business.facebook.com',
},
'google_analytics': {
'display_name': 'Google Analytics',
'icon': 'google-analytics',
'providerUrl': 'https://analytics.google.com',
},
'servicetitan': {
'display_name': 'ServiceTitan',
'icon': 'servicetitan',
'providerUrl': 'https://www.servicetitan.com',
},
'hubspot': {
'display_name': 'HubSpot',
'icon': 'hubspot',
'providerUrl': 'https://www.hubspot.com',
},
'quickbooks': {
'display_name': 'QuickBooks',
'icon': 'quickbooks',
'providerUrl': 'https://quickbooks.intuit.com',
},
}
result = []
for conn in connectors:
info = service_info.get(conn.service, {
'display_name': conn.service.replace('_', ' ').title(),
'icon': conn.service,
'providerUrl': '',
})
result.append({
'service': conn.service,
'display_name': info['display_name'],
'status': conn.status if conn.status else 'inactive',
'connected_at': conn.created_at.isoformat() if conn.created_at else None,
'last_sync': conn.last_sync_at.isoformat() if conn.last_sync_at else None,
'icon': info['icon'],
'providerUrl': info['providerUrl'],
})
return jsonify({'success': True, 'data': result})
@api_proxy_bp.route('/api/security/connected-accounts/<service>/disconnect', methods=['POST'])
@require_auth_json()
def api_security_disconnect_account(service):
"""Disconnect (disable) a connector for the user's company."""
from app.models import db, Connector
company, _ = _get_user_company()
if not company:
return jsonify({'error': 'No company found'}), 404
connector = Connector.query.filter_by(
company_id=company.id, service=service
).first()
if not connector:
return jsonify({'error': f'No {service} connection found'}), 404
connector.status = 'inactive'
connector.config_encrypted = None
connector.config_json = None
db.session.commit()
return jsonify({'success': True, 'message': f'{service} disconnected'})
# ═══════════════════════════════════════════════════════════════════════
# Active Sessions
# ═══════════════════════════════════════════════════════════════════════
@api_proxy_bp.route('/api/security/sessions', methods=['GET'])
@require_auth_json()
def api_security_sessions():
"""List active sessions for the current user."""
from app.models import UserSession
sessions = UserSession.query.filter_by(
user_id=current_user.id
).order_by(UserSession.created_at.desc()).all()
result = []
for s in sessions:
result.append({
'id': s.id,
'session_id': s.session_id,
'device_info': s.device_info or '',
'ip_address': s.ip_address or '',
'is_current': s.is_current,
'last_seen_at': s.last_seen_at.isoformat() if s.last_seen_at else None,
'created_at': s.created_at.isoformat() if s.created_at else None,
})
return jsonify({'success': True, 'data': result})
@api_proxy_bp.route('/api/security/sessions/<session_id>/revoke', methods=['POST'])
@require_auth_json()
def api_security_revoke_session(session_id):
"""Revoke a specific session by ID."""
from app.models import db, UserSession
session_rec = UserSession.query.filter_by(
id=session_id, user_id=current_user.id
).first()
if not session_rec:
return jsonify({'error': 'Session not found'}), 404
# Prevent revoking the current session
if session_rec.is_current:
return jsonify({'error': 'Cannot revoke the current session'}), 400
db.session.delete(session_rec)
db.session.commit()
return jsonify({'success': True, 'message': 'Session revoked'})
@api_proxy_bp.route('/api/security/sessions/revoke-other', methods=['POST'])
@require_auth_json()
def api_security_revoke_other_sessions():
"""Revoke all sessions except the current one."""
from app.models import db, UserSession
now = datetime.now(timezone.utc)
# Set is_current=False on all, then set True on current
# First find current session
current = UserSession.query.filter_by(
user_id=current_user.id, is_current=True
).first()
if current:
# Delete all non-current sessions
deleted = UserSession.query.filter(
UserSession.user_id == current_user.id,
UserSession.id != current.id,
).delete(synchronize_session='fetch')
# Update last_seen on current
current.last_seen_at = now
else:
# No current session marked; delete all
deleted = UserSession.query.filter_by(
user_id=current_user.id
).delete(synchronize_session='fetch')
db.session.commit()
return jsonify({
'success': True,
'message': f'Revoked {deleted} session(s)',
'count': deleted,
})
# ═══════════════════════════════════════════════════════════════════════
# Login History
# ═══════════════════════════════════════════════════════════════════════
@api_proxy_bp.route('/api/security/login-history', methods=['GET'])
@require_auth_json()
def api_security_login_history():
"""Return the last 20 session records (login history) for the user."""
from app.models import UserSession
sessions = UserSession.query.filter_by(
user_id=current_user.id
).order_by(UserSession.created_at.desc()).limit(20).all()
result = []
for s in sessions:
result.append({
'id': s.id,
'session_id': s.session_id,
'device_info': s.device_info or '',
'ip_address': s.ip_address or '',
'is_current': s.is_current,
'last_seen_at': s.last_seen_at.isoformat() if s.last_seen_at else None,
'created_at': s.created_at.isoformat() if s.created_at else None,
})
return jsonify({'success': True, 'data': result})