Compare commits

...

2 Commits

Author SHA1 Message Date
Benjamin Admin c3db56ddb6 feat: Live cookie banner overlay in SDK — auto-open + FAB reopen button
Build + Deploy / build-admin-compliance (push) Successful in 2m16s
Build + Deploy / build-backend-compliance (push) Failing after 4m47s
Build + Deploy / build-ai-sdk (push) Successful in 51s
Build + Deploy / build-developer-portal (push) Successful in 1m17s
Build + Deploy / build-tts (push) Successful in 2m30s
Build + Deploy / build-document-crawler (push) Successful in 45s
Build + Deploy / build-dsms-gateway (push) Successful in 29s
Build + Deploy / build-dsms-node (push) Successful in 11s
Build + Deploy / trigger-orca (push) Has been skipped
CI / branch-name (push) Has been skipped
CI / guardrail-integrity (push) Has been skipped
CI / loc-budget (push) Failing after 28s
CI / secret-scan (push) Has been skipped
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 2m56s
CI / dep-audit (push) Has been skipped
CI / sbom-scan (push) Has been skipped
CI / test-go (push) Successful in 53s
CI / test-python-backend (push) Successful in 43s
CI / test-python-document-crawler (push) Successful in 33s
CI / test-python-dsms-gateway (push) Successful in 26s
CI / validate-canonical-controls (push) Successful in 19s
- CookieBannerOverlay: opens automatically on first visit (localStorage check)
- CookieBannerFAB: shield icon button at right-[10rem] to reopen settings
- 3 consent modes: accept all, reject all (nur notwendige), custom settings
- 4 categories: Notwendig (locked on), Statistik, Marketing, Funktional
- Category toggles with descriptions in settings view
- Datenschutzerklaerung + Impressum links in banner
- Consent persisted to localStorage, custom event fired on change
- Comprehensive Playwright E2E tests (16 tests):
  - First visit auto-open, button visibility, category toggles
  - Accept all / reject all / custom settings persistence
  - FAB reopen behavior, disabled toggle for necessary category

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-05-02 19:55:13 +02:00
Benjamin Admin 44acd68c96 feat: Cookie-Banner ↔ Backend Integration (DSR, Retention, Consent Proof)
Phase 1: Vendor sync from service registry (82+ services → banner vendors)
Phase 2: Category-based retention (marketing=90d, statistics=790d, not hardcoded 365d)
Phase 3: DSR ↔ Banner email linking (link-email, by-email, Art.17 erasure, Art.15/20 export)
Phase 4: Consent sync (Banner → Einwilligungen bridge)
Phase 6: Consent proof (SHA256 config hash + config_version in audit log, Art. 7(1) DSGVO)

New files:
- banner_dsr_service.py — email linking + DSR integration
- vendor_banner_sync.py — service registry → vendor configs
- migration 106 — linked_email, banner_config_hash, consent_version columns

