Artikel
23. Sicherheits- und Sperrlogik
Die Sicherheits- und Sperrlogik verhindert, dass die Automatisierung bei ungewöhnlichen Zuständen weiterarbeitet.
Grundregel:
Wenn die Plattform anders reagiert als erwartet, stoppt das System.
Nicht zulässig:
CAPTCHA umgehen
Login-Probleme automatisch bekämpfen
Sicherheitswarnungen wegklicken
mehrfach blind neu versuchen
UI-Fehler mit Klickketten überbrücken
Account-Sperren ignorieren
Die Sicherheitslogik sitzt vor und während jeder Plattformaktion.
Scheduler
↓
SafetyGuard
↓
Publisher
↓
PlatformAdapter
↓
SafetyGuard
↓
Status: published / failed / blocked
23.1 Globaler Kill-Switch
Der globale Kill-Switch deaktiviert das gesamte System.
Konfiguration:
kill_switch:
enabled: false
reason: ""
Wenn enabled: true gesetzt ist, darf keine Veröffentlichung, kein Like, kein Kommentar und keine Interaktion ausgeführt werden.
kill_switch:
enabled: true
reason: "Manuelle Prüfung nach LinkedIn-Sicherheitswarnung."
Prüfung:
class GlobalKillSwitch:
def assert_disabled(self, raw_config: dict[str, object]) -> None:
kill_switch = raw_config.get("kill_switch")
if isinstance(kill_switch, dict) is False:
raise SafetyGuardError("Missing kill_switch configuration.")
enabled = kill_switch.get("enabled")
reason = kill_switch.get("reason")
if isinstance(enabled, bool) is False:
raise SafetyGuardError("kill_switch.enabled must be boolean.")
if isinstance(reason, str) is False:
raise SafetyGuardError("kill_switch.reason must be string.")
if enabled is True:
raise SafetyGuardBlockedError("Global kill-switch enabled: " + reason)
Der globale Kill-Switch ist die einfachste Notbremse.
23.2 Plattform-Kill-Switch
Der Plattform-Kill-Switch deaktiviert nur eine einzelne Plattform.
Konfiguration:
platform_kill_switch:
linkedin:
enabled: false
reason: ""
instagram:
enabled: false
reason: ""
facebook:
enabled: false
reason: ""
Beispiel:
platform_kill_switch:
linkedin:
enabled: true
reason: "LinkedIn zeigt Sicherheitsprüfung."
instagram:
enabled: false
reason: ""
facebook:
enabled: false
reason: ""
Prüfung:
class PlatformKillSwitch:
def assert_disabled(self, raw_config: dict[str, object], platform: str) -> None:
platform_kill_switch = raw_config.get("platform_kill_switch")
if isinstance(platform_kill_switch, dict) is False:
raise SafetyGuardError("Missing platform_kill_switch configuration.")
platform_config = platform_kill_switch.get(platform)
if isinstance(platform_config, dict) is False:
raise SafetyGuardError("Missing platform kill-switch: " + platform)
enabled = platform_config.get("enabled")
reason = platform_config.get("reason")
if isinstance(enabled, bool) is False:
raise SafetyGuardError("platform_kill_switch.enabled must be boolean for " + platform)
if isinstance(reason, str) is False:
raise SafetyGuardError("platform_kill_switch.reason must be string for " + platform)
if enabled is True:
raise SafetyGuardBlockedError(
"Platform kill-switch enabled for "
+ platform
+ ": "
+ reason
)
23.3 Account-Lock bei Fehlern
Ein Account-Lock wird gesetzt, wenn zu viele Fehler auftreten.
Typische Auslöser:
mehrere fehlgeschlagene Veröffentlichungen
wiederholte Timeouts
mehrfach Login erforderlich
Plattformwarnung erkannt
CAPTCHA erkannt
unerwartete Weiterleitung
Konfiguration:
safety:
stop_after_consecutive_failures: 2
disable_platform_after_daily_failures: 3
Fehlerzähler:
from datetime import datetime
from zoneinfo import ZoneInfo
class PlatformFailureCounter:
def __init__(self, timezone: str) -> None:
self.timezone = timezone
def count_failures_today(
self,
posts: list[dict[str, object]],
platform: str,
) -> int:
today = datetime.now(ZoneInfo(self.timezone)).date().isoformat()
count = 0
for post in posts:
if post.get("platform") != platform:
continue
failed_at = post.get("failed_at")
blocked_at = post.get("blocked_at")
if isinstance(failed_at, str) is True and failed_at.startswith(today):
count += 1
continue
if isinstance(blocked_at, str) is True and blocked_at.startswith(today):
count += 1
continue
return count
Lock-Prüfung:
class AccountLockPolicy:
def assert_failure_limit_not_reached(
self,
failure_count: int,
max_failures: int,
platform: str,
) -> None:
if failure_count >= max_failures:
raise SafetyGuardBlockedError(
"Daily failure limit reached for platform: " + platform
)
Im MVP wird der Account-Lock nicht automatisch in die YAML geschrieben. Der Scheduler blockiert die Plattform für den aktuellen Lauf. Später kann ein Laufzeitstatus gespeichert werden.
{
"platform": "linkedin",
"status": "locked",
"reason": "Daily failure limit reached.",
"locked_at": "2026-05-12T10:00:00+02:00"
}
23.4 Stop bei CAPTCHA
CAPTCHA ist ein harter Stop.
Regel:
CAPTCHA wird nicht gelöst.
CAPTCHA wird nicht umgangen.
CAPTCHA setzt Status blocked.
Screenshot wird gespeichert.
Plattform wird nicht weiter bearbeitet.
Erkennung über URL und sichtbaren Text:
class CaptchaDetector:
def detect(self, url: str, body_text: str) -> bool:
lowered_url = url.lower()
lowered_body_text = body_text.lower()
fragments = [
"captcha",
"security check",
"sicherheitsüberprüfung",
"verify you are human",
"bestätigen sie, dass sie ein mensch sind",
]
for fragment in fragments:
if fragment in lowered_url:
return True
if fragment in lowered_body_text:
return True
return False
Verwendung:
if captcha_detector.detect(page.url, page.locator("body").inner_text(timeout=10_000)) is True:
raise SafetyGuardBlockedError("CAPTCHA detected.")
23.5 Stop bei Login-Problemen
Wenn eine Plattform zur Login-Seite weiterleitet, ist die Session abgelaufen oder ungültig.
Regel:
nicht automatisch neu einloggen
nicht Passwort eintragen
nicht 2FA automatisieren
Status blocked
Screenshot speichern
manueller Session-Refresh
Detector:
class LoginRequiredDetector:
def detect(self, url: str, body_text: str) -> bool:
lowered_url = url.lower()
lowered_body_text = body_text.lower()
url_fragments = [
"login",
"signin",
"checkpoint",
]
body_fragments = [
"einloggen",
"anmelden",
"log in",
"sign in",
"passwort",
"password",
]
for fragment in url_fragments:
if fragment in lowered_url:
return True
for fragment in body_fragments:
if fragment in lowered_body_text:
return True
return False
Behandlung:
if login_required_detector.detect(current_url, body_text) is True:
raise SafetyGuardBlockedError("Login required.")
23.6 Stop bei ungewöhnlicher Plattformreaktion
Ungewöhnliche Plattformreaktionen sind Zustände, die nicht eindeutig als Erfolg, normaler Fehler oder Login erkannt werden.
Beispiele:
unerwartete Weiterleitung
leerer Feed
ungewöhnlicher Dialog
Button fehlt
Post-Dialog öffnet nicht
Upload verschwindet
Seite zeigt Warntext
URL enthält challenge oder checkpoint
Detector:
class PlatformAnomalyDetector:
def detect(self, url: str, body_text: str) -> bool:
lowered_url = url.lower()
lowered_body_text = body_text.lower()
url_fragments = [
"checkpoint",
"challenge",
"recover",
"restricted",
"help/contact",
]
body_fragments = [
"ungewöhnliche aktivität",
"unusual activity",
"account restricted",
"konto eingeschränkt",
"temporarily blocked",
"vorübergehend blockiert",
"try again later",
"versuchen sie es später erneut",
]
for fragment in url_fragments:
if fragment in lowered_url:
return True
for fragment in body_fragments:
if fragment in lowered_body_text:
return True
return False
Behandlung:
Screenshot speichern
Status blocked
Fehler loggen
keine weiteren Aktionen
23.7 Keine Umgehungslogik
Dieses System enthält ausdrücklich keine Umgehungslogik.
Nicht implementieren:
CAPTCHA-Solver
Proxy-Rotation
Account-Rotation
Fingerprint-Spoofing
zufällige Mausbewegungen
zufällige Tippfehler
Anti-Bot-Stealth als Kernlogik
automatisches Wegklicken von Sicherheitswarnungen
aggressives Retry
Zulässig:
normale Browser-Session
manueller Login
storage_state
Dry-Run
Screenshots
Timeouts
saubere Stop-Regeln
begrenzte Tageslimits
Die Maschine soll wiederkehrende Handarbeit übernehmen, aber nicht gegen Schutzmechanismen arbeiten.
23.8 Beispiel: safety_guard.py
Datei:
src/safety_guard.py
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo
from playwright.sync_api import Page
class SafetyGuardError(Exception):
pass
class SafetyGuardBlockedError(SafetyGuardError):
pass
@dataclass(frozen=True)
class PlatformState:
platform: str
url: str
title: str
body_text: str
@dataclass(frozen=True)
class SafetyCheckResult:
passed: bool
reason: str | None
class GlobalKillSwitch:
def assert_disabled(self, raw_config: dict[str, Any]) -> None:
kill_switch = raw_config.get("kill_switch")
if isinstance(kill_switch, dict) is False:
raise SafetyGuardError("Missing kill_switch configuration.")
enabled = kill_switch.get("enabled")
reason = kill_switch.get("reason")
if isinstance(enabled, bool) is False:
raise SafetyGuardError("kill_switch.enabled must be boolean.")
if isinstance(reason, str) is False:
raise SafetyGuardError("kill_switch.reason must be string.")
if enabled is True:
raise SafetyGuardBlockedError("Global kill-switch enabled: " + reason)
class PlatformKillSwitch:
def assert_disabled(self, raw_config: dict[str, Any], platform: str) -> None:
platform_kill_switch = raw_config.get("platform_kill_switch")
if isinstance(platform_kill_switch, dict) is False:
raise SafetyGuardError("Missing platform_kill_switch configuration.")
platform_config = platform_kill_switch.get(platform)
if isinstance(platform_config, dict) is False:
raise SafetyGuardError("Missing platform kill-switch: " + platform)
enabled = platform_config.get("enabled")
reason = platform_config.get("reason")
if isinstance(enabled, bool) is False:
raise SafetyGuardError("platform_kill_switch.enabled must be boolean for " + platform)
if isinstance(reason, str) is False:
raise SafetyGuardError("platform_kill_switch.reason must be string for " + platform)
if enabled is True:
raise SafetyGuardBlockedError(
"Platform kill-switch enabled for "
+ platform
+ ": "
+ reason
)
class CaptchaDetector:
def detect(self, platform_state: PlatformState) -> bool:
lowered_url = platform_state.url.lower()
lowered_body_text = platform_state.body_text.lower()
fragments = [
"captcha",
"security check",
"sicherheitsüberprüfung",
"verify you are human",
"bestätigen sie, dass sie ein mensch sind",
]
for fragment in fragments:
if fragment in lowered_url:
return True
if fragment in lowered_body_text:
return True
return False
class LoginRequiredDetector:
def detect(self, platform_state: PlatformState) -> bool:
lowered_url = platform_state.url.lower()
lowered_body_text = platform_state.body_text.lower()
url_fragments = [
"login",
"signin",
"checkpoint",
]
body_fragments = [
"einloggen",
"anmelden",
"log in",
"sign in",
"passwort",
"password",
]
for fragment in url_fragments:
if fragment in lowered_url:
return True
for fragment in body_fragments:
if fragment in lowered_body_text:
return True
return False
class PlatformAnomalyDetector:
def detect(self, platform_state: PlatformState) -> bool:
lowered_url = platform_state.url.lower()
lowered_body_text = platform_state.body_text.lower()
url_fragments = [
"checkpoint",
"challenge",
"recover",
"restricted",
"help/contact",
]
body_fragments = [
"ungewöhnliche aktivität",
"unusual activity",
"account restricted",
"konto eingeschränkt",
"temporarily blocked",
"vorübergehend blockiert",
"try again later",
"versuchen sie es später erneut",
]
for fragment in url_fragments:
if fragment in lowered_url:
return True
for fragment in body_fragments:
if fragment in lowered_body_text:
return True
return False
class PlatformFailureCounter:
def __init__(self, timezone: str) -> None:
self.timezone = timezone
def count_failures_today(
self,
posts: list[dict[str, Any]],
platform: str,
) -> int:
today = datetime.now(ZoneInfo(self.timezone)).date().isoformat()
count = 0
for post in posts:
if post.get("platform") != platform:
continue
failed_at = post.get("failed_at")
blocked_at = post.get("blocked_at")
if isinstance(failed_at, str) is True and failed_at.startswith(today):
count += 1
continue
if isinstance(blocked_at, str) is True and blocked_at.startswith(today):
count += 1
continue
return count
class AccountLockPolicy:
def assert_failure_limit_not_reached(
self,
failure_count: int,
max_failures: int,
platform: str,
) -> None:
if failure_count >= max_failures:
raise SafetyGuardBlockedError(
"Daily failure limit reached for platform: " + platform
)
class PlatformStateReader:
def read(self, page: Page, platform: str) -> PlatformState:
title = page.title()
body_text = page.locator("body").inner_text(timeout=10_000)
return PlatformState(
platform=platform,
url=page.url,
title=title,
body_text=body_text,
)
class SafetyGuard:
def __init__(self, timezone: str) -> None:
self.timezone = timezone
self.global_kill_switch = GlobalKillSwitch()
self.platform_kill_switch = PlatformKillSwitch()
self.platform_state_reader = PlatformStateReader()
self.captcha_detector = CaptchaDetector()
self.login_required_detector = LoginRequiredDetector()
self.platform_anomaly_detector = PlatformAnomalyDetector()
self.platform_failure_counter = PlatformFailureCounter(timezone)
self.account_lock_policy = AccountLockPolicy()
def assert_runtime_allowed(
self,
raw_config: dict[str, Any],
posts: list[dict[str, Any]],
platform: str,
) -> None:
self.global_kill_switch.assert_disabled(raw_config)
self.platform_kill_switch.assert_disabled(raw_config, platform)
max_failures = self._read_max_failures(raw_config)
failure_count = self.platform_failure_counter.count_failures_today(
posts=posts,
platform=platform,
)
self.account_lock_policy.assert_failure_limit_not_reached(
failure_count=failure_count,
max_failures=max_failures,
platform=platform,
)
def assert_page_state_allowed(self, page: Page, platform: str) -> None:
platform_state = self.platform_state_reader.read(page, platform)
if self.captcha_detector.detect(platform_state) is True:
raise SafetyGuardBlockedError("CAPTCHA detected.")
if self.login_required_detector.detect(platform_state) is True:
raise SafetyGuardBlockedError("Login required.")
if self.platform_anomaly_detector.detect(platform_state) is True:
raise SafetyGuardBlockedError("Unusual platform reaction detected.")
def check_page_state(self, page: Page, platform: str) -> SafetyCheckResult:
try:
self.assert_page_state_allowed(page, platform)
return SafetyCheckResult(
passed=True,
reason=None,
)
except SafetyGuardBlockedError as exception:
return SafetyCheckResult(
passed=False,
reason=str(exception),
)
def _read_max_failures(self, raw_config: dict[str, Any]) -> int:
safety_config = raw_config.get("safety")
if isinstance(safety_config, dict) is False:
raise SafetyGuardError("Missing safety configuration.")
value = safety_config.get("disable_platform_after_daily_failures")
if isinstance(value, int) is False:
raise SafetyGuardError("safety.disable_platform_after_daily_failures must be integer.")
if value <= 0:
raise SafetyGuardError("safety.disable_platform_after_daily_failures must be greater than zero.")
return value
class SafetyStatusApplier:
def apply_blocked_status(
self,
post: dict[str, Any],
reason: str,
screenshot_path: Path | None,
) -> dict[str, Any]:
post["status"] = "blocked"
post["blocked_at"] = datetime.now(ZoneInfo("Europe/Berlin")).isoformat()
post["error_message"] = reason
if screenshot_path is not None:
post["screenshot_path"] = str(screenshot_path)
return post
Verwendung im Publisher
Ausschnitt:
from src.safety_guard import SafetyGuard
from src.safety_guard import SafetyGuardBlockedError
safety_guard = SafetyGuard(timezone="Europe/Berlin")
safety_guard.assert_runtime_allowed(
raw_config=raw_config,
posts=posts,
platform=request.platform,
)
page.goto(
"https://www.linkedin.com/feed/",
wait_until="domcontentloaded",
timeout=30_000,
)
try:
safety_guard.assert_page_state_allowed(
page=page,
platform=request.platform,
)
except SafetyGuardBlockedError as exception:
screenshot_path = screenshot_service.save(
page=page,
screenshot_dir=request.screenshot_dir,
post_id=request.id,
status="blocked",
)
return PublishResult(
post_id=request.id,
platform=request.platform,
status="blocked",
screenshot_path=str(screenshot_path),
platform_post_url=None,
error_message=str(exception),
)
Ergebnis dieses Kapitels
Die Sicherheitslogik stellt bereit:
globalen Kill-Switch
Plattform-Kill-Switch
Account-Lock bei Fehlerhäufung
Stop bei CAPTCHA
Stop bei Login-Problemen
Stop bei ungewöhnlichen Plattformreaktionen
expliziten Verzicht auf Umgehungslogik
zentrale SafetyGuard-Klasse
Status blocked für kritische Zustände
Damit arbeitet die Automatisierung nicht gegen Plattformsignale, sondern beendet den Lauf kontrolliert.
0 Kommentare