accesspath3r/privacy.py

"""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