breakpilot-compliance/backend-compliance/scripts/seed_demo_customer.py

#!/usr/bin/env python3
"""Seed ONE coherent demo customer so the whole CRA<->IACE frontend journey is
walkable: CompanyProfile (complete) + an IACE CE risk-assessment project
(Kistenhubgeraet, with components + hazards + mitigations) + a CRA project
(intake + scope-check) + a persisted CRA assessment snapshot driven by faked
repo findings + the machine's networked components + its safety functions.

Run inside the backend container (reaches both APIs):
  docker exec -i bp-compliance-backend python3 - < scripts/seed_demo_customer.py

Idempotency: not enforced — re-running creates duplicates. Resilient: each step
prints its outcome and continues on error. The source-repo layer is FAKED
(findings below) so the entire frontend can be seen once.
"""
import json
import urllib.request

TENANT = "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
BACKEND = "http://localhost:8002"
AISDK = "http://ai-compliance-sdk:8090"
H = {"Content-Type": "application/json", "X-Tenant-ID": TENANT, "X-Tenant-Id": TENANT}


def call(method, url, body=None):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, headers=H, method=method)
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            txt = r.read().decode()
            return r.getcode(), (json.loads(txt) if txt.strip() else {})
    except urllib.error.HTTPError as e:
        return e.code, {"error": e.read().decode()[:300]}
    except Exception as e:
        return 0, {"error": str(e)}


def step(label, method, url, body=None):
    code, resp = call(method, url, body)
    ok = 200 <= code < 300
    rid = resp.get("id") if isinstance(resp, dict) else None
    print(("  OK " if ok else "  !! ") + f"[{code}] {label}" + (f" id={rid}" if rid else "")
          + ("" if ok else f" {resp.get('error','')[:160]}"))
    return resp if ok else None


print("== 1. CompanyProfile ==")
step("company-profile", "POST", f"{BACKEND}/api/v1/company-profile", {
    "company_name": "DEMO Kistenhub GmbH", "legal_form": "GmbH", "industry": "Maschinenbau",
    "founded_year": 2015, "business_model": "B2B", "company_size": "medium", "employee_count": "50-99",
    "headquarters_country": "DE", "headquarters_city": "Augsburg", "is_data_controller": True,
    "is_complete": True,
})

print("== 2. IACE project (CE risk assessment) ==")
proj = step("iace project", "POST", f"{AISDK}/sdk/v1/iace/projects", {
    "machine_name": "Kistenhubgeraet Demo", "machine_type": "Hydraulischer Kistenhubwagen (vernetzt)",
    "manufacturer": "DEMO Kistenhub GmbH",
    "description": "Palettierte-Gueter-Hubgeraet mit IoT-Modul, App-Parametrierung und Fernwartung.",
    "narrative_text": "Zwei-Hub-System, hydraulisch, mit vernetzter Steuerung und Fernzugriff.",
    "ce_marking_target": "Maschinenverordnung (EU) 2023/1230",
})
pid = proj.get("id") if proj else None
if not pid:  # IACE create response may not surface "id" — fall back to the list
    _, lst = call("GET", f"{AISDK}/sdk/v1/iace/projects")
    for p in (lst.get("projects") if isinstance(lst, dict) else []) or []:
        if p.get("machine_name") == "Kistenhubgeraet Demo":
            pid = p.get("id")
            break
    print(f"  (fallback) iace project id={pid}")

