nixos/scripts/guildhall-join.py

#!/usr/bin/env python3
"""guildhall-join: one-time invite links that create Authelia accounts.

    guildhall-join invite [NOTE]   mint a link, valid for a week, print it
    guildhall-join list            show open invites
    guildhall-join revoke ID       cancel an invite (ID from `list`)
    guildhall-join serve           run the /join/ page behind nginx

Accounts land in the Authelia file database with exactly one group,
`guildhall`, which the access rules only let into the game. The file is only
ever appended to: before the new version replaces the old one, it is parsed
back and every existing user must come out unchanged. Invites are stored as
SHA-256 digests, so the state file alone can't be used to sign up.
"""

from __future__ import annotations

import fcntl
import hashlib
import html
import json
import os
import re
import secrets
import sys
import tempfile
import time
from contextlib import contextmanager
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs

import yaml
from argon2 import PasswordHasher

STATE_DIR = Path(os.environ.get("GUILDHALL_JOIN_STATE", "/var/lib/guildhall-join"))
USERS_FILE = Path(os.environ.get("GUILDHALL_JOIN_USERS", "/var/lib/authelia-users/users_database.yml"))
SITE = os.environ.get("GUILDHALL_JOIN_SITE", "https://guildhall.nordhammer.it")
PORT = int(os.environ.get("GUILDHALL_JOIN_PORT", "8099"))
GROUP = "guildhall"
INVITE_DAYS = 7
USERNAME = re.compile(r"^[a-z][a-z0-9_-]{2,19}$")
TOKEN_PATH = re.compile(r"^/join/([A-Za-z0-9_-]{16,64})/?$")
MIN_PASSWORD = 10
MAX_BODY = 4096

# Authelia's own argon2id defaults, so its hashes and ours look the same.
HASHER = PasswordHasher(time_cost=3, memory_cost=65536, parallelism=4, hash_len=32, salt_len=16)


class Refused(Exception):
    """A reason to show the person filling in the form."""


# --- invites ---------------------------------------------------------------

