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/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 new file mode 100644 index 0000000000..b2a0ce6f7a --- /dev/null +++ b/airbyte_cdk/test/standard_tests/_spec_lint.py @@ -0,0 +1,657 @@ +# 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 dataclasses import dataclass +from typing import Any +from urllib.parse import quote + +from airbyte_cdk.models import ( + AirbyteMessage, + AirbyteMessageSerializer, + ConnectorSpecification, + Type, +) +from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput + +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. + +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") + +_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. + +Short values, such as `123` in a test config, occur in ordinary output by coincidence. +""" + + +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 _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. A property + that declares its types only under a combinator can hold a secret if any variant can. + """ + if "const" in schema: + return False + 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. + + 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(), + 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 _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): + child_pointer = f"{pointer}/properties/{property_name}" + yield from _iter_spec_properties( + property_schema, + 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 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_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. 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 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( + 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 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 = _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`. " + "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"`{_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 + 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 _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 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 item in value.values(): + yield from _scalars(item) + elif isinstance(value, list): + for item in value: + yield from _scalars(item) + elif value is not None: + 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, + pointer: str = "", +) -> Iterator[tuple[str, Any]]: + """Yield `(config_pointer, value)` for each value in `config` that `schema` marks secret. + + 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): + if keyword == "oneOf": + variants = _selected_variants(variants, config) + 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 `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))) + + +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 + test configs, appear in ordinary output by coincidence. + """ + 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 _string_values(value: Any) -> Iterator[str]: + """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) + elif isinstance(value, list): + for item in value: + 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 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. + """ + secrets = list(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 = set().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 number, message in enumerate(output.get_message_iterator(), start=1): + if message.type == Type.CONTROL: + continue + 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}`" + 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() + 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 + + +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. + + 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) + 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) + hint + ) diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py index 1c9bf24e10..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 @@ -15,7 +16,14 @@ ConnectorTestScenario, ) from airbyte_cdk.test.standard_tests._job_runner import IConnector, run_test_job -from airbyte_cdk.test.standard_tests.docker_base import DockerConnectorTestSuite +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 ( + _SPEC_FAILED_SKIP_REASON, + DockerConnectorTestSuite, +) if TYPE_CHECKING: from collections.abc import Callable @@ -100,6 +108,40 @@ 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`. + + Pass this to `run_test_job` as `inspect_output`, so it runs before the job's own + 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. + """ + 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=spec, + 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( @@ -112,6 +154,9 @@ 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. " diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index 84cfab2b00..aab32e154e 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,46 @@ def _assert_check_outcome( ) +_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 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. + """ + 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: """Base class for connector test suites.""" @@ -227,7 +273,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 +294,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, @@ -305,11 +348,28 @@ 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 + 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=spec, + 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 @@ -317,7 +377,7 @@ 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_name, ) @pytest.mark.skipif( diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py index 18f3fa1792..6146a3e354 100644 --- a/airbyte_cdk/test/standard_tests/source_base.py +++ b/airbyte_cdk/test/standard_tests/source_base.py @@ -25,6 +25,7 @@ format_catalog_problems, ) 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,9 @@ 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, ( @@ -117,7 +121,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. @@ -130,10 +134,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/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/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/cdk-migrations.md b/cdk-migrations.md index 249b6256b2..5a87684f58 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -2,7 +2,7 @@ ## Upgrading to the next minor release (unreleased) -The next minor release of the CDK adds catalog checks to the `test_discover` standard test (`SourceTestSuiteBase.test_discover`). Every source connector that runs the standard tests, for example through `airbyte-cdk connector test`, picks them up when it moves to this release. There is no change to how connectors run. +The next minor release of the CDK adds 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. ### New mandatory checks in `test_discover` @@ -31,6 +31,25 @@ Migration steps: fix the stream schema, primary key or cursor field that the fai Rationale: these checks restore the `TestDiscovery` assertions of the retired Connector Acceptance Tests, so that catalog problems are caught in the connector's own tests instead of in a destination or in a customer's sync. +### `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. 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/test_spec_lint.py b/unit_tests/test/test_spec_lint.py new file mode 100644 index 0000000000..e36704c794 --- /dev/null +++ b/unit_tests/test/test_spec_lint.py @@ -0,0 +1,1305 @@ +# 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 +import logging +import subprocess +from collections.abc import Iterator, Mapping +from pathlib import Path +from typing import Any, Optional +from urllib.parse import quote + +import pytest +from pydantic import BaseModel, Field + +from airbyte_cdk.models import ( + AirbyteCatalog, + AirbyteConnectionStatus, + AirbyteMessage, + ConnectorSpecification, + Status, +) +from airbyte_cdk.sources.source import Source +from airbyte_cdk.test.entrypoint_wrapper import AirbyteEntrypointException, EntrypointOutput +from airbyte_cdk.test.models import ConnectorTestScenario +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, + find_config_secrets, + find_leaked_secrets, + find_secret_marking_errors, + 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 import docker as docker_utils +from airbyte_cdk.utils.airbyte_secrets_utils import get_secrets + +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"), + 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", + [ + 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", + ), + 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( + 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, location", + [ + pytest.param( + _log(f"Calling https://api.example.com/?key={SECRET}"), + "message #1 (LOG)", + id="log", + ), + pytest.param( + { + "type": "TRACE", + "trace": { + "type": "ERROR", + "emitted_at": 0, + "error": {"message": "Request failed", "stack_trace": f"token={SECRET}"}, + }, + }, + "message #1 (TRACE)", + id="trace_stack_trace", + ), + pytest.param( + { + "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_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_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", + "control": { + "type": "CONNECTOR_CONFIG", + "emitted_at": 0, + "connectorConfig": {"config": {"api_key": SECRET}}, + }, + } + assert find_leaked_secrets(_output(control_message), [("/api_key", 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(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", secret)]) == [] + + +_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 _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())), + ], + stderr=text, + ) + 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 "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 + 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", "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", + ), + ], +) +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, 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 + + +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 + + +@pytest.fixture(autouse=True) +def _clear_docker_spec_cache() -> Iterator[None]: + docker_base._docker_spec_cache.clear() + yield + docker_base._docker_spec_cache.clear() + + +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) + if raise_if_errors: + outputs[verb].raise_if_errors() + 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, + ) + + +_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", + [ + 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: + 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") + + if should_pass: + _run_docker_check(scenario) + 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) + + +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.""" + + 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) + + +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" 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$'" + )