74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import re
|
|
from typing import Iterable
|
|
from urllib.parse import urlsplit
|
|
|
|
REDACTION = "[REDACTED]"
|
|
|
|
_URL_RE = re.compile(r"https?://[^\s<>\]\[\)\(\"']+", re.IGNORECASE)
|
|
_EMAIL_RE = re.compile(r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b", re.IGNORECASE)
|
|
_IPV4_RE = re.compile(r"(?<![\d.])(?:\d{1,3}\.){3}\d{1,3}(?![\d.])")
|
|
_IPV6_CANDIDATE_RE = re.compile(r"(?<![0-9A-Fa-f:])(?:[0-9A-Fa-f]{0,4}:){2,}[0-9A-Fa-f]{0,4}(?![0-9A-Fa-f:])")
|
|
_PHONE_RE = re.compile(r"(?<!\w)\+?\d(?:[\s().-]*\d){7,14}(?!\w)")
|
|
_PATH_RE = re.compile(r"(?:/(?:Users|home|root|opt|etc|var)/[^\s,;]+|[A-Za-z]:\\Users\\[^\s,;]+)")
|
|
_CREDENTIAL_RE = re.compile(
|
|
r"(?i)\b(?:password|passwd|secret|token|api[_-]?key|authorization)\s*[:=]\s*[^\s,;]+"
|
|
)
|
|
_PRIVATE_HOST_RE = re.compile(r"\b(?:[A-Z0-9-]+\.)+(?:local|internal|lan|home)\b", re.IGNORECASE)
|
|
_FQDN_RE = re.compile(r"\b(?:[A-Z0-9-]+\.)+[A-Z]{2,63}\b", re.IGNORECASE)
|
|
|
|
|
|
def _normalized_hosts(hosts: Iterable[str]) -> set[str]:
|
|
return {host.strip().lower().rstrip(".") for host in hosts if host.strip()}
|
|
|
|
|
|
def _redact_url(match: re.Match[str], allowed_hosts: set[str]) -> str:
|
|
value = match.group(0)
|
|
host = (urlsplit(value).hostname or "").lower().rstrip(".")
|
|
return value if host in allowed_hosts else REDACTION
|
|
|
|
|
|
def _redact_ipv4(match: re.Match[str]) -> str:
|
|
value = match.group(0)
|
|
try:
|
|
ipaddress.IPv4Address(value)
|
|
except ipaddress.AddressValueError:
|
|
return value
|
|
return REDACTION
|
|
|
|
|
|
def _redact_ipv6(match: re.Match[str]) -> str:
|
|
value = match.group(0)
|
|
try:
|
|
ipaddress.IPv6Address(value)
|
|
except ipaddress.AddressValueError:
|
|
return value
|
|
return REDACTION
|
|
|
|
|
|
def _redact_fqdn(match: re.Match[str], allowed_hosts: set[str]) -> str:
|
|
host = match.group(0)
|
|
return host if host.lower().rstrip(".") in allowed_hosts else REDACTION
|
|
|
|
|
|
def sanitize_text(text: str, *, allowed_hosts: Iterable[str]) -> str:
|
|
hosts = _normalized_hosts(allowed_hosts)
|
|
cleaned = _URL_RE.sub(lambda match: _redact_url(match, hosts), text)
|
|
cleaned = _CREDENTIAL_RE.sub(REDACTION, cleaned)
|
|
cleaned = _EMAIL_RE.sub(REDACTION, cleaned)
|
|
cleaned = _PATH_RE.sub(REDACTION, cleaned)
|
|
cleaned = _IPV4_RE.sub(_redact_ipv4, cleaned)
|
|
cleaned = _IPV6_CANDIDATE_RE.sub(_redact_ipv6, cleaned)
|
|
cleaned = _PHONE_RE.sub(REDACTION, cleaned)
|
|
cleaned = _PRIVATE_HOST_RE.sub(REDACTION, cleaned)
|
|
cleaned = _FQDN_RE.sub(lambda match: _redact_fqdn(match, hosts), cleaned)
|
|
return cleaned
|
|
|
|
|
|
def assert_publishable(text: str, *, allowed_hosts: Iterable[str]) -> None:
|
|
sanitized = sanitize_text(text, allowed_hosts=allowed_hosts)
|
|
if sanitized != text:
|
|
raise ValueError("content still contains private or unapproved data")
|