feat: validate Anwarpy publication metadata

2026-10-09 15:52:08 +03:00
parent 00f8e4aee7
commit 4aeb747d56
8 changed files with 457 additions and 0 deletions
@@ -0,0 +1,73 @@
+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")