From 0966df2df28ab212a6f4736a4e17a31d690a6173 Mon Sep 17 00:00:00 2001
From: yyamashita <yyamashita@hetzner.yyamashita.com>
Date: Sat, 31 Oct 2026 19:51:15 +0900
Subject: hetzner-admin: add passkey login (password login kept)

Replace Basic auth with a session-cookie login page. After password
login, register a passkey from the bottom of the page; later logins
can use it. Passkeys are stored in hetzner-admin/data/ (gitignored).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 hetzner-admin/app.py | 232 ++++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 221 insertions(+), 11 deletions(-)

(limited to 'hetzner-admin/app.py')

diff --git a/hetzner-admin/app.py b/hetzner-admin/app.py
index abef6a9..90a1cca 100644
--- a/hetzner-admin/app.py
+++ b/hetzner-admin/app.py
@@ -1,10 +1,26 @@
 import functools
+import hmac
+import json
 import os
+import secrets
+import time
+from datetime import timedelta
 
 import paramiko
-from flask import Flask, Response, render_template_string, request
-
-app = Flask(__name__)
+from flask import Flask, redirect, render_template_string, request, session
+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,
+)
 
 ADMIN_USER = os.environ["ADMIN_USER"]
 ADMIN_PASS = os.environ["ADMIN_PASS"]
@@ -12,6 +28,30 @@ SSH_HOST = os.environ.get("SSH_HOST", "host.docker.internal")
 SSH_PORT = int(os.environ.get("SSH_PORT", "22"))
 SSH_USER = os.environ.get("SSH_USER", "root")
 SSH_KEY_PATH = os.environ.get("SSH_KEY_PATH", "/app/id_rsa")
