Moon-Household-Budget/auth_routes.py

"""WebAuthn passkey routes + the login gate.

Pages: /login (passkey sign-in), /enroll (register a device), /logout.
JSON ceremony endpoints under /auth/*. init_auth(app) wires the blueprint and a
before_request gate that protects everything when AUTH_ENABLED is on.
"""
import json
from datetime import timedelta

from flask import (Blueprint, abort, jsonify, redirect, render_template,
                   request, session, url_for)
from webauthn import (generate_authentication_options,
                      generate_registration_options, options_to_json,
                      verify_authentication_response,
                      verify_registration_response)
from webauthn.helpers import base64url_to_bytes, bytes_to_base64url
from webauthn.helpers.structs import (AuthenticatorSelectionCriteria,
                                      PublicKeyCredentialDescriptor,
                                      ResidentKeyRequirement,
                                      UserVerificationRequirement)

import auth

auth_bp = Blueprint("auth", __name__)

# Endpoints reachable WITHOUT a session (the login/enroll flow itself).
OPEN_ENDPOINTS = {
    "auth.login", "auth.enroll", "auth.logout",
    "auth.register_begin", "auth.register_complete",
    "auth.login_begin", "auth.login_complete",
    "static",
}


def _json_body():
    return request.get_json(silent=True) or {}


def _enroll_authorized():
    """May the caller register a new device right now?"""
    if session.get("authed"):
        return True
    code = auth.enrollment_code()
    if code:
        supplied = (_json_body().get("code")
                    or request.args.get("code")
                    or request.headers.get("X-Enroll-Code"))
        return supplied == code
    # No code configured: allow only to bootstrap the very first device.
    return not auth.has_credentials()


# ---- pages ----
@auth_bp.route("/login")
def login():
    if session.get("authed"):
        return redirect(url_for("index"))
    return render_template("login.html", rp_id=auth.rp_id(),
                           has_credentials=auth.has_credentials())


@auth_bp.route("/enroll")
def enroll():
    return render_template("enroll.html", rp_id=auth.rp_id(),
                           needs_code=bool(auth.enrollment_code()) and not session.get("authed"),
                           logged_in=bool(session.get("authed")))


@auth_bp.route("/logout", methods=["GET", "POST"])
def logout():
    session.clear()
    return redirect(url_for("auth.login"))


# ---- registration ceremony ----
@auth_bp.route("/auth/register/begin", methods=["POST"])
def register_begin():
    if not _enroll_authorized():
        return jsonify({"error": "not_authorized"}), 403
    creds = auth.list_credentials()
    opts = generate_registration_options(
        rp_id=auth.rp_id(),
        rp_name=auth.rp_name(),
        user_name="household",
        user_id=auth.user_handle_bytes(),
        user_display_name="Moon Household",
        exclude_credentials=[
            PublicKeyCredentialDescriptor(id=base64url_to_bytes(c["id"])) for c in creds
        ],
        authenticator_selection=AuthenticatorSelectionCriteria(
            resident_key=ResidentKeyRequirement.PREFERRED,
            user_verification=UserVerificationRequirement.PREFERRED,
        ),
    )
    session["reg_challenge"] = bytes_to_base64url(opts.challenge)
    session["reg_label"] = (_json_body().get("label") or "Device")[:60]
    session["enroll_ok"] = True
    return options_to_json(opts), 200, {"Content-Type": "application/json"}


@auth_bp.route("/auth/register/complete", methods=["POST"])
def register_complete():
    if not session.pop("enroll_ok", False):
        return jsonify({"error": "not_authorized"}), 403
    challenge = session.pop("reg_challenge", None)
    if not challenge:
        return jsonify({"error": "no_challenge"}), 400
    try:
        v = verify_registration_response(
            credential=request.get_data(as_text=True),
            expected_challenge=base64url_to_bytes(challenge),
            expected_rp_id=auth.rp_id(),
            expected_origin=auth.origin(),
            require_user_verification=False,
        )
    except Exception as e:
        return jsonify({"error": f"verify_failed: {e}"}), 400
    auth.add_credential(
        bytes_to_base64url(v.credential_id),
        bytes_to_base64url(v.credential_public_key),
        v.sign_count,
        session.pop("reg_label", "Device"),
    )
    session["authed"] = True
    session.permanent = True
    return jsonify({"ok": True})


# ---- authentication ceremony ----
@auth_bp.route("/auth/login/begin", methods=["POST"])
def login_begin():
    creds = auth.list_credentials()
    opts = generate_authentication_options(
        rp_id=auth.rp_id(),
        allow_credentials=[
            PublicKeyCredentialDescriptor(id=base64url_to_bytes(c["id"])) for c in creds
        ],
        user_verification=UserVerificationRequirement.PREFERRED,
    )
    session["auth_challenge"] = bytes_to_base64url(opts.challenge)
    return options_to_json(opts), 200, {"Content-Type": "application/json"}


@auth_bp.route("/auth/login/complete", methods=["POST"])
def login_complete():
    challenge = session.pop("auth_challenge", None)
    if not challenge:
        return jsonify({"error": "no_challenge"}), 400
    body = request.get_data(as_text=True)
    cred = auth.get_credential(json.loads(body).get("id"))
    if not cred:
        return jsonify({"error": "unknown_credential"}), 400
    try:
        v = verify_authentication_response(
            credential=body,
            expected_challenge=base64url_to_bytes(challenge),
            expected_rp_id=auth.rp_id(),
            expected_origin=auth.origin(),
            credential_public_key=base64url_to_bytes(cred["public_key"]),
            credential_current_sign_count=cred["sign_count"],
            require_user_verification=False,
        )
    except Exception as e:
        return jsonify({"error": f"verify_failed: {e}"}), 400
    auth.update_sign_count(cred["id"], v.new_sign_count)
    session["authed"] = True
    session.permanent = True
    return jsonify({"ok": True})


@auth_bp.route("/auth/device/remove", methods=["POST"])
def device_remove():
    if auth.auth_enabled() and not session.get("authed"):
        abort(403)
    cid = request.form.get("id")
    if cid:
        auth.remove_credential(cid)
    return redirect(url_for("settings_view"))


def init_auth(app):
    app.permanent_session_lifetime = timedelta(days=30)
    app.register_blueprint(auth_bp)

    @app.before_request
    def _gate():
        if not auth.auth_enabled():
            return
        ep = request.endpoint or ""
        if ep in OPEN_ENDPOINTS or ep.startswith("auth."):
            return
        if request.path.startswith(("/static", "/suggestions-static", "/auth/")):
            return
        if session.get("authed"):
            return
        if request.path.startswith("/api/") or request.is_json:
            return jsonify({"error": "unauthorized"}), 401
        return redirect(url_for("auth.login", next=request.path))