comp_ids = {}
if pid:
    print("== 3. IACE components ==")
    for c in [
        {"name": "Hub-Steuerung (SPS)", "component_type": "controller", "is_safety_relevant": True, "is_networked": True},
        {"name": "Bedienpanel (HMI)", "component_type": "hmi", "is_networked": True},
        {"name": "Hydraulikpumpe", "component_type": "hydraulic", "is_safety_relevant": True},
        {"name": "Fernwartungs-Gateway", "component_type": "gateway", "is_networked": True},
    ]:
        c["project_id"] = pid
        r = step(f"component {c['name']}", "POST", f"{AISDK}/sdk/v1/iace/projects/{pid}/components", c)
        if r:
            comp_ids[c["name"]] = r.get("id")

    print("== 4. IACE hazards + mitigations ==")
    hazards = [
        {"name": "Quetschen zwischen Last und Rahmen", "category": "Mechanische Gefaehrdungen",
         "scenario": "Bediener greift bei Hubbewegung zwischen Last und Rahmen",
         "comp": "Hub-Steuerung (SPS)", "measure": "Zweihandschaltung + trennende Schutzeinrichtung (PL d)"},
        {"name": "Lastabsturz durch Ueberlast", "category": "Mechanische Gefaehrdungen",
         "scenario": "Ueberschreiten der zulaessigen Last fuehrt zum Absturz",
         "comp": "Hydraulikpumpe", "measure": "Ueberlastsicherung / Lastmomentbegrenzer"},
        {"name": "Schneiden an scharfen Kanten", "category": "Mechanische Gefaehrdungen",
         "scenario": "Kontakt mit scharfen Gehaeusekanten bei Wartung",
         "comp": "Bedienpanel (HMI)", "measure": "Brechen oder Runden aller zugaenglichen Kanten"},
    ]
    for hz in hazards:
        body = {"project_id": pid, "name": hz["name"], "category": hz["category"], "scenario": hz["scenario"]}
        cid = comp_ids.get(hz["comp"])
        if cid:
            body["component_id"] = cid
        r = step(f"hazard {hz['name']}", "POST", f"{AISDK}/sdk/v1/iace/projects/{pid}/hazards", body)
        hid = r.get("id") if r else None
        if hid:
            step("  mitigation", "POST", f"{AISDK}/sdk/v1/iace/projects/{pid}/hazards/{hid}/mitigations",
                 {"hazard_id": hid, "reduction_type": "design_measure", "name": hz["measure"]})

print("== 5. CRA project + intake + scope-check ==")
cra = step("cra project", "POST", f"{BACKEND}/api/v1/cra/projects",
           {"name": "DEMO Kistenhub CRA", "description": "Cyber-Risikobeurteilung fuer das Kistenhubgeraet"})
cra_id = cra.get("id") if cra else None
if cra_id:
    step("intake (flags)", "PATCH", f"{BACKEND}/api/v1/cra/projects/{cra_id}", {
        "intended_use": "Vernetztes Kistenhubgeraet mit App-Parametrierung und Fernwartung",
        "connected_to_internet": True, "has_software_updates": True, "has_firmware": True,
        "processes_personal_data": False, "is_critical_infra_supplier": False,
    })
    step("scope-check", "POST", f"{BACKEND}/api/v1/cra/projects/{cra_id}/scope-check")

    print("== 6. CRA assessment snapshot (faked repo findings + components + safety functions) ==")
    snap_body = {
        "findings": [
            {"id": "REPO-001", "title": "Default-Passwort in Fernsteuer-UI", "cwe": "CWE-259", "severity": "critical", "location": "remote-ui/login"},
            {"id": "REPO-002", "title": "Schwache TLS-Konfiguration", "cwe": "CWE-327", "severity": "high", "location": "telemetry/tls"},
            {"id": "REPO-003", "title": "Veraltete Abhaengigkeit libmodbus", "category": "dependency", "cwe": "CWE-1104", "severity": "high", "location": "deps/libmodbus"},
        ],
        "components": [
            {"name": "Hub-Steuerung (SPS)", "component_class": "controller", "networked": True},
            {"name": "Bedienpanel (HMI)", "component_class": "hmi", "networked": True},
            {"name": "Fernwartungs-Gateway", "component_class": "gateway", "networked": True},
        ],
        "safety_functions": [
            {"name": "Zweihandschaltung + trennende Schutzeinrichtung", "hazard": "Quetschen zwischen Last und Rahmen", "kind": "prevent_unexpected_actuation"},
            {"name": "Ueberlastsicherung / Lastmomentbegrenzer", "hazard": "Lastabsturz durch Ueberlast", "kind": "signal_integrity"},
        ],
        "weights": {"access": "high", "data": "high"},
    }
    step("assess-snapshot", "POST", f"{BACKEND}/api/v1/cra/projects/{cra_id}/assess-snapshot", snap_body)

print("\nDONE. tenant=%s | iace_project=%s | cra_project=%s" % (TENANT, pid, cra_id))