Write Pulumi CrossGuard Policies in Python

This how-to, part of the Pulumi policy as code topic under Pulumi patterns and provider management, walks through building a resource validation policy from an empty directory to a rule that fails a preview — the smallest useful CrossGuard pack.

Context

You have a stack and a rule you want enforced: no security group may expose port 22 to the internet. Rather than trust code review, you will encode it as a policy that runs on every preview and blocks the change automatically.

CrossGuard is not a linter over your source code. It runs against the planned resource graph, after the Pulumi program has executed and every input has been computed, but before a single API call reaches the cloud provider. That placement is the whole value: a rule written against the resource model catches a violation regardless of how the resource got there — written by hand, produced by a component resource, rendered by a Helm chart, or generated by a loop over a config list. A grep over .py files catches none of those reliably.

The pack itself is an ordinary Python project with its own directory, its own requirements.txt, and its own virtual environment. Pulumi launches it as a separate process and speaks to it over gRPC, handing it each planned resource in turn; your callbacks report violations back. Because it is a real Python process, a policy can do anything Python can — read a JSON allowlist off disk, compute a CIDR containment check with ipaddress, or share a helper module with your other guardrails. Nothing about it is a bespoke expression language.

Two kinds of callback exist and they are not interchangeable. A ResourceValidationPolicy sees exactly one resource's properties at a time and knows nothing about the rest of the stack, which makes it fast and easy to unit test. A StackValidationPolicy receives the whole planned graph in one call, which is what you need for rules of the form "every bucket must be referenced by a logging configuration" or "no more than one NAT gateway per stack". Start with the resource form; reach for the stack form only when a rule genuinely spans resources.

Enforcement level is the third axis. EnforcementLevel.ADVISORY prints the violation and lets the update proceed, MANDATORY fails the run, and DISABLED keeps the rule loaded but silent. The level can be set on the pack as a default and overridden per policy, which is what makes it practical to ship a new rule to a large organisation without breaking every pipeline on the first morning.

The rule The rule: No public SSH with 4 facets. No public SSH Type aws SecurityGroup Field ingress Check 0.0.0.0/0 + 22 Level mandatory
The policy matches security groups and fails any that open port 22 to the world.

Prerequisites

Prerequisites Prerequisites: layered from Python interface / API down to Cloud runtime. Python interface / API Typed resource model Provider plugin State backend Cloud runtime
Prerequisites: the stack from the Python interface down to the cloud runtime.
  • Python 3.9+ with pulumi-policy installed in a dedicated policy/ project — its own virtual environment, not the stack's, so the pack's dependencies can never shift what the stack resolves.
  • pulumi>=3.0 and a stack you can preview, ideally a scratch one you are free to break.
  • The exact Pulumi resource type token for the resource you are guarding. Tokens are <package>:<module>/<resource>:<Resource> and they are case-sensitive; a typo produces a policy that silently never fires.
  • A stack that currently violates the rule, so you can prove the pack fails before you trust it to pass.
# CLI: scaffold and confirm the policy SDK
mkdir policy && cd policy && python -c "import pulumi_policy"

Find the token from the state you already have rather than from memory — this is the single most common reason a first policy does nothing at all:

# CLI: list the distinct type tokens present in the stack you want to guard
pulumi stack export | python -c "import json,sys; print(*sorted({r['type'] for r in json.load(sys.stdin)['deployment']['resources']}), sep='\n')"

Implementation

Step 1 — scaffold the pack

A policy pack is a directory containing PulumiPolicy.yaml, a requirements.txt, and a __main__.py. The YAML file is what tells the CLI that the directory is a pack and which runtime to launch; without it, --policy-pack fails with error: failed to load policy pack: no PulumiPolicy.yaml.

# policy/PulumiPolicy.yaml
# CLI: pulumi preview --policy-pack ./policy
name: network-baseline
runtime: python
description: Baseline network guardrails for AWS stacks.
version: 0.0.1
# CLI: create the pack's own virtualenv and install the policy SDK
cd policy && python -m venv venv && ./venv/bin/pip install pulumi-policy>=1.9.0

Keep version under version control and bump it on every rule change. It is what appears next to each violation in the output, so an engineer looking at a failed pipeline can tell which revision of the rules rejected their change.

Step 2 — write the resource validation policy

Write the pack's entry point. The validate function inspects the ingress rules and reports a violation for any that pairs an open CIDR with port 22.

