not
400bdcafc4
verify-domain already drives a full approval; it now also asserts Objecten holds exactly one RegisterRecord for that registration — matched on its own reference, because the shared verify stack carries records from earlier runs. The check covers the three things that can silently go wrong: the record is missing (the ACL's Objecten hop never ran), duplicated (the upsert is not idempotent), or carries a field outside the public-safe schema.
89 lines
3.5 KiB
Python
89 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""S-19a (#149): prove the approval path wrote the register record to Objecten.
|
|
|
|
Given the registration whose Beoordelen task the caller just completed with `goedkeuren`, assert
|
|
that Objecten holds exactly one RegisterRecord object for it, with status INGESCHREVEN and the
|
|
registration's reference — i.e. the ACL's Objecten hop ran, the record validates against the
|
|
objecttype schema (Objecten rejects a mismatch), and it carries no personal data (ADR-0027/0028).
|
|
|
|
Stdlib only so it runs in a bare python:3-slim container on the compose network.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
OBJECTTYPEN = os.environ["OBJECTTYPEN"] # http://<ip>:8000
|
|
OBJECTTYPEN_TOKEN = os.environ["OBJECTTYPEN_TOKEN"]
|
|
OBJECTEN = os.environ["OBJECTEN"] # http://<ip>:8000
|
|
OBJECTEN_TOKEN = os.environ["OBJECTEN_TOKEN"]
|
|
REFERENCE = os.environ["REGISTRATION_REFERENCE"]
|
|
TIMEOUT = int(os.environ.get("REGISTER_RECORD_TIMEOUT", "60"))
|
|
NAME = "RegisterRecord"
|
|
# The register is world-readable: a record must never carry anything identifying (ADR-0027).
|
|
ALLOWED_FIELDS = {"id", "status", "reference"}
|
|
|
|
|
|
def get(base, token, path, crs=False):
|
|
headers = {"Authorization": f"Token {token}"}
|
|
if crs:
|
|
headers["Accept-Crs"] = "EPSG:4326"
|
|
req = urllib.request.Request(f"{base}{path}", headers=headers)
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
return json.load(r)
|
|
|
|
|
|
def objecttype_url():
|
|
"""The RegisterRecord objecttype URL, or None while registerrecord-init has yet to run."""
|
|
ots = get(OBJECTTYPEN, OBJECTTYPEN_TOKEN, "/api/v2/objecttypes").get("results", [])
|
|
match = next((o for o in ots if o.get("name") == NAME), None)
|
|
return match["url"] if match else None
|
|
|
|
|
|
def check():
|
|
"""Return (ok, detail). Raises on transport errors so the caller can retry."""
|
|
type_url = objecttype_url()
|
|
if not type_url:
|
|
return False, f"no objecttype named {NAME!r} in Objecttypen yet"
|
|
|
|
query = urllib.parse.urlencode({"type": type_url, "data_attrs": f"reference__exact__{REFERENCE}"})
|
|
results = get(OBJECTEN, OBJECTEN_TOKEN, f"/api/v2/objects?{query}", crs=True).get("results", [])
|
|
if not results:
|
|
return False, f"no RegisterRecord object with reference {REFERENCE}"
|
|
if len(results) > 1:
|
|
# The ACL upserts, so a replayed approval must update rather than duplicate (§8.6).
|
|
return False, f"{len(results)} RegisterRecord objects for reference {REFERENCE} — the write is not idempotent"
|
|
|
|
data = (results[0].get("record") or {}).get("data") or {}
|
|
if data.get("status") != "INGESCHREVEN":
|
|
return False, f"record status is {data.get('status')!r}, expected 'INGESCHREVEN'"
|
|
if not data.get("id"):
|
|
return False, "record carries no id (the zaak the projection keys on)"
|
|
extra = set(data) - ALLOWED_FIELDS
|
|
if extra:
|
|
return False, f"record leaks non-public fields: {sorted(extra)}"
|
|
return True, f"id={data['id']} status={data['status']} reference={data['reference']}"
|
|
|
|
|
|
def main():
|
|
deadline = time.time() + TIMEOUT
|
|
detail = "no attempt"
|
|
while time.time() < deadline:
|
|
try:
|
|
ok, detail = check()
|
|
if ok:
|
|
print(f"OK — approval wrote the register record to Objecten: {detail}")
|
|
return 0
|
|
except (urllib.error.URLError, ConnectionError, TimeoutError) as e:
|
|
detail = f"transport: {e}"
|
|
time.sleep(3)
|
|
print(f"FAIL — {detail}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|