139 lines
7.3 KiB
Python
139 lines
7.3 KiB
Python
#!/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)
|