fix(privacy): omit provider IDs from routine batch logs

pg_llm_batch/batch_api_client.py

@@ -29,7 +29,7 @@
 
 import aiohttp
 
-from .db import load_virtual_payload
+from .db import MAX_ENDPOINT_ALIAS_CHARACTERS, load_virtual_payload
 from .exceptions import GatewayError, ValidationError
 
 logger = logging.getLogger(__name__)
@@ -49,6 +49,9 @@
 BATCH_ENDPOINT_PATTERN = re.compile(
     r"/[A-Za-z0-9_~-]+(?:/[A-Za-z0-9._~-]+){0,15}\Z"
 )
+ENDPOINT_ALIAS_PATTERN = re.compile(
+    rf"[A-Za-z0-9][A-Za-z0-9._:-]{{0,{MAX_ENDPOINT_ALIAS_CHARACTERS - 1}}}\Z"
+)
 
 
 def _utc_now() -> datetime:
@@ -153,6 +156,29 @@ def _validate_batch_endpoint(value: Any) -> str:
     return value
 
 
+def _validate_credential_endpoint_alias(value: Any) -> str:
+    """Normalize one bounded ASCII endpoint alias before credential resolution.
+
+    Credential aliases are configuration selectors, not diagnostic payloads. This
+    boundary trims surrounding whitespace and then accepts only the same finite
+    identifier alphabet used by durable remote identities. Rejected input is never
+    reflected through ``ValidationError`` or handed to configuration/secret stores.
+    """
+    if type(value) is str:
+        normalized = value.strip()
+        if ENDPOINT_ALIAS_PATTERN.fullmatch(normalized) is not None:
+            return normalized
+    raise ValidationError(
+        field="endpoint_alias",
+        value="<redacted>",
+        reason=(
+            "must be 1-128 ASCII characters beginning with an alphanumeric "
+            "character and containing only letters, digits, dot, underscore, "
+            "colon, or hyphen"
+        ),
+    )
+
+
 def _is_loopback_host(hostname: str) -> bool:
     """Return whether a hostname is an explicit local-loopback destination."""
     if hostname.lower() in LOOPBACK_HOSTNAMES:
