66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
"""Local privacy gateway stub. Personal LLM egress is not allowed to skip this layer."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
|
|
class PrivacyGatewayError(Exception):
|
|
def __init__(self, code: str, message: str, status_code: int = 503):
|
|
super().__init__(message)
|
|
self.code = code
|
|
self.message = message
|
|
self.status_code = status_code
|
|
|
|
|
|
@dataclass
|
|
class GatewayRequest:
|
|
prompt_id: str | None
|
|
purpose: str
|
|
data_class: str
|
|
payload: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class GatewayResult:
|
|
allowed: bool
|
|
reason: str
|
|
provider: str | None = None
|
|
content: str | None = None
|
|
checked_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
|
|
|
|
|
|
ALLOWED_CLASSES = {"A", "B", "C"}
|
|
LOCAL_ONLY_CLASS = "A"
|
|
|
|
|
|
def inspect(request: GatewayRequest) -> GatewayResult:
|
|
"""Fail closed. No provider is configured in the local frame."""
|
|
data_class = (request.data_class or "").upper()
|
|
if data_class not in ALLOWED_CLASSES:
|
|
return GatewayResult(
|
|
allowed=False,
|
|
reason="unknown_data_class",
|
|
)
|
|
if data_class == LOCAL_ONLY_CLASS:
|
|
return GatewayResult(
|
|
allowed=False,
|
|
reason="class_a_never_leaves_local_zone",
|
|
)
|
|
return GatewayResult(
|
|
allowed=False,
|
|
reason="no_egress_provider_configured",
|
|
)
|
|
|
|
|
|
def complete(request: GatewayRequest) -> GatewayResult:
|
|
result = inspect(request)
|
|
if not result.allowed:
|
|
raise PrivacyGatewayError(
|
|
result.reason,
|
|
"Persönlicher KI-Aufruf wurde vom Privacy Gateway blockiert. "
|
|
"Es ist kein Egress-Provider konfiguriert.",
|
|
)
|
|
return result
|