File size: 4,961 Bytes
2f382c4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | """Privacy-safe formatting for console output, logs, and public artifacts."""
from __future__ import annotations
import os
import re
import socket
from pathlib import Path
from typing import Iterable
EMAIL_PATTERN = re.compile(
r"(?<![A-Za-z0-9._%+-])[A-Za-z0-9._%+-]+@"
r"[A-Za-z0-9.-]+\.[A-Za-z]{2,}(?![A-Za-z0-9.-])"
)
HOME_PATH_PATTERN = re.compile(r"/(?:home|Users)/[^/\s:'\"]+")
DATA_USER_PATH_PATTERN = re.compile(r"/data\d*/userdata/[^/\s:'\"]+")
SECRET_QUERY_PATTERN = re.compile(
r"(?i)(\b(?:access[_-]?token|api[_-]?key|password|secret|token)=)"
r"[^&\s]+"
)
IPV4_PATTERN = re.compile(r"(?<![\d.])(?:\d{1,3}\.){3}\d{1,3}(?![\d.])")
def _resolved(path: str | os.PathLike[str] | Path) -> Path:
return Path(path).expanduser().resolve(strict=False)
def public_path(
path: str | os.PathLike[str] | Path,
*,
project_root: str | os.PathLike[str] | Path,
output_root: str | os.PathLike[str] | Path | None = None,
) -> str:
"""Return a useful path label without exposing a workstation/user prefix."""
resolved = _resolved(path)
project = _resolved(project_root)
try:
relative = resolved.relative_to(project)
return relative.as_posix() or "."
except ValueError:
pass
if output_root is not None:
output = _resolved(output_root)
try:
relative = resolved.relative_to(output)
suffix = relative.as_posix()
return "${OUTPUT_DIR}" + (f"/{suffix}" if suffix else "")
except ValueError:
pass
home_value = os.environ.get("HOME")
if home_value:
home = _resolved(home_value)
try:
relative = resolved.relative_to(home)
suffix = relative.as_posix()
return "~" + (f"/{suffix}" if suffix else "")
except ValueError:
pass
return f"<external>/{resolved.name}" if resolved.name else "<external>"
def _replacement_pairs(
*,
project_root: str | os.PathLike[str] | Path | None,
output_root: str | os.PathLike[str] | Path | None,
sensitive_paths: Iterable[str | os.PathLike[str] | Path],
) -> list[tuple[str, str]]:
pairs: list[tuple[str, str]] = []
def add(value: str | os.PathLike[str] | Path | None, label: str) -> None:
if value is None or not str(value).strip():
return
raw = str(value).rstrip("/")
# Relative values such as "." are meaningful CLI inputs but unsafe
# replacement needles ("." would rewrite every decimal and suffix).
if raw and Path(raw).expanduser().is_absolute():
pairs.append((raw, label))
try:
resolved = str(_resolved(value)).rstrip("/")
except (OSError, RuntimeError, ValueError):
return
if resolved and resolved not in {".", "/"} and resolved != raw:
pairs.append((resolved, label))
add(output_root, "${OUTPUT_DIR}")
add(project_root, "${PROJECT_ROOT}")
add(os.environ.get("HOME"), "${HOME}")
for index, path in enumerate(sensitive_paths, start=1):
add(path, f"${{PRIVATE_PATH_{index}}}")
# Replace longer prefixes first so OUTPUT_DIR wins over PROJECT_ROOT.
return sorted(set(pairs), key=lambda pair: len(pair[0]), reverse=True)
def sanitize_text(
text: str,
*,
project_root: str | os.PathLike[str] | Path | None = None,
output_root: str | os.PathLike[str] | Path | None = None,
sensitive_paths: Iterable[str | os.PathLike[str] | Path] = (),
) -> str:
"""Redact local identity/path details while retaining diagnostic meaning."""
result = text
for value, label in _replacement_pairs(
project_root=project_root,
output_root=output_root,
sensitive_paths=sensitive_paths,
):
result = result.replace(value, label)
result = HOME_PATH_PATTERN.sub("${HOME}", result)
result = DATA_USER_PATH_PATTERN.sub("${HOME}", result)
result = EMAIL_PATTERN.sub("<redacted-email>", result)
result = SECRET_QUERY_PATTERN.sub(r"\1<redacted>", result)
def redact_ipv4(match: re.Match[str]) -> str:
value = match.group(0)
octets = value.split(".")
if any(int(octet) > 255 for octet in octets) or value.startswith("127."):
return value
return "<redacted-ip>"
result = IPV4_PATTERN.sub(redact_ipv4, result)
identity_values = {
os.environ.get("USER", ""),
os.environ.get("LOGNAME", ""),
os.environ.get("HOSTNAME", ""),
socket.gethostname(),
}
for value in sorted(identity_values, key=len, reverse=True):
# Avoid replacing common words such as "root" in ordinary diagnostics.
if len(value) < 4 or value.lower() in {"root", "user", "admin"}:
continue
result = re.sub(
rf"(?<![A-Za-z0-9_.-]){re.escape(value)}(?![A-Za-z0-9_.-])",
"<redacted-identity>",
result,
)
return result
|