feat(ISV-7642): static check for leaks in operator files

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:

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

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)"
 

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/<operator>/` 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

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

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:

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