Improve secret redaction transparency
This commit is contained in:
Mikei386
@@ -1,8 +1,8 @@
|
||||
"""
|
||||
title: MikeAI Secret Redaction
|
||||
author: MikeAI
|
||||
version: 1.0.0
|
||||
description: Redacts common credentials from tool results and final assistant output.
|
||||
version: 1.1.0
|
||||
description: Redacts likely credentials while reporting only safe match categories and origins.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -20,63 +20,155 @@ class Filter:
|
||||
self.valves = self.Valves()
|
||||
self.toggle = False
|
||||
|
||||
def _redact(self, text: str) -> tuple[str, int]:
|
||||
@staticmethod
|
||||
def _merge(target: dict[str, int], source: dict[str, int]) -> None:
|
||||
for category, count in source.items():
|
||||
target[category] = target.get(category, 0) + count
|
||||
|
||||
@staticmethod
|
||||
def _safe_documentation_value(value: str) -> bool:
|
||||
"""Ignore obvious examples, references and placeholders, never real values."""
|
||||
candidate = value.strip().strip("'\"")
|
||||
lowered = candidate.casefold()
|
||||
if not candidate:
|
||||
return True
|
||||
if lowered in {
|
||||
"[redacted]", "redacted", "[configured]", "configured",
|
||||
"placeholder", "example", "example-value", "changeme", "unknown",
|
||||
"none", "null", "true", "false",
|
||||
}:
|
||||
return True
|
||||
if re.fullmatch(r"\$\{[A-Za-z_][A-Za-z0-9_]*(?::[^}]*)?\}", candidate):
|
||||
return True
|
||||
if re.fullmatch(r"<[A-Za-z0-9_.:/ -]+>", candidate):
|
||||
return True
|
||||
if re.fullmatch(r"[A-Z][A-Z0-9_]{7,}", candidate):
|
||||
return True
|
||||
if candidate.startswith(("/etc/", "/var/", "/opt/", "/data/", "/srv/", "./", "../")):
|
||||
return True
|
||||
if candidate.startswith(("http://", "https://", "ssh://")):
|
||||
return True
|
||||
return False
|
||||
|
||||
def _redact(self, text: str) -> tuple[str, dict[str, int]]:
|
||||
if not isinstance(text, str) or not text:
|
||||
return text, 0
|
||||
count = 0
|
||||
return text, {}
|
||||
counts: dict[str, int] = {}
|
||||
|
||||
def replace_full(match):
|
||||
nonlocal count
|
||||
count += 1
|
||||
return self.valves.replacement
|
||||
def mark(category: str) -> None:
|
||||
counts[category] = counts.get(category, 0) + 1
|
||||
|
||||
def replace_value(match):
|
||||
nonlocal count
|
||||
count += 1
|
||||
full_patterns = [
|
||||
(
|
||||
r"-----BEGIN (?:OPENSSH |RSA |EC |DSA )?PRIVATE KEY-----.*?"
|
||||
r"-----END (?:OPENSSH |RSA |EC |DSA )?PRIVATE KEY-----",
|
||||
"privater Schlüssel",
|
||||
),
|
||||
(
|
||||
r"\beyJ[A-Za-z0-9_-]{20,}\.[A-Za-z0-9_-]{20,}\."
|
||||
r"[A-Za-z0-9_-]{10,}\b",
|
||||
"JWT",
|
||||
),
|
||||
]
|
||||
for pattern, category in full_patterns:
|
||||
def replace_full(match, category=category):
|
||||
mark(category)
|
||||
return self.valves.replacement
|
||||
|
||||
text = re.sub(
|
||||
pattern, replace_full, text,
|
||||
flags=re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
|
||||
bearer_pattern = re.compile(
|
||||
r"((?:authorization\s*[:=]\s*)?(?:bearer\s+))"
|
||||
r"([A-Za-z0-9._~+/=-]{16,})",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
def replace_bearer(match):
|
||||
value = match.group(2)
|
||||
if self._safe_documentation_value(value):
|
||||
return match.group(0)
|
||||
mark("Bearer-Token")
|
||||
return match.group(1) + self.valves.replacement
|
||||
|
||||
patterns_full = [
|
||||
r"-----BEGIN (?:OPENSSH |RSA |EC |DSA )?PRIVATE KEY-----.*?-----END (?:OPENSSH |RSA |EC |DSA )?PRIVATE KEY-----",
|
||||
r"\beyJ[A-Za-z0-9_-]{20,}\.[A-Za-z0-9_-]{20,}\.[A-Za-z0-9_-]{10,}\b",
|
||||
]
|
||||
for pattern in patterns_full:
|
||||
text = re.sub(pattern, replace_full, text, flags=re.IGNORECASE | re.DOTALL)
|
||||
text = bearer_pattern.sub(replace_bearer, text)
|
||||
|
||||
patterns_value = [
|
||||
r"((?:authorization\s*[:=]\s*)?(?:bearer\s+))[A-Za-z0-9._~+/=-]{16,}",
|
||||
r"((?:\"|')?(?:api[_-]?key|access[_-]?token|refresh[_-]?token|password|passwd|secret|token)(?:\"|')?\s*[:=]\s*(?:\"|')?)[^\s\"',;&}]{8,}",
|
||||
r"([?&](?:api[_-]?key|access[_-]?token|token|key)=)[^&\s]+",
|
||||
]
|
||||
for pattern in patterns_value:
|
||||
text = re.sub(pattern, replace_value, text, flags=re.IGNORECASE)
|
||||
return text, count
|
||||
named_pattern = re.compile(
|
||||
r"((?:\"|')?"
|
||||
r"(api[_-]?key|access[_-]?token|refresh[_-]?token|password|passwd|secret|token)"
|
||||
r"(?:\"|')?\s*[:=]\s*(?:\"|')?)"
|
||||
r"(\$\{[^}\r\n]{1,200}\}|<[^>\r\n]{1,100}>|[^\s\"',;&}]{8,})",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
def _redact_content(self, content) -> tuple[object, int]:
|
||||
def replace_named(match):
|
||||
value = match.group(3)
|
||||
if self._safe_documentation_value(value):
|
||||
return match.group(0)
|
||||
name = match.group(2).casefold().replace("-", "_")
|
||||
if "password" in name or name == "passwd":
|
||||
category = "Passwort"
|
||||
elif "api" in name and "key" in name:
|
||||
category = "API-Key"
|
||||
elif "token" in name:
|
||||
category = "Token"
|
||||
else:
|
||||
category = "Secret"
|
||||
mark(category)
|
||||
return match.group(1) + self.valves.replacement
|
||||
|
||||
text = named_pattern.sub(replace_named, text)
|
||||
|
||||
query_pattern = re.compile(
|
||||
r"([?&](?:api[_-]?key|access[_-]?token|token|key)=)([^&\s]+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
def replace_query(match):
|
||||
value = match.group(2)
|
||||
if self._safe_documentation_value(value):
|
||||
return match.group(0)
|
||||
mark("URL-Zugangswert")
|
||||
return match.group(1) + self.valves.replacement
|
||||
|
||||
text = query_pattern.sub(replace_query, text)
|
||||
return text, counts
|
||||
|
||||
def _redact_content(self, content) -> tuple[object, dict[str, int]]:
|
||||
if isinstance(content, str):
|
||||
return self._redact(content)
|
||||
if not isinstance(content, list):
|
||||
return content, 0
|
||||
return content, {}
|
||||
result = []
|
||||
total = 0
|
||||
counts: dict[str, int] = {}
|
||||
for part in content:
|
||||
if isinstance(part, dict) and isinstance(part.get("text"), str):
|
||||
item = dict(part)
|
||||
item["text"], found = self._redact(item["text"])
|
||||
total += found
|
||||
self._merge(counts, found)
|
||||
result.append(item)
|
||||
else:
|
||||
result.append(part)
|
||||
return result, total
|
||||
return result, counts
|
||||
|
||||
async def _notify(self, emitter, count: int) -> None:
|
||||
if not count or emitter is None:
|
||||
async def _notify(self, emitter, counts: dict[str, int], origin: str) -> None:
|
||||
total = sum(counts.values())
|
||||
if not total or emitter is None:
|
||||
return
|
||||
categories = ", ".join(
|
||||
f"{category}: {count}" for category, count in sorted(counts.items())
|
||||
)
|
||||
try:
|
||||
await emitter(
|
||||
{
|
||||
"type": "status",
|
||||
"data": {
|
||||
"description": f"Secret-Schutz: {count} mögliche Zugangsdaten entfernt.",
|
||||
"description": (
|
||||
f"Secret-Schutz: {total} Treffer in {origin} entfernt "
|
||||
f"({categories}). Werte werden nicht protokolliert."
|
||||
),
|
||||
"done": True,
|
||||
},
|
||||
}
|
||||
@@ -85,21 +177,21 @@ class Filter:
|
||||
pass
|
||||
|
||||
async def inlet(self, body: dict, __event_emitter__=None, **kwargs) -> dict:
|
||||
total = 0
|
||||
counts: dict[str, int] = {}
|
||||
for message in body.get("messages") or []:
|
||||
if message.get("role") != "tool":
|
||||
continue
|
||||
message["content"], count = self._redact_content(message.get("content", ""))
|
||||
total += count
|
||||
await self._notify(__event_emitter__, total)
|
||||
message["content"], found = self._redact_content(message.get("content", ""))
|
||||
self._merge(counts, found)
|
||||
await self._notify(__event_emitter__, counts, "Werkzeugausgaben")
|
||||
return body
|
||||
|
||||
async def outlet(self, body: dict, __event_emitter__=None, **kwargs) -> dict:
|
||||
total = 0
|
||||
counts: dict[str, int] = {}
|
||||
for message in body.get("messages") or []:
|
||||
if message.get("role") != "assistant":
|
||||
continue
|
||||
message["content"], count = self._redact_content(message.get("content", ""))
|
||||
total += count
|
||||
await self._notify(__event_emitter__, total)
|
||||
message["content"], found = self._redact_content(message.get("content", ""))
|
||||
self._merge(counts, found)
|
||||
await self._notify(__event_emitter__, counts, "Modellantworten")
|
||||
return body
|
||||
|
||||
Reference in New Issue
Block a user