feat(general): add Azure Pipelines framework

.github/pr-title-checker-config.json

@@ -7,11 +7,11 @@
     "prefixes": [
       "chore: "
     ],
-    "regexp": "^(fix|feat|break|docs|chore|platform)\\((argo|arm|bicep|bitbucket|circleci|cloudformation|dockerfile|github|gha|gitlab|helm|kubernetes|kustomize|openapi|sca|secrets|serverless|terraform|general|graph)\\): "
+    "regexp": "^(fix|feat|break|docs|chore|platform)\\((argo|arm|azure|bicep|bitbucket|circleci|cloudformation|dockerfile|github|gha|gitlab|helm|kubernetes|kustomize|openapi|sca|secrets|serverless|terraform|general|graph)\\): "
   },
   "MESSAGES": {
     "success": "PR title is valid",
     "failure": "PR title is invalid",
-    "notice": "Title needs to pass regex '(fix|feat|break|docs|chore|platform)\\((argo|arm|bicep|bitbucket|circleci|cloudformation|dockerfile|github|gha|gitlab|helm|kubernetes|kustomize|openapi|sca|secrets|serverless|terraform|general|graph)\\): '"
+    "notice": "Title needs to pass regex '(fix|feat|break|docs|chore|platform)\\((argo|arm|azure|bicep|bitbucket|circleci|cloudformation|dockerfile|github|gha|gitlab|helm|kubernetes|kustomize|openapi|sca|secrets|serverless|terraform|general|graph)\\): '"
   }
 }

.github/workflows/build.yml

@@ -155,7 +155,7 @@ jobs:
           export PYTHONPATH='.'
           git pull
 
-          for i in cloudformation terraform kubernetes serverless arm dockerfile secrets github_configuration gitlab_configuration bitbucket_configuration github_actions gitlab_ci bicep openapi bitbucket_pipelines argo_workflows circleci_pipelines all
+          for i in cloudformation terraform kubernetes serverless arm dockerfile secrets github_configuration gitlab_configuration bitbucket_configuration github_actions gitlab_ci bicep openapi bitbucket_pipelines argo_workflows circleci_pipelines azure_pipelines all
           do
             export scansdoc="docs/5.Policy Index/$i.md"
             echo "---" > "$scansdoc"

README.md

