Custom Checks Best Practices¶

Tips and guidelines for writing effective custom validation checks.

Naming¶

Use Descriptive Check IDs¶

# Good
check_id: ClassVar[str] = "require_mfa_for_iam_actions"

# Bad
check_id: ClassVar[str] = "check1"

Use Clear Descriptions¶

# Good
description: ClassVar[str] = "Ensures IAM write actions require MFA authentication"

# Bad
description: ClassVar[str] = "MFA check"

Error Messages¶

Be Specific¶

# Good
message=f"Action '{action}' requires MFA but no aws:MultiFactorAuthPresent condition found"

# Bad
message="MFA required"

Provide Actionable Suggestions¶

# Good
suggestion='Add condition: {"Bool": {"aws:MultiFactorAuthPresent": "true"}}'

# Bad
suggestion="Fix this"

Handle Edge Cases¶

Check for None Values¶

async def execute(self, statement, statement_idx, fetcher, config):
    issues = []

    # Handle None action
    if statement.action is None and statement.not_action is None:
        return issues

    # Use helper method for safe list access
    actions = statement.get_actions()

Handle Wildcards¶

for action in actions:
    if action == "*":
        # Handle full wildcard separately
        continue

    if "*" in action:
        # Handle partial wildcard
        expanded = await fetcher.expand_wildcard_action(action)

Severity Levels¶

Choose appropriate severity:

# Critical: Full admin access, public exposure
default_severity: ClassVar[str] = "critical"

# High: Missing security controls, sensitive actions
default_severity: ClassVar[str] = "high"

# Medium: Best practice violations
default_severity: ClassVar[str] = "medium"

# Low: Style issues, minor improvements
default_severity: ClassVar[str] = "low"

Configuration¶

Support Config Overrides¶

async def execute(self, statement, statement_idx, fetcher, config):
    # Get config with defaults
    required_tags = config.config.get("required_tags", ["Environment", "Owner"])
    max_wildcards = config.config.get("max_wildcards", 3)

    # Use configured severity
    severity = self.get_severity(config)

Document Config Options¶

# iam-validator.yaml
checks:
  my_check:
    enabled: true
    severity: high
    # Custom options
    required_tags:
      - Environment
      - Owner
      - CostCenter

Deduplicating Against Another Check¶

If your check would report an action that action_condition_enforcement already reports, ask that check what it enforces — never read its requirements config yourself:

from iam_validator.checks.action_condition_enforcement import ActionConditionEnforcementCheck
from iam_validator.checks.utils.aws_matching import action_matches

covered = ActionConditionEnforcementCheck.enforced_actions(config.root_config)
if any(action_matches(action, pattern) for pattern in covered):
    return []  # already reported there

enforced_actions(root_config, policy_file=None) returns the action patterns that check will actually enforce, which is not the same as what its requirements list contains. It returns an empty set when the check is disabled, and it honours merge_strategy (user_only, defaults_only, replace_all, …), user-supplied action_condition_requirements, and ignore_patterns. A requirement that any of those takes out of play is not returned — so you never suppress your own finding on the assumption that another check covers it, when it does not.

Two things to get right at the call site:

Match with action_matches, not ==. The returned values are patterns as configured, so iam:Pass* must cover iam:PassRole, and IAM action names are case-insensitive.
Pass policy_file when you have it. Without it, a requirement scoped by an ignore_patterns filepath cannot be evaluated and is treated as not enforced. That errs toward reporting a possible duplicate rather than dropping a real finding. Statement-level checks receive no policy_file; policy-level checks get it as the second argument to execute_policy.

Testing¶

Write Unit Tests¶

import pytest
from iam_validator.core.aws_service import AWSServiceFetcher
from iam_validator.core.check_registry import CheckConfig
from iam_validator.core.models import Statement
from my_checks.mfa_check import MFARequiredCheck


@pytest.mark.asyncio
async def test_mfa_check_detects_missing_mfa():
    check = MFARequiredCheck()

    statement = Statement(
        effect="Allow",
        action=["iam:DeleteUser"],
        resource="*",
        condition=None,
    )

    config = CheckConfig(
        check_id="mfa_required",
        config={"require_mfa_for": ["iam:DeleteUser"]}
    )

    async with AWSServiceFetcher() as fetcher:
        issues = await check.execute(statement, 0, fetcher, config)

    assert len(issues) == 1
    assert "MFA" in issues[0].message


@pytest.mark.asyncio
async def test_mfa_check_passes_with_condition():
    check = MFARequiredCheck()

    statement = Statement(
        effect="Allow",
        action=["iam:DeleteUser"],
        resource="*",
        condition={"Bool": {"aws:MultiFactorAuthPresent": "true"}},
    )

    config = CheckConfig(
        check_id="mfa_required",
        config={"require_mfa_for": ["iam:DeleteUser"]}
    )

    async with AWSServiceFetcher() as fetcher:
        issues = await check.execute(statement, 0, fetcher, config)

    assert len(issues) == 0

Test Edge Cases¶

Empty statements
Wildcard actions
Multiple actions in one statement
Deny statements (usually skip)
Missing conditions

Performance¶

Avoid Unnecessary AWS Calls¶

# Good - only call when needed
if needs_validation:
    is_valid, error, _ = await fetcher.validate_action(action)

# Bad - always calls API
for action in actions:
    await fetcher.validate_action(action)  # Even if not needed

Cache Results¶

class MyCheck(PolicyCheck):
    def __init__(self):
        self._cache = {}

    async def execute(self, statement, idx, fetcher, config):
        cache_key = statement.sid or idx
        if cache_key in self._cache:
            return self._cache[cache_key]

        # ... check logic ...

        self._cache[cache_key] = issues
        return issues

Documentation¶

Add Docstrings¶

class MyCheck(PolicyCheck):
    """Ensures S3 buckets require encryption.

    This check verifies that all S3 PutObject actions include
    the s3:x-amz-server-side-encryption condition to enforce
    encryption at rest.

    Configuration:
        allowed_encryption: List of allowed encryption types
            Default: ["AES256", "aws:kms"]

    Example:
        checks:
          s3_encryption:
            enabled: true
            allowed_encryption:
              - AES256
    """