Merge pull request #3 from alimasri/unused-secrets-validator

Add validators for undefined and unused secrets

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "cloudbuild-validator"
-version = "0.1.0"
+version = "0.1.1"
 description = "A robust and extensible tool for validating Google Cloud Build YAML configuration files against schema specifications and custom rules."
 readme = "README.md"
 authors = [

src/cloudbuild_validator/main.py

@@ -7,16 +7,15 @@
 
 
 def main(schema: Path, content: Path):
-    logger.info("Program started")
-
-    logger.info(f"Validating {content} against {schema}...")
+    log_prefix = f"[{content.name}]"
+    logger.info(f"{log_prefix} validating...")
     validator = CloudBuildValidator(schema)
     errors = validator.validate(content)
     if not errors:
-        logger.info("Validation passed")
+        logger.info(f"{log_prefix} passed")
         raise SystemExit(0)
 
-    logger.error("Validation failed")
+    logger.error(f"{log_prefix} failed")
     for error_msg in errors:
         logger.error(f"\t{error_msg}")
 
@@ -26,6 +25,11 @@ def main(schema: Path, content: Path):
 def run():
     default_schema = Path(__file__).parent / "data" / "cloudbuild-specifications.yaml"
     parser = ArgumentParser()
+    parser.add_argument(
+        "file",
+        type=Path,
+        help="Path to the content file to validate",
+    )
     parser.add_argument(
         "-s",
         "--schema",
@@ -34,13 +38,6 @@ def run():
         required=False,
         default=default_schema,
     )
-    parser.add_argument(
-        "-f",
-        "--file",
-        type=Path,
-        help="Path to the content file to validate",
-        required=True,
-    )
     args = parser.parse_args()
     main(args.schema, args.file)
 

src/cloudbuild_validator/validators.py

@@ -8,8 +8,7 @@
 
 class Validator(ABC):
     @abstractmethod
-    def validate(self, yaml_file_content: str) -> List[str]:
-        ...
+    def validate(self, yaml_file_content: str) -> List[str]: ...
 
 
 class CloudBuildValidationError(Exception):
@@ -73,3 +72,55 @@ def validate(self, content: dict) -> None:
                         raise CloudBuildValidationError(
                             f"Undefined substitution variable {variable} in step `{step['id']}`"
                         )
+
+
+class UndefinedSecretsValidator(Validator):
+    """Check for undefined secrets in the content"""
+
+    def validate(self, content: dict) -> None:
+        secrets = content.get("availableSecrets", [])
+        if not secrets:
+            all_secrets = set()
+        else:
+            all_secrets = {
+                secret.get("env")
+                for secret_store in secrets.values()
+                for secret in secret_store
+            }
+
+        errors = []
+        for step in content["steps"]:
+            step_secrets = step.get("secretEnv", [])
+            if not step_secrets:
+                continue
+            for secret in step_secrets:
+                if secret not in all_secrets:
+                    errors.append(f"Undefined secret `{secret}` in step `{step['id']}`")
+        if errors:
+            raise CloudBuildValidationError("\n".join(errors))
+
+
+class UnusedSecretsValidator(Validator):
+    """Check for unused secrets in the content"""
+
+    def validate(self, content: dict) -> None:
+        secrets = content.get("availableSecrets", [])
+        if not secrets:
+            return
+        all_secrets = {
+            secret.get("env")
+            for secret_store in secrets.values()
+            for secret in secret_store
+        }
+
+        for step in content["steps"]:
+            step_secrets = step.get("secretEnv", [])
+            if not step_secrets:
+                continue
+            for secret in step_secrets:
+                all_secrets.discard(secret)
+
+        if all_secrets:
+            raise CloudBuildValidationError(
+                f"Secret(s) {all_secrets} are defined but not used"
+            )

tests/test_default_validators.py

@@ -1,4 +1,5 @@
 import pytest
+
 from cloudbuild_validator import validators
 from cloudbuild_validator.validators import CloudBuildValidationError
 
@@ -103,3 +104,54 @@ def test_invalid_substitution_variable_names():
     }
     with pytest.raises(CloudBuildValidationError):
         validators.SubstitutionVariablesValidator().validate(content)
+
+
+def test_undefined_secret_undefined():
+    content = {
+        "steps": [
+            {"id": "step1"},
+            {"id": "step2", "secretEnv": ["SECRET"]},
+        ],
+    }
+    with pytest.raises(CloudBuildValidationError):
+        validators.UndefinedSecretsValidator().validate(content)
+
+
+def test_unsed_secret_defined():
+    content = {
+        "steps": [
+            {"id": "step1", "secretEnv": ["SECRET"]},
+            {"id": "step2"},
+        ],
+        "availableSecrets": {
+            "secretManager": [{"env": "SECRET"}],
+        },
+    }
+    validators.UndefinedSecretsValidator().validate(content)
+
+
+def test_unsed_secret_unused():
+    content = {
+        "steps": [
+            {"id": "step1"},
+            {"id": "step2"},
+        ],
+        "availableSecrets": {
+            "secretManager": [{"env": "SECRET"}],
+        },
+    }
+    with pytest.raises(CloudBuildValidationError):
+        validators.UnusedSecretsValidator().validate(content)
+
+
+def test_unsed_secret_used():
+    content = {
+        "steps": [
+            {"id": "step1", "secretEnv": ["SECRET"]},
+            {"id": "step2"},
+        ],
+        "availableSecrets": {
+            "secretManager": [{"env": "SECRET"}],
+        },
+    }
+    validators.UnusedSecretsValidator().validate(content)

tests/test_main.py

@@ -2,6 +2,7 @@
 from typing import List
 
 import pytest
+
 from cloudbuild_validator import validators
 from cloudbuild_validator.core import CloudBuildValidator
 from cloudbuild_validator.validators import Validator