Add WebAuthn passkey auth (dormant behind AUTH_ENABLED)
- auth.py: passkey credential store (public keys only) + env-driven RP config. - auth_routes.py: /login, /enroll, /logout + /auth/* ceremony endpoints + before_request gate (no-op unless AUTH_ENABLED=1). - login/enroll pages (SimpleWebAuthn browser); settings page lists + removes enrolled devices. requirements: webauthn==2.7.1. - Deploys OFF: no behavior change until enabled after the hostname/cert exist. Closes groundwork for #5. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
198
auth_routes.py
Normal file
198
auth_routes.py
Normal file
@@ -0,0 +1,198 @@
|
||||
"""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))
|
||||
Reference in New Issue
Block a user