"""
Command Center - Flask Application Factory
B2B revenue forecasting SaaS for home improvement companies.
"""
import logging
import os
from flask import Flask, redirect, url_for, request, send_from_directory, jsonify
from flask_login import LoginManager
from flask_mail import Mail
from logging.handlers import RotatingFileHandler
from .models import db
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask import request as flask_request
from sqlalchemy import event
from sqlalchemy.engine import Engine
# SQLite: enforce foreign keys + WAL mode on every new connection
@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
"""Set production pragmas for SQLite connections.
- foreign_keys=ON: enforce FK constraints
- journal_mode=WAL: readers don't block writers (persistent setting)
- synchronous=NORMAL: safe with WAL, much faster than FULL
- busy_timeout=5000: wait up to 5s on lock contention instead of
immediately raising 'database is locked'
"""
import sqlite3
if not isinstance(dbapi_connection, sqlite3.Connection):
return # non-SQLite backend (e.g. PostgreSQL)
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.execute("PRAGMA journal_mode=WAL")
cursor.execute("PRAGMA synchronous=NORMAL")
cursor.execute("PRAGMA busy_timeout=5000")
cursor.close()
def _get_real_ip():
"""Rate limit key function that respects Cloudflare's CF-Connecting-IP.
Behind a Cloudflare Tunnel the real client IP is lost without this.
Falls back to X-Forwarded-For → X-Real-IP → remote_addr.
Returns None when RATELIMIT_ENABLED is False to disable per-route limits
(flask_limiter skips limits when key_func returns None).
"""
from flask import current_app
if current_app.testing or not current_app.config.get('RATELIMIT_ENABLED', True):
return None
cf_ip = flask_request.headers.get('CF-Connecting-IP')
if cf_ip:
return cf_ip
xff = flask_request.headers.get('X-Forwarded-For')
if xff:
return xff.split(',')[0].strip()
xri = flask_request.headers.get('X-Real-IP')
if xri:
return xri
return flask_request.remote_addr or '127.0.0.1'
mail = Mail()
limiter = Limiter(
key_func=_get_real_ip,
storage_uri=os.environ.get("RATELIMIT_STORAGE_URI", "redis://localhost:6379/0")
)
def create_app():
# Templates live at project root, not inside app/
import os
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
app = Flask(__name__, template_folder=os.path.join(project_root, 'templates'))
# SECURITY: Fail loudly if SECRET_KEY is not set
secret_key = os.environ.get('SECRET_KEY')
if not secret_key:
raise RuntimeError('SECRET_KEY environment variable is required')
app.config['SECRET_KEY'] = secret_key
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get(
'DATABASE_URL',
'sqlite:///auth.db'
)
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SESSION_COOKIE_HTTPONLY'] = True
app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'
app.config['SESSION_COOKIE_SECURE'] = os.environ.get('SESSION_COOKIE_SECURE', 'true').lower() in ('true', '1', 'yes')
app.config['PERMANENT_SESSION_LIFETIME'] = 86400 * 7 # 7 days
app.config['SESSION_REFRESH_EACH_REQUEST'] = True # re-issue session ID each request (session fixation)
# Session revocation: bind Flask sessions to UserSession records (us_id).
# If the UserSession DB record is deleted/revoked, the session is invalidated.
@app.before_request
def check_session_revocation():
from flask import session as flask_session
from flask_login import current_user
from .models import UserSession
if current_user.is_authenticated:
us_id = flask_session.get('us_id')
if us_id:
record = db.session.get(UserSession, us_id)
if not record:
from flask import redirect, url_for
from flask import flash
flask_session.clear()
return redirect(url_for('auth.login'))
# Security: cap request body size (uploads) at 16MB
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB
# Rate limiting config
app.config['RATELIMIT_ENABLED'] = os.environ.get('RATELIMIT_ENABLED', 'true').lower() not in ('false', '0', 'no')
app.config['RATELIMIT_DEFAULT'] = '100 per hour'
app.config['RATELIMIT_STRATEGY'] = 'fixed-window'
# Mail config - Hetzner Postfix relay
app.config['MAIL_SERVER'] = os.environ.get('MAIL_SERVER', '5.161.95.32')
app.config['MAIL_PORT'] = int(os.environ.get('MAIL_PORT', '587'))
app.config['MAIL_USE_TLS'] = True
app.config['MAIL_USERNAME'] = os.environ.get('MAIL_USERNAME', 'cs_marketing')
app.config['MAIL_PASSWORD'] = os.environ.get('MAIL_PASSWORD', '')
app.config['MAIL_DEFAULT_SENDER'] = os.environ.get('MAIL_DEFAULT_SENDER', 'marketing@commandsovereignty.com')
# OAuth 2.0 credentials — loaded from env vars, empty string if not set (graceful degradation)
for _svc in ('QUICKBOOKS', 'GOOGLE_ADS', 'GOOGLE_SHEETS', 'HUBSPOT', 'SLACK', 'FACEBOOK_ADS', 'ZAPIER', 'SQUARE', 'JOBBER'):
app.config[f'OAUTH_{_svc}_CLIENT_ID'] = os.environ.get(f'OAUTH_{_svc}_CLIENT_ID', '')
app.config[f'OAUTH_{_svc}_CLIENT_SECRET'] = os.environ.get(f'OAUTH_{_svc}_CLIENT_SECRET', '')
# Connector redirect base URI for OAuth callbacks
app.config['CONNECTOR_REDIRECT_URI'] = os.environ.get('CONNECTOR_REDIRECT_URI', '')
# Initialize extensions
db.init_app(app)
mail.init_app(app)
limiter.init_app(app)
# Disable rate limiting in test mode or when explicitly disabled via config/env
if app.testing or not app.config.get('RATELIMIT_ENABLED', True):
limiter.enabled = False
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = '/auth/login'
# Structured logging (rotating file, production only)
if not app.debug:
log_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'logs')
os.makedirs(log_dir, exist_ok=True)
file_handler = RotatingFileHandler(
os.path.join(log_dir, 'app.log'),
maxBytes=10 * 1024 * 1024, # 10MB
backupCount=5
)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s %(name)s: %(message)s'
))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
# Pre-flight DB health check: warn if critical tables are missing/empty
if not app.debug:
try:
from sqlalchemy import text
with app.app_context():
uri = app.config['SQLALCHEMY_DATABASE_URI']
if uri.startswith('sqlite'):
result = db.session.execute(text(
"SELECT COALESCE((SELECT count(*) FROM profiles), -1) as c"
)).scalar()
if result == -1:
app.logger.warning('DB pre-flight: profiles table does not exist. Run migrations.')
elif result == 0:
app.logger.warning('DB pre-flight: profiles table is empty — check backups at backups/')
else:
app.logger.info('DB pre-flight: %d profiles found.', result)
except Exception as e:
app.logger.warning('DB pre-flight check failed: %s', e)
@login_manager.user_loader
def load_user(user_id):
from .models import User
# UUIDs are strings - don't convert to int
return db.session.get(User, user_id)
@app.errorhandler(404)
def not_found(e):
# Serve SPA for all non-API routes
if not request.path.startswith('/api/'):
SPA_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'static', 'dist')
return send_from_directory(SPA_DIR, 'index.html')
from flask import jsonify
return jsonify({'error': 'Not found'}), 404
@app.errorhandler(429)
def ratelimit_handler(e):
"""Return JSON error for rate limit hits instead of HTML."""
return jsonify({
'error': 'Too many attempts. Please wait a moment and try again.',
'retry_after': e.description
}), 429
@app.errorhandler(Exception)
def handle_exception(e):
"""Global catch-all: standardized JSON errors, no stack traces in prod."""
from werkzeug.exceptions import HTTPException
# Let HTTP errors (400/401/403/405/...) keep their semantics, as JSON
if isinstance(e, HTTPException):
return jsonify({
'error': e.description or e.name,
'code': e.name.upper().replace(' ', '_'),
}), e.code
# In debug mode, re-raise so the interactive debugger/traceback shows
if app.debug:
raise e
# Log the full traceback server-side, return a sanitized response
app.logger.exception('Unhandled exception: %s', request.path)
return jsonify({'error': 'Internal server error', 'code': 'INTERNAL_ERROR'}), 500
@app.after_request
def set_security_headers(response):
"""Attach security headers to every response."""
# CSP audit (July 2026): built SPA bundle has NO inline <script> tags
# (single external module at /dist/assets/), so script-src stays 'self'
# plus js.stripe.com (loadStripe in BillingPage injects Stripe.js).
# style-src keeps 'unsafe-inline': React components use style={} attrs
# (19 files) and libs inject <style> elements dynamically at runtime —
# nonces/hashes are not practical for those. Stripe checkout/elements
# also needs frame-src and connect-src for js.stripe.com/api.stripe.com.
response.headers.setdefault(
'Content-Security-Policy',
"default-src 'self'; "
"script-src 'self' https://js.stripe.com; "
"style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; "
"font-src 'self' data: https://fonts.gstatic.com; "
"img-src 'self' data: blob:; "
"connect-src 'self' https://api.stripe.com; "
"frame-src https://js.stripe.com https://hooks.stripe.com; "
"frame-ancestors 'none'; "
"object-src 'none'; "
"base-uri 'self'; "
"form-action 'self'"
)
response.headers.setdefault(
'Strict-Transport-Security', 'max-age=31536000; includeSubDomains'
)
response.headers.setdefault('X-Content-Type-Options', 'nosniff')
response.headers.setdefault('X-Frame-Options', 'DENY')
response.headers.setdefault('X-XSS-Protection', '1; mode=block')
response.headers.setdefault('Referrer-Policy', 'strict-origin-when-cross-origin')
return response
# Register connectors (imports submodules to populate the registry)
from .connectors import _import_submodules
_import_submodules()
# Register blueprints
from .routes.auth import auth_bp
from .routes.dashboard import dashboard_bp
from .routes.marketing import marketing_bp
from .routes.api_proxy import api_proxy_bp
# Split blueprints from api_proxy.py monolith (July 2026)
from .routes.auth_api import auth_api_bp
from .routes.security_api import security_api_bp
from .routes.admin_api import admin_api_bp
from .routes.business_api import business_api_bp
from .routes.connectors import connectors_bp
from .routes.analytics import analytics_bp
from .routes.multi_market import multi_market_bp
from .routes.coaching import coaching_bp
from .routes.billing import billing_bp
from .routes.portfolios import portfolios_bp
from .routes.email_campaigns import email_campaigns_bp
from .routes.sms_webhooks import sms_webhooks_bp
from .routes.sms import sms_bp
from .routes.sms_call import sms_call_bp
from .routes.enterprise import enterprise_bp
from .routes.super_admin_proxy import super_admin_proxy_bp
from .routes.lead_gen import lead_gen_bp
from .routes.estimates import estimates_bp
app.register_blueprint(auth_bp)
app.register_blueprint(dashboard_bp)
app.register_blueprint(marketing_bp)
app.register_blueprint(api_proxy_bp)
# Split blueprints (July 2026) — routes at same URLs as before
app.register_blueprint(auth_api_bp)
app.register_blueprint(security_api_bp)
app.register_blueprint(admin_api_bp)
app.register_blueprint(business_api_bp)
app.register_blueprint(connectors_bp)
app.register_blueprint(analytics_bp)
app.register_blueprint(multi_market_bp)
app.register_blueprint(coaching_bp)
app.register_blueprint(billing_bp)
app.register_blueprint(enterprise_bp)
app.register_blueprint(portfolios_bp)
app.register_blueprint(email_campaigns_bp)
app.register_blueprint(sms_webhooks_bp)
app.register_blueprint(sms_bp)
app.register_blueprint(sms_call_bp, url_prefix='/app/api/sms-call')
app.register_blueprint(super_admin_proxy_bp, url_prefix='/api/super-admin')
app.register_blueprint(lead_gen_bp)
app.register_blueprint(estimates_bp)
# CSRF audit: scan all state-changing routes for @require_csrf coverage
from app.utils.csrf import audit_exempt_routes
audit_exempt_routes(app)
# Health check endpoint for uptime monitoring
@app.route('/health')
def health_check():
"""Health check endpoint for uptime monitoring.\n\n Returns 503 if critical DB tables are missing (wiped DB detection).\n Monitored by command-sovereignty-watchdog.timer for auto-remediation.\n """
from flask import jsonify
try:
db.session.execute(db.text('SELECT 1'))
db_status = 'healthy'
except Exception as e:
db_status = f'unhealthy: {str(e)}'
return jsonify({
'status': 'unhealthy',
'database': db_status,
'version': '0.1.0',
'error': 'Database connection failed'
}), 503
# Wiped DB detection: verify critical tables exist and have data
# Skip in test mode (test DBs are in-memory and empty) and debug mode
if not app.testing and not app.debug:
uri = app.config['SQLALCHEMY_DATABASE_URI']
if uri.startswith('sqlite'):
try:
result = db.session.execute(db.text(
"SELECT COALESCE((SELECT count(*) FROM profiles), -1) as c"
)).scalar()
if result == -1:
return jsonify({
'status': 'unhealthy',
'database': 'wiped: profiles table missing',
'version': '0.1.0',
'error': 'Database appears wiped — critical tables missing'
}), 503
elif result == 0:
return jsonify({
'status': 'unhealthy',
'database': 'wiped: profiles table empty',
'version': '0.1.0',
'error': 'Database appears wiped — profiles table is empty'
}), 503
except Exception as e:
# Table check failed — DB likely wiped
return jsonify({
'status': 'unhealthy',
'database': f'wiped: table check failed ({str(e)})',
'version': '0.1.0',
'error': 'Database integrity check failed'
}), 503
return jsonify({
'status': 'healthy',
'database': db_status,
'version': '0.1.0'
})
# Start background sync scheduler (APScheduler; singleton lock ensures
# only one gunicorn worker runs it)
from .scheduler import init_scheduler
init_scheduler(app)
return app