#!/usr/bin/env python3 """Prepare PocketBase users for named logins. Dry-run unless --apply is supplied. Run with --env-file pointing to the deployment's private env file. Applying requires --confirm-host and --bootstrap-user so the target and first app administrator are explicit. Back up PocketBase before applying. """ import argparse import os from pathlib import Path import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from pb_env import http, load_env_local # noqa: E402 def api(method, path, token, body=None): status, data = http(method, path, body, token) if status < 200 or status >= 300: raise RuntimeError(f"{method} {path} failed ({status})") return data def field_names(collection): return {field["name"] for field in collection["fields"]} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--env-file", type=Path) parser.add_argument("--apply", action="store_true") parser.add_argument("--confirm-host") parser.add_argument("--bootstrap-user", help="Existing user's exact email to make first app admin") parser.add_argument("--backup-confirmed", type=Path, help="Path to a verified PocketBase backup") parser.add_argument("--backup-key", help="Name of a verified backup in PocketBase's backup service") args = parser.parse_args() if args.env_file: load_env_local(args.env_file) base = (os.environ.get("POCKETBASE_URL") or os.environ.get("NEXT_PUBLIC_POCKETBASE_URL") or "").rstrip("/") if not base: parser.error("PocketBase URL is required") if args.backup_confirmed and args.backup_key: parser.error("Use either --backup-confirmed or --backup-key, not both") if args.apply and (args.confirm_host != base or not args.bootstrap_user or not (args.backup_confirmed and args.backup_confirmed.is_file()) and not args.backup_key): parser.error("--apply requires matching --confirm-host, --bootstrap-user, and a verified backup path or key") print(f"Target: {base} | {'APPLY' if args.apply else 'DRY RUN'}") email = os.environ.get("POCKETBASE_ADMIN_EMAIL") password = os.environ.get("POCKETBASE_ADMIN_PASSWORD") if not email or not password: parser.error("PocketBase superuser credentials are required") status, auth = http("POST", "/api/collections/_superusers/auth-with-password", {"identity": email, "password": password}) if status != 200: raise RuntimeError(f"PocketBase superuser authentication failed ({status})") token = auth["token"] if args.backup_key: backups = api("GET", "/api/backups", token) backup = next((row for row in backups if row.get("key") == args.backup_key), None) if not backup or not isinstance(backup.get("size"), int) or backup["size"] <= 0: raise RuntimeError("Named PocketBase backup is missing or empty") print(f"Verified PocketBase backup: {backup['key']} ({backup['size']} bytes)") users = api("GET", "/api/collections/users", token) sites = api("GET", "/api/collections/sites", token) if users["type"] != "auth": raise RuntimeError("users is not an auth collection") expected = [ {"name": "role", "type": "select", "required": True, "maxSelect": 1, "values": ["admin", "site_user", "clock_only"]}, {"name": "sites", "type": "relation", "required": False, "collectionId": sites["id"], "cascadeDelete": False, "maxSelect": 99}, {"name": "is_active", "type": "bool", "required": False}, {"name": "must_change_password", "type": "bool", "required": False}, ] present = field_names(users) missing = [field for field in expected if field["name"] not in present] for field in expected: old = next((item for item in users["fields"] if item["name"] == field["name"]), None) if old and old["type"] != field["type"]: raise RuntimeError(f"users.{field['name']} exists with unexpected type") print("users: add " + (", ".join(field["name"] for field in missing) or "no fields")) print("users: disable public signup and direct record changes; require active login") collections = api("GET", "/api/collections?perPage=500", token) if collections.get("totalItems", 0) > 500: raise RuntimeError("More than 500 collections; review rules manually") rule_names = ("listRule", "viewRule", "createRule", "updateRule", "deleteRule") open_collections = [row for row in collections["items"] if row["type"] == "base" and not row["name"].startswith("_") and any(row.get(name) is not None for name in rule_names)] print(f"Base collections with direct user API rules to lock: {len(open_collections)}") for row in open_collections: print(f" {row['name']}") exposed_files = [(row, [field["name"] for field in row["fields"] if field["type"] == "file" and not field.get("protected")]) for row in collections["items"] if row["type"] == "base"] exposed_files = [(row, names) for row, names in exposed_files if names] print(f"Data collections with public file fields to protect: {len(exposed_files)}") for row, names in exposed_files: print(f" {row['name']}: {', '.join(names)}") records = api("GET", "/api/collections/users/records?perPage=100&fields=id,email,role,is_active", token) if records.get("totalItems", 0) > 100: raise RuntimeError("More than 100 users; review bootstrap manually") target = next((row for row in records["items"] if row["email"].lower() == (args.bootstrap_user or "").lower()), None) if args.bootstrap_user and not target: raise RuntimeError("Bootstrap user email does not match an existing PocketBase user") print(f"Existing users: {records.get('totalItems', 0)}; bootstrap: {'matched' if target else 'not selected'}") if not args.apply: print("No writes. Back up PocketBase, then run with --apply, --confirm-host, --bootstrap-user, and --backup-confirmed or --backup-key.") return if missing: api("PATCH", "/api/collections/users", token, {"fields": users["fields"] + missing}) api("PATCH", f"/api/collections/users/records/{target['id']}", token, {"role": "admin", "sites": [], "is_active": True, "must_change_password": False}) api("PATCH", "/api/collections/users", token, {**{name: None for name in rule_names}, "authRule": "is_active = true"}) for row in open_collections: api("PATCH", f"/api/collections/{row['id']}", token, {name: None for name in rule_names}) for row, names in exposed_files: changed = [{**field, "protected": True} if field["name"] in names else field for field in row["fields"]] api("PATCH", f"/api/collections/{row['id']}", token, {"fields": changed}) verified = api("GET", f"/api/collections/users/records/{target['id']}", token) if verified.get("role") != "admin" or verified.get("is_active") is not True: raise RuntimeError("Bootstrap admin verification failed") print("Users schema, bootstrap administrator, direct API rules, and file protection updated. AUTH_MODE remains unchanged.") if __name__ == "__main__": try: main() except RuntimeError as error: print(f"ERROR: {error}", file=sys.stderr) sys.exit(1)