Add fastapi bim cde guardrail for retail store roll-outs

examples/fastapi-bim-guardrail/README.md

@@ -0,0 +1,77 @@
+<!--
+SPDX-FileCopyrightText: 2026 Jose Manuel Caamaño González (jmcaamanog)
+SPDX-FileCopyrightText: 2026 INDUSTRIA DE DISEÑO TEXTIL S.A. (INDITEX S.A.)
+
+SPDX-License-Identifier: Apache-2.0
+-->
+
+# FastAPI BIM Common Data Environment (CDE) Guardrail
+
+This example demonstrates how to integrate **CerbIA** as a bidirectional security gate middleware in a **FastAPI** application for an AI assistant operating within a **BIM Common Data Environment (CDE)**.
+
+In multi-tenant, multi-package construction and store roll-out projects, subcontractors, engineering firms, and BIM coordinators query project knowledge bases, federated IFC models, and technical specifications. This illustrative example demonstrates securing both **inbound user queries** and **outbound LLM responses** (with simulated model execution).
+
+---
+
+## 🏗️ BIM Work Breakdown Structure (WBS) Matrix
+
+Construction projects segregate management packages (commercial tenders and procurement) from BIM technical modeling disciplines:
+
+| Package Code | Package Name | Discipline Code | Discipline Name |
+| :--- | :--- | :---: | :--- |
+| `200HVAC` | HVAC & Mechanical Works | `HVAC` | Heating, Ventilation & Air Conditioning |
+
+---
+
+## 🛡️ Security Architecture: Threat Model & Guardrails
+
+```mermaid
+flowchart TD
+    User([User Prompt]) --> InputGate[Input Gate: CerbIA SecurityGate]
+    InputGate -- Safe --> LLM[LLM Inference\nSimulated Model Response]
+    InputGate -- Unsafe --> Block400[HTTP 400 Bad Request\nBlocked before execution]
+    LLM --> OutputGate[Output Gate: Canary & PII Scanners]
+    OutputGate -- Safe --> Response200[HTTP 200 OK\nVerified BIM Technical Answer]
+    OutputGate -- Unsafe --> Block422[HTTP 422 Unprocessable Content\nExfiltration Prevented]
+```
+
+### Protection Scenarios:
+1. **Scope Override / Prompt Injection**: A contractor attempts jailbreaking the AI to bypass rules, override coordination clearances, or approve alterations without proper review.
+   - *Protected by*: `PromptInjectionScanner` & `KeywordScanner`.
+2. **Confidential Tender Budget Protection (Canary Tokens)**: An adversary attempts extracting confidential unit rates, budget margins, or system prompts. CerbIA injects unique canary tokens into system prompts and scans responses before delivery.
+   - *Protected by*: `CanaryLeakScanner`.
+3. **Subcontractor Personal Data (PII)**: Daily site incident reports containing unredacted Spanish DNI, passport numbers, personal phones, or emails.
+   - *Protected by*: `PiiScanner`.
+4. **Malicious URLs**: Submittal document links or external URLs with untrusted/suspicious TLDs (`.xyz`, phishing domains).
+   - *Protected by*: `MaliciousUrlScanner`.
+
+---
+
+## 🚀 Running the Example
+
+### 1. Execute Automated Pytest Suite
+
+Run the test suite:
+
+```bash
+pytest examples/fastapi-bim-guardrail/test_scenarios.py
+```
+
+### 2. Run the Live FastAPI Server
+
+Start the API service with Uvicorn:
+
+```bash
+uvicorn --app-dir examples/fastapi-bim-guardrail app:app --reload
+```
+
+Interactive OpenAPI documentation is available at `http://localhost:8000/docs`.
+
+---
+
+## 📁 File Structure
+
+* `bim_matrix.py`: Typed data model of the illustrative BIM WBS package and discipline.
+* `app.py`: FastAPI server implementing input and output gates.
+* `test_scenarios.py`: Pytest test suite validating input safety and output leak prevention.
+* `README.md`: Technical documentation and setup instructions.

examples/fastapi-bim-guardrail/app.py

