Compare commits

..

2 Commits

Author SHA1 Message Date
Benjamin Admin d4d9b60007 feat(email): P18 — Critical-Findings-Box + Banner-Deep-Block
CI / detect-changes (push) Successful in 12s
CI / branch-name (push) Has been skipped
CI / guardrail-integrity (push) Has been skipped
CI / secret-scan (push) Has been skipped
CI / dep-audit (push) Has been skipped
CI / sbom-scan (push) Has been skipped
CI / validate-canonical-controls (push) Successful in 15s
CI / loc-budget (push) Successful in 20s
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / nodejs-build (push) Successful in 3m8s
CI / test-go (push) Has been skipped
CI / iace-gt-coverage (push) Has been skipped
CI / test-python-backend (push) Successful in 47s
CI / test-python-document-crawler (push) Has been skipped
CI / test-python-dsms-gateway (push) Has been skipped
Backend wirft 90% der consent-tester-Daten weg — nur 4 Felder von einem
vollen Banner-Scan landeten im Email. Phases (before_consent / after_reject
/ after_accept), banner_checks.violations mit Rechtsgrundlagen,
category_tests, 46 structured_checks, completeness/correctness-Scores
waren alle nicht sichtbar.

Backend: agent_compliance_check_routes leitet jetzt das volle banner_result
durch (15 Felder statt 4).

Renderer (2 neue Module):
1) agent_doc_check_critical.build_critical_findings_html
   - ROTER Sofortmassnahmen-Block GANZ OBEN in der Email
   - Erkennt: banner-violations (HIGH/CRITICAL), leere Per-Category-Lists,
     DSE-Score <30%, fehlende Cookie-Richtlinie, US-Tracker ohne SCC/DPF
   - Pro Issue: konkrete Sofortmassnahme + Rechtsgrundlage + Bussgeld-
     Praezedenz (CNIL TikTok 5 Mio, LfDI BW 30k, EuGH Schrems II, ...)
   - Wird nur gerendert wenn echte Issues vorliegen

2) agent_doc_check_banner.build_banner_deep_html
   - Banner-Quality-Score-Cards (Vollstaendigkeit / Korrektheit / Verstoesse)
   - 3-Phasen-Cookie-Tabelle: vor Consent / nach Ablehnung / nach Annahme
     mit Cookie-Count, Tracker-Count, Auffaelligkeiten
   - Per-Category-Tracker-Listing (Statistik/Marketing) — zeigt explizit
     wenn eine Kategorie keine Provider listet (Safetykon-Pattern)
   - Violations-Liste mit Severity-Badge + Quellen-Hint (LG Rostock, EDPB)

Smoke-Test Safetykon: alle 6 neuen Blocks rendern, kein Regression.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-19 13:34:17 +02:00
Benjamin Admin e536247c20 feat(quaidal): backend API + frontend tab for BSI QUAIDAL data-quality controls
Wire the 195 Clean-Room QUAIDAL controls (from breakpilot-core migration 011)
into the compliance SaaS UI.

Backend:
- GET /api/v1/quaidal/stats           - counts by kind + source provenance
- GET /api/v1/quaidal/controls        - list, optional kind= filter
- GET /api/v1/quaidal/controls/{id}   - single derived control
- GET /api/v1/quaidal/criteria        - 10 QKB criteria
- GET /api/v1/quaidal/criteria/{id}   - QKB with QB/MA/QM tree

Frontend:
- /sdk/quality: new "Trainingsdaten-Qualität (BSI QUAIDAL)" tab with
  10 QKB cards and a drill-down modal showing the full QB→MA→QM tree
  plus original BSI source link and license note.
- /sdk/ai-act: Art. 10 tile on each high-risk/unacceptable result,
  linking to /sdk/quality?category=data_quality.

