#!/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 = """ Join Guildhall
{body}
""" FORM = """

🏰 Join Guildhall

You've been invited to a small co-op idle game. Pick a login for the city gates.

{error}

What you type when signing in. Lowercase, 3 to 20 characters.

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

This invite doesn't work

It may have been used already, or it expired. " "Ask whoever sent it for a new one.

", HTTPStatus.NOT_FOUND) def form(self, error: str = "", username: str = "", display: str = "") -> None: self.page(FORM.format( error=f'

{html.escape(error)}

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

Welcome, {html.escape(name)}!

Your account is ready. Sign in with it at the city gates " f"and create your adventurer.

Go to Guildhall") # --- 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:]))