Write Pulumi CrossGuard Policies in Python

This how-to, part of the Pulumi policy as code cluster 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.

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
  • pulumi>=3.0 and a stack you can preview
  • The exact Pulumi resource type token for the resource you are guarding
# CLI: scaffold and confirm the policy SDK
mkdir policy && cd policy && python -c "import pulumi_policy"

Implementation

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.

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 pulumi_policy import (
    PolicyPack, ResourceValidationPolicy, EnforcementLevel, ReportViolation)

SG = "aws:ec2/securityGroup:SecurityGroup"

def no_public_ssh(args, report: ReportViolation) -> None:
    if args.resource_type != SG:
        return
    for rule in args.props.get("ingress", []):
        cidrs = rule.get("cidrBlocks", [])
        opens_ssh = rule.get("fromPort", 0) <= 22 <= rule.get("toPort", 0)
        if opens_ssh and "0.0.0.0/0" in cidrs:
            report("Security group opens port 22 to 0.0.0.0/0.")

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)])

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

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.

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.