diff --git a/ansible/roles/operator-pipeline/templates/openshift/pipelines/operator-hosted-pipeline.yml b/ansible/roles/operator-pipeline/templates/openshift/pipelines/operator-hosted-pipeline.yml index 7e9577956..727f841d5 100644 --- a/ansible/roles/operator-pipeline/templates/openshift/pipelines/operator-hosted-pipeline.yml +++ b/ansible/roles/operator-pipeline/templates/openshift/pipelines/operator-hosted-pipeline.yml @@ -501,6 +501,9 @@ spec: - name: output workspace: results subPath: summary + - name: changes + workspace: results + subPath: changes - name: content-hash taskRef: diff --git a/ansible/roles/operator-pipeline/templates/openshift/tasks/parse-repo-changes.yml b/ansible/roles/operator-pipeline/templates/openshift/tasks/parse-repo-changes.yml index 673fe80a2..7792d4895 100644 --- a/ansible/roles/operator-pipeline/templates/openshift/tasks/parse-repo-changes.yml +++ b/ansible/roles/operator-pipeline/templates/openshift/tasks/parse-repo-changes.yml @@ -111,3 +111,6 @@ spec: deleted_catalog_operators="$(jq -r '.deleted_catalog_operators | join(",")' < changes.json)" echo -n $deleted_catalog_operators > "$(results.deleted_catalog_operators.path)" + + # Newline-separated paths under operators/ (avoids Tekton result size limits). + jq -r '.affected_operator_files[]?' < changes.json > affected_operator_files.txt diff --git a/ansible/roles/operator-pipeline/templates/openshift/tasks/run-static-tests.yml b/ansible/roles/operator-pipeline/templates/openshift/tasks/run-static-tests.yml index db6d4fa28..4d1a2fd39 100644 --- a/ansible/roles/operator-pipeline/templates/openshift/tasks/run-static-tests.yml +++ b/ansible/roles/operator-pipeline/templates/openshift/tasks/run-static-tests.yml @@ -54,6 +54,11 @@ spec: - name: output description: Storage for the results file (json) + - name: changes + description: | + Output from parse-repo-changes. Expected file: + affected_operator_files.txt (newline-separated repo-relative paths). + steps: - name: run-suite image: "$(params.pipeline_image)" @@ -68,8 +73,12 @@ spec: #!/usr/bin/env bash set -xe - if [ -z "$(params.bundle_path)" ] && [ -z "$(params.affected_catalog_operators)" ]; then - echo "No bundle or catalog added/changed, skipping tests" + AFFECTED_OPERATOR_FILES_FILE="$(workspaces.changes.path)/affected_operator_files.txt" + + if [ -z "$(params.bundle_path)" ] \ + && [ -z "$(params.affected_catalog_operators)" ] \ + && { [ ! -f "$AFFECTED_OPERATOR_FILES_FILE" ] || [ ! -s "$AFFECTED_OPERATOR_FILES_FILE" ]; }; then + echo "No bundle, catalog, or operator files added/changed, skipping tests" echo -n "" | tee $(results.messages_count.path) echo -n "" | tee $(results.failures_count.path) exit 0 @@ -88,12 +97,21 @@ spec: EXTRA_ARGS+=" --skip-tests $SKIP_TESTS" fi + AFFECTED_FILES_ARGS=() + if [ -f "$AFFECTED_OPERATOR_FILES_FILE" ]; then + AFFECTED_FILES_ARGS=( + --affected-operator-files-path "$AFFECTED_OPERATOR_FILES_FILE" + ) + fi + JSON_RESULTS_FILE="$(workspaces.output.path)/static-test-results.json" static-tests --verbose \ --output-file "$JSON_RESULTS_FILE" \ --repo-path "$(workspaces.source.path)" \ - --suites "$(params.test_suites)" $EXTRA_ARGS \ + --suites "$(params.test_suites)" \ + "${AFFECTED_FILES_ARGS[@]}" \ + $EXTRA_ARGS \ "$(params.operator_name)" "$(params.bundle_version)" \ "$(params.affected_catalog_operators)" diff --git a/docs/users/static_checks.md b/docs/users/static_checks.md index ed55c6090..a9e7a3c5b 100644 --- a/docs/users/static_checks.md +++ b/docs/users/static_checks.md @@ -129,6 +129,19 @@ name in the CSV definition. The source of these values are: - `operators.operatorframework.io.bundle.package.v1` (`metadata/annotation.yaml`) - `csv.metadata.name` - the name without a version (`manifests/.*.clusterserviceversion.yaml`) +#### check_leaks_in_changed_files +The test scans files under `operators//` that were added or modified +in the pull request for potential secret leaks (tokens, credentials, and similar +secrets) using [LeakTK](https://github.com/leaktk/leaktk). + +Only paths under the operator directory are checked; catalog files are not +scanned. The check fails if any leak is detected and reports the relative file +path without including secret content. + +If this is intentional or a false positive, you can skip the check by adding +the `tests/skip/check_leaks_in_changed_files` label to the pull request +(or `/test skip check_leaks_in_changed_files`). + #### check_bundle_images_in_fbc This check will ensure that all bundle images in the file based catalog for given operator catalog(s) use allowed image registry. Allowed registries are configured diff --git a/operatorcert/entrypoints/detect_changed_operators.py b/operatorcert/entrypoints/detect_changed_operators.py index f78e52151..20fcb522e 100644 --- a/operatorcert/entrypoints/detect_changed_operators.py +++ b/operatorcert/entrypoints/detect_changed_operators.py @@ -497,12 +497,21 @@ def detect_changes( head_repo, base_repo, all_affected_catalog_operators ) + # Operator-directory files from the PR that still exist on head (for leak scanning). + # Catalog paths and deleted files are intentionally excluded. + affected_operator_files = sorted( + path + for path in pr_files + if path.startswith("operators/") and (head_repo.root / path).is_file() + ) + parsed_results = ParserResults( affected_operators=operators, affected_bundles=bundles, affected_catalogs=catalogs, affected_catalog_operators=catalog_operators, extra_files=non_operator_files, + affected_operator_files=affected_operator_files, ) return parsed_results diff --git a/operatorcert/entrypoints/static_tests.py b/operatorcert/entrypoints/static_tests.py index be6eadfc1..232f17ba9 100644 --- a/operatorcert/entrypoints/static_tests.py +++ b/operatorcert/entrypoints/static_tests.py @@ -3,11 +3,13 @@ import argparse import json import logging +from pathlib import Path from typing import Any, Dict, List, Optional from operatorcert.logger import setup_logger from operatorcert.operator_repo import OperatorCatalogList, Repo from operatorcert.operator_repo.checks import Fail, run_suite +from operatorcert.static_tests.helpers import set_affected_operator_files from operatorcert.utils import SplitArgs LOGGER = logging.getLogger("operator-cert") @@ -46,6 +48,14 @@ def setup_argparser() -> argparse.ArgumentParser: default=[], action=SplitArgs, ) + parser.add_argument( + "--affected-operator-files-path", + help=( + "Path to a file listing repo-relative operator file paths " + "(one path per line) that were added or modified in the pull request" + ), + default=None, + ) parser.add_argument("--verbose", action="store_true", help="Verbose output") parser.add_argument("operator") parser.add_argument("bundle") @@ -54,6 +64,29 @@ def setup_argparser() -> argparse.ArgumentParser: return parser +def load_affected_operator_files(path: Optional[str]) -> List[str]: + """ + Load repo-relative operator file paths from a newline-separated file. + + Args: + path: Filesystem path to the list file, or None/empty for no files. + + Returns: + Non-empty stripped lines from the file, or an empty list. + """ + if not path: + return [] + file_path = Path(path) + if not file_path.is_file(): + LOGGER.warning("Affected operator files list not found: %s", path) + return [] + return [ + line.strip() + for line in file_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + def get_objects_to_test( repo: Repo, operator_name: str, bundle_version: str, affected_catalogs: str ) -> List[Any]: @@ -76,13 +109,14 @@ def get_objects_to_test( # We need to skip the test for that resource # Check if the operator and bundle exist in the repository - # and add them to the list of objects to test - if repo.has(operator_name): + # and add them to the list of objects to test. + # Empty names mean no bundle/operator change (e.g. operator-only file edits). + if operator_name and repo.has(operator_name): operator = repo.operator(operator_name) test_objects.append(operator) - if operator.has(bundle_version): + if bundle_version and operator.has(bundle_version): test_objects.append(operator.bundle(bundle_version)) - else: + elif operator_name: LOGGER.warning("Operator %s not found in the repository", operator_name) # Check if the affected catalogs and catalog operators exist in the repository @@ -93,16 +127,16 @@ def get_objects_to_test( affected_catalogs_list = [item for item in affected_catalogs.split(",") if item] for operator_catalog_path in affected_catalogs_list: - catalog_name, operator_name = operator_catalog_path.split("/") + catalog_name, catalog_operator_name = operator_catalog_path.split("/") if not repo.has_catalog(catalog_name) or not repo.catalog(catalog_name).has( - operator_name + catalog_operator_name ): LOGGER.warning( "Catalog %s not found in the repository", operator_catalog_path ) continue catalog = repo.catalog(catalog_name) - operator_catalog = catalog.operator_catalog(operator_name) + operator_catalog = catalog.operator_catalog(catalog_operator_name) operator_catalogs.append(operator_catalog) test_objects.append(OperatorCatalogList(operator_catalogs)) @@ -116,6 +150,7 @@ def execute_checks( # pylint: disable=too-many-arguments,too-many-positional-ar affected_catalogs: str, suite_names: List[str], skip_tests: Optional[List[str]] = None, + affected_operator_files: Optional[List[str]] = None, ) -> Dict[str, Any]: """ Run a check suite against the given target and return warnings and @@ -129,36 +164,42 @@ def execute_checks( # pylint: disable=too-many-arguments,too-many-positional-ar affected_catalogs (str): Coma separated list of affected catalogs suite_name (str): name of the suite to use skip_tests (Optional[List]): List of checks to skip + affected_operator_files (Optional[List]): Repo-relative paths under + operators/ affected by the pull request Returns: The results of the checks in the suite applied to the given bundle """ - repo = Repo(repo_path) - test_objects = get_objects_to_test( - repo, operator_name, bundle_version, affected_catalogs - ) - - outputs = [] - passed = True - - for suite_name in suite_names: - for result in run_suite( - test_objects, - suite_name, - skip_tests=skip_tests, - ): - if isinstance(result, Fail): - passed = False - item = { - "type": "error" if isinstance(result, Fail) else "warning", - "message": result.reason, - "test_suite": suite_name, - } - if result.check: - item["check"] = result.check - outputs.append(item) - - return {"passed": passed, "outputs": outputs} + set_affected_operator_files(affected_operator_files or []) + try: + repo = Repo(repo_path) + test_objects = get_objects_to_test( + repo, operator_name, bundle_version, affected_catalogs + ) + + outputs = [] + passed = True + + for suite_name in suite_names: + for result in run_suite( + test_objects, + suite_name, + skip_tests=skip_tests, + ): + if isinstance(result, Fail): + passed = False + item = { + "type": "error" if isinstance(result, Fail) else "warning", + "message": result.reason, + "test_suite": suite_name, + } + if result.check: + item["check"] = result.check + outputs.append(item) + + return {"passed": passed, "outputs": outputs} + finally: + set_affected_operator_files([]) def main() -> None: @@ -183,6 +224,7 @@ def main() -> None: args.affected_catalogs, args.suites, args.skip_tests, + load_affected_operator_files(args.affected_operator_files_path), ) if args.output_file: diff --git a/operatorcert/parsed_file.py b/operatorcert/parsed_file.py index 7f606a26a..8a920122e 100644 --- a/operatorcert/parsed_file.py +++ b/operatorcert/parsed_file.py @@ -1,7 +1,7 @@ """Module containing data classes and validators for PR parsed files""" from dataclasses import dataclass, field -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional import yaml from operatorcert import utils @@ -166,12 +166,16 @@ def __init__( # pylint: disable=too-many-arguments,too-many-positional-argument affected_catalogs: AffectedCatalogCollection, affected_catalog_operators: AffectedCatalogOperatorCollection, extra_files: set[str], + affected_operator_files: Optional[list[str]] = None, ): self.affected_operators = affected_operators self.affected_bundles = affected_bundles self.affected_catalogs = affected_catalogs self.affected_catalog_operators = affected_catalog_operators self.extra_files = extra_files + self.affected_operator_files = ( + list(affected_operator_files) if affected_operator_files is not None else [] + ) def to_dict(self) -> Dict[str, Any]: """ @@ -186,6 +190,7 @@ def to_dict(self) -> Dict[str, Any]: **(self.affected_catalogs.to_dict()), **(self.affected_catalog_operators.to_dict()), "extra_files": list(self.extra_files), + "affected_operator_files": list(self.affected_operator_files), } self.enrich_result(result) return result diff --git a/operatorcert/redact.py b/operatorcert/redact.py index 3d2d70f85..a29bf2ef0 100644 --- a/operatorcert/redact.py +++ b/operatorcert/redact.py @@ -12,7 +12,7 @@ from pathlib import Path import json -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict LOGGER = logging.getLogger("operator-cert") @@ -22,6 +22,8 @@ class RedactLocation(BaseModel): Class for tracking information about a chunk of data in files. """ + model_config = ConfigDict(extra="ignore") + path: Path @@ -30,6 +32,8 @@ class ResultModel(BaseModel): Model for parsing a single result from scanning. """ + model_config = ConfigDict(extra="ignore") + location: RedactLocation @@ -38,6 +42,8 @@ class ResultSetModel(BaseModel): Model for parsing all results from scanning. """ + model_config = ConfigDict(extra="ignore") + results: list[ResultModel] @@ -54,12 +60,30 @@ def scan(*input_paths: Path) -> list[ResultSetModel]: + "\n" for i, input_path in enumerate(input_paths) ) - results_jsonl = subprocess.check_output( - ["leaktk", "listen"], input=requests, text=True - ) - return list( - map(ResultSetModel.model_validate, map(json.loads, results_jsonl.splitlines())) - ) + try: + results_jsonl = subprocess.check_output( + ["leaktk", "listen"], + input=requests, + text=True, + stderr=subprocess.DEVNULL, + ) + except subprocess.CalledProcessError as exc: + # LeakTK output may contain secret match text; do not propagate it. + raise RuntimeError( + f"LeakTK scan failed with exit code {exc.returncode}" + ) from None + + parsed: list[ResultSetModel] = [] + for line in results_jsonl.splitlines(): + if not line.strip(): + continue + try: + parsed.append(ResultSetModel.model_validate(json.loads(line))) + except (json.JSONDecodeError, ValueError): + raise RuntimeError( + "LeakTK returned output that could not be parsed safely" + ) from None + return parsed def _redact( @@ -121,6 +145,25 @@ def redact_results(*result_sets: ResultSetModel) -> dict[Path, Path]: return _redact(*file_locations) +def paths_with_leaks(*input_paths: Path) -> set[Path]: + """ + Return absolute paths that LeakTK reported as containing leaks. + + Args: + *input_paths: Any path (directory or file) to scan. + + Returns: + Absolute paths of files with detected leaks. + """ + if not input_paths: + return set() + leaky: set[Path] = set() + for result_set in scan(*input_paths): + for result in result_set.results: + leaky.add(Path(result.location.path).absolute()) + return leaky + + def scan_and_redact(*input_paths: Path) -> dict[Path, Path]: """ Checks which files have leaked information and redacts them. diff --git a/operatorcert/static_tests/common/operator.py b/operatorcert/static_tests/common/operator.py index ff6d381bd..968637273 100644 --- a/operatorcert/static_tests/common/operator.py +++ b/operatorcert/static_tests/common/operator.py @@ -1,13 +1,79 @@ """A common test suite for operators""" import json +import logging import os from collections import defaultdict from collections.abc import Iterator +from pathlib import Path from jsonschema.validators import Draft202012Validator from operatorcert.operator_repo import Operator from operatorcert.operator_repo.checks import CheckResult, Fail +from operatorcert.redact import paths_with_leaks +from operatorcert.static_tests.helpers import get_affected_operator_files + +LOGGER = logging.getLogger("operator-cert") + + +def check_leaks_in_changed_files(operator: Operator) -> Iterator[CheckResult]: + """ + Scan pull-request-affected files under this operator for secret leaks using LeakTK. + + The list of affected files is provided by detect-changes via the static-tests + entrypoint (see set_affected_operator_files). Catalog paths are not included. + Fail messages report relative paths only and never include secret content. + """ + affected_files = get_affected_operator_files() + if not affected_files: + return + + operator_prefix = f"operators/{operator.operator_name}/" + repo_root = operator.repo.root + paths_to_scan: list[Path] = [] + for rel_path in affected_files: + if not rel_path.startswith(operator_prefix): + continue + absolute_path = repo_root / rel_path + if absolute_path.is_file(): + paths_to_scan.append(absolute_path) + + if not paths_to_scan: + return + + LOGGER.info( + "Scanning %d affected file(s) under %s for secret leaks", + len(paths_to_scan), + operator_prefix, + ) + try: + leaky_paths = paths_with_leaks(*paths_to_scan) + except Exception as exc: # pylint: disable=broad-except + # LeakTK stdout/errors can contain secret match text; never log or yield it. + LOGGER.error( + "LeakTK failed while scanning %d file(s) under %s (%s)", + len(paths_to_scan), + operator_prefix, + type(exc).__name__, + ) + yield Fail( + "Secret leak scan failed due to an internal error. " + "Re-run the pipeline or skip this check with the label " + "tests/skip/check_leaks_in_changed_files if needed." + ) + return + + for leaky_path in sorted(leaky_paths): + try: + rel_path = str(leaky_path.relative_to(repo_root)) + except ValueError: + rel_path = str(leaky_path) + yield Fail( + f"Potential secret leak detected in {rel_path}. " + "Remove secrets from the pull request before merging. " + "To skip this check, add the label " + "tests/skip/check_leaks_in_changed_files to the pull request." + ) def check_schema_operator_ci_config( diff --git a/operatorcert/static_tests/helpers.py b/operatorcert/static_tests/helpers.py index f006d5044..8d8bc659e 100644 --- a/operatorcert/static_tests/helpers.py +++ b/operatorcert/static_tests/helpers.py @@ -2,12 +2,28 @@ import logging from functools import wraps -from typing import Any, Callable, Iterator +from typing import Any, Callable, Iterator, Sequence from operatorcert.operator_repo import Bundle, Operator LOGGER = logging.getLogger("operator-cert") +_affected_operator_files: tuple[str, ...] = () + + +def set_affected_operator_files(files: Sequence[str]) -> None: + """ + Set the list of repo-relative operator files affected by the pull request. + Used by check_leaks_in_changed_files; call before run_suite. + """ + global _affected_operator_files # pylint: disable=global-statement + _affected_operator_files = tuple(files) + + +def get_affected_operator_files() -> tuple[str, ...]: + """Return repo-relative operator files affected by the pull request.""" + return _affected_operator_files + def skip_fbc(func: Callable[..., Any]) -> Callable[..., Any]: """ diff --git a/tests/conftest.py b/tests/conftest.py index 25c751981..dfc7f9f03 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,7 +4,16 @@ from operatorcert.catalog.package import CatalogPackage from operatorcert.catalog.channel import CatalogChannel from operatorcert.catalog.bundle import CatalogBundle -from typing import Optional +from operatorcert.static_tests.helpers import set_affected_operator_files +from typing import Iterator, Optional + + +@pytest.fixture(autouse=True) +def reset_affected_operator_files() -> Iterator[None]: + """Ensure module-global affected-file state does not leak across tests.""" + set_affected_operator_files([]) + yield + set_affected_operator_files([]) @pytest.fixture diff --git a/tests/entrypoints/test_detect_changed_operators.py b/tests/entrypoints/test_detect_changed_operators.py index 6d075cd76..dece220ea 100644 --- a/tests/entrypoints/test_detect_changed_operators.py +++ b/tests/entrypoints/test_detect_changed_operators.py @@ -339,6 +339,11 @@ def test_detect_changes( "added_or_modified_catalogs": [], "deleted_catalogs": [], "catalogs_with_added_or_modified_operators": [], + "affected_operator_files": sorted( + path + for path in affected_files + if path.startswith("operators/") and (after_dir / path).is_file() + ), } expected = {**default_expected, **expected} diff --git a/tests/entrypoints/test_static_tests.py b/tests/entrypoints/test_static_tests.py index 6cdb7c319..05cacde50 100644 --- a/tests/entrypoints/test_static_tests.py +++ b/tests/entrypoints/test_static_tests.py @@ -1,9 +1,10 @@ +from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch import pytest from operatorcert.entrypoints import static_tests -from operatorcert.operator_repo import Repo +from operatorcert.operator_repo import Bundle, Operator, OperatorCatalogList, Repo from operatorcert.operator_repo.checks import Fail, Warn from tests.utils import bundle_files, catalog_files, create_files @@ -128,6 +129,35 @@ def test_execute_checks( assert result == expected +def test_load_affected_operator_files(tmp_path: Path) -> None: + assert static_tests.load_affected_operator_files(None) == [] + assert static_tests.load_affected_operator_files(str(tmp_path / "missing")) == [] + + list_file = tmp_path / "affected_operator_files.txt" + list_file.write_text( + "operators/foo/ci.yaml\n\noperators/foo/bar.yaml\n", + encoding="utf-8", + ) + assert static_tests.load_affected_operator_files(str(list_file)) == [ + "operators/foo/ci.yaml", + "operators/foo/bar.yaml", + ] + + +def test_get_objects_to_test_operator_only_no_bundle(tmp_path: Path) -> None: + create_files(tmp_path, bundle_files("test-operator", "0.0.1")) + repo = Repo(tmp_path) + + objects = static_tests.get_objects_to_test(repo, "test-operator", "", "") + + assert len(objects) == 2 + assert isinstance(objects[0], Operator) + assert objects[0].operator_name == "test-operator" + assert isinstance(objects[1], OperatorCatalogList) + assert list(objects[1]) == [] + assert not any(isinstance(obj, Bundle) for obj in objects) + + @patch("operatorcert.entrypoints.static_tests.execute_checks") @patch("operatorcert.entrypoints.static_tests.setup_logger") def test_static_tests_main( @@ -150,6 +180,7 @@ def test_static_tests_main( ["v4.14/test-operator"], ["operatorcert.static_tests.community", "operatorcert.static_tests.common"], [], + [], ) assert capsys.readouterr().out.strip() == '{"foo": ["bar"]}' mock_logger.assert_called_once_with(level="INFO") @@ -159,6 +190,8 @@ def test_static_tests_main( out_file = tmpdir / "out.json" out_file_name = str(out_file) + affected_file = tmpdir / "affected_operator_files.txt" + affected_file.write_text("operators/other/ci.yaml\n", encoding="utf-8") args = [ "static-tests", "--repo-path=/tmp/other_repo", @@ -168,6 +201,7 @@ def test_static_tests_main( ["v4.14/test-operator"], "--suites=other_suite", f"--output-file={out_file_name}", + f"--affected-operator-files-path={affected_file}", "--verbose", ] mock_execute_checks.return_value = {"bar": ["baz"]} @@ -180,6 +214,7 @@ def test_static_tests_main( ["v4.14/test-operator"], ["other_suite"], ["check_123", "check_456"], + ["operators/other/ci.yaml"], ) assert out_file.read().strip() == '{"bar": ["baz"]}' mock_logger.assert_called_once_with(level="DEBUG") diff --git a/tests/static_tests/common/test_operator.py b/tests/static_tests/common/test_operator.py index bd3871bf1..2712ee865 100644 --- a/tests/static_tests/common/test_operator.py +++ b/tests/static_tests/common/test_operator.py @@ -1,13 +1,16 @@ from pathlib import Path from typing import Any +from unittest.mock import patch import pytest from operatorcert.operator_repo import Repo from operatorcert.operator_repo.checks import Fail from operatorcert.static_tests.common.operator import ( check_catalog_usage_ci_config, + check_leaks_in_changed_files, check_schema_operator_ci_config, ) +from operatorcert.static_tests.helpers import set_affected_operator_files from tests.utils import bundle_files, create_files @@ -275,3 +278,100 @@ def test_check_catalog_usage_ci_config( assert { (x.__class__, x.reason) for x in check_catalog_usage_ci_config(operator) } == expected_results + + +def test_check_leaks_in_changed_files_no_affected_files(tmp_path: Path) -> None: + create_files(tmp_path, bundle_files("hello", "0.0.1")) + repo = Repo(tmp_path) + operator = repo.operator("hello") + assert list(check_leaks_in_changed_files(operator)) == [] + + +def test_check_leaks_in_changed_files_filters_other_operators(tmp_path: Path) -> None: + create_files( + tmp_path, + bundle_files("hello", "0.0.1"), + bundle_files("other", "0.0.1"), + ) + repo = Repo(tmp_path) + operator = repo.operator("hello") + set_affected_operator_files( + [ + "operators/other/0.0.1/manifests/other.clusterserviceversion.yaml", + "catalogs/v4.15/hello/catalog.yaml", + ] + ) + assert list(check_leaks_in_changed_files(operator)) == [] + + +@patch("operatorcert.static_tests.common.operator.paths_with_leaks") +def test_check_leaks_in_changed_files_reports_leaks( + mock_paths_with_leaks: Any, tmp_path: Path +) -> None: + create_files(tmp_path, bundle_files("hello", "0.0.1")) + repo = Repo(tmp_path) + operator = repo.operator("hello") + leaky_rel = "operators/hello/0.0.1/manifests/hello.clusterserviceversion.yaml" + leaky_abs = (tmp_path / leaky_rel).resolve() + set_affected_operator_files([leaky_rel, "operators/hello/missing.yaml"]) + mock_paths_with_leaks.return_value = {leaky_abs} + + results = list(check_leaks_in_changed_files(operator)) + assert len(results) == 1 + assert isinstance(results[0], Fail) + assert leaky_rel in results[0].reason + assert "Potential secret leak detected" in results[0].reason + assert "SUPERSECRET" not in results[0].reason + mock_paths_with_leaks.assert_called_once() + scanned_paths = mock_paths_with_leaks.call_args[0] + assert leaky_abs in scanned_paths + + +@patch("operatorcert.static_tests.common.operator.paths_with_leaks") +def test_check_leaks_in_changed_files_outside_repo_path( + mock_paths_with_leaks: Any, tmp_path: Path +) -> None: + create_files(tmp_path, bundle_files("hello", "0.0.1")) + repo = Repo(tmp_path) + operator = repo.operator("hello") + leaky_rel = "operators/hello/0.0.1/manifests/hello.clusterserviceversion.yaml" + set_affected_operator_files([leaky_rel]) + outside = Path("/tmp/outside-repo-leak") + mock_paths_with_leaks.return_value = {outside} + + results = list(check_leaks_in_changed_files(operator)) + assert len(results) == 1 + assert str(outside) in results[0].reason + + +@patch("operatorcert.static_tests.common.operator.paths_with_leaks") +def test_check_leaks_in_changed_files_clean( + mock_paths_with_leaks: Any, tmp_path: Path +) -> None: + create_files(tmp_path, bundle_files("hello", "0.0.1")) + repo = Repo(tmp_path) + operator = repo.operator("hello") + leaky_rel = "operators/hello/0.0.1/manifests/hello.clusterserviceversion.yaml" + set_affected_operator_files([leaky_rel]) + mock_paths_with_leaks.return_value = set() + + assert list(check_leaks_in_changed_files(operator)) == [] + + +@patch("operatorcert.static_tests.common.operator.paths_with_leaks") +def test_check_leaks_in_changed_files_scan_error_is_generic( + mock_paths_with_leaks: Any, tmp_path: Path +) -> None: + create_files(tmp_path, bundle_files("hello", "0.0.1")) + repo = Repo(tmp_path) + operator = repo.operator("hello") + leaky_rel = "operators/hello/0.0.1/manifests/hello.clusterserviceversion.yaml" + set_affected_operator_files([leaky_rel]) + secret = "SUPERSECRET_TOKEN_VALUE" + mock_paths_with_leaks.side_effect = RuntimeError(f"scanner boom {secret}") + + results = list(check_leaks_in_changed_files(operator)) + assert len(results) == 1 + assert isinstance(results[0], Fail) + assert "internal error" in results[0].reason + assert secret not in results[0].reason diff --git a/tests/static_tests/test_helpers.py b/tests/static_tests/test_helpers.py index 9e01e31af..17949d5ec 100644 --- a/tests/static_tests/test_helpers.py +++ b/tests/static_tests/test_helpers.py @@ -2,7 +2,20 @@ from unittest.mock import MagicMock, call, patch from operatorcert.operator_repo import Bundle, Operator -from operatorcert.static_tests.helpers import skip_fbc +from operatorcert.static_tests.helpers import ( + get_affected_operator_files, + set_affected_operator_files, + skip_fbc, +) + + +def test_affected_operator_files_context() -> None: + assert get_affected_operator_files() == () + set_affected_operator_files(["operators/foo/ci.yaml", "operators/foo/bar.yaml"]) + assert get_affected_operator_files() == ( + "operators/foo/ci.yaml", + "operators/foo/bar.yaml", + ) @patch("operatorcert.static_tests.helpers.LOGGER") diff --git a/tests/test_redact.py b/tests/test_redact.py index 425107216..09de7b9ef 100644 --- a/tests/test_redact.py +++ b/tests/test_redact.py @@ -1,5 +1,6 @@ import json import os +import subprocess import tempfile from base64 import b64encode from pathlib import Path @@ -9,6 +10,7 @@ import pytest from operatorcert.redact import ( + paths_with_leaks, scan_and_redact, scan, ) @@ -19,8 +21,10 @@ REDACTED_CONTENT = b"curl -H 'Authorization: bearer ***[REDACTED]*** example.com\n" -def _scan_result_str(path: str) -> str: - return json.dumps({"results": [{"location": {"path": path}}]}) +def _scan_result_str(path: str, **extra: Any) -> str: + result: dict[str, Any] = {"location": {"path": path}} + result.update(extra) + return json.dumps({"results": [result]}) def _make_run_side_effect( @@ -45,7 +49,8 @@ def test_scan(mock_check_output: MagicMock) -> None: input_path = Path("/fake/path/testfile") scan_result = _scan_result_str(str(input_path)) - mock_check_output.return_value = scan_result + # Blank lines in JSONL output are skipped (leaktk may emit them). + mock_check_output.return_value = f"\n{scan_result}\n\n" results = scan(input_path) @@ -67,9 +72,67 @@ def test_scan(mock_check_output: MagicMock) -> None: ["leaktk", "listen"], input=expected_input, text=True, + stderr=subprocess.DEVNULL, ) +@patch("operatorcert.redact.subprocess.check_output") +def test_scan_ignores_secret_fields(mock_check_output: MagicMock) -> None: + """Extra LeakTK fields (e.g. secret/match) must not be retained on models.""" + input_path = Path("/fake/path/leaky-file") + secret = "SUPERSECRET_TOKEN_VALUE" + mock_check_output.return_value = _scan_result_str( + str(input_path), secret=secret, match=secret + ) + + results = scan(input_path) + + assert results[0].results[0].location.path == input_path + dumped = results[0].model_dump_json() + assert secret not in dumped + + +@patch("operatorcert.redact.subprocess.check_output") +def test_scan_called_process_error_is_sanitized(mock_check_output: MagicMock) -> None: + """Non-zero LeakTK exit must not propagate stdout that may contain secrets.""" + secret = "SUPERSECRET_TOKEN_VALUE" + mock_check_output.side_effect = subprocess.CalledProcessError( + 1, + ["leaktk", "listen"], + output=json.dumps({"secret": secret}), + ) + + with pytest.raises(RuntimeError, match="LeakTK scan failed") as exc_info: + scan(Path("/fake/path/file")) + + assert secret not in str(exc_info.value) + assert secret not in repr(exc_info.value) + + +@patch("operatorcert.redact.subprocess.check_output") +def test_scan_invalid_json_is_sanitized(mock_check_output: MagicMock) -> None: + """Unparseable LeakTK lines raise a generic error without raw payload.""" + secret = "SUPERSECRET_TOKEN_VALUE" + mock_check_output.return_value = f'{{"results": "not-a-list-{secret}"}}' + + with pytest.raises(RuntimeError, match="could not be parsed safely") as exc_info: + scan(Path("/fake/path/file")) + + assert secret not in str(exc_info.value) + + +@patch("operatorcert.redact.subprocess.check_output") +def test_paths_with_leaks(mock_check_output: MagicMock) -> None: + """paths_with_leaks returns absolute paths reported by the scan.""" + input_path = Path("/fake/path/leaky-file") + mock_check_output.return_value = _scan_result_str(str(input_path)) + + assert paths_with_leaks() == set() + + result = paths_with_leaks(input_path) + assert result == {input_path.absolute()} + + @patch("operatorcert.redact.subprocess.run") @patch("operatorcert.redact.subprocess.check_output") def test_scan_and_redact(mock_check_output: MagicMock, mock_run: MagicMock) -> None: