authelia: household group; guildhall invite links
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
791a3bdf34
commit
782a1c96a6
6 changed files with 502 additions and 8 deletions
94
scripts/guildhall-join-check.py
Normal file
94
scripts/guildhall-join-check.py
Normal file
|
|
@ -0,0 +1,94 @@
|
|||
"""Self-check for guildhall-join.py: invites, the append-only users file, and
|
||||
that Authelia accepts the password hashes it writes.
|
||||
|
||||
Run it against the same python the service uses, with authelia on PATH:
|
||||
|
||||
py=$(nix build --no-link --print-out-paths --impure --expr \
|
||||
'with import <nixpkgs> {}; python3.withPackages (ps: [ ps.pyyaml ps.argon2-cffi ])')
|
||||
nix shell nixpkgs#authelia -c $py/bin/python3 scripts/guildhall-join-check.py
|
||||
"""
|
||||
import importlib.util
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
TMP = Path(tempfile.mkdtemp())
|
||||
os.environ["GUILDHALL_JOIN_STATE"] = str(TMP / "state")
|
||||
os.environ["GUILDHALL_JOIN_USERS"] = str(TMP / "users_database.yml")
|
||||
|
||||
spec = importlib.util.spec_from_file_location("join", HERE / "guildhall-join.py")
|
||||
j = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(j)
|
||||
|
||||
ORIGINAL = """users:
|
||||
fred:
|
||||
password: "$argon2id$v=19$m=65536,t=3,p=4$c2FsdA$aGFzaA"
|
||||
displayname: Fred
|
||||
email: fred@example.org
|
||||
groups: [household]
|
||||
|
||||
kayla:
|
||||
password: "$argon2id$v=19$m=65536,t=3,p=4$c2FsdA$aGFzaA"
|
||||
displayname: Kayla
|
||||
groups: [household]
|
||||
"""
|
||||
j.USERS_FILE.write_text(ORIGINAL)
|
||||
failures = 0
|
||||
|
||||
|
||||
def check(label, ok):
|
||||
global failures
|
||||
print(("ok " if ok else "FAIL ") + label)
|
||||
failures += not ok
|
||||
|
||||
|
||||
def refused(fn, *args):
|
||||
try:
|
||||
fn(*args)
|
||||
except j.Refused as reason:
|
||||
return str(reason)
|
||||
return None
|
||||
|
||||
|
||||
token = j.mint("for alex")
|
||||
check("fresh invite is open", j.open_invite(token) is not None)
|
||||
check("state holds digests, not tokens", token not in (TMP / "state" / "invites.json").read_text())
|
||||
check("unknown token is closed", j.open_invite("x" * 24) is None)
|
||||
|
||||
check("bad username refused", refused(j.sign_up, token, "Al", "", "longpassword", "longpassword"))
|
||||
check("existing username refused", refused(j.sign_up, token, "Kayla", "", "longpassword", "longpassword"))
|
||||
check("short password refused", refused(j.sign_up, token, "alex", "", "short", "short"))
|
||||
check("mismatched passwords refused", refused(j.sign_up, token, "alex", "", "longpassword", "longpassword2"))
|
||||
check("refusals leave the file alone", j.USERS_FILE.read_text() == ORIGINAL)
|
||||
check("refusals leave the invite open", j.open_invite(token) is not None)
|
||||
|
||||
name = j.sign_up(token, " Alex ", 'Alex "the Bold" Æ', "correct horse battery", "correct horse battery")
|
||||
users = yaml.safe_load(j.USERS_FILE.read_text())["users"]
|
||||
check("username normalised", name == "alex")
|
||||
check("old users unchanged", {k: users[k] for k in ("fred", "kayla")} == yaml.safe_load(ORIGINAL)["users"])
|
||||
check("original text kept byte for byte", j.USERS_FILE.read_text().startswith(ORIGINAL.rstrip("\n")))
|
||||
check("new user only in guildhall group", users["alex"]["groups"] == ["guildhall"])
|
||||
check("display name survives quoting", users["alex"]["displayname"] == 'Alex "the Bold" Æ')
|
||||
check("invite is spent", j.open_invite(token) is None)
|
||||
check("spent invite refused", refused(j.sign_up, token, "sam", "", "longpassword", "longpassword"))
|
||||
|
||||
expired = j.mint("old", now=1.0)
|
||||
check("expired invite is closed", j.open_invite(expired) is None)
|
||||
|
||||
if shutil.which("authelia"):
|
||||
result = subprocess.run(
|
||||
["authelia", "crypto", "hash", "validate", "--password", "correct horse battery", "--", users["alex"]["password"]],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
check("authelia accepts the hash", result.returncode == 0 and "matches" in result.stdout.lower())
|
||||
else:
|
||||
print("skip authelia hash check (authelia not on PATH)")
|
||||
|
||||
shutil.rmtree(TMP)
|
||||
sys.exit(1 if failures else 0)
|
||||
294
scripts/guildhall-join.py
Normal file
294
scripts/guildhall-join.py
Normal file
|
|
@ -0,0 +1,294 @@
|
|||
#!/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 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.")
|
||||
block = (
|
||||
f" {username}:\n"
|
||||
f" displayname: {json.dumps(display)}\n"
|
||||
f" password: {json.dumps(password_hash)}\n"
|
||||
f" groups:\n"
|
||||
f" - {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:]))
|
||||
Loading…
Add table
Add a link
Reference in a new issue