path: root/hetzner-admin/app.py

import functools
import hmac
import json
import os
import secrets
import time
from datetime import timedelta

import paramiko
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"]
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"),
    ("Restart Caddy", "docker compose -f /app/infra/caddy/docker-compose.yml restart"),
    ("Restart Mail", "docker compose -f /app/infra/mail/docker-compose.yml restart"),
    ("Restart cgit", "docker compose -f /app/infra/cgit/docker-compose.yml restart"),
    ("Restart Admin", "docker compose -f /app/infra/hetzner-admin/docker-compose.yml restart"),
    ("Restart Claude Sessions", "bash /app/infra/claude/restart-sessions.sh"),
    ("Reboot Server", "reboot"),
]

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><meta name="viewport" content="width=device-width"></head>
<body>
<h3>Server Admin</h3>
<p><a href="https://status.yyamashita.com">Status</a></p>
{% for label, _ in commands %}
<form method="post" action="/run" style="display:block;margin-bottom:8px">
  <input type="hidden" name="cmd" value="{{ label }}">
  {% if label == "Reboot Server" %}
  <button onclick="return confirm('Really reboot?')">{{ label }}</button>
  {% else %}
  <button>{{ label }}</button>
  {% endif %}
</form>
{% endfor %}
{% if output is not none %}
<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):
        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()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(SSH_HOST, port=SSH_PORT, username=SSH_USER, pkey=key, timeout=10)
    try:
        _, stdout, stderr = client.exec_command(command, timeout=30)
        out = stdout.read().decode(errors="replace")
        err = stderr.read().decode(errors="replace")
        return (out + err).strip() or "(no output)"
    finally:
        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_index()


@app.route("/run", methods=["POST"])
@requires_auth
def run():
    label = request.form.get("cmd", "")
    command = COMMAND_MAP.get(label)
    if not command:
        output = f"Unknown command: {label}"
    else:
        try:
            output = run_ssh(command)
        except Exception as e:
            output = f"Error: {e}"
    return render_index(output)