#!/usr/bin/env python3 """Relax legacy read/create rules without opening clock audit writes. Pattern: list/view/create → any authenticated user update/delete → superuser only (audit-safe for OCR-verified records) Override per-collection with the RULES dict if you need different behaviour. """ import json import os import sys import urllib.request import urllib.error PB = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/") EMAIL = os.environ.get("PB_ADMIN_EMAIL") PASSWORD = os.environ.get("PB_ADMIN_PASSWORD") if not EMAIL or not PASSWORD: sys.exit("Set PB_ADMIN_EMAIL and PB_ADMIN_PASSWORD before updating collection rules") AUTH_RULE = "@request.auth.id != ''" # any logged-in user NONE_RULE = None # superuser only RULES = { "SITES": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "EMPLOYEES": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "PURCHASE_ORDERS": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "PURCHASE_ORDER_ITEMS": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "FIXED_COSTS": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "VEHICLES": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "FUEL_PRICES": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "TIMESHEETS": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "TIMESHEET_ENTRIES": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "CLOCK_EVENTS": (NONE_RULE, NONE_RULE, NONE_RULE, NONE_RULE, NONE_RULE), "clock_shifts": (NONE_RULE, NONE_RULE, NONE_RULE, NONE_RULE, NONE_RULE), "clock_shift_corrections": (NONE_RULE, NONE_RULE, NONE_RULE, NONE_RULE, NONE_RULE), "clock_employee_locks": (NONE_RULE, NONE_RULE, NONE_RULE, NONE_RULE, NONE_RULE), "FUEL_SHEETS": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "FUEL_LOGS": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "PLANT_HOURS_SHEETS": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "PLANT_HOURS_LOG": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "INCOME_CLAIMS": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "DELIVERY_NOTES": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), "DELIVERY_NOTE_ITEMS": (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE), } def http(method, path, body=None, token=None): url = f"{PB}{path}" data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method) if body is not None: req.add_header("Content-Type", "application/json") if token: req.add_header("Authorization", token) try: with urllib.request.urlopen(req) as r: txt = r.read().decode() return r.status, json.loads(txt) if txt else None except urllib.error.HTTPError as e: return e.code, json.loads(e.read().decode() or "{}") code, auth = http("POST", "/api/collections/_superusers/auth-with-password", {"identity": EMAIL, "password": PASSWORD}) if code != 200: sys.exit(f"AUTH FAILED: {code} {auth}") TOKEN = auth["token"] # fetch all collections to map name → id code, listing = http("GET", "/api/collections?perPage=100", token=TOKEN) if code != 200: sys.exit(f"could not list collections: {code}") by_name = {c["name"]: c["id"] for c in listing["items"]} for name, (l, v, c, u, d) in RULES.items(): cid = by_name.get(name) if not cid: print(f" ✗ {name:25s} not found, skipping") continue code, resp = http("PATCH", f"/api/collections/{cid}", {"listRule": l, "viewRule": v, "createRule": c, "updateRule": u, "deleteRule": d}, TOKEN) if code == 200: print(f" ✓ {name:25s} updated") else: print(f" ✗ {name:25s} FAILED: {code} {resp}") print("\nDone.")