glo/app.py

import html
import mimetypes
import os
import secrets
from pathlib import Path

from flask import Flask, g, jsonify, render_template, request, session
from werkzeug.exceptions import RequestEntityTooLarge
from werkzeug.middleware.proxy_fix import ProxyFix
from werkzeug.utils import secure_filename

from config import (
    APP_ROOT,
    CALLS_BASE,
    CALL_HELPER_SHARED_SECRET,
    CALL_TOKEN_TTL_SECONDS,
    CLIENT_HASH_KEY,
    CONSTANTS,
    DB_PATH,
    PRIVATE_ACCESS_KEY,
    RUNTIME_STATE,
    SECRET_KEY,
    TRUSTED_PROXY_HOPS,
    UPLOAD_DIR,
    ATTACHMENT_CLEANUP_INTERVAL_SECONDS,
    ATTACHMENT_ORPHAN_TTL_HOURS,
    flask_config,
    get_asset_version,
    now_iso,
    parse_iso,
    utc_now,
)
from db import get_db, db_write, init_db, bootstrap_admin
from auth import csrf_token, request_blacklisted
from attachments import maybe_cleanup_orphaned_attachments, cleanup_orphaned_attachments
from routes import register_routes

app = Flask(__name__)
app.secret_key = SECRET_KEY

if TRUSTED_PROXY_HOPS > 0:
    app.wsgi_app = ProxyFix(
        app.wsgi_app,
        x_for=TRUSTED_PROXY_HOPS,
        x_proto=TRUSTED_PROXY_HOPS,
        x_host=TRUSTED_PROXY_HOPS,
    )

app.config.update(flask_config())

app.os = os
app.secrets = secrets
app.mimetypes = mimetypes
app.secure_filename = secure_filename
app.timedelta = __import__("datetime").timedelta


def get_asset_template_values():
    return {
        "asset_version": get_asset_version(),
        "csp_nonce": getattr(g, "csp_nonce", ""),
    }


@app.before_request
def setup_request_security():
    g.csp_nonce = __import__("base64").b64encode(secrets.token_bytes(18)).decode("ascii")
    if request_blacklisted():
        if request.path.startswith("/api/"):
            return jsonify({"error": "Access to this service has been blocked.", "code": "blacklisted"}), 403
        try:
            return render_template("blocked.html", asset_version=get_asset_version()), 403
        except Exception:
            return (
                "<!doctype html><html><head><meta charset='utf-8'><title>access blocked</title></head>"
                "<body><h1>access blocked</h1><p>access is unavailable.</p></body></html>",
                403,
            )


@app.context_processor
def inject_template_values():
    return get_asset_template_values()


@app.after_request
def add_security_headers(response):
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["Referrer-Policy"] = "same-origin"
    response.headers["Permissions-Policy"] = "camera=(self), microphone=(self), geolocation=()"
    response.headers["Cross-Origin-Opener-Policy"] = "same-origin"
    response.headers["Cross-Origin-Resource-Policy"] = "same-origin"

    calls_connect_origin = CALLS_BASE
    calls_ws_origin = CALLS_BASE.replace("https://", "wss://").replace("http://", "ws://")

    response.headers["Content-Security-Policy"] = (
        "default-src 'self'; "
        "base-uri 'self'; "
        "form-action 'self'; "
        "frame-ancestors 'none'; "
        "object-src 'none'; "
        "img-src 'self' data: blob:; "
        "media-src 'self' blob:; "
        "style-src 'self' 'unsafe-inline'; "
        "script-src 'self'; "
        f"connect-src 'self' {calls_connect_origin} {calls_ws_origin};"
    )

    if request.path.startswith("/static/"):
        response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
    else:
        response.headers["Cache-Control"] = "no-store"

    return response


@app.errorhandler(RequestEntityTooLarge)
def request_too_large(_error):
    return jsonify({"error": "Request is too large."}), 413


@app.errorhandler(__import__("sqlite3").OperationalError)
def sqlite_error(error):
    app.logger.exception("Database error: %s", error)
    return jsonify({"error": "A database error occurred."}), 500


@app.route("/")
def index():
    return render_template("index.html")


@app.route("/api/csrf")
def api_csrf():
    return jsonify({"csrf_token": csrf_token()})


@app.route("/api/me")
def api_me():
    from auth import current_user

    user = current_user()
    return jsonify(
        {
            "authenticated": bool(user),
            "user": user,
            "csrf_token": csrf_token(),
        }
    )


@app.route("/api/register", methods=["POST"])
def register():
    from auth import rate_limited, csrf_valid, safe_json, authenticate_account, audit, login_user, register_account

    limited = rate_limited("register")
    if limited:
        return limited

    if not csrf_valid():
        return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

    try:
        data = safe_json()
    except ValueError as error:
        return jsonify({"error": str(error)}), 400

    authenticated, error = register_account(data)
    if error:
        return error

    return jsonify(
        {
            "status": "registered",
            "user": authenticated,
            "csrf_token": csrf_token(),
        }
    ), 201


@app.route("/api/login", methods=["POST"])
def login():
    from auth import rate_limited, csrf_valid, safe_json, authenticate_account

    limited = rate_limited("login")
    if limited:
        return limited

    if not csrf_valid():
        return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

    try:
        data = safe_json()
    except ValueError as error:
        return jsonify({"error": str(error)}), 400

    authenticated, error = authenticate_account(data)
    if error:
        return error

    return jsonify(
        {
            "status": "logged_in",
            "user": authenticated,
            "csrf_token": csrf_token(),
        }
    )


@app.route("/api/logout", methods=["POST"])
def logout():
    from auth import csrf_valid

    if not csrf_valid():
        return jsonify({"error": "Security token expired.", "code": "csrf"}), 403
    session.clear()
    return jsonify({"status": "logged_out", "csrf_token": csrf_token()})


def _maybe_cleanup_orphaned_attachments():
    maybe_cleanup_orphaned_attachments(
        RUNTIME_STATE,
        ATTACHMENT_CLEANUP_INTERVAL_SECONDS,
        lambda: cleanup_orphaned_attachments(
            UPLOAD_DIR,
            utc_now,
            ATTACHMENT_ORPHAN_TTL_HOURS,
            app.logger,
        ),
        app.logger,
    )


app.maybe_cleanup_orphaned_attachments = _maybe_cleanup_orphaned_attachments

register_routes(app)

init_db()
bootstrap_admin()

if __name__ == "__main__":
    app.run(
        host="0.0.0.0",
        port=int(os.environ.get("PORT", 5000)),
        debug=os.environ.get("FLASK_DEBUG", "0") == "1",
    )