jarca-ledger-mcp/jarca_ledger_mcp/provenance_policy.py

from __future__ import annotations

import hashlib
import json
import re
from pathlib import Path
from typing import Any, Iterable, Mapping

SOURCE_HASH_PREFIX_LENGTH = 12
CSV_PREVIEW_ALLOWED_FIELDS = (
    "transaction_date",
    "date",
    "amount",
    "type",
    "category",
    "subcategory",
    "merchant",
    "description",
)

EXPOSURE_LEVEL_MINIMAL = "minimal"
EXPOSURE_LEVEL_OPERATOR_SUMMARY = "operator-summary"
EXPOSURE_LEVEL_DEBUG_GATED = "debug-gated"
EXPOSURE_LEVELS = (
    EXPOSURE_LEVEL_MINIMAL,
    EXPOSURE_LEVEL_OPERATOR_SUMMARY,
    EXPOSURE_LEVEL_DEBUG_GATED,
)

_FORBIDDEN_KEY_HINTS = (
    "password",
    "pwd",
    "token",
    "secret",
    "credential",
    "raw_document",
    "raw_text",
    "raw_dump",
    "raw_payload_dump",
)
_PATH_KEY_HINTS = ("path", "filename", "filepath", "file_path", "source_path", "local_path")
_PATH_PREFIX_RE = re.compile(r"^[A-Za-z]:[\\/]|^(?:/|\\)")

_TOPLEVEL_FIELDS = {
    EXPOSURE_LEVEL_MINIMAL: {
        "source",
        "source_hash",
        "source_hash_prefix",
        "parser",
        "import_batch_id",
        "external_id",
        "dedupe_key",
    },
    EXPOSURE_LEVEL_OPERATOR_SUMMARY: {
        "source",
        "source_hash",
        "source_hash_prefix",
        "parser",
        "filename",
        "import_batch_id",
        "external_id",
        "dedupe_key",
        "page",
        "sheet",
        "row_number",
        "sheet_name",
        "validation",
    },
    EXPOSURE_LEVEL_DEBUG_GATED: {
        "source",
        "source_hash",
        "source_hash_prefix",
        "parser",
        "filename",
        "import_batch_id",
        "external_id",
        "dedupe_key",
        "page",
        "sheet",
        "row_number",
        "sheet_name",
        "validation",
        "raw_payload",
        "debug",
        "diagnostics",
    },
}

# Default field allowlists intentionally stay intentionally narrow and source-agnostic.
_RAW_PAYLOAD_ALLOWLIST = {
    EXPOSURE_LEVEL_MINIMAL: frozenset(),
    EXPOSURE_LEVEL_OPERATOR_SUMMARY: frozenset(
        {
            "row_index",
            "page",
            "sheet",
            "transaction_date",
            "transaction_datetime",
            "type",
            "amount",
            "code",
            "section",
            "action",
            "name",
            "counterparty",
            "merchant",
        }
    ),
    EXPOSURE_LEVEL_DEBUG_GATED: frozenset(
        {
            "row_index",
            "page",
            "sheet",
            "transaction_date",
            "transaction_datetime",
            "type",
            "amount",
            "code",
            "section",
            "action",
            "name",
            "counterparty",
            "merchant",
            "transaction_no",
            "business_no",
            "merchant_id",
            "channel",
            "balance_after",
            "quantity",
            "unit_price",
            "fee",
            "tax",
            "row_number",
            "row_ref",
        }
    ),
}


def source_hash_prefix(source_hash: str, length: int = SOURCE_HASH_PREFIX_LENGTH) -> str:
    return (source_hash or "").strip()[:length]


def redacted_source_file_label(file_path: str, source_hash: str = "") -> str:
    basename = Path(str(file_path or "")).name or "source"
    token = source_hash_prefix(source_hash)
    return f"{basename}#{token}" if token else basename


def file_sha256(file_path: str | Path) -> str:
    return hashlib.sha256(Path(file_path).read_bytes()).hexdigest()


def allowlisted_json_payload(payload: Mapping[str, Any], allowed_keys: Iterable[str]) -> str:
    cleaned = {
        key: payload[key]
        for key in allowed_keys
        if key in payload and payload[key] not in (None, "", [], {})
    }
    return json.dumps(cleaned, ensure_ascii=False, separators=(",", ":"))


def sanitize_csv_preview_rows(rows: list[Mapping[str, Any]], limit: int = 20) -> list[dict[str, Any]]:
    preview: list[dict[str, Any]] = []
    for row_number, row in enumerate(rows[: max(0, limit)], start=1):
        item = {"row_number": row_number}
        for key in CSV_PREVIEW_ALLOWED_FIELDS:
            value = row.get(key)
            if value not in (None, ""):
                item[key] = value
        preview.append(item)
    return preview


def _is_forbidden_key(key: str) -> bool:
    lower = key.lower()
    return any(token in lower for token in _FORBIDDEN_KEY_HINTS)


def _is_path_key(key: str) -> bool:
    lower = key.lower()
    return any(lower.endswith(path_key) or lower == path_key for path_key in _PATH_KEY_HINTS)


def _is_full_path(value: str) -> bool:
    if not value:
        return False
    return bool(_PATH_PREFIX_RE.search(value)) and ("/" in value or "\\" in value)


