"""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))