Pattern matches existing IACE module DIN-reference handling:
own wording, source section + URL preserved for due diligence.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-19 13:03:54 +02:00
16 changed files with 1266 additions and 24 deletions
@@ -0,0 +1,27 @@
import { NextRequest, NextResponse } from 'next/server'
const BACKEND_URL = process.env.BACKEND_URL || 'http://backend-compliance:8002'
function tenantHeader(request: NextRequest): string {
return request.headers.get('x-tenant-id') || '00000000-0000-0000-0000-000000000001'
}
export async function GET(
request: NextRequest,
{ params }: { params: Promise<{ derived_id: string }> }
) {
const { derived_id } = await params
try {
const resp = await fetch(
`${BACKEND_URL}/api/v1/quaidal/controls/${encodeURIComponent(derived_id)}`,
{ headers: { 'X-Tenant-ID': tenantHeader(request) }, cache: 'no-store' }
)
const body = await resp.text()
return new NextResponse(body, {
status: resp.status,
headers: { 'Content-Type': resp.headers.get('Content-Type') || 'application/json' },
})
} catch (err) {
return NextResponse.json({ error: 'Backend unreachable', details: String(err) }, { status: 502 })
}
}
@@ -0,0 +1,25 @@
import { NextRequest, NextResponse } from 'next/server'
const BACKEND_URL = process.env.BACKEND_URL || 'http://backend-compliance:8002'
function tenantHeader(request: NextRequest): string {
return request.headers.get('x-tenant-id') || '00000000-0000-0000-0000-000000000001'
}
export async function GET(request: NextRequest) {
const { searchParams } = new URL(request.url)
const qs = searchParams.toString()
try {
const resp = await fetch(
`${BACKEND_URL}/api/v1/quaidal/controls${qs ? `?${qs}` : ''}`,
{ headers: { 'X-Tenant-ID': tenantHeader(request) }, cache: 'no-store' }
)
const body = await resp.text()
return new NextResponse(body, {
status: resp.status,
headers: { 'Content-Type': resp.headers.get('Content-Type') || 'application/json' },
})
} catch (err) {
return NextResponse.json({ error: 'Backend unreachable', details: String(err) }, { status: 502 })
}
}
@@ -0,0 +1,27 @@
import { NextRequest, NextResponse } from 'next/server'
const BACKEND_URL = process.env.BACKEND_URL || 'http://backend-compliance:8002'
function tenantHeader(request: NextRequest): string {
return request.headers.get('x-tenant-id') || '00000000-0000-0000-0000-000000000001'
}
export async function GET(
request: NextRequest,
{ params }: { params: Promise<{ section_id: string }> }
) {
const { section_id } = await params
try {
const resp = await fetch(
`${BACKEND_URL}/api/v1/quaidal/criteria/${encodeURIComponent(section_id)}`,
{ headers: { 'X-Tenant-ID': tenantHeader(request) }, cache: 'no-store' }
)
const body = await resp.text()
return new NextResponse(body, {
status: resp.status,
headers: { 'Content-Type': resp.headers.get('Content-Type') || 'application/json' },
})
} catch (err) {
return NextResponse.json({ error: 'Backend unreachable', details: String(err) }, { status: 502 })
}
}
@@ -0,0 +1,23 @@
import { NextRequest, NextResponse } from 'next/server'
const BACKEND_URL = process.env.BACKEND_URL || 'http://backend-compliance:8002'
function tenantHeader(request: NextRequest): string {
return request.headers.get('x-tenant-id') || '00000000-0000-0000-0000-000000000001'
}
export async function GET(request: NextRequest) {
try {
const resp = await fetch(`${BACKEND_URL}/api/v1/quaidal/criteria`, {
headers: { 'X-Tenant-ID': tenantHeader(request) },
cache: 'no-store',
})
const body = await resp.text()
return new NextResponse(body, {
status: resp.status,
headers: { 'Content-Type': resp.headers.get('Content-Type') || 'application/json' },
})
} catch (err) {
return NextResponse.json({ error: 'Backend unreachable', details: String(err) }, { status: 502 })
}
}
@@ -0,0 +1,23 @@
import { NextRequest, NextResponse } from 'next/server'
const BACKEND_URL = process.env.BACKEND_URL || 'http://backend-compliance:8002'
function tenantHeader(request: NextRequest): string {
return request.headers.get('x-tenant-id') || '00000000-0000-0000-0000-000000000001'
}
export async function GET(request: NextRequest) {
try {
const resp = await fetch(`${BACKEND_URL}/api/v1/quaidal/stats`, {
headers: { 'X-Tenant-ID': tenantHeader(request) },
cache: 'no-store',
})
const body = await resp.text()
return new NextResponse(body, {
status: resp.status,
headers: { 'Content-Type': resp.headers.get('Content-Type') || 'application/json' },
})
} catch (err) {
return NextResponse.json({ error: 'Backend unreachable', details: String(err) }, { status: 502 })
}
}
@@ -0,0 +1,45 @@
'use client'
import Link from 'next/link'
interface Props {
/** Risk classification of the AI system. Tile is only rendered for high_risk / unacceptable. */
riskLevel: string
}
/**
* Renders a tile pointing to the BSI QUAIDAL-based data-quality control tab.
* AI Act Article 10 obligations (training-data quality) apply only to high-risk
* systems, so the tile is skipped for limited / minimal / not-applicable classes.
*/
export function Art10Tile({ riskLevel }: Props) {
if (riskLevel !== 'high_risk' && riskLevel !== 'unacceptable') return null
return (
<Link
href="/sdk/quality?category=data_quality"
className="block mt-3 p-3 rounded-lg border border-purple-200 bg-purple-50 hover:bg-purple-100 transition-colors"
>
<div className="flex items-start gap-3">
<div className="w-9 h-9 rounded-full bg-purple-200 text-purple-700 flex items-center justify-center shrink-0">
<svg className="w-5 h-5" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2}
d="M3 7v10a2 2 0 002 2h14a2 2 0 002-2V7M3 7l9 6 9-6M3 7l9-4 9 4" />
</svg>
</div>
<div className="flex-1 min-w-0">
<div className="text-sm font-semibold text-purple-900">
Art. 10 Datenqualität (Hochrisiko-KI)
</div>
<div className="text-xs text-purple-700 mt-0.5">
BSI QUAIDAL Controls: 10 Kriterien, 15 Bausteine, 30 Maßnahmen, 140 Metriken.
Klicken zum Öffnen des Trainingsdaten-Qualität-Moduls.
</div>
</div>
<svg className="w-4 h-4 text-purple-500 shrink-0 mt-1" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M9 5l7 7-7 7" />
</svg>
</div>
</Link>
)
}
+2
View File
@@ -9,6 +9,7 @@ import { RiskPyramid } from './_components/RiskPyramid'
import { AddSystemForm } from './_components/AddSystemForm'
import { AISystemCard } from './_components/AISystemCard'
import DecisionTreeWizard from '@/components/sdk/ai-act/DecisionTreeWizard'
import { Art10Tile } from './_components/Art10Tile'
type TabId = 'overview' | 'decision-tree' | 'results'
@@ -136,6 +137,7 @@ function SavedResultsTab() {
Löschen
</button>
</div>
<Art10Tile riskLevel={r.high_risk_result} />
</div>
))}
</div>
@@ -0,0 +1,152 @@
'use client'
import { useEffect, useState } from 'react'
import { fetchCriterionTree, type QuaidalControl, type QuaidalCriterionTree } from '../_hooks/useQuaidalData'
interface Props {
sectionId: string
onClose: () => void
}
function ControlBlock({ ctrl, badgeColor }: { ctrl: QuaidalControl; badgeColor: string }) {
return (
<div className="border border-gray-200 rounded-lg p-4 bg-white">
<div className="flex items-start justify-between gap-3 mb-2">
<h4 className="font-semibold text-gray-900">{ctrl.canonical_name}</h4>
<span className={`px-2 py-0.5 text-xs rounded-full ${badgeColor} shrink-0`}>{ctrl.source.section}</span>
</div>
<p className="text-sm text-gray-600 mb-3 whitespace-pre-line">{ctrl.description}</p>
{ctrl.source.url && (
<a
href={ctrl.source.url}
target="_blank"
rel="noreferrer noopener"
className="text-xs text-purple-600 hover:text-purple-800 underline"
>
BSI-Quelle ansehen ({ctrl.source.framework})
</a>
)}
</div>
)
}
export function QuaidalCriterionDetail({ sectionId, onClose }: Props) {
const [tree, setTree] = useState<QuaidalCriterionTree | null>(null)
const [loading, setLoading] = useState(true)
useEffect(() => {
let active = true
setLoading(true)
fetchCriterionTree(sectionId).then(t => {
if (active) {
setTree(t)
setLoading(false)
}
})
return () => { active = false }
}, [sectionId])
return (
<div className="fixed inset-0 z-50 flex items-center justify-center bg-black/40 p-4">
<div className="bg-white rounded-2xl shadow-xl w-full max-w-4xl max-h-[90vh] overflow-hidden flex flex-col">
<div className="flex items-center justify-between px-6 py-4 border-b border-gray-200">
<div>
<div className="text-xs text-gray-500 uppercase tracking-wide">QUAIDAL Kriterium</div>
<h2 className="text-xl font-bold text-gray-900">
{tree?.criterion.canonical_name || sectionId}
</h2>
</div>
<button
onClick={onClose}
className="w-8 h-8 rounded-full hover:bg-gray-100 flex items-center justify-center text-gray-500"
aria-label="Schliessen"
>×</button>
</div>
<div className="overflow-y-auto p-6 space-y-6">
{loading && <div className="text-center text-gray-400 py-12">Lade...</div>}
{tree && (
<>
<div>
<h3 className="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-2">
Anforderung (eigene Formulierung)
</h3>
<div className="bg-purple-50 border border-purple-200 rounded-lg p-4">
<p className="text-gray-800 whitespace-pre-line">{tree.criterion.description}</p>
</div>
<div className="mt-3 flex flex-wrap items-center gap-3 text-xs text-gray-500">
<span>Regulierung: <span className="font-medium text-gray-700">{tree.criterion.regulation_anchor || '—'}</span></span>
<span>Quelle: <span className="font-medium text-gray-700">{tree.criterion.source.framework} {tree.criterion.source.section}</span></span>
{tree.criterion.source.url && (
<a href={tree.criterion.source.url} target="_blank" rel="noreferrer noopener" className="text-purple-600 hover:text-purple-800 underline">
Originalquelle
</a>
)}
</div>
</div>
{tree.criterion.external_refs.length > 0 && (
<div>
<h3 className="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-2">
Externe Referenzen (nicht ingestiert, nur Verweis)
</h3>
<div className="flex flex-wrap gap-2">
{tree.criterion.external_refs.map((ref, i) => (
<span key={i} className="px-2 py-1 text-xs bg-gray-100 text-gray-700 rounded">
{ref.framework}{ref.citation ? `${ref.citation}` : ''}
</span>
))}
</div>
</div>
)}
{tree.building_blocks.length > 0 && (
<div>
<h3 className="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-3">
Bausteine ({tree.building_blocks.length})
</h3>
<div className="grid grid-cols-1 md:grid-cols-2 gap-3">
{tree.building_blocks.map(qb => (
<ControlBlock key={qb.derived_id} ctrl={qb} badgeColor="bg-blue-100 text-blue-700" />
))}
</div>
</div>
)}
{tree.measures.length > 0 && (
<div>
<h3 className="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-3">
Maßnahmen ({tree.measures.length})
</h3>
<div className="grid grid-cols-1 md:grid-cols-2 gap-3">
{tree.measures.map(m => (
<ControlBlock key={m.derived_id} ctrl={m} badgeColor="bg-green-100 text-green-700" />
))}
</div>
</div>
)}
{tree.metrics.length > 0 && (
<div>
<h3 className="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-3">
Metriken & Methoden ({tree.metrics.length})
</h3>
<div className="grid grid-cols-1 md:grid-cols-2 gap-3">
{tree.metrics.map(qm => (
<ControlBlock key={qm.derived_id} ctrl={qm} badgeColor="bg-amber-100 text-amber-700" />
))}
</div>
</div>
)}
</>
)}
</div>
<div className="px-6 py-3 border-t border-gray-200 bg-gray-50 text-xs text-gray-500">
Eigene Clean-Room-Ableitung von BSI QUAIDAL. Quellverweis und Lizenz-Note pro Eintrag.
</div>
</div>
</div>
)
}
@@ -0,0 +1,109 @@
'use client'
import { useState } from 'react'
import { useQuaidalData, type QuaidalControl } from '../_hooks/useQuaidalData'
import { QuaidalCriterionDetail } from './QuaidalCriterionDetail'
function CriterionCard({ ctrl, onOpen }: { ctrl: QuaidalControl; onOpen: () => void }) {
return (
<button
onClick={onOpen}
className="text-left bg-white rounded-xl border border-gray-200 p-5 hover:border-purple-400 hover:shadow-sm transition-all"
>
<div className="flex items-start justify-between mb-2">
<h3 className="font-semibold text-gray-900">{ctrl.canonical_name}</h3>
<span className="px-2 py-0.5 text-xs rounded-full bg-purple-100 text-purple-700">
{ctrl.source.section}
</span>
</div>
<p className="text-sm text-gray-600 line-clamp-3">{ctrl.description}</p>
<div className="mt-3 flex flex-wrap items-center gap-2 text-xs">
<span className="text-gray-500">Bausteine: <span className="font-medium text-gray-700">{ctrl.related_quaidal_ids.length}</span></span>
{ctrl.external_refs.slice(0, 2).map((r, i) => (
<span key={i} className="px-1.5 py-0.5 bg-gray-100 text-gray-600 rounded">
{r.framework}
</span>
))}
</div>
</button>
)
}
export function TrainingDataQualityTab() {
const { criteria, stats, loading, error } = useQuaidalData()
const [openSection, setOpenSection] = useState<string | null>(null)
if (loading) {
return <div className="text-center text-gray-400 py-12">Lade QUAIDAL-Katalog...</div>
}
if (error) {
return (
<div className="bg-red-50 border border-red-200 rounded-lg p-4 text-red-700">
QUAIDAL-Daten konnten nicht geladen werden: {error}
</div>
)
}
return (
<div className="space-y-6">
<div className="bg-purple-50 border border-purple-200 rounded-xl p-5">
<h2 className="text-lg font-semibold text-gray-900">Trainingsdaten-Qualität nach BSI QUAIDAL</h2>
<p className="text-sm text-gray-600 mt-1">
Operative Umsetzung von EU AI Act Art. 10 (Datenqualität für Hochrisiko-KI) auf Basis des
BSI-Katalogs QUAIDAL. Alle Controls sind eigenständig formuliert (Clean-Room) und verweisen
auf die jeweilige QUAIDAL-Sektion.
</p>
{stats && (
<div className="mt-4 grid grid-cols-2 md:grid-cols-4 gap-3 text-sm">
<div>
<div className="text-xs text-gray-500">Qualitätskriterien</div>
<div className="text-xl font-semibold text-gray-900">{stats.counts_by_kind.criterion ?? 0}</div>
</div>
<div>
<div className="text-xs text-gray-500">Bausteine</div>
<div className="text-xl font-semibold text-gray-900">{stats.counts_by_kind.building_block ?? 0}</div>
</div>
<div>
<div className="text-xs text-gray-500">Maßnahmen</div>
<div className="text-xl font-semibold text-gray-900">{stats.counts_by_kind.measure ?? 0}</div>
</div>
<div>
<div className="text-xs text-gray-500">Metriken & Methoden</div>
<div className="text-xl font-semibold text-gray-900">{stats.counts_by_kind.metric ?? 0}</div>
</div>
</div>
)}
</div>
<div>
<h3 className="text-lg font-semibold text-gray-900 mb-4">10 Qualitätskriterien</h3>
{criteria.length === 0 ? (
<div className="bg-white rounded-xl border border-gray-200 p-8 text-center text-gray-400">
Keine Kriterien gefunden. Bitte Backend-Ingest prüfen.
</div>
) : (
<div className="grid grid-cols-1 md:grid-cols-2 lg:grid-cols-3 gap-4">
{criteria.map(c => (
<CriterionCard
key={c.derived_id}
ctrl={c}
onOpen={() => setOpenSection(c.source.section)}
/>
))}
</div>
)}
</div>
{stats?.license_note && (
<div className="text-xs text-gray-500 italic">{stats.license_note}</div>
)}
{openSection && (
<QuaidalCriterionDetail
sectionId={openSection}
onClose={() => setOpenSection(null)}
/>
)}
</div>
)
}
@@ -0,0 +1,86 @@
'use client'
import { useCallback, useEffect, useState } from 'react'
export interface QuaidalExternalRef {
framework: string
citation: string | null
}
export interface QuaidalSource {
framework: string
section: string
url: string | null
commit_sha: string | null
title_original: string | null
license_note: string | null
}
export interface QuaidalControl {
derived_id: string
kind: 'criterion' | 'building_block' | 'measure' | 'metric'
canonical_name: string
description: string
regulation_anchor: string | null
related_quaidal_ids: string[]
external_refs: QuaidalExternalRef[]
source: QuaidalSource
plagiarism_score: number | null
}
export interface QuaidalStats {
counts_by_kind: Record<string, number>
source_framework: string
source_commit_sha: string | null
license_note: string | null
}
export interface QuaidalCriterionTree {
criterion: QuaidalControl
building_blocks: QuaidalControl[]
measures: QuaidalControl[]
metrics: QuaidalControl[]
}
const API_BASE = '/api/sdk/v1/quaidal'
export function useQuaidalData() {
const [criteria, setCriteria] = useState<QuaidalControl[]>([])
const [stats, setStats] = useState<QuaidalStats | null>(null)
const [loading, setLoading] = useState(true)
const [error, setError] = useState<string | null>(null)
const loadAll = useCallback(async () => {
setLoading(true)
setError(null)
try {
const [criteriaRes, statsRes] = await Promise.all([
fetch(`${API_BASE}/criteria`, { cache: 'no-store' }),
fetch(`${API_BASE}/stats`, { cache: 'no-store' }),
])
if (criteriaRes.ok) {
const data = (await criteriaRes.json()) as QuaidalControl[]
setCriteria(Array.isArray(data) ? data : [])
} else {
setError(`Criteria endpoint returned ${criteriaRes.status}`)
}
if (statsRes.ok) {
setStats(await statsRes.json())
}
} catch (err) {
setError(String(err))
} finally {
setLoading(false)
}
}, [])
useEffect(() => { loadAll() }, [loadAll])
return { criteria, stats, loading, error, reload: loadAll }
}
export async function fetchCriterionTree(sectionId: string): Promise<QuaidalCriterionTree | null> {
const res = await fetch(`${API_BASE}/criteria/${encodeURIComponent(sectionId)}`, { cache: 'no-store' })
if (!res.ok) return null
return (await res.json()) as QuaidalCriterionTree
}
+56 -16
View File
@@ -1,15 +1,23 @@
'use client'
import { useState, useEffect } from 'react'
import { useSearchParams } from 'next/navigation'
import { useSDK } from '@/lib/sdk'
import { useQualityData } from './_hooks/useQualityData'
import { MetricCard, type QualityMetric } from './_components/MetricCard'
import { TestRow } from './_components/TestRow'
import { MetricModal } from './_components/MetricModal'
import { TestModal } from './_components/TestModal'
import { TrainingDataQualityTab } from './_components/TrainingDataQualityTab'
type TabId = 'model_quality' | 'data_quality'
export default function QualityPage() {
const { state } = useSDK()
const searchParams = useSearchParams()
const initialTab: TabId = searchParams?.get('category') === 'data_quality' ? 'data_quality' : 'model_quality'
const [tab, setTab] = useState<TabId>(initialTab)
const {
metrics,
tests,
@@ -41,24 +49,54 @@ export default function QualityPage() {
<h1 className="text-2xl font-bold text-gray-900">AI Quality Dashboard</h1>
<p className="mt-1 text-gray-500">Ueberwachen Sie die Qualitaet und Fairness Ihrer KI-Systeme</p>
</div>
<div className="flex items-center gap-2">
<button
onClick={() => setShowTestModal(true)}
className="flex items-center gap-2 px-4 py-2 border border-purple-300 text-purple-700 rounded-lg hover:bg-purple-50 transition-colors"
>
<svg className="w-5 h-5" fill="none" stroke="currentColor" viewBox="0 0 24 24"><path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M4 4v5h.582m15.356 2A8.001 8.001 0 004.582 9m0 0H9m11 11v-5h-.581m0 0a8.003 8.003 0 01-15.357-2m15.357 2H15" /></svg>
Test hinzufuegen
</button>
<button
onClick={() => { setEditMetric(undefined); setShowMetricModal(true) }}
className="flex items-center gap-2 px-4 py-2 bg-purple-600 text-white rounded-lg hover:bg-purple-700 transition-colors"
>
<svg className="w-5 h-5" fill="none" stroke="currentColor" viewBox="0 0 24 24"><path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M12 6v6m0 0v6m0-6h6m-6 0H6" /></svg>
Messung hinzufuegen
</button>
</div>
{tab === 'model_quality' && (
<div className="flex items-center gap-2">
<button
onClick={() => setShowTestModal(true)}
className="flex items-center gap-2 px-4 py-2 border border-purple-300 text-purple-700 rounded-lg hover:bg-purple-50 transition-colors"
>
<svg className="w-5 h-5" fill="none" stroke="currentColor" viewBox="0 0 24 24"><path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M4 4v5h.582m15.356 2A8.001 8.001 0 004.582 9m0 0H9m11 11v-5h-.581m0 0a8.003 8.003 0 01-15.357-2m15.357 2H15" /></svg>
Test hinzufuegen
</button>
<button
onClick={() => { setEditMetric(undefined); setShowMetricModal(true) }}
className="flex items-center gap-2 px-4 py-2 bg-purple-600 text-white rounded-lg hover:bg-purple-700 transition-colors"
>
<svg className="w-5 h-5" fill="none" stroke="currentColor" viewBox="0 0 24 24"><path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M12 6v6m0 0v6m0-6h6m-6 0H6" /></svg>
Messung hinzufuegen
</button>
</div>
)}
</div>
<div className="border-b border-gray-200">
<nav className="-mb-px flex gap-6">
<button
onClick={() => setTab('model_quality')}
className={`pb-3 px-1 text-sm font-medium border-b-2 transition-colors ${
tab === 'model_quality'
? 'border-purple-500 text-purple-600'
: 'border-transparent text-gray-500 hover:text-gray-700'
}`}
>
Modell-Qualität
</button>
<button
onClick={() => setTab('data_quality')}
className={`pb-3 px-1 text-sm font-medium border-b-2 transition-colors ${
tab === 'data_quality'
? 'border-purple-500 text-purple-600'
: 'border-transparent text-gray-500 hover:text-gray-700'
}`}
>
Trainingsdaten-Qualität (BSI QUAIDAL)
</button>
</nav>
</div>
{tab === 'data_quality' && <TrainingDataQualityTab />}
{tab === 'model_quality' && (
<>
<div className="grid grid-cols-1 md:grid-cols-4 gap-4">
<div className="bg-white rounded-xl border border-gray-200 p-6">
<div className="text-sm text-gray-500">Durchschnittlicher Score</div>
@@ -141,6 +179,8 @@ export default function QualityPage() {
</div>
</div>
</div>
</>
)}
{showMetricModal && (
<MetricModal
@@ -614,6 +614,9 @@ async def _run_compliance_check(check_id: str, req: ComplianceCheckRequest):
summary_html = build_management_summary(results)
scanned_html = build_scanned_urls_html(doc_entries)
providers_html = build_provider_list_html(banner_result, vvt_entries)
# P18: Deep-Block mit Phases + Quality-Score + Per-Category-Tracker
from .agent_doc_check_banner import build_banner_deep_html
banner_deep_html = build_banner_deep_html(banner_result)
vvt_html = build_vvt_table_html(cmp_vendors)
# MC scorecard aggregated across ALL docs in this run (DSGVO/TDDDG/
@@ -676,6 +679,20 @@ async def _run_compliance_check(check_id: str, req: ComplianceCheckRequest):
site_name=site_name_for_exec,
)
# P18: Critical-Findings-Block (rot oben, mit Sofortmassnahmen +
# Quellen + Bussgeld-Praezedenz). Wird nur gerendert wenn echte
# kritische Verstoesse vorliegen.
critical_html = ""
try:
from .agent_doc_check_critical import build_critical_findings_html
critical_html = build_critical_findings_html(
banner_result=banner_result,
scorecard=scorecard,
results=results,
)
except Exception as e:
logger.warning("Critical-findings block skipped: %s", e)
# P10: Cookie-Policy-Architecture-Detection (BMW-Pattern erkennen)
cookie_arch_html = ""
try:
@@ -726,10 +743,10 @@ async def _run_compliance_check(check_id: str, req: ComplianceCheckRequest):
# 7) providers_html + vvt_html (Vendor-Liste)
# 8) report_html (Doc-Pruefung Details)
full_html = (
exec_summary_html + cookie_arch_html + summary_html
+ scanned_html + profile_html
critical_html + exec_summary_html + cookie_arch_html
+ summary_html + scanned_html + profile_html
+ scorecard_html + redundancy_html
+ providers_html + vvt_html + report_html
+ providers_html + banner_deep_html + vvt_html + report_html
)
# Step 6: Send email — derive site name primarily from entered URL.
@@ -753,12 +770,23 @@ async def _run_compliance_check(check_id: str, req: ComplianceCheckRequest):
"results": [_result_to_dict(r) for r in results],
"business_profile": profile_dict,
"extracted_profile": extracted_profile,
"banner_result": {
"detected": banner_result.get("banner_detected", False) if banner_result else False,
"provider": banner_result.get("banner_provider", "") if banner_result else "",
"violations": len(banner_result.get("banner_checks", {}).get("violations", [])) if banner_result else 0,
# P18: vollen consent-tester-Output durchreichen statt nur 4 Felder.
# phases (before/after-accept/reject) + banner_checks.violations +
# category_tests werden vom Renderer + Critical-Findings-Block genutzt.
"banner_result": ({
"detected": banner_result.get("banner_detected", False),
"provider": banner_result.get("banner_provider", ""),
"violations": len((banner_result.get("banner_checks") or {})
.get("violations", [])),
"tcf_vendor_count": len(tcf_vendors),
} if banner_result else None,
"completeness_pct": banner_result.get("completeness_pct"),
"correctness_pct": banner_result.get("correctness_pct"),
"phases": banner_result.get("phases", {}),
"banner_checks": banner_result.get("banner_checks", {}),
"category_tests": banner_result.get("category_tests", []),
"structured_checks": banner_result.get("structured_checks", []),
"summary": banner_result.get("summary", {}),
} if banner_result else None),
"tcf_vendors": vvt_entries if tcf_vendors else [],
"cmp_vendors": cmp_vendors,
"total_documents": len(results),
@@ -0,0 +1,201 @@
"""
P18 Erweiterter Banner-Block fuer die Email.
Rendert die Daten aus dem consent-tester die heute weggeworfen wurden:
- 3-Phasen-Cookie-Tabelle (before_consent / after_reject / after_accept)
- Banner-Quality-Score (completeness/correctness/violations)
- Per-Category-Tracker-Listing
- Violations-Liste mit Rechtsgrundlagen
"""
from __future__ import annotations
def _color_for(pct: int) -> str:
return ("#16a34a" if pct >= 80 else
"#d97706" if pct >= 50 else "#dc2626")
def _short_phase_label(key: str) -> str:
return {
"before_consent": "Vor Consent",
"after_reject": "Nach Ablehnung",
"after_accept": "Nach Annahme",
}.get(key, key)
def _phase_color(key: str, cookie_count: int) -> str:
if key == "before_consent":
return "#16a34a" if cookie_count == 0 else "#dc2626"
if key == "after_reject":
return "#16a34a" if cookie_count <= 1 else "#d97706"
return "#94a3b8"
def build_banner_deep_html(banner_result: dict | None) -> str:
"""Render: Banner-Quality + Phases + Violations.
Konsumiert das volle consent-tester-Response. Komplementiert
`build_provider_list_html` (das nur Summary + TCF-Vendor-Tabelle macht).
"""
if not banner_result:
return ""
parts: list[str] = [
'<div style="font-family:-apple-system,BlinkMacSystemFont,sans-serif;'
'max-width:700px;margin:0 auto 16px;padding:14px 18px;'
'background:#fff;border:1px solid #cbd5e1;border-radius:8px">'
'<h3 style="margin:0 0 12px;font-size:14px;color:#0f172a">'
'Cookie-Banner — technische Analyse</h3>'
]
# 1) Quality-Score-Cards
compl = banner_result.get("completeness_pct")
corr = banner_result.get("correctness_pct")
summary = banner_result.get("summary") or {}
n_critical = summary.get("critical", 0)
n_high = summary.get("high", 0)
if compl is not None or corr is not None:
parts.append(
'<table style="width:100%;border-collapse:separate;'
'border-spacing:6px;margin-bottom:10px"><tr>'
)
if compl is not None:
c = _color_for(int(compl))
parts.append(
f'<td style="width:33%;padding:8px 10px;background:#f8fafc;'
f'border-radius:5px;border-left:3px solid {c}">'
f'<div style="font-size:10px;color:#64748b;text-transform:uppercase">'
f'Vollstaendigkeit</div>'
f'<div style="font-size:18px;font-weight:700;color:{c}">{compl}%</div>'
f'</td>'
)
if corr is not None:
c = _color_for(int(corr))
parts.append(
f'<td style="width:33%;padding:8px 10px;background:#f8fafc;'
f'border-radius:5px;border-left:3px solid {c}">'
f'<div style="font-size:10px;color:#64748b;text-transform:uppercase">'
f'Korrektheit</div>'
f'<div style="font-size:18px;font-weight:700;color:{c}">{corr}%</div>'
f'</td>'
)
viol_c = ("#dc2626" if n_critical + n_high > 0 else
"#d97706" if (summary.get("total_violations") or 0) > 0 else
"#16a34a")
parts.append(
f'<td style="width:33%;padding:8px 10px;background:#f8fafc;'
f'border-radius:5px;border-left:3px solid {viol_c}">'
f'<div style="font-size:10px;color:#64748b;text-transform:uppercase">'
f'Verstoesse</div>'
f'<div style="font-size:18px;font-weight:700;color:{viol_c}">'
f'{summary.get("total_violations", 0)}'
f'<span style="font-size:11px;color:#64748b;margin-left:6px">'
f'(crit:{n_critical} high:{n_high})</span></div></td>'
)
parts.append('</tr></table>')
# 2) 3-Phasen-Tabelle
phases = banner_result.get("phases") or {}
if phases:
parts.append(
'<div style="font-size:11px;color:#475569;margin:8px 0 4px;'
'font-weight:600">Cookie-Setzungen pro Phase '
'(echter Browser-Test):</div>'
'<table style="width:100%;border-collapse:collapse;font-size:11px;'
'margin-bottom:10px;border:1px solid #e2e8f0">'
'<thead><tr style="background:#f1f5f9;color:#475569;text-align:left">'
'<th style="padding:5px 8px">Phase</th>'
'<th style="padding:5px 8px;text-align:center">Cookies</th>'
'<th style="padding:5px 8px;text-align:center">Tracker</th>'
'<th style="padding:5px 8px">Auffaelligkeiten</th>'
'</tr></thead><tbody>'
)
for key in ("before_consent", "after_reject", "after_accept"):
ph = phases.get(key) or {}
if not isinstance(ph, dict): continue
cookies = ph.get("cookies") or []
trackers = ph.get("tracking_services") or []
new_track = ph.get("new_tracking") or []
violations = ph.get("violations") or []
undoc = ph.get("undocumented") or []
color = _phase_color(key, len(cookies))
issues_parts = []
if violations: issues_parts.append(f"{len(violations)} Verstoss")
if new_track: issues_parts.append(f"{len(new_track)} neue Tracker")
if undoc: issues_parts.append(f"{len(undoc)} undokumentiert")
issues_str = ", ".join(issues_parts) or ""
parts.append(
f'<tr style="border-top:1px solid #e2e8f0">'
f'<td style="padding:5px 8px;color:#1e293b;font-weight:600">'
f'<span style="display:inline-block;width:6px;height:6px;'
f'border-radius:50%;background:{color};margin-right:6px"></span>'
f'{_short_phase_label(key)}</td>'
f'<td style="padding:5px 8px;text-align:center;color:{color};'
f'font-weight:600">{len(cookies)}</td>'
f'<td style="padding:5px 8px;text-align:center">{len(trackers)}</td>'
f'<td style="padding:5px 8px;color:#475569">{issues_str}</td>'
f'</tr>'
)
parts.append('</tbody></table>')
# 3) Per-Category-Tracker
cats = banner_result.get("category_tests") or []
if cats:
non_essential = [c for c in cats if c.get("category") != "necessary"]
if non_essential:
parts.append(
'<div style="font-size:11px;color:#475569;margin:8px 0 4px;'
'font-weight:600">Provider-Listing pro Banner-Kategorie:</div>'
'<table style="width:100%;border-collapse:collapse;font-size:11px;'
'margin-bottom:10px;border:1px solid #e2e8f0">'
'<thead><tr style="background:#f1f5f9;color:#475569;text-align:left">'
'<th style="padding:5px 8px">Kategorie</th>'
'<th style="padding:5px 8px;text-align:center">Anbieter</th>'
'<th style="padding:5px 8px">Hinweis</th>'
'</tr></thead><tbody>'
)
for c in non_essential:
n = len(c.get("tracking_services") or [])
label = c.get("category_label") or c.get("category", "?")
if n == 0:
color = "#dc2626"
hint = ("Keine Anbieter sichtbar — Nutzer kann nicht "
"informiert einwilligen (Art. 7 DSGVO)")
else:
color = "#16a34a"
hint = ""
parts.append(
f'<tr style="border-top:1px solid #e2e8f0">'
f'<td style="padding:5px 8px">{label}</td>'
f'<td style="padding:5px 8px;text-align:center;color:{color};'
f'font-weight:600">{n}</td>'
f'<td style="padding:5px 8px;color:#dc2626;font-size:10px">'
f'{hint}</td></tr>'
)
parts.append('</tbody></table>')
# 4) Violations mit Rechtsgrundlage
violations = (banner_result.get("banner_checks") or {}).get("violations", [])
if violations:
parts.append(
'<div style="font-size:11px;color:#475569;margin:8px 0 4px;'
'font-weight:600">Erkannte Banner-Verstoesse:</div>'
'<ul style="margin:0 0 8px 18px;padding:0;font-size:11px;color:#1e293b">'
)
for v in violations[:8]:
sev = (v.get("severity") or "MEDIUM").upper()
sev_c = ("#dc2626" if sev in ("CRITICAL", "HIGH") else
"#d97706" if sev == "MEDIUM" else "#94a3b8")
parts.append(
f'<li style="margin-bottom:6px">'
f'<span style="display:inline-block;background:{sev_c};color:#fff;'
f'font-size:9px;padding:1px 5px;border-radius:3px;margin-right:6px">'
f'{sev}</span>{v.get("text", "")[:200]}'
f'<div style="font-size:10px;color:#94a3b8;margin-top:2px;'
f'font-style:italic">Quelle: {v.get("legal_ref", "")}</div></li>'
)
parts.append('</ul>')
parts.append('</div>')
return "".join(parts)
@@ -0,0 +1,208 @@
"""
P18 Critical-Findings-Block fuer die Executive-Summary.
Analysiert die echten Daten (banner_checks, phases, scorecard, results) und
rendert einen ROTEN Sofortmassnahmen-Block GANZ OBEN in der Email mit
Quellenangaben (DSK, EDPB, EuGH, Behoerden-Buessgeld-Faelle) und konkreten
Sofortmassnahmen.
Regel: Block wird nur gerendert wenn echte kritische Verstoesse vorliegen.
Bei sauberen Sites bleibt er weg.
"""
from __future__ import annotations
# Bekannte Buessgeld-Praezedenzfaelle als Quellen-Hint
_BUSSGELD_REFS = {
"no_provider_per_category": "CNIL France 2023 — TikTok 5 Mio EUR (fehlende Vendor-Transparenz)",
"dse_unvollstaendig": "BayLDA 2024 — diverse Mittelstand-Faelle, 5k50k EUR",
"cookie_doc_missing": "LfDI BW 2023 — fehlende Cookie-Erklaerung, 30k EUR",
"dark_pattern_reject": "EDPB Guidelines 3/2022 + DSK 2024 — Bussgeldrahmen Art. 83 DSGVO",
"schrems_ii": "EuGH C-311/18 (Schrems II) — Bussgeldrahmen bis 4% Konzern-Umsatz",
"impressum_im_banner": "LG Rostock 3 O 22/19 — Impressum-Pflicht ueberlagernder Banner",
}
def _detect_critical_issues(
banner_result: dict | None,
scorecard: dict | None,
results: list,
) -> list[dict]:
"""Erkenne kritische Verstoesse aus den vorliegenden Daten."""
issues: list[dict] = []
br = banner_result or {}
sc = scorecard or {}
# 1) Banner-Violations (HIGH/CRITICAL) aus consent-tester
for v in (br.get("banner_checks") or {}).get("violations", []):
sev = (v.get("severity") or "").upper()
if sev in ("CRITICAL", "HIGH"):
issues.append({
"key": "banner_violation",
"title": v.get("text", "")[:120],
"severity": sev,
"action": _action_for_banner_violation(v),
"source": v.get("legal_ref", ""),
"bussgeld": _BUSSGELD_REFS.get("impressum_im_banner")
if "impressum" in (v.get("text") or "").lower()
else _BUSSGELD_REFS.get("dark_pattern_reject"),
})
# 2) Category-Tests: leere Vendor-Liste pro Kategorie (Safetykon-Pattern)
cat_tests = br.get("category_tests") or []
empty_cats = [c for c in cat_tests
if not c.get("tracking_services")
and c.get("category") != "necessary"]
if empty_cats and len(cat_tests) >= 2:
cats = ", ".join(c.get("category_label", c.get("category", "?"))
for c in empty_cats)
issues.append({
"key": "no_provider_per_category",
"title": f"Cookie-Banner Kategorien ({cats}) ohne Provider-Listing",
"severity": "HIGH",
"action": ("Pro Banner-Kategorie eine Liste der eingebundenen "
"Anbieter + Cookie-Details (Name, Zweck, Speicherdauer, "
"Drittlandtransfer) ergaenzen. Sonst ist die "
"Einwilligung nicht 'informiert' nach Art. 7 DSGVO."),
"source": "Art. 7 DSGVO, EDPB Guidelines 2/2023, DSK 2024",
"bussgeld": _BUSSGELD_REFS["no_provider_per_category"],
})
# 3) DSGVO/TDDDG-Score < 30%: DSE rechtswidrig
pct = int((sc.get("totals") or {}).get("pct", 100))
if pct and pct < 30:
issues.append({
"key": "dse_unvollstaendig",
"title": f"Datenschutzerklaerung erfuellt nur {pct}% der Pflichten",
"severity": "HIGH",
"action": ("Vollstaendig nach Art. 13 DSGVO ueberarbeiten: "
"Verantwortlicher, Zwecke, Rechtsgrundlage, "
"Speicherdauer, Drittland-Transfers, alle Betroffenen-"
"rechte, konkrete Aufsichtsbehoerde."),
"source": "Art. 13 DSGVO + Art. 14 (alternativ), DSK-OH Telemedien 2024",
"bussgeld": _BUSSGELD_REFS["dse_unvollstaendig"],
})
# 4) Cookie-Richtlinie fehlt komplett (nicht erreichbar)
cookie_missing = any(
(r.doc_type == "cookie" if hasattr(r, "doc_type") else
r.get("doc_type") == "cookie")
and ((r.error if hasattr(r, "error") else r.get("error", "")) or "")
.startswith("Auf der Website nicht gefunden")
for r in (results or [])
)
cookie_deduped = any(
(r.doc_type == "cookie" if hasattr(r, "doc_type") else
r.get("doc_type") == "cookie")
and "Nicht separat vorhanden" in
((r.error if hasattr(r, "error") else r.get("error", "")) or "")
for r in (results or [])
)
if cookie_missing or cookie_deduped:
issues.append({
"key": "cookie_doc_missing",
"title": ("Keine eigenstaendige Cookie-Richtlinie"
if cookie_deduped
else "Cookie-Richtlinie nicht auffindbar"),
"severity": "HIGH",
"action": ("Separate Cookie-Richtlinie-Seite erstellen mit "
"tabellarischer Auflistung aller Cookies (Name, "
"Anbieter, Zweck, Speicherdauer, Drittlandtransfer). "
"Direkt aus dem Banner verlinken."),
"source": "Art. 13 DSGVO, §25 TDDDG, DSK-OH Telemedien 2024",
"bussgeld": _BUSSGELD_REFS["cookie_doc_missing"],
})
# 5) Schrems-II-Risiko: Google/Meta/Microsoft im Banner, aber keine SCC/DPF
# Detection: pre-/post-consent-cookies in den phases enthalten US-Tracker
phases = br.get("phases") or {}
has_us_tracker = False
for ph in phases.values():
if not isinstance(ph, dict):
continue
for t in (ph.get("tracking_services") or []):
if isinstance(t, dict):
name = (t.get("name", "") or "").lower()
else:
name = str(t).lower()
if any(w in name for w in ("google", "meta", "facebook",
"microsoft", "linkedin", "tiktok")):
has_us_tracker = True
break
if has_us_tracker:
issues.append({
"key": "schrems_ii",
"title": "US-Tracker geladen — Schrems-II-Risiko",
"severity": "HIGH",
"action": ("Pro Drittland-Anbieter dokumentieren: SCC (Art. 46 "
"DSGVO) ODER DPF-Zertifizierung pruefen + in der "
"Datenschutzerklaerung explizit benennen."),
"source": "Art. 44 ff. DSGVO, EuGH C-311/18 (Schrems II)",
"bussgeld": _BUSSGELD_REFS["schrems_ii"],
})
return issues
def _action_for_banner_violation(v: dict) -> str:
text = (v.get("text") or "").lower()
if "impressum" in text:
return ("Impressum-Link direkt im Banner ergaenzen — bei "
"ueberlagerndem Banner Pflicht nach §5 TMG.")
if "ablehnen" in text or "dark pattern" in text:
return ("'Ablehnen'-Button visuell gleichwertig zu 'Akzeptieren' "
"gestalten (gleiche Groesse, Farbe, Position).")
if "widerruf" in text or "cookie-einstellungen" in text:
return ("Floating-Icon oder Footer-Link 'Cookie-Einstellungen' "
"permanent einblenden — Widerruf so einfach wie Erteilung.")
return ("Banner-Verstoss beheben gemaess der genannten Rechtsgrundlage.")
def build_critical_findings_html(
banner_result: dict | None,
scorecard: dict | None,
results: list,
) -> str:
"""Render der Critical-Findings-Box. Leerer String wenn keine Issues."""
issues = _detect_critical_issues(banner_result, scorecard, results)
if not issues:
return ""
items = []
for i in issues:
items.append(
f'<div style="margin-bottom:10px;padding:10px 12px;'
f'background:rgba(255,255,255,0.06);border-radius:4px;'
f'border-left:3px solid #fca5a5">'
f'<div style="font-size:13px;font-weight:700;color:#fff;'
f'margin-bottom:4px">'
f'<span style="display:inline-block;background:#dc2626;color:#fff;'
f'padding:1px 6px;border-radius:3px;font-size:9px;'
f'margin-right:6px">{i["severity"]}</span>{i["title"]}</div>'
f'<div style="font-size:11px;color:#fecaca;margin-top:4px">'
f'<strong>Sofortmassnahme:</strong> {i["action"]}</div>'
f'<div style="font-size:10px;color:#fca5a5;margin-top:4px;'
f'font-style:italic">Rechtsgrundlage: {i.get("source","")}'
+ (f' &middot; Praezedenz: {i["bussgeld"]}'
if i.get("bussgeld") else "") +
f'</div></div>'
)
n = len(issues)
return (
'<div style="font-family:-apple-system,BlinkMacSystemFont,sans-serif;'
'max-width:700px;margin:0 auto 18px;padding:18px 22px;'
'background:#7f1d1d;border-radius:10px;color:white">'
'<div style="font-size:12px;color:#fecaca;text-transform:uppercase;'
'letter-spacing:1.5px;margin-bottom:6px;font-weight:700">'
'🚨 Sofortmassnahmen erforderlich</div>'
f'<h2 style="margin:0 0 10px;font-size:18px;color:white">'
f'{n} kritische Compliance-Risiken mit Bussgeldpotenzial</h2>'
'<p style="margin:0 0 12px;font-size:12px;color:#fecaca">'
'Die folgenden Verstoesse sind durch Tool-Analyse belegt und '
'erfordern Sofortmassnahmen. Bussgeldrahmen nach Art. 83 DSGVO: '
'<strong>bis 4% des weltweiten Jahresumsatzes</strong>.</p>'
+ "".join(items) +
'</div>'
)
@@ -0,0 +1,244 @@
"""FastAPI routes for QUAIDAL-derived Controls (AI Trainingsdaten-Qualität).
Endpoints:
- GET /v1/quaidal/stats - Counts by kind + source provenance
- GET /v1/quaidal/controls - List all controls, optional kind= filter
- GET /v1/quaidal/controls/{id} - Single derived control by derived_id
- GET /v1/quaidal/criteria - The 10 QKB criteria with linked QB/MA IDs
- GET /v1/quaidal/criteria/{id} - Single QKB with full child tree (QB MA QM)
The controls are Clean-Room derived from BSI QUAIDAL. See
control-pipeline/scripts/derive_quaidal_mcs.py and migration 011.
"""
from __future__ import annotations
import logging
from typing import Optional
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import text
from database import SessionLocal
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/v1/quaidal", tags=["quaidal"])
# ---------------------------------------------------------------------------
# Response shapes
# ---------------------------------------------------------------------------
class ExternalRef(BaseModel):
framework: str
citation: Optional[str] = None
class SourceProvenance(BaseModel):
framework: str
section: str
url: Optional[str] = None
commit_sha: Optional[str] = None
title_original: Optional[str] = None
license_note: Optional[str] = None
class DerivedControl(BaseModel):
derived_id: str
kind: str
canonical_name: str
description: str
regulation_anchor: Optional[str] = None
related_quaidal_ids: list[str]
external_refs: list[ExternalRef]
source: SourceProvenance
plagiarism_score: Optional[float] = None
class ControlsListResponse(BaseModel):
total: int
controls: list[DerivedControl]
class CriterionWithChildren(BaseModel):
"""A QKB criterion with the IDs of its linked building blocks, measures and metrics."""
criterion: DerivedControl
building_blocks: list[DerivedControl]
measures: list[DerivedControl]
metrics: list[DerivedControl]
class StatsResponse(BaseModel):
counts_by_kind: dict[str, int]
source_framework: str
source_commit_sha: Optional[str]
license_note: Optional[str]
# ---------------------------------------------------------------------------
# DB helpers
# ---------------------------------------------------------------------------
def _row_to_control(row) -> DerivedControl:
return DerivedControl(
derived_id=row.derived_id,
kind=row.kind,
canonical_name=row.canonical_name,
description=row.description,
regulation_anchor=row.regulation_anchor,
related_quaidal_ids=row.related_quaidal_ids or [],
external_refs=[ExternalRef(**r) for r in (row.external_refs or [])],
source=SourceProvenance(
framework=row.source_framework,
section=row.source_section,
url=row.source_url,
commit_sha=row.source_commit_sha,
title_original=row.source_title_original,
license_note=row.source_license_note,
),
plagiarism_score=float(row.plagiarism_score_at_generation) if row.plagiarism_score_at_generation is not None else None,
)
_SELECT_COLUMNS = """
derived_id, kind, canonical_name, description, regulation_anchor,
related_quaidal_ids, external_refs,
source_framework, source_section, source_url, source_commit_sha,
source_title_original, source_license_note,
plagiarism_score_at_generation
"""
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@router.get("/stats", response_model=StatsResponse)
def get_stats() -> StatsResponse:
"""Counts by kind + the QUAIDAL source provenance (single source today)."""
with SessionLocal() as db:
counts = db.execute(text(
"SELECT kind, COUNT(*) AS n FROM compliance.derived_controls "
"WHERE source_framework = :fw GROUP BY kind"
), {"fw": "BSI QUAIDAL"}).all()
meta = db.execute(text(
"SELECT source_commit_sha, source_license_note FROM compliance.derived_controls "
"WHERE source_framework = :fw LIMIT 1"
), {"fw": "BSI QUAIDAL"}).first()
return StatsResponse(
counts_by_kind={r.kind: r.n for r in counts},
source_framework="BSI QUAIDAL",
source_commit_sha=meta.source_commit_sha if meta else None,
license_note=meta.source_license_note if meta else None,
)
@router.get("/controls", response_model=ControlsListResponse)
def list_controls(
kind: Optional[str] = Query(None, description="criterion | building_block | measure | metric"),
limit: int = Query(500, ge=1, le=2000),
offset: int = Query(0, ge=0),
) -> ControlsListResponse:
"""List QUAIDAL-derived controls, optionally filtered by kind."""
where = ["source_framework = :fw"]
params: dict = {"fw": "BSI QUAIDAL", "limit": limit, "offset": offset}
if kind:
where.append("kind = :kind")
params["kind"] = kind
sql = (
f"SELECT {_SELECT_COLUMNS} FROM compliance.derived_controls "
f"WHERE {' AND '.join(where)} "
"ORDER BY source_section LIMIT :limit OFFSET :offset"
)
count_sql = f"SELECT COUNT(*) FROM compliance.derived_controls WHERE {' AND '.join(where)}"
with SessionLocal() as db:
rows = db.execute(text(sql), params).all()
total = db.execute(text(count_sql), {k: v for k, v in params.items() if k not in ("limit", "offset")}).scalar() or 0
return ControlsListResponse(total=int(total), controls=[_row_to_control(r) for r in rows])
@router.get("/controls/{derived_id}", response_model=DerivedControl)
def get_control(derived_id: str) -> DerivedControl:
with SessionLocal() as db:
row = db.execute(text(
f"SELECT {_SELECT_COLUMNS} FROM compliance.derived_controls WHERE derived_id = :id"
), {"id": derived_id}).first()
if not row:
raise HTTPException(status_code=404, detail=f"Control {derived_id} not found")
return _row_to_control(row)
@router.get("/criteria", response_model=list[DerivedControl])
def list_criteria() -> list[DerivedControl]:
"""Returns the 10 QKB criteria. Use /criteria/{section_id} for the full child tree."""
with SessionLocal() as db:
rows = db.execute(text(
f"SELECT {_SELECT_COLUMNS} FROM compliance.derived_controls "
"WHERE source_framework = :fw AND kind = 'criterion' ORDER BY source_section"
), {"fw": "BSI QUAIDAL"}).all()
return [_row_to_control(r) for r in rows]
@router.get("/criteria/{section_id}", response_model=CriterionWithChildren)
def get_criterion_tree(section_id: str) -> CriterionWithChildren:
"""Single QKB with the building blocks it references and the measures/metrics those reference.
`section_id` is the canonical QUAIDAL ID, e.g. `QKB-01`.
"""
section_id_upper = section_id.upper()
with SessionLocal() as db:
criterion_row = db.execute(text(
f"SELECT {_SELECT_COLUMNS} FROM compliance.derived_controls "
"WHERE source_framework = :fw AND source_section = :sid AND kind = 'criterion'"
), {"fw": "BSI QUAIDAL", "sid": section_id_upper}).first()
if not criterion_row:
raise HTTPException(status_code=404, detail=f"Criterion {section_id_upper} not found")
building_block_ids = criterion_row.related_quaidal_ids or []
building_blocks = []
if building_block_ids:
qb_rows = db.execute(text(
f"SELECT {_SELECT_COLUMNS} FROM compliance.derived_controls "
"WHERE source_framework = :fw AND kind = 'building_block' "
"AND source_section = ANY(:ids) ORDER BY source_section"
), {"fw": "BSI QUAIDAL", "ids": building_block_ids}).all()
building_blocks = [_row_to_control(r) for r in qb_rows]
# Collect measure IDs from each building block, then fetch them
measure_ids: list[str] = []
for qb in building_blocks:
measure_ids.extend(mid for mid in qb.related_quaidal_ids if mid.startswith("MA-"))
measures = []
if measure_ids:
ma_rows = db.execute(text(
f"SELECT {_SELECT_COLUMNS} FROM compliance.derived_controls "
"WHERE source_framework = :fw AND kind = 'measure' "
"AND source_section = ANY(:ids) ORDER BY source_section"
), {"fw": "BSI QUAIDAL", "ids": list(set(measure_ids))}).all()
measures = [_row_to_control(r) for r in ma_rows]
# Collect metric IDs from each measure
metric_ids: list[str] = []
for ma in measures:
metric_ids.extend(mid for mid in ma.related_quaidal_ids if mid.startswith("QM-"))
metrics = []
if metric_ids:
qm_rows = db.execute(text(
f"SELECT {_SELECT_COLUMNS} FROM compliance.derived_controls "
"WHERE source_framework = :fw AND kind = 'metric' "
"AND source_section = ANY(:ids) ORDER BY source_section"
), {"fw": "BSI QUAIDAL", "ids": list(set(metric_ids))}).all()
metrics = [_row_to_control(r) for r in qm_rows]
return CriterionWithChildren(
criterion=_row_to_control(criterion_row),
building_blocks=building_blocks,
measures=measures,
metrics=metrics,
)
+2
View File
@@ -55,6 +55,7 @@ from compliance.api.saving_scan_routes import router as saving_scan_router
from compliance.api.agent_migration_routes import router as agent_migration_router
from compliance.api.vendor_assessment_routes import router as vendor_assessment_router
from compliance.api.cra_routes import router as cra_router
from compliance.api.quaidal_routes import router as quaidal_router
# Middleware
from middleware import (
@@ -168,6 +169,7 @@ app.include_router(vendor_assessment_router, prefix="/api")
# CRA (Cyber Resilience Act) Compliance
app.include_router(cra_router, prefix="/api")
app.include_router(quaidal_router, prefix="/api")
if __name__ == "__main__":