@@ -42,7 +42,7 @@ Checkov also powers [**Bridgecrew**](https://bridgecrew.io/?utm_source=github&ut
 
  * [Over 1000 built-in policies](docs/5.Policy%20Index/all.md) cover security and compliance best practices for AWS, Azure and Google Cloud.
  * Scans Terraform, Terraform Plan, CloudFormation, AWS SAM, Kubernetes, Dockerfile, Serverless framework, Bicep and ARM template files.
- * Scans Argo Workflows, BitBucket Pipelines, GitHub Actions and GitLab CI workflow files
+ * Scans Argo Workflows, Azure Pipelines, BitBucket Pipelines, Circle CI Pipelines, GitHub Actions and GitLab CI workflow files
  * Supports Context-awareness policies based on in-memory graph-based scanning.
  * Supports Python format for attribute policies and YAML format for both attribute and composite policies.
  * Detects [AWS credentials](docs/2.Basics/Scanning%20Credentials%20and%20Secrets.md) in EC2 Userdata, Lambda environment variables and Terraform providers.

checkov/azure_pipelines/__init__.py

@@ -0,0 +1 @@
+from checkov.azure_pipelines.checks import *  # noqa

checkov/azure_pipelines/checks/__init__.py

@@ -0,0 +1 @@
+from checkov.azure_pipelines.checks.job import *  # noqa

checkov/azure_pipelines/checks/base_azure_pipelines_check.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from abc import abstractmethod
+from collections.abc import Iterable
+from typing import TYPE_CHECKING, Any
+
+from checkov.common.checks.base_check import BaseCheck
+from checkov.azure_pipelines.checks.registry import registry
+
+if TYPE_CHECKING:
+    from checkov.common.models.enums import CheckCategories, CheckResult
+
+
+class BaseAzurePipelinesCheck(BaseCheck):
+    def __init__(
+        self,
+        name: str,
+        id: str,
+        categories: Iterable[CheckCategories],
+        supported_entities: Iterable[str],
+        block_type: str,
+        path: str | None = None,
+    ) -> None:
+        super().__init__(
+            name=name,
+            id=id,
+            categories=categories,
+            supported_entities=supported_entities,
+            block_type=block_type,
+        )
+        self.path = path
+        registry.register(self)
+
+    def scan_entity_conf(self, conf: dict[str, Any], entity_type: str) -> tuple[CheckResult, dict[str, Any]]:  # type:ignore[override]  # multi_signature decorator is problematic
+        self.entity_type = entity_type
+
+        return self.scan_conf(conf)
+
+    @abstractmethod
+    def scan_conf(self, conf: dict[str, Any]) -> tuple[CheckResult, dict[str, Any]]:
+        pass

checkov/azure_pipelines/checks/job/ContainerDigest.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from typing import Any
+
+from checkov.azure_pipelines.checks.base_azure_pipelines_check import BaseAzurePipelinesCheck
+from checkov.common.models.enums import CheckResult, CheckCategories
+from checkov.yaml_doc.enums import BlockType
+
+
+class ContainerDigest(BaseAzurePipelinesCheck):
+    def __init__(self) -> None:
+        name = "Ensure container job uses a version digest"
+        id = "CKV_AZUREPIPELINES_2"
+        super().__init__(
+            name=name,
+            id=id,
+            categories=(CheckCategories.SUPPLY_CHAIN,),
+            supported_entities=("jobs", "stages[].jobs[]"),
+            block_type=BlockType.ARRAY,
+        )
+
+    def scan_conf(self, conf: dict[str, Any]) -> tuple[CheckResult, dict[str, Any]]:
+        container = conf.get("container")
+        if container and isinstance(container, str):
+            if "@" in container:
+                return CheckResult.PASSED, conf
+
+            return CheckResult.FAILED, conf
+
+        return CheckResult.UNKNOWN, conf
+
+
+check = ContainerDigest()

checkov/azure_pipelines/checks/job/ContainerLatestTag.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from typing import Any
+
+from checkov.azure_pipelines.checks.base_azure_pipelines_check import BaseAzurePipelinesCheck
+from checkov.common.models.enums import CheckResult, CheckCategories
+from checkov.yaml_doc.enums import BlockType
+
+
+class ContainerLatestTag(BaseAzurePipelinesCheck):
+    def __init__(self) -> None:
+        name = "Ensure container job uses a non latest version tag"
+        id = "CKV_AZUREPIPELINES_1"
+        super().__init__(
+            name=name,
+            id=id,
+            categories=(CheckCategories.SUPPLY_CHAIN,),
+            supported_entities=("jobs", "stages[].jobs[]"),
+            block_type=BlockType.ARRAY,
+        )
+
+    def scan_conf(self, conf: dict[str, Any]) -> tuple[CheckResult, dict[str, Any]]:
+        container = conf.get("container")
+        if container and isinstance(container, str):
+            if ":" in container:
+                # some image tag
+                if container.split(":")[1] == "latest":
+                    # latest image tag
+                    return CheckResult.FAILED, conf
+            elif "@" not in container:
+                # no image tag
+                return CheckResult.FAILED, conf
+
+            # image tag is either not latest or a digest
+            return CheckResult.PASSED, conf
+
+        return CheckResult.UNKNOWN, conf
+
+
+check = ContainerLatestTag()

checkov/azure_pipelines/checks/job/SetSecretVariable.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from typing import Any
+
+from checkov.azure_pipelines.checks.base_azure_pipelines_check import BaseAzurePipelinesCheck
+from checkov.common.models.enums import CheckResult, CheckCategories
+from checkov.yaml_doc.enums import BlockType
+
+
+class SetSecretVariable(BaseAzurePipelinesCheck):
+    def __init__(self) -> None:
+        name = "Ensure set variable is not marked as a secret"
+        id = "CKV_AZUREPIPELINES_3"
+        super().__init__(
+            name=name,
+            id=id,
+            categories=(CheckCategories.SUPPLY_CHAIN,),
+            supported_entities=("jobs[].steps[]", "stages[].jobs[].steps[]"),
+            block_type=BlockType.ARRAY,
+        )
+
+    def scan_conf(self, conf: dict[str, Any]) -> tuple[CheckResult, dict[str, Any]]:
+        run_cmd = conf.get("bash") or conf.get("powershell")
+        if run_cmd and isinstance(run_cmd, str):
+            variable_found = False
+
+            for line in run_cmd.splitlines():
+                if "task.setvariable" in line:
+                    variable_found = True
+
+                    if "issecret=true" in line:
+                        return CheckResult.FAILED, conf
+
+            if variable_found:
+                # should only pass, if it really found a set variable, otherwise unknown
+                return CheckResult.PASSED, conf
+
+        return CheckResult.UNKNOWN, conf
+
+
+check = SetSecretVariable()

checkov/azure_pipelines/checks/job/__init__.py

@@ -0,0 +1,4 @@
+from pathlib import Path
+
+modules = Path(__file__).parent.glob("*.py")
+__all__ = [f.stem for f in modules if f.is_file() and not f.stem == "__init__"]

checkov/azure_pipelines/checks/registry.py

@@ -0,0 +1,4 @@
+from checkov.common.bridgecrew.check_type import CheckType
+from checkov.yaml_doc.base_registry import Registry
+
+registry = Registry(CheckType.AZURE_PIPELINES)

checkov/azure_pipelines/runner.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from checkov.azure_pipelines.checks.registry import registry
+from checkov.common.output.report import CheckType
+from checkov.yaml_doc.runner import Runner as YamlRunner
+
+
+if TYPE_CHECKING:
+    from checkov.common.checks.base_check_registry import BaseCheckRegistry
+
+
+class Runner(YamlRunner):
+    check_type = CheckType.AZURE_PIPELINES  # noqa: CCE003  # a static attribute
+
+    def require_external_checks(self) -> bool:
+        return False
+
+    def import_registry(self) -> BaseCheckRegistry:
+        return registry
+
+    def _parse_file(
+        self, f: str, file_content: str | None = None
+    ) -> tuple[dict[str, Any] | list[dict[str, Any]], list[tuple[int, str]]] | None:
+        if self.is_workflow_file(f):
+            return super()._parse_file(f=f)
+
+        return None
+
+    def is_workflow_file(self, file_path: str) -> bool:
+        return file_path.endswith("azure-pipelines.yml")

checkov/common/bridgecrew/check_type.py

@@ -7,6 +7,7 @@ class CheckType:
     CIRCLECI_PIPELINES = "circleci_pipelines"
     ARGO_WORKFLOWS = "argo_workflows"
     ARM = "arm"
+    AZURE_PIPELINES = "azure_pipelines"
     BICEP = "bicep"
     CLOUDFORMATION = "cloudformation"
     DOCKERFILE = "dockerfile"

checkov/common/bridgecrew/code_categories.py

@@ -17,6 +17,7 @@ class CodeCategoryType:
     CheckType.BITBUCKET_PIPELINES: CodeCategoryType.SUPPLY_CHAIN,
     CheckType.CIRCLECI_PIPELINES: CodeCategoryType.SUPPLY_CHAIN,
     CheckType.ARM: CodeCategoryType.IAC,
+    CheckType.AZURE_PIPELINES: CodeCategoryType.SUPPLY_CHAIN,
     CheckType.BICEP: CodeCategoryType.IAC,
     CheckType.CLOUDFORMATION: CodeCategoryType.IAC,
     CheckType.DOCKERFILE: CodeCategoryType.IAC,

checkov/common/util/docs_generator.py

@@ -9,6 +9,7 @@
 
 from checkov.argo_workflows.checks.registry import registry as argo_workflows_registry
 from checkov.arm.registry import arm_resource_registry, arm_parameter_registry
+from checkov.azure_pipelines.checks.registry import registry as azure_pipelines_registry
 from checkov.bicep.checks.param.registry import registry as bicep_param_registry
 from checkov.bicep.checks.resource.registry import registry as bicep_resource_registry
 from checkov.bitbucket.registry import registry as bitbucket_configuration_registry
@@ -123,6 +124,8 @@ def add_from_repository(registry: Union[BaseCheckRegistry, BaseGraphRegistry], c
         add_from_repository(circleci_pipelines_registry, "circleci_pipelines", "circleci_pipelines")
     if any(x in framework_list for x in ("all", "argo_workflows")):
         add_from_repository(argo_workflows_registry, "argo_workflows", "Argo Workflows")
+    if any(x in framework_list for x in ("all", "azure_pipelines")):
+        add_from_repository(azure_pipelines_registry, "azure_pipelines", "Azure Pipelines")
     if any(x in framework_list for x in ("all", "arm")):
         add_from_repository(arm_resource_registry, "resource", "arm")
         add_from_repository(arm_parameter_registry, "parameter", "arm")

checkov/main.py

@@ -19,6 +19,7 @@
 
 from checkov.argo_workflows.runner import Runner as argo_workflows_runner
 from checkov.arm.runner import Runner as arm_runner
+from checkov.azure_pipelines.runner import Runner as azure_pipelines_runner
 from checkov.bitbucket.runner import Runner as bitbucket_configuration_runner
 from checkov.bitbucket_pipelines.runner import Runner as bitbucket_pipelines_runner
 from checkov.cloudformation.runner import Runner as cfn_runner
@@ -99,6 +100,7 @@
     sca_image_runner(),
     argo_workflows_runner(),
     circleci_pipelines_runner(),
+    azure_pipelines_runner(),
 )
 
 

checkov/yaml_doc/base_registry.py

@@ -157,10 +157,10 @@ def scan(  # type:ignore[override]  # return type is different than the base cla
             )
 
         if self.wildcard_checks:
-            for wildcard_pattern in self.wildcard_checks:
+            for wildcard_pattern, checks in self.wildcard_checks.items():
                 self._scan_yaml(
                     scanned_file=scanned_file,
-                    checks=self.wildcard_checks[wildcard_pattern],
+                    checks=checks,
                     skipped_checks=skipped_checks,
                     runner_filter=runner_filter,
                     entity=entity,