f3e44cf59f
banner_detector.py, script_analyzer.py, category_tester.py, authenticated_scanner.py were only on the feature branch — needed for consent-tester to start. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
231 lines
7.9 KiB
Python
231 lines
7.9 KiB
Python
"""
|
|
Authenticated Scanner — tests post-login functionality.
|
|
|
|
Checks §312k BGB (cancellation), Art. 17 (deletion), Art. 20 (export),
|
|
Art. 7(3) (consent withdrawal), Art. 15 (data access).
|
|
|
|
Credentials are NEVER stored, logged, or transmitted beyond the browser context.
|
|
"""
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
|
|
from playwright.async_api import async_playwright, Page
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
USER_AGENT = (
|
|
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class CheckResult:
|
|
found: bool = False
|
|
selector: str = ""
|
|
text: str = ""
|
|
clicks_needed: int = 0
|
|
screenshot: bytes = b""
|
|
|
|
|
|
@dataclass
|
|
class AuthTestResult:
|
|
authenticated: bool = False
|
|
login_error: str = ""
|
|
cancel_subscription: CheckResult = field(default_factory=CheckResult)
|
|
delete_account: CheckResult = field(default_factory=CheckResult)
|
|
export_data: CheckResult = field(default_factory=CheckResult)
|
|
consent_settings: CheckResult = field(default_factory=CheckResult)
|
|
profile_visible: CheckResult = field(default_factory=CheckResult)
|
|
|
|
|
|
# Search patterns for each check (DE + EN)
|
|
CANCEL_PATTERNS = [
|
|
"kündigen", "kuendigen", "vertrag beenden", "abo beenden",
|
|
"mitgliedschaft kündigen", "cancel subscription", "unsubscribe",
|
|
"cancel membership", "vertrag kündigen",
|
|
]
|
|
|
|
DELETE_PATTERNS = [
|
|
"konto löschen", "konto loeschen", "account löschen", "delete account",
|
|
"account deaktivieren", "profil löschen", "remove account",
|
|
]
|
|
|
|
EXPORT_PATTERNS = [
|
|
"daten exportieren", "daten herunterladen", "export data", "download data",
|
|
"meine daten", "datenauskunft", "data download", "daten anfordern",
|
|
]
|
|
|
|
CONSENT_PATTERNS = [
|
|
"einwilligung", "einstellungen", "datenschutz-einstellungen",
|
|
"consent", "privacy settings", "cookie-einstellungen",
|
|
"werbeeinstellungen", "marketing preferences",
|
|
]
|
|
|
|
PROFILE_PATTERNS = [
|
|
"profil", "mein konto", "kontodaten", "persönliche daten",
|
|
"profile", "my account", "account settings", "personal data",
|
|
]
|
|
|
|
|
|
async def run_authenticated_test(
|
|
url: str,
|
|
username: str,
|
|
password: str,
|
|
username_selector: str = "",
|
|
password_selector: str = "",
|
|
submit_selector: str = "",
|
|
) -> AuthTestResult:
|
|
"""Run authenticated area test. Credentials are destroyed after test."""
|
|
result = AuthTestResult()
|
|
|
|
async with async_playwright() as p:
|
|
browser = await p.chromium.launch(
|
|
headless=True,
|
|
args=["--no-sandbox", "--disable-dev-shm-usage"],
|
|
)
|
|
context = await browser.new_context(user_agent=USER_AGENT)
|
|
page = await context.new_page()
|
|
|
|
try:
|
|
# Step 1: Login
|
|
await page.goto(url, wait_until="networkidle", timeout=30000)
|
|
await page.wait_for_timeout(2000)
|
|
|
|
login_ok = await _try_login(
|
|
page, username, password,
|
|
username_selector, password_selector, submit_selector,
|
|
)
|
|
|
|
if not login_ok:
|
|
result.login_error = "Login fehlgeschlagen — Formular nicht gefunden oder Credentials falsch"
|
|
await context.close()
|
|
await browser.close()
|
|
return result
|
|
|
|
result.authenticated = True
|
|
await page.wait_for_timeout(3000)
|
|
|
|
# Step 2: Check cancellation (§312k BGB)
|
|
result.cancel_subscription = await _check_patterns(page, CANCEL_PATTERNS, "cancel")
|
|
logger.info("Cancel check: found=%s", result.cancel_subscription.found)
|
|
|
|
# Step 3: Check delete account (Art. 17 DSGVO)
|
|
result.delete_account = await _check_patterns(page, DELETE_PATTERNS, "delete")
|
|
|
|
# Step 4: Check data export (Art. 20 DSGVO)
|
|
result.export_data = await _check_patterns(page, EXPORT_PATTERNS, "export")
|
|
|
|
# Step 5: Check consent settings (Art. 7(3) DSGVO)
|
|
result.consent_settings = await _check_patterns(page, CONSENT_PATTERNS, "consent")
|
|
|
|
# Step 6: Check profile visibility (Art. 15 DSGVO)
|
|
result.profile_visible = await _check_patterns(page, PROFILE_PATTERNS, "profile")
|
|
|
|
except Exception as e:
|
|
logger.error("Authenticated test failed: %s", e)
|
|
result.login_error = str(e)
|
|
finally:
|
|
# CRITICAL: Destroy context — wipes all credentials, cookies, session
|
|
await context.close()
|
|
await browser.close()
|
|
|
|
return result
|
|
|
|
|
|
async def _try_login(
|
|
page: Page, username: str, password: str,
|
|
user_sel: str, pass_sel: str, submit_sel: str,
|
|
) -> bool:
|
|
"""Attempt to fill and submit login form."""
|
|
try:
|
|
# Auto-detect selectors if not provided
|
|
if not user_sel:
|
|
for sel in ['input[type="email"]', 'input[name="email"]', 'input[name="username"]',
|
|
'input[name="login"]', 'input[id="email"]', 'input[id="username"]']:
|
|
if await page.locator(sel).count() > 0:
|
|
user_sel = sel
|
|
break
|
|
if not pass_sel:
|
|
for sel in ['input[type="password"]', 'input[name="password"]', 'input[id="password"]']:
|
|
if await page.locator(sel).count() > 0:
|
|
pass_sel = sel
|
|
break
|
|
if not submit_sel:
|
|
for sel in ['button[type="submit"]', 'input[type="submit"]',
|
|
'button:has-text("Anmelden")', 'button:has-text("Login")',
|
|
'button:has-text("Sign in")', 'button:has-text("Einloggen")']:
|
|
if await page.locator(sel).count() > 0:
|
|
submit_sel = sel
|
|
break
|
|
|
|
if not user_sel or not pass_sel:
|
|
return False
|
|
|
|
await page.fill(user_sel, username)
|
|
await page.fill(pass_sel, password)
|
|
|
|
if submit_sel:
|
|
await page.click(submit_sel)
|
|
else:
|
|
await page.press(pass_sel, "Enter")
|
|
|
|
await page.wait_for_timeout(5000)
|
|
|
|
# Check if login succeeded (URL changed or login form disappeared)
|
|
still_on_login = await page.locator('input[type="password"]').count() > 0
|
|
return not still_on_login
|
|
|
|
except Exception as e:
|
|
logger.warning("Login attempt failed: %s", e)
|
|
return False
|
|
|
|
|
|
async def _check_patterns(page: Page, patterns: list[str], check_name: str) -> CheckResult:
|
|
"""Search current page and navigation for patterns."""
|
|
result = CheckResult()
|
|
|
|
# Check current page text
|
|
for pattern in patterns:
|
|
try:
|
|
locator = page.get_by_text(pattern, exact=False)
|
|
count = await locator.count()
|
|
if count > 0:
|
|
text = await locator.first.text_content()
|
|
result.found = True
|
|
result.text = (text or "").strip()[:100]
|
|
return result
|
|
except Exception:
|
|
continue
|
|
|
|
# Check links/buttons
|
|
for pattern in patterns:
|
|
try:
|
|
for sel in [f'a:has-text("{pattern}")', f'button:has-text("{pattern}")',
|
|
f'[href*="{pattern.replace(" ", "-")}"]']:
|
|
locator = page.locator(sel)
|
|
if await locator.count() > 0:
|
|
result.found = True
|
|
result.selector = sel
|
|
result.text = pattern
|
|
return result
|
|
except Exception:
|
|
continue
|
|
|
|
# Check navigation menus (common locations for account management)
|
|
for nav_sel in ['nav', '[role="navigation"]', '.sidebar', '.account-menu', '#account']:
|
|
try:
|
|
nav = page.locator(nav_sel)
|
|
if await nav.count() > 0:
|
|
nav_text = (await nav.first.text_content() or "").lower()
|
|
for pattern in patterns:
|
|
if pattern.lower() in nav_text:
|
|
result.found = True
|
|
result.text = f"In Navigation: {pattern}"
|
|
return result
|
|
except Exception:
|
|
continue
|
|
|
|
return result
|