From b30bb50c4264e24695f0cfe8b431117939e54454 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Thu, 8 Oct 2026 21:01:56 +0300 Subject: [PATCH 1/8] feat(standard-tests): lint `airbyte_secret` marking and secret leaks Add the first slice of the credential-free spec lint from airbytehq/airbyte-internal-issues#17136. - `test_docker_image_build_and_spec` and `SourceTestSuiteBase.test_spec` now require exactly one SPEC message and fail when a property whose name denotes a secret (`password`, `api_key`, `*_token`, ...) is not marked `airbyte_secret: true`, when a boolean/object property is marked secret, or when any `airbyte_secret` value is not a boolean (the CDK's secret filter ignores a string `"true"`). `airbyte-cdk image test` builds with `no_verify=True`, so the Docker spec test is the only place a Python connector's spec gets checked. - `test_docker_image_build_and_check` and both in-process `test_check` tests fail when a non-CONTROL message contains an `airbyte_secret` value from the scenario's config. The excerpt in the failure message is masked. Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/test/standard_tests/_spec_lint.py | 273 +++++++++++++ .../test/standard_tests/connector_base.py | 29 ++ .../test/standard_tests/docker_base.py | 48 ++- .../test/standard_tests/source_base.py | 10 +- unit_tests/test/test_spec_lint.py | 382 ++++++++++++++++++ 5 files changed, 726 insertions(+), 16 deletions(-) create mode 100644 airbyte_cdk/test/standard_tests/_spec_lint.py create mode 100644 unit_tests/test/test_spec_lint.py diff --git a/airbyte_cdk/test/standard_tests/_spec_lint.py b/airbyte_cdk/test/standard_tests/_spec_lint.py new file mode 100644 index 0000000000..2fbd45b61b --- /dev/null +++ b/airbyte_cdk/test/standard_tests/_spec_lint.py @@ -0,0 +1,273 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Credential-free assertions on a connector's spec. + +These checks need only the connector's SPEC message, plus a scenario's config and the +connector's output for the secret leakage check. That lets them run in both standard-test +paths: in-process, and against the connector image. +""" + +from __future__ import annotations + +import json +import re +from collections.abc import Iterable, Iterator, Mapping +from typing import Any + +from airbyte_cdk.models import AirbyteMessageSerializer, ConnectorSpecification, Type +from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput +from airbyte_cdk.utils.airbyte_secrets_utils import get_secrets + +SECRET_PROPERTY_NAMES = frozenset( + { + "access_token", + "api_token", + "certificate", + "client_secret", + "client_token", + "credentials", + "jwt", + "key", + "password", + "refresh_token", + "secret", + "service_account", + "service_account_info", + "token", + } +) +"""Property names that always denote a secret. + +Ported from the connector acceptance tests (CAT), minus `tenant_id`, `app_id` and `appid`: +those identify an account or an app rather than authenticate it. +""" + +SECRET_PROPERTY_NAME_SUFFIXES = ( + "_token", + "_secret", + "password", + "api_key", + "apikey", + "secret_key", + "private_key", + "access_key", +) +"""Name suffixes that denote a secret, such as `api_key` or `personal_access_token`. + +CAT matched exact names only, so it never flagged an unmarked `api_key`, the most common +secret property name in the connector fleet. +""" + +_SECRET_CAPABLE_TYPES = {"string", "integer", "number"} + +_MASK = "****" +_EXCERPT_CONTEXT = 150 + +_MIN_LEAK_CHECK_LENGTH = 8 +"""Secret values shorter than this are not searched for in the output. + +Short values, such as `123` in a test config, occur in ordinary output by coincidence. +""" + + +def is_secret_property_name(name: str) -> bool: + """Return whether a spec property with this name is expected to hold a secret.""" + normalized_name = name.lower().replace("-", "_") + return normalized_name in SECRET_PROPERTY_NAMES or normalized_name.endswith( + SECRET_PROPERTY_NAME_SUFFIXES + ) + + +def _can_hold_secret(property_schema: Mapping[str, Any]) -> bool: + """Return whether a property can hold a secret value. + + Booleans and nulls cannot, and the UI cannot render secret objects or arrays. A property + with a `const` value cannot either, since every config holds the same value. + """ + property_type = property_schema.get("type") + if isinstance(property_type, str): + types = {property_type} + elif isinstance(property_type, list): + types = set(property_type) + else: + return False + return bool(types & _SECRET_CAPABLE_TYPES) and "const" not in property_schema + + +def _iter_named_properties( + schema: Mapping[str, Any], + pointer: str = "", + name: str | None = None, +) -> Iterator[tuple[str, str, Mapping[str, Any]]]: + """Yield `(json_pointer, property_name, property_schema)` for each property in a schema. + + Variants of a `oneOf`/`anyOf`/`allOf` and the `items` of an array are yielded under the + name of the property that declares them. + """ + if name is not None: + yield pointer, name, schema + + properties = schema.get("properties") + if isinstance(properties, Mapping): + for property_name, property_schema in properties.items(): + if isinstance(property_schema, Mapping): + yield from _iter_named_properties( + property_schema, + f"{pointer}/properties/{property_name}", + property_name, + ) + + for keyword in ("oneOf", "anyOf", "allOf"): + variants = schema.get(keyword) + if isinstance(variants, list): + for index, variant in enumerate(variants): + if isinstance(variant, Mapping): + yield from _iter_named_properties(variant, f"{pointer}/{keyword}/{index}", name) + + items = schema.get("items") + if isinstance(items, Mapping): + yield from _iter_named_properties(items, f"{pointer}/items", name) + + +def find_secret_marking_errors(connection_specification: Mapping[str, Any]) -> list[str]: + """Return one error per property whose `airbyte_secret` flag is wrong. + + Any `airbyte_secret` value must be a boolean: the CDK's secret filter only masks values + whose flag is `True`, so a string such as `"true"` leaves the value unmasked. A property + whose name denotes a secret (see `is_secret_property_name`) and that can hold a secret + value must set `airbyte_secret: true`. One that cannot hold a secret value, such as a + boolean or an object, must not set it. + """ + errors: list[str] = [] + for pointer, name, property_schema in _iter_named_properties(connection_specification): + marking = property_schema.get("airbyte_secret") + if marking is not None and not isinstance(marking, bool): + errors.append( + f"`{pointer}` sets `airbyte_secret` to `{json.dumps(marking)}`, which is not a " + "boolean, so the CDK does not treat the value as a secret." + ) + continue + if "type" not in property_schema or not is_secret_property_name(name): + continue + + marked_as_secret = marking is True + can_hold_secret = _can_hold_secret(property_schema) + if can_hold_secret and not marked_as_secret: + errors.append( + f"`{pointer}` looks like a secret but is not marked `airbyte_secret: true`." + ) + elif marked_as_secret and not can_hold_secret: + errors.append( + f"`{pointer}` is marked `airbyte_secret: true` but its type " + f"`{property_schema.get('type')}` cannot hold a secret value." + ) + return errors + + +def get_single_spec(result: EntrypointOutput, *, connector_name: str) -> ConnectorSpecification: + """Assert that `spec` emitted exactly one SPEC message, and return its specification.""" + spec_messages = result.spec_messages + assert len(spec_messages) == 1, ( + f"`spec` for connector '{connector_name}' emitted {len(spec_messages)} SPEC messages, " + f"expected exactly 1. Logs: {result.logs}" + ) + spec = spec_messages[0].spec + assert spec is not None, ( + f"`spec` for connector '{connector_name}' emitted an empty SPEC message." + ) + return spec + + +def assert_spec_is_valid(spec: ConnectorSpecification, *, connector_name: str) -> None: + """Assert that a connector spec passes every credential-free lint.""" + errors = find_secret_marking_errors(spec.connectionSpecification) + assert not errors, ( + f"The spec of connector '{connector_name}' marks secrets incorrectly. A property that " + "holds a credential must set `airbyte_secret: true` so the platform masks it in the UI " + "and logs and stores it in the secrets manager:\n" + "\n".join(errors) + ) + + +def _string_values(value: Any) -> Iterator[str]: + """Yield every string found in a JSON-like value, recursively.""" + if isinstance(value, str): + yield value + elif isinstance(value, Mapping): + for item in value.values(): + yield from _string_values(item) + elif isinstance(value, list): + for item in value: + yield from _string_values(item) + + +def _searchable_secrets(secrets: Iterable[Any]) -> list[str]: + """Return the secret values worth searching for in the output. + + Only values of at least `_MIN_LEAK_CHECK_LENGTH` characters that contain a digit are + kept. Shorter or word-like values, such as the `invalid_api_key` placeholders found in + test configs, appear in ordinary output by coincidence. + """ + searchable: list[str] = [] + for secret in secrets: + if isinstance(secret, bool) or not isinstance(secret, (str, int, float)): + continue + value = str(secret) + if len(value) >= _MIN_LEAK_CHECK_LENGTH and re.search(r"\d", value): + searchable.append(value) + return searchable + + +def _leak_excerpt(text: str, secrets: Iterable[Any]) -> str: + """Return an excerpt of `text` around its first secret, with every secret masked. + + Secrets are masked longest first so that a secret containing another one is masked whole, + and before the excerpt is cut so that the cut cannot expose part of a secret. + """ + secret_values = sorted( + {str(secret) for secret in secrets if secret and not isinstance(secret, bool)}, + key=len, + reverse=True, + ) + for secret in secret_values: + text = text.replace(secret, _MASK) + start = max(text.find(_MASK) - _EXCERPT_CONTEXT, 0) + return text[start : start + 2 * _EXCERPT_CONTEXT] + + +def find_leaked_secrets(output: EntrypointOutput, secrets: Iterable[Any]) -> list[str]: + """Return a masked excerpt of each message in `output` that contains one of `secrets`. + + CONTROL messages are skipped: a connector config update carries the full config, + including its secrets, by design. + """ + secrets = list(secrets) + searchable_secrets = _searchable_secrets(secrets) + if not searchable_secrets: + return [] + + leaks: list[str] = [] + for message in output.get_message_iterator(): + if message.type == Type.CONTROL: + continue + for text in _string_values(AirbyteMessageSerializer.dump(message)): + if any(secret in text for secret in searchable_secrets): + leaks.append(f"{message.type.value} message: {_leak_excerpt(text, secrets)}") + break + return leaks + + +def assert_no_secrets_in_output( + output: EntrypointOutput, + *, + spec: ConnectorSpecification, + config: Mapping[str, Any], + verb: str, + connector_name: str, +) -> None: + """Assert that no `airbyte_secret` value from `config` appears in the connector's output.""" + secrets = get_secrets(spec.connectionSpecification, config) + leaks = find_leaked_secrets(output, secrets) + assert not leaks, ( + f"`{verb}` for connector '{connector_name}' printed secret values from its config. " + "Secrets must never appear in logs, traces or connection status messages:\n" + + "\n".join(leaks) + ) diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py index 1c9bf24e10..acbd6d4352 100644 --- a/airbyte_cdk/test/standard_tests/connector_base.py +++ b/airbyte_cdk/test/standard_tests/connector_base.py @@ -15,6 +15,10 @@ ConnectorTestScenario, ) from airbyte_cdk.test.standard_tests._job_runner import IConnector, run_test_job +from airbyte_cdk.test.standard_tests._spec_lint import ( + assert_no_secrets_in_output, + get_single_spec, +) from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite if TYPE_CHECKING: @@ -100,6 +104,30 @@ def create_connector( "override `cls.create_connector()` to define a custom initialization process." ) + def assert_no_secrets_in_output( + self, + scenario: ConnectorTestScenario, + output: entrypoint_wrapper.EntrypointOutput, + *, + verb: str, + ) -> None: + """Assert that no `airbyte_secret` value from the scenario's config is in `output`.""" + spec_result = run_test_job( + self.create_connector(scenario), + "spec", + connector_root=self.get_connector_root_dir(), + ) + assert_no_secrets_in_output( + output, + spec=get_single_spec(spec_result, connector_name=self.connector_name), + config=scenario.get_config_dict( + connector_root=self.get_connector_root_dir(), + empty_if_missing=True, + ), + verb=verb, + connector_name=self.connector_name, + ) + # Test Definitions def test_check( @@ -117,3 +145,4 @@ def test_check( "Expected exactly one CONNECTION_STATUS message. " f"Got: {result.connection_status_messages!s}" ) + self.assert_no_secrets_in_output(scenario, result, verb="check") diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index 84cfab2b00..08ce09ff5d 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -21,6 +21,7 @@ AirbyteCatalog, ConfiguredAirbyteCatalog, ConfiguredAirbyteStream, + ConnectorSpecification, DestinationSyncMode, Status, SyncMode, @@ -28,6 +29,11 @@ from airbyte_cdk.models.connector_metadata import MetadataFile from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput from airbyte_cdk.test.models import ConnectorTestScenario, ExpectedOutcome +from airbyte_cdk.test.standard_tests._spec_lint import ( + assert_no_secrets_in_output, + assert_spec_is_valid, + get_single_spec, +) from airbyte_cdk.utils.connector_paths import ( ACCEPTANCE_TEST_CONFIG, find_connector_root, @@ -76,6 +82,21 @@ def _assert_check_outcome( ) +def _run_docker_spec(connector_image: str, *, connector_name: str) -> ConnectorSpecification: + """Run `spec` in the connector image and return the single spec it emits.""" + spec_result = run_docker_airbyte_command( + [ + "docker", + "run", + "--rm", + connector_image, + "spec", + ], + raise_if_errors=True, + ) + return get_single_spec(spec_result, connector_name=connector_name) + + class DockerConnectorTestSuite: """Base class for connector test suites.""" @@ -227,7 +248,12 @@ def test_docker_image_build_and_spec( connector_image_override: str | None, connector_base_image_override: str | None, ) -> None: - """Run `docker_image` acceptance tests.""" + """Run `spec` in the connector image and lint the spec it emits. + + The image must emit exactly one SPEC message. `airbyte-cdk image test` builds the + image with `no_verify=True`, which skips the build-time spec check, so this test is + what catches an image that emits no spec on that path. + """ connector_root = self.get_connector_root_dir().absolute() metadata = MetadataFile.from_file(connector_root / "metadata.yaml") @@ -243,16 +269,8 @@ def test_docker_image_build_and_spec( base_image_override=connector_base_image_override, ) - _ = run_docker_airbyte_command( - [ - "docker", - "run", - "--rm", - connector_image, - "spec", - ], - raise_if_errors=True, - ) + spec = _run_docker_spec(connector_image, connector_name=connector_root.name) + assert_spec_is_valid(spec, connector_name=connector_root.name) @pytest.mark.skipif( shutil.which("docker") is None, @@ -319,6 +337,14 @@ def test_docker_image_build_and_check( expected_outcome=scenario.expected_outcome, connector_name=connector_root.absolute().name, ) + connector_name = connector_root.absolute().name + assert_no_secrets_in_output( + check_result, + spec=_run_docker_spec(connector_image, connector_name=connector_name), + config=scenario.get_config_dict(connector_root=connector_root, empty_if_missing=True), + verb="check", + connector_name=connector_name, + ) @pytest.mark.skipif( shutil.which("docker") is None, diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py index ce9b861726..1580b83af4 100644 --- a/airbyte_cdk/test/standard_tests/source_base.py +++ b/airbyte_cdk/test/standard_tests/source_base.py @@ -19,6 +19,7 @@ ConnectorTestScenario, ) from airbyte_cdk.test.standard_tests._job_runner import run_test_job +from airbyte_cdk.test.standard_tests._spec_lint import assert_spec_is_valid, get_single_spec from airbyte_cdk.test.standard_tests.connector_base import ( ConnectorTestSuiteBase, ) @@ -55,6 +56,7 @@ def test_check( f"Expected exactly one CONNECTION_STATUS message. Got {num_status_messages}: \n" + "\n".join([str(m) for m in result.get_message_iterator()]) ) + self.assert_no_secrets_in_output(scenario, result, verb="check") def test_discover( self, @@ -82,7 +84,7 @@ def test_spec(self) -> None: does not require any inputs. We assume `spec` should always succeed and it should always generate - a valid `SPEC` message. + a single valid `SPEC` message, which must pass the credential-free spec lint. Note: the parsing of messages by type also implicitly validates that the generated `SPEC` message is valid JSON. @@ -95,10 +97,8 @@ def test_spec(self) -> None: ) # If an error occurs, it will be raised above. - assert len(result.spec_messages) == 1, ( - f"Expected exactly 1 spec message but got {len(result.spec_messages)}. " - f"Errors: {result.errors!s}" - ) + spec = get_single_spec(result, connector_name=self.connector_name) + assert_spec_is_valid(spec, connector_name=self.connector_name) def test_basic_read( self, diff --git a/unit_tests/test/test_spec_lint.py b/unit_tests/test/test_spec_lint.py new file mode 100644 index 0000000000..f51f33da95 --- /dev/null +++ b/unit_tests/test/test_spec_lint.py @@ -0,0 +1,382 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Unit tests for the credential-free spec lint in `airbyte_cdk.test.standard_tests._spec_lint`.""" + +import json +from pathlib import Path +from typing import Any + +import pytest + +from airbyte_cdk.models import ConnectorSpecification +from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput +from airbyte_cdk.test.models import ConnectorTestScenario +from airbyte_cdk.test.standard_tests import docker_base +from airbyte_cdk.test.standard_tests._spec_lint import ( + assert_no_secrets_in_output, + assert_spec_is_valid, + find_leaked_secrets, + find_secret_marking_errors, + get_single_spec, + is_secret_property_name, +) +from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite + +SECRET = "sk_live_51UWRAsFuIbeygfIY3" +POKEAPI_CONNECTOR_ROOT = ( + Path(__file__).parent.parent / "resources" / "source_pokeapi_w_components_py" +) + + +def _spec(properties: dict[str, Any]) -> dict[str, Any]: + return {"type": "object", "properties": properties} + + +def _output(*messages: dict[str, Any]) -> EntrypointOutput: + return EntrypointOutput(messages=[json.dumps(message) for message in messages]) + + +def _log(message: str) -> dict[str, Any]: + return {"type": "LOG", "log": {"level": "INFO", "message": message}} + + +def _spec_message(properties: dict[str, Any]) -> dict[str, Any]: + return {"type": "SPEC", "spec": {"connectionSpecification": _spec(properties)}} + + +@pytest.mark.parametrize( + "name, expected", + [ + pytest.param("password", True, id="exact_name"), + pytest.param("Client_Secret", True, id="case_insensitive"), + pytest.param("api_key", True, id="api_key_suffix"), + pytest.param("x-api-key", True, id="dashes_normalized"), + pytest.param("personal_access_token", True, id="token_suffix"), + pytest.param("aws_secret_access_key", True, id="access_key_suffix"), + pytest.param("tunnel_user_password", True, id="password_suffix"), + pytest.param("tenant_id", False, id="tenant_id_is_an_identifier"), + pytest.param("app_id", False, id="app_id_is_an_identifier"), + pytest.param("token_expiry_date", False, id="token_prefix_only"), + pytest.param("primary_key", False, id="primary_key"), + pytest.param("access_key_id", False, id="access_key_id"), + pytest.param("start_date", False, id="unrelated"), + ], +) +def test_is_secret_property_name(name: str, expected: bool) -> None: + assert is_secret_property_name(name) == expected + + +@pytest.mark.parametrize( + "properties, expected_errors", + [ + pytest.param( + {"api_key": {"type": "string", "airbyte_secret": True}}, + [], + id="marked_secret_passes", + ), + pytest.param( + {"start_date": {"type": "string"}, "tenant_id": {"type": "string"}}, + [], + id="non_secret_names_pass", + ), + pytest.param( + {"api_key": {"type": "string"}}, + ["`/properties/api_key` looks like a secret"], + id="unmarked_api_key_fails", + ), + pytest.param( + {"password": {"type": ["null", "integer"]}}, + ["`/properties/password` looks like a secret"], + id="unmarked_nullable_type_list_fails", + ), + pytest.param( + { + "credentials": { + "type": "object", + "oneOf": [ + { + "type": "object", + "properties": { + "auth_type": {"type": "string", "const": "oauth2.0"}, + "client_secret": {"type": "string", "airbyte_secret": True}, + }, + }, + { + "type": "object", + "properties": { + "auth_type": {"type": "string", "const": "api_key"}, + "access_token": {"type": "string"}, + }, + }, + ], + } + }, + ["`/properties/credentials/oneOf/1/properties/access_token` looks like a secret"], + id="unmarked_secret_inside_one_of_fails", + ), + pytest.param( + {"tokens": {"type": "array", "items": {"type": "string"}}}, + [], + id="array_of_non_secret_name_passes", + ), + pytest.param( + {"refresh_token": {"type": "array", "items": {"type": "string"}}}, + ["`/properties/refresh_token/items` looks like a secret"], + id="unmarked_items_of_secret_array_fail", + ), + pytest.param( + {"token": {"type": "string", "const": "fixed"}}, + [], + id="const_secret_name_passes", + ), + pytest.param( + {"password": {"type": "boolean", "airbyte_secret": True}}, + ["`/properties/password` is marked `airbyte_secret: true` but its type `boolean`"], + id="marked_boolean_fails", + ), + pytest.param( + {"credentials": {"type": "object", "airbyte_secret": True, "properties": {}}}, + ["`/properties/credentials` is marked `airbyte_secret: true` but its type `object`"], + id="marked_object_fails", + ), + pytest.param( + {"credentials": {"type": "object", "properties": {"type": {"type": "string"}}}}, + [], + id="property_named_type_inside_secret_object_passes", + ), + pytest.param( + {"access_key": {"type": "string", "airbyte_secret": "true"}}, + ['`/properties/access_key` sets `airbyte_secret` to `"true"`, which is not a boolean'], + id="string_marking_on_secret_name_fails_once", + ), + pytest.param( + {"deployment_url": {"type": "string", "airbyte_secret": "yes"}}, + [ + '`/properties/deployment_url` sets `airbyte_secret` to `"yes"`, which is not a boolean' + ], + id="string_marking_on_any_name_fails", + ), + pytest.param( + {"start_date": {"type": "string", "airbyte_secret": False}}, + [], + id="false_marking_on_non_secret_name_passes", + ), + ], +) +def test_find_secret_marking_errors( + properties: dict[str, Any], + expected_errors: list[str], +) -> None: + errors = find_secret_marking_errors(_spec(properties)) + assert len(errors) == len(expected_errors), errors + for error, expected_error in zip(errors, expected_errors): + assert error.startswith(expected_error) + + +def test_assert_spec_is_valid_lists_every_unmarked_secret() -> None: + spec = ConnectorSpecification( + connectionSpecification=_spec( + {"api_key": {"type": "string"}, "client_secret": {"type": "string"}} + ) + ) + with pytest.raises(AssertionError) as error: + assert_spec_is_valid(spec, connector_name="source-test") + assert "/properties/api_key" in str(error.value) + assert "/properties/client_secret" in str(error.value) + + +@pytest.mark.parametrize("spec_count", [0, 2]) +def test_get_single_spec_requires_exactly_one_spec_message(spec_count: int) -> None: + output = _output(*[_spec_message({}) for _ in range(spec_count)], _log("done")) + with pytest.raises(AssertionError, match=f"emitted {spec_count} SPEC messages"): + get_single_spec(output, connector_name="source-test") + + +def test_get_single_spec_returns_the_spec() -> None: + spec = get_single_spec(_output(_spec_message({"a": {"type": "string"}})), connector_name="x") + assert spec.connectionSpecification["properties"] == {"a": {"type": "string"}} + + +@pytest.mark.parametrize( + "message", + [ + pytest.param(_log(f"Calling https://api.example.com/?key={SECRET}"), id="log"), + pytest.param( + { + "type": "TRACE", + "trace": { + "type": "ERROR", + "emitted_at": 0, + "error": {"message": "Request failed", "stack_trace": f"token={SECRET}"}, + }, + }, + id="trace_stack_trace", + ), + pytest.param( + { + "type": "CONNECTION_STATUS", + "connectionStatus": {"status": "FAILED", "message": f"401 for {SECRET}"}, + }, + id="connection_status", + ), + ], +) +def test_find_leaked_secrets_reports_masked_excerpt(message: dict[str, Any]) -> None: + leaks = find_leaked_secrets(_output(message), [SECRET]) + assert len(leaks) == 1 + assert leaks[0].startswith(message["type"]) + assert "****" in leaks[0] + assert SECRET not in leaks[0] + + +def test_find_leaked_secrets_ignores_control_messages() -> None: + control_message = { + "type": "CONTROL", + "control": { + "type": "CONNECTOR_CONFIG", + "emitted_at": 0, + "connectorConfig": {"config": {"api_key": SECRET}}, + }, + } + assert find_leaked_secrets(_output(control_message), [SECRET]) == [] + + +@pytest.mark.parametrize( + "secret, text", + [ + pytest.param("123", "Fetched 123 records", id="too_short"), + pytest.param("invalid_api_key", "Got invalid_api_key from the API", id="no_digit"), + pytest.param(True, "True", id="boolean"), + pytest.param({"nested": SECRET}, SECRET, id="not_a_scalar"), + ], +) +def test_find_leaked_secrets_skips_values_that_match_by_coincidence(secret: Any, text: str) -> None: + assert find_leaked_secrets(_output(_log(text)), [secret]) == [] + + +def test_leak_excerpt_masks_overlapping_secrets_whole() -> None: + short_secret = "abc12345" + long_secret = short_secret + "6789xyz" + leaks = find_leaked_secrets(_output(_log(f"token={long_secret}")), [short_secret, long_secret]) + assert leaks == ["LOG message: token=****"] + + +def test_leak_excerpt_is_centered_on_the_leak() -> None: + text = "x" * 1000 + SECRET + "y" * 1000 + [leak] = find_leaked_secrets(_output(_log(text)), [SECRET]) + assert "****" in leak + assert len(leak) < 400 + + +def test_assert_no_secrets_in_output_reads_secret_paths_from_the_spec() -> None: + spec = ConnectorSpecification( + connectionSpecification=_spec( + { + "credentials": { + "type": "object", + "oneOf": [ + { + "type": "object", + "properties": {"api_key": {"type": "string", "airbyte_secret": True}}, + } + ], + }, + "account_id": {"type": "string"}, + } + ) + ) + config = {"credentials": {"api_key": SECRET}, "account_id": "acct_12345678"} + + # The non-secret account ID may appear in the output. + assert_no_secrets_in_output( + _output(_log("Connected to acct_12345678")), + spec=spec, + config=config, + verb="check", + connector_name="source-test", + ) + with pytest.raises(AssertionError, match="`check` for connector 'source-test' printed"): + assert_no_secrets_in_output( + _output(_log(f"Connected with {SECRET}")), + spec=spec, + config=config, + verb="check", + connector_name="source-test", + ) + + +class _DockerSuite(DockerConnectorTestSuite): + @classmethod + def get_connector_root_dir(cls) -> Path: + return POKEAPI_CONNECTOR_ROOT + + +def _fake_docker( + monkeypatch: pytest.MonkeyPatch, + outputs: dict[str, EntrypointOutput], +) -> list[list[str]]: + """Make `run_docker_airbyte_command` return `outputs[verb]` and record each command.""" + commands: list[list[str]] = [] + + def run_docker_airbyte_command(cmd: list[str], *, raise_if_errors: bool) -> EntrypointOutput: + commands.append(cmd) + verb = next(verb for verb in outputs if verb in cmd) + return outputs[verb] + + monkeypatch.setattr(docker_base, "run_docker_airbyte_command", run_docker_airbyte_command) + return commands + + +def test_docker_spec_test_fails_when_the_image_emits_no_spec( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _fake_docker(monkeypatch, {"spec": _output(_log("starting up"))}) + with pytest.raises(AssertionError, match="emitted 0 SPEC messages"): + _DockerSuite().test_docker_image_build_and_spec( + connector_image_override="source-test:dev", + connector_base_image_override=None, + ) + + +def test_docker_spec_test_lints_the_spec(monkeypatch: pytest.MonkeyPatch) -> None: + _fake_docker(monkeypatch, {"spec": _output(_spec_message({"api_key": {"type": "string"}}))}) + with pytest.raises(AssertionError, match="/properties/api_key"): + _DockerSuite().test_docker_image_build_and_spec( + connector_image_override="source-test:dev", + connector_base_image_override=None, + ) + + +@pytest.mark.parametrize( + "status_message, should_pass", + [ + pytest.param("Invalid API key", True, id="no_leak"), + pytest.param(f"Invalid API key {SECRET}", False, id="leak"), + ], +) +def test_docker_check_test_fails_when_check_prints_a_secret( + monkeypatch: pytest.MonkeyPatch, + status_message: str, + should_pass: bool, +) -> None: + secret_spec = _spec_message({"api_key": {"type": "string", "airbyte_secret": True}}) + check_output = _output( + { + "type": "CONNECTION_STATUS", + "connectionStatus": {"status": "FAILED", "message": status_message}, + } + ) + _fake_docker(monkeypatch, {"spec": _output(secret_spec), "check": check_output}) + scenario = ConnectorTestScenario(config_dict={"api_key": SECRET}, status="failed") + + def run_check() -> None: + _DockerSuite().test_docker_image_build_and_check( + scenario=scenario, + connector_image_override="source-test:dev", + connector_base_image_override=None, + ) + + if should_pass: + run_check() + else: + with pytest.raises(AssertionError, match="printed secret values"): + run_check() From da368209331393e230993d31ee2036da7f5167dc Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Fri, 9 Oct 2026 14:59:59 +0300 Subject: [PATCH 2/8] fix(standard-tests): report leaked secrets by config path, never by excerpt The leak report printed a masked excerpt of the message that held a secret. Masking replaced only the literal value of each secret, so a second secret in the same message passed through whenever it appeared in another form: a PEM key or service-account JSON inside a json.dumps(config), a repr, or a URL. The failure text reaches the JUnit XML that CI publishes, outside the runner's log masking. The report now holds no text from the output at all. Each line names the config paths of the secrets found and where they were found: the message number, its type and its stream. The stream name is left out when it overlaps any secret. Detection also searches the JSON-escaped, repr-escaped and URL-encoded forms of each secret and each line of a multi-line secret, so a key that only appears escaped is still caught. The secrets now come from a local walker that also descends into array items and every anyOf/allOf/oneOf variant. get_secrets, which the runtime log filter uses, skips those, so a per-account API key such as source-onesignal's applications[].app_api_key was never searched. The walker builds paths from spec property names and array indices only. get_secrets itself is unchanged. Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/test/standard_tests/_spec_lint.py | 202 ++++++++--- unit_tests/test/test_spec_lint.py | 338 ++++++++++++++++-- 2 files changed, 454 insertions(+), 86 deletions(-) diff --git a/airbyte_cdk/test/standard_tests/_spec_lint.py b/airbyte_cdk/test/standard_tests/_spec_lint.py index 2fbd45b61b..087b97cac9 100644 --- a/airbyte_cdk/test/standard_tests/_spec_lint.py +++ b/airbyte_cdk/test/standard_tests/_spec_lint.py @@ -12,10 +12,10 @@ import re from collections.abc import Iterable, Iterator, Mapping from typing import Any +from urllib.parse import quote from airbyte_cdk.models import AirbyteMessageSerializer, ConnectorSpecification, Type from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput -from airbyte_cdk.utils.airbyte_secrets_utils import get_secrets SECRET_PROPERTY_NAMES = frozenset( { @@ -59,8 +59,7 @@ _SECRET_CAPABLE_TYPES = {"string", "integer", "number"} -_MASK = "****" -_EXCERPT_CONTEXT = 150 +_COMBINATORS = ("oneOf", "anyOf", "allOf") _MIN_LEAK_CHECK_LENGTH = 8 """Secret values shorter than this are not searched for in the output. @@ -187,71 +186,178 @@ def assert_spec_is_valid(spec: ConnectorSpecification, *, connector_name: str) - ) -def _string_values(value: Any) -> Iterator[str]: - """Yield every string found in a JSON-like value, recursively.""" - if isinstance(value, str): - yield value - elif isinstance(value, Mapping): - for item in value.values(): - yield from _string_values(item) +def _pointer_token(name: str) -> str: + return name.replace("~", "~0").replace("/", "~1") + + +def _scalars(value: Any) -> Iterator[Any]: + """Yield every scalar inside a JSON-like value, mapping keys included.""" + if isinstance(value, Mapping): + for key, item in value.items(): + yield key + yield from _scalars(item) elif isinstance(value, list): for item in value: - yield from _string_values(item) + yield from _scalars(item) + elif value is not None: + yield value + +def _iter_config_secrets( + schema: Any, + config: Any, + pointer: str = "", +) -> Iterator[tuple[str, Any]]: + """Yield `(config_pointer, value)` for each value in `config` that `schema` marks secret. -def _searchable_secrets(secrets: Iterable[Any]) -> list[str]: - """Return the secret values worth searching for in the output. + The pointer is built from spec property names and array indices only, never from config + keys or values, so it is safe to print. Every scalar inside a secret object or array is + yielded under the pointer of that object or array. + """ + if not isinstance(schema, Mapping): + return + if schema.get("airbyte_secret") is True: + for value in _scalars(config): + yield pointer, value + + properties = schema.get("properties") + if isinstance(properties, Mapping) and isinstance(config, Mapping): + for name, property_schema in properties.items(): + if name in config: + yield from _iter_config_secrets( + property_schema, config[name], f"{pointer}/{_pointer_token(name)}" + ) + + for keyword in _COMBINATORS: + variants = schema.get(keyword) + if isinstance(variants, list): + for variant in variants: + yield from _iter_config_secrets(variant, config, pointer) + + items = schema.get("items") + if isinstance(items, Mapping) and isinstance(config, list): + for index, item in enumerate(config): + yield from _iter_config_secrets(items, item, f"{pointer}/{index}") + + +def find_config_secrets( + connection_specification: Mapping[str, Any], + config: Mapping[str, Any], +) -> list[tuple[str, Any]]: + """Return `(config_pointer, value)` for every `airbyte_secret` value in `config`. + Unlike `airbyte_cdk.utils.airbyte_secrets_utils.get_secrets`, which the CDK uses to mask + logs at runtime, this also walks array `items` and every `anyOf`/`allOf`/`oneOf` variant, + so it finds a secret in an array of objects, such as one API key per account. It finds + every value `get_secrets` finds for the spec shapes connectors use. + """ + return list(dict.fromkeys(_iter_config_secrets(connection_specification, config))) + + +def _is_searchable(value: str) -> bool: + return len(value) >= _MIN_LEAK_CHECK_LENGTH and re.search(r"\d", value) is not None + + +def _secret_variants(secret: Any) -> set[str]: + """Return every form of `secret` worth searching for in the output. + + Besides the literal value, a secret shows up JSON-escaped when a connector dumps its + config, `repr`-escaped when it formats a Python object, and URL-encoded in a logged URL. + Each line of a multi-line secret, such as a PEM private key, is searched on its own too. Only values of at least `_MIN_LEAK_CHECK_LENGTH` characters that contain a digit are - kept. Shorter or word-like values, such as the `invalid_api_key` placeholders found in + kept: shorter or word-like values, such as the `invalid_api_key` placeholders found in test configs, appear in ordinary output by coincidence. """ - searchable: list[str] = [] - for secret in secrets: - if isinstance(secret, bool) or not isinstance(secret, (str, int, float)): - continue - value = str(secret) - if len(value) >= _MIN_LEAK_CHECK_LENGTH and re.search(r"\d", value): - searchable.append(value) - return searchable + if isinstance(secret, bool) or not isinstance(secret, (str, int, float)): + return set() + value = str(secret) + if not _is_searchable(value): + return set() + variants = {value, json.dumps(value)[1:-1], repr(value)[1:-1], quote(value, safe="")} + variants.update(line for line in re.split(r"\r?\n|\\n", value) if _is_searchable(line)) + return variants + + +def _stream_name(message: Mapping[str, Any]) -> str | None: + """Return the name of the stream a serialized message is about, if any.""" + record = message.get("record") + if isinstance(record, Mapping) and isinstance(record.get("stream"), str): + return str(record["stream"]) + for value in message.values(): + if isinstance(value, Mapping): + descriptor = value.get("stream_descriptor") + if isinstance(descriptor, Mapping) and isinstance(descriptor.get("name"), str): + return str(descriptor["name"]) + nested_name = _stream_name(value) + if nested_name is not None: + return nested_name + return None -def _leak_excerpt(text: str, secrets: Iterable[Any]) -> str: - """Return an excerpt of `text` around its first secret, with every secret masked. +def _string_values(value: Any) -> Iterator[str]: + """Yield every string found in a JSON-like value, recursively.""" + if isinstance(value, str): + yield value + elif isinstance(value, Mapping): + for item in value.values(): + yield from _string_values(item) + elif isinstance(value, list): + for item in value: + yield from _string_values(item) - Secrets are masked longest first so that a secret containing another one is masked whole, - and before the excerpt is cut so that the cut cannot expose part of a secret. - """ - secret_values = sorted( - {str(secret) for secret in secrets if secret and not isinstance(secret, bool)}, - key=len, - reverse=True, - ) - for secret in secret_values: - text = text.replace(secret, _MASK) - start = max(text.find(_MASK) - _EXCERPT_CONTEXT, 0) - return text[start : start + 2 * _EXCERPT_CONTEXT] +def find_leaked_secrets( + output: EntrypointOutput, + secrets: Iterable[tuple[str, Any]], +) -> list[str]: + """Return one line per place in `output` that contains a secret from `secrets`. -def find_leaked_secrets(output: EntrypointOutput, secrets: Iterable[Any]) -> list[str]: - """Return a masked excerpt of each message in `output` that contains one of `secrets`. + `secrets` holds `(config_pointer, value)` pairs, as `find_config_secrets` returns. A line names the pointers of the secrets found and where they were found: the + message number (on the Docker path, the stdout line number), and its type and stream. It + never contains any text from the output, so it cannot print a + secret in any encoding. The stream name is left out if it overlaps any secret. CONTROL messages are skipped: a connector config update carries the full config, including its secrets, by design. """ secrets = list(secrets) - searchable_secrets = _searchable_secrets(secrets) - if not searchable_secrets: + variants_by_pointer: dict[str, set[str]] = {} + for pointer, secret in secrets: + if variants := _secret_variants(secret): + variants_by_pointer.setdefault(pointer, set()).update(variants) + if not variants_by_pointer: return [] + every_secret_form = {str(secret) for _, secret in secrets if str(secret)}.union( + *variants_by_pointer.values() + ) + + def leaked_pointers(text: str) -> list[str]: + return [ + pointer + for pointer, variants in variants_by_pointer.items() + if any(variant in text for variant in variants) + ] + + def describe(pointers: Iterable[str]) -> str: + return ", ".join(f"`{pointer}`" for pointer in sorted(set(pointers))) leaks: list[str] = [] - for message in output.get_message_iterator(): + for number, message in enumerate(output.get_message_iterator(), start=1): if message.type == Type.CONTROL: continue - for text in _string_values(AirbyteMessageSerializer.dump(message)): - if any(secret in text for secret in searchable_secrets): - leaks.append(f"{message.type.value} message: {_leak_excerpt(text, secrets)}") - break + serialized_message = AirbyteMessageSerializer.dump(message) + pointers = [ + pointer + for text in _string_values(serialized_message) + for pointer in leaked_pointers(text) + ] + if not pointers: + continue + location = message.type.value + stream = _stream_name(serialized_message) + if stream and not any(form in stream or stream in form for form in every_secret_form): + location += f", stream `{stream}`" + leaks.append(f"{describe(pointers)} in message #{number} ({location})") return leaks @@ -264,10 +370,10 @@ def assert_no_secrets_in_output( connector_name: str, ) -> None: """Assert that no `airbyte_secret` value from `config` appears in the connector's output.""" - secrets = get_secrets(spec.connectionSpecification, config) + secrets = find_config_secrets(spec.connectionSpecification, config) leaks = find_leaked_secrets(output, secrets) assert not leaks, ( f"`{verb}` for connector '{connector_name}' printed secret values from its config. " - "Secrets must never appear in logs, traces or connection status messages:\n" - + "\n".join(leaks) + "Secrets must never appear in logs, traces or connection status messages. The values " + "at these config paths were found in the output:\n" + "\n".join(leaks) ) diff --git a/unit_tests/test/test_spec_lint.py b/unit_tests/test/test_spec_lint.py index f51f33da95..5179c95678 100644 --- a/unit_tests/test/test_spec_lint.py +++ b/unit_tests/test/test_spec_lint.py @@ -1,6 +1,7 @@ # Copyright (c) 2026 Airbyte, Inc., all rights reserved. """Unit tests for the credential-free spec lint in `airbyte_cdk.test.standard_tests._spec_lint`.""" +import base64 import json from pathlib import Path from typing import Any @@ -14,12 +15,14 @@ from airbyte_cdk.test.standard_tests._spec_lint import ( assert_no_secrets_in_output, assert_spec_is_valid, + find_config_secrets, find_leaked_secrets, find_secret_marking_errors, get_single_spec, is_secret_property_name, ) from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite +from airbyte_cdk.utils.airbyte_secrets_utils import get_secrets SECRET = "sk_live_51UWRAsFuIbeygfIY3" POKEAPI_CONNECTOR_ROOT = ( @@ -197,9 +200,13 @@ def test_get_single_spec_returns_the_spec() -> None: @pytest.mark.parametrize( - "message", + "message, location", [ - pytest.param(_log(f"Calling https://api.example.com/?key={SECRET}"), id="log"), + pytest.param( + _log(f"Calling https://api.example.com/?key={SECRET}"), + "message #1 (LOG)", + id="log", + ), pytest.param( { "type": "TRACE", @@ -209,6 +216,7 @@ def test_get_single_spec_returns_the_spec() -> None: "error": {"message": "Request failed", "stack_trace": f"token={SECRET}"}, }, }, + "message #1 (TRACE)", id="trace_stack_trace", ), pytest.param( @@ -216,16 +224,58 @@ def test_get_single_spec_returns_the_spec() -> None: "type": "CONNECTION_STATUS", "connectionStatus": {"status": "FAILED", "message": f"401 for {SECRET}"}, }, + "message #1 (CONNECTION_STATUS)", id="connection_status", ), + pytest.param( + { + "type": "RECORD", + "record": {"stream": "users", "data": {"token": SECRET}, "emitted_at": 0}, + }, + "message #1 (RECORD, stream `users`)", + id="record_names_the_stream", + ), + pytest.param( + { + "type": "TRACE", + "trace": { + "type": "ERROR", + "emitted_at": 0, + "error": { + "message": f"Failed with {SECRET}", + "stream_descriptor": {"name": "users"}, + }, + }, + }, + "message #1 (TRACE, stream `users`)", + id="trace_names_the_stream", + ), + pytest.param( + { + "type": "RECORD", + "record": {"stream": f"users_{SECRET}", "data": {}, "emitted_at": 0}, + }, + "message #1 (RECORD)", + id="stream_name_holding_a_secret_is_left_out", + ), ], ) -def test_find_leaked_secrets_reports_masked_excerpt(message: dict[str, Any]) -> None: - leaks = find_leaked_secrets(_output(message), [SECRET]) - assert len(leaks) == 1 - assert leaks[0].startswith(message["type"]) - assert "****" in leaks[0] - assert SECRET not in leaks[0] +def test_find_leaked_secrets_reports_where_without_any_text( + message: dict[str, Any], location: str +) -> None: + leaks = find_leaked_secrets(_output(message), [("/api_key", SECRET)]) + assert leaks == [f"`/api_key` in {location}"] + + +def test_find_leaked_secrets_numbers_messages_and_groups_pointers() -> None: + other_secret = "pk_test_99887766554433" + output = _output( + _log("starting"), + _log(f"{SECRET} and {other_secret}"), + _log(f"again {SECRET}"), + ) + leaks = find_leaked_secrets(output, [("/b", other_secret), ("/a", SECRET)]) + assert leaks == ["`/a`, `/b` in message #2 (LOG)", "`/a` in message #3 (LOG)"] def test_find_leaked_secrets_ignores_control_messages() -> None: @@ -237,7 +287,7 @@ def test_find_leaked_secrets_ignores_control_messages() -> None: "connectorConfig": {"config": {"api_key": SECRET}}, }, } - assert find_leaked_secrets(_output(control_message), [SECRET]) == [] + assert find_leaked_secrets(_output(control_message), [("/api_key", SECRET)]) == [] @pytest.mark.parametrize( @@ -246,25 +296,235 @@ def test_find_leaked_secrets_ignores_control_messages() -> None: pytest.param("123", "Fetched 123 records", id="too_short"), pytest.param("invalid_api_key", "Got invalid_api_key from the API", id="no_digit"), pytest.param(True, "True", id="boolean"), - pytest.param({"nested": SECRET}, SECRET, id="not_a_scalar"), + pytest.param(None, "None", id="none"), ], ) def test_find_leaked_secrets_skips_values_that_match_by_coincidence(secret: Any, text: str) -> None: - assert find_leaked_secrets(_output(_log(text)), [secret]) == [] + assert find_leaked_secrets(_output(_log(text)), [("/secret", secret)]) == [] -def test_leak_excerpt_masks_overlapping_secrets_whole() -> None: - short_secret = "abc12345" - long_secret = short_secret + "6789xyz" - leaks = find_leaked_secrets(_output(_log(f"token={long_secret}")), [short_secret, long_secret]) - assert leaks == ["LOG message: token=****"] +_PRIVATE_KEY = ( + "-----BEGIN PRIVATE KEY-----\n" + "MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC7VJTUt9Us8cKj\n" + "MzEfYyjiWA4R4/M2bS1GB4t7NXp98C3SC6dVMvDuictGeurT8jNbvJZHtCSuYEvu\n" + 'NMoSfm76oqFvAp8Gy0iz5sxjZmSnXyCdPEovGhLa0VzMaQ8s+CLOyS56YyCFGeJZ"q\n' + "-----END PRIVATE KEY-----\n" +) +_SERVICE_ACCOUNT_INFO = json.dumps( + { + "type": "service_account", + "project_id": "test-project-424242", + "private_key_id": "0f1e2d3c4b5a69788796a5b4c3d2e1f0aabbccdd", + "private_key": _PRIVATE_KEY, + "client_email": "airbyte@test-project-424242.iam.gserviceaccount.com", + } +) +_MULTI_LINE_SECRETS_SPEC = _spec( + { + "api_key": {"type": "string", "airbyte_secret": True}, + "credentials": { + "type": "object", + "properties": { + "private_key": {"type": "string", "airbyte_secret": True}, + "service_account_info": {"type": "string", "airbyte_secret": True}, + }, + }, + } +) -def test_leak_excerpt_is_centered_on_the_leak() -> None: - text = "x" * 1000 + SECRET + "y" * 1000 - [leak] = find_leaked_secrets(_output(_log(text)), [SECRET]) - assert "****" in leak - assert len(leak) < 400 +def _secret_fragments() -> list[str]: + """Return every piece of the multi-line secrets that the failure text must not contain.""" + fragments: list[str] = [] + for secret in (_PRIVATE_KEY, _SERVICE_ACCOUNT_INFO): + fragments += [ + secret, + json.dumps(secret)[1:-1], + repr(secret)[1:-1], + base64.b64encode(secret.encode()).decode(), + ] + fragments += [line for line in _PRIVATE_KEY.splitlines() if "PRIVATE KEY" not in line] + fragments += [line for line in json.dumps(_PRIVATE_KEY)[1:-1].split("\\n") if line] + fragments += ["0f1e2d3c4b5a69788796a5b4c3d2e1f0aabbccdd", "test-project-424242"] + return fragments + + +@pytest.mark.parametrize( + "printed_text, expected_pointers", + [ + pytest.param( + lambda config: "bad config: " + json.dumps(config), + ["/api_key", "/credentials/private_key", "/credentials/service_account_info"], + id="json_dumped_config", + ), + pytest.param( + lambda config: f"bad config: {config!r}", + ["/api_key", "/credentials/private_key", "/credentials/service_account_info"], + id="repr_of_config", + ), + pytest.param( + lambda config: "key: " + json.dumps(config["credentials"]["private_key"]), + ["/credentials/private_key", "/credentials/service_account_info"], + id="json_escaped_key_alone", + ), + pytest.param( + lambda config: "line: " + config["credentials"]["private_key"].splitlines()[2], + ["/credentials/private_key", "/credentials/service_account_info"], + id="single_line_of_the_key", + ), + ], +) +def test_leak_report_contains_no_part_of_a_multi_line_secret( + printed_text: Any, + expected_pointers: list[str], +) -> None: + config = { + "api_key": SECRET, + "credentials": { + "private_key": _PRIVATE_KEY, + "service_account_info": _SERVICE_ACCOUNT_INFO, + "auth_type": "service_account", + }, + } + text = printed_text(config) + output = EntrypointOutput( + messages=[ + json.dumps( + { + "type": "CONNECTION_STATUS", + "connectionStatus": {"status": "FAILED", "message": text}, + } + ), + json.dumps(_log(base64.b64encode(_PRIVATE_KEY.encode()).decode())), + ], + ) + with pytest.raises(AssertionError) as error: + assert_no_secrets_in_output( + output, + spec=ConnectorSpecification(connectionSpecification=_MULTI_LINE_SECRETS_SPEC), + config=config, + verb="check", + connector_name="source-test", + ) + + failure_text = str(error.value) + pointers = ", ".join(f"`{pointer}`" for pointer in expected_pointers) + assert f"{pointers} in message #1 (CONNECTION_STATUS)" in failure_text + assert "message #2" not in failure_text, "base64 is not searched, so it is not reported" + for fragment in _secret_fragments(): + assert fragment not in failure_text + for line in failure_text.splitlines()[1:]: + assert line.startswith("`/"), line + + +_ONESIGNAL_STYLE_SPEC = _spec( + { + "user_auth_key": {"type": "string", "airbyte_secret": True}, + "applications": { + "type": "array", + "items": { + "type": "object", + "properties": { + "app_id": {"type": "string"}, + "app_api_key": {"type": "string", "airbyte_secret": True}, + }, + }, + }, + } +) + + +@pytest.mark.parametrize( + "spec, config, expected", + [ + pytest.param( + _ONESIGNAL_STYLE_SPEC, + { + "user_auth_key": "uak_0000000000000001", + "applications": [ + {"app_id": "app-1", "app_api_key": "key_0000000000000001"}, + {"app_id": "app-2", "app_api_key": "key_0000000000000002"}, + ], + }, + [ + ("/user_auth_key", "uak_0000000000000001"), + ("/applications/0/app_api_key", "key_0000000000000001"), + ("/applications/1/app_api_key", "key_0000000000000002"), + ], + id="secrets_in_array_items", + ), + pytest.param( + _spec( + { + "auth": { + "anyOf": [ + {"type": "object", "properties": {"token": {"airbyte_secret": True}}}, + {"type": "object", "properties": {"user": {"type": "string"}}}, + ], + "allOf": [ + {"properties": {"password": {"airbyte_secret": True}}}, + ], + }, + } + ), + {"auth": {"token": "t-1", "password": "p-1", "user": "u"}}, + [("/auth/token", "t-1"), ("/auth/password", "p-1")], + id="any_of_and_all_of_variants", + ), + pytest.param( + _spec({"a/b": {"type": "string", "airbyte_secret": True}}), + {"a/b": "s-1"}, + [("/a~1b", "s-1")], + id="pointer_is_escaped", + ), + pytest.param( + _spec({"headers": {"type": "object", "airbyte_secret": True}}), + {"headers": {"X-Token": "tok-1", "extra": ["tok-2"]}}, + [ + ("/headers", "X-Token"), + ("/headers", "tok-1"), + ("/headers", "extra"), + ("/headers", "tok-2"), + ], + id="secret_object_never_names_its_keys", + ), + ], +) +def test_find_config_secrets( + spec: dict[str, Any], config: dict[str, Any], expected: list[tuple[str, Any]] +) -> None: + assert find_config_secrets(spec, config) == expected + + +@pytest.mark.parametrize( + "spec, config", + [ + pytest.param(_ONESIGNAL_STYLE_SPEC, {"user_auth_key": "a", "applications": []}, id="items"), + pytest.param( + _spec( + { + "credentials": { + "type": "object", + "oneOf": [ + {"properties": {"client_secret": {"airbyte_secret": True}}}, + {"properties": {"api_key": {"airbyte_secret": True}}}, + ], + }, + "password": {"type": "string", "airbyte_secret": True}, + } + ), + {"credentials": {"api_key": "k"}, "password": "p"}, + id="one_of", + ), + pytest.param(_MULTI_LINE_SECRETS_SPEC, {"api_key": "k", "credentials": {}}, id="nested"), + ], +) +def test_find_config_secrets_finds_every_runtime_secret( + spec: dict[str, Any], config: dict[str, Any] +) -> None: + """Every value the runtime log filter masks is also searched for in the output.""" + found = {value for _, value in find_config_secrets(spec, config)} + assert set(get_secrets(spec, config)) <= found def test_assert_no_secrets_in_output_reads_secret_paths_from_the_spec() -> None: @@ -346,6 +606,21 @@ def test_docker_spec_test_lints_the_spec(monkeypatch: pytest.MonkeyPatch) -> Non ) +_SECRET_SPEC_MESSAGE = _spec_message({"api_key": {"type": "string", "airbyte_secret": True}}) + + +def _connection_status(status: str, message: str) -> dict[str, Any]: + return {"type": "CONNECTION_STATUS", "connectionStatus": {"status": status, "message": message}} + + +def _run_docker_check(scenario: ConnectorTestScenario) -> None: + _DockerSuite().test_docker_image_build_and_check( + scenario=scenario, + connector_image_override="source-test:dev", + connector_base_image_override=None, + ) + + @pytest.mark.parametrize( "status_message, should_pass", [ @@ -358,25 +633,12 @@ def test_docker_check_test_fails_when_check_prints_a_secret( status_message: str, should_pass: bool, ) -> None: - secret_spec = _spec_message({"api_key": {"type": "string", "airbyte_secret": True}}) - check_output = _output( - { - "type": "CONNECTION_STATUS", - "connectionStatus": {"status": "FAILED", "message": status_message}, - } - ) - _fake_docker(monkeypatch, {"spec": _output(secret_spec), "check": check_output}) + check_output = _output(_connection_status("FAILED", status_message)) + _fake_docker(monkeypatch, {"spec": _output(_SECRET_SPEC_MESSAGE), "check": check_output}) scenario = ConnectorTestScenario(config_dict={"api_key": SECRET}, status="failed") - def run_check() -> None: - _DockerSuite().test_docker_image_build_and_check( - scenario=scenario, - connector_image_override="source-test:dev", - connector_base_image_override=None, - ) - if should_pass: - run_check() + _run_docker_check(scenario) else: with pytest.raises(AssertionError, match="printed secret values"): - run_check() + _run_docker_check(scenario) From 218b0441696a6a0126ea02baf7d9ff940a592e6d Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Fri, 9 Oct 2026 15:00:23 +0300 Subject: [PATCH 3/8] fix(standard-tests): run the secret leak check before the check assertions The leak check ran after the outcome and message-count assertions, and those build their failure text from raw messages: the last CONNECTION_STATUS, the error traces, or every message. A success scenario whose check failed with a token echoed in its status message therefore printed the token, and the leak check never ran. The Docker check test now runs `check` without raising, runs the leak check, and only then raises on errors and asserts the outcome. In-process, run_test_job takes an `inspect_output` callback that it calls before any of its own assertions; both test_check methods pass the leak check through it, so it runs whatever the outcome. The method that does this is renamed to `_assert_no_secrets_in_output`: it was a new public method on a base class that connectors subclass, and it shared its name with the module function it calls. EntrypointOutput gains an optional `stderr`, which run_docker_airbyte_command fills, and the leak check searches it line by line. In-process, a raw print() to stdout is still not captured; the docstring says which output each path can see. The new tests cover the in-process failing path for both suites. Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/test/entrypoint_wrapper.py | 10 ++ .../test/standard_tests/_job_runner.py | 11 +- airbyte_cdk/test/standard_tests/_spec_lint.py | 19 ++- .../test/standard_tests/connector_base.py | 12 +- .../test/standard_tests/docker_base.py | 27 ++-- .../test/standard_tests/source_base.py | 4 +- airbyte_cdk/utils/docker.py | 1 + unit_tests/test/test_spec_lint.py | 141 +++++++++++++++++- 8 files changed, 205 insertions(+), 20 deletions(-) diff --git a/airbyte_cdk/test/entrypoint_wrapper.py b/airbyte_cdk/test/entrypoint_wrapper.py index abe2970308..683e2fffd2 100644 --- a/airbyte_cdk/test/entrypoint_wrapper.py +++ b/airbyte_cdk/test/entrypoint_wrapper.py @@ -89,6 +89,7 @@ def __init__( *, command: list[str] | None = None, message_file: Path | None = None, + stderr: str | None = None, ) -> None: if messages is None and message_file is None: raise ValueError("Either messages or message_file must be provided") @@ -96,6 +97,7 @@ def __init__( raise ValueError("Only one of messages or message_file can be provided") self._command = command + self._stderr = stderr self._messages: list[AirbyteMessage] | None = None self._message_file: Path | None = message_file if messages: @@ -124,6 +126,14 @@ def _parse_message(message: str) -> AirbyteMessage: type=Type.LOG, log=AirbyteLogMessage(level=Level.INFO, message=message) ) + @property + def stderr(self) -> str | None: + """The raw stderr of the connector process, or `None` if it was not captured. + + Only a run in a separate process, such as `run_docker_airbyte_command`, captures it. + """ + return self._stderr + @property def records_and_state_messages( self, diff --git a/airbyte_cdk/test/standard_tests/_job_runner.py b/airbyte_cdk/test/standard_tests/_job_runner.py index ddd293ffa2..bc729190dc 100644 --- a/airbyte_cdk/test/standard_tests/_job_runner.py +++ b/airbyte_cdk/test/standard_tests/_job_runner.py @@ -41,8 +41,14 @@ def run_test_job( connector_root: Path, test_scenario: ConnectorTestScenario | None = None, catalog: ConfiguredAirbyteCatalog | dict[str, Any] | None = None, + inspect_output: Callable[[entrypoint_wrapper.EntrypointOutput], None] | None = None, ) -> entrypoint_wrapper.EntrypointOutput: - """Run a test scenario from provided CLI args and return the result.""" + """Run a test scenario from provided CLI args and return the result. + + `inspect_output`, if given, is called with the output before any assertion on it, so it + runs whatever the outcome. The secret leak check uses it, since the assertions below + print raw messages. + """ # Use default (empty) scenario if not provided: test_scenario = test_scenario or ConnectorTestScenario() @@ -104,6 +110,9 @@ def run_test_job( args=args, expected_outcome=test_scenario.expected_outcome, ) + if inspect_output: + inspect_output(result) + if result.errors and test_scenario.expected_outcome.expect_success(): raise result.as_exception() diff --git a/airbyte_cdk/test/standard_tests/_spec_lint.py b/airbyte_cdk/test/standard_tests/_spec_lint.py index 087b97cac9..5181224060 100644 --- a/airbyte_cdk/test/standard_tests/_spec_lint.py +++ b/airbyte_cdk/test/standard_tests/_spec_lint.py @@ -313,8 +313,8 @@ def find_leaked_secrets( """Return one line per place in `output` that contains a secret from `secrets`. `secrets` holds `(config_pointer, value)` pairs, as `find_config_secrets` returns. A line names the pointers of the secrets found and where they were found: the - message number (on the Docker path, the stdout line number), and its type and stream. It - never contains any text from the output, so it cannot print a + message number (on the Docker path, the stdout line number), its type and stream, or the + stderr line number. It never contains any text from the output, so it cannot print a secret in any encoding. The stream name is left out if it overlaps any secret. CONTROL messages are skipped: a connector config update carries the full config, @@ -358,6 +358,12 @@ def describe(pointers: Iterable[str]) -> str: if stream and not any(form in stream or stream in form for form in every_secret_form): location += f", stream `{stream}`" leaks.append(f"{describe(pointers)} in message #{number} ({location})") + + stderr_lines = (output.stderr or "").splitlines() + for number, line in enumerate(stderr_lines, start=1): + pointers = leaked_pointers(line) + if pointers: + leaks.append(f"{describe(pointers)} in stderr line {number}") return leaks @@ -369,7 +375,14 @@ def assert_no_secrets_in_output( verb: str, connector_name: str, ) -> None: - """Assert that no `airbyte_secret` value from `config` appears in the connector's output.""" + """Assert that no `airbyte_secret` value from `config` appears in the connector's output. + + This searches every message the connector emitted and, on the Docker path, its raw + stdout and stderr. In-process, a `print()` to stdout is not captured: only the messages + the connector yields and the records of the root logger are searched. + + Run this before any other assertion on `output`, since those print raw messages. + """ secrets = find_config_secrets(spec.connectionSpecification, config) leaks = find_leaked_secrets(output, secrets) assert not leaks, ( diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py index acbd6d4352..829e793cb2 100644 --- a/airbyte_cdk/test/standard_tests/connector_base.py +++ b/airbyte_cdk/test/standard_tests/connector_base.py @@ -104,14 +104,18 @@ def create_connector( "override `cls.create_connector()` to define a custom initialization process." ) - def assert_no_secrets_in_output( + def _assert_no_secrets_in_output( self, scenario: ConnectorTestScenario, output: entrypoint_wrapper.EntrypointOutput, *, verb: str, ) -> None: - """Assert that no `airbyte_secret` value from the scenario's config is in `output`.""" + """Assert that no `airbyte_secret` value from the scenario's config is in `output`. + + Pass this to `run_test_job` as `inspect_output`, so it runs before the job's own + assertions print raw messages. + """ spec_result = run_test_job( self.create_connector(scenario), "spec", @@ -140,9 +144,11 @@ def test_check( "check", test_scenario=scenario, connector_root=self.get_connector_root_dir(), + inspect_output=lambda output: self._assert_no_secrets_in_output( + scenario, output, verb="check" + ), ) assert len(result.connection_status_messages) == 1, ( "Expected exactly one CONNECTION_STATUS message. " f"Got: {result.connection_status_messages!s}" ) - self.assert_no_secrets_in_output(scenario, result, verb="check") diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index 08ce09ff5d..cfd74f0c2c 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -323,11 +323,24 @@ def test_docker_image_build_and_check( "--config", container_config_path, ], - # For expected-failure scenarios, a non-zero exit or trace error is an - # acceptable way for `check` to fail; don't raise before we assert on it. - raise_if_errors=not scenario.expected_outcome.expect_exception(), + raise_if_errors=False, ) + # The leak check runs first, whatever the outcome: the assertions below print the raw + # CONNECTION_STATUS and error messages, which are the likeliest to carry a secret. + connector_name = connector_root.absolute().name + assert_no_secrets_in_output( + check_result, + spec=_run_docker_spec(connector_image, connector_name=connector_name), + config=scenario.get_config_dict(connector_root=connector_root, empty_if_missing=True), + verb="check", + connector_name=connector_name, + ) + # For expected-failure scenarios, a non-zero exit or trace error is an acceptable way + # for `check` to fail, so only the other scenarios raise on errors. + if not scenario.expected_outcome.expect_exception(): + check_result.raise_if_errors() + # This makes the image test exercise the connector's actual `check` outcome inside the # container, in both directions (e.g. it fails if bundled custom components are rejected # by the CDK baked into the base image, and it fails if a `check` that is expected to @@ -335,14 +348,6 @@ def test_docker_image_build_and_check( _assert_check_outcome( check_result=check_result, expected_outcome=scenario.expected_outcome, - connector_name=connector_root.absolute().name, - ) - connector_name = connector_root.absolute().name - assert_no_secrets_in_output( - check_result, - spec=_run_docker_spec(connector_image, connector_name=connector_name), - config=scenario.get_config_dict(connector_root=connector_root, empty_if_missing=True), - verb="check", connector_name=connector_name, ) diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py index 1580b83af4..9600c17b0d 100644 --- a/airbyte_cdk/test/standard_tests/source_base.py +++ b/airbyte_cdk/test/standard_tests/source_base.py @@ -50,13 +50,15 @@ def test_check( "check", test_scenario=scenario, connector_root=self.get_connector_root_dir(), + inspect_output=lambda output: self._assert_no_secrets_in_output( + scenario, output, verb="check" + ), ) num_status_messages = len(result.connection_status_messages) assert num_status_messages == 1, ( f"Expected exactly one CONNECTION_STATUS message. Got {num_status_messages}: \n" + "\n".join([str(m) for m in result.get_message_iterator()]) ) - self.assert_no_secrets_in_output(scenario, result, verb="check") def test_discover( self, diff --git a/airbyte_cdk/utils/docker.py b/airbyte_cdk/utils/docker.py index 9e66728573..cf65b149d9 100644 --- a/airbyte_cdk/utils/docker.py +++ b/airbyte_cdk/utils/docker.py @@ -483,6 +483,7 @@ def run_docker_airbyte_command( result_output = EntrypointOutput( command=cmd, messages=process_result.stdout.splitlines(), + stderr=process_result.stderr, uncaught_exception=( subprocess.CalledProcessError( cmd=cmd, diff --git a/unit_tests/test/test_spec_lint.py b/unit_tests/test/test_spec_lint.py index 5179c95678..81c7039f6c 100644 --- a/unit_tests/test/test_spec_lint.py +++ b/unit_tests/test/test_spec_lint.py @@ -3,12 +3,21 @@ import base64 import json +import logging +from collections.abc import Iterator, Mapping from pathlib import Path from typing import Any import pytest -from airbyte_cdk.models import ConnectorSpecification +from airbyte_cdk.models import ( + AirbyteCatalog, + AirbyteConnectionStatus, + AirbyteMessage, + ConnectorSpecification, + Status, +) +from airbyte_cdk.sources.source import Source from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput from airbyte_cdk.test.models import ConnectorTestScenario from airbyte_cdk.test.standard_tests import docker_base @@ -21,7 +30,9 @@ get_single_spec, is_secret_property_name, ) +from airbyte_cdk.test.standard_tests.connector_base import ConnectorTestSuiteBase from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite +from airbyte_cdk.test.standard_tests.source_base import SourceTestSuiteBase from airbyte_cdk.utils.airbyte_secrets_utils import get_secrets SECRET = "sk_live_51UWRAsFuIbeygfIY3" @@ -278,6 +289,14 @@ def test_find_leaked_secrets_numbers_messages_and_groups_pointers() -> None: assert leaks == ["`/a`, `/b` in message #2 (LOG)", "`/a` in message #3 (LOG)"] +def test_find_leaked_secrets_searches_stderr() -> None: + output = EntrypointOutput( + messages=[json.dumps(_log("ok"))], + stderr=f"Traceback (most recent call last):\n KeyError: {SECRET}\n", + ) + assert find_leaked_secrets(output, [("/api_key", SECRET)]) == ["`/api_key` in stderr line 2"] + + def test_find_leaked_secrets_ignores_control_messages() -> None: control_message = { "type": "CONTROL", @@ -397,6 +416,7 @@ def test_leak_report_contains_no_part_of_a_multi_line_secret( ), json.dumps(_log(base64.b64encode(_PRIVATE_KEY.encode()).decode())), ], + stderr=text, ) with pytest.raises(AssertionError) as error: assert_no_secrets_in_output( @@ -410,6 +430,7 @@ def test_leak_report_contains_no_part_of_a_multi_line_secret( failure_text = str(error.value) pointers = ", ".join(f"`{pointer}`" for pointer in expected_pointers) assert f"{pointers} in message #1 (CONNECTION_STATUS)" in failure_text + assert "stderr line 1" in failure_text assert "message #2" not in failure_text, "base64 is not searched, so it is not reported" for fragment in _secret_fragments(): assert fragment not in failure_text @@ -580,6 +601,8 @@ def _fake_docker( def run_docker_airbyte_command(cmd: list[str], *, raise_if_errors: bool) -> EntrypointOutput: commands.append(cmd) verb = next(verb for verb in outputs if verb in cmd) + if raise_if_errors: + outputs[verb].raise_if_errors() return outputs[verb] monkeypatch.setattr(docker_base, "run_docker_airbyte_command", run_docker_airbyte_command) @@ -642,3 +665,119 @@ def test_docker_check_test_fails_when_check_prints_a_secret( else: with pytest.raises(AssertionError, match="printed secret values"): _run_docker_check(scenario) + + +@pytest.mark.parametrize( + "check_output", + [ + pytest.param( + _output(_connection_status("FAILED", f"401 Unauthorized: token {SECRET}")), + id="failed_status_on_a_success_scenario", + ), + pytest.param( + _output( + { + "type": "TRACE", + "trace": { + "type": "ERROR", + "emitted_at": 0, + "error": {"message": "boom", "stack_trace": f"KeyError: {SECRET}"}, + }, + } + ), + id="trace_error_on_a_success_scenario", + ), + pytest.param( + EntrypointOutput( + messages=[json.dumps(_connection_status("SUCCEEDED", "ok"))], + stderr=f"warning: retrying with {SECRET}", + ), + id="stderr", + ), + ], +) +def test_docker_check_test_runs_the_leak_check_before_the_outcome_assertions( + monkeypatch: pytest.MonkeyPatch, + check_output: EntrypointOutput, +) -> None: + """The outcome assertions print raw messages, so the leak check must fail first.""" + _fake_docker(monkeypatch, {"spec": _output(_SECRET_SPEC_MESSAGE), "check": check_output}) + scenario = ConnectorTestScenario(config_dict={"api_key": SECRET}, status="succeed") + + with pytest.raises(AssertionError, match="printed secret values") as error: + _run_docker_check(scenario) + assert SECRET not in str(error.value) + + +def test_docker_check_test_still_asserts_the_outcome_without_a_leak( + monkeypatch: pytest.MonkeyPatch, +) -> None: + check_output = _output(_connection_status("FAILED", "401 Unauthorized")) + _fake_docker(monkeypatch, {"spec": _output(_SECRET_SPEC_MESSAGE), "check": check_output}) + scenario = ConnectorTestScenario(config_dict={"api_key": SECRET}, status="succeed") + + with pytest.raises(AssertionError, match="did not succeed"): + _run_docker_check(scenario) + + +class _LeakySource(Source): + """A source whose `check` echoes its API key in the connection status message.""" + + succeed = False + + def spec(self, logger: logging.Logger) -> ConnectorSpecification: + return ConnectorSpecification( + connectionSpecification=_spec({"api_key": {"type": "string", "airbyte_secret": True}}) + ) + + def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteConnectionStatus: + status = Status.SUCCEEDED if self.succeed else Status.FAILED + return AirbyteConnectionStatus(status=status, message=f"Used key {config['api_key']}") + + def discover(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCatalog: + return AirbyteCatalog(streams=[]) + + def read(self, *args: Any, **kwargs: Any) -> Iterator[AirbyteMessage]: + yield from () + + +class _SucceedingLeakySource(_LeakySource): + succeed = True + + +class _InProcessSuite(SourceTestSuiteBase): + connector = _LeakySource + + @classmethod + def get_connector_root_dir(cls) -> Path: + return POKEAPI_CONNECTOR_ROOT + + +@pytest.mark.parametrize( + "connector, status", + [ + pytest.param(_LeakySource, "succeed", id="failed_check_on_a_success_scenario"), + pytest.param(_SucceedingLeakySource, "failed", id="succeeded_check_on_a_failure_scenario"), + ], +) +@pytest.mark.parametrize( + "suite_test", + [ + pytest.param(SourceTestSuiteBase.test_check, id="source_suite"), + pytest.param(ConnectorTestSuiteBase.test_check, id="connector_suite"), + ], +) +def test_in_process_check_test_fails_when_check_prints_a_secret( + monkeypatch: pytest.MonkeyPatch, + connector: type[Source], + status: str, + suite_test: Any, +) -> None: + """The leak check fails first, whatever the outcome, and prints no part of the secret.""" + monkeypatch.setattr(_InProcessSuite, "connector", connector) + scenario = ConnectorTestScenario(config_dict={"api_key": SECRET}, status=status) + + with pytest.raises(AssertionError, match="printed secret values") as error: + suite_test(_InProcessSuite(), scenario) + assert "`/api_key` in message #" in str(error.value) + assert SECRET not in str(error.value) From 8b7e8617a07a8db2f54e90d8caa47be407bb03f9 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Fri, 9 Oct 2026 15:00:41 +0300 Subject: [PATCH 4/8] fix(standard-tests): match camelCase secret names, exempt key-pair access keys Property names were only lowercased, so `clientSecret`, `accessToken` and `privateKey` never matched. They are now split into snake case first. No connector in the fleet has an unmarked camelCase secret today, so this adds no failures. The `access_key` suffix also matched the public half of a key pair, such as source-100ms `app_access_key`, which sits next to the marked `app_secret` and plays the role of an AWS access key ID. An `*access_key` with a sibling named `secret`, `*_secret` or `*secret_key` is now exempt. A lone `access_key` is still required to be secret, since many APIs take a single `access_key` parameter as their only credential. Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/test/standard_tests/_spec_lint.py | 75 +++++++++++++++---- unit_tests/test/test_spec_lint.py | 66 ++++++++++++++++ 2 files changed, 125 insertions(+), 16 deletions(-) diff --git a/airbyte_cdk/test/standard_tests/_spec_lint.py b/airbyte_cdk/test/standard_tests/_spec_lint.py index 5181224060..a32523e1f4 100644 --- a/airbyte_cdk/test/standard_tests/_spec_lint.py +++ b/airbyte_cdk/test/standard_tests/_spec_lint.py @@ -55,8 +55,14 @@ CAT matched exact names only, so it never flagged an unmarked `api_key`, the most common secret property name in the connector fleet. + +An `access_key` that sits next to a `secret` or `secret_key` property is exempt: it is the +public half of a key pair, like an AWS access key ID (see `is_secret_property_name`). """ +_ACCESS_KEY_SUFFIX = "access_key" +_KEY_PAIR_SECRET_SUFFIXES = ("_secret", "secret_key") + _SECRET_CAPABLE_TYPES = {"string", "integer", "number"} _COMBINATORS = ("oneOf", "anyOf", "allOf") @@ -68,12 +74,41 @@ """ -def is_secret_property_name(name: str) -> bool: - """Return whether a spec property with this name is expected to hold a secret.""" - normalized_name = name.lower().replace("-", "_") - return normalized_name in SECRET_PROPERTY_NAMES or normalized_name.endswith( - SECRET_PROPERTY_NAME_SUFFIXES - ) +def _normalize_property_name(name: str) -> str: + """Return `name` in snake case: `clientSecret` and `client-secret` become `client_secret`.""" + return re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", name).lower().replace("-", "_") + + +def _is_key_pair_secret_name(normalized_name: str) -> bool: + return normalized_name == "secret" or normalized_name.endswith(_KEY_PAIR_SECRET_SUFFIXES) + + +def is_secret_property_name(name: str, sibling_names: Iterable[str] = ()) -> bool: + """Return whether a spec property with this name is expected to hold a secret. + + Names are compared in snake case, so `clientSecret`, `client-secret` and `Client_Secret` + all match `client_secret`. + + `sibling_names` are the names of the other properties of the same object. An + `*access_key` with a sibling such as `app_secret` or `secret_key` is the public half of + a key pair: it identifies the key rather than authenticates with it, so it is not + expected to be secret. An `*access_key` on its own is the credential, as with the many + APIs that take a single `access_key` parameter. + """ + normalized_name = _normalize_property_name(name) + if normalized_name in SECRET_PROPERTY_NAMES: + return True + if not normalized_name.endswith(SECRET_PROPERTY_NAME_SUFFIXES): + return False + if normalized_name.endswith(_ACCESS_KEY_SUFFIX) and not normalized_name.endswith( + f"secret_{_ACCESS_KEY_SUFFIX}" + ): + return not any( + _is_key_pair_secret_name(_normalize_property_name(sibling)) + for sibling in sibling_names + if sibling != name + ) + return True def _can_hold_secret(property_schema: Mapping[str, Any]) -> bool: @@ -96,35 +131,41 @@ def _iter_named_properties( schema: Mapping[str, Any], pointer: str = "", name: str | None = None, -) -> Iterator[tuple[str, str, Mapping[str, Any]]]: - """Yield `(json_pointer, property_name, property_schema)` for each property in a schema. + sibling_names: frozenset[str] = frozenset(), +) -> Iterator[tuple[str, str, frozenset[str], Mapping[str, Any]]]: + """Yield `(json_pointer, name, sibling_names, property_schema)` for each property. - Variants of a `oneOf`/`anyOf`/`allOf` and the `items` of an array are yielded under the - name of the property that declares them. + `sibling_names` holds the names of every property of the object that declares the + property, itself included. Variants of a `oneOf`/`anyOf`/`allOf` and the `items` of an + array are yielded under the name and siblings of the property that declares them. """ if name is not None: - yield pointer, name, schema + yield pointer, name, sibling_names, schema properties = schema.get("properties") if isinstance(properties, Mapping): + property_names = frozenset(properties) for property_name, property_schema in properties.items(): if isinstance(property_schema, Mapping): yield from _iter_named_properties( property_schema, f"{pointer}/properties/{property_name}", property_name, + property_names, ) - for keyword in ("oneOf", "anyOf", "allOf"): + for keyword in _COMBINATORS: variants = schema.get(keyword) if isinstance(variants, list): for index, variant in enumerate(variants): if isinstance(variant, Mapping): - yield from _iter_named_properties(variant, f"{pointer}/{keyword}/{index}", name) + yield from _iter_named_properties( + variant, f"{pointer}/{keyword}/{index}", name, sibling_names + ) items = schema.get("items") if isinstance(items, Mapping): - yield from _iter_named_properties(items, f"{pointer}/items", name) + yield from _iter_named_properties(items, f"{pointer}/items", name, sibling_names) def find_secret_marking_errors(connection_specification: Mapping[str, Any]) -> list[str]: @@ -137,7 +178,9 @@ def find_secret_marking_errors(connection_specification: Mapping[str, Any]) -> l boolean or an object, must not set it. """ errors: list[str] = [] - for pointer, name, property_schema in _iter_named_properties(connection_specification): + for pointer, name, sibling_names, property_schema in _iter_named_properties( + connection_specification + ): marking = property_schema.get("airbyte_secret") if marking is not None and not isinstance(marking, bool): errors.append( @@ -145,7 +188,7 @@ def find_secret_marking_errors(connection_specification: Mapping[str, Any]) -> l "boolean, so the CDK does not treat the value as a secret." ) continue - if "type" not in property_schema or not is_secret_property_name(name): + if "type" not in property_schema or not is_secret_property_name(name, sibling_names): continue marked_as_secret = marking is True diff --git a/unit_tests/test/test_spec_lint.py b/unit_tests/test/test_spec_lint.py index 81c7039f6c..b234a0616e 100644 --- a/unit_tests/test/test_spec_lint.py +++ b/unit_tests/test/test_spec_lint.py @@ -73,12 +73,42 @@ def _spec_message(properties: dict[str, Any]) -> dict[str, Any]: pytest.param("primary_key", False, id="primary_key"), pytest.param("access_key_id", False, id="access_key_id"), pytest.param("start_date", False, id="unrelated"), + pytest.param("clientSecret", True, id="camel_case_client_secret"), + pytest.param("accessToken", True, id="camel_case_access_token"), + pytest.param("privateKey", True, id="camel_case_private_key"), + pytest.param("apiKey", True, id="camel_case_api_key"), + pytest.param("APIKey", True, id="upper_case_acronym"), + pytest.param("OAuthRefreshToken", True, id="pascal_case"), + pytest.param("tokenExpiryDate", False, id="camel_case_token_prefix_only"), + pytest.param("accessKeyId", False, id="camel_case_access_key_id"), + pytest.param("access_key", True, id="lone_access_key"), ], ) def test_is_secret_property_name(name: str, expected: bool) -> None: assert is_secret_property_name(name) == expected +@pytest.mark.parametrize( + "name, sibling_names, expected", + [ + pytest.param("app_access_key", ["app_secret"], False, id="100ms_key_pair"), + pytest.param("access_key", ["secret_key"], False, id="access_key_with_secret_key"), + pytest.param("accessKey", ["secretKey"], False, id="camel_case_key_pair"), + pytest.param("access_key", ["secret"], False, id="access_key_with_secret"), + pytest.param("access_key", ["bucket", "region"], True, id="no_secret_sibling"), + pytest.param("access_key", ["access_key"], True, id="itself_is_not_a_sibling"), + pytest.param( + "aws_secret_access_key", ["aws_access_key_id", "client_secret"], True, id="secret_half" + ), + pytest.param("api_key", ["api_secret"], True, id="only_access_keys_are_exempt"), + ], +) +def test_is_secret_property_name_exempts_the_public_half_of_a_key_pair( + name: str, sibling_names: list[str], expected: bool +) -> None: + assert is_secret_property_name(name, sibling_names) == expected + + @pytest.mark.parametrize( "properties, expected_errors", [ @@ -174,6 +204,42 @@ def test_is_secret_property_name(name: str, expected: bool) -> None: [], id="false_marking_on_non_secret_name_passes", ), + pytest.param( + { + "app_access_key": {"type": "string"}, + "app_secret": {"type": "string", "airbyte_secret": True}, + }, + [], + id="unmarked_public_half_of_a_key_pair_passes", + ), + pytest.param( + {"access_key": {"type": "string"}}, + ["`/properties/access_key` looks like a secret"], + id="unmarked_lone_access_key_fails", + ), + pytest.param( + { + "keys": { + "type": "object", + "oneOf": [ + { + "type": "object", + "properties": { + "access_key": {"type": "string"}, + "secret_key": {"type": "string", "airbyte_secret": True}, + }, + } + ], + } + }, + [], + id="key_pair_inside_one_of_passes", + ), + pytest.param( + {"clientSecret": {"type": "string"}}, + ["`/properties/clientSecret` looks like a secret"], + id="unmarked_camel_case_secret_fails", + ), ], ) def test_find_secret_marking_errors( From c31a023dfb565c3b1ca38b9937c4fa33079f48a1 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Fri, 9 Oct 2026 15:00:55 +0300 Subject: [PATCH 5/8] fix(standard-tests): run `spec` once per image per test session test_docker_image_build_and_check ran an extra `docker run spec` for every check scenario, only to read the secret paths for the leak check. _run_docker_spec is now cached per image and connector name, so the spec test and every check scenario share one run. The image under a tag does not change within a session, since each test builds it before it runs. Co-Authored-By: Claude Opus 5.5 --- .../test/standard_tests/docker_base.py | 8 ++++++- unit_tests/test/test_spec_lint.py | 22 +++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index cfd74f0c2c..e6b664ad53 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -3,6 +3,7 @@ from __future__ import annotations +import functools import inspect import shutil import sys @@ -82,8 +83,13 @@ def _assert_check_outcome( ) +@functools.cache def _run_docker_spec(connector_image: str, *, connector_name: str) -> ConnectorSpecification: - """Run `spec` in the connector image and return the single spec it emits.""" + """Run `spec` in the connector image and return the single spec it emits. + + The result is cached per image for the test session, so the spec test and every check + scenario share one `docker run`. The image under a tag does not change within a session. + """ spec_result = run_docker_airbyte_command( [ "docker", diff --git a/unit_tests/test/test_spec_lint.py b/unit_tests/test/test_spec_lint.py index b234a0616e..5d0111cc3e 100644 --- a/unit_tests/test/test_spec_lint.py +++ b/unit_tests/test/test_spec_lint.py @@ -657,6 +657,13 @@ def get_connector_root_dir(cls) -> Path: return POKEAPI_CONNECTOR_ROOT +@pytest.fixture(autouse=True) +def _clear_docker_spec_cache() -> Iterator[None]: + docker_base._run_docker_spec.cache_clear() + yield + docker_base._run_docker_spec.cache_clear() + + def _fake_docker( monkeypatch: pytest.MonkeyPatch, outputs: dict[str, EntrypointOutput], @@ -786,6 +793,21 @@ def test_docker_check_test_still_asserts_the_outcome_without_a_leak( _run_docker_check(scenario) +def test_docker_check_tests_share_one_spec_run(monkeypatch: pytest.MonkeyPatch) -> None: + check_output = _output(_connection_status("FAILED", "Invalid API key")) + commands = _fake_docker( + monkeypatch, {"spec": _output(_SECRET_SPEC_MESSAGE), "check": check_output} + ) + _DockerSuite().test_docker_image_build_and_spec( + connector_image_override="source-test:dev", + connector_base_image_override=None, + ) + for _ in range(2): + _run_docker_check(ConnectorTestScenario(config_dict={"api_key": SECRET}, status="failed")) + + assert [command[-1] for command in commands if "spec" in command] == ["spec"] + + class _LeakySource(Source): """A source whose `check` echoes its API key in the connection status message.""" From 5a6ede5283ed1d2e9041c6639f68f4eef90822f8 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Fri, 9 Oct 2026 19:31:15 +0300 Subject: [PATCH 6/8] fix: mask secrets in config-error CONNECTION_STATUS messages `check_config_against_spec_or_exit` builds "Config validation error: ..." from jsonschema's message, which quotes the rejected value, and `AirbyteTracedException.as_connection_status_message` put it into the CONNECTION_STATUS without the secret filter. A secret that failed a `pattern`, `enum` or length check was printed in clear. - `as_connection_status_message` now passes the message through `filter_secrets`. - `AirbyteEntrypoint.check` emits a connector-raised config error as a sanitized TRACE and builds its CONNECTION_STATUS through the same method. - `Destination.run_cmd` registers the config's secrets with the filter before validating the config, as the source entrypoint already does. Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/destinations/destination.py | 4 ++ airbyte_cdk/entrypoint.py | 11 ++-- airbyte_cdk/utils/traced_exception.py | 8 ++- unit_tests/destinations/test_destination.py | 32 +++++++++++ unit_tests/test_entrypoint.py | 59 +++++++++++++++++++++ unit_tests/utils/test_traced_exception.py | 17 ++++++ 6 files changed, 123 insertions(+), 8 deletions(-) diff --git a/airbyte_cdk/destinations/destination.py b/airbyte_cdk/destinations/destination.py index 547f966849..956fad3bb7 100644 --- a/airbyte_cdk/destinations/destination.py +++ b/airbyte_cdk/destinations/destination.py @@ -21,6 +21,7 @@ Type, ) from airbyte_cdk.sources.utils.schema_helpers import check_config_against_spec_or_exit +from airbyte_cdk.utils.airbyte_secrets_utils import get_secrets, update_secrets from airbyte_cdk.utils.traced_exception import AirbyteTracedException logger = logging.getLogger("airbyte") @@ -125,6 +126,9 @@ def run_cmd(self, parsed_args: argparse.Namespace) -> Iterable[AirbyteMessage]: yield AirbyteMessage(type=Type.SPEC, spec=spec) return config = self.read_config(config_path=parsed_args.config) + # Register the config's secrets with the log filter before anything can print them, + # as the source entrypoint does. A config validation error quotes config values. + update_secrets(get_secrets(spec.connectionSpecification, config)) if self.check_config_against_spec or cmd == "check": try: check_config_against_spec_or_exit(config, spec) diff --git a/airbyte_cdk/entrypoint.py b/airbyte_cdk/entrypoint.py index 57820f0053..c2432237bc 100644 --- a/airbyte_cdk/entrypoint.py +++ b/airbyte_cdk/entrypoint.py @@ -238,18 +238,15 @@ def check( try: check_result = self.source.check(self.logger, config) except AirbyteTracedException as traced_exc: - yield traced_exc.as_airbyte_message() + yield traced_exc.as_sanitized_airbyte_message() # The platform uses the exit code to surface unexpected failures so we raise the exception if the failure type not a config error # If the failure is not exceptional, we'll emit a failed connection status message and return if traced_exc.failure_type != FailureType.config_error: raise traced_exc else: - yield AirbyteMessage( - type=Type.CONNECTION_STATUS, - connectionStatus=AirbyteConnectionStatus( - status=Status.FAILED, message=traced_exc.message - ), - ) + connection_status = traced_exc.as_connection_status_message() + if connection_status: + yield connection_status return if check_result.status == Status.SUCCEEDED: self.logger.info("Check succeeded") diff --git a/airbyte_cdk/utils/traced_exception.py b/airbyte_cdk/utils/traced_exception.py index cb4072b534..7a9acb0f04 100644 --- a/airbyte_cdk/utils/traced_exception.py +++ b/airbyte_cdk/utils/traced_exception.py @@ -89,11 +89,17 @@ def as_airbyte_message( return AirbyteMessage(type=MessageType.TRACE, trace=trace_message) def as_connection_status_message(self) -> Optional[AirbyteMessage]: + """Build a failed CONNECTION_STATUS from a config error, with secrets masked. + + The message can quote config values, as a config validation error does, so it goes + through the same secret filter as logs and traces. + """ if self.failure_type == FailureType.config_error: return AirbyteMessage( type=MessageType.CONNECTION_STATUS, connectionStatus=AirbyteConnectionStatus( - status=Status.FAILED, message=self.message + status=Status.FAILED, + message=filter_secrets(self.message) if self.message else self.message, ), ) return None diff --git a/unit_tests/destinations/test_destination.py b/unit_tests/destinations/test_destination.py index 1f8f6573fb..826de76763 100644 --- a/unit_tests/destinations/test_destination.py +++ b/unit_tests/destinations/test_destination.py @@ -32,6 +32,7 @@ SyncMode, Type, ) +from airbyte_cdk.utils.airbyte_secrets_utils import update_secrets @pytest.fixture(name="destination") @@ -248,6 +249,37 @@ def test_run_check_with_invalid_config( # the specific phrasing is not relevant, so only check for the keywords assert "validation error" in returned_check_result.connectionStatus.message + def test_run_check_masks_secrets_in_the_config_validation_error( + self, + mocker, + destination: Destination, + tmp_path: Path, + ) -> None: + file_path = tmp_path / "config.json" + write_file(file_path, {"password": "abcd1234"}) + spec = { + "type": "object", + "properties": { + "password": {"type": "string", "pattern": "^[a-z]+$", "airbyte_secret": True} + }, + } + mocker.patch.object( + destination, "spec", return_value=ConnectorSpecification(connectionSpecification=spec) + ) + mocker.patch.object(destination, "check") + + try: + parsed_args = argparse.Namespace(command="check", config=file_path) + returned_check_result = next(iter(destination.run_cmd(parsed_args))) + finally: + update_secrets([]) + + destination.check.assert_not_called() # type: ignore + message = returned_check_result.connectionStatus.message + assert message.startswith("Config validation error:") + assert "abcd1234" not in message + assert "****" in message + def test_run_write(self, mocker, destination: Destination, tmp_path, monkeypatch): config_path, dummy_config = tmp_path / "config.json", {"user": "sherif"} write_file(config_path, dummy_config) diff --git a/unit_tests/test_entrypoint.py b/unit_tests/test_entrypoint.py index fcfb449151..58b94a85a6 100644 --- a/unit_tests/test_entrypoint.py +++ b/unit_tests/test_entrypoint.py @@ -46,6 +46,7 @@ from airbyte_cdk.sources import Source from airbyte_cdk.sources.connector_state_manager import HashableStreamDescriptor from airbyte_cdk.utils import AirbyteTracedException +from airbyte_cdk.utils.airbyte_secrets_utils import update_secrets class MockSource(Source): @@ -944,3 +945,61 @@ def test_memory_failfast_flushes_queued_state_before_raising(mocker): with pytest.raises(AirbyteTracedException) as exc_info: next(gen) assert exc_info.value is fail_fast_exc + + +@pytest.fixture +def reset_secret_filter(): + yield + update_secrets([]) + + +_PASSWORD_SPEC = ConnectorSpecification( + connectionSpecification={ + "type": "object", + "properties": { + "password": {"type": "string", "pattern": "^[a-z]+$", "airbyte_secret": True} + }, + } +) + + +@pytest.mark.parametrize("config_mock", [{"password": "abcd1234"}], indirect=True) +def test_check_masks_secrets_in_the_config_validation_error( + entrypoint: AirbyteEntrypoint, mocker, config_mock, reset_secret_filter +): + """jsonschema quotes the rejected value, so the CONNECTION_STATUS must mask it.""" + mocker.patch.object(MockSource, "spec", return_value=_PASSWORD_SPEC) + check_mock = mocker.patch.object(MockSource, "check") + + messages = list(entrypoint.run(Namespace(command="check", config="config_path"))) + + check_mock.assert_not_called() + status_message = orjson.loads(messages[-1])["connectionStatus"]["message"] + assert status_message.startswith("Config validation error:") + assert "abcd1234" not in status_message + assert "****" in status_message + + +@freezegun.freeze_time("1970-01-01T00:00:00.001Z") +@pytest.mark.parametrize("config_mock", [{"password": "secretpw"}], indirect=True) +def test_check_masks_secrets_in_a_config_error_raised_by_the_connector( + entrypoint: AirbyteEntrypoint, mocker, config_mock, reset_secret_filter +): + spec = ConnectorSpecification( + connectionSpecification={ + "type": "object", + "properties": {"password": {"type": "string", "airbyte_secret": True}}, + } + ) + mocker.patch.object(MockSource, "spec", return_value=spec) + exception = AirbyteTracedException( + message="Login failed for secretpw", + internal_message="401 for secretpw", + failure_type=FailureType.config_error, + ) + mocker.patch.object(MockSource, "check", side_effect=exception) + + messages = list(entrypoint.run(Namespace(command="check", config="config_path"))) + + assert "secretpw" not in "".join(messages) + assert orjson.loads(messages[-1])["connectionStatus"]["message"] == "Login failed for ****" diff --git a/unit_tests/utils/test_traced_exception.py b/unit_tests/utils/test_traced_exception.py index 670d109534..ac29fb6deb 100644 --- a/unit_tests/utils/test_traced_exception.py +++ b/unit_tests/utils/test_traced_exception.py @@ -17,6 +17,7 @@ TraceType, ) from airbyte_cdk.models import Type as MessageType +from airbyte_cdk.utils.airbyte_secrets_utils import update_secrets from airbyte_cdk.utils.traced_exception import AirbyteTracedException _AN_EXCEPTION = ValueError("An exception") @@ -336,3 +337,19 @@ def test_from_exception_wrapping_regular_exception_with_message(self) -> None: ) assert wrapped.internal_message == "runtime failure" assert wrapped.message == "A runtime error occurred." + + +def test_config_error_as_connection_status_message_masks_secrets(): + update_secrets(["s3cr3t-value"]) + try: + traced_exc = AirbyteTracedException( + message="Config validation error: 's3cr3t-value' does not match '^a$'", + failure_type=FailureType.config_error, + ) + airbyte_message = traced_exc.as_connection_status_message() + finally: + update_secrets([]) + + assert airbyte_message.connectionStatus.message == ( + "Config validation error: '****' does not match '^a$'" + ) From 6225ea920b1a00d920e7ce0c326a0c8e617b70cc Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Fri, 9 Oct 2026 19:31:15 +0300 Subject: [PATCH 7/8] fix(standard-tests): align the spec lint with the runtime secret filter Marking lint: - Accept the pydantic-v2 optional secret (`airbyte_secret` next to `anyOf`): variants are no longer linted as the property, and a property typed only by its variants is checked against their types. - Report `airbyte_secret: true` under `anyOf`/`allOf` or a top-level combinator, where `get_secret_paths` cannot resolve it, and count a flag on a `oneOf` variant, which the filter does resolve. - Treat an explicit `airbyte_secret: false` as a reviewed non-secret, and say so in the failure text. Leak check: - Walk only the `oneOf` variant the config selects by its discriminators. - Stop yielding the field names of a secret object. - Search numbers emitted as JSON numbers. - Hide a stream name only when it overlaps a searched secret form. - Label a hit in the CDK's own "Config validation error" CONNECTION_STATUS. Harness: - Cache a failing Docker `spec` per image, and skip the leak check (Docker and in-process) when `spec` fails, since the spec tests report it. Tests cover each harness change the review found untested. Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/test/standard_tests/_spec_lint.py | 332 ++++++++++--- .../test/standard_tests/connector_base.py | 26 +- .../test/standard_tests/docker_base.py | 55 ++- unit_tests/test/test_spec_lint.py | 462 +++++++++++++++++- 4 files changed, 782 insertions(+), 93 deletions(-) diff --git a/airbyte_cdk/test/standard_tests/_spec_lint.py b/airbyte_cdk/test/standard_tests/_spec_lint.py index a32523e1f4..b2a0ce6f7a 100644 --- a/airbyte_cdk/test/standard_tests/_spec_lint.py +++ b/airbyte_cdk/test/standard_tests/_spec_lint.py @@ -11,10 +11,16 @@ import json import re from collections.abc import Iterable, Iterator, Mapping +from dataclasses import dataclass from typing import Any from urllib.parse import quote -from airbyte_cdk.models import AirbyteMessageSerializer, ConnectorSpecification, Type +from airbyte_cdk.models import ( + AirbyteMessage, + AirbyteMessageSerializer, + ConnectorSpecification, + Type, +) from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput SECRET_PROPERTY_NAMES = frozenset( @@ -67,6 +73,16 @@ _COMBINATORS = ("oneOf", "anyOf", "allOf") +_UNMASKED_COMBINATORS = ("anyOf", "allOf") +"""Combinators that the CDK's runtime secret filter does not look through. + +`airbyte_cdk.utils.airbyte_secrets_utils.get_secret_paths` drops only `properties` and `oneOf` +from the schema path of an `airbyte_secret` flag and looks the rest up in the config. A flag +under `anyOf` or `allOf` therefore resolves to no config value, and the secret is never masked +in logs. `get_secrets` also starts at the top-level `properties`, so no combinator at the root +of the spec is looked through either. +""" + _MIN_LEAK_CHECK_LENGTH = 8 """Secret values shorter than this are not searched for in the output. @@ -111,76 +127,171 @@ def is_secret_property_name(name: str, sibling_names: Iterable[str] = ()) -> boo return True -def _can_hold_secret(property_schema: Mapping[str, Any]) -> bool: +def _variants(schema: Mapping[str, Any], keywords: Iterable[str] = _COMBINATORS) -> Iterator[Any]: + for keyword in keywords: + variants = schema.get(keyword) + if isinstance(variants, list): + yield from (variant for variant in variants if isinstance(variant, Mapping)) + + +def _declared_types(schema: Mapping[str, Any]) -> set[str]: + """Return the types `schema` declares, itself or through its combinator variants. + + A pydantic-v2 `Optional[str]` field declares its types only under `anyOf`. + """ + property_type = schema.get("type") + if isinstance(property_type, str): + return {property_type} + if isinstance(property_type, list): + return {str(item) for item in property_type} + return set().union(*(_declared_types(variant) for variant in _variants(schema))) + + +def _can_hold_secret(schema: Mapping[str, Any]) -> bool: """Return whether a property can hold a secret value. Booleans and nulls cannot, and the UI cannot render secret objects or arrays. A property - with a `const` value cannot either, since every config holds the same value. + with a `const` value cannot either, since every config holds the same value. A property + that declares its types only under a combinator can hold a secret if any variant can. """ - property_type = property_schema.get("type") - if isinstance(property_type, str): - types = {property_type} - elif isinstance(property_type, list): - types = set(property_type) - else: + if "const" in schema: return False - return bool(types & _SECRET_CAPABLE_TYPES) and "const" not in property_schema + if "type" in schema: + return bool(_declared_types(schema) & _SECRET_CAPABLE_TYPES) + return any(_can_hold_secret(variant) for variant in _variants(schema)) + +def _is_marked_secret(schema: Mapping[str, Any]) -> bool: + """Return whether the runtime secret filter masks the value of this property. -def _iter_named_properties( + The flag counts on the property itself, or on a `oneOf` variant of it, because the + filter drops `oneOf` from the schema path. A flag under `anyOf` or `allOf` does not count + (see `_UNMASKED_COMBINATORS`). + """ + if schema.get("airbyte_secret") is True: + return True + return any(_is_marked_secret(variant) for variant in _variants(schema, ("oneOf",))) + + +@dataclass(frozen=True) +class _SpecProperty: + pointer: str + """JSON pointer of this schema in the connection specification.""" + name: str + """Name of the property this schema belongs to.""" + sibling_names: frozenset[str] + """Names of every property of the object that declares the property, itself included.""" + schema: Mapping[str, Any] + property_pointer: str + """JSON pointer of the property itself, which differs from `pointer` for a variant.""" + is_variant: bool + """Whether this schema is a `oneOf`/`anyOf`/`allOf` variant of the property.""" + unmasked_under: str | None + """The first combinator on the path that the runtime secret filter does not look through.""" + unmasked_at_root: bool + """Whether `unmasked_under` sits at the root of the spec, which the filter never reads.""" + + +def _iter_spec_properties( schema: Mapping[str, Any], + *, pointer: str = "", name: str | None = None, sibling_names: frozenset[str] = frozenset(), -) -> Iterator[tuple[str, str, frozenset[str], Mapping[str, Any]]]: - """Yield `(json_pointer, name, sibling_names, property_schema)` for each property. - - `sibling_names` holds the names of every property of the object that declares the - property, itself included. Variants of a `oneOf`/`anyOf`/`allOf` and the `items` of an - array are yielded under the name and siblings of the property that declares them. + property_pointer: str = "", + is_variant: bool = False, + unmasked_under: str | None = None, + unmasked_at_root: bool = False, +) -> Iterator[_SpecProperty]: + """Yield every property of `schema`, and every combinator variant of each property. + + The `items` of an array are yielded as the property itself, under its name and siblings, + since an array of secret strings marks its items. """ if name is not None: - yield pointer, name, sibling_names, schema + yield _SpecProperty( + pointer, + name, + sibling_names, + schema, + property_pointer, + is_variant, + unmasked_under, + unmasked_at_root, + ) properties = schema.get("properties") if isinstance(properties, Mapping): property_names = frozenset(properties) for property_name, property_schema in properties.items(): if isinstance(property_schema, Mapping): - yield from _iter_named_properties( + child_pointer = f"{pointer}/properties/{property_name}" + yield from _iter_spec_properties( property_schema, - f"{pointer}/properties/{property_name}", - property_name, - property_names, + pointer=child_pointer, + name=property_name, + sibling_names=property_names, + property_pointer=child_pointer, + unmasked_under=unmasked_under, + unmasked_at_root=unmasked_at_root, ) for keyword in _COMBINATORS: variants = schema.get(keyword) - if isinstance(variants, list): - for index, variant in enumerate(variants): - if isinstance(variant, Mapping): - yield from _iter_named_properties( - variant, f"{pointer}/{keyword}/{index}", name, sibling_names - ) + if not isinstance(variants, list): + continue + at_root = name is None + hides_secrets = unmasked_under is None and (keyword in _UNMASKED_COMBINATORS or at_root) + for index, variant in enumerate(variants): + if isinstance(variant, Mapping): + yield from _iter_spec_properties( + variant, + pointer=f"{pointer}/{keyword}/{index}", + name=name, + sibling_names=sibling_names, + property_pointer=property_pointer, + is_variant=not at_root, + unmasked_under=keyword if hides_secrets else unmasked_under, + unmasked_at_root=at_root if hides_secrets else unmasked_at_root, + ) items = schema.get("items") if isinstance(items, Mapping): - yield from _iter_named_properties(items, f"{pointer}/items", name, sibling_names) + yield from _iter_spec_properties( + items, + pointer=f"{pointer}/items", + name=name, + sibling_names=sibling_names, + property_pointer=f"{pointer}/items", + unmasked_under=unmasked_under, + unmasked_at_root=unmasked_at_root, + ) + + +def _describe_types(schema: Mapping[str, Any]) -> str: + return ", ".join(sorted(_declared_types(schema))) or "none" def find_secret_marking_errors(connection_specification: Mapping[str, Any]) -> list[str]: """Return one error per property whose `airbyte_secret` flag is wrong. Any `airbyte_secret` value must be a boolean: the CDK's secret filter only masks values - whose flag is `True`, so a string such as `"true"` leaves the value unmasked. A property - whose name denotes a secret (see `is_secret_property_name`) and that can hold a secret - value must set `airbyte_secret: true`. One that cannot hold a secret value, such as a - boolean or an object, must not set it. + whose flag is `True`, so a string such as `"true"` leaves the value unmasked. The flag + must also sit where the filter finds it, so not under `anyOf` or `allOf` (see + `_UNMASKED_COMBINATORS`). + + A property whose name denotes a secret (see `is_secret_property_name`) and that can hold + a secret value must set `airbyte_secret: true`, on itself or on a `oneOf` variant. One + that cannot hold a secret value, such as a boolean or an object, must not set it. + + A property that sets `airbyte_secret: false` explicitly is not reported as an unmarked + secret: that is how a spec records that a secret-looking name, such as a sort `key`, + holds no secret. """ errors: list[str] = [] - for pointer, name, sibling_names, property_schema in _iter_named_properties( - connection_specification - ): + for spec_property in _iter_spec_properties(connection_specification): + pointer = spec_property.pointer + property_schema = spec_property.schema marking = property_schema.get("airbyte_secret") if marking is not None and not isinstance(marking, bool): errors.append( @@ -188,23 +299,59 @@ def find_secret_marking_errors(connection_specification: Mapping[str, Any]) -> l "boolean, so the CDK does not treat the value as a secret." ) continue - if "type" not in property_schema or not is_secret_property_name(name, sibling_names): + if marking is True and spec_property.unmasked_under: + if spec_property.is_variant: + fix = f"Set it on `{spec_property.property_pointer}` instead." + elif spec_property.unmasked_at_root: + fix = "The filter reads only the top-level `properties` of the spec." + else: + combinator = spec_property.unmasked_under + fix = f"Declare the variants with `oneOf` instead of `{combinator}`." + errors.append( + f"`{pointer}` sets `airbyte_secret: true` under " + f"`{spec_property.unmasked_under}`, where the CDK's secret filter does not look, " + f"so the value is not masked in logs. {fix}" + ) + continue + if spec_property.is_variant: + # The property that declares the variant is checked as a whole. + continue + if not _declared_types(property_schema) or not is_secret_property_name( + spec_property.name, spec_property.sibling_names + ): continue - marked_as_secret = marking is True + marked_as_secret = _is_marked_secret(property_schema) can_hold_secret = _can_hold_secret(property_schema) if can_hold_secret and not marked_as_secret: + if marking is False or _has_unmasked_variant_marking(property_schema): + # An explicit `false` is a reviewed decision. A flag under `anyOf` is reported + # on the variant above. + continue errors.append( - f"`{pointer}` looks like a secret but is not marked `airbyte_secret: true`." + f"`{pointer}` looks like a secret but is not marked `airbyte_secret: true`. " + "If it holds no secret, set `airbyte_secret: false` on it explicitly." ) elif marked_as_secret and not can_hold_secret: errors.append( f"`{pointer}` is marked `airbyte_secret: true` but its type " - f"`{property_schema.get('type')}` cannot hold a secret value." + f"`{_describe_types(property_schema)}` cannot hold a secret value." ) return errors +def _has_unmasked_variant_marking(schema: Mapping[str, Any], under_unmasked: bool = False) -> bool: + """Return whether a variant of `schema` sets the flag under `anyOf` or `allOf`.""" + for keyword in _COMBINATORS: + unmasked = under_unmasked or keyword in _UNMASKED_COMBINATORS + for variant in _variants(schema, (keyword,)): + if unmasked and variant.get("airbyte_secret") is True: + return True + if _has_unmasked_variant_marking(variant, unmasked): + return True + return False + + def get_single_spec(result: EntrypointOutput, *, connector_name: str) -> ConnectorSpecification: """Assert that `spec` emitted exactly one SPEC message, and return its specification.""" spec_messages = result.spec_messages @@ -234,10 +381,13 @@ def _pointer_token(name: str) -> str: def _scalars(value: Any) -> Iterator[Any]: - """Yield every scalar inside a JSON-like value, mapping keys included.""" + """Yield every scalar inside a JSON-like value. + + Mapping keys are not yielded: they name the fields of a secret object, and the runtime + secret filter does not mask them either. + """ if isinstance(value, Mapping): - for key, item in value.items(): - yield key + for item in value.values(): yield from _scalars(item) elif isinstance(value, list): for item in value: @@ -246,6 +396,42 @@ def _scalars(value: Any) -> Iterator[Any]: yield value +def _discriminators(variant: Mapping[str, Any]) -> dict[str, Any]: + """Return the properties of a `oneOf` variant that have one fixed value, such as `auth_type`.""" + properties = variant.get("properties") + if not isinstance(properties, Mapping): + return {} + discriminators: dict[str, Any] = {} + for name, property_schema in properties.items(): + if not isinstance(property_schema, Mapping): + continue + enum = property_schema.get("enum") + if "const" in property_schema: + discriminators[name] = property_schema["const"] + elif isinstance(enum, list) and len(enum) == 1: + discriminators[name] = enum[0] + return discriminators + + +def _selected_variants(variants: list[Any], config: Any) -> list[Mapping[str, Any]]: + """Return the `oneOf` variants that `config` selects through their discriminators. + + A value that is secret only in a variant the user did not choose is not a secret of this + config. If no variant's discriminators match the config, for example because no variant + has any, every variant is returned. + """ + mappings = [variant for variant in variants if isinstance(variant, Mapping)] + if not isinstance(config, Mapping): + return mappings + selected = [ + variant + for variant in mappings + if (discriminators := _discriminators(variant)) + and all(name in config and config[name] == value for name, value in discriminators.items()) + ] + return selected or mappings + + def _iter_config_secrets( schema: Any, config: Any, @@ -274,6 +460,8 @@ def _iter_config_secrets( for keyword in _COMBINATORS: variants = schema.get(keyword) if isinstance(variants, list): + if keyword == "oneOf": + variants = _selected_variants(variants, config) for variant in variants: yield from _iter_config_secrets(variant, config, pointer) @@ -290,9 +478,10 @@ def find_config_secrets( """Return `(config_pointer, value)` for every `airbyte_secret` value in `config`. Unlike `airbyte_cdk.utils.airbyte_secrets_utils.get_secrets`, which the CDK uses to mask - logs at runtime, this also walks array `items` and every `anyOf`/`allOf`/`oneOf` variant, - so it finds a secret in an array of objects, such as one API key per account. It finds - every value `get_secrets` finds for the spec shapes connectors use. + logs at runtime, this also walks array `items` and `anyOf`/`allOf` variants, so it finds a + secret in an array of objects, such as one API key per account. Of a `oneOf`, it walks only + the variants the config selects (see `_selected_variants`). For the variant the config + selects, it finds every value `get_secrets` finds. """ return list(dict.fromkeys(_iter_config_secrets(connection_specification, config))) @@ -338,9 +527,14 @@ def _stream_name(message: Mapping[str, Any]) -> str | None: def _string_values(value: Any) -> Iterator[str]: - """Yield every string found in a JSON-like value, recursively.""" + """Yield every string found in a JSON-like value, recursively. + + Numbers are yielded as text too, so a numeric secret emitted as a JSON number is found. + """ if isinstance(value, str): yield value + elif isinstance(value, (int, float)) and not isinstance(value, bool): + yield str(value) elif isinstance(value, Mapping): for item in value.values(): yield from _string_values(item) @@ -349,16 +543,36 @@ def _string_values(value: Any) -> Iterator[str]: yield from _string_values(item) +_CONFIG_VALIDATION_ERROR_PREFIX = "Config validation error:" +"""How `check_config_against_spec_or_exit` starts the message of a failed config validation.""" + +_CONFIG_VALIDATION_LABEL = "built by the CDK's config validation" + + +def _is_config_validation_status(message: AirbyteMessage) -> bool: + status = message.connectionStatus + return ( + message.type == Type.CONNECTION_STATUS + and status is not None + and (status.message or "").startswith(_CONFIG_VALIDATION_ERROR_PREFIX) + ) + + def find_leaked_secrets( output: EntrypointOutput, secrets: Iterable[tuple[str, Any]], ) -> list[str]: """Return one line per place in `output` that contains a secret from `secrets`. - `secrets` holds `(config_pointer, value)` pairs, as `find_config_secrets` returns. A line names the pointers of the secrets found and where they were found: the - message number (on the Docker path, the stdout line number), its type and stream, or the - stderr line number. It never contains any text from the output, so it cannot print a - secret in any encoding. The stream name is left out if it overlaps any secret. + `secrets` holds `(config_pointer, value)` pairs, as `find_config_secrets` returns. A line + names the pointers of the secrets found and where they were found: the message number (on + the Docker path, the stdout line number), its type and stream, or the stderr line number. + It never contains any text from the output, so it cannot print a secret in any encoding. + The stream name is left out if it overlaps any searched form of a secret. + + A CONNECTION_STATUS that the CDK builds from a failed config validation is labeled as + such, since the connector's own code never ran: the fix is a newer CDK, which masks + secrets in that message, or a scenario config that passes the spec. CONTROL messages are skipped: a connector config update carries the full config, including its secrets, by design. @@ -370,9 +584,7 @@ def find_leaked_secrets( variants_by_pointer.setdefault(pointer, set()).update(variants) if not variants_by_pointer: return [] - every_secret_form = {str(secret) for _, secret in secrets if str(secret)}.union( - *variants_by_pointer.values() - ) + every_secret_form = set().union(*variants_by_pointer.values()) def leaked_pointers(text: str) -> list[str]: return [ @@ -400,6 +612,8 @@ def describe(pointers: Iterable[str]) -> str: stream = _stream_name(serialized_message) if stream and not any(form in stream or stream in form for form in every_secret_form): location += f", stream `{stream}`" + if _is_config_validation_status(message): + location += f", {_CONFIG_VALIDATION_LABEL}" leaks.append(f"{describe(pointers)} in message #{number} ({location})") stderr_lines = (output.stderr or "").splitlines() @@ -428,8 +642,16 @@ def assert_no_secrets_in_output( """ secrets = find_config_secrets(spec.connectionSpecification, config) leaks = find_leaked_secrets(output, secrets) + hint = "" + if any(_CONFIG_VALIDATION_LABEL in leak for leak in leaks): + hint = ( + "\nA message built by the CDK's config validation comes from the CDK, not from " + "connector code: the CDK in use predates masking secrets in it, or the secret sits " + "where the CDK's secret filter does not look (the spec lint reports that). Upgrade " + "`airbyte-cdk`, or make the scenario config pass the spec." + ) assert not leaks, ( f"`{verb}` for connector '{connector_name}' printed secret values from its config. " "Secrets must never appear in logs, traces or connection status messages. The values " - "at these config paths were found in the output:\n" + "\n".join(leaks) + "at these config paths were found in the output:\n" + "\n".join(leaks) + hint ) diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py index 829e793cb2..8c251c835c 100644 --- a/airbyte_cdk/test/standard_tests/connector_base.py +++ b/airbyte_cdk/test/standard_tests/connector_base.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import TYPE_CHECKING, cast +import pytest from boltons.typeutils import classproperty from airbyte_cdk.test import entrypoint_wrapper @@ -19,7 +20,10 @@ assert_no_secrets_in_output, get_single_spec, ) -from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite +from airbyte_cdk.test.standard_tests.docker_base import ( + _SPEC_FAILED_SKIP_REASON, + DockerConnectorTestSuite, +) if TYPE_CHECKING: from collections.abc import Callable @@ -114,16 +118,22 @@ def _assert_no_secrets_in_output( """Assert that no `airbyte_secret` value from the scenario's config is in `output`. Pass this to `run_test_job` as `inspect_output`, so it runs before the job's own - assertions print raw messages. + assertions print raw messages. It runs `spec` in-process once per call, to learn which + config paths are secret. If `spec` fails, the test is skipped: the spec tests report + that failure, and the outcome assertions must not print raw messages unchecked. """ - spec_result = run_test_job( - self.create_connector(scenario), - "spec", - connector_root=self.get_connector_root_dir(), - ) + try: + spec_result = run_test_job( + self.create_connector(scenario), + "spec", + connector_root=self.get_connector_root_dir(), + ) + spec = get_single_spec(spec_result, connector_name=self.connector_name) + except Exception: + pytest.skip(_SPEC_FAILED_SKIP_REASON) assert_no_secrets_in_output( output, - spec=get_single_spec(spec_result, connector_name=self.connector_name), + spec=spec, config=scenario.get_config_dict( connector_root=self.get_connector_root_dir(), empty_if_missing=True, diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index e6b664ad53..aab32e154e 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -3,7 +3,6 @@ from __future__ import annotations -import functools import inspect import shutil import sys @@ -83,24 +82,44 @@ def _assert_check_outcome( ) -@functools.cache +_docker_spec_cache: dict[str, ConnectorSpecification | Exception] = {} +"""The spec, or the error from running `spec`, of each connector image in this test session.""" + + def _run_docker_spec(connector_image: str, *, connector_name: str) -> ConnectorSpecification: """Run `spec` in the connector image and return the single spec it emits. - The result is cached per image for the test session, so the spec test and every check - scenario share one `docker run`. The image under a tag does not change within a session. + The outcome is cached per image for the test session, so the spec test and every check + scenario share one `docker run`. A failure is cached too and raised again, so an image + whose `spec` fails is not run again for each scenario. The image under a tag does not + change within a session. """ - spec_result = run_docker_airbyte_command( - [ - "docker", - "run", - "--rm", - connector_image, - "spec", - ], - raise_if_errors=True, - ) - return get_single_spec(spec_result, connector_name=connector_name) + outcome = _docker_spec_cache.get(connector_image) + if outcome is None: + try: + spec_result = run_docker_airbyte_command( + [ + "docker", + "run", + "--rm", + connector_image, + "spec", + ], + raise_if_errors=True, + ) + outcome = get_single_spec(spec_result, connector_name=connector_name) + except Exception as error: # `get_single_spec` raises `AssertionError`. + outcome = error + _docker_spec_cache[connector_image] = outcome + if isinstance(outcome, Exception): + raise outcome + return outcome + + +_SPEC_FAILED_SKIP_REASON = ( + "`spec` failed, so the secret leak check cannot tell which config values are secret, and " + "the check outcome is not asserted without it. The spec tests report the `spec` failure." +) class DockerConnectorTestSuite: @@ -335,9 +354,13 @@ def test_docker_image_build_and_check( # The leak check runs first, whatever the outcome: the assertions below print the raw # CONNECTION_STATUS and error messages, which are the likeliest to carry a secret. connector_name = connector_root.absolute().name + try: + spec = _run_docker_spec(connector_image, connector_name=connector_name) + except Exception: + pytest.skip(_SPEC_FAILED_SKIP_REASON) assert_no_secrets_in_output( check_result, - spec=_run_docker_spec(connector_image, connector_name=connector_name), + spec=spec, config=scenario.get_config_dict(connector_root=connector_root, empty_if_missing=True), verb="check", connector_name=connector_name, diff --git a/unit_tests/test/test_spec_lint.py b/unit_tests/test/test_spec_lint.py index 5d0111cc3e..e36704c794 100644 --- a/unit_tests/test/test_spec_lint.py +++ b/unit_tests/test/test_spec_lint.py @@ -4,11 +4,14 @@ import base64 import json import logging +import subprocess from collections.abc import Iterator, Mapping from pathlib import Path -from typing import Any +from typing import Any, Optional +from urllib.parse import quote import pytest +from pydantic import BaseModel, Field from airbyte_cdk.models import ( AirbyteCatalog, @@ -18,9 +21,9 @@ Status, ) from airbyte_cdk.sources.source import Source -from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput +from airbyte_cdk.test.entrypoint_wrapper import AirbyteEntrypointException, EntrypointOutput from airbyte_cdk.test.models import ConnectorTestScenario -from airbyte_cdk.test.standard_tests import docker_base +from airbyte_cdk.test.standard_tests import connector_base, docker_base, source_base from airbyte_cdk.test.standard_tests._spec_lint import ( assert_no_secrets_in_output, assert_spec_is_valid, @@ -33,6 +36,7 @@ from airbyte_cdk.test.standard_tests.connector_base import ConnectorTestSuiteBase from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite from airbyte_cdk.test.standard_tests.source_base import SourceTestSuiteBase +from airbyte_cdk.utils import docker as docker_utils from airbyte_cdk.utils.airbyte_secrets_utils import get_secrets SECRET = "sk_live_51UWRAsFuIbeygfIY3" @@ -566,14 +570,78 @@ def test_leak_report_contains_no_part_of_a_multi_line_secret( ), pytest.param( _spec({"headers": {"type": "object", "airbyte_secret": True}}), - {"headers": {"X-Token": "tok-1", "extra": ["tok-2"]}}, - [ - ("/headers", "X-Token"), - ("/headers", "tok-1"), - ("/headers", "extra"), - ("/headers", "tok-2"), - ], - id="secret_object_never_names_its_keys", + {"headers": {"X-Token": "tok-1", "region2025x": ["tok-2"]}}, + [("/headers", "tok-1"), ("/headers", "tok-2")], + id="secret_object_yields_values_not_keys", + ), + pytest.param( + _spec( + { + "credentials": { + "type": "object", + "oneOf": [ + { + "properties": { + "auth": {"const": "oauth"}, + "client_id": {"type": "string", "airbyte_secret": True}, + } + }, + { + "properties": { + "auth": {"enum": ["basic"]}, + "client_id": {"type": "string"}, + "password": {"type": "string", "airbyte_secret": True}, + } + }, + ], + } + } + ), + {"credentials": {"auth": "basic", "client_id": "cid-12345678", "password": "p-1"}}, + [("/credentials/password", "p-1")], + id="one_of_walks_only_the_selected_variant", + ), + pytest.param( + _spec( + { + "credentials": { + "type": "object", + "oneOf": [ + {"properties": {"client_secret": {"airbyte_secret": True}}}, + {"properties": {"api_key": {"airbyte_secret": True}}}, + ], + } + } + ), + {"credentials": {"client_secret": "s-1", "api_key": "k-1"}}, + [("/credentials/client_secret", "s-1"), ("/credentials/api_key", "k-1")], + id="one_of_without_discriminators_walks_every_variant", + ), + pytest.param( + _spec( + { + "credentials": { + "type": "object", + "oneOf": [ + { + "properties": { + "auth": {"const": "oauth"}, + "client_secret": {"airbyte_secret": True}, + } + }, + { + "properties": { + "auth": {"const": "token"}, + "api_key": {"airbyte_secret": True}, + } + }, + ], + } + } + ), + {"credentials": {"client_secret": "s-1", "api_key": "k-1"}}, + [("/credentials/client_secret", "s-1"), ("/credentials/api_key", "k-1")], + id="one_of_with_no_matching_discriminator_walks_every_variant", ), ], ) @@ -609,7 +677,7 @@ def test_find_config_secrets( def test_find_config_secrets_finds_every_runtime_secret( spec: dict[str, Any], config: dict[str, Any] ) -> None: - """Every value the runtime log filter masks is also searched for in the output.""" + """Every value the runtime log filter masks, in the variant the config selects, is searched.""" found = {value for _, value in find_config_secrets(spec, config)} assert set(get_secrets(spec, config)) <= found @@ -659,9 +727,9 @@ def get_connector_root_dir(cls) -> Path: @pytest.fixture(autouse=True) def _clear_docker_spec_cache() -> Iterator[None]: - docker_base._run_docker_spec.cache_clear() + docker_base._docker_spec_cache.clear() yield - docker_base._run_docker_spec.cache_clear() + docker_base._docker_spec_cache.clear() def _fake_docker( @@ -869,3 +937,369 @@ def test_in_process_check_test_fails_when_check_prints_a_secret( suite_test(_InProcessSuite(), scenario) assert "`/api_key` in message #" in str(error.value) assert SECRET not in str(error.value) + + +class _OptionalSecretSpec(BaseModel): + """How a pydantic-v2 spec declares an optional secret: the flag sits next to `anyOf`.""" + + api_key: Optional[str] = Field(None, json_schema_extra={"airbyte_secret": True}) + + +class _OptionalUnmarkedSpec(BaseModel): + api_key: Optional[str] = None + + +_NULLABLE_STRING = [{"type": "string"}, {"type": "null"}] +_MARKED_STRING_OR_NULL = [{"type": "string", "airbyte_secret": True}, {"type": "null"}] + + +@pytest.mark.parametrize( + "connection_specification, expected_errors", + [ + pytest.param( + _OptionalSecretSpec.model_json_schema(), + [], + id="pydantic_v2_optional_secret_passes", + ), + pytest.param( + _OptionalUnmarkedSpec.model_json_schema(), + ["`/properties/api_key` looks like a secret"], + id="pydantic_v2_optional_unmarked_secret_fails_once_at_the_property", + ), + pytest.param( + _spec({"api_key": {"anyOf": _MARKED_STRING_OR_NULL}}), + [ + "`/properties/api_key/anyOf/0` sets `airbyte_secret: true` under `anyOf`, where " + "the CDK's secret filter does not look, so the value is not masked in logs. Set it " + "on `/properties/api_key` instead." + ], + id="flag_inside_any_of_variant_fails", + ), + pytest.param( + _spec({"api_key": {"allOf": _MARKED_STRING_OR_NULL[:1]}}), + ["`/properties/api_key/allOf/0` sets `airbyte_secret: true` under `allOf`"], + id="flag_inside_all_of_variant_fails", + ), + pytest.param( + _spec({"api_key": {"oneOf": _MARKED_STRING_OR_NULL}}), + [], + id="flag_inside_one_of_variant_passes", + ), + pytest.param( + _spec( + { + "credentials": { + "anyOf": [ + { + "type": "object", + "properties": { + "token": {"type": "string", "airbyte_secret": True}, + }, + } + ] + } + } + ), + [ + "`/properties/credentials/anyOf/0/properties/token` sets `airbyte_secret: true` " + "under `anyOf`, where the CDK's secret filter does not look, so the value is not " + "masked in logs. Declare the variants with `oneOf` instead of `anyOf`." + ], + id="flag_on_a_property_of_an_any_of_object_fails", + ), + pytest.param( + { + "type": "object", + "oneOf": [ + {"properties": {"api_key": {"type": "string", "airbyte_secret": True}}}, + ], + }, + [ + "`/oneOf/0/properties/api_key` sets `airbyte_secret: true` under `oneOf`, where " + "the CDK's secret filter does not look, so the value is not masked in logs. The " + "filter reads only the top-level `properties` of the spec." + ], + id="flag_under_a_top_level_one_of_fails", + ), + pytest.param( + _spec({"password": {"airbyte_secret": True, "anyOf": [{"type": "boolean"}]}}), + [ + "`/properties/password` is marked `airbyte_secret: true` but its type `boolean` " + "cannot hold a secret value." + ], + id="marked_property_with_only_boolean_variants_fails", + ), + pytest.param( + _spec({"api_key": {"anyOf": _NULLABLE_STRING}}), + ["`/properties/api_key` looks like a secret"], + id="unmarked_property_typed_by_its_variants_fails", + ), + pytest.param( + _spec({"key": {"type": "string", "airbyte_secret": False}}), + [], + id="explicit_false_records_a_non_secret", + ), + pytest.param( + _spec({"sort": {"type": "object", "properties": {"key": {"type": "string"}}}}), + [ + "`/properties/sort/properties/key` looks like a secret but is not marked " + "`airbyte_secret: true`. If it holds no secret, set `airbyte_secret: false` on it " + "explicitly." + ], + id="failure_text_names_the_opt_out", + ), + ], +) +def test_find_secret_marking_errors_follows_the_runtime_secret_filter( + connection_specification: dict[str, Any], + expected_errors: list[str], +) -> None: + errors = find_secret_marking_errors(connection_specification) + assert len(errors) == len(expected_errors), errors + for error, expected_error in zip(errors, expected_errors): + assert error.startswith(expected_error), error + + +@pytest.mark.parametrize( + "connection_specification, config", + [ + pytest.param(_OptionalSecretSpec.model_json_schema(), {"api_key": SECRET}, id="optional"), + pytest.param(_spec({"api_key": {"anyOf": _MARKED_STRING_OR_NULL}}), {"api_key": SECRET}), + pytest.param( + _spec({"api_key": {"allOf": _MARKED_STRING_OR_NULL[:1]}}), {"api_key": SECRET} + ), + pytest.param(_spec({"api_key": {"oneOf": _MARKED_STRING_OR_NULL}}), {"api_key": SECRET}), + pytest.param( + _spec( + { + "credentials": { + "anyOf": [ + {"properties": {"token": {"type": "string", "airbyte_secret": True}}} + ] + } + } + ), + {"credentials": {"token": SECRET}}, + id="any_of_object", + ), + pytest.param( + _spec( + { + "credentials": { + "oneOf": [ + {"properties": {"token": {"type": "string", "airbyte_secret": True}}} + ] + } + } + ), + {"credentials": {"token": SECRET}}, + id="one_of_object", + ), + pytest.param( + { + "type": "object", + "oneOf": [{"properties": {"api_key": {"type": "string", "airbyte_secret": True}}}], + }, + {"api_key": SECRET}, + id="top_level_one_of", + ), + ], +) +def test_marking_lint_agrees_with_the_runtime_secret_filter( + connection_specification: dict[str, Any], config: dict[str, Any] +) -> None: + """The lint reports a flag the runtime cannot resolve exactly when `get_secrets` misses it.""" + errors = find_secret_marking_errors(connection_specification) + lint_says_unmasked = any("where the CDK's secret filter does not look" in e for e in errors) + runtime_masks = SECRET in get_secrets(connection_specification, config) + assert lint_says_unmasked != runtime_masks, errors + + +def test_find_leaked_secrets_finds_a_numeric_secret_emitted_as_a_json_number() -> None: + record = { + "type": "RECORD", + "record": {"stream": "pins", "data": {"pin": 12345678}, "emitted_at": 0}, + } + leaks = find_leaked_secrets(_output(record), [("/pin", 12345678)]) + assert leaks == ["`/pin` in message #1 (RECORD, stream `pins`)"] + + +@pytest.mark.parametrize( + "secret, printed_text", + [ + pytest.param('ab"cd\\1234ef', lambda config: json.dumps(config), id="json_escaped"), + pytest.param( + "p@ss/w0rd&123", + lambda config: "GET https://api.example.com/?token=" + + quote(config["api_key"], safe=""), + id="url_encoded", + ), + pytest.param("it's\"s3cret", lambda config: f"config: {config!r}", id="repr_escaped"), + ], +) +def test_find_leaked_secrets_finds_each_encoded_form(secret: str, printed_text: Any) -> None: + text = printed_text({"api_key": secret}) + assert secret not in text, "the literal value must not match, so only the encoded form can" + leaks = find_leaked_secrets(_output(_log(text)), [("/api_key", secret)]) + assert leaks == ["`/api_key` in message #1 (LOG)"] + + +def test_find_leaked_secrets_shows_the_stream_despite_a_short_unrelated_secret() -> None: + record = { + "type": "RECORD", + "record": {"stream": "users", "data": {"t": SECRET}, "emitted_at": 0}, + } + leaks = find_leaked_secrets(_output(record), [("/api_key", SECRET), ("/pin", "s")]) + assert leaks == ["`/api_key` in message #1 (RECORD, stream `users`)"] + + +def test_leak_in_a_cdk_config_validation_status_is_labeled() -> None: + """A CDK older than the masking fix leaks the value; the failure says the CDK built it.""" + output = _output( + _connection_status("FAILED", f"Config validation error: '{SECRET}' does not match '^a$'") + ) + with pytest.raises(AssertionError) as error: + assert_no_secrets_in_output( + output, + spec=ConnectorSpecification( + connectionSpecification=_SECRET_SPEC_MESSAGE["spec"]["connectionSpecification"] + ), + config={"api_key": SECRET}, + verb="check", + connector_name="source-test", + ) + failure_text = str(error.value) + assert ( + "`/api_key` in message #1 (CONNECTION_STATUS, built by the CDK's config validation)" + in failure_text + ) + assert "Upgrade `airbyte-cdk`, or make the scenario config pass the spec." in failure_text + assert SECRET not in failure_text + + +class _PatternSecretSource(_LeakySource): + """A source whose spec rejects the scenario's API key, so only the CDK reports on it.""" + + def spec(self, logger: logging.Logger) -> ConnectorSpecification: + return ConnectorSpecification( + connectionSpecification=_spec( + {"api_key": {"type": "string", "pattern": "^[a-z]+$", "airbyte_secret": True}} + ) + ) + + def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteConnectionStatus: + raise AssertionError("connector code must not run when the config fails validation") + + +@pytest.mark.parametrize( + "suite_test", + [ + pytest.param(SourceTestSuiteBase.test_check, id="source_suite"), + pytest.param(ConnectorTestSuiteBase.test_check, id="connector_suite"), + ], +) +def test_in_process_check_passes_when_the_config_fails_validation( + monkeypatch: pytest.MonkeyPatch, suite_test: Any +) -> None: + """The CDK masks the secret in its own `Config validation error` CONNECTION_STATUS.""" + monkeypatch.setattr(_InProcessSuite, "connector", _PatternSecretSource) + scenario = ConnectorTestScenario(config_dict={"api_key": "abcd1234"}, status="failed") + suite_test(_InProcessSuite(), scenario) + + +class _UnmarkedSpecSource(_LeakySource): + def spec(self, logger: logging.Logger) -> ConnectorSpecification: + return ConnectorSpecification( + connectionSpecification=_spec({"api_key": {"type": "string"}}) + ) + + +def test_in_process_spec_test_lints_the_spec(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(_InProcessSuite, "connector", _UnmarkedSpecSource) + with pytest.raises(AssertionError, match="/properties/api_key"): + _InProcessSuite().test_spec() + + +def test_in_process_spec_test_requires_a_spec_message(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(source_base, "run_test_job", lambda *args, **kwargs: _output(_log("hi"))) + with pytest.raises(AssertionError, match="emitted 0 SPEC messages"): + _InProcessSuite().test_spec() + + +class _BrokenSpecSource(_LeakySource): + def spec(self, logger: logging.Logger) -> ConnectorSpecification: + raise ValueError("No module named 'source_test.run'") + + +def test_in_process_check_is_skipped_when_spec_fails(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(_InProcessSuite, "connector", _BrokenSpecSource) + scenario = ConnectorTestScenario(config_dict={"api_key": SECRET}, status="failed") + with pytest.raises(pytest.skip.Exception, match="`spec` failed"): + _InProcessSuite().test_check(scenario) + + +def test_in_process_check_is_skipped_when_the_spec_run_raises( + monkeypatch: pytest.MonkeyPatch, +) -> None: + real_run_test_job = connector_base.run_test_job + + def run_test_job(connector: Any, verb: str, **kwargs: Any) -> EntrypointOutput: + if verb == "spec": + raise AirbyteEntrypointException("spec failed") + return real_run_test_job(connector, verb, **kwargs) + + monkeypatch.setattr(connector_base, "run_test_job", run_test_job) + scenario = ConnectorTestScenario(config_dict={"api_key": SECRET}, status="failed") + with pytest.raises(pytest.skip.Exception, match="`spec` failed"): + ConnectorTestSuiteBase.test_check(_InProcessSuite(), scenario) + + +_TRACE_ERROR = { + "type": "TRACE", + "trace": {"type": "ERROR", "emitted_at": 0, "error": {"message": "Invalid API key"}}, +} + + +def test_docker_check_accepts_a_trace_error_on_a_failure_scenario( + monkeypatch: pytest.MonkeyPatch, +) -> None: + check_output = _output(_TRACE_ERROR, _connection_status("FAILED", "Invalid API key")) + _fake_docker(monkeypatch, {"spec": _output(_SECRET_SPEC_MESSAGE), "check": check_output}) + _run_docker_check(ConnectorTestScenario(config_dict={"api_key": SECRET}, status="failed")) + + +def test_docker_check_raises_a_trace_error_on_a_success_scenario( + monkeypatch: pytest.MonkeyPatch, +) -> None: + check_output = _output(_TRACE_ERROR, _connection_status("SUCCEEDED", "ok")) + _fake_docker(monkeypatch, {"spec": _output(_SECRET_SPEC_MESSAGE), "check": check_output}) + with pytest.raises(AirbyteEntrypointException, match="Invalid API key"): + _run_docker_check(ConnectorTestScenario(config_dict={"api_key": SECRET}, status="succeed")) + + +def test_docker_spec_failure_runs_spec_once_and_skips_the_check_tests( + monkeypatch: pytest.MonkeyPatch, +) -> None: + check_output = _output(_connection_status("FAILED", "Invalid API key")) + commands = _fake_docker(monkeypatch, {"spec": _output(_TRACE_ERROR), "check": check_output}) + with pytest.raises(AirbyteEntrypointException): + _DockerSuite().test_docker_image_build_and_spec( + connector_image_override="source-test:dev", + connector_base_image_override=None, + ) + for _ in range(2): + with pytest.raises(pytest.skip.Exception, match="`spec` failed"): + _run_docker_check( + ConnectorTestScenario(config_dict={"api_key": SECRET}, status="failed") + ) + + assert [command[-1] for command in commands if "spec" in command] == ["spec"] + + +def test_run_docker_airbyte_command_keeps_stderr(monkeypatch: pytest.MonkeyPatch) -> None: + def run_docker_command(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="warning: retrying\n") + + monkeypatch.setattr(docker_utils, "run_docker_command", run_docker_command) + output = docker_utils.run_docker_airbyte_command(["docker", "run", "image", "spec"]) + assert output.stderr == "warning: retrying\n" From fff068e84e4356a2fde0a4d38fa2f145b9f1538c Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Fri, 9 Oct 2026 19:31:15 +0300 Subject: [PATCH 8/8] docs: describe the new spec lint and secret leak check in cdk-migrations.md Co-Authored-By: Claude Opus 5.5 --- cdk-migrations.md | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/cdk-migrations.md b/cdk-migrations.md index 3445a04b88..f79dcbe7b1 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -1,5 +1,28 @@ # CDK Migration Guide +## Upgrading to the next minor release (unreleased) + +The next minor release of the CDK adds two checks to the standard tests. Every Python and manifest-only connector that runs the standard tests, for example through `airbyte-cdk connector test` or `airbyte-cdk image test`, picks them up when it moves to this release. There is no change to how connectors run. + +### `airbyte_secret` marking in the spec tests + +`SourceTestSuiteBase.test_spec` and `test_docker_image_build_and_spec` now require exactly one `SPEC` message and lint its `connectionSpecification`. The test fails when: + +- a property whose name denotes a credential (such as `api_key`, `client_secret`, `password` or `*_token`) can hold a string or number but is not marked `airbyte_secret: true`; +- a property is marked `airbyte_secret: true` but its type (boolean, object, array, or a `const`) cannot hold a secret; +- `airbyte_secret` is not a boolean, such as the string `"true"`; +- `airbyte_secret: true` sits under `anyOf` or `allOf`, or under a combinator at the top level of the spec, where the CDK's secret filter does not look. + +Migration steps: mark the property `airbyte_secret: true`, or move the flag to the property that declares the `anyOf`. If a flagged property holds no secret, set `airbyte_secret: false` on it explicitly; the lint accepts that as a reviewed decision. + +### Secret leak check in the `check` tests + +`test_check` and `test_docker_image_build_and_check` now fail when an `airbyte_secret` value from the scenario's config, of at least 8 characters and containing a digit, appears in the connector's output: any non-`CONTROL` message, and stderr on the Docker path. The failure names the config path of each leaked secret and where it was found, never the secret itself. + +Migration steps: stop printing the value; a `CONNECTION_STATUS` message that the connector builds itself is the usual cause. A leak in a `CONNECTION_STATUS` labeled as built by the CDK's config validation comes from an older CDK in the connector image, which did not mask secrets there; upgrade `airbyte-cdk`. + +Rationale: these checks restore the `test_secret_is_properly_marked` assertion of the retired Connector Acceptance Tests, and add the leak check that `test_secret_never_in_the_output` never implemented. + ## Upgrading to 7.0.0 [Version 7.0.0](https://github.com/airbytehq/airbyte-python-cdk/releases/tag/v7.0.0) of the CDK migrates the CDK to the Concurrent CDK by removing some of the Declarative CDK concepts that are better expressed in the Concurrent CDK or that are outright incompatible with it. This changes mostly impact the Python implementations although the concept of CustomIncrementalSync has been removed from the declarative language as well.