Add static check for NetworkPolicy presence

A new static check for all incoming bundles that rejects any bundle
that contains a NetworkPolicy.

Based on the decision from OLM team this resource is temporarily
not allowed globally for all OCP versions until OLM releases a backport.

JIRA: ISV-6226

Signed-off-by: Ales Raszka <araszka@redhat.com>

Author: Allda

docs/users/static_checks.md

@@ -173,6 +173,19 @@ To prevent the test from failing download the latest [Makefile](https://raw.gith
 and re-render the catalog again with `make catalogs` command. The Makefile uses
 extra arguments `--migrate-level bundle-object-to-csv-metadata` for opm when rendering
 catalogs for `>=4.17` version.
+
+#### check_network_policy_presence
+
+The test checks if there is a NetworkPolicy object defined in the bundle manifests.
+```yamlyaml
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+...
+```
+Network policies are not a supported resource that Operator Lifecycle Manager(OLM)
+can install and manage. Static checks will raise an error if such resource is
+found in the bundle manifests.
+
 ## Running tests locally
 
 ```bash

operator-pipeline-images/operatorcert/operator_repo/core.py

@@ -1,3 +1,4 @@
+# pylint: disable=too-many-lines
 """
 Definition of Repo, Operator, Bundle, Catalog and CatalogOperator classes
 """
@@ -7,7 +8,7 @@
 from dataclasses import dataclass
 from functools import cached_property, total_ordering
 from pathlib import Path
-from typing import Any, Optional, SupportsIndex, Union
+from typing import Any, Optional, SupportsIndex, Union, Generator
 
 from semantic_version import NpmSpec, Version
 
@@ -179,6 +180,36 @@ def csv_file_name(self) -> Path:
             f"CSV file for {self.operator_name}/{self.operator_version} not found"
         )
 
+    def manifest_files(self) -> list[Path]:
+        """
+        Get a list of all manifest files in the bundle.
+        The manifest files are considered to be all .yaml and .yml files within
+        the manifests/ directory of the bundle.
+
+        Returns:
+            list[Path]: List of paths to manifest files.
+        """
+        return list(self._manifests_path.glob("*.yaml")) + list(
+            self._manifests_path.glob("*.yml")
+        )
+
+    def manifest_files_content(
+        self,
+    ) -> Generator[tuple[Path, dict[str, Any]], None, None]:
+        """
+        Iterate over all manifest files in the bundle and yield their content.
+
+        Yields:
+            Generator[tuple[Path, dict[str, Any]], None, None]: A generator yielding tuples
+            with the file path and its parsed content as a dictionary.
+        """
+        for file in self.manifest_files():
+            content = load_yaml(file)
+            if not isinstance(content, dict):
+                log.warning("Manifest file %s does not contain a valid yaml document")
+                continue
+            yield file, content
+
     @property
     def channels(self) -> set[str]:
         """

operator-pipeline-images/operatorcert/static_tests/common/bundle.py

@@ -187,3 +187,25 @@ def check_validate_schema_bundle_release_config(
             "Bundle's 'release-config.yaml' contains invalid data "
             f"which does not comply with the schema: {ve.message}"
         )
+
+
+def check_network_policy_presence(bundle: Bundle) -> Iterator[CheckResult]:
+    """
+    Check if the bundle contains a network policy object within its manifests.
+    Network policies are not supported by OpenShift and their presence
+    in the bundle will lead to bundle rejection during the certification process.
+    """
+    for manifest_file, content in bundle.manifest_files_content():
+        kind = content.get("kind") or ""
+        api_version = content.get("apiVersion") or ""
+
+        if not (
+            kind == "NetworkPolicy" and api_version.startswith("networking.k8s.io/")
+        ):
+            continue
+
+        yield Fail(
+            f"Bundle contains a NetworkPolicy in {manifest_file.name}. "
+            "Network policies are not a supported resource that Operator "
+            "Lifecycle Manager(OLM) can install and manage."
+        )

operator-pipeline-images/tests/operator_repo/test_bundle.py

@@ -142,3 +142,38 @@ def test_bundle_invalid(tmp_path: Path) -> None:
 def test_bundle_caching(mock_bundle: Bundle) -> None:
     assert Bundle(mock_bundle.root).operator == mock_bundle.operator
     assert Bundle(mock_bundle.root).operator is not mock_bundle.operator
+
+
+def test_bundle_manifest_files(tmp_path: Path) -> None:
+    create_files(
+        tmp_path,
+        bundle_files("hello", "0.0.1"),
+        {
+            "operators/hello/0.0.1/manifests/sample_doc.yaml": "---",
+            "operators/hello/0.0.1/manifests/sample_doc_2.yml": "---\nfoo: bar",
+        },
+    )
+    repo = Repo(tmp_path)
+    operator = repo.operator("hello")
+    bundle = operator.bundle("0.0.1")
+
+    manifest_files = list(bundle.manifest_files())
+    assert len(manifest_files) == 3
+    assert sorted(manifest_files) == sorted(
+        [
+            bundle.root / "manifests" / "hello.clusterserviceversion.yaml",
+            bundle.root / "manifests" / "sample_doc.yaml",
+            bundle.root / "manifests" / "sample_doc_2.yml",
+        ]
+    )
+    for manifest_file, content in bundle.manifest_files_content():
+        assert manifest_file in manifest_files
+        assert isinstance(content, dict)
+        if manifest_file.name == "sample_doc.yaml":
+            assert content == {}
+        elif manifest_file.name == "sample_doc_2.yml":
+            assert content == {"foo": "bar"}
+        elif manifest_file.name == "hello.clusterserviceversion.yaml":
+            assert content["metadata"]["name"] == "hello.v0.0.1"
+        else:
+            pytest.fail(f"Unexpected manifest file {manifest_file}")

operator-pipeline-images/tests/static_tests/common/test_bundle.py

@@ -8,6 +8,7 @@
     check_bundle_release_config,
     check_operator_name,
     check_validate_schema_bundle_release_config,
+    check_network_policy_presence,
 )
 from tests.utils import bundle_files, create_files
 
@@ -656,3 +657,53 @@ def test_check_validate_schema_bundle_release_config(
         (x.__class__, x.reason)
         for x in check_validate_schema_bundle_release_config(bundle)
     } == expected_results
+
+
+@pytest.mark.parametrize(
+    "files, bundle_to_check, expected_results",
+    [
+        pytest.param(
+            [
+                bundle_files("hello", "0.0.1"),
+            ],
+            ("hello", "0.0.1"),
+            set(),
+            id="pass: network policy not present",
+        ),
+        pytest.param(
+            [
+                bundle_files("hello", "0.0.1"),
+                {
+                    "operators/hello/0.0.1/manifests/dummy_.k8s.io_v1_networkpolicy.yaml": {
+                        "kind": "NetworkPolicy",
+                        "apiVersion": "networking.k8s.io/v1",
+                    }
+                },
+            ],
+            ("hello", "0.0.1"),
+            {
+                (
+                    Fail,
+                    f"Bundle contains a NetworkPolicy in dummy_.k8s.io_v1_networkpolicy.yaml. "
+                    "Network policies are not a supported resource that Operator "
+                    "Lifecycle Manager(OLM) can install and manage.",
+                ),
+            },
+            id="fail: A bundle contains a NetworkPolicy",
+        ),
+    ],
+)
+def test_check_network_policy_presence(
+    tmp_path: Path,
+    files: list[dict[str, Any]],
+    bundle_to_check: tuple[str, str],
+    expected_results: set[tuple[type, str]],
+) -> None:
+    create_files(tmp_path, *files)
+    repo = Repo(tmp_path)
+    operator_name, bundle_version = bundle_to_check
+    operator = repo.operator(operator_name)
+    bundle = operator.bundle(bundle_version)
+    assert {
+        (x.__class__, x.reason) for x in check_network_policy_presence(bundle)
+    } == expected_results