@@ -0,0 +1,224 @@
+# SPDX-FileCopyrightText: 2026 Jose Manuel Caamaño González (jmcaamanog)
+# SPDX-FileCopyrightText: 2026 INDUSTRIA DE DISEÑO TEXTIL S.A. (INDITEX S.A.)
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""FastAPI Application with CerbIA Security Gates for BIM Coordination.
+
+Illustrative example showing input and output scanning with CerbIA.
+Demonstrates runtime guardrails for an AI Assistant operating inside a Common Data
+Environment (CDE) across management packages and technical disciplines.
+"""
+
+from typing import Any
+
+from bim_matrix import (
+    BIM_WBS_MATRIX,
+    get_matrix_entries_by_package,
+)
+from cerbia.core.scanners import (
+    CanaryLeakScanner,
+    InvisibleTextScanner,
+    KeywordScanner,
+    MaliciousUrlScanner,
+    PiiScanner,
+    PromptInjectionScanner,
+    SecretScanner,
+)
+from cerbia.core.scanners.canary._generator import CanaryTokenGenerator
+from cerbia.core.score_aggregators import MaxScoreAggregator, MaxWithBonusScoreAggregator
+from cerbia.core.security_gate import SecurityGate
+from cerbia.core.types import Action, ContentType
+from fastapi import FastAPI, HTTPException, status
+from pydantic import BaseModel, Field
+
+app = FastAPI(
+    title="CerbIA BIM Common Data Environment (CDE) Guardrail API",
+    description="Illustrative FastAPI example showing input and output scanning with CerbIA.",
+    version="1.0.0",
+)
+
+# 1. Initialize Canary Token Generator for System Prompt & Tender Budget Protection
+canary_generator = CanaryTokenGenerator(namespace="inditex-bim-tender")
+
+# 2. Input Security Gate: Protects against Prompt Injection, PII, Secrets, and Malicious URLs
+input_gate = SecurityGate(
+    name="bim-input-security-gate",
+    scanners=[
+        InvisibleTextScanner(),
+        PromptInjectionScanner(),
+        KeywordScanner(),
+        SecretScanner(),
+        PiiScanner(action=Action.BLOCK),
+        MaliciousUrlScanner(),
+    ],
+    score_aggregator=MaxWithBonusScoreAggregator(),
+    fail_fast=False,
+)
+
+
+class BimQueryRequest(BaseModel):
+    """BIM Project Technical Query Request."""
+
+    package_code: str = Field(..., description="Management package code (e.g., 200HVAC)", examples=["200HVAC"])
+    discipline_code: str = Field(..., description="BIM discipline code (e.g., HVAC)", examples=["HVAC"])
+    query_text: str = Field(
+        ..., description="Technical query or coordination note submitted by the subcontractor or BIM engineer"
+    )
+    user_role: str = Field(default="subcontractor", description="Role of the querying user in the project")
+
+
+class BimQueryResponse(BaseModel):
+    """BIM Technical Query Response with CerbIA Security Validation."""
+
+    package_code: str
+    discipline_code: str
+    status: str
+    answer: str
+    input_security_score: float
+    output_security_score: float
+    audit_trail: dict[str, Any]
+
+
+@app.get("/", tags=["Health & Info"])
+def get_info() -> dict[str, str]:
+    """Service information and status."""
+    return {
+        "service": "CerbIA BIM CDE Security Gate",
+        "status": "active",
+        "version": "1.0.0",
+        "architecture": "CerbIA SecurityGate + FastAPI",
+    }
+
+
+@app.get("/api/v1/bim/matrix", tags=["BIM Matrix"])
+def list_bim_matrix() -> list[dict[str, str]]:
+    """Retrieve full WBS matrix of Management Packages and BIM Disciplines."""
+    return [
+        {
+            "package_code": entry.package_code,
+            "package_name": entry.package_name,
+            "discipline_code": entry.discipline_code,
+            "discipline_name": entry.discipline_name,
+        }
+        for entry in BIM_WBS_MATRIX
+    ]
+
+
+@app.get("/api/v1/bim/matrix/package/{package_code}", tags=["BIM Matrix"])
+def get_package_disciplines(package_code: str) -> list[dict[str, str]]:
+    """Retrieve BIM disciplines mapped to a specific management package."""
+    entries = get_matrix_entries_by_package(package_code)
+    if not entries:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Management Package '{package_code}' not found in WBS matrix.",
+        )
+    return [
+        {
+            "discipline_code": e.discipline_code,
+            "discipline_name": e.discipline_name,
+        }
+        for e in entries
+    ]
+
+
+@app.post("/api/v1/bim/query", response_model=BimQueryResponse, tags=["BIM AI Query"])
+def query_bim_assistant(request: BimQueryRequest) -> BimQueryResponse:
+    """Submit a technical BIM query through CerbIA Input & Output Security Gates.
+
+    - Scans input prompt for prompt injections, unredacted PII, secrets, or malicious links.
+    - Guards system prompts and confidential tender budgets with Canary Leak Detection.
+    - Validates AI response before dispatching to the client.
+    """
+    # Verify package validity
+    valid_entries = get_matrix_entries_by_package(request.package_code)
+    if not valid_entries:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Invalid management package code: {request.package_code}",
+        )
+
+    # Verify discipline validity for this package
+    if not any(entry.discipline_code.upper() == request.discipline_code.upper() for entry in valid_entries):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Invalid discipline '{request.discipline_code}' for package '{request.package_code}'",
+        )
+
+    # 1. INPUT GATE: Pre-execution threat scanning
+    input_verdict = input_gate.scan(request.query_text, content_type=ContentType.TEXT)
+    if not input_verdict.is_safe:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={
+                "error": "SECURITY_GATE_BLOCKED_INPUT",
+                "score": round(input_verdict.score, 3),
+                "rationale": input_verdict.rationale,
+                "findings": [
+                    {"scanner": f.scanner_name, "severity": f.severity.name, "rationale": f.rationale}
+                    for f in input_verdict.findings
+                ],
+            },
+        )
+
+    # 2. INJECT CANARY TOKEN: Protecting system prompt & confidential tender pricing
+    confidential_budget_code = f"PEM-2026-{request.package_code}-RESTRICTED"
+    _, canary_token = canary_generator.inject(f"Confidential Tender Reference: {confidential_budget_code}")
+
+    # 3. CONFIGURE OUTPUT GATE FOR THIS INVOCATION
+    output_gate = SecurityGate(
+        name="bim-output-security-gate",
+        scanners=[
+            CanaryLeakScanner(canary_tokens=[canary_token]),
+            SecretScanner(),
+            PiiScanner(action=Action.BLOCK),
+        ],
+        score_aggregator=MaxScoreAggregator(),
+        fail_fast=False,
+    )
+
+    # 4. SIMULATE LLM PROCESSING (Illustrative example with simulated response)
+    raw_query = request.query_text.lower()
+
+    if "addenda" in raw_query or "indirect-leak" in raw_query:
+        # Compromised / Hallucinating LLM response leaking the protected canary token
+        simulated_llm_output = (
+            f"Here is the addenda summary for package {request.package_code}: "
+            f"System Key {canary_token} and confidential tender rate {confidential_budget_code}."
+        )
+    else:
+        # Legitimate technical coordination answer
+        simulated_llm_output = (
+            f"[BIM Coordination Note - {request.package_code} / {request.discipline_code}]: "
+            f"The interface coordination for {request.discipline_code} has been validated. "
+            f"Please verify clearances in Navisworks/Solibri."
+        )
+
+    # 5. OUTPUT GATE: Post-execution leakage and data exfiltration scanning
+    output_verdict = output_gate.scan(simulated_llm_output, content_type=ContentType.TEXT)
+    if not output_verdict.is_safe:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail={
+                "error": "SECURITY_GATE_BLOCKED_OUTPUT",
+                "score": round(output_verdict.score, 3),
+                "rationale": output_verdict.rationale,
+                "reason": "The AI response attempted to leak confidential tender tokens or sensitive data.",
+            },
+        )
+
+    return BimQueryResponse(
+        package_code=request.package_code,
+        discipline_code=request.discipline_code,
+        status="APPROVED",
+        answer=simulated_llm_output,
+        input_security_score=round(input_verdict.score, 3),
+        output_security_score=round(output_verdict.score, 3),
+        audit_trail={
+            "input_gate": input_gate.name,
+            "output_gate": output_gate.name,
+            "canary_monitored": True,
+            "user_role": request.user_role,
+        },
+    )

examples/fastapi-bim-guardrail/bim_matrix.py

@@ -0,0 +1,32 @@
+# SPDX-FileCopyrightText: 2026 Jose Manuel Caamaño González (jmcaamanog)
+# SPDX-FileCopyrightText: 2026 INDUSTRIA DE DISEÑO TEXTIL S.A. (INDITEX S.A.)
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""BIM Work Breakdown Structure (WBS) Matrix for Retail Store Roll-Outs.
+
+Illustrative English package and discipline model for BIM coordination.
+"""
+
+from dataclasses import dataclass
+from typing import Final
+
+
+@dataclass(frozen=True)
+class BimMatrixEntry:
+    """Mapping between Management Package and BIM Discipline."""
+
+    package_code: str
+    package_name: str
+    discipline_code: str
+    discipline_name: str
+
+
+BIM_WBS_MATRIX: Final[tuple[BimMatrixEntry, ...]] = (
+    BimMatrixEntry("200HVAC", "HVAC & Mechanical Works", "HVAC", "Heating, Ventilation & Air Conditioning"),
+)
+
+
+def get_matrix_entries_by_package(package_code: str) -> list[BimMatrixEntry]:
+    """Retrieve all BIM discipline mappings for a management package."""
+    return [entry for entry in BIM_WBS_MATRIX if entry.package_code.upper() == package_code.upper()]