Checkov 新增 Argo Workflows 策略开发全指南从 Check 编写、测试到运行原理【免费下载链接】checkovPrevent cloud misconfigurations and find vulnerabilities during build-time in infrastructure as code, container images and open source packages with Checkov by Bridgecrew.项目地址: https://gitcode.com/GitHub_Trending/ch/checkov本文以 Checkov 的 Argo Workflows 框架支持为背景完整讲解如何为一个自定义的 Argo Workflows 配置编写新的策略Check包括策略类设计、示例文件与单元测试的编写以及策略注册、扫描执行与资源标识的底层原理。读者读完本文后将掌握在 Checkov 中新增一个 Argo Workflows 配置检查项如校验 Workflow 是否使用默认 ServiceAccount的完整闭环流程并能理解新策略从编写到合并后自动生效的机制。一、Argo Workflows 策略在 Checkov 中的定位Argo Workflows 是 Kubernetes 生态中常用的工作流引擎其工作流定义以apiVersion: argoproj.io/v1alpha1kind: Workflow的 YAML 清单呈现。Checkov 通过checkov/argo_workflows模块为其提供了独立的策略扫描能力对应的检查 ID 前缀为CKV_ARGO_。从仓库结构看Argo Workflows 支持主要包含四个组成部分checkov/argo_workflows/runner.py扫描入口负责识别 Argo Workflows 文件并驱动扫描checkov/argo_workflows/checks/base_argo_workflows_check.py所有 Argo 策略必须继承的基类checkov/argo_workflows/checks/registry.py策略注册表在模块导入时收集所有 Check 实例checkov/argo_workflows/checks/template/存放具体策略实现的目录目前已有DefaultServiceAccount.pyCKV_ARGO_1与RunAsNonRoot.pyCKV_ARGO_2两个内置策略。新策略的开发流程可以概括为三步编写 Check 类 → 添加示例 YAML 与单元测试 → 运行测试验证。下面以校验 Workflow Pod 未使用默认 ServiceAccount这一策略为例逐步展开。二、编写第一个策略DefaultServiceAccountCKV_ARGO_12.1 创建 Check 文件在checkov/argo_workflows/checks/template目录下新建DefaultServiceAccount.py完整代码如下与仓库中实际实现的 DefaultServiceAccount.py 一致from __future__ import annotations from typing import Any from checkov.argo_workflows.checks.base_argo_workflows_check import BaseArgoWorkflowsCheck from checkov.common.models.enums import CheckResult, CheckCategories from checkov.yaml_doc.enums import BlockType class DefaultServiceAccount(BaseArgoWorkflowsCheck): def __init__(self) - None: name Ensure Workflow pods are not using the default ServiceAccount id CKV_ARGO_1 super().__init__( namename, idid, categories(CheckCategories.IAM,), supported_entities(spec,), block_typeBlockType.OBJECT, ) def scan_conf(self, conf: dict[str, Any]) - tuple[CheckResult, dict[str, Any]]: if serviceAccountName in conf.keys() and conf[serviceAccountName] ! default: return CheckResult.PASSED, conf return CheckResult.FAILED, conf check DefaultServiceAccount()2.2 逐行解析 Check 类构造函数中的五个关键参数参数取值示例含义nameEnsure Workflow pods are not using the default ServiceAccount策略的人类可读描述会出现在扫描报告中idCKV_ARGO_1策略唯一标识遵循CKV_ARGO_N命名规则用于命令行过滤与抑制categories(CheckCategories.IAM,)策略所属类别此处归类为 IAM身份与访问管理supported_entities(spec,)声明该策略作用于 Workflow YAML 中的哪个实体即工作流的spec段block_typeBlockType.OBJECT声明实体的 YAML 块类型OBJECT表示以键值对象方式扫描另一个可选项是BlockType.ARRAY数组块从 base_argo_workflows_check.py 的源码可以看到基类BaseArgoWorkflowsCheck继承自 Checkov 通用基类BaseCheck它除了透传上述参数外还会在初始化时执行registry.register(self)即策略在实例化的一瞬间即被自动注册进全局注册表无需手动添加注册代码。scan_conf判定逻辑def scan_conf(self, conf: dict[str, Any]) - tuple[CheckResult, dict[str, Any]]: if serviceAccountName in conf.keys() and conf[serviceAccountName] ! default: return CheckResult.PASSED, conf return CheckResult.FAILED, confconf是当前被扫描实体即spec块解析后的字典判定规则若spec中显式声明了serviceAccountName且其值不为default则返回CheckResult.PASSED否则未声明或值为default返回CheckResult.FAILED返回值是(CheckResult, dict)二元组第二个元素为关联的配置对象会被记录进扫描结果供报告展示。模块末尾的实例化语句check DefaultServiceAccount()这是 Checkov YAML/策略框架的惯用写法——策略文件即模块导入模块即触发注册check实例同时可供测试文件直接引用。三、为策略添加示例与测试3.1 组织示例目录在tests/argo_workflows/checks/template下新建与策略同名的目录example_DefaultServiceAccount用于存放示例配置。规范要求至少提供两个用例一个通过、一个失败。仓库中该目录实际包含三个文件见 example_DefaultServiceAccount/pass.yaml、fail_default.yaml、fail_none.yaml。通过用例pass.yamlapiVersion: argoproj.io/v1alpha1 kind: Workflow metadata: generateName: hello-world- spec: entrypoint: whalesay serviceAccountName: custom-sa templates: - name: whalesay container: image: docker/whalesay:latest command: [cowsay] args: [hello world]该用例在spec中声明了自定义 ServiceAccountcustom-sa满足策略要求应判定通过。失败用例fail_default.yaml与通过用例的唯一差别是serviceAccountName: default显式使用默认 ServiceAccount应判定失败。失败用例fail_none.yamlspec中完全未声明serviceAccountName字段按策略定义同样视为违规应判定失败。这一用例恰好覆盖了字段缺失的边界场景值得保留。3.2 编写测试文件在tests/argo_workflows/checks/template/下创建test_DefaultServiceAccount.py与仓库中的 test_DefaultServiceAccount.py 一致from pathlib import Path from checkov.argo_workflows.runner import Runner from checkov.argo_workflows.checks.template.DefaultServiceAccount import check from checkov.runner_filter import RunnerFilter def test_examples(): # given test_files_dir Path(__file__).parent / example_DefaultServiceAccount # when report Runner().run(root_folderstr(test_files_dir), runner_filterRunnerFilter(checks[check.id])) # then summary report.get_summary() passing_resources { f{test_files_dir}/pass.yaml.spec.spec.CKV_ARGO_1[6:14], } failing_resources { f{test_files_dir}/fail_default.yaml.spec.spec.CKV_ARGO_1[6:14], f{test_files_dir}/fail_none.yaml.spec.spec.CKV_ARGO_1[6:13], } passed_check_resources {c.resource for c in report.passed_checks} failed_check_resources {c.resource for c in report.failed_checks} assert summary[passed] len(passing_resources) assert summary[failed] len(failing_resources) assert summary[skipped] 0 assert summary[parsing_errors] 0 assert passed_check_resources passing_resources assert failed_check_resources failing_resources测试要点拆解仅运行目标策略RunnerFilter(checks[check.id])让扫描只执行CKV_ARGO_1避免其他策略干扰断言报告摘要断言summary[passed]、summary[failed]、summary[skipped]、summary[parsing_errors]四个维度全部校验确保 1 个通过、2 个失败、0 跳过、0 解析错误资源级精确断言将report.passed_checks/report.failed_checks中的resource字段与期望集合比对精确到每个文件、每个实体。关于资源标识符pass.yaml.spec.spec.CKV_ARGO_1[6:14]的解读该标识符格式为文件路径.实体类型.实体名.检查ID[起始行:结束行]。从 YAML 注册表基类 checkov/yaml_doc/base_registry.py 的get_result_key源码可知行号区间来自解析器注入的__startline__与__endline__标记[6:14]表示pass.yaml中spec实体占据第 6 至 14 行fail_none.yaml因缺少serviceAccountName字段spec块结束于第 13 行故区间为[6:13]。测试正是通过这一文件 实体 策略 行号区间的四元组完成确定性匹配。四、理解策略的注册与扫描执行链路4.1 注册机制checkov/argo_workflows/checks/registry.py 内容极简本质上是 YAML 注册表的一个实例from checkov.common.bridgecrew.check_type import CheckType from checkov.yaml_doc.base_registry import Registry registry Registry(CheckType.ARGO_WORKFLOWS)每个BaseArgoWorkflowsCheck子类在实例化时调用registry.register(self)将自身按supported_entities归入注册表的checks字典。同时模块级check DefaultServiceAccount()保证了导入策略模块即完成注册。4.2 Runner 的文件识别与扫描入口checkov/argo_workflows/runner.py 中定义了两个关键正则API_VERSION_PATTERN re.compile(r^apiVersion:\s*argoproj.io/, re.MULTILINE) KIND_PATTERN re.compile(r^kind:\s*Workflow, re.MULTILINE)Runner._get_workflow_file_contentrunner.py负责文件级筛选只有扩展名为.yaml/.yml、内容包含argoproj.io、且同时匹配上述两个正则的文件才会被识别为 Argo Workflows 清单并进入解析流程。这意味着其他 Kubernetes 资源如 Deployment即使放在同目录也不会被误扫Argo CD 的Application清单kind: Application会被自动排除——这一点在 tests/argo_workflows/test_runner.py 的test_runner_ignore_argo_cd用例中有明确验证。Runner 通过block_type_registries {template: template_registry}将注册表挂载到 YAML 解析的template块类型上import_registry返回该注册表供外部检查external checks使用。4.3 YAML 块扫描OBJECT 与 ARRAY在 base_registry.py 中注册表维护了块类型到扫描函数的映射self._scanner: dict[str, _ScannerCallableAlias] { BlockType.ARRAY: self._scan_yaml_array, BlockType.OBJECT: self._scan_yaml_object, }BlockType.OBJECT策略如本文的 CKV_ARGO_1通过_scan_yaml_object定位entity_name in entity将entity[entity_name]即spec字典交给策略的scan_confBlockType.ARRAY策略则通过 jmespath 表达式在实体中搜索数组结构后逐项扫描适用于templates这类列表实体。五、第二个策略示例RunAsNonRootCKV_ARGO_2仓库中还内置了第二个 Argo Workflows 策略 RunAsNonRoot.py校验 Workflow Pod 是否以非 root 用户运行可作为编写类似策略的参考class RunAsNonRoot(BaseArgoWorkflowsCheck): def __init__(self) - None: name Ensure Workflow pods are running as non-root user id CKV_ARGO_2 super().__init__( namename, idid, categories(CheckCategories.IAM,), supported_entities(spec,), block_typeBlockType.OBJECT, ) def scan_conf(self, conf: dict[str, Any]) - tuple[CheckResult, dict[str, Any]]: security_context conf.get(securityContext) if isinstance(security_context, dict) and security_context.get(runAsNonRoot) is True: return CheckResult.PASSED, conf return CheckResult.FAILED, conf该示例展示了嵌套结构的判定技巧先通过conf.get(securityContext)取出嵌套的安全上下文字典再用isinstance做防御性类型检查最后判定runAsNonRoot is True。其配套测试见 example_RunAsNonRoot/ 与 test_RunAsNonRoot.py测试模式与 CKV_ARGO_1 完全一致。两个内置策略在 docs/5.Policy Index/argo_workflows.md 中有自动生成的索引记录该索引文档由checkov/docs_generator.py生成新增策略合并后索引会自动更新。六、验证与运行测试完成 Check 与测试编写后在仓库根目录运行 pytest 即可验证# 运行单个策略的测试 pytest tests/argo_workflows/checks/template/test_DefaultServiceAccount.py -v # 运行整个 Argo Workflows 模块的测试 pytest tests/argo_workflows/ -vtests/argo_workflows/test_runner.py还提供了 Runner 层面的集成测试覆盖了注册表报告类型CheckType.ARGO_WORKFLOWS、强制规则enforcement rules生效、单策略过滤运行RunnerFilter(checks[CKV_ARGO_1])预期 1 个失败、checks[CKV_ARGO_2]预期 1 个通过以及 Argo CD 文件被忽略等场景。这些用例同样值得在新增策略时参照补充。七、对完整清单的进一步验证在实际使用中新策略合入后可以通过 CLI 直接对示例目录发起扫描验证策略生效# 仅运行 CKV_ARGO_1 扫描示例目录 checkov -d tests/argo_workflows/checks/template/example_DefaultServiceAccount --check CKV_ARGO_1 # 或对该测试示例目录执行全部 Argo Workflows 策略 checkov -d tests/argo_workflows/checks/template/example_DefaultServiceAccount --framework argo_workflows需要说明的是checkov命令会扫描整个目录包括内置策略而单元测试通过RunnerFilter(checks[check.id])精确限定范围二者互补——前者验证真实 CLI 体验后者保证测试的确定性。另外Runner 还实现了镜像引用Image Referencer能力runner.py 中的get_images会提取templates下每个模板的container与script块中的image字段用于后续的镜像漏洞扫描如 tests/argo_workflows/test_runner.py 验证的alpine:latest、python:alpine3.6镜像提取。这意味着新增策略若涉及容器镜像字段可以参考该实现理解镜像数据的获取方式。结语至此一个完整的 Argo Workflows 策略贡献流程已经走通创建DefaultServiceAccount.py策略类 → 在基类中自动注册 → 编写pass.yaml与失败用例 → 用RunnerFilter限定范围运行 Runner 并断言摘要与资源标识符 → 合并后新策略随每次扫描自动生效。掌握了BaseArgoWorkflowsCheck、supported_entities、BlockType与资源标识符文件.实体类型.实体名.检查ID[行区间]的语义后你可以按相同模式为 Argo Workflows 清单添加任意配置校验策略为 CI 流水线中的工作流安全把关。【免费下载链接】checkovPrevent cloud misconfigurations and find vulnerabilities during build-time in infrastructure as code, container images and open source packages with Checkov by Bridgecrew.项目地址: https://gitcode.com/GitHub_Trending/ch/checkov创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考