Tests: 20+ new backend tests + 2 Playwright E2E test suites (API + UI)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-05-02 19:55:13 +02:00
15 changed files with 2063 additions and 8 deletions
+8 -3
View File
@@ -7,6 +7,7 @@ import { SDKSidebar } from '@/components/sdk/Sidebar/SDKSidebar'
import { CommandBar } from '@/components/sdk/CommandBar'
import { SDKPipelineSidebar } from '@/components/sdk/SDKPipelineSidebar'
import { ComplianceAdvisorWidget } from '@/components/sdk/ComplianceAdvisorWidget'
import { CookieBannerOverlay, CookieBannerFAB } from '@/components/sdk/CookieBannerOverlay'
import { useSDK } from '@/lib/sdk'
// =============================================================================
@@ -208,10 +209,14 @@ function SDKInnerLayout({ children }: { children: React.ReactNode }) {
{isCommandBarOpen && <CommandBar onClose={() => setCommandBarOpen(false)} />}
{/* Pipeline Sidebar (FAB on mobile/tablet, fixed on desktop xl+) */}
{projectId && <SDKPipelineSidebar />}
<SDKPipelineSidebar />
{/* Compliance Advisor Widget */}
{projectId && <ComplianceAdvisorWidget currentStep={currentStep} />}
{/* Compliance Advisor Widget — immer sichtbar, auch ohne Projekt */}
<ComplianceAdvisorWidget currentStep={currentStep} />
{/* Cookie Banner — opens on first visit, reopenable via FAB */}
<CookieBannerOverlay />
<CookieBannerFAB />
</div>
)
}
@@ -0,0 +1,277 @@
'use client'
import { useState, useEffect, useCallback } from 'react'
/**
* CookieBannerOverlay — Live cookie consent banner for the Compliance SDK.
*
* - Opens automatically on first visit (localStorage check)
* - Can be reopened via FAB button (right-[10rem])
* - Records consent choice to localStorage
* - Fires custom event 'sdkCookieConsentUpdated' for other components
*/
const STORAGE_KEY = 'bp-sdk-cookie-consent'
interface ConsentState {
necessary: boolean
statistics: boolean
marketing: boolean
functional: boolean
timestamp: string
}
function getStoredConsent(): ConsentState | null {
if (typeof window === 'undefined') return null
try {
const raw = localStorage.getItem(STORAGE_KEY)
if (!raw) return null
return JSON.parse(raw)
} catch {
return null
}
}
export function CookieBannerOverlay() {
const [isOpen, setIsOpen] = useState(false)
const [showSettings, setShowSettings] = useState(false)
const [consent, setConsent] = useState<ConsentState>({
necessary: true,
statistics: false,
marketing: false,
functional: false,
timestamp: '',
})
// Check on mount if consent was already given
useEffect(() => {
const stored = getStoredConsent()
if (!stored) {
// First visit — show banner
setIsOpen(true)
} else {
setConsent(stored)
}
}, [])
// Listen for reopen event from FAB button
useEffect(() => {
const handler = () => {
setIsOpen(true)
setShowSettings(true)
}
window.addEventListener('openCookieBanner', handler)
return () => window.removeEventListener('openCookieBanner', handler)
}, [])
const saveConsent = useCallback((state: ConsentState) => {
const withTimestamp = { ...state, timestamp: new Date().toISOString() }
localStorage.setItem(STORAGE_KEY, JSON.stringify(withTimestamp))
setConsent(withTimestamp)
setIsOpen(false)
setShowSettings(false)
window.dispatchEvent(new CustomEvent('sdkCookieConsentUpdated', { detail: withTimestamp }))
}, [])
const handleAcceptAll = () => {
saveConsent({ necessary: true, statistics: true, marketing: true, functional: true, timestamp: '' })
}
const handleRejectAll = () => {
saveConsent({ necessary: true, statistics: false, marketing: false, functional: false, timestamp: '' })
}
const handleSaveSettings = () => {
saveConsent(consent)
}
if (!isOpen) return null
return (
<>
{/* Overlay */}
<div
className="fixed inset-0 bg-black/40 z-[9998] transition-opacity duration-300"
onClick={() => {/* Don't close on overlay click — consent is required */}}
/>
{/* Banner */}
<div className="fixed bottom-0 left-0 right-0 z-[9999] animate-in slide-in-from-bottom duration-300">
<div className="max-w-3xl mx-auto m-4 bg-white rounded-2xl shadow-2xl border border-gray-200 overflow-hidden">
{/* Header */}
<div className="px-6 pt-6 pb-4">
<h2 className="text-lg font-semibold text-gray-900 flex items-center gap-2">
<svg className="w-5 h-5 text-purple-600" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M9 12l2 2 4-4m5.618-4.016A11.955 11.955 0 0112 2.944a11.955 11.955 0 01-8.618 3.04A12.02 12.02 0 003 9c0 5.591 3.824 10.29 9 11.622 5.176-1.332 9-6.03 9-11.622 0-1.042-.133-2.052-.382-3.016z" />
</svg>
Cookie-Einstellungen
</h2>
<p className="text-sm text-gray-600 mt-2 leading-relaxed">
Wir verwenden Cookies und aehnliche Technologien, um Ihnen die bestmoegliche Erfahrung zu bieten.
Sie koennen Ihre Praeferenzen jederzeit aendern.
</p>
<div className="flex items-center gap-4 mt-2 text-xs text-gray-500">
<a href="/sdk/einwilligungen/cookie-banner" className="hover:text-purple-600 underline">
Datenschutzerklaerung
</a>
<span>|</span>
<a href="/sdk/einwilligungen" className="hover:text-purple-600 underline">
Impressum
</a>
</div>
</div>
{/* Category Settings (expandable) */}
{showSettings && (
<div className="px-6 pb-4 space-y-3 border-t border-gray-100 pt-4">
{/* Necessary — always on */}
<CategoryToggle
label="Notwendig"
description="Fuer die Grundfunktionen der Website erforderlich."
checked={true}
disabled={true}
onChange={() => {}}
/>
<CategoryToggle
label="Statistik"
description="Helfen uns zu verstehen, wie Besucher mit der Website interagieren."
checked={consent.statistics}
onChange={(v) => setConsent(prev => ({ ...prev, statistics: v }))}
/>
<CategoryToggle
label="Marketing"
description="Werden verwendet, um Besuchern relevante Werbung zu zeigen."
checked={consent.marketing}
onChange={(v) => setConsent(prev => ({ ...prev, marketing: v }))}
/>
<CategoryToggle
label="Funktional"
description="Ermoeglichen erweiterte Funktionen und Personalisierung."
checked={consent.functional}
onChange={(v) => setConsent(prev => ({ ...prev, functional: v }))}
/>
</div>
)}
{/* Buttons */}
<div className="px-6 py-4 bg-gray-50 border-t border-gray-100 flex flex-wrap items-center gap-3">
{!showSettings ? (
<>
<button
onClick={handleAcceptAll}
className="flex-1 min-w-[140px] px-4 py-2.5 bg-purple-600 text-white rounded-lg font-medium hover:bg-purple-700 transition-colors text-sm"
>
Alle akzeptieren
</button>
<button
onClick={handleRejectAll}
className="flex-1 min-w-[140px] px-4 py-2.5 bg-gray-200 text-gray-700 rounded-lg font-medium hover:bg-gray-300 transition-colors text-sm"
>
Nur notwendige
</button>
<button
onClick={() => setShowSettings(true)}
className="flex-1 min-w-[140px] px-4 py-2.5 border border-gray-300 text-gray-700 rounded-lg font-medium hover:bg-gray-100 transition-colors text-sm"
>
Einstellungen
</button>
</>
) : (
<>
<button
onClick={handleSaveSettings}
className="flex-1 min-w-[140px] px-4 py-2.5 bg-purple-600 text-white rounded-lg font-medium hover:bg-purple-700 transition-colors text-sm"
>
Auswahl speichern
</button>
<button
onClick={handleAcceptAll}
className="flex-1 min-w-[140px] px-4 py-2.5 bg-gray-200 text-gray-700 rounded-lg font-medium hover:bg-gray-300 transition-colors text-sm"
>
Alle akzeptieren
</button>
<button
onClick={() => setShowSettings(false)}
className="px-4 py-2.5 text-gray-500 hover:text-gray-700 text-sm"
>
Zurueck
</button>
</>
)}
</div>
</div>
</div>
</>
)
}
/**
* FAB button to reopen the cookie banner settings.
* Positioned next to ComplianceAdvisor and PipelineSidebar.
*/
export function CookieBannerFAB() {
const [hasConsent, setHasConsent] = useState(false)
useEffect(() => {
setHasConsent(!!getStoredConsent())
const handler = () => setHasConsent(true)
window.addEventListener('sdkCookieConsentUpdated', handler)
return () => window.removeEventListener('sdkCookieConsentUpdated', handler)
}, [])
// Only show FAB after consent was given (banner is closed)
if (!hasConsent) return null
return (
<button
onClick={() => window.dispatchEvent(new Event('openCookieBanner'))}
className="fixed bottom-6 right-[10rem] w-14 h-14 bg-gray-700 hover:bg-gray-800 text-white rounded-full shadow-lg flex items-center justify-center transition-all duration-200 hover:scale-110 z-50"
aria-label="Cookie-Einstellungen oeffnen"
title="Cookie-Einstellungen"
>
<svg className="w-6 h-6" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M9 12l2 2 4-4m5.618-4.016A11.955 11.955 0 0112 2.944a11.955 11.955 0 01-8.618 3.04A12.02 12.02 0 003 9c0 5.591 3.824 10.29 9 11.622 5.176-1.332 9-6.03 9-11.622 0-1.042-.133-2.052-.382-3.016z" />
</svg>
</button>
)
}
function CategoryToggle({
label,
description,
checked,
disabled,
onChange,
}: {
label: string
description: string
checked: boolean
disabled?: boolean
onChange: (v: boolean) => void
}) {
return (
<div className="flex items-start justify-between gap-4 py-2">
<div className="flex-1">
<div className="text-sm font-medium text-gray-900">{label}</div>
<div className="text-xs text-gray-500 mt-0.5">{description}</div>
</div>
<button
onClick={() => !disabled && onChange(!checked)}
disabled={disabled}
className={`relative inline-flex h-6 w-11 items-center rounded-full transition-colors shrink-0 ${
checked
? disabled ? 'bg-gray-400' : 'bg-purple-600'
: 'bg-gray-200'
} ${disabled ? 'cursor-not-allowed opacity-60' : 'cursor-pointer'}`}
>
<span
className={`inline-block h-4 w-4 transform rounded-full bg-white shadow-sm transition-transform ${
checked ? 'translate-x-6' : 'translate-x-1'
}`}
/>
</button>
</div>
)
}
@@ -0,0 +1,515 @@
import { test, expect } from '@playwright/test'
/**
* Banner Consent API Integration Tests
*
* Tests the complete lifecycle of cookie banner consents:
* - Record/retrieve/withdraw consent
* - Email linking for DSR integration
* - Consent export (Art. 15/20 DSGVO)
* - Consent deletion (Art. 17 DSGVO erasure)
* - Consent sync to Einwilligungen
* - Vendor sync from service registry
* - Site config with config_version for consent proof
*/
const API_BASE = process.env.PLAYWRIGHT_API_URL || 'https://macmini:8093'
const TENANT_ID = '9282a473-5c95-4b3a-bf78-0ecc0ec71d3e'
const HEADERS = {
'Content-Type': 'application/json',
'X-Tenant-ID': TENANT_ID,
}
// Test data
const TEST_SITE_ID = `e2e-banner-test-${Date.now()}`
const TEST_DEVICE_FP = `e2e-device-${Date.now()}`
const TEST_EMAIL = `e2e-test-${Date.now()}@example.com`
test.describe('Banner Consent API — Full Lifecycle', () => {
let siteConfigId: string | null = null
let consentId: string | null = null
// ─── Setup: Create site config ───────────────────────────
test('01 — Create site config', async ({ request }) => {
const res = await request.post(`${API_BASE}/banner/admin/sites`, {
headers: HEADERS,
data: {
site_id: TEST_SITE_ID,
site_name: 'E2E Test Site',
site_url: 'https://e2e-test.example.com',
banner_title: 'Cookie-Einstellungen',
banner_description: 'Wir verwenden Cookies fuer E2E Tests.',
privacy_url: '/datenschutz',
imprint_url: '/impressum',
},
})
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.site_id).toBe(TEST_SITE_ID)
expect(body.config_version).toBe(1)
siteConfigId = body.id
})
test('02 — Create categories for site', async ({ request }) => {
const categories = [
{ category_key: 'necessary', name_de: 'Notwendig', is_required: true, sort_order: 0 },
{ category_key: 'statistics', name_de: 'Statistik', is_required: false, sort_order: 1 },
{ category_key: 'marketing', name_de: 'Marketing', is_required: false, sort_order: 2 },
]
for (const cat of categories) {
const res = await request.post(`${API_BASE}/banner/admin/sites/${TEST_SITE_ID}/categories`, {
headers: HEADERS,
data: cat,
})
expect(res.ok()).toBeTruthy()
}
})
test('03 — Create vendor configs', async ({ request }) => {
const vendors = [
{ vendor_name: 'Google Analytics', category_key: 'statistics', retention_days: 790, cookie_names: ['_ga', '_gid'] },
{ vendor_name: 'Facebook Pixel', category_key: 'marketing', retention_days: 90, cookie_names: ['_fbp'] },
]
for (const v of vendors) {
const res = await request.post(`${API_BASE}/banner/admin/sites/${TEST_SITE_ID}/vendors`, {
headers: HEADERS,
data: v,
})
expect(res.ok()).toBeTruthy()
}
})
// ─── Core Consent CRUD ───────────────────────────────────
test('04 — Get site config for banner display', async ({ request }) => {
const res = await request.get(`${API_BASE}/banner/config/${TEST_SITE_ID}`, { headers: HEADERS })
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.site_id).toBe(TEST_SITE_ID)
expect(body.banner_title).toBe('Cookie-Einstellungen')
expect(body.categories.length).toBe(3)
expect(body.vendors.length).toBe(2)
expect(body.config_version).toBe(1)
})
test('05 — Record consent (accept statistics only)', async ({ request }) => {
const res = await request.post(`${API_BASE}/banner/consent`, {
headers: HEADERS,
data: {
site_id: TEST_SITE_ID,
device_fingerprint: TEST_DEVICE_FP,
categories: ['necessary', 'statistics'],
vendors: ['Google Analytics'],
ip_address: '192.168.1.100',
user_agent: 'Playwright E2E Test',
},
})
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.device_fingerprint).toBe(TEST_DEVICE_FP)
expect(body.categories).toEqual(['necessary', 'statistics'])
expect(body.vendors).toEqual(['Google Analytics'])
expect(body.ip_hash).toBeTruthy() // IP is hashed
expect(body.linked_email).toBeNull()
consentId = body.id
})
test('06 — Retrieve consent for device', async ({ request }) => {
const res = await request.get(
`${API_BASE}/banner/consent?site_id=${TEST_SITE_ID}&device_fingerprint=${TEST_DEVICE_FP}`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.has_consent).toBe(true)
expect(body.consent.categories).toEqual(['necessary', 'statistics'])
})
test('07 — Update consent (accept all categories)', async ({ request }) => {
const res = await request.post(`${API_BASE}/banner/consent`, {
headers: HEADERS,
data: {
site_id: TEST_SITE_ID,
device_fingerprint: TEST_DEVICE_FP,
categories: ['necessary', 'statistics', 'marketing'],
vendors: ['Google Analytics', 'Facebook Pixel'],
ip_address: '192.168.1.100',
user_agent: 'Playwright E2E Test',
},
})
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.categories).toEqual(['necessary', 'statistics', 'marketing'])
expect(body.id).toBe(consentId) // Same consent row (upsert)
})
// ─── Phase 3: Email Linking ──────────────────────────────
test('08 — Link email to device fingerprint', async ({ request }) => {
const res = await request.post(`${API_BASE}/banner/consent/link-email`, {
headers: HEADERS,
data: {
site_id: TEST_SITE_ID,
device_fingerprint: TEST_DEVICE_FP,
email: TEST_EMAIL,
},
})
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.linked_email).toBe(TEST_EMAIL.toLowerCase())
expect(body.device_fingerprint).toBe(TEST_DEVICE_FP)
})
test('09 — Find consents by email (Art. 15)', async ({ request }) => {
const res = await request.get(
`${API_BASE}/banner/consent/by-email/${encodeURIComponent(TEST_EMAIL)}`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.length).toBe(1)
expect(body[0].linked_email).toBe(TEST_EMAIL.toLowerCase())
expect(body[0].device_fingerprint).toBe(TEST_DEVICE_FP)
})
test('10 — Export consent data for DSR (Art. 15/20)', async ({ request }) => {
const res = await request.get(
`${API_BASE}/banner/consent/dsr-export/${encodeURIComponent(TEST_EMAIL)}`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.email).toBe(TEST_EMAIL.toLowerCase())
expect(body.banner_consents.length).toBe(1)
expect(body.audit_trail.length).toBeGreaterThan(0)
// Verify consent proof fields in audit trail
const lastAudit = body.audit_trail[0]
expect(lastAudit.action).toBeTruthy()
expect(lastAudit.site_id).toBe(TEST_SITE_ID)
})
// ─── Phase 4: Consent Sync ──────────────────────────────
test('11 — Sync banner consent to Einwilligungen', async ({ request }) => {
const res = await request.post(`${API_BASE}/banner/consent/sync`, {
headers: HEADERS,
data: {
site_id: TEST_SITE_ID,
device_fingerprint: TEST_DEVICE_FP,
email: TEST_EMAIL,
},
})
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.synced).toBeGreaterThan(0)
expect(body.categories).toContain('necessary')
expect(body.categories).toContain('statistics')
expect(body.categories).toContain('marketing')
expect(body.email).toBe(TEST_EMAIL.toLowerCase())
})
// ─── DSGVO Export ────────────────────────────────────────
test('12 — Export consent per device (existing endpoint)', async ({ request }) => {
const res = await request.get(
`${API_BASE}/banner/consent/export?site_id=${TEST_SITE_ID}&device_fingerprint=${TEST_DEVICE_FP}`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.device_fingerprint).toBe(TEST_DEVICE_FP)
expect(body.consents.length).toBe(1)
expect(body.audit_trail.length).toBeGreaterThan(0)
})
// ─── Stats ───────────────────────────────────────────────
test('13 — Get site consent statistics', async ({ request }) => {
const res = await request.get(`${API_BASE}/banner/admin/stats/${TEST_SITE_ID}`, {
headers: HEADERS,
})
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.site_id).toBe(TEST_SITE_ID)
expect(body.total_consents).toBeGreaterThan(0)
expect(body.category_acceptance.necessary).toBeTruthy()
expect(body.category_acceptance.statistics).toBeTruthy()
})
// ─── Withdraw + Cleanup ─────────────────────────────────
test('14 — Withdraw consent', async ({ request }) => {
expect(consentId).toBeTruthy()
const res = await request.delete(`${API_BASE}/banner/consent/${consentId}`, {
headers: HEADERS,
})
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.success).toBe(true)
})
test('15 — Verify consent withdrawn (no consent found)', async ({ request }) => {
const res = await request.get(
`${API_BASE}/banner/consent?site_id=${TEST_SITE_ID}&device_fingerprint=${TEST_DEVICE_FP}`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.has_consent).toBe(false)
})
// ─── Cleanup: Delete site config ────────────────────────
test('16 — Cleanup: Delete test site', async ({ request }) => {
const res = await request.delete(`${API_BASE}/banner/admin/sites/${TEST_SITE_ID}`, {
headers: HEADERS,
})
expect(res.status()).toBe(204)
})
})
test.describe('Banner Consent API — Art. 17 Erasure via Email', () => {
const ERASURE_SITE = `e2e-erasure-${Date.now()}`
const ERASURE_DEVICE_1 = `e2e-dev1-${Date.now()}`
const ERASURE_DEVICE_2 = `e2e-dev2-${Date.now()}`
const ERASURE_EMAIL = `erasure-${Date.now()}@example.com`
test.beforeAll(async ({ request }) => {
// Create site
await request.post(`${API_BASE}/banner/admin/sites`, {
headers: HEADERS,
data: { site_id: ERASURE_SITE, site_name: 'Erasure Test Site' },
})
// Record consent on two devices with same email
for (const fp of [ERASURE_DEVICE_1, ERASURE_DEVICE_2]) {
await request.post(`${API_BASE}/banner/consent`, {
headers: HEADERS,
data: {
site_id: ERASURE_SITE,
device_fingerprint: fp,
categories: ['necessary', 'statistics'],
vendors: [],
},
})
await request.post(`${API_BASE}/banner/consent/link-email`, {
headers: HEADERS,
data: { site_id: ERASURE_SITE, device_fingerprint: fp, email: ERASURE_EMAIL },
})
}
})
test('should find 2 consents for email', async ({ request }) => {
const res = await request.get(
`${API_BASE}/banner/consent/by-email/${encodeURIComponent(ERASURE_EMAIL)}`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.length).toBe(2)
})
test('should delete all consents by email (Art. 17)', async ({ request }) => {
const res = await request.delete(
`${API_BASE}/banner/consent/by-email/${encodeURIComponent(ERASURE_EMAIL)}`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.deleted).toBe(2)
expect(body.email).toBe(ERASURE_EMAIL.toLowerCase())
})
test('should find 0 consents after erasure', async ({ request }) => {
const res = await request.get(
`${API_BASE}/banner/consent/by-email/${encodeURIComponent(ERASURE_EMAIL)}`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.length).toBe(0)
})
test.afterAll(async ({ request }) => {
await request.delete(`${API_BASE}/banner/admin/sites/${ERASURE_SITE}`, {
headers: HEADERS,
})
})
})
test.describe('Banner Consent API — Vendor Sync from Registry', () => {
const SYNC_SITE = `e2e-sync-${Date.now()}`
test.beforeAll(async ({ request }) => {
await request.post(`${API_BASE}/banner/admin/sites`, {
headers: HEADERS,
data: { site_id: SYNC_SITE, site_name: 'Vendor Sync Test' },
})
})
test('should sync vendors from service registry', async ({ request }) => {
const res = await request.post(
`${API_BASE}/banner/admin/sites/${SYNC_SITE}/sync-vendors`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.total).toBeGreaterThan(0)
expect(body.created).toBeGreaterThan(0)
})
test('should list synced vendors', async ({ request }) => {
const res = await request.get(
`${API_BASE}/banner/admin/sites/${SYNC_SITE}/vendors`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const vendors = await res.json()
expect(vendors.length).toBeGreaterThan(0)
// Verify vendor structure
const ga = vendors.find((v: any) => v.vendor_name === 'Google Analytics')
if (ga) {
expect(ga.category_key).toBe('statistics')
expect(ga.retention_days).toBe(790) // 26 months
}
})
test.afterAll(async ({ request }) => {
await request.delete(`${API_BASE}/banner/admin/sites/${SYNC_SITE}`, {
headers: HEADERS,
})
})
})
test.describe('Banner Consent API — Edge Cases & Validation', () => {
test('should return has_consent=false for unknown device', async ({ request }) => {
const res = await request.get(
`${API_BASE}/banner/consent?site_id=nonexistent&device_fingerprint=unknown`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.has_consent).toBe(false)
})
test('should return default config for unconfigured site', async ({ request }) => {
const res = await request.get(`${API_BASE}/banner/config/nonexistent-site`, {
headers: HEADERS,
})
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body.site_id).toBe('nonexistent-site')
expect(body.banner_title).toBe('Cookie-Einstellungen')
expect(body.categories).toEqual([])
})
test('should reject invalid email for link-email', async ({ request }) => {
const res = await request.post(`${API_BASE}/banner/consent/link-email`, {
headers: HEADERS,
data: {
site_id: 'test',
device_fingerprint: 'test',
email: 'not-an-email',
},
})
// Should fail — either 400 or 404 (no consent found)
expect(res.status()).toBeGreaterThanOrEqual(400)
})
test('should return empty list for unknown email in by-email', async ({ request }) => {
const res = await request.get(
`${API_BASE}/banner/consent/by-email/nobody@nowhere.test`,
{ headers: HEADERS },
)
expect(res.ok()).toBeTruthy()
const body = await res.json()
expect(body).toEqual([])
})
test('should hash IP address (never store plain IP)', async ({ request }) => {
const siteId = `e2e-ip-test-${Date.now()}`
const fp = `e2e-ip-fp-${Date.now()}`
// Create site
await request.post(`${API_BASE}/banner/admin/sites`, {
headers: HEADERS,
data: { site_id: siteId, site_name: 'IP Test' },
})
// Record consent with IP
const res = await request.post(`${API_BASE}/banner/consent`, {
headers: HEADERS,
data: {
site_id: siteId,
device_fingerprint: fp,
categories: ['necessary'],
vendors: [],
ip_address: '10.0.0.42',
},
})
expect(res.ok()).toBeTruthy()
const body = await res.json()
// IP should be hashed, not stored plain
expect(body.ip_hash).toBeTruthy()
expect(body.ip_hash).not.toBe('10.0.0.42')
expect(body.ip_hash.length).toBe(16) // SHA256[:16]
// Cleanup
await request.delete(`${API_BASE}/banner/consent/${body.id}`, { headers: HEADERS })
await request.delete(`${API_BASE}/banner/admin/sites/${siteId}`, { headers: HEADERS })
})
})
test.describe('Banner Consent API — Retention per Category', () => {
const RET_SITE = `e2e-retention-${Date.now()}`
test.beforeAll(async ({ request }) => {
// Create site with categories and vendors
await request.post(`${API_BASE}/banner/admin/sites`, {
headers: HEADERS,
data: { site_id: RET_SITE, site_name: 'Retention Test' },
})
await request.post(`${API_BASE}/banner/admin/sites/${RET_SITE}/categories`, {
headers: HEADERS,
data: { category_key: 'necessary', name_de: 'Notwendig', is_required: true },
})
await request.post(`${API_BASE}/banner/admin/sites/${RET_SITE}/categories`, {
headers: HEADERS,
data: { category_key: 'marketing', name_de: 'Marketing', is_required: false },
})
await request.post(`${API_BASE}/banner/admin/sites/${RET_SITE}/vendors`, {
headers: HEADERS,
data: { vendor_name: 'FB Pixel', category_key: 'marketing', retention_days: 90 },
})
})
test('consent expiry should match max vendor retention', async ({ request }) => {
const fp = `e2e-ret-fp-${Date.now()}`
const res = await request.post(`${API_BASE}/banner/consent`, {
headers: HEADERS,
data: {
site_id: RET_SITE,
device_fingerprint: fp,
categories: ['necessary', 'marketing'],
vendors: ['FB Pixel'],
},
})
expect(res.ok()).toBeTruthy()
const body = await res.json()
// Expiry should be based on max vendor retention (90 days for marketing)
const expiresAt = new Date(body.expires_at)
const createdAt = new Date(body.created_at)
const diffDays = Math.round((expiresAt.getTime() - createdAt.getTime()) / (1000 * 60 * 60 * 24))
// Should be 90 days (marketing) not 365 (default)
expect(diffDays).toBeGreaterThanOrEqual(89)
expect(diffDays).toBeLessThanOrEqual(91)
// Cleanup
await request.delete(`${API_BASE}/banner/consent/${body.id}`, { headers: HEADERS })
})
test.afterAll(async ({ request }) => {
await request.delete(`${API_BASE}/banner/admin/sites/${RET_SITE}`, {
headers: HEADERS,
})
})
})
@@ -0,0 +1,256 @@
import { test, expect } from '@playwright/test'
/**
* Cookie Banner Overlay E2E Tests
*
* Tests the live cookie consent banner in the Compliance SDK:
* - Auto-opens on first visit
* - Consent choices (accept all, reject all, custom settings)
* - FAB button to reopen after consent
* - Persistence in localStorage
* - Correct toggle behavior for categories
*/
const SDK_URL = '/sdk'
const STORAGE_KEY = 'bp-sdk-cookie-consent'
test.describe('Cookie Banner — First Visit', () => {
test.beforeEach(async ({ page }) => {
// Clear localStorage to simulate first visit
await page.goto(SDK_URL)
await page.evaluate(() => localStorage.removeItem('bp-sdk-cookie-consent'))
await page.reload()
await page.waitForLoadState('networkidle')
})
test('should show banner on first visit', async ({ page }) => {
// Banner should be visible
await expect(page.getByText('Cookie-Einstellungen')).toBeVisible({ timeout: 10000 })
await expect(page.getByText('Wir verwenden Cookies')).toBeVisible()
})
test('should show three buttons: accept all, reject, settings', async ({ page }) => {
await page.waitForTimeout(500)
await expect(page.getByRole('button', { name: 'Alle akzeptieren' })).toBeVisible()
await expect(page.getByRole('button', { name: 'Nur notwendige' })).toBeVisible()
await expect(page.getByRole('button', { name: 'Einstellungen' })).toBeVisible()
})
test('should have Datenschutzerklaerung and Impressum links', async ({ page }) => {
await page.waitForTimeout(500)
await expect(page.getByText('Datenschutzerklaerung')).toBeVisible()
await expect(page.getByText('Impressum')).toBeVisible()
})
test('should show overlay backdrop', async ({ page }) => {
await page.waitForTimeout(500)
// Check for the dark overlay behind the banner
const overlay = page.locator('.bg-black\\/40')
await expect(overlay).toBeVisible()
})
})
test.describe('Cookie Banner — Accept All', () => {
test('should close banner and save consent on "Alle akzeptieren"', async ({ page }) => {
await page.goto(SDK_URL)
await page.evaluate(() => localStorage.removeItem('bp-sdk-cookie-consent'))
await page.reload()
await page.waitForLoadState('networkidle')
// Click accept all
await page.getByRole('button', { name: 'Alle akzeptieren' }).click()
// Banner should close
await expect(page.getByText('Cookie-Einstellungen').first()).not.toBeVisible({ timeout: 5000 })
// Verify localStorage
const consent = await page.evaluate(() => {
const raw = localStorage.getItem('bp-sdk-cookie-consent')
return raw ? JSON.parse(raw) : null
})
expect(consent).toBeTruthy()
expect(consent.necessary).toBe(true)
expect(consent.statistics).toBe(true)
expect(consent.marketing).toBe(true)
expect(consent.functional).toBe(true)
expect(consent.timestamp).toBeTruthy()
})
})
test.describe('Cookie Banner — Reject All', () => {
test('should save only necessary on "Nur notwendige"', async ({ page }) => {
await page.goto(SDK_URL)
await page.evaluate(() => localStorage.removeItem('bp-sdk-cookie-consent'))
await page.reload()
await page.waitForLoadState('networkidle')
// Click reject all
await page.getByRole('button', { name: 'Nur notwendige' }).click()
// Banner should close
await expect(page.getByText('Cookie-Einstellungen').first()).not.toBeVisible({ timeout: 5000 })
// Verify localStorage — only necessary enabled
const consent = await page.evaluate(() => {
const raw = localStorage.getItem('bp-sdk-cookie-consent')
return raw ? JSON.parse(raw) : null
})
expect(consent).toBeTruthy()
expect(consent.necessary).toBe(true)
expect(consent.statistics).toBe(false)
expect(consent.marketing).toBe(false)
expect(consent.functional).toBe(false)
})
})
test.describe('Cookie Banner — Custom Settings', () => {
test('should expand category toggles on "Einstellungen"', async ({ page }) => {
await page.goto(SDK_URL)
await page.evaluate(() => localStorage.removeItem('bp-sdk-cookie-consent'))
await page.reload()
await page.waitForLoadState('networkidle')
// Click settings
await page.getByRole('button', { name: 'Einstellungen' }).click()
// Category toggles should appear
await expect(page.getByText('Notwendig')).toBeVisible()
await expect(page.getByText('Statistik')).toBeVisible()
await expect(page.getByText('Marketing')).toBeVisible()
await expect(page.getByText('Funktional')).toBeVisible()
})
test('should have "Auswahl speichern" button in settings view', async ({ page }) => {
await page.goto(SDK_URL)
await page.evaluate(() => localStorage.removeItem('bp-sdk-cookie-consent'))
await page.reload()
await page.waitForLoadState('networkidle')
await page.getByRole('button', { name: 'Einstellungen' }).click()
await expect(page.getByRole('button', { name: 'Auswahl speichern' })).toBeVisible()
})
test('should save custom selection', async ({ page }) => {
await page.goto(SDK_URL)
await page.evaluate(() => localStorage.removeItem('bp-sdk-cookie-consent'))
await page.reload()
await page.waitForLoadState('networkidle')
// Open settings
await page.getByRole('button', { name: 'Einstellungen' }).click()
await page.waitForTimeout(300)
// Toggle statistics on (click the toggle next to "Statistik")
const statistikRow = page.locator('text=Statistik').locator('..')
const toggle = statistikRow.locator('button').last()
await toggle.click()
// Save
await page.getByRole('button', { name: 'Auswahl speichern' }).click()
// Verify
const consent = await page.evaluate(() => {
const raw = localStorage.getItem('bp-sdk-cookie-consent')
return raw ? JSON.parse(raw) : null
})
expect(consent).toBeTruthy()
expect(consent.necessary).toBe(true)
expect(consent.statistics).toBe(true)
expect(consent.marketing).toBe(false)
})
test('necessary toggle should be disabled', async ({ page }) => {
await page.goto(SDK_URL)
await page.evaluate(() => localStorage.removeItem('bp-sdk-cookie-consent'))
await page.reload()
await page.waitForLoadState('networkidle')
await page.getByRole('button', { name: 'Einstellungen' }).click()
await page.waitForTimeout(300)
// The "Notwendig" toggle should be disabled (can't be turned off)
const necessaryRow = page.locator('text=Notwendig').locator('..')
const toggle = necessaryRow.locator('button[disabled]')
await expect(toggle).toBeVisible()
})
})
test.describe('Cookie Banner — Persistence', () => {
test('should NOT show banner on subsequent visits', async ({ page }) => {
await page.goto(SDK_URL)
// Set consent in localStorage
await page.evaluate(() => {
localStorage.setItem('bp-sdk-cookie-consent', JSON.stringify({
necessary: true, statistics: true, marketing: false, functional: false,
timestamp: new Date().toISOString(),
}))
})
await page.reload()
await page.waitForLoadState('networkidle')
await page.waitForTimeout(1000)
// Banner should NOT appear
const bannerVisible = await page.getByText('Wir verwenden Cookies').isVisible().catch(() => false)
expect(bannerVisible).toBe(false)
})
})
test.describe('Cookie Banner — FAB Reopen Button', () => {
test('should show cookie FAB after consent is given', async ({ page }) => {
await page.goto(SDK_URL)
await page.evaluate(() => {
localStorage.setItem('bp-sdk-cookie-consent', JSON.stringify({
necessary: true, statistics: true, marketing: false, functional: false,
timestamp: new Date().toISOString(),
}))
})
await page.reload()
await page.waitForLoadState('networkidle')
await page.waitForTimeout(1000)
// FAB button should be visible (shield icon)
const fab = page.locator('button[aria-label="Cookie-Einstellungen oeffnen"]')
await expect(fab).toBeVisible({ timeout: 5000 })
})
test('should reopen banner when FAB is clicked', async ({ page }) => {
await page.goto(SDK_URL)
await page.evaluate(() => {
localStorage.setItem('bp-sdk-cookie-consent', JSON.stringify({
necessary: true, statistics: true, marketing: false, functional: false,
timestamp: new Date().toISOString(),
}))
})
await page.reload()
await page.waitForLoadState('networkidle')
await page.waitForTimeout(1000)
// Click FAB
const fab = page.locator('button[aria-label="Cookie-Einstellungen oeffnen"]')
await fab.click()
// Banner should reopen with settings expanded
await expect(page.getByText('Cookie-Einstellungen')).toBeVisible({ timeout: 5000 })
// Settings should be shown (since reopening goes straight to settings)
await expect(page.getByText('Statistik')).toBeVisible()
await expect(page.getByText('Marketing')).toBeVisible()
})
test('should NOT show FAB before consent is given', async ({ page }) => {
await page.goto(SDK_URL)
await page.evaluate(() => localStorage.removeItem('bp-sdk-cookie-consent'))
await page.reload()
await page.waitForLoadState('networkidle')
await page.waitForTimeout(1000)
// FAB should NOT be visible (banner is showing instead)
const fab = page.locator('button[aria-label="Cookie-Einstellungen oeffnen"]')
const isVisible = await fab.isVisible().catch(() => false)
expect(isVisible).toBe(false)
})
})
@@ -0,0 +1,139 @@
import { test, expect } from '@playwright/test'
import { navigateToSDK, waitForPageLoad } from '../utils/test-helpers'
/**
* Cookie Banner UI E2E Tests
*
* Tests the cookie banner configuration page and admin UI.
* Verifies that the banner builder, category management,
* and code export work correctly.
*/
test.describe('Cookie Banner Configuration Page', () => {
test.beforeEach(async ({ page }) => {
await navigateToSDK(page, '/cookie-banner')
})
test('should load cookie banner page', async ({ page }) => {
// The page should load with the step header
await expect(page.getByText('Cookie-Banner')).toBeVisible({ timeout: 10000 })
})
test('should display category cards', async ({ page }) => {
// Wait for content to load
await page.waitForTimeout(1000)
// Check for standard cookie categories
const pageContent = await page.textContent('body')
expect(pageContent).toBeTruthy()
// At minimum the "Notwendig" category should exist
const hasCategories = pageContent?.includes('Notwendig') ||
pageContent?.includes('necessary') ||
pageContent?.includes('Kategorie')
expect(hasCategories).toBeTruthy()
})
test('should have banner preview section', async ({ page }) => {
await page.waitForTimeout(1000)
// Check for preview or banner-related UI elements
const hasPreview = await page.locator('text=Vorschau').isVisible().catch(() => false) ||
await page.locator('text=Preview').isVisible().catch(() => false) ||
await page.locator('text=Banner').isVisible().catch(() => false)
expect(hasPreview).toBeTruthy()
})
test('should have export/publish buttons', async ({ page }) => {
// Check for action buttons
const exportBtn = page.getByRole('button', { name: /Code exportieren|Export/ })
const publishBtn = page.getByRole('button', { name: /Veroeffentlichen|Speichern|Publish/ })
const hasExport = await exportBtn.isVisible().catch(() => false)
const hasPublish = await publishBtn.isVisible().catch(() => false)
// At least one action button should be present
expect(hasExport || hasPublish).toBeTruthy()
})
test('should display cookie statistics', async ({ page }) => {
await page.waitForTimeout(1000)
// Check for statistics display (cookie count, third-party count)
const pageContent = await page.textContent('body')
const hasStats = pageContent?.includes('Cookie') || pageContent?.includes('Vendor')
expect(hasStats).toBeTruthy()
})
})
test.describe('Cookie Banner Navigation', () => {
test('should be reachable from SDK sidebar', async ({ page }) => {
// Navigate to SDK dashboard first
await navigateToSDK(page, '')
await page.waitForTimeout(1000)
// Look for cookie banner link in sidebar or navigation
const bannerLink = page.locator('a[href*="cookie-banner"]').first()
if (await bannerLink.isVisible().catch(() => false)) {
await bannerLink.click()
await waitForPageLoad(page)
await expect(page).toHaveURL(/cookie-banner/)
}
})
test('should navigate between consent management pages', async ({ page }) => {
await navigateToSDK(page, '/einwilligungen')
await page.waitForTimeout(1000)
// Check if tabs/links to cookie-banner exist
const cookieBannerTab = page.locator('a[href*="cookie-banner"], button:has-text("Cookie")')
const isVisible = await cookieBannerTab.first().isVisible().catch(() => false)
if (isVisible) {
await cookieBannerTab.first().click()
await waitForPageLoad(page)
}
})
})
test.describe('Consent Management Page', () => {
test.beforeEach(async ({ page }) => {
await navigateToSDK(page, '/consent-management')
})
test('should load consent management page', async ({ page }) => {
await page.waitForTimeout(1000)
const pageContent = await page.textContent('body')
// Page should have consent-related content
const hasContent = pageContent?.includes('Consent') ||
pageContent?.includes('Einwilligung') ||
pageContent?.includes('consent')
expect(hasContent).toBeTruthy()
})
})
test.describe('DSR Module — Banner Integration', () => {
test.beforeEach(async ({ page }) => {
await navigateToSDK(page, '/dsr')
})
test('should load DSR portal', async ({ page }) => {
await expect(page.getByRole('heading', { name: /DSR|Betroffenen/ })).toBeVisible({
timeout: 10000,
})
})
test('should have tab navigation for DSR workflow', async ({ page }) => {
await page.waitForTimeout(1000)
// DSR should have workflow tabs
const pageContent = await page.textContent('body')
const hasTabs = pageContent?.includes('Eingang') ||
pageContent?.includes('Bearbeitung') ||
pageContent?.includes('Abgeschlossen')
expect(hasTabs).toBeTruthy()
})
})
@@ -20,12 +20,16 @@ from compliance.api._http_errors import translate_domain_errors
from compliance.schemas.banner import (
CategoryConfigCreate,
ConsentCreate,
ConsentSyncRequest,
LinkEmailRequest,
SiteConfigCreate,
SiteConfigUpdate,
VendorConfigCreate,
)
from compliance.services.banner_admin_service import BannerAdminService
from compliance.services.banner_consent_service import BannerConsentService
from compliance.services.banner_dsr_service import BannerDSRService
from compliance.services.vendor_banner_sync import get_banner_vendors_from_registry
router = APIRouter(prefix="/banner", tags=["compliance-banner"])
@@ -48,6 +52,10 @@ def get_admin_service(db: Session = Depends(get_db)) -> BannerAdminService:
return BannerAdminService(db)
def get_dsr_service(db: Session = Depends(get_db)) -> BannerDSRService:
return BannerDSRService(db)
# =============================================================================
# Public SDK Endpoints (fuer Einbettung in Kunden-Websites)
# =============================================================================
@@ -118,6 +126,69 @@ async def export_consent(
return service.export_consent(tenant_id, site_id, device_fingerprint)
# =============================================================================
# DSR Integration — Email Linking + Consent Sync (Phase 3 + 4)
# =============================================================================
@router.post("/consent/link-email")
async def link_email(
body: LinkEmailRequest,
tenant_id: str = Depends(_get_tenant),
service: BannerDSRService = Depends(get_dsr_service),
) -> dict[str, Any]:
"""Link an email to a device fingerprint (e.g. after signup/login)."""
with translate_domain_errors():
return service.link_email(
tenant_id, body.site_id, body.device_fingerprint, body.email,
)
@router.get("/consent/by-email/{email}")
async def get_consents_by_email(
email: str,
tenant_id: str = Depends(_get_tenant),
service: BannerDSRService = Depends(get_dsr_service),
) -> list[dict[str, Any]]:
"""Find all banner consents linked to an email (Art. 15 DSGVO)."""
with translate_domain_errors():
return service.get_consents_by_email(tenant_id, email)
@router.delete("/consent/by-email/{email}")
async def delete_consents_by_email(
email: str,
tenant_id: str = Depends(_get_tenant),
service: BannerDSRService = Depends(get_dsr_service),
) -> dict[str, Any]:
"""Delete all banner consents for an email (Art. 17 DSGVO erasure)."""
with translate_domain_errors():
return service.delete_consents_by_email(tenant_id, email)
@router.get("/consent/dsr-export/{email}")
async def export_for_dsr(
email: str,
tenant_id: str = Depends(_get_tenant),
service: BannerDSRService = Depends(get_dsr_service),
) -> dict[str, Any]:
"""Export all banner consent data for DSR (Art. 15/20 DSGVO)."""
with translate_domain_errors():
return service.export_for_dsr(tenant_id, email)
@router.post("/consent/sync")
async def sync_consent(
body: ConsentSyncRequest,
tenant_id: str = Depends(_get_tenant),
service: BannerDSRService = Depends(get_dsr_service),
) -> dict[str, Any]:
"""Sync banner consent to Einwilligungen (Phase 4 — user-based bridge)."""
with translate_domain_errors():
return service.sync_consent_to_einwilligungen(
tenant_id, body.device_fingerprint, body.email, body.site_id,
)
# =============================================================================
# Admin — Stats
# =============================================================================
@@ -253,3 +324,43 @@ async def delete_vendor(
"""Delete a vendor."""
with translate_domain_errors():
service.delete_vendor(vendor_id)
# =============================================================================
# Admin — Vendor Sync from Service Registry (Phase 1)
# =============================================================================
@router.post("/admin/sites/{site_id}/sync-vendors")
async def sync_vendors_from_registry(
site_id: str,
tenant_id: str = Depends(_get_tenant),
service: BannerAdminService = Depends(get_admin_service),
) -> dict[str, Any]:
"""Sync 82+ services from service registry to banner vendor configs."""
with translate_domain_errors():
vendors = get_banner_vendors_from_registry()
created = 0
updated = 0
for v in vendors:
try:
existing = service.list_vendors(tenant_id, site_id)
match = next(
(e for e in existing if e["vendor_name"] == v["vendor_name"]),
None,
)
if match:
updated += 1
else:
from compliance.schemas.banner import VendorConfigCreate
service.create_vendor(tenant_id, site_id, VendorConfigCreate(
vendor_name=v["vendor_name"],
category_key=v["category_key"],
description_de=v["description_de"],
description_en=v["description_en"],
cookie_names=v["cookie_names"],
retention_days=v["retention_days"],
))
created += 1
except Exception:
continue
return {"created": created, "updated": updated, "total": len(vendors)}
@@ -34,6 +34,7 @@ class BannerConsentDB(Base):
ip_hash = Column(Text)
user_agent = Column(Text)
consent_string = Column(Text)
linked_email = Column(Text)
expires_at = Column(DateTime)
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
@@ -42,6 +43,8 @@ class BannerConsentDB(Base):
Index('idx_banner_consent_tenant', 'tenant_id'),
Index('idx_banner_consent_site', 'site_id'),
Index('idx_banner_consent_device', 'device_fingerprint'),
Index('idx_banner_consent_email', 'linked_email',
postgresql_where='linked_email IS NOT NULL'),
)
@@ -58,6 +61,8 @@ class BannerConsentAuditLogDB(Base):
device_fingerprint = Column(Text)
categories = Column(JSON, default=list)
ip_hash = Column(Text)
banner_config_hash = Column(Text)
consent_version = Column(Integer)
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
__table_args__ = (
@@ -85,6 +90,7 @@ class BannerSiteConfigDB(Base):
dsb_email = Column(Text)
theme = Column(JSON, default=dict)
tcf_enabled = Column(Boolean, default=False)
config_version = Column(Integer, nullable=False, default=1)
is_active = Column(Boolean, nullable=False, default=True)
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
@@ -76,10 +76,26 @@ class VendorConfigCreate(BaseModel):
retention_days: int = 365
class LinkEmailRequest(BaseModel):
"""Request body for linking an email to a device fingerprint."""
site_id: str
device_fingerprint: str
email: str
class ConsentSyncRequest(BaseModel):
"""Request body for syncing banner consent to Einwilligungen."""
site_id: str
device_fingerprint: str
email: str
__all__ = [
"ConsentCreate",
"SiteConfigCreate",
"SiteConfigUpdate",
"CategoryConfigCreate",
"VendorConfigCreate",
"LinkEmailRequest",
"ConsentSyncRequest",
]
@@ -25,6 +25,7 @@ def consent_to_dict(c: BannerConsentDB) -> dict[str, Any]:
"vendors": c.vendors or [],
"ip_hash": c.ip_hash,
"consent_string": c.consent_string,
"linked_email": c.linked_email,
"expires_at": c.expires_at.isoformat() if c.expires_at else None,
"created_at": c.created_at.isoformat() if c.created_at else None,
"updated_at": c.updated_at.isoformat() if c.updated_at else None,
@@ -45,6 +46,7 @@ def site_config_to_dict(s: BannerSiteConfigDB) -> dict[str, Any]:
"dsb_email": s.dsb_email,
"theme": s.theme or {},
"tcf_enabled": s.tcf_enabled,
"config_version": s.config_version,
"is_active": s.is_active,
"created_at": s.created_at.isoformat() if s.created_at else None,
"updated_at": s.updated_at.isoformat() if s.updated_at else None,
@@ -9,9 +9,12 @@ display), export, and per-site consent statistics.
Admin-side site/category/vendor management lives in
``compliance.services.banner_admin_service.BannerAdminService``.
DSR-facing email linking lives in
``compliance.services.banner_dsr_service.BannerDSRService``.
"""
import hashlib
import json
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any, Optional
@@ -33,6 +36,15 @@ from compliance.services._banner_serializers import (
vendor_to_dict,
)
# Default consent expiration per banner category (days).
# Based on: DSGVO Art. 5(1)(e), CNIL guidelines, EDPB recommendations.
CATEGORY_RETENTION_DAYS = {
"necessary": 365, # Session + functional = max 12 months
"statistics": 790, # Max 26 months (Google Analytics default)
"marketing": 90, # Max 90 days for retargeting
"functional": 365, # Max 12 months
}
class BannerConsentService:
"""Business logic for public SDK banner consent endpoints."""
@@ -59,6 +71,8 @@ class BannerConsentService:
device_fingerprint: Optional[str] = None,
categories: Optional[list[str]] = None,
ip_hash: Optional[str] = None,
banner_config_hash: Optional[str] = None,
consent_version: Optional[int] = None,
) -> None:
entry = BannerConsentAuditLogDB(
tenant_id=tenant_id,
@@ -68,9 +82,53 @@ class BannerConsentService:
device_fingerprint=device_fingerprint,
categories=categories or [],
ip_hash=ip_hash,
banner_config_hash=banner_config_hash,
consent_version=consent_version,
)
self.db.add(entry)
def _compute_config_hash(self, tenant_id: uuid.UUID, site_id: str) -> tuple[Optional[str], Optional[int]]:
"""Compute SHA256 hash of current site config for consent proof (Art. 7(1) DSGVO)."""
config = (
self.db.query(BannerSiteConfigDB)
.filter(
BannerSiteConfigDB.tenant_id == tenant_id,
BannerSiteConfigDB.site_id == site_id,
)
.first()
)
if not config:
return None, None
snapshot = json.dumps({
"banner_title": config.banner_title,
"banner_description": config.banner_description,
"privacy_url": config.privacy_url,
"imprint_url": config.imprint_url,
}, sort_keys=True)
return hashlib.sha256(snapshot.encode()).hexdigest()[:32], config.config_version
def _get_max_retention(self, tenant_id: uuid.UUID, site_id: str, categories: list[str]) -> int:
"""Determine consent expiration based on accepted categories and vendor retention."""
config = (
self.db.query(BannerSiteConfigDB)
.filter(BannerSiteConfigDB.tenant_id == tenant_id, BannerSiteConfigDB.site_id == site_id)
.first()
)
if not config:
return 365
vendors = (
self.db.query(BannerVendorConfigDB)
.filter(
BannerVendorConfigDB.site_config_id == config.id,
BannerVendorConfigDB.category_key.in_(categories),
BannerVendorConfigDB.is_active,
)
.all()
)
if vendors:
return max(v.retention_days for v in vendors if v.retention_days)
return max((CATEGORY_RETENTION_DAYS.get(c, 365) for c in categories), default=365)
# ------------------------------------------------------------------
# Consent CRUD (public SDK)
# ------------------------------------------------------------------
@@ -86,11 +144,19 @@ class BannerConsentService:
user_agent: Optional[str],
consent_string: Optional[str],
) -> dict[str, Any]:
"""Upsert a device consent row for (tenant, site, device_fingerprint)."""
"""Upsert a device consent row for (tenant, site, device_fingerprint).
Expiration is derived from the maximum vendor retention for the
accepted categories (Phase 2 — DSGVO Art. 5(1)(e)).
A SHA256 hash of the banner config is stored in the audit log
for consent proof (Phase 6 — Art. 7(1) DSGVO).
"""
tid = uuid.UUID(tenant_id)
ip_hash = self._hash_ip(ip_address)
now = datetime.now(timezone.utc)
expires_at = now + timedelta(days=365)
retention = self._get_max_retention(tid, site_id, categories)
expires_at = now + timedelta(days=retention)
config_hash, config_ver = self._compute_config_hash(tid, site_id)
existing = (
self.db.query(BannerConsentDB)
@@ -113,7 +179,7 @@ class BannerConsentService:
self.db.flush()
self._log(
tid, existing.id, "consent_updated", site_id, device_fingerprint,
categories, ip_hash,
categories, ip_hash, config_hash, config_ver,
)
self.db.commit()
self.db.refresh(existing)
@@ -134,7 +200,7 @@ class BannerConsentService:
self.db.flush()
self._log(
tid, consent.id, "consent_given", site_id, device_fingerprint,
categories, ip_hash,
categories, ip_hash, config_hash, config_ver,
)
self.db.commit()
self.db.refresh(consent)
@@ -0,0 +1,266 @@
# mypy: disable-error-code="arg-type,assignment"
"""
Banner DSR service — bridges device-based banner consents with
user-based DSR (Data Subject Request) processing.
Phase 3: Email linking allows correlating anonymous device fingerprints
with user emails (e.g. after newsletter signup or account creation).
This enables Art. 15 (access), Art. 17 (erasure), and Art. 20
(portability) requests to include/delete banner consent data.
Phase 4: Consent sync bridges banner consents (device-based) with
Einwilligungen (user-based) for unified consent management.
"""
import uuid
from datetime import datetime, timezone
from typing import Any, Optional
from sqlalchemy.orm import Session
from compliance.db.banner_models import (
BannerConsentAuditLogDB,
BannerConsentDB,
)
from compliance.db.einwilligungen_models import (
EinwilligungenConsentDB,
EinwilligungenConsentHistoryDB,
)
from compliance.domain import NotFoundError, ValidationError
from compliance.services._banner_serializers import consent_to_dict
class BannerDSRService:
"""Email linking + DSR integration for banner consents."""
def __init__(self, db: Session) -> None:
self.db = db
# ------------------------------------------------------------------
# Phase 3: Email linking
# ------------------------------------------------------------------
def link_email(
self,
tenant_id: str,
site_id: str,
device_fingerprint: str,
email: str,
) -> dict[str, Any]:
"""Link an email address to a device fingerprint's consent.
Typically called after newsletter signup, account creation, or
login — any point where the user's email becomes known.
"""
tid = uuid.UUID(tenant_id)
if not email or "@" not in email:
raise ValidationError("Invalid email address")
consent = (
self.db.query(BannerConsentDB)
.filter(
BannerConsentDB.tenant_id == tid,
BannerConsentDB.site_id == site_id,
BannerConsentDB.device_fingerprint == device_fingerprint,
)
.first()
)
if not consent:
raise NotFoundError("No consent found for this device")
consent.linked_email = email.lower().strip()
consent.updated_at = datetime.now(timezone.utc)
# Audit the linking
self.db.add(BannerConsentAuditLogDB(
tenant_id=tid,
consent_id=consent.id,
action="email_linked",
site_id=site_id,
device_fingerprint=device_fingerprint,
categories=consent.categories or [],
))
self.db.commit()
self.db.refresh(consent)
return consent_to_dict(consent)
def get_consents_by_email(
self, tenant_id: str, email: str,
) -> list[dict[str, Any]]:
"""Find all banner consents linked to an email (Art. 15 DSGVO)."""
tid = uuid.UUID(tenant_id)
consents = (
self.db.query(BannerConsentDB)
.filter(
BannerConsentDB.tenant_id == tid,
BannerConsentDB.linked_email == email.lower().strip(),
)
.all()
)
return [consent_to_dict(c) for c in consents]
def delete_consents_by_email(
self, tenant_id: str, email: str,
) -> dict[str, Any]:
"""Delete all banner consents for an email (Art. 17 DSGVO erasure).
Creates audit log entries before deletion for compliance proof.
"""
tid = uuid.UUID(tenant_id)
normalized = email.lower().strip()
consents = (
self.db.query(BannerConsentDB)
.filter(
BannerConsentDB.tenant_id == tid,
BannerConsentDB.linked_email == normalized,
)
.all()
)
deleted = 0
for c in consents:
self.db.add(BannerConsentAuditLogDB(
tenant_id=tid,
consent_id=c.id,
action="consent_deleted_dsr",
site_id=c.site_id,
device_fingerprint=c.device_fingerprint,
categories=c.categories or [],
))
self.db.delete(c)
deleted += 1
self.db.commit()
return {"deleted": deleted, "email": normalized}
def export_for_dsr(
self, tenant_id: str, email: str,
) -> dict[str, Any]:
"""Export all banner consent data for a DSR (Art. 15/20 DSGVO).
Returns consent records + audit trail for the email.
"""
tid = uuid.UUID(tenant_id)
normalized = email.lower().strip()
consents = (
self.db.query(BannerConsentDB)
.filter(
BannerConsentDB.tenant_id == tid,
BannerConsentDB.linked_email == normalized,
)
.all()
)
consent_ids = [c.id for c in consents]
audit = []
if consent_ids:
audit = (
self.db.query(BannerConsentAuditLogDB)
.filter(BannerConsentAuditLogDB.consent_id.in_(consent_ids))
.order_by(BannerConsentAuditLogDB.created_at.desc())
.all()
)
return {
"email": normalized,
"banner_consents": [consent_to_dict(c) for c in consents],
"audit_trail": [
{
"id": str(a.id),
"action": a.action,
"site_id": a.site_id,
"categories": a.categories or [],
"banner_config_hash": a.banner_config_hash,
"consent_version": a.consent_version,
"created_at": a.created_at.isoformat() if a.created_at else None,
}
for a in audit
],
}
# ------------------------------------------------------------------
# Phase 4: Consent sync (Banner ↔ Einwilligungen)
# ------------------------------------------------------------------
def sync_consent_to_einwilligungen(
self,
tenant_id: str,
device_fingerprint: str,
email: str,
site_id: str,
) -> dict[str, Any]:
"""Sync banner consent categories to user-based Einwilligungen.
Called when a user logs in and their email becomes known.
Creates/updates EinwilligungenConsent entries for each accepted
banner category, bridging device-based and user-based systems.
"""
tid = uuid.UUID(tenant_id)
consent = (
self.db.query(BannerConsentDB)
.filter(
BannerConsentDB.tenant_id == tid,
BannerConsentDB.site_id == site_id,
BannerConsentDB.device_fingerprint == device_fingerprint,
)
.first()
)
if not consent:
raise NotFoundError("No banner consent found for this device")
# Link email if not already linked
normalized = email.lower().strip()
if not consent.linked_email:
consent.linked_email = normalized
consent.updated_at = datetime.now(timezone.utc)
synced = 0
categories = consent.categories or []
for cat in categories:
data_point_id = f"banner_{cat}"
existing = (
self.db.query(EinwilligungenConsentDB)
.filter(
EinwilligungenConsentDB.tenant_id == tid,
EinwilligungenConsentDB.user_id == normalized,
EinwilligungenConsentDB.data_point_id == data_point_id,
)
.first()
)
now = datetime.now(timezone.utc)
if existing:
if not existing.granted:
existing.granted = True
existing.granted_at = now
existing.revoked_at = None
existing.source = "banner_sync"
self.db.add(EinwilligungenConsentHistoryDB(
consent_id=existing.id,
tenant_id=tid,
action="granted",
source="banner_sync",
))
synced += 1
else:
new_consent = EinwilligungenConsentDB(
tenant_id=tid,
user_id=normalized,
data_point_id=data_point_id,
granted=True,
granted_at=now,
consent_version="1",
source="banner_sync",
)
self.db.add(new_consent)
self.db.flush()
self.db.add(EinwilligungenConsentHistoryDB(
consent_id=new_consent.id,
tenant_id=tid,
action="granted",
source="banner_sync",
))
synced += 1
self.db.commit()
return {
"synced": synced,
"categories": categories,
"email": normalized,
}
@@ -142,9 +142,33 @@ class DSRWorkflowService:
if body.result_data:
dsr.data_export = body.result_data
dsr.updated_at = now
# Phase 3: Auto-delete banner consents on Art. 17 erasure
banner_result = None
if dsr.request_type == "erasure" and dsr.requester_email:
from compliance.services.banner_dsr_service import BannerDSRService
banner_svc = BannerDSRService(self._db)
banner_result = banner_svc.delete_consents_by_email(
tenant_id, dsr.requester_email,
)
# Phase 3: Include banner consents in data export for access/portability
if dsr.request_type in ("access", "portability") and dsr.requester_email:
from compliance.services.banner_dsr_service import BannerDSRService
banner_svc = BannerDSRService(self._db)
export = banner_svc.export_for_dsr(tenant_id, dsr.requester_email)
if export.get("banner_consents"):
existing_export = dsr.data_export or {}
if isinstance(existing_export, dict):
existing_export["banner_consents"] = export
dsr.data_export = existing_export
self._db.commit()
self._db.refresh(dsr)
return _dsr_to_dict(dsr)
result = _dsr_to_dict(dsr)
if banner_result:
result["banner_consents_deleted"] = banner_result["deleted"]
return result
# -- Reject --------------------------------------------------------------
@@ -0,0 +1,127 @@
"""
Vendor-Banner Sync maps the 82-service registry to banner vendor configs.
Automatically creates vendor entries in the cookie banner with correct
category assignment and legally required retention periods.
"""
import logging
import os
import uuid
logger = logging.getLogger(__name__)
# Service category → Banner category mapping
CATEGORY_MAP = {
"tracking": "statistics",
"heatmap": "statistics",
"tag_manager": "statistics",
"marketing": "marketing",
"social": "marketing",
"push": "marketing",
"crm": "marketing",
"chatbot": "functional",
"support": "functional",
"video": "functional",
"testing": "functional",
"cdn": "necessary",
"payment": "necessary",
"error_tracking": "necessary",
"accessibility": "necessary",
"cmp": "necessary",
"other": "functional",
}
# Legally required max retention per category (in days)
# Based on: DSGVO Art. 5(1)(e), CNIL guidelines, EDPB recommendations
RETENTION_DEFAULTS = {
"necessary": 365, # Session + functional = max 12 months
"statistics": 790, # Max 26 months (Google Analytics default)
"marketing": 90, # Max 90 days for retargeting
"functional": 365, # Max 12 months
}
# Specific service retention overrides
SERVICE_RETENTION = {
"google_analytics": 790, # 26 months (GA4 default)
"matomo": 790, # 26 months
"hotjar": 365, # 12 months
"facebook_pixel": 90, # 90 days (Meta default)
"google_ads": 90, # 90 days
"stripe": 0, # Session only (payment)
"paypal": 0, # Session only
"klarna": 0, # Session only
}
def get_banner_vendors_from_registry() -> list[dict]:
"""Convert service registry entries to banner vendor configs."""
from compliance.services.service_registry import SERVICE_REGISTRY
vendors = []
for pattern, meta in SERVICE_REGISTRY.items():
service_id = meta.get("id", "")
category = meta.get("category", "other")
banner_category = CATEGORY_MAP.get(category, "functional")
# Skip CMP — consent managers are not vendor entries
if service_id == "cmp":
continue
retention = SERVICE_RETENTION.get(service_id, RETENTION_DEFAULTS.get(banner_category, 365))
vendors.append({
"vendor_name": meta["name"],
"vendor_url": "", # Would need manual entry
"category_key": banner_category,
"description_de": f"{meta['name']} ({meta.get('provider', '')})",
"description_en": f"{meta['name']} ({meta.get('provider', '')})",
"cookie_names": [], # Service-specific, populated later
"retention_days": retention,
"is_active": True,
"country": meta.get("country", ""),
"eu_adequate": meta.get("eu_adequate", False),
"requires_consent": meta.get("requires_consent", True),
"legal_ref": meta.get("legal_ref", ""),
"service_id": service_id,
})
logger.info("Generated %d banner vendors from service registry", len(vendors))
return vendors
async def sync_vendors_to_site(pool, site_config_id: str, tenant_id: str) -> dict:
"""Sync service registry vendors to a site's banner vendor configs."""
vendors = get_banner_vendors_from_registry()
created = 0
updated = 0
async with pool.acquire() as conn:
for v in vendors:
# Check if vendor already exists for this site
existing = await conn.fetchrow("""
SELECT id FROM compliance_banner_vendor_configs
WHERE site_config_id = $1 AND vendor_name = $2
""", uuid.UUID(site_config_id), v["vendor_name"])
if existing:
await conn.execute("""
UPDATE compliance_banner_vendor_configs
SET category_key = $1, retention_days = $2, is_active = $3
WHERE id = $4
""", v["category_key"], v["retention_days"], v["is_active"], existing["id"])
updated += 1
else:
import json
await conn.execute("""
INSERT INTO compliance_banner_vendor_configs
(site_config_id, vendor_name, category_key, description_de,
description_en, cookie_names, retention_days, is_active)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
""", uuid.UUID(site_config_id), v["vendor_name"], v["category_key"],
v["description_de"], v["description_en"],
json.dumps(v["cookie_names"]), v["retention_days"], v["is_active"])
created += 1
logger.info("Synced vendors to site %s: %d created, %d updated", site_config_id, created, updated)
return {"created": created, "updated": updated, "total": len(vendors)}
@@ -0,0 +1,23 @@
-- Migration 106: Banner Email Linking + Consent Proof
-- Phase 3: linked_email for DSR ↔ Banner-Consent correlation
-- Phase 6: banner_config_hash + consent_version for Art. 7(1) DSGVO proof
-- 1. Add linked_email to banner consents (optional, nullable)
-- Allows correlating device-based consents with user email for DSR processing
ALTER TABLE compliance_banner_consents
ADD COLUMN IF NOT EXISTS linked_email TEXT;
CREATE INDEX IF NOT EXISTS idx_banner_consent_email
ON compliance_banner_consents (linked_email)
WHERE linked_email IS NOT NULL;
-- 2. Add consent proof columns to audit log
-- banner_config_hash: SHA256 of the site config at consent time (Art. 7(1) DSGVO)
-- consent_version: incremented per site on config change, tracks which banner version was shown
ALTER TABLE compliance_banner_consent_audit_log
ADD COLUMN IF NOT EXISTS banner_config_hash TEXT,
ADD COLUMN IF NOT EXISTS consent_version INTEGER;
-- 3. Add config_version counter to site configs (auto-incremented on config change)
ALTER TABLE compliance_banner_site_configs
ADD COLUMN IF NOT EXISTS config_version INTEGER NOT NULL DEFAULT 1;
@@ -312,3 +312,225 @@ class TestStats:
assert data["category_acceptance"]["necessary"]["count"] == 3
assert data["category_acceptance"]["analytics"]["count"] == 2
assert data["category_acceptance"]["marketing"]["count"] == 1
# =============================================================================
# Phase 3: Email Linking
# =============================================================================
class TestEmailLinking:
def test_link_email(self):
_record_consent()
r = client.post("/api/compliance/banner/consent/link-email", json={
"site_id": "example.com",
"device_fingerprint": "fp-123",
"email": "user@example.com",
}, headers=HEADERS)
assert r.status_code == 200
assert r.json()["linked_email"] == "user@example.com"
def test_link_email_normalizes(self):
_record_consent()
r = client.post("/api/compliance/banner/consent/link-email", json={
"site_id": "example.com",
"device_fingerprint": "fp-123",
"email": " User@Example.COM ",
}, headers=HEADERS)
assert r.status_code == 200
assert r.json()["linked_email"] == "user@example.com"
def test_link_email_invalid(self):
_record_consent()
r = client.post("/api/compliance/banner/consent/link-email", json={
"site_id": "example.com",
"device_fingerprint": "fp-123",
"email": "not-an-email",
}, headers=HEADERS)
assert r.status_code == 400
def test_link_email_no_consent(self):
r = client.post("/api/compliance/banner/consent/link-email", json={
"site_id": "example.com",
"device_fingerprint": "nonexistent",
"email": "user@example.com",
}, headers=HEADERS)
assert r.status_code == 404
def test_get_consents_by_email(self):
_record_consent(fingerprint="dev-a")
_record_consent(fingerprint="dev-b")
# Link both to same email
for fp in ["dev-a", "dev-b"]:
client.post("/api/compliance/banner/consent/link-email", json={
"site_id": "example.com",
"device_fingerprint": fp,
"email": "multi@example.com",
}, headers=HEADERS)
r = client.get("/api/compliance/banner/consent/by-email/multi@example.com", headers=HEADERS)
assert r.status_code == 200
assert len(r.json()) == 2
def test_get_consents_by_email_empty(self):
r = client.get("/api/compliance/banner/consent/by-email/nobody@nowhere.com", headers=HEADERS)
assert r.status_code == 200
assert r.json() == []
def test_delete_consents_by_email_art17(self):
_record_consent(fingerprint="del-a")
_record_consent(fingerprint="del-b")
for fp in ["del-a", "del-b"]:
client.post("/api/compliance/banner/consent/link-email", json={
"site_id": "example.com",
"device_fingerprint": fp,
"email": "erasure@example.com",
}, headers=HEADERS)
r = client.delete("/api/compliance/banner/consent/by-email/erasure@example.com", headers=HEADERS)
assert r.status_code == 200
data = r.json()
assert data["deleted"] == 2
assert data["email"] == "erasure@example.com"
# Verify gone
r2 = client.get("/api/compliance/banner/consent/by-email/erasure@example.com", headers=HEADERS)
assert r2.json() == []
def test_dsr_export_by_email(self):
_record_consent(fingerprint="exp-1")
client.post("/api/compliance/banner/consent/link-email", json={
"site_id": "example.com",
"device_fingerprint": "exp-1",
"email": "export@example.com",
}, headers=HEADERS)
r = client.get("/api/compliance/banner/consent/dsr-export/export@example.com", headers=HEADERS)
assert r.status_code == 200
data = r.json()
assert data["email"] == "export@example.com"
assert len(data["banner_consents"]) == 1
assert len(data["audit_trail"]) >= 1
# =============================================================================
# Phase 4: Consent Sync (Banner → Einwilligungen)
# =============================================================================
class TestConsentSync:
def test_sync_consent(self):
_record_consent(categories=["necessary", "analytics"])
r = client.post("/api/compliance/banner/consent/sync", json={
"site_id": "example.com",
"device_fingerprint": "fp-123",
"email": "sync@example.com",
}, headers=HEADERS)
assert r.status_code == 200
data = r.json()
assert data["synced"] >= 1
assert "necessary" in data["categories"]
assert data["email"] == "sync@example.com"
def test_sync_consent_links_email(self):
_record_consent()
# Sync should auto-link email
client.post("/api/compliance/banner/consent/sync", json={
"site_id": "example.com",
"device_fingerprint": "fp-123",
"email": "autolink@example.com",
}, headers=HEADERS)
r = client.get("/api/compliance/banner/consent/by-email/autolink@example.com", headers=HEADERS)
assert len(r.json()) == 1
def test_sync_no_consent(self):
r = client.post("/api/compliance/banner/consent/sync", json={
"site_id": "example.com",
"device_fingerprint": "nonexistent",
"email": "test@example.com",
}, headers=HEADERS)
assert r.status_code == 404
# =============================================================================
# Phase 2: Retention per Category
# =============================================================================
class TestRetention:
def test_retention_uses_vendor_max(self):
"""Consent expiry should use max vendor retention, not hardcoded 365."""
_create_site()
# Add marketing vendor with 90 days retention
client.post("/api/compliance/banner/admin/sites/example.com/vendors", json={
"vendor_name": "FB Pixel",
"category_key": "marketing",
"retention_days": 90,
}, headers=HEADERS)
c = _record_consent(categories=["necessary", "marketing"])
# Consent should expire in 90 days (max of vendor retentions)
from datetime import datetime
created = datetime.fromisoformat(c["created_at"])
expires = datetime.fromisoformat(c["expires_at"])
diff_days = (expires - created).days
assert 89 <= diff_days <= 91, f"Expected ~90 days, got {diff_days}"
def test_retention_default_365(self):
"""Without vendor config, should use category default."""
c = _record_consent(categories=["necessary"])
from datetime import datetime
created = datetime.fromisoformat(c["created_at"])
expires = datetime.fromisoformat(c["expires_at"])
diff_days = (expires - created).days
assert 364 <= diff_days <= 366, f"Expected ~365 days, got {diff_days}"
# =============================================================================
# Phase 6: Consent Proof (Art. 7(1) DSGVO)
# =============================================================================
class TestConsentProof:
def test_consent_has_linked_email_field(self):
"""New consent should include linked_email in response."""
c = _record_consent()
assert "linked_email" in c
assert c["linked_email"] is None # Not linked yet
def test_site_config_has_version(self):
"""Site config should have config_version field."""
s = _create_site()
assert "config_version" in s
assert s["config_version"] == 1
def test_audit_trail_after_consent(self):
"""Audit trail should exist after recording consent."""
_record_consent()
r = client.get(
"/api/compliance/banner/consent/export?site_id=example.com&device_fingerprint=fp-123",
headers=HEADERS,
)
data = r.json()
assert len(data["audit_trail"]) >= 1
audit = data["audit_trail"][0]
assert audit["action"] in ("consent_given", "consent_updated")
# =============================================================================
# IP Hashing
# =============================================================================
class TestIPHashing:
def test_ip_is_hashed(self):
"""IP address should never be stored plain — only SHA256[:16]."""
c = _record_consent()
assert c["ip_hash"] is not None
assert c["ip_hash"] != "1.2.3.4"
assert len(c["ip_hash"]) == 16
def test_no_ip_returns_none(self):
r = client.post("/api/compliance/banner/consent", json={
"site_id": "example.com",
"device_fingerprint": "fp-no-ip",
"categories": ["necessary"],
}, headers=HEADERS)
assert r.status_code == 200
assert r.json()["ip_hash"] is None