def digest(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


@contextmanager
def locked():
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    with open(STATE_DIR / "lock", "w") as lock:
        fcntl.flock(lock, fcntl.LOCK_EX)
        yield


def load_invites() -> dict:
    try:
        return json.loads((STATE_DIR / "invites.json").read_text())
    except FileNotFoundError:
        return {}


def save_invites(invites: dict) -> None:
    write_atomic(STATE_DIR / "invites.json", json.dumps(invites, indent=2) + "\n", 0o600)


def mint(note: str, now: float | None = None) -> str:
    now = now or time.time()
    token = secrets.token_urlsafe(18)
    with locked():
        invites = load_invites()
        invites[digest(token)] = {"note": note, "created": now, "expires": now + INVITE_DAYS * 86400, "used_by": None}
        save_invites(invites)
    return token


def open_invite(token: str, now: float | None = None) -> dict | None:
    invite = load_invites().get(digest(token))
    if not invite or invite["used_by"] or invite["expires"] < (now or time.time()):
        return None
    return invite


# --- the Authelia users file -------------------------------------------------

def write_atomic(path: Path, text: str, mode: int, gid: int = -1) -> None:
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.")
    try:
        with os.fdopen(fd, "w") as out:
            out.write(text)
            out.flush()
            os.fsync(out.fileno())
        os.chmod(tmp, mode)
        if gid != -1:
            os.chown(tmp, -1, gid)
        os.replace(tmp, path)
    except BaseException:
        os.unlink(tmp)
        raise


def user_indent(text: str) -> str:
    """The indent of the existing user keys. Authelia rewrites the whole file
    with four spaces when someone resets their password; by hand it's two."""
    match = re.search(r"^users:[ \t]*\n(?:[ \t]*(?:#.*)?\n)*( +)\S", text, re.M)
    return match.group(1) if match else "  "


def add_user(path: Path, username: str, display: str, password_hash: str) -> None:
    """Append one user; refuse unless every existing user parses back unchanged."""
    text = path.read_text()
    before = yaml.safe_load(text)["users"]
    if username.lower() in (name.lower() for name in before):
        raise Refused("That username is taken.")
    i = user_indent(text)
    block = (
        f"{i}{username}:\n"
        f"{i * 2}displayname: {json.dumps(display)}\n"
        f"{i * 2}password: {json.dumps(password_hash)}\n"
        f"{i * 2}groups:\n"
        f"{i * 3}- {GROUP}\n"
    )
    new_text = text.rstrip("\n") + "\n\n" + block
    after = yaml.safe_load(new_text)["users"]
    expected = {**before, username: {"displayname": display, "password": password_hash, "groups": [GROUP]}}
    if after != expected:
        raise RuntimeError("appending to the users file would change other entries; refusing")
    info = path.stat()
    write_atomic(path, new_text, info.st_mode & 0o777, info.st_gid)


def sign_up(token: str, username: str, display: str, password: str, confirm: str, now: float | None = None) -> str:
    username = username.strip().lower()
    display = " ".join(display.split()) or username.capitalize()
    if not USERNAME.match(username):
        raise Refused("Usernames are 3 to 20 characters: lowercase letters, digits, - and _, starting with a letter.")
    if len(display) > 30 or not display.isprintable():
        raise Refused("Keep the display name under 30 characters.")
    if len(password) < MIN_PASSWORD:
        raise Refused(f"Use at least {MIN_PASSWORD} characters for the password.")
    if len(password) > 128:
        raise Refused("That password is too long.")
    if password != confirm:
        raise Refused("The two passwords don't match.")
    password_hash = HASHER.hash(password)
    with locked():
        invites = load_invites()
        if not open_invite(token, now):
            raise Refused("This invite has already been used or has expired.")
        add_user(USERS_FILE, username, display, password_hash)
        invites[digest(token)]["used_by"] = username
        save_invites(invites)
    return username


# --- the page ----------------------------------------------------------------

PAGE = """<!doctype html>
<html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<meta name="referrer" content="no-referrer">
<title>Join Guildhall</title>
<style>
  :root {{ color-scheme: dark; }}
  body {{ margin: 0; min-height: 100vh; display: grid; place-items: center; padding: 16px; box-sizing: border-box;
         background: #14161d; color: #e7e9f0; font: 15px/1.5 system-ui, -apple-system, "Segoe UI", Roboto, sans-serif; }}
  main {{ width: 100%; max-width: 400px; background: #212532; border: 1px solid #353b4d; border-radius: 14px; padding: 24px; }}
  h1 {{ margin: 0 0 4px; font-size: 22px; }}
  p {{ color: #98a1b4; margin: 0 0 16px; }}
  label {{ display: block; font-size: 13px; color: #98a1b4; margin: 12px 0 4px; }}
  input {{ width: 100%; box-sizing: border-box; background: #14161d; color: inherit; border: 1px solid #353b4d;
          border-radius: 8px; padding: 9px 11px; font: inherit; }}
  button, .go {{ display: inline-block; margin-top: 18px; width: 100%; box-sizing: border-box; text-align: center;
                background: #2f6b3a; border: 1px solid #3f8a4c; color: #fff; border-radius: 8px; padding: 10px;
                font: inherit; font-weight: 600; cursor: pointer; text-decoration: none; }}
  .error {{ color: #e2645c; background: rgba(226,100,92,.1); border: 1px solid rgba(226,100,92,.4);
           border-radius: 8px; padding: 8px 12px; }}
  .hint {{ font-size: 12.5px; margin: 4px 0 0; }}
</style></head>
<body><main>{body}</main></body></html>"""

FORM = """<h1>🏰 Join Guildhall</h1>
<p>You've been invited to a small co-op idle game. Pick a login for the city gates.</p>
{error}
<form method="post">
  <label for="username">Username</label>
  <input id="username" name="username" value="{username}" autocomplete="username" autocapitalize="none" required>
  <p class="hint">What you type when signing in. Lowercase, 3 to 20 characters.</p>
  <label for="display">Display name <span>(optional)</span></label>
  <input id="display" name="display" value="{display}" maxlength="30">
  <label for="password">Password</label>
  <input id="password" name="password" type="password" autocomplete="new-password" minlength="{min}" required>
  <label for="confirm">Password again</label>
  <input id="confirm" name="confirm" type="password" autocomplete="new-password" minlength="{min}" required>
  <button>Create my account</button>
</form>"""


class Handler(BaseHTTPRequestHandler):
    server_version = "guildhall-join"

    def log_message(self, fmt: str, *args) -> None:
        # Never log paths: they carry the invite token.
        pass

    def page(self, body: str, status: HTTPStatus = HTTPStatus.OK) -> None:
        data = PAGE.format(body=body).encode()
        self.send_response(status)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.send_header("Referrer-Policy", "no-referrer")
        self.send_header("X-Frame-Options", "DENY")
        self.end_headers()
        self.wfile.write(data)

    def dead_end(self) -> None:
        self.page("<h1>This invite doesn't work</h1><p>It may have been used already, or it expired. "
                  "Ask whoever sent it for a new one.</p>", HTTPStatus.NOT_FOUND)

    def form(self, error: str = "", username: str = "", display: str = "") -> None:
        self.page(FORM.format(
            error=f'<p class="error">{html.escape(error)}</p>' if error else "",
            username=html.escape(username), display=html.escape(display), min=MIN_PASSWORD,
        ), HTTPStatus.BAD_REQUEST if error else HTTPStatus.OK)

    def do_GET(self) -> None:
        match = TOKEN_PATH.match(self.path.split("?")[0])
        if not match or not open_invite(match.group(1)):
            return self.dead_end()
        self.form()

    def do_POST(self) -> None:
        match = TOKEN_PATH.match(self.path.split("?")[0])
        if not match or not open_invite(match.group(1)):
            return self.dead_end()
        length = int(self.headers.get("Content-Length") or 0)
        if length > MAX_BODY:
            return self.form("That was too much to send.")
        fields = {k: v[0] for k, v in parse_qs(self.rfile.read(length).decode(errors="replace")).items()}
        username, display = fields.get("username", ""), fields.get("display", "")
        try:
            name = sign_up(match.group(1), username, display, fields.get("password", ""), fields.get("confirm", ""))
        except Refused as reason:
            return self.form(str(reason), username, display)
        print(f"created account {name}", file=sys.stderr, flush=True)
        self.page(f"<h1>Welcome, {html.escape(name)}!</h1><p>Your account is ready. Sign in with it at the city gates "
                  f"and create your adventurer.</p><a class=\"go\" href=\"{html.escape(SITE)}\">Go to Guildhall</a>")


# --- command line --------------------------------------------------------------

def main(argv: list[str]) -> int:
    command = argv[0] if argv else "invite"
    if command == "invite":
        token = mint(" ".join(argv[1:]))
        print(f"{SITE}/join/{token}")
        print(f"Single use, valid for {INVITE_DAYS} days.", file=sys.stderr)
    elif command == "list":
        now = time.time()
        for key, invite in sorted(load_invites().items(), key=lambda kv: kv[1]["created"]):
            if invite["used_by"]:
                state = f"used by {invite['used_by']}"
            elif invite["expires"] < now:
                state = "expired"
            else:
                state = f"open, {(invite['expires'] - now) / 86400:.1f} days left"
            print(f"{key[:8]}  {time.strftime('%Y-%m-%d', time.localtime(invite['created']))}  {state:28}  {invite['note']}")
    elif command == "revoke" and len(argv) == 2:
        with locked():
            invites = load_invites()
            hits = [key for key in invites if key.startswith(argv[1])]
            if len(hits) != 1:
                print("No single invite matches that ID.", file=sys.stderr)
                return 1
            del invites[hits[0]]
            save_invites(invites)
        print("Revoked.")
    elif command == "serve":
        ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever()
    else:
        print(__doc__, file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))