Files
breakpilot-compliance/backend-compliance/main.py
T
Benjamin Admin 0326d5baab feat(vendor-assessment): AVV/SCC/TOM/Sub-Processor checklists + assessment service
Phase 1-3 of the Vendor Contract Assessment:

Backend checklists (Doc-Check L1/L2 engine compatible):
- avv_checks.py: 28 checks (11 L1 + 17 L2) for Art. 28(3) DSGVO
- scc_checks.py: 7 checks for EU SCC 2021 (modules, annexes, TIA)
- tom_annex_checks.py: 12 checks for Art. 32 (8 control objectives)
- sub_processor_checks.py: 7 checks for sub-processor list completeness

Assessment service:
- POST /vendor-compliance/assessments — async contract analysis
- GET /vendor-compliance/assessments/{id} — poll status
- Cross-check engine: detects missing SCC when AVV mentions third-country,
  missing TOM annex, missing sub-processor list

All checklists registered in runner.py CHECKLIST_MAP (27 doc_types total).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-05-12 23:14:54 +02:00

172 lines
5.5 KiB
Python

"""
BreakPilot Compliance Backend
Extracted compliance-specific APIs from the monorepo backend.
Provides: Compliance Framework, Consent Admin, DSR, GDPR Export.
Runs on port 8002 with DB search_path=compliance,core,public.
"""
import logging
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
# Configure root logging so all modules' logger.info() etc. are visible
logging.basicConfig(
level=logging.INFO,
format="%(levelname)s:%(name)s: %(message)s",
)
logger = logging.getLogger(__name__)
# Compliance-specific API routers
from consent_api import router as consent_router
from consent_admin_api import router as consent_admin_router
from gdpr_api import router as gdpr_router, admin_router as gdpr_admin_router
# DSR proxy removed — now handled natively in compliance/api/dsr_routes.py
# Compliance framework sub-package
from compliance.api import router as compliance_framework_router
# Source Policy
from compliance.api.source_policy_router import router as source_policy_router
# Document Import & Screening
from compliance.api.import_routes import router as import_router
from compliance.api.screening_routes import router as screening_router
# Company Profile
from compliance.api.company_profile_routes import router as company_profile_router
# Agent (ZeroClaw compliance agent)
from compliance.api.agent_notification_routes import router as agent_notify_router
from compliance.api.agent_analyze_routes import router as agent_analyze_router
from compliance.api.agent_scan_routes import router as agent_scan_router
from compliance.api.agent_history_routes import router as agent_history_router
from compliance.api.agent_recurring_routes import router as agent_recurring_router
from compliance.api.agent_compare_routes import router as agent_compare_router
from compliance.api.agent_doc_check_routes import router as agent_doc_check_router
from compliance.api.agent_compliance_check_routes import router as agent_compliance_check_router
from compliance.api.vendor_assessment_routes import router as vendor_assessment_router
# Middleware
from middleware import (
RequestIDMiddleware,
SecurityHeadersMiddleware,
)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Run migrations on startup."""
from migration_runner import run_migrations
try:
run_migrations()
logger.info("Database migrations completed")
except Exception as e:
logger.error("Migration failed: %s — backend starting anyway", e)
yield
app = FastAPI(
title="BreakPilot Compliance Backend",
description="GDPR/DSGVO Compliance, Consent Management, Data Subject Requests, and Regulatory Compliance Framework",
version="1.0.0",
lifespan=lifespan,
)
# --- CORS ---
ALLOWED_ORIGINS = os.getenv("CORS_ORIGINS", "*").split(",")
app.add_middleware(
CORSMiddleware,
allow_origins=ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# --- Security & Tracing Middleware ---
app.add_middleware(RequestIDMiddleware)
app.add_middleware(SecurityHeadersMiddleware)
# --- Health Endpoint ---
@app.get("/health", tags=["system"])
async def health():
"""Health check endpoint for load balancers and orchestration."""
return {
"status": "healthy",
"service": "backend-compliance",
"version": "1.0.0",
}
@app.get("/debug/routers", tags=["system"])
async def debug_routers():
"""Diagnostic: show which sub-routers loaded and which failed."""
from compliance.api import _ROUTER_MODULES, _failed_routers, _loaded_count
return {
"total": len(_ROUTER_MODULES),
"loaded": _loaded_count,
"failed_count": len(_failed_routers),
"failed": _failed_routers,
}
# --- Compliance-specific Routers ---
# Consent (user-facing)
app.include_router(consent_router, prefix="/api")
# Consent Admin
app.include_router(consent_admin_router, prefix="/api")
# GDPR / Privacy (user-facing)
app.include_router(gdpr_router, prefix="/api")
# GDPR Admin
app.include_router(gdpr_admin_router, prefix="/api")
# DSR now handled natively via compliance_framework_router (dsr_routes.py)
# Compliance Framework (regulations, controls, evidence, risks, audits, ISMS)
app.include_router(compliance_framework_router, prefix="/api")
# Source Policy (allowed sources, PII rules, audit)
app.include_router(source_policy_router, prefix="/api")
# Document Import (PDF analysis, gap detection)
app.include_router(import_router, prefix="/api")
# System Screening (SBOM generation, vulnerability scan)
app.include_router(screening_router, prefix="/api")
# Company Profile (CRUD with audit logging)
app.include_router(company_profile_router, prefix="/api")
# Agent (ZeroClaw compliance agent → analyze + email via SMTP)
app.include_router(agent_notify_router, prefix="/api")
app.include_router(agent_analyze_router, prefix="/api")
app.include_router(agent_scan_router, prefix="/api")
app.include_router(agent_history_router, prefix="/api")
app.include_router(agent_recurring_router, prefix="/api")
app.include_router(agent_compare_router, prefix="/api")
app.include_router(agent_doc_check_router, prefix="/api")
app.include_router(agent_compliance_check_router, prefix="/api")
# Vendor Contract Assessment
app.include_router(vendor_assessment_router, prefix="/api")
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"main:app",
host="0.0.0.0",
port=int(os.getenv("PORT", "8002")),
reload=os.getenv("ENVIRONMENT", "development") == "development",
)