Write Custom Checkov Policies in Python

Checkov ships hundreds of built-in checks, but your organisation has rules no vendor knows about. This guide, part of security and compliance basics under Python IaC fundamentals and strategy, writes a custom Checkov policy in Python and runs it against your infrastructure alongside the built-ins covered in scanning Python IaC with Checkov.

Context

A built-in scan enforces community baselines, but internal rules — an approved AMI list, a mandatory logging bucket, a naming convention — need custom checks. Checkov lets you write these as Python classes that plug into the same scan, so one command enforces both community and in-house policy.

The mechanism is deliberately simple. Checkov keeps a global registry keyed by resource type; instantiating a check subclass registers it as a side effect of __init__. When you pass --external-checks-dir, Checkov imports every .py file in that directory before the scan starts, your module-level check = S3AccessLogging() line runs, and from that point the check is indistinguishable from a built-in. There is no manifest to maintain and no plugin interface to version — but it also means a check that is never instantiated is silently never run, which is the first thing to verify when a new policy appears to do nothing.

Checkov also supports custom policies written in YAML, and it is worth knowing when each is appropriate. A YAML policy expresses attribute conditions declaratively and, importantly, can express graph conditions — "every aws_s3_bucket must be connected to an aws_s3_bucket_logging" — which a per-resource Python check cannot see, because scan_resource_conf receives one resource block at a time with no view of the rest of the file. Python is the right choice when the rule needs real logic: parsing a CIDR, comparing a date, calling a regular expression across a naming convention, or reading an allow-list from a file. Anything expressible as "attribute X equals Y" is cheaper in YAML.

Prerequisites

• Python 3.9+ with checkov >= 3.2 installed, in the same interpreter that will run the checks — a check importing a library Checkov's environment does not have fails at load time, not at scan time.
• A directory of Terraform or CDKTF-synthesized JSON to scan, plus at least one deliberately non-compliant fixture so you can prove the check fires.
• A rule expressed concretely: which resource type, which attribute, what value is compliant, and what an engineer should do to fix it.
• A stable identifier prefix reserved for your organisation, decided once and written down.

