client/python/apache_polaris/cli/log_sanitizer.py

#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
"""
Sanitize Polaris CLI debug log output.

``polaris --debug`` logs HTTP traffic to stderr. Known credential fields and
OAuth token endpoint payloads are redacted before writing.
"""

import json
import sys
from typing import Any
from urllib.parse import parse_qsl, urlencode

REDACTED = "***REDACTED***"
OAUTH_TOKEN_BODY_REDACTED = "<redacted sensitive authentication payload>"
SANITIZE_FAILURE_MESSAGE = "<redacted: unable to sanitize payload>"

# Explicit list of credential field names that appear on the wire.
# - OAuth credentials: the Polaris management API serializes bodies by alias
#   (camelCase); the OAuth token endpoint uses snake_case; Iceberg REST config
#   uses a bare ``token`` for the bearer.
# - Iceberg vended-credential storage properties: keys returned in
#   ``LoadTableResult.config`` (see spec/iceberg-rest-catalog-open-api.yaml).
#   ``adls.sas-token`` is emitted with a ``.<hostname>`` / ``.<account>``
#   suffix; the prefix check in ``_is_sensitive_key`` handles those variants.
SENSITIVE_BODY_KEYS = frozenset(
    {
        "client_secret",
        "clientSecret",
        "access_token",
        "accessToken",
        "refresh_token",
        "refreshToken",
        "bearerToken",
        "token",
        "password",
        "secret",
        "s3.secret-access-key",
        "s3.session-token",
        "gcs.oauth2.token",
        "adls.sas-token",
    }
)


def _is_sensitive_key(key: Any) -> bool:
    # Prefix match catches ``adls.sas-token.<hostname>`` / ``.<account>`` variants.
    return key in SENSITIVE_BODY_KEYS or (
        isinstance(key, str) and key.startswith("adls.sas-token")
    )


def sanitize_data(data: Any) -> Any:
    if isinstance(data, dict):
        return {
            key: REDACTED if _is_sensitive_key(key) else sanitize_data(value)
            for key, value in data.items()
        }
    if isinstance(data, list):
        return [sanitize_data(item) for item in data]
    if isinstance(data, tuple):
        return tuple(sanitize_data(item) for item in data)
    return data


def sanitize_headers(headers: dict[str, Any] | None) -> dict[str, Any] | None:
    if headers is None:
        return headers
    return {
        key: REDACTED if key == "Authorization" else value
        for key, value in headers.items()
    }


def is_oauth_token_endpoint(url: str) -> bool:
    return url.endswith("/oauth/tokens")


def _sanitize_form_body(body: str) -> str:
    sanitized_pairs = [
        (key, REDACTED if _is_sensitive_key(key) else value)
        for key, value in parse_qsl(body, keep_blank_values=True)
    ]
    return urlencode(sanitized_pairs, safe="*")


def sanitize_body(body: Any) -> Any:
    if body is None:
        return body
    if isinstance(body, (dict, list, tuple)):
        return sanitize_data(body)
    if isinstance(body, bytes):
        try:
            return sanitize_body(body.decode("utf-8"))
        except UnicodeDecodeError:
            return REDACTED
    if isinstance(body, str):
        try:
            parsed = json.loads(body)
            return json.dumps(sanitize_data(parsed))
        except json.JSONDecodeError:
            pass
        if "=" in body:
            return _sanitize_form_body(body)
        return body
    return str(body)


def sanitize_body_for_log(body: Any, url: str) -> Any:
    if is_oauth_token_endpoint(url):
        return OAUTH_TOKEN_BODY_REDACTED
    return sanitize_body(body)


def safe_sanitize_headers(headers: Any) -> Any:
    try:
        return sanitize_headers(headers)
    except Exception as e:
        sys.stderr.write(f"Failed to sanitize debug log headers: {e}\n")
        return SANITIZE_FAILURE_MESSAGE


def safe_sanitize_body_for_log(body: Any, url: str) -> Any:
    try:
        return sanitize_body_for_log(body, url)
    except Exception as e:
        sys.stderr.write(f"Failed to sanitize debug log body: {e}\n")
        return SANITIZE_FAILURE_MESSAGE