PassKeeper/app/__init__.py

from flask import Flask, render_template, request as _flask_request
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_wtf.csrf import CSRFProtect
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask_cors import CORS
from werkzeug.middleware.proxy_fix import ProxyFix

from .config import config

db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
csrf = CSRFProtect()
limiter = Limiter(key_func=get_remote_address)


def client_ip() -> str:
    """
    Return the real client IP address, trusted only from one upstream proxy
    (Nginx). ProxyFix — applied in create_app() — ensures request.remote_addr
    is already set to the correct value; we do NOT parse X-Forwarded-For
    manually here, which would be spoofable.
    """
    return _flask_request.remote_addr or ''

# APScheduler is used for the background token-blacklist cleanup job.
# Imported here so it is available at module level; started inside create_app().
try:
    from apscheduler.schedulers.background import BackgroundScheduler
    _scheduler_available = True
except ImportError:  # pragma: no cover — optional dependency
    _scheduler_available = False


def create_app(config_name: str = 'development') -> Flask:
    app = Flask(__name__)
    # Trust exactly one upstream proxy (Nginx) for X-Forwarded-For / X-Forwarded-Proto.
    # x_for=1 means only the rightmost hop in XFF is trusted, making rate-limit
    # IP keys spoof-resistant — a client cannot bypass per-IP limits by injecting
    # an arbitrary IP into the XFF header.
    app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)
    selected_config = config[config_name]
    # Reject insecure production settings before the app is handed back, so no
    # request is ever served under one. Deliberately not done at class-definition
    # time — that made importing app.config impossible without production
    # secrets, breaking tests and local tooling.
    if hasattr(selected_config, 'validate'):
        selected_config.validate()
    app.config.from_object(selected_config)

    # Extensions
    db.init_app(app)
    migrate.init_app(app, db)
    login_manager.init_app(app)
    csrf.init_app(app)
    limiter.init_app(app)

    # Restrict CORS to the configured origin (locked to production domain in prod)
    cors_origins = app.config.get('CORS_ORIGINS', '*')
    CORS(app, resources={r'/api/*': {'origins': cors_origins}})

    # Inject static asset version into every template for cache-busting.
    # Usage in templates: {{ url_for('static', filename='css/app.css') }}?v={{ sv }}
    app.jinja_env.globals['sv'] = app.config.get('STATIC_VERSION', '1')

    # Attach security headers to every response
    @app.after_request
    def set_security_headers(response):
        # HSTS: enforce HTTPS for 1 year across all subdomains.
        # 'preload' enables browser preload-list submission so first-time
        # HTTP visitors are also protected before the first redirect.
        response.headers['Strict-Transport-Security'] = (
            'max-age=31536000; includeSubDomains; preload'
        )
        # Prevent clickjacking (belt-and-suspenders alongside CSP frame-ancestors)
        response.headers['X-Frame-Options'] = 'DENY'
        # Prevent MIME-type sniffing attacks
        response.headers['X-Content-Type-Options'] = 'nosniff'
        # Limit referrer information sent on navigation
        response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
        # Permissions policy — disable browser features the app does not use
        response.headers['Permissions-Policy'] = (
            'geolocation=(), camera=(), microphone=()'
        )
        # Content-Security-Policy (HTTP header is authoritative — overrides meta tag).
        # base-uri 'self' — blocks <base href> injection that would redirect all
        #   relative URLs to an attacker-controlled origin.
        # upgrade-insecure-requests — instructs browsers to rewrite http:// sub-resource
        #   requests to https:// to avoid mixed-content warnings.
        response.headers['Content-Security-Policy'] = (
            "default-src 'self'; "
            "script-src 'self'; "
            "style-src 'self'; "
            "img-src 'self' data:; "
            "font-src 'self'; "
            "connect-src 'self' https://api.pwnedpasswords.com; "
            "frame-ancestors 'none'; "
            "base-uri 'self'; "
            "upgrade-insecure-requests;"
        )
        return response

    # Ensure all models are imported so SQLAlchemy knows about them
    from .models.user import User
    from .models.folder import Folder
    from .models.vault_item import VaultItem
    from .models.token_blacklist import TokenBlacklist
    from .models.shared_item import SharedItem
    from .models.emergency_access import EmergencyAccess
    from .models.audit_log import AuditLog
    from .models.recovery_challenge import RecoveryChallenge
    from .models.totp_used_code import TotpUsedCode
    from .models.webauthn_credential import WebAuthnCredential

    @login_manager.user_loader
    def load_user(user_id):
        return db.session.get(User, int(user_id))

    # Blueprints
    from .routes.auth import auth_bp
    from .routes.vault import vault_bp
    from .routes.folders import folders_bp
    from .routes.sharing import sharing_bp
    from .routes.emergency import emergency_bp
    from .routes.webauthn import webauthn_bp

    app.register_blueprint(auth_bp,       url_prefix='/api/auth')
    app.register_blueprint(vault_bp,      url_prefix='/api/vault')
    app.register_blueprint(folders_bp,    url_prefix='/api/folders')
    app.register_blueprint(sharing_bp,    url_prefix='/api/sharing')
    app.register_blueprint(emergency_bp,  url_prefix='/api/emergency')
    app.register_blueprint(webauthn_bp,   url_prefix='/api/webauthn')

    # Exempt all API blueprints from CSRF — JWT bearer tokens make CSRF irrelevant
    csrf.exempt(auth_bp)
    csrf.exempt(vault_bp)
    csrf.exempt(folders_bp)
    csrf.exempt(sharing_bp)
    csrf.exempt(emergency_bp)
    csrf.exempt(webauthn_bp)

    # ── Global JSON error handlers ─────────────────────────────────────────────
    # All /api/* routes must return JSON even on unhandled exceptions.
    # Without this, Flask returns an HTML 500 page which breaks JSON.parse()
    # in the extension and surfaces as "Unexpected token '<'" to the user.
    import logging as _logging
    from flask import request as _request, jsonify as _jsonify
    _api_log = _logging.getLogger(__name__)

    @app.errorhandler(404)
    def not_found(e):
        if _request.path.startswith('/api/'):
            return _jsonify({'error': 'Endpoint not found'}), 404
        return render_template('auth/login.html'), 404

    @app.errorhandler(405)
    def method_not_allowed(e):
        if _request.path.startswith('/api/'):
            return _jsonify({'error': 'Method not allowed'}), 405
        return render_template('auth/login.html'), 405

    @app.errorhandler(Exception)
    def handle_exception(e):
        from werkzeug.exceptions import HTTPException
        if isinstance(e, HTTPException):
            if _request.path.startswith('/api/'):
                return _jsonify({'error': e.description}), e.code
            return e
        # Unhandled exception — log it and return JSON for API routes.
        _api_log.exception('[PassKeeper] Unhandled exception on %s %s', _request.method, _request.path)
        if _request.path.startswith('/api/'):
            return _jsonify({'error': 'An internal server error occurred. Please try again.'}), 500
        return render_template('auth/login.html'), 500

    # Page-serving routes
    @app.route('/')
    @app.route('/login')
    def login_page():
        return render_template('auth/login.html')

    @app.route('/register')
    def register_page():
        return render_template('auth/register.html')

    @app.route('/vault')
    def vault_page():
        return render_template('vault/index.html')

    @app.route('/recover')
    def recover_page():
        return render_template('auth/recover.html')

    # ── Background scheduler — token blacklist cleanup ─────────────────────────
    # Runs cleanup_expired() every hour so the token_blacklist table never
    # accumulates unbounded rows. Runs in a daemon thread — no request context.
    if _scheduler_available:
        def _cleanup_expired_tokens():
            with app.app_context():
                try:
                    from app.models.token_blacklist import TokenBlacklist
                    from app.models.recovery_challenge import RecoveryChallenge
                    from app.models.totp_used_code import TotpUsedCode
                    from app.models.shared_item import SharedItem
                    TokenBlacklist.cleanup_expired()
                    RecoveryChallenge.cleanup_expired()
                    TotpUsedCode.cleanup_expired()
                    # Delete expired unaccepted shares.
                    from datetime import datetime, timezone
                    SharedItem.query.filter(
                        SharedItem.accepted == False,
                        SharedItem.expires_at != None,
                        SharedItem.expires_at <= datetime.now(timezone.utc).replace(tzinfo=None),
                    ).delete()
                    db.session.commit()
                    import logging
                    logging.getLogger(__name__).debug(
                        '[PassKeeper] token_blacklist + recovery_challenges cleanup completed'
                    )
                except Exception as exc:  # pragma: no cover
                    import logging
                    logging.getLogger(__name__).warning(
                        '[PassKeeper] cleanup job failed: %s', exc
                    )

        scheduler = BackgroundScheduler(daemon=True)
        scheduler.add_job(
            _cleanup_expired_tokens,
            trigger='interval',
            hours=1,
            id='token_blacklist_cleanup',
            replace_existing=True,
        )
        # In Flask debug mode the Werkzeug reloader spawns two processes.
        # WERKZEUG_RUN_MAIN == 'true' only in the child (the actual server),
        # so we skip the scheduler in the parent reloader to avoid duplicate jobs.
        import os
        if app.config.get('SCHEDULER_ENABLED', True) and (
            not app.debug or os.environ.get('WERKZEUG_RUN_MAIN') == 'true'
        ):
            scheduler.start()

    # ── Production safety checks ───────────────────────────────────────────────
    # Warn loudly at startup when running in production with settings that are
    # only appropriate for development.
    if not app.config.get('DEBUG', False):
        import logging
        _log = logging.getLogger(__name__)
        storage_uri = app.config.get('RATELIMIT_STORAGE_URI', 'memory://')
        if storage_uri.startswith('memory://'):
            _log.warning(
                '[PassKeeper] WARNING: RATELIMIT_STORAGE_URI is set to "memory://" '
                'in a production environment. Rate limits are tracked per-worker '
                'and will not be shared across Gunicorn processes. '
                'Set RATELIMIT_STORAGE_URI to a Redis URL (e.g. redis://localhost:6379) '
                'in your production .env to enforce global rate limits.'
            )

    # ── Critical config validation — all environments ──────────────────────────
    # TOTP_ENCRYPTION_KEY is required whenever MFA is in use. Validate it at
    # startup so a misconfiguration produces a clear error immediately rather
    # than a cryptic RuntimeError inside a request handler hours later.
    import logging as _startup_log
    _slog = _startup_log.getLogger(__name__)
    totp_key = app.config.get('TOTP_ENCRYPTION_KEY', '')
    if not totp_key:
        _slog.warning(
            '[PassKeeper] TOTP_ENCRYPTION_KEY is not set. MFA setup and verification '
            'will fail. Generate a key with: '
            'python -c "import secrets; print(secrets.token_hex(32))" '
            'and add it to your .env file.'
        )
    elif len(totp_key) != 64:
        _slog.error(
            '[PassKeeper] TOTP_ENCRYPTION_KEY must be exactly 64 hex characters (32 bytes). '
            f'Current value has {len(totp_key)} characters. MFA will not function correctly.'
        )
    else:
        try:
            bytes.fromhex(totp_key)
        except ValueError:
            _slog.error(
                '[PassKeeper] TOTP_ENCRYPTION_KEY contains non-hex characters. '
                'MFA will not function correctly.'
            )

    return app