# CLI: confirm Checkov and point it at a synthesized config
checkov --version && ls cdktf.out/stacks/*/cdk.tf.json

Before writing any logic, decide the identifier and the guideline text. The id appears in every report, every suppression comment, and every ticket that references the finding, so it must never be reused for a different rule. The guideline is a short sentence or an internal URL path telling the engineer what to change; a check that reports CKV_ACME_1 FAILED with no guidance generates arguments, not fixes.

Implementation

1. Look at the parsed configuration before writing logic

The single largest source of broken custom checks is guessing the shape of conf. Checkov's HCL parser wraps almost every value in a list, and nested blocks become lists of dicts, so a rule written against what the HCL looks like fails against what the parser produces. Dump it once:

# checks/_inspect.py — temporary, delete after you have the shape
# CLI: checkov -d cdktf.out --external-checks-dir checks --check CKV_ACME_0
from __future__ import annotations
import json
from typing import Any

from checkov.common.models.enums import CheckCategories, CheckResult
from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck


class DumpConf(BaseResourceCheck):
    def __init__(self) -> None:
        super().__init__(
            name="Debug: print the parsed resource configuration",
            id="CKV_ACME_0",
            categories=[CheckCategories.CONVENTION],
            supported_resources=["aws_s3_bucket"],
        )

    def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
        print(json.dumps(conf, indent=2, default=str))
        return CheckResult.PASSED


check = DumpConf()

For a bucket declared with bucket = "acme-logs" the output is {"bucket": ["acme-logs"], "__start_line__": 4, ...} — the value is a one-element list, and Checkov adds __start_line__ and __end_line__ keys it uses for reporting. Write conf["bucket"][0], never conf["bucket"]. The equivalent mistake in the other direction produces AttributeError: 'list' object has no attribute 'get' when a nested block is treated as a dict.

2. Write the check

Subclass BaseResourceCheck, declare the resource types it applies to, and implement scan_resource_conf returning PASSED or FAILED. This check requires S3 buckets to have access logging configured.

# checks/s3_logging.py — require S3 access logging
# CLI: checkov -d . --external-checks-dir checks
from __future__ import annotations
from typing import Any

from checkov.common.models.enums import CheckCategories, CheckResult
from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck


class S3AccessLogging(BaseResourceCheck):
    def __init__(self) -> None:
        super().__init__(
            name="Ensure S3 buckets enable access logging",
            id="CKV_ACME_1",
            categories=[CheckCategories.LOGGING],
            supported_resources=["aws_s3_bucket"],
            guideline="Set a logging block or attach an aws_s3_bucket_logging resource.",
        )

    def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
        # conf holds the resource block; 'logging' must be present and non-empty.
        logging = conf.get("logging")
        return CheckResult.PASSED if logging else CheckResult.FAILED


check = S3AccessLogging()

Two subtleties are worth calling out. conf.get("logging") returns [{}] — a truthy list containing an empty dict — when the block is present but has no attributes, so this check passes a bucket whose logging block is empty. If that matters, reach into the first element and require target_bucket. And because AWS split S3 configuration into separate resources, a modern configuration sets logging via aws_s3_bucket_logging rather than an inline block; a per-resource check cannot see that sibling resource, which is precisely the case where a YAML graph check is the correct tool.

3. Prefer a narrower base class where the rule fits

Most internal rules are "attribute X must equal Y", and BaseResourceValueCheck handles the traversal, the missing-block behaviour, and the report wording for you. You supply a dotted path and an expected value:

# checks/rds_deletion_protection.py — one attribute, one expected value
# CLI: checkov -d cdktf.out --external-checks-dir checks --check CKV_ACME_2
from __future__ import annotations
from typing import Any

from checkov.common.models.enums import CheckCategories, CheckResult
from checkov.terraform.checks.resource.base_resource_value_check import (
    BaseResourceValueCheck,
)


class RdsDeletionProtection(BaseResourceValueCheck):
    def __init__(self) -> None:
        super().__init__(
            name="Ensure production RDS instances enable deletion protection",
            id="CKV_ACME_2",
            categories=[CheckCategories.BACKUP_AND_RECOVERY],
            supported_resources=["aws_db_instance", "aws_rds_cluster"],
            # An absent attribute is a failure, not an exemption.
            missing_block_result=CheckResult.FAILED,
        )

    def get_inspected_key(self) -> str:
        return "deletion_protection"

    def get_expected_value(self) -> Any:
        return True


check = RdsDeletionProtection()

missing_block_result is the parameter people get wrong. Its default is CheckResult.FAILED, which is usually what you want, but setting it to PASSED turns the check into "if the attribute is present it must be true" — a meaningfully weaker rule, and one that quietly exempts every resource that simply omits the field.

4. Write a rule that needs real Python

The case for a Python check rather than YAML is logic. Here the rule is an approved-AMI allow-list loaded from a file that the platform team owns, combined with a naming convention — neither of which a declarative condition can express:

# checks/approved_ami.py — allow-list plus naming convention
# CLI: checkov -d cdktf.out --external-checks-dir checks --check CKV_ACME_3
from __future__ import annotations
import os
import re
from pathlib import Path
from typing import Any

from checkov.common.models.enums import CheckCategories, CheckResult
from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck

NAME_RE = re.compile(r"^acme-(dev|stg|prod)-[a-z0-9-]{3,40}$")


def _approved_amis() -> frozenset[str]:
    """Read the allow-list once, from a path CI controls."""
    path = Path(os.environ.get("ACME_AMI_ALLOWLIST", "policy/approved_amis.txt"))
    if not path.is_file():
        return frozenset()
    return frozenset(
        line.strip() for line in path.read_text().splitlines() if line.strip()
    )


class ApprovedAmiAndName(BaseResourceCheck):
    def __init__(self) -> None:
        super().__init__(
            name="Ensure EC2 instances use an approved AMI and the standard name",
            id="CKV_ACME_3",
            categories=[CheckCategories.GENERAL_SECURITY],
            supported_resources=["aws_instance"],
            guideline="Pick an AMI from policy/approved_amis.txt and name it acme-<env>-<role>.",
        )

    def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
        allowed = _approved_amis()
        if not allowed:
            # Fail closed: an unreadable allow-list must not silently pass a scan.
            return CheckResult.FAILED

        ami_values = conf.get("ami") or []
        ami = ami_values[0] if ami_values else None
        if not isinstance(ami, str) or ami not in allowed:
            return CheckResult.FAILED

        tags = (conf.get("tags") or [{}])[0]
        name = tags.get("Name") if isinstance(tags, dict) else None
        if not isinstance(name, str) or not NAME_RE.match(name):
            return CheckResult.FAILED
        return CheckResult.PASSED


check = ApprovedAmiAndName()

Failing closed when the allow-list is missing is a deliberate choice. The alternative — returning PASSED when the file cannot be read — means a mistyped path or a missing checkout in CI disables the policy for every resource without a single warning in the log.

Note also what happens when the AMI is a Terraform reference rather than a literal: the parsed value is a string like "${data.aws_ami.base.id}", which is not in the allow-list and therefore fails. That is arguably correct — the check cannot verify an unresolved reference — but it must be a decision you make consciously, and it should be in the guideline text so engineers know why.

Verification

Run Checkov with your external checks directory and confirm your custom id appears in the results, failing a non-compliant bucket.

# CLI: expect CKV_ACME_1 in the output, FAILED for an unlogged bucket
checkov -d cdktf.out --external-checks-dir checks --check CKV_ACME_1

Eyeballing CLI output is fine once. After that, test the check the way you would test any other Python: drive Checkov's runner directly against a fixture directory and assert on the report. This is the step that turns a policy from a script into something you can refactor safely.

# tests/test_custom_checks.py
# CLI: pytest tests/test_custom_checks.py -q
from __future__ import annotations
from pathlib import Path

import pytest
from checkov.runner_filter import RunnerFilter
from checkov.terraform.runner import Runner

FIXTURES = Path(__file__).parent / "fixtures"


def _ids(records: list) -> set[str]:
    return {r.resource for r in records}


@pytest.fixture(scope="module", autouse=True)
def _load_external_checks() -> None:
    """Import the checks package so each class registers itself."""
    import checks.s3_logging  # noqa: F401
    import checks.rds_deletion_protection  # noqa: F401


def test_unlogged_bucket_fails_and_logged_bucket_passes() -> None:
    report = Runner().run(
        root_folder=str(FIXTURES / "s3"),
        runner_filter=RunnerFilter(framework=["terraform"], checks=["CKV_ACME_1"]),
    )
    assert "aws_s3_bucket.unlogged" in _ids(report.failed_checks)
    assert "aws_s3_bucket.logged" in _ids(report.passed_checks)
    # State implication: none — Checkov reads files and never calls a cloud API.
    assert not report.parsing_errors, report.parsing_errors

Keep two fixtures per check — one that must fail and one that must pass — in a directory the scan never reaches during a real run. A check with only a failing fixture is the most common way a policy ends up rejecting everything, including compliant configurations, and nobody notices until a release is blocked.

Finally, confirm the exit code, because that is what CI actually consumes. Checkov exits 0 when nothing failed and 1 when any check failed; --soft-fail-on CKV_ACME_3 keeps a newly introduced policy visible in the report while it exits 0, which is how you roll a rule out without breaking every open pull request on day one.

# CLI: prove the gate behaves before wiring it into the pipeline
checkov -d cdktf.out --external-checks-dir checks --compact ; echo "exit=$?"
checkov -d cdktf.out --external-checks-dir checks --soft-fail-on CKV_ACME_3 ; echo "exit=$?"

Gotchas & Edge Cases

Attribute shape varies. Terraform wraps block attributes in lists; conf.get('logging') may be [{...}], so inspect the real conf before asserting on it.

Unique ids matter. Custom ids must not collide with built-ins; prefix them (CKV_ACME_*) so upgrades never clash.

Wire it into CI. A custom check only protects you if it runs on every change; add the --external-checks-dir flag to the same pipeline stage that runs the built-in scan.

A check module that raises at import time is skipped, not reported. Checkov logs a load failure and carries on scanning with the checks it did manage to import, so a ModuleNotFoundError: No module named 'yaml' in one file leaves you with a green pipeline and no policy. Run with LOG_LEVEL=DEBUG once after adding a check and confirm your id appears in the loaded-checks output.

--check and --skip-check silently accept unknown ids. Typing --check CKV_ACME_11 when the check is CKV_ACME_1 produces a report with zero passed and zero failed results and an exit code of 0. Assert on the number of evaluated resources in your test rather than trusting a clean run.

Suppression comments are per-resource and survive refactors. #checkov:skip=CKV_ACME_1:logs go to the central account sits in the resource block, and because CDKTF regenerates cdk.tf.json on every synth, a skip written into the generated file is lost. Suppressions for synthesized stacks belong in the .checkov.yaml config or as a baseline file, not in the output directory.

The same id in two files is a coin flip. Both modules import, both instantiate, and the registry keeps one. Reserve a numeric range per team and keep a single index file so CKV_ACME_7 means one thing forever, including after the person who wrote it has left.

Operational Notes

A custom policy has a lifecycle: proposed, enforced in warn mode, enforced hard, and eventually retired when the platform makes it structurally impossible to violate. Skipping the warn stage is what makes engineers resent policy tooling — a rule that appears in CI as a hard failure on a Friday afternoon gets suppressed rather than fixed.

Roll out in three steps. Ship the check with --soft-fail-on <id> so it reports without blocking, and watch the failure count for a week. Generate a baseline of the existing violations with checkov -d cdktf.out --create-baseline and commit .checkov.baseline, which lets new code be held to the rule while existing debt is tracked separately. Then remove the soft-fail and let it block.

# CLI: baseline today's violations, block anything new from here on
checkov -d cdktf.out --external-checks-dir checks --create-baseline
git add cdktf.out/.checkov.baseline
checkov -d cdktf.out --external-checks-dir checks --baseline cdktf.out/.checkov.baseline

Distribute the checks the same way you distribute any other Python. A directory works for one repository; across many, either publish the checks as a package on your internal index or point Checkov at a repository with --external-checks-git, which pins them for everyone at once. Whichever you choose, the checks need their own tests and their own review — they are code that can block every deployment in the organisation, and they deserve the same care as the infrastructure they police.

Two operational details are easy to miss. Scan time grows with the number of checks multiplied by the number of resources, and an expensive scan_resource_conf — one that reads a file or compiles a regular expression on every call — turns a two-second scan into a two-minute one; hoist that work to module scope as the allow-list example does. And output format matters for anything downstream: -o sarif uploads into code-scanning dashboards, -o junitxml renders in CI test panes, and -o json is what you parse when you want to route findings to the owning team rather than to whoever opened the pull request. The complementary preview-time approach is covered in Pulumi CrossGuard policy as code.

FAQ

Can I write checks for CDKTF output?

Yes — Checkov scans the synthesized cdk.tf.json, so a check on aws_s3_bucket works whether the JSON came from HCL or CDKTF. Pass --framework terraform_json for that directory; the HCL parser does not recognise .tf.json files and will report nothing at all.

How do I suppress a check on one resource?

Add a #checkov:skip=CKV_ACME_1:<reason> comment inside the resource block, or list the id in a .checkov.yaml config file. Always include the reason — Checkov prints it in the report, which is what makes a suppression auditable rather than an unexplained hole.

Is this a replacement for CrossGuard?

They overlap but differ: Checkov scans static config, while Pulumi CrossGuard evaluates Pulumi's planned graph at preview. CrossGuard sees resolved values and resource relationships that a static file scan cannot, at the cost of running only for Pulumi.

Why does my check pass when the attribute is obviously missing?

Almost always missing_block_result, or a truthy empty structure. BaseResourceValueCheck treats an absent key according to that parameter, and a conf.get("logging") that returns [{}] is truthy even though the block is empty. Assert on the value inside the list, not on the list.

Can a custom check look at more than one resource?

Not with BaseResourceCheck — it receives one resource block at a time. Rules that span resources, such as "every bucket must have a matching public access block", need a YAML graph check, which walks the parsed relationship graph and can express connection conditions.

How do I roll out a new policy without blocking every open pull request?

Ship it with --soft-fail-on <id> so it reports but exits zero, then create a baseline with --create-baseline so pre-existing violations are recorded and only new ones fail. Remove the soft-fail once the failure count on new code has been zero for a full sprint.

Related

• Scanning Python IaC with Checkov — running the built-in checks this guide extends.
• Security & Compliance Basics — the parent topic on securing Python IaC.
• Pulumi Policy as Code with CrossGuard — the Pulumi-native alternative for preview-time enforcement.