+RP_ID = os.environ.get("RP_ID", "hetzner-admin.yyamashita.com")
+ORIGIN = f"https://{RP_ID}"
+DATA_DIR = os.environ.get("DATA_DIR", "/app/data")
+PASSKEYS_PATH = os.path.join(DATA_DIR, "passkeys.json")
+SECRET_PATH = os.path.join(DATA_DIR, "secret_key")
+
+
+def load_secret():
+    os.makedirs(DATA_DIR, exist_ok=True)
+    if not os.path.exists(SECRET_PATH):
+        with open(SECRET_PATH, "w") as f:
+            f.write(secrets.token_hex(32))
+    with open(SECRET_PATH) as f:
+        return f.read().strip()
+
+
+app = Flask(__name__)
+app.secret_key = load_secret()
+app.config.update(
+    SESSION_COOKIE_SECURE=True,
+    SESSION_COOKIE_HTTPONLY=True,
+    SESSION_COOKIE_SAMESITE="Lax",
+    PERMANENT_SESSION_LIFETIME=timedelta(days=30),
+)
 
 COMMANDS = [
     ("Docker Status", "docker ps"),
@@ -25,9 +65,45 @@ COMMANDS = [
 
 COMMAND_MAP = {label: cmd for label, cmd in COMMANDS}
 
+# ブラウザ側: JSON ⇔ WebAuthn の変換はブラウザ組み込みの
+# parseCreationOptionsFromJSON / parseRequestOptionsFromJSON / toJSON に任せる
+PASSKEY_JS = """<script>
+async function passkey(kind) {
+  try {
+    const opts = await (await fetch('/passkey/' + kind + '/options', {method: 'POST'})).json();
+    const cred = kind === 'register'
+      ? await navigator.credentials.create({publicKey: PublicKeyCredential.parseCreationOptionsFromJSON(opts)})
+      : await navigator.credentials.get({publicKey: PublicKeyCredential.parseRequestOptionsFromJSON(opts)});
+    const res = await fetch('/passkey/' + kind + '/verify', {
+      method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify(cred.toJSON())});
+    const msg = await res.text();
+    if (!res.ok) throw new Error(msg);
+    if (kind === 'login') location.href = '/'; else alert(msg);
+  } catch (e) {
+    alert('Passkey error: ' + e.message);
+  }
+}
+</script>"""
+
+LOGIN_HTML = """<!DOCTYPE html>
+<html>
+<head><title>Server Admin</title><meta name="viewport" content="width=device-width"></head>
+<body>
+<h3>Server Admin</h3>
+<p><button onclick="passkey('login')">Login with Passkey</button></p>
+<form method="post" action="/login">
+  <input name="username" placeholder="user" autocomplete="username">
+  <input name="password" type="password" placeholder="password" autocomplete="current-password">
+  <button>Login</button>
+</form>
+{% if error %}<p style="color:red">{{ error }}</p>{% endif %}
+""" + PASSKEY_JS + """
+</body>
+</html>"""
+
 HTML = """<!DOCTYPE html>
 <html>
-<head><title>Server Admin</title></head>
+<head><title>Server Admin</title><meta name="viewport" content="width=device-width"></head>
 <body>
 <h3>Server Admin</h3>
 <p><a href="https://status.yyamashita.com">Status</a></p>
@@ -45,23 +121,48 @@ HTML = """<!DOCTYPE html>
 <h4>Output:</h4>
 <pre>{{ output }}</pre>
 {% endif %}
+<hr>
+<p>
+  Passkeys registered: {{ passkey_count }}
+  <button onclick="passkey('register')">Register this device's passkey</button>
+  <a href="/logout">Logout</a>
+</p>
+""" + PASSKEY_JS + """
 </body>
 </html>"""
 
 
+def load_passkeys():
+    try:
+        with open(PASSKEYS_PATH) as f:
+            return json.load(f)
+    except FileNotFoundError:
+        return []
+
+
+def save_passkeys(keys):
+    tmp = PASSKEYS_PATH + ".tmp"
+    with open(tmp, "w") as f:
+        json.dump(keys, f, indent=2)
+    os.replace(tmp, PASSKEYS_PATH)
+
+
 def requires_auth(f):
     @functools.wraps(f)
     def decorated(*args, **kwargs):
-        auth = request.authorization
-        if not auth or auth.username != ADMIN_USER or auth.password != ADMIN_PASS:
-            return Response(
-                "Unauthorized", 401, {"WWW-Authenticate": 'Basic realm="Admin"'}
-            )
+        if not session.get("auth"):
+            return redirect("/login")
         return f(*args, **kwargs)
 
     return decorated
 
 
+def log_in():
+    session.clear()
+    session.permanent = True
+    session["auth"] = True
+
+
 def run_ssh(command):
     key = paramiko.RSAKey.from_private_key_file(SSH_KEY_PATH)
     client = paramiko.SSHClient()
@@ -76,10 +177,119 @@ def run_ssh(command):
         client.close()
 
 
+def render_index(output=None):
+    return render_template_string(
+        HTML, commands=COMMANDS, output=output, passkey_count=len(load_passkeys())
+    )
+
+
+@app.route("/login", methods=["GET", "POST"])
+def login():
+    error = None
+    if request.method == "POST":
+        user_ok = hmac.compare_digest(request.form.get("username", ""), ADMIN_USER)
+        pass_ok = hmac.compare_digest(request.form.get("password", ""), ADMIN_PASS)
+        if user_ok and pass_ok:
+            log_in()
+            return redirect("/")
+        error = "Wrong user or password"
+    return render_template_string(LOGIN_HTML, error=error)
+
+
+@app.route("/logout")
+def logout():
+    session.clear()
+    return redirect("/login")
+
+
+@app.route("/passkey/register/options", methods=["POST"])
+@requires_auth
+def passkey_register_options():
+    opts = generate_registration_options(
+        rp_id=RP_ID,
+        rp_name="Server Admin",
+        user_name=ADMIN_USER,
+        user_id=ADMIN_USER.encode(),
+        authenticator_selection=AuthenticatorSelectionCriteria(
+            resident_key=ResidentKeyRequirement.REQUIRED
+        ),
+        exclude_credentials=[
+            PublicKeyCredentialDescriptor(id=base64url_to_bytes(k["id"]))
+            for k in load_passkeys()
+        ],
+    )
+    session["challenge"] = bytes_to_base64url(opts.challenge)
+    return app.response_class(options_to_json(opts), mimetype="application/json")
+
+
+@app.route("/passkey/register/verify", methods=["POST"])
+@requires_auth
+def passkey_register_verify():
+    challenge = session.pop("challenge", None)
+    if not challenge:
+        return "No challenge", 400
+    try:
+        v = verify_registration_response(
+            credential=request.get_data(as_text=True),
+            expected_challenge=base64url_to_bytes(challenge),
+            expected_rp_id=RP_ID,
+            expected_origin=ORIGIN,
+        )
+    except Exception as e:
+        return f"Registration failed: {e}", 400
+    keys = load_passkeys()
+    keys.append(
+        {
+            "id": bytes_to_base64url(v.credential_id),
+            "public_key": bytes_to_base64url(v.credential_public_key),
+            "sign_count": v.sign_count,
+            "created": time.strftime("%Y-%m-%d %H:%M:%S"),
+            "user_agent": request.headers.get("User-Agent", ""),
+        }
+    )
+    save_passkeys(keys)
+    return "Passkey registered"
+
+
+@app.route("/passkey/login/options", methods=["POST"])
+def passkey_login_options():
+    # allow_credentials を空にしてブラウザ側で保存済みパスキーを選ばせる
+    opts = generate_authentication_options(rp_id=RP_ID)
+    session["challenge"] = bytes_to_base64url(opts.challenge)
+    return app.response_class(options_to_json(opts), mimetype="application/json")
+
+
+@app.route("/passkey/login/verify", methods=["POST"])
+def passkey_login_verify():
+    challenge = session.pop("challenge", None)
+    if not challenge:
+        return "No challenge", 400
+    cred = request.get_json(silent=True) or {}
+    keys = load_passkeys()
+    key = next((k for k in keys if k["id"] == cred.get("id")), None)
+    if not key:
+        return "Unknown passkey (register it first after password login)", 400
+    try:
+        v = verify_authentication_response(
+            credential=cred,
+            expected_challenge=base64url_to_bytes(challenge),
+            expected_rp_id=RP_ID,
+            expected_origin=ORIGIN,
+            credential_public_key=base64url_to_bytes(key["public_key"]),
+            credential_current_sign_count=key["sign_count"],
+        )
+    except Exception as e:
+        return f"Login failed: {e}", 400
+    key["sign_count"] = v.new_sign_count
+    save_passkeys(keys)
+    log_in()
+    return "OK"
+
+
 @app.route("/")
 @requires_auth
 def index():
-    return render_template_string(HTML, commands=COMMANDS, output=None)
+    return render_index()
 
 
 @app.route("/run", methods=["POST"])
@@ -94,4 +304,4 @@ def run():
             output = run_ssh(command)
         except Exception as e:
             output = f"Error: {e}"
-    return render_template_string(HTML, commands=COMMANDS, output=output)
+    return render_index(output)
-- 
cgit v1.3