The signature matters: validate receives a ResourceValidationArgs and a report callable. The args object carries resource_type (the token), props (the resource's planned inputs as a plain dict, in camelCase — the wire format, not the Python snake_case you wrote), name (the logical name), and urn. That camelCase detail catches everyone once: the program said cidr_blocks=, the policy must read cidrBlocks.

Preview with policy Preview with policy: preview → engine → policy pack. preview engine policy pack plan graph validate each violations fail run
The engine hands each planned resource to the pack, which fails the preview on a violation.
# policy/__main__.py — block public SSH
# CLI: pulumi preview --policy-pack ./policy
from __future__ import annotations

from typing import Any

from pulumi_policy import (
    EnforcementLevel,
    PolicyPack,
    ReportViolation,
    ResourceValidationArgs,
    ResourceValidationPolicy,
)

SG = "aws:ec2/securityGroup:SecurityGroup"
OPEN_V4 = "0.0.0.0/0"
OPEN_V6 = "::/0"


def no_public_ssh(args: ResourceValidationArgs, report: ReportViolation) -> None:
    if args.resource_type != SG:
        return
    rules: list[dict[str, Any]] = args.props.get("ingress") or []
    for rule in rules:
        # Provider note: props arrive in the provider's wire shape — cidrBlocks,
        # not the cidr_blocks you typed in the Pulumi program.
        cidrs: list[str] = (rule.get("cidrBlocks") or []) + (rule.get("ipv6CidrBlocks") or [])
        from_port: int = rule.get("fromPort") or 0
        to_port: int = rule.get("toPort") or 0
        opens_ssh = from_port <= 22 <= to_port
        exposed = [c for c in cidrs if c in (OPEN_V4, OPEN_V6)]
        if opens_ssh and exposed:
            report(
                f"Security group opens port 22 to {', '.join(exposed)}. "
                "Restrict ingress to a bastion security group or an internal CIDR."
            )


PolicyPack(name="network-baseline",
           enforcement_level=EnforcementLevel.MANDATORY,
           policies=[ResourceValidationPolicy(
               name="no-public-ssh",
               description="SSH must not be open to the internet.",
               validate=no_public_ssh)])

Three details in that code are load-bearing. args.props.get("ingress") or [] rather than .get("ingress", []) handles the case where the key exists with a None value, which happens on a resource whose ingress list is an unresolved output. The IPv6 list is checked because ::/0 is exactly as open as 0.0.0.0/0 and is the standard way to route around a rule that only looks at IPv4. And the message names the fix, not just the fault — a violation that says only "not allowed" generates a support conversation, one that says "use a bastion security group" does not.

Step 3 — decide what to do about unknown values

During a preview, any property derived from a resource that does not exist yet is unknown: Pulumi has no value to give you, and the property arrives as None. A rule that reads a port range off a security group whose CIDR list is unknown cannot conclude anything. You have to choose, per rule, whether that is a pass or a fail.

# policy/unknowns.py — fail closed for a network rule that cannot be evaluated
# CLI: pulumi preview --policy-pack ./policy
from __future__ import annotations

from pulumi_policy import ReportViolation, ResourceValidationArgs

SG = "aws:ec2/securityGroup:SecurityGroup"


def ingress_must_be_knowable(args: ResourceValidationArgs, report: ReportViolation) -> None:
    if args.resource_type != SG:
        return
    if args.props.get("ingress") is None and "ingress" in args.props:
        # State implication: this fires only at preview. The same policy on
        # `pulumi up` sees the resolved value and passes silently.
        report(
            "Ingress rules are computed from another resource and cannot be "
            "checked before the update. Move the CIDR into stack config, or "
            "run the pack again after the dependency exists."
        )

For a mandatory network rule, failing closed is right: an unevaluable rule that reports success is worse than no rule. For an advisory tagging rule, failing closed just trains people to ignore the output. The point is to make the choice explicitly rather than inherit it from a .get() default.

Step 4 — reach for stack validation when the rule spans resources

Some rules cannot be expressed one resource at a time. "Every security group in this stack must be attached to something" needs the whole graph, which is what StackValidationPolicy provides.

Two validation callbacks, two jobs Two validation callbacks, two jobs: comparison across Sees, Use it for, Runs. Policy type Sees Use it for Runs ResourceValidationPolicy One resource at a time Per-resource shape rules Once per resource StackValidationPolicy Every resource plus URNs Rules about relationships Once per stack
Pick the callback by what the rule needs to see: a single set of properties, or the whole planned graph.
# policy/stack_rules.py — a rule that needs to see every resource at once
# CLI: pulumi preview --policy-pack ./policy
from __future__ import annotations

from pulumi_policy import (
    EnforcementLevel,
    ReportViolation,
    StackValidationArgs,
    StackValidationPolicy,
)

SG = "aws:ec2/securityGroup:SecurityGroup"
INSTANCE = "aws:ec2/instance:Instance"


def no_orphan_security_groups(args: StackValidationArgs, report: ReportViolation) -> None:
    groups = {r.urn: r for r in args.resources if r.resource_type == SG}
    referenced: set[str] = set()
    for resource in args.resources:
        if resource.resource_type != INSTANCE:
            continue
        for dep in resource.dependencies:
            referenced.add(str(dep))
    for urn, group in groups.items():
        if str(urn) not in referenced:
            # Provider note: report() takes the URN so the violation points at
            # the offending resource rather than at the stack as a whole.
            report(f"Security group '{group.name}' is not attached to any instance.", urn)


orphan_policy = StackValidationPolicy(
    name="no-orphan-security-groups",
    description="Security groups must be attached to a workload.",
    enforcement_level=EnforcementLevel.ADVISORY,
    validate=no_orphan_security_groups,
)

Stack validation runs once, after every resource has been planned, so it is the more expensive callback — but it is also the only place a relationship rule can live. Keep the resource-level rules doing the bulk of the work and use stack validation sparingly.

Verification

Point the pack at a stack with an offending rule and confirm the preview fails, then fix the rule and confirm it passes.

Verification Verification: Test → Program → Mock/Cloud. Test Program Mock/Cloud invoke declare resolve assert
Verification: the test drives the program and asserts on resolved values.
# CLI: expect a mandatory violation, then a clean run after fixing the SG
pulumi preview --policy-pack ./policy --stack dev

A firing mandatory rule ends the preview with a violation block naming the pack, its version, the rule, and the resource, followed by the description and your message:

Policy Violations:
    [mandatory]  network-baseline v0.0.1  no-public-ssh  (aws:ec2/securityGroup:SecurityGroup: web-sg)
    SSH must not be open to the internet.
    Security group opens port 22 to 0.0.0.0/0. Restrict ingress to a bastion security group or an internal CIDR.

Check the exit code, not the text — that is what a pipeline will act on:

# CLI: prove the pack fails the run rather than just printing something
pulumi preview --policy-pack ./policy --stack dev ; echo "exit=$?"   # expect nonzero

The faster loop is a unit test. Because validate is a plain function over a dict, you can exercise every branch without a stack, a cloud account, or the CLI — which is what makes a rule worth trusting before it blocks anyone else's deploy.

# policy/tests/test_no_public_ssh.py
# CLI: pytest policy/tests -q
from __future__ import annotations

from typing import Any

import pytest
from pulumi_policy import ResourceValidationArgs

from policy.__main__ import SG, no_public_ssh


def args_for(props: dict[str, Any]) -> ResourceValidationArgs:
    return ResourceValidationArgs(resource_type=SG, props=props, urn="urn:pulumi:dev::x::sg::web-sg",
                                  name="web-sg", opts=None, provider=None)


def test_open_ssh_is_reported() -> None:
    seen: list[str] = []
    no_public_ssh(
        args_for({"ingress": [{"fromPort": 22, "toPort": 22, "cidrBlocks": ["0.0.0.0/0"]}]}),
        lambda message, urn=None: seen.append(message),
    )
    assert seen and "port 22" in seen[0]


@pytest.mark.parametrize("props", [
    {"ingress": [{"fromPort": 22, "toPort": 22, "cidrBlocks": ["10.0.0.0/8"]}]},
    {"ingress": [{"fromPort": 443, "toPort": 443, "cidrBlocks": ["0.0.0.0/0"]}]},
    {"ingress": None},
    {},
])
def test_allowed_shapes_are_silent(props: dict[str, Any]) -> None:
    seen: list[str] = []
    no_public_ssh(args_for(props), lambda message, urn=None: seen.append(message))
    assert seen == []

The parametrised negative cases are the important half. A rule that fires on the bad input is easy; a rule that stays quiet on a wide port range, an internal CIDR, an unresolved output and a resource with no ingress at all is one that will not be disabled a week later for crying wolf.

Gotchas & Edge Cases

Gotchas & Edge Cases Gotchas & Edge Cases: Where it breaks with 4 facets. Where it breaks None watch this boundary ingress watch this boundary SecurityGroupR watch this boundary args.resource_ watch this boundary
Gotchas & Edge Cases: the boundaries where things break and what to check.

Computed CIDRs. If the CIDR is an unresolved output at preview, it may appear as None; treat unknown values conservatively and prefer failing closed for mandatory network rules.

Multiple ingress shapes. Inline ingress and standalone SecurityGroupRule resources both exist; guard both types or attackers route around your rule.

Wrong token. aws:ec2/securityGroup:SecurityGroup is case- and path-sensitive; print args.resource_type if the rule never fires.

camelCase properties. args.props carries the provider's wire names, so a program written with cidr_blocks is inspected with cidrBlocks. A rule reading the snake_case key finds nothing and reports a clean stack forever.

The pack has its own dependency tree. Pulumi launches the pack with its own interpreter, so importing a helper that only exists in the stack's virtualenv fails with ModuleNotFoundError at policy load time, and the run aborts before a single resource is evaluated. Vendor shared helpers into the pack or publish them as a package both projects install.

Component resources are evaluated too. A ComponentResource appears in the stream with its own type token and an empty-looking props, alongside the real children it created. Filtering strictly on the provider token you care about avoids reporting the same logical violation twice.

A pack that raises kills the run. An uncaught exception inside validate is not a violation; it surfaces as a policy-pack failure and stops the update entirely. Guard every dict access and prefer returning early over assuming a shape.

Operational Notes

Writing the rule is an afternoon; landing it across an organisation without being routed around is the actual project.

Ship every new rule as ADVISORY first. An advisory violation prints in the same block and the update continues, so a week of previews tells you exactly how many stacks are non-compliant and who owns them — information you cannot get from reading code. Fix or exempt those, then promote the same rule to MANDATORY in a version bump.

Getting a rule to mandatory without breaking teams Getting a rule to mandatory without breaking teams: Advisory then Measure then Remediate then Mandatory Advisory warn only Measure count violations Remediate fix the stacks Mandatory block the update
Rollout order: a rule that lands as mandatory on day one is a rule that gets disabled on day two.

For anything beyond one repository, publish the pack rather than passing --policy-pack ./policy on every command. pulumi policy publish <org> uploads the versioned pack, pulumi policy enable <org>/network-baseline latest applies it to a policy group, and stacks in that group are then checked whether or not the engineer running pulumi up remembered a flag. A local flag is a suggestion; an enabled group is a control.

# CLI: publish the pack and enable it for the default policy group
cd policy && pulumi policy publish acme
pulumi policy enable acme/network-baseline latest
pulumi policy ls acme

Version the pack like a library, because that is what it is. Pinning a group to 0.0.1 rather than latest means a rule change reaches teams when someone chooses to move them, which is the difference between a guardrail and an outage. Keep a CHANGELOG entry per rule promotion, and treat a level change from advisory to mandatory as a breaking change.

Exemptions need a design, or they will be invented for you. The two workable shapes are a resource-level opt-out — a tag or annotation the policy reads, such as policy-exempt: no-public-ssh — and a stack allowlist checked into the pack. Both are inspectable and reviewable. What does not work is disabling the rule for the whole organisation because one legacy stack cannot comply; that turns a control into decoration and nobody notices until an audit.

Finally, run the pack in CI on the pull request, not only on the deploy. A violation caught in a preview against a scratch stack costs a comment; the same violation caught during pulumi up on production costs a rollback conversation. Wire it into the same job that already runs pulumi preview, and make the exit code the gate.

FAQ

Can one pack hold many rules?

Yes — pass a list of ResourceValidationPolicy objects. Group a baseline (SSH, encryption, tagging) into one pack so teams adopt it as a unit.

How do I test the rule?

Call no_public_ssh with a fake args object and a stub report, asserting it fires only for the bad case — a plain pytest, no cloud needed.

Advisory or mandatory?

Start advisory to gauge impact, then promote to mandatory once you have fixed existing violations.

Why does my policy never fire?

Almost always one of two things: the type token does not match, or the property name is snake_case where args.props uses camelCase. Print args.resource_type and sorted(args.props) from inside validate on a run you know should fail — the answer is visible in one preview.

Does CrossGuard run on pulumi destroy?

Resource validation runs against planned resources, so a pure delete gives the callbacks nothing to inspect and the pack effectively passes. Rules about what may be removed belong in protect on the resource or in branch protection, not in a validation policy.

Can a policy read stack configuration?

Not directly — the pack is a separate process and does not receive the stack's config. Pass what a rule needs through the pack's own configuration or a file inside the pack directory, and keep the rule's inputs explicit so its unit tests stay honest.