def _sanitize_mapping(
    data: Mapping[str, Any],
    *,
    allowed_keys: set[str] | None,
    source_hash: str,
) -> dict[str, Any]:
    sanitized: dict[str, Any] = {}
    for key, value in data.items():
        key_name = str(key)
        if allowed_keys is not None and key_name not in allowed_keys:
            continue
        if _is_forbidden_key(key_name):
            continue
        if _is_path_key(key_name) and isinstance(value, str) and _is_full_path(value):
            sanitized[key_name] = redacted_source_file_label(value, source_hash)
            continue
        if isinstance(value, Mapping):
            nested = _sanitize_mapping(value, allowed_keys=allowed_keys, source_hash=source_hash)
            if nested:
                sanitized[key_name] = nested
            continue
        if isinstance(value, list):
            if value == []:
                sanitized[key_name] = []
                continue

            nested_list: list[Any] = []
            for item in value:
                if isinstance(item, Mapping):
                    nested_item = _sanitize_mapping(item, allowed_keys=allowed_keys, source_hash=source_hash)
                    if nested_item:
                        nested_list.append(nested_item)
                elif isinstance(item, str):
                    if _is_forbidden_key(item) or _is_full_path(item):
                        continue
                    nested_list.append(item)
                else:
                    nested_list.append(item)
            if nested_list:
                sanitized[key_name] = nested_list
            continue
        if isinstance(value, str):
            if _is_forbidden_key(value) or _is_full_path(value):
                continue
        if value is not None:
            sanitized[key_name] = value
    return sanitized


def _normalize_raw_payload(raw_payload: Any, exposure_level: str, source_hash: str) -> dict[str, Any]:
    if isinstance(raw_payload, str):
        try:
            parsed_payload = json.loads(raw_payload)
        except Exception:
            return {}
    else:
        parsed_payload = raw_payload

    if not isinstance(parsed_payload, Mapping):
        return {}

    allowlist = set(_RAW_PAYLOAD_ALLOWLIST[exposure_level])
    if not allowlist:
        return {}

    return _sanitize_mapping(parsed_payload, allowed_keys=allowlist, source_hash=source_hash)


def sanitize_provenance(
    provenance: Mapping[str, Any],
    exposure_level: str = EXPOSURE_LEVEL_MINIMAL,
) -> dict[str, Any]:
    if exposure_level not in EXPOSURE_LEVELS:
        raise ValueError(f"unsupported exposure level: {exposure_level}")

    source_hash = str(provenance.get("source_hash", ""))
    source_hash_prefix_value = source_hash_prefix(source_hash)
    allowed_fields = _TOPLEVEL_FIELDS[exposure_level]

    sanitized: dict[str, Any] = {"source_hash_prefix": source_hash_prefix_value}

    for key in sorted(allowed_fields):
        if key == "source_hash_prefix":
            sanitized[key] = source_hash_prefix_value
            continue
        if key not in provenance:
            continue
        if key in {"raw_payload", "debug", "diagnostics"} and key in provenance:
            if key in {"raw_payload"}:
                value = _normalize_raw_payload(
                    provenance[key],
                    exposure_level=exposure_level,
                    source_hash=source_hash,
                )
            else:
                value = _sanitize_mapping(provenance[key], allowed_keys=None, source_hash=source_hash) if isinstance(provenance[key], Mapping) else provenance[key]
            if value:
                sanitized[key] = value
            continue

        value = provenance[key]
        if _is_forbidden_key(str(key)):
            continue
        if isinstance(value, Mapping):
            value = _sanitize_mapping(value, allowed_keys=None, source_hash=source_hash)
            if not value:
                continue
            sanitized[key] = value
            continue

        if isinstance(value, list):
            nested_list: list[Any] = []
            for item in value:
                if isinstance(item, Mapping):
                    item_value = _sanitize_mapping(item, allowed_keys=None, source_hash=source_hash)
                    if item_value:
                        nested_list.append(item_value)
                elif isinstance(item, str):
                    if _is_forbidden_key(item) or _is_full_path(item):
                        continue
                    nested_list.append(item)
                else:
                    nested_list.append(item)
            if nested_list:
                sanitized[key] = nested_list
            continue

        if _is_path_key(str(key)) and isinstance(value, str) and _is_full_path(value):
            sanitized[key] = redacted_source_file_label(value, source_hash)
            continue

        if isinstance(value, str):
            if _is_forbidden_key(value) or _is_full_path(value):
                continue

        if value is not None:
            sanitized[key] = value

    if "parser" not in sanitized and "source" in provenance:
        sanitized["parser"] = provenance["source"]

    if exposure_level != EXPOSURE_LEVEL_MINIMAL:
        for path_key in ("file_path", "source_path", "local_path", "filename"):
            raw_path = provenance.get(path_key)
            if isinstance(raw_path, str) and _is_full_path(raw_path):
                sanitized["filename"] = redacted_source_file_label(raw_path, source_hash)
                break

        # Keep raw document/text dumps suppressed even on debug output.
        for forbidden_key in ("raw_document", "raw_text", "raw_dump", "raw_payload_dump"):
            sanitized.pop(forbidden_key, None)

    return sanitized


def sanitize_provenance_envelope(
    provenance: Mapping[str, Any],
    exposure_level: str = EXPOSURE_LEVEL_MINIMAL,
) -> dict[str, Any]:
    return sanitize_provenance(provenance, exposure_level=exposure_level)


def sanitize_provenance_for_exposure(
    provenance: Mapping[str, Any],
    exposure_level: str = EXPOSURE_LEVEL_MINIMAL,
) -> dict[str, Any]:
    return sanitize_provenance(provenance, exposure_level=exposure_level)