@@ -221,16 +247,15 @@ def config_credentials_provider(
     """
 
     def _provider(endpoint_alias: str) -> GatewayCredentials:
-        """Resolve the base URL and API key for one endpoint alias from the stores."""
-        url = config_store.get("gateway", endpoint_alias, None)
+        """Resolve one validated endpoint alias from configuration and secret stores."""
+        normalized_alias = _validate_credential_endpoint_alias(endpoint_alias)
+        url = config_store.get("gateway", normalized_alias, None)
         if url is None:
             url = config_store.get("gateway", "base_url", None)
         if url is None:
-            raise GatewayError(
-                f"No gateway base_url configured for alias '{endpoint_alias}'"
-            )
+            raise GatewayError("No gateway base_url configured for endpoint alias")
         normalized_url = _normalize_gateway_url(url)
-        api_key = secret_store.require_secret(f"gateway_api_key.{endpoint_alias}")
+        api_key = secret_store.require_secret(f"gateway_api_key.{normalized_alias}")
         return GatewayCredentials(url=normalized_url, api_key=api_key)
 
     return _provider
@@ -317,8 +342,9 @@ def __init__(
         self.postgres_dsn = postgres_dsn
 
         def _validated_credentials(endpoint_alias: str) -> GatewayCredentials:
-            """Revalidate custom credential destinations before authenticated I/O."""
-            resolved = credentials(endpoint_alias)
+            """Validate alias authority and custom destinations before authenticated I/O."""
+            normalized_alias = _validate_credential_endpoint_alias(endpoint_alias)
+            resolved = credentials(normalized_alias)
             return GatewayCredentials(
                 url=_normalize_gateway_url(resolved.url),
                 api_key=resolved.api_key,
@@ -669,7 +695,7 @@ async def upload_jsonl(
                     response_data={"error_type": "ProviderHTTPError"},
                 )
             result = await self._read_json_object(response, "Files API upload")
-            logger.info("Uploaded JSONL file: %s", result.get("id"))
+            logger.info("Uploaded JSONL file")
             return result
 
     async def delete_file(
@@ -749,7 +775,7 @@ async def create_batch_job(
                     response_data={"error_type": "ProviderHTTPError"},
                 )
             result = await self._read_json_object(response, "Batch creation")
-            logger.info("Created batch job: %s", result.get("id"))
+            logger.info("Created batch job")
             return result
 
     async def get_batch_status(

tests/test_batch_api_client_log_privacy.py

@@ -0,0 +1,92 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Privacy regressions for ordinary Batch API client logging."""
+
+from __future__ import annotations
+
+import json
+import logging
+
+from pg_llm_batch import batch_api_client as client_mod
+from pg_llm_batch.batch_api_client import BatchAPIClient, GatewayCredentials
+
+
+class _FakeContent:
+    """Expose deterministic response bytes through a bounded stream."""
+
+    def __init__(self, payload: bytes) -> None:
+        self._payload = payload
+
+    async def iter_chunked(self, size: int):
+        """Yield the payload in chunks no larger than ``size``."""
+        for index in range(0, len(self._payload), size):
+            yield self._payload[index : index + size]
+
+
+class _FakeResponse:
+    """Provide one bounded JSON response to the client."""
+
+    def __init__(self, status: int, payload: dict[str, object]) -> None:
+        encoded = json.dumps(payload).encode("utf-8")
+        self.status = status
+        self.content_length = len(encoded)
+        self.content = _FakeContent(encoded)
+
+    async def __aenter__(self):
+        """Enter the asynchronous response context."""
+        return self
+
+    async def __aexit__(self, *_exc):
+        """Leave the asynchronous response context."""
+        return None
+
+
+class _FakeSession:
+    """Route provider POST operations to deterministic responses."""
+
+    def __init__(self, file_id: str, batch_id: str) -> None:
+        self._file_response = _FakeResponse(200, {"id": file_id})
+        self._batch_response = _FakeResponse(
+            201, {"id": batch_id, "status": "validating"}
+        )
+
+    def post(self, url: str, **_kwargs):
+        """Return the response matching one Files or Batches endpoint."""
+        if url.endswith("/files"):
+            return self._file_response
+        if url.endswith("/batches"):
+            return self._batch_response
+        raise AssertionError(f"unexpected POST URL: {url}")
+
+
+def _credentials(_alias: str) -> GatewayCredentials:
+    """Return deterministic HTTPS credentials for the focused regression."""
+    return GatewayCredentials(url="https://gw.example/v1", api_key="sk-test")
+
+
+async def test_success_info_logs_omit_provider_resource_ids(caplog) -> None:
+    """Keep provider IDs in API results while excluding them from routine INFO logs."""
+    provider_file_id = "provider-file-id-sensitive"
+    provider_batch_id = "provider-batch-id-sensitive"
+    client = BatchAPIClient("postgresql://x", _credentials)
+    client._session = _FakeSession(provider_file_id, provider_batch_id)
+
+    async def _payload(_file_id: str) -> bytes:
+        return b'{"custom_id":"r1"}\n'
+
+    client._load_payload_bytes = _payload  # type: ignore[method-assign]
+
+    with caplog.at_level(logging.INFO, logger=client_mod.__name__):
+        uploaded = await client.upload_jsonl("memory://local-file", "default")
+        created = await client.create_batch_job(provider_file_id, "default")
+
+    assert uploaded["id"] == provider_file_id
+    assert created["id"] == provider_batch_id
+
+    messages = [
+        record.getMessage()
+        for record in caplog.records
+        if record.name == client_mod.__name__ and record.levelno == logging.INFO
+    ]
+    assert messages == ["Uploaded JSONL file", "Created batch job"]
+    assert all(provider_file_id not in message for message in messages)
+    assert all(provider_batch_id not in message for message in messages)

tests/test_endpoint_alias_credential_boundary.py

@@ -0,0 +1,118 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Privacy and authority regressions for endpoint-alias credential resolution."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from pg_llm_batch.batch_api_client import (
+    BatchAPIClient,
+    GatewayCredentials,
+    config_credentials_provider,
+)
+from pg_llm_batch.db import MAX_ENDPOINT_ALIAS_CHARACTERS
+from pg_llm_batch.exceptions import GatewayError, ValidationError
+
+
+class _ConfigStore:
+    """Record configuration lookups without granting any fallback behavior."""
+
+    def __init__(self, values: dict[tuple[str, str], Any] | None = None) -> None:
+        self.values = values or {}
+        self.calls: list[tuple[str, str, Any]] = []
+
+    def get(self, category: str, key: str, default: Any) -> Any:
+        """Return one configured value while preserving the exact lookup key."""
+        self.calls.append((category, key, default))
+        return self.values.get((category, key), default)
+
+
+class _SecretStore:
+    """Record secret-key lookups and return one deterministic test credential."""
+
+    def __init__(self) -> None:
+        self.calls: list[str] = []
+
+    def require_secret(self, key: str) -> str:
+        """Return a fixed secret after recording only the requested key."""
+        self.calls.append(key)
+        return "test-secret"
+
+
+@pytest.mark.parametrize(
+    "alias",
+    [
+        "bad/alias",
+        "bad alias",
+        "한글",
+        "bad\nalias",
+        "a" * (MAX_ENDPOINT_ALIAS_CHARACTERS + 1),
+    ],
+)
+def test_config_provider_rejects_noncanonical_alias_before_store_access(
+    alias: str,
+) -> None:
+    """Reject aliases outside the bounded ASCII grammar before any store can observe them."""
+    config = _ConfigStore()
+    secrets = _SecretStore()
+    provider = config_credentials_provider(config, secrets)
+
+    with pytest.raises(ValidationError) as raised:
+        provider(alias)
+
+    assert alias not in str(raised.value)
+    assert raised.value.details["value"] == "<redacted>"
+    assert config.calls == []
+    assert secrets.calls == []
+
+
+async def test_batch_client_rejects_alias_before_custom_credential_resolution() -> None:
+    """Invalid aliases must fail before an injected credential provider can observe them."""
+    credential_calls: list[str] = []
+
+    def _credentials(alias: str) -> GatewayCredentials:
+        credential_calls.append(alias)
+        raise AssertionError("credential provider must not receive an invalid alias")
+
+    client = BatchAPIClient("postgresql://test", _credentials)
+
+    with pytest.raises(ValidationError):
+        await client.get_batch_status("batch-1", "operator/secret")
+
+    assert credential_calls == []
+
+
+def test_config_credentials_provider_uses_normalized_alias_for_all_store_keys() -> None:
+    """Whitespace normalization must happen once before configuration or secret lookup."""
+    config = _ConfigStore(
+        {("gateway", "default"): "https://gateway.example.test/v1"}
+    )
+    secrets = _SecretStore()
+    provider = config_credentials_provider(config, secrets)
+
+    credentials = provider(" default ")
+
+    assert credentials.url == "https://gateway.example.test/v1"
+    assert credentials.api_key == "test-secret"
+    assert config.calls == [("gateway", "default", None)]
+    assert secrets.calls == ["gateway_api_key.default"]
+
+
+def test_missing_gateway_configuration_does_not_echo_valid_alias() -> None:
+    """Missing configuration diagnostics must not turn an alias into loggable content."""
+    config = _ConfigStore()
+    secrets = _SecretStore()
+    provider = config_credentials_provider(config, secrets)
+    alias = "private-admin"
+
+    with pytest.raises(GatewayError) as raised:
+        provider(alias)
+
+    assert alias not in str(raised.value)
+    assert config.calls == [
+        ("gateway", alias, None),
+        ("gateway", "base_url", None),
+    ]
+    assert secrets.calls == []