MEGAINFRA-WEBAPP/scripts/pb-relax-rules.py

#!/usr/bin/env python3
"""Relax API rules on the 17 MEGA INFRASTRUCTURE collections.

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 sys
import urllib.request
import urllib.error

PB = "https://pb.mega.infra.apphub.co.za"
EMAIL = "admin@mega.co.za"
PASSWORD = "12345678"

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":            (AUTH_RULE, AUTH_RULE, AUTH_RULE, NONE_RULE, NONE_RULE),
    "clock_missed_out_corrections": (NONE_RULE, NONE_RULE, NONE_RULE, NONE_RULE, NONE_RULE),
    "clock_correction_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.")