diff --git a/pg_llm_batch/batch_api_client.py b/pg_llm_batch/batch_api_client.py index a438de87..e918afcc 100644 --- a/pg_llm_batch/batch_api_client.py +++ b/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="", + 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( diff --git a/tests/test_batch_api_client_log_privacy.py b/tests/test_batch_api_client_log_privacy.py new file mode 100644 index 00000000..54f6f376 --- /dev/null +++ b/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) diff --git a/tests/test_endpoint_alias_credential_boundary.py b/tests/test_endpoint_alias_credential_boundary.py new file mode 100644 index 00000000..eb28db7d --- /dev/null +++ b/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"] == "" + 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 == []