""" Public response redaction helpers. These helpers preserve committed evidence semantics while preventing internal LAN topology from being returned to browser-facing API responses. """ from __future__ import annotations import re from typing import Any _ENDPOINT_ALIASES = { "192.168.0.110:3001": "host:public-gateway/gitea", "192.168.0.110:3002": "host:public-gateway/grafana", "192.168.0.110:3100": "host:public-gateway/langfuse", "192.168.0.110:5000": "host:public-gateway/registry", "192.168.0.110:9000": "host:public-gateway/sentry", "192.168.0.112:8080": "host:kali-readonly/scanner", "192.168.0.188:11434": "host:observability-a/ollama", "192.168.0.188:8088": "host:observability-a/openclaw-legacy", "192.168.0.188:8089": "host:observability-a/openclaw", "192.168.0.188:3301": "host:observability-a/signoz", "192.168.0.188:5432": "host:observability-a/postgres", "192.168.0.188:6380": "host:observability-a/redis", } _HOST_ALIASES = { "192.168.0.110": "host:public-gateway", "192.168.0.111": "host:dev-a", "192.168.0.112": "host:kali-readonly", "192.168.0.120": "host:k3s-control-a", "192.168.0.121": "host:k3s-control-b", "192.168.0.125": "host:edge-vip", "192.168.0.168": "host:dev-b", "192.168.0.188": "host:observability-a", } _PRIVATE_LAN_RE = re.compile(r"192\.168\.0\.\d{1,3}(?::\d{1,5})?") _WORK_CONTEXT_REPLACEMENTS = { "工作視窗": "內部協作環境", "批准!繼續": "內部短訊指令", "批准!": "內部短訊指令", "source_thread_id": "redacted_thread_id", "codex_delegation": "redacted_delegation", } def redact_public_lan_text(value: str) -> str: """Replace internal LAN addresses with public-safe asset aliases.""" redacted = value for endpoint, alias in _ENDPOINT_ALIASES.items(): redacted = redacted.replace(f"http://{endpoint}", alias) redacted = redacted.replace(f"https://{endpoint}", alias) redacted = redacted.replace(endpoint, alias) for host, alias in _HOST_ALIASES.items(): redacted = redacted.replace(f"http://{host}", alias) redacted = redacted.replace(f"https://{host}", alias) redacted = redacted.replace(host, alias) redacted = _PRIVATE_LAN_RE.sub("host:internal-node", redacted) for phrase, replacement in _WORK_CONTEXT_REPLACEMENTS.items(): redacted = redacted.replace(phrase, replacement) return redacted def redact_public_lan_topology(value: Any) -> Any: """Recursively redact internal LAN topology from JSON-compatible values.""" if isinstance(value, str): return redact_public_lan_text(value) if isinstance(value, list): return [redact_public_lan_topology(item) for item in value] if isinstance(value, dict): return {key: redact_public_lan_topology(nested) for key, nested in value.items()} return value