"""Auth blueprint — login, signup, logout, user profile, settings, company endpoints.
Split from api_proxy.py monolith for maintainability (July 2026).
Shared utilities (require_auth_json, require_csrf, etc.) are still in api_proxy.py
and imported from there.
"""
from flask import Blueprint, request, jsonify, session as flask_session, send_file
from flask_login import login_user, logout_user, current_user
from werkzeug.security import check_password_hash, generate_password_hash
from datetime import datetime, timezone, timedelta
import re
import secrets
import uuid
from app import db, limiter
from app.routes.api_proxy import (
require_auth_json,
require_super_admin,
_check_company_access,
)
from app.utils.csrf import require_csrf
auth_api_bp = Blueprint('auth_api', __name__)
# ─── Password policy + account lockout ─────────────────────────────────
MAX_FAILED_LOGINS = 5
LOCKOUT_MINUTES = 15
def validate_password_policy(password: str):
"""Validate password complexity.
Rules: minimum 8 characters, and at least 2 of the 4 character classes
(uppercase, lowercase, digit, special).
Returns None if valid, else an error message string.
"""
if not password or len(password) < 8:
return 'Password must be at least 8 characters'
classes = sum([
bool(re.search(r'[A-Z]', password)),
bool(re.search(r'[a-z]', password)),
bool(re.search(r'\d', password)),
bool(re.search(r'[^A-Za-z0-9]', password)),
])
if classes < 2:
return ('Password must contain at least 2 of: uppercase letters, '
'lowercase letters, digits, special characters')
return None
def _as_aware_utc(dt):
"""Ensure a datetime is timezone-aware UTC (SQLite may return naive)."""
if dt is not None and dt.tzinfo is None:
return dt.replace(tzinfo=timezone.utc)
return dt
@auth_api_bp.route('/api/auth/user', methods=['GET'])
def api_auth_user():
"""Return current user info or empty user if not authenticated.
SPA expects { user: { role, tenantId, tenantSlug, companyRole, ... } }
Returns empty response instead of 401 to prevent SPA crash.
"""
if not current_user.is_authenticated:
return jsonify({'user': None})
from app.models import Company, UserCompany
# Get user's companies for tenant info
user_companies = UserCompany.query.filter_by(user_id=current_user.id).all()
tenant_id = None
tenant_slug = None
company_role = None
if user_companies:
first_uc = user_companies[0]
first_company = first_uc.company
if first_company:
tenant_id = first_company.id
tenant_slug = first_company.name.lower().replace(' ', '-').replace('&', 'and').replace(',', '')
company_role = first_uc.role
return jsonify({
'user': {
'id': current_user.id,
'email': current_user.email,
'full_name': current_user.full_name,
'role': current_user.role,
'isSuperAdmin': current_user.role == 'super_admin',
'tenantId': tenant_id,
'tenantSlug': tenant_slug,
'companyRole': company_role,
}
})
@auth_api_bp.route('/api/auth/user', methods=['PUT'])
@require_auth_json()
@require_csrf
def api_auth_update_user():
"""Update current user's profile (full_name)."""
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
full_name = data.get('full_name')
if not full_name or not full_name.strip():
return jsonify({'error': 'full_name is required'}), 400
current_user.full_name = full_name.strip()
from app.models import db
db.session.commit()
return jsonify({
'user': {
'id': current_user.id,
'email': current_user.email,
'full_name': current_user.full_name,
'role': current_user.role,
}
})
@auth_api_bp.route('/api/auth/login', methods=['POST'])
@limiter.limit("10 per minute")
def api_auth_login():
"""SPA login endpoint - accepts JSON, uses Flask-Login for session."""
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
email = data.get('email', '').strip()
password = data.get('password', '')
if not email or not password:
return jsonify({'error': 'Email and password required'}), 400
from app.models import db, User, UserSession
user = User.query.filter_by(email=email).first()
now = datetime.now(timezone.utc)
# Account lockout check (5 failures → 15 min lock)
if user and user.locked_until:
locked_until = _as_aware_utc(user.locked_until)
if locked_until > now:
return jsonify({
'error': 'Account temporarily locked due to too many failed '
'login attempts. Please try again later.'
}), 423
# Lock expired — reset counters
user.locked_until = None
user.failed_login_attempts = 0
db.session.commit()
if not user or not check_password_hash(user.password_hash, password):
if user:
user.failed_login_attempts = (user.failed_login_attempts or 0) + 1
if user.failed_login_attempts >= MAX_FAILED_LOGINS:
user.locked_until = now + timedelta(minutes=LOCKOUT_MINUTES)
user.failed_login_attempts = 0
db.session.commit()
return jsonify({'error': 'Invalid credentials'}), 401
# Check if 2FA is enabled — require TOTP code
totp_code = data.get('totp_code', '').replace(' ', '')
if user.totp_enabled and user.totp_secret:
import pyotp
totp = pyotp.TOTP(user.totp_secret)
if not totp.verify(totp_code):
user.failed_login_attempts = (user.failed_login_attempts or 0) + 1
if user.failed_login_attempts >= MAX_FAILED_LOGINS:
user.locked_until = now + timedelta(minutes=LOCKOUT_MINUTES)
user.failed_login_attempts = 0
db.session.commit()
return jsonify({'error': 'Invalid 2FA code', 'requires_totp': True}), 401
# Successful login — reset lockout counters, track last login
user.failed_login_attempts = 0
user.locked_until = None
user.last_login = now
# Create session record
ua = request.headers.get('User-Agent', 'Unknown')[:255]
ip = request.remote_addr or ''
# Mark other sessions as not current
UserSession.query.filter_by(user_id=user.id).update({'is_current': False})
new_session = UserSession(
user_id=user.id,
session_id=str(uuid.uuid4()),
device_info=ua[:100],
ip_address=ip,
is_current=True,
)
db.session.add(new_session)
db.session.commit()
# Use Flask-Login to properly set session cookie
login_user(user)
flask_session.permanent = True # Ensure session persists beyond request
# Bind this Flask session to the UserSession record for revocation
flask_session['us_id'] = new_session.id
# SPA expects { user: {...} } with tenantId, tenantSlug, isSuperAdmin
from app.models import UserCompany
uc = UserCompany.query.filter_by(user_id=user.id).first()
# SECURITY FIX: derive tenant_slug dynamically from the company name
# instead of using the previous hardcoded 'greenfield-landscaping' value.
if uc and uc.company:
tenant_slug = uc.company.name.lower().replace(' ', '-').replace('&', 'and').replace(',', '')
else:
tenant_slug = None
tenant_id = str(uc.company_id) if uc else 'default'
company_role = uc.role if uc else None
resp = jsonify({
'success': True,
'user': {
'id': user.id,
'email': user.email,
'full_name': user.full_name,
'role': user.role,
'isSuperAdmin': user.role == 'super_admin',
'tenantId': tenant_id,
'tenantSlug': tenant_slug,
'companyRole': company_role,
}
})
return resp
@auth_api_bp.route('/api/auth/signup', methods=['POST'])
@limiter.limit("10 per hour")
def api_auth_signup():
"""SPA signup endpoint — creates User, Company, and UserCompany membership.
The first person to sign up with a company name becomes the company owner.
They can invite others with lower roles later.
"""
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
email = data.get('email', '').strip()
password = data.get('password', '')
# Accept both camelCase (SPA) and snake_case (tests/scripts) field names
company_name = data.get('companyName', data.get('company_name', ''))
full_name = data.get('fullName', data.get('full_name', ''))
plan_interest = data.get('planInterest', 'launch')
# Validation
errors = []
if not email:
errors.append('Email is required')
elif not re.match(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$', email):
errors.append('Please enter a valid email address')
if not password:
errors.append('Password is required')
else:
policy_error = validate_password_policy(password)
if policy_error:
errors.append(policy_error)
if not company_name or not company_name.strip():
# For testing: use a default company name if not provided
company_name = data.get('company', f"{full_name}'s Company" if full_name else 'My Company')
if not full_name or not full_name.strip():
errors.append('Full name is required')
if errors:
return jsonify({'error': errors[0], 'errors': errors}), 400
from app.models import db, User, Company, UserCompany
# Check if user exists
existing = User.query.filter_by(email=email).first()
if existing:
return jsonify({'error': 'An account with this email already exists'}), 409
# Create user (platform role = 'user' by default)
new_user = User(
email=email,
full_name=full_name,
company=company_name
)
new_user.set_password(password)
db.session.add(new_user)
db.session.flush() # Get the user ID
# Create company record
company_slug = company_name.lower().replace(' ', '-').replace('&', 'and').replace(',', '').strip('-')
new_company = Company(
name=company_name,
settings_json={'tier': plan_interest} if plan_interest else {}
)
db.session.add(new_company)
db.session.flush()
# Create UserCompany membership — new user is the owner
membership = UserCompany(
user_id=new_user.id,
company_id=new_company.id,
role='owner'
)
db.session.add(membership)
db.session.commit()
# Use Flask-Login to set session
login_user(new_user)
return jsonify({
'success': True,
'user': {
'id': new_user.id,
'email': new_user.email,
'full_name': new_user.full_name,
'role': new_user.role,
'isSuperAdmin': new_user.role == 'super_admin',
'tenantId': new_company.id,
'tenantSlug': company_slug,
'companyRole': 'owner',
},
'redirect': '/app'
})
@auth_api_bp.route('/api/auth/logout', methods=['POST'])
def api_auth_logout():
"""SPA logout endpoint — always clears session if one exists."""
logout_user()
return jsonify({'success': True, 'redirect': '/auth/login'})
@auth_api_bp.route('/api/auth/invite', methods=['POST'])
@require_auth_json()
@require_csrf
def api_auth_invite():
"""Send an invite to add a user to a company."""
from app.models import db, Company, UserCompany, Invite, User
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
email = data.get('email', '').strip().lower()
role = data.get('role', 'member')
company_id = data.get('company_id')
if not email:
return jsonify({'error': 'Email is required'}), 400
# Validate role
valid_roles = ('owner', 'admin', 'manager', 'member', 'viewer')
if role not in valid_roles:
return jsonify({'error': f'Invalid role. Must be one of: {", ".join(valid_roles)}'}), 400
# Check company access
if not company_id:
memberships = UserCompany.query.filter_by(user_id=current_user.id).all()
if not memberships:
return jsonify({'error': 'No company membership found'}), 403
company_id = memberships[0].company_id
membership = UserCompany.query.filter_by(user_id=current_user.id, company_id=company_id).first()
if not membership:
return jsonify({'error': 'No access to this company'}), 403
# Only owners and admins can invite
if membership.role not in ('owner', 'admin'):
return jsonify({'error': 'Insufficient permissions to invite members'}), 403
# Check company exists
company = Company.query.get(company_id)
if not company:
return jsonify({'error': 'Company not found'}), 404
# Check if already a member (bug fix: previously passed user_id=None)
invited_user = User.query.filter(db.func.lower(User.email) == email).first()
if invited_user:
existing = UserCompany.query.filter_by(
user_id=invited_user.id, company_id=company_id
).first()
if existing:
return jsonify({'error': 'This user is already a member of the company'}), 409
# Check for existing pending invite
existing_invite = Invite.query.filter_by(company_id=company_id, email=email, status='pending').first()
if existing_invite:
return jsonify({'error': 'An active invite already exists for this email'}), 409
# Create invite
invite = Invite(
company_id=company_id,
email=email,
role=role,
token=secrets.token_urlsafe(48),
expires_at=datetime.now(timezone.utc) + timedelta(days=7),
created_by=current_user.id,
)
db.session.add(invite)
db.session.commit()
# Send the invite email
import os
base_url = os.environ.get('APP_BASE_URL', request.host_url.rstrip('/'))
invite_link = f'{base_url}/invite/{invite.token}'
from app.utils.mail import send_email
inviter = current_user.full_name or current_user.email
email_result = send_email(
to=[email],
subject=f'You have been invited to join {company.name} on Command Sovereignty',
html_body=(
f'<p>Hi,</p>'
f'<p>{inviter} has invited you to join <strong>{company.name}</strong> '
f'on Command Sovereignty as a <strong>{role}</strong>.</p>'
f'<p><a href="{invite_link}">Accept your invite</a></p>'
f'<p>This invite expires in 7 days.</p>'
),
text_body=(
f'{inviter} has invited you to join {company.name} on Command Sovereignty '
f'as a {role}.\n\nAccept your invite (expires in 7 days): {invite_link}\n'
),
)
return jsonify({
'success': True,
'invite': {
'id': invite.id,
'email': invite.email,
'role': invite.role,
'expires_at': invite.expires_at.isoformat(),
'company_name': company.name,
},
'invite_link': invite_link,
'email_sent': bool(email_result.get('success')),
}), 201
@auth_api_bp.route('/api/auth/accept', methods=['POST'])
def api_auth_accept():
"""Accept a company invite via token."""
from app.models import db, Invite, UserCompany, User
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
token = data.get('token', '').strip()
user_id = data.get('user_id')
if not token:
return jsonify({'error': 'Invite token is required'}), 400
invite = Invite.query.filter_by(token=token).first()
if not invite:
return jsonify({'error': 'Invalid invite token'}), 404
if invite.status == 'accepted':
return jsonify({'error': 'This invite has already been accepted'}), 409
if invite.status == 'revoked':
return jsonify({'error': 'This invite has been revoked'}), 410
if datetime.now(timezone.utc) > _as_aware_utc(invite.expires_at):
invite.status = 'expired'
db.session.commit()
return jsonify({'error': 'This invite has expired'}), 410
# If user_id provided, check they match the invite email
if user_id:
user = User.query.get(user_id)
if not user:
return jsonify({'error': 'User not found'}), 404
if user.email != invite.email:
return jsonify({'error': 'User email does not match invite'}), 403
uid = user_id
elif current_user.is_authenticated:
if current_user.email != invite.email:
return jsonify({'error': 'Your email does not match this invite'}), 403
uid = current_user.id
else:
return jsonify({'error': 'Must be logged in to accept invite'}), 401
# Add to company
existing = UserCompany.query.filter_by(user_id=uid, company_id=invite.company_id).first()
if existing:
return jsonify({'error': 'You are already a member of this company'}), 409
membership = UserCompany(
user_id=uid,
company_id=invite.company_id,
role=invite.role,
)
db.session.add(membership)
invite.status = 'accepted'
invite.accepted_by = uid
invite.accepted_at = datetime.now(timezone.utc)
db.session.commit()
return jsonify({
'success': True,
'membership': {
'company_id': invite.company_id,
'role': invite.role,
},
'redirect': '/app',
})
@auth_api_bp.route('/api/auth/change', methods=['POST'])
@require_auth_json()
@require_csrf
def api_auth_change():
"""Change password endpoint — verifies current password, then sets new one."""
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
current_password = data.get('currentPassword', '')
new_password = data.get('newPassword', '')
if not current_password or not new_password:
return jsonify({'error': 'Current and new passwords are required'}), 400
if len(new_password) < 8:
return jsonify({'error': 'New password must be at least 8 characters'}), 400
policy_error = validate_password_policy(new_password)
if policy_error:
return jsonify({'error': policy_error}), 400
from app.models import db
if not current_user.check_password(current_password):
return jsonify({'error': 'Current password is incorrect'}), 403
current_user.set_password(new_password)
# Security: revoke all other sessions on password change
from app.models import UserSession
current_session = UserSession.query.filter_by(
user_id=current_user.id, is_current=True
).first()
if current_session:
UserSession.query.filter(
UserSession.user_id == current_user.id,
UserSession.id != current_session.id,
).delete(synchronize_session='fetch')
else:
UserSession.query.filter_by(user_id=current_user.id).delete(
synchronize_session='fetch'
)
db.session.commit()
return jsonify({'success': True, 'message': 'Password updated successfully'})
# ─── User Settings (notification preferences + appearance) ────────────
@auth_api_bp.route('/api/user/settings', methods=['GET'])
@require_auth_json()
def api_user_settings_get():
"""Get all user-level settings as a flat dict { key: value }."""
from app.models import UserSetting
rows = UserSetting.query.filter_by(user_id=current_user.id).all()
result: dict[str, object] = {}
for row in rows:
result[row.key] = row.value if isinstance(row.value, dict) else (row.value or {})
return jsonify({'success': True, 'data': result})
@auth_api_bp.route('/api/user/settings', methods=['PUT'])
@require_auth_json()
@require_csrf
def api_user_settings_put():
"""Bulk upsert user-level settings."""
from app.models import db, UserSetting
data = request.get_json()
if not data or 'settings' not in data:
return jsonify({'error': 'Missing settings payload'}), 400
settings = data['settings']
if not isinstance(settings, dict):
return jsonify({'error': 'settings must be a JSON object'}), 400
for key, value in settings.items():
us = UserSetting.query.filter_by(user_id=current_user.id, key=key).first()
if us:
us.value = value
else:
us = UserSetting(user_id=current_user.id, key=key, value=value)
db.session.add(us)
db.session.commit()
return jsonify({'success': True, 'message': 'Settings updated'})
# ─── Company Settings ─────────────────────────────────────────────────
def _get_user_company():
"""Return (company, membership) for the current user's primary company."""
from app.models import Company, UserCompany, db
membership = UserCompany.query.filter_by(user_id=current_user.id).first()
if not membership:
return None, None
company = db.session.get(Company, membership.company_id)
return company, membership
def _require_company_admin(company, membership):
"""Raise 403 JSON if membership role is not owner or admin."""
if membership and membership.role not in ('owner', 'admin'):
return jsonify({'error': 'Forbidden — requires owner or admin role'}), 403
if not company:
return jsonify({'error': 'Company not found'}), 404
return None
@auth_api_bp.route('/api/user/company', methods=['GET'])
@require_auth_json()
def api_user_company_get():
"""Return the current user's company info."""
from app.models import db
company, membership = _get_user_company()
if not company:
return jsonify({'error': 'No company found'}), 404
tier = (company.settings_json or {}).get('tier', 'starter')
return jsonify({
'success': True,
'data': {
'id': company.id,
'name': company.name,
'industry': company.industry or '',
'size': company.size or '',
'annual_revenue': company.annual_revenue,
'target_revenue': company.target_revenue,
'address': company.address or '',
'city': company.city or '',
'state': company.state or '',
'zip_code': company.zip_code or '',
'website': company.website or '',
'logo_url': company.logo_url or '',
'tier': tier,
'created_at': company.created_at.isoformat() if company.created_at else None,
}
})
@auth_api_bp.route('/api/user/company', methods=['PUT'])
@require_auth_json()
@require_csrf
def api_user_company_put():
"""Update editable company fields. Requires owner/admin role."""
from app.models import db
company, membership = _get_user_company()
blocked = _require_company_admin(company, membership)
if blocked:
return blocked
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request'}), 400
editable_fields = {
'name', 'industry', 'size', 'annual_revenue', 'target_revenue',
'address', 'city', 'state', 'zip_code', 'website', 'logo_url',
}
updated = False
for field in editable_fields:
if field in data:
setattr(company, field, data[field])
updated = True
if not updated:
return jsonify({'success': True, 'data': None})
db.session.commit()
return jsonify({
'success': True,
'data': {
'id': company.id,
'name': company.name,
'industry': company.industry or '',
'size': company.size or '',
'annual_revenue': company.annual_revenue,
'target_revenue': company.target_revenue,
'address': company.address or '',
'city': company.city or '',
'state': company.state or '',
'zip_code': company.zip_code or '',
'website': company.website or '',
'logo_url': company.logo_url or '',
'tier': (company.settings_json or {}).get('tier', 'starter'),
'created_at': company.created_at.isoformat() if company.created_at else None,
}
})
@auth_api_bp.route('/api/user/company-settings', methods=['GET'])
@require_auth_json()
def api_user_company_settings_get():
"""Return all Setting records for the user's company as { key: value }."""
from app.models import Setting
company, _ = _get_user_company()
if not company:
return jsonify({'error': 'No company found'}), 404
rows = Setting.query.filter_by(company_id=company.id).all()
result: dict[str, object] = {}
for row in rows:
result[row.key] = row.value if isinstance(row.value, dict) else (row.value or {})
return jsonify({'success': True, 'data': result})
@auth_api_bp.route('/api/user/company-settings', methods=['PUT'])
@require_auth_json()
@require_csrf
def api_user_company_settings_put():
"""Bulk upsert company-level settings. Requires owner/admin."""
from app.models import db, Setting
company, membership = _get_user_company()
blocked = _require_company_admin(company, membership)
if blocked:
return blocked
data = request.get_json()
if not data or 'settings' not in data:
return jsonify({'error': 'Missing settings payload'}), 400
settings = data['settings']
if not isinstance(settings, dict):
return jsonify({'error': 'settings must be a JSON object'}), 400
for key, value in settings.items():
s = Setting.query.filter_by(company_id=company.id, key=key).first()
if s:
s.value = value
else:
s = Setting(company_id=company.id, key=key, value=value)
db.session.add(s)
db.session.commit()
return jsonify({'success': True, 'message': 'Company settings updated'})
# ═══════════════════════════════════════════════════════════════════════
# Password Reset
# ═══════════════════════════════════════════════════════════════════════
_RESET_TOKEN_TTL = timedelta(hours=1)
_GENERIC_RESET_RESPONSE = {
'success': True,
'message': 'If an account with that email exists, a password reset link has been sent.',
}
def _hash_reset_token(token: str) -> str:
import hashlib
return hashlib.sha256(token.encode('utf-8')).hexdigest()
@auth_api_bp.route('/api/auth/forgot-password', methods=['POST'])
@limiter.limit("5 per hour")
def api_auth_forgot_password():
"""Request a password reset email.
Always returns the same generic response whether or not the email
exists (prevents user enumeration).
"""
data = request.get_json(silent=True) or {}
email = (data.get('email') or '').strip().lower()
if not email:
return jsonify({'error': 'Email is required'}), 400
from app.models import db, User, PasswordResetToken
user = User.query.filter(db.func.lower(User.email) == email).first()
if not user:
return jsonify(_GENERIC_RESET_RESPONSE)
# Invalidate any outstanding tokens for this user
now = datetime.now(timezone.utc)
PasswordResetToken.query.filter_by(user_id=user.id, used_at=None).update(
{'used_at': now}
)
# Create new token — store only the sha256 hash
raw_token = secrets.token_urlsafe(32)
prt = PasswordResetToken(
user_id=user.id,
token_hash=_hash_reset_token(raw_token),
expires_at=now + _RESET_TOKEN_TTL,
)
db.session.add(prt)
db.session.commit()
# Build reset link (APP_BASE_URL env override, else request host)
import os
base_url = os.environ.get('APP_BASE_URL', request.host_url.rstrip('/'))
reset_url = f"{base_url}/reset-password?token={raw_token}"
from app.utils.mail import send_email
send_email(
to=[user.email],
subject='Reset your Command Sovereignty password',
html_body=(
f"<p>Hi {user.full_name or 'there'},</p>"
f"<p>We received a request to reset your password. "
f"Click the link below to choose a new one. "
f"This link expires in 1 hour.</p>"
f'<p><a href="{reset_url}">Reset your password</a></p>'
f"<p>If you didn't request this, you can safely ignore this email — "
f"your password will not be changed.</p>"
),
text_body=(
f"We received a request to reset your password.\n\n"
f"Reset link (expires in 1 hour): {reset_url}\n\n"
f"If you didn't request this, you can ignore this email."
),
)
return jsonify(_GENERIC_RESET_RESPONSE)
@auth_api_bp.route('/api/auth/reset-password', methods=['POST'])
@limiter.limit("10 per hour")
def api_auth_reset_password():
"""Complete a password reset with a valid token + new password."""
data = request.get_json(silent=True) or {}
token = (data.get('token') or '').strip()
new_password = data.get('password') or data.get('new_password') or ''
if not token:
return jsonify({'error': 'Reset token is required'}), 400
policy_error = validate_password_policy(new_password)
if policy_error:
return jsonify({'error': policy_error}), 400
from app.models import db, User, PasswordResetToken, UserSession
prt = PasswordResetToken.query.filter_by(
token_hash=_hash_reset_token(token)
).first()
if not prt or not prt.is_valid:
return jsonify({'error': 'Invalid or expired reset token'}), 400
user = db.session.get(User, prt.user_id)
if not user:
return jsonify({'error': 'Invalid or expired reset token'}), 400
now = datetime.now(timezone.utc)
# Update password
user.password_hash = generate_password_hash(new_password)
# Mark token used; invalidate any other outstanding tokens
prt.used_at = now
PasswordResetToken.query.filter_by(user_id=user.id, used_at=None).update(
{'used_at': now}
)
# Revoke ALL active sessions for this user (force re-login everywhere)
UserSession.query.filter_by(user_id=user.id).delete()
db.session.commit()
return jsonify({
'success': True,
'message': 'Password has been reset. Please log in with your new password.',
})
# ─── 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.
# ═══════════════════════════════════════════════════════════════════════
# Team invites — list, send, cancel
# ═══════════════════════════════════════════════════════════════════════
@auth_api_bp.route('/api/user/invites', methods=['GET'])
@require_auth_json()
def api_user_invites_list():
"""List all invites for the current user's company."""
from app.models import Invite, User, UserCompany
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({'error': 'No company membership found'}), 403
invites = Invite.query.filter_by(company_id=uc.company_id).order_by(Invite.created_at.desc()).all()
result = []
for inv in invites:
result.append({
'id': inv.id,
'email': inv.email,
'role': inv.role,
'status': inv.status,
'created_at': inv.created_at.isoformat(),
'expires_at': inv.expires_at.isoformat() if inv.expires_at else None,
})
return jsonify({'invites': result})
@auth_api_bp.route('/api/user/invites', methods=['POST'])
@require_auth_json()
@require_csrf
def api_user_invites_send():
"""Send an invite for a new team member."""
from app.models import Invite, db, User, UserCompany
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({'error': 'No company membership found'}), 403
error, membership = _check_company_access(uc.company_id, min_permission='manage_team')
if error:
return error
data = request.get_json(silent=True) or {}
email = (data.get('email') or '').strip()
role = (data.get('role') or 'member').strip().lower()
valid_roles = ['owner', 'admin', 'manager', 'member', 'viewer']
if role not in valid_roles:
role = 'member'
if not email or '@' not in email:
return jsonify({'error': 'Valid email is required'}), 400
existing = Invite.query.filter_by(company_id=uc.company_id, email=email, status='pending').first()
if existing:
return jsonify({'error': 'Invite already sent to this email'}), 409
invite = Invite(
company_id=uc.company_id,
email=email,
role=role,
invited_by=current_user.id,
status='pending',
token=secrets.token_urlsafe(32),
expires_at=datetime.now(timezone.utc) + timedelta(days=14),
)
db.session.add(invite)
db.session.commit()
return jsonify({'success': True, 'invite': {'email': email, 'role': role}}), 201
@auth_api_bp.route('/api/user/invites/<int:invite_id>', methods=['DELETE'])
@require_auth_json()
@require_csrf
def api_user_invites_cancel(invite_id):
"""Cancel a pending invite."""
from app.models import Invite, db, UserCompany
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({'error': 'No company membership found'}), 403
invite = Invite.query.filter_by(id=invite_id, company_id=uc.company_id).first()
if not invite:
return jsonify({'error': 'Invite not found'}), 404
if invite.status != 'pending':
return jsonify({'error': 'Invite is no longer pending'}), 400
invite.status = 'cancelled'
db.session.commit()
return jsonify({'success': True, 'message': 'Invite cancelled'})
# ═══════════════════════════════════════════════════════════════════════
# Subscription / plan info
# ═══════════════════════════════════════════════════════════════════════
PLANS = {
'free': {
'name': 'Free',
'price': 0,
'currency': 'USD',
'features': ['1 market', '1 seat', '2 integrations', '5 GB storage'],
'limits': {'markets': 1, 'seats': 1, 'integrations': 2, 'storageGb': 5},
},
'pro': {
'name': 'Pro',
'price': 49,
'currency': 'USD',
'features': ['5 markets', '5 seats', '5 integrations', '25 GB storage'],
'limits': {'markets': 5, 'seats': 5, 'integrations': 5, 'storageGb': 25},
},
'business': {
'name': 'Business',
'price': 149,
'currency': 'USD',
'features': ['Unlimited markets', 'Unlimited seats', '99 integrations', '100 GB storage'],
'limits': {'markets': 999, 'seats': 999, 'integrations': 99, 'storageGb': 100},
},
'enterprise': {
'name': 'Enterprise',
'price': 0,
'currency': 'USD',
'features': ['Everything', 'Custom limits', 'Dedicated support', 'SLA'],
'limits': {'markets': 999, 'seats': 999, 'integrations': 99, 'storageGb': 999},
},
}
@auth_api_bp.route('/api/user/subscription', methods=['GET'])
@require_auth_json()
def api_user_subscription():
"""Return the current user's subscription info."""
from app.models import Company, UserCompany
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({'error': 'No company membership found'}), 403
company = db.session.get(Company, uc.company_id)
if not company:
return jsonify({'error': 'Company not found'}), 404
tier = company.subscription_tier or 'free'
plan = PLANS.get(tier, PLANS['free'])
# TODO: Replace with Stripe API when billing is wired up.
return jsonify({
'current_plan': tier,
'plan': plan,
'stripe_customer_id': company.stripe_customer_id,
'status': 'active', # TODO: Pull from Stripe
'billing_cycle': 'monthly', # TODO: Pull from Stripe
})
@auth_api_bp.route('/api/user/subscription', methods=['PUT'])
@require_auth_json()
@require_csrf
def api_user_subscription_update():
"""Update subscription plan — placeholder until Stripe is wired up."""
return jsonify({
'error': 'Subscription changes are not yet available. Stripe integration in progress.',
'todo': 'stripe',
}), 501
# ═══════════════════════════════════════════════════════════════════════
# Usage / quota stats
# ═══════════════════════════════════════════════════════════════════════
@auth_api_bp.route('/api/user/usage', methods=['GET'])
@require_auth_json()
def api_user_usage():
"""Return current usage stats vs plan limits."""
from app.models import Project, UserCompany, Connector, Company
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if not uc:
return jsonify({'error': 'No company membership found'}), 403
company = db.session.get(Company, uc.company_id)
tier = company.subscription_tier or 'free'
plan = PLANS.get(tier, PLANS['free'])
limits = plan['limits']
seats_used = UserCompany.query.filter_by(company_id=uc.company_id).count()
markets_used = Project.query.filter_by(company_id=uc.company_id).count()
integrations_used = Connector.query.filter_by(company_id=uc.company_id, status='connected').count()
return jsonify({
'plan': tier,
'usage': {
'seats': seats_used,
'markets': markets_used,
'integrations': integrations_used,
'storageGb': 0, # TODO: Calculate actual storage usage
},
'limits': limits,
'percentages': {
'seats': min(seats_used / max(limits['seats'], 1) * 100, 100),
'markets': min(markets_used / max(limits['markets'], 1) * 100, 100),
'integrations': min(integrations_used / max(limits['integrations'], 1) * 100, 100),
'storageGb': 0,
},
})
# ═══════════════════════════════════════════════════════════════════════
# Data export
# ═══════════════════════════════════════════════════════════════════════
@auth_api_bp.route('/api/user/data-export', methods=['POST'])
@require_auth_json()
@require_csrf
def api_user_data_export():
"""Generate and return a full data export as a downloadable file."""
import json
import csv
import io
import zipfile
from datetime import datetime, timezone
from flask import make_response
from app.models import (
User, UserCompany, Company, Setting, UserSetting, UserSession,
Notification, Invite, UserGroup, GroupMember,
Project, Goal, GoalMetric, KPIValue, Forecast,
RevenueLeak, OptimizationMove, CoachingAssignment, CoachingScorecard,
RevenueRecord, Location, AdCampaign, AdMetric, AdTemplate,
AdCreative, AdKeyword, LeadAttribution, OptimizationRule,
OptimizationLog, AngiLead, AngiLeadAction, AccountingRecord,
QuickbooksInvoice, QuickbooksTransaction, QuickbooksCustomer,
QuickbooksExpense, CrmContact, CrmCompany, CrmDeal,
Template, CustomField, ROICalculation, SlackChannel,
ExternalSyncRecord, DemoRequest, SSOConfig,
ActivityLog, AuditLog, Connector, ConnectorLog, PasswordResetToken,
)
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
company_id = uc.company_id if uc else None
data = request.get_json(silent=True) or {}
fmt = data.get('format', 'json').lower()
if fmt not in ('json', 'csv'):
return jsonify({'error': 'Format must be json or csv'}), 400
# ── Helper: serialize model row to dict ──────────────────────────
def row_to_dict(obj):
d = {}
for col in obj.__table__.columns:
val = getattr(obj, col.name)
if isinstance(val, (datetime,)):
d[col.name] = val.isoformat()
elif isinstance(val, dict):
d[col.name] = val # JSON columns pass through
else:
d[col.name] = val
return d
# ── Collect data ─────────────────────────────────────────────────
export = {}
# Profile (exclude password_hash, totp_secret)
profile = row_to_dict(current_user)
profile.pop('password_hash', None)
profile.pop('totp_secret', None)
export['profile'] = profile
# Company membership
memberships = UserCompany.query.filter_by(user_id=current_user.id).all()
export['company_memberships'] = [row_to_dict(m) for m in memberships]
if not company_id:
# No company — return what we have
payload = {
'exported_at': datetime.now(timezone.utc).isoformat(),
'exported_by': current_user.email,
'user_id': current_user.id,
}
payload.update(export)
else:
# ── Company data ─────────────────────────────────────────────
company = Company.query.get(company_id)
export['company'] = row_to_dict(company) if company else None
# User settings (user-scoped)
export['user_settings'] = [
row_to_dict(s) for s in UserSetting.query.filter_by(user_id=current_user.id).all()
]
# Company settings
export['company_settings'] = [
row_to_dict(s) for s in Setting.query.filter_by(company_id=company_id).all()
]
# Team
export['invites'] = [
row_to_dict(i) for i in Invite.query.filter_by(company_id=company_id).all()
]
export['user_groups'] = [
row_to_dict(g) for g in UserGroup.query.filter_by(company_id=company_id).all()
]
export['group_members'] = [
row_to_dict(m) for m in GroupMember.query.filter(
GroupMember.group_id.in_(
UserGroup.query.filter_by(company_id=company_id).with_entities(UserGroup.id)
)
).all()
]
# Projects
export['projects'] = [
row_to_dict(p) for p in Project.query.filter_by(company_id=company_id).all()
]
# Goals + metrics
export['goals'] = [
row_to_dict(g) for g in Goal.query.filter_by(company_id=company_id).all()
]
export['goal_metrics'] = [
row_to_dict(m) for m in GoalMetric.query.filter(
GoalMetric.goal_id.in_(
Goal.query.filter_by(company_id=company_id).with_entities(Goal.id)
)
).all()
]
# KPIs & forecasts
export['kpi_values'] = [
row_to_dict(k) for k in KPIValue.query.filter_by(company_id=company_id).all()
]
export['forecasts'] = [
row_to_dict(f) for f in Forecast.query.filter_by(company_id=company_id).all()
]
# Revenue & optimization
export['revenue_leaks'] = [
row_to_dict(r) for r in RevenueLeak.query.filter_by(company_id=company_id).all()
]
export['optimization_moves'] = [
row_to_dict(o) for o in OptimizationMove.query.filter_by(company_id=company_id).all()
]
export['revenue_records'] = [
row_to_dict(r) for r in RevenueRecord.query.filter_by(company_id=company_id).all()
]
export['locations'] = [
row_to_dict(l) for l in Location.query.filter_by(company_id=company_id).all()
]
# Coaching
export['coaching_assignments'] = [
row_to_dict(c) for c in CoachingAssignment.query.filter_by(company_id=company_id).all()
]
export['coaching_scorecards'] = [
row_to_dict(s) for s in CoachingScorecard.query.filter_by(company_id=company_id).all()
]
# Connectors
export['connectors'] = [
row_to_dict(c) for c in Connector.query.filter_by(company_id=company_id).all()
]
export['connector_logs'] = [
row_to_dict(l) for l in ConnectorLog.query.filter_by(company_id=company_id).all()
]
# Ads & marketing
export['ad_campaigns'] = [
row_to_dict(a) for a in AdCampaign.query.filter_by(company_id=company_id).all()
]
export['ad_metrics'] = [
row_to_dict(m) for m in AdMetric.query.filter_by(company_id=company_id).all()
]
# Ad templates (global reference data — no company_id)
export['ad_templates'] = [
row_to_dict(t) for t in AdTemplate.query.all()
]
export['ad_creatives'] = [
row_to_dict(c) for c in AdCreative.query.filter_by(company_id=company_id).all()
]
export['ad_keywords'] = [
row_to_dict(k) for k in AdKeyword.query.filter_by(company_id=company_id).all()
]
# Leads & CRM
export['angi_leads'] = [
row_to_dict(l) for l in AngiLead.query.filter_by(company_id=company_id).all()
]
export['lead_actions'] = [
row_to_dict(a) for a in AngiLeadAction.query.filter(
AngiLeadAction.lead_id.in_(
AngiLead.query.filter_by(company_id=company_id).with_entities(AngiLead.id)
)
).all()
]
export['lead_attribution'] = [
row_to_dict(l) for l in LeadAttribution.query.filter_by(company_id=company_id).all()
]
export['crm_contacts'] = [
row_to_dict(c) for c in CrmContact.query.filter_by(company_id=company_id).all()
]
export['crm_companies'] = [
row_to_dict(c) for c in CrmCompany.query.filter_by(company_id=company_id).all()
]
export['crm_deals'] = [
row_to_dict(d) for d in CrmDeal.query.filter_by(company_id=company_id).all()
]
# Accounting (QuickBooks)
export['accounting_records'] = [
row_to_dict(r) for r in AccountingRecord.query.filter_by(company_id=company_id).all()
]
export['qb_invoices'] = [
row_to_dict(i) for i in QuickbooksInvoice.query.filter_by(company_id=company_id).all()
]
export['qb_transactions'] = [
row_to_dict(t) for t in QuickbooksTransaction.query.filter_by(company_id=company_id).all()
]
export['qb_customers'] = [
row_to_dict(c) for c in QuickbooksCustomer.query.filter_by(company_id=company_id).all()
]
export['qb_expenses'] = [
row_to_dict(e) for e in QuickbooksExpense.query.filter_by(company_id=company_id).all()
]
# Templates & fields
export['templates'] = [
row_to_dict(t) for t in Template.query.filter_by(company_id=company_id).all()
]
export['custom_fields'] = [
row_to_dict(f) for f in CustomField.query.filter_by(company_id=company_id).all()
]
export['roi_calculations'] = [
row_to_dict(r) for r in ROICalculation.query.filter_by(company_id=company_id).all()
]
# Slack
export['slack_channels'] = [
row_to_dict(s) for s in SlackChannel.query.filter_by(company_id=company_id).all()
]
# Optimization rules/logs
export['optimization_rules'] = [
row_to_dict(r) for r in OptimizationRule.query.filter_by(company_id=company_id).all()
]
export['optimization_logs'] = [
row_to_dict(l) for l in OptimizationLog.query.filter_by(company_id=company_id).all()
]
# Notifications
export['notifications'] = [
row_to_dict(n) for n in Notification.query.filter_by(company_id=company_id).all()
]
# Activity & audit logs
export['activity_logs'] = [
row_to_dict(l) for l in ActivityLog.query.filter_by(company_id=company_id).all()
]
export['audit_logs'] = [
row_to_dict(l) for l in AuditLog.query.all()
]
# Misc
export['external_sync_records'] = [
row_to_dict(r) for r in ExternalSyncRecord.query.filter_by(company_id=company_id).all()
]
export['sso_configs'] = [
row_to_dict(s) for s in SSOConfig.query.filter_by(company_id=company_id).all()
]
# Global tables (no company_id — export all but sanitize sensitive data)
export['demo_requests'] = [
row_to_dict(d) for d in DemoRequest.query.all()
]
# Sessions (user-scoped)
export['sessions'] = [
row_to_dict(s) for s in UserSession.query.filter_by(user_id=current_user.id).all()
]
# Build final payload
payload = {
'exported_at': datetime.now(timezone.utc).isoformat(),
'exported_by': current_user.email,
'user_id': current_user.id,
'company_id': company_id,
}
payload.update(export)
# ── Generate response ────────────────────────────────────────────
timestamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
if fmt == 'json':
content = json.dumps(payload, indent=2, default=str).encode('utf-8')
mime = 'application/json'
ext = 'json'
else:
# CSV: flatten the JSON into a single CSV with sections
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(['section', 'key', 'value'])
for section, data_item in payload.items():
if isinstance(data_item, list):
for idx, row in enumerate(data_item):
if isinstance(row, dict):
for k, v in row.items():
writer.writerow([section, f'row_{idx}.{k}', json.dumps(v) if isinstance(v, (dict, list)) else v])
else:
writer.writerow([section, f'row_{idx}', row])
elif isinstance(data_item, dict):
for k, v in data_item.items():
writer.writerow([section, k, json.dumps(v) if isinstance(v, (dict, list)) else v])
else:
writer.writerow([section, '', data_item])
content = buf.getvalue().encode('utf-8')
mime = 'text/csv'
ext = 'csv'
# Build a safe filename from company name
company_name = company.name if company else 'command-sovereignty'
slug = re.sub(r'[^a-zA-Z0-9]+', '-', company_name).strip('-').lower()
filename = f'{slug}-export-{timestamp}.{ext}'
response = make_response(content)
response.headers['Content-Type'] = mime
response.headers['Content-Disposition'] = f'attachment; filename="{filename}"'
return response
# ═══════════════════════════════════════════════════════════════════════
# Data deletion (GDPR "right to be forgotten")
# ═══════════════════════════════════════════════════════════════════════
@auth_api_bp.route('/api/user/data-delete', methods=['POST'])
@require_auth_json()
@require_csrf
def api_user_data_delete():
"""Permanently delete the user's account and all associated data."""
from app.models import User, UserCompany
# Prevent users from deleting themselves
if current_user.id == current_user.id:
owners = UserCompany.query.filter_by(
user_id=current_user.id, role='owner'
).count()
if owners > 0:
return jsonify({
'error': 'Cannot delete an account that owns a company. Transfer ownership first.',
}), 400
# TODO: Cascade delete user data, sessions, settings, etc.
# For now return a placeholder.
return jsonify({
'error': 'Account deletion is not yet implemented. Contact support.',
}), 501
# ═══════════════════════════════════════════════════════════════════════
# 2FA — TOTP
# ═══════════════════════════════════════════════════════════════════════