Compliance Testing

Automated Compliance Testing with Policy-as-Code

HelpMeTest

HelpMeTest

— 7 min read
Automated Compliance Testing with Policy-as-Code

The problem with compliance testing is that most teams treat it as a layer separate from their regular test suite. GDPR tests live in one place, SOC 2 evidence collection in another, PCI tests in a third. Each uses a different toolchain, runs on a different schedule, and is owned by a different person. When your architecture changes, three separate suites need updating. When a new regulation arrives, you build a fourth silo.

Policy-as-code solves this. The idea is simple: encode your compliance requirements as machine-readable rules in the same repository as your code, execute those rules automatically as part of CI, and use a single framework that can express requirements across multiple regulations.

This post covers how to implement that using Open Policy Agent (OPA), Conftest, and custom assertion frameworks.

The Core Concept: Policies Are Tests

A compliance policy is a statement like:

"All API endpoints that return personal data must require authentication."

A test is a statement like:

"When I call GET /api/users without a token, I get a 401."

These are the same statement. Policy-as-code frameworks make this explicit by letting you write the policy once and execute it as a test automatically.

# policies/api-security.rego (Open Policy Agent)
package api.security

import future.keywords.if
import future.keywords.in

# All personal data endpoints must require authentication
deny[msg] if {
    endpoint := input.endpoints[_]
    endpoint.returns_personal_data == true
    not endpoint.requires_authentication
    msg := sprintf("Endpoint %v returns personal data but does not require auth", [endpoint.path])
}

# All endpoints must use HTTPS
deny[msg] if {
    endpoint := input.endpoints[_]
    not startswith(endpoint.url, "https://")
    msg := sprintf("Endpoint %v does not use HTTPS", [endpoint.path])
}

# Session tokens must expire within policy window
deny[msg] if {
    config := input.auth_config
    config.session_ttl_seconds > 28800  # 8 hours
    msg := sprintf("Session TTL %vs exceeds maximum 8 hours", [config.session_ttl_seconds])
}

Setting Up OPA for API Compliance Testing

The workflow is: extract a machine-readable description of your API configuration, pass it to OPA, and let your policies evaluate it.

# scripts/generate_api_manifest.py
"""
Generate a machine-readable manifest of your API's security configuration.
This is the 'input' that OPA policies will evaluate.
"""
import json
import requests

def generate_api_manifest(base_url: str, admin_token: str) -> dict:
    """
    Query your API's introspection or OpenAPI spec endpoint and
    augment it with runtime security information.
    """
    # Get the OpenAPI spec
    spec_resp = requests.get(f"{base_url}/api/openapi.json")
    spec = spec_resp.json()

    # Get runtime config (auth settings, session config, etc.)
    config_resp = requests.get(
        f"{base_url}/api/admin/security-config",
        headers={"Authorization": f"Bearer {admin_token}"}
    )
    security_config = config_resp.json()

    endpoints = []
    for path, path_item in spec.get("paths", {}).items():
        for method, operation in path_item.items():
            if method in ("get", "post", "put", "patch", "delete"):
                endpoints.append({
                    "path": path,
                    "method": method.upper(),
                    "url": f"{base_url}{path}",
                    "requires_authentication": bool(operation.get("security")),
                    "returns_personal_data": any(
                        tag in operation.get("tags", [])
                        for tag in ["users", "patients", "personal-data", "pii"]
                    ),
                    "scopes_required": [
                        scope
                        for sec in operation.get("security", [])
                        for scope in sec.get("oauth2", [])
                    ]
                })

    return {
        "endpoints": endpoints,
        "auth_config": security_config.get("auth", {}),
        "encryption_config": security_config.get("encryption", {}),
        "logging_config": security_config.get("logging", {}),
    }

if __name__ == "__main__":
    manifest = generate_api_manifest(
        base_url="https://api.example.com",
        admin_token="your-token"
    )
    with open("/tmp/api-manifest.json", "w") as f:
        json.dump(manifest, f, indent=2)
    print("Manifest generated")

Then run OPA against it:

# Install OPA
curl -L -o opa https://openpolicyagent.org/downloads/latest/opa_linux_amd64
chmod +x opa

# Evaluate policies against manifest
opa eval \
  --input /tmp/api-manifest.json \
  --data policies/ \
  --format pretty \
  "data.api.security.deny"

Infrastructure Compliance with Conftest

Conftest applies OPA policies to configuration files — Kubernetes manifests, Terraform plans, Dockerfiles. This is where infrastructure compliance lives.

# policies/kubernetes-security.rego
package kubernetes.security

# Containers must not run as root
deny[msg] if {
    container := input.spec.containers[_]
    not container.securityContext.runAsNonRoot
    msg := sprintf("Container %v does not set runAsNonRoot=true", [container.name])
}

# Containers must have resource limits (availability requirement)
deny[msg] if {
    container := input.spec.containers[_]
    not container.resources.limits
    msg := sprintf("Container %v has no resource limits set", [container.name])
}

# Secrets must not be passed as environment variables in plaintext
deny[msg] if {
    container := input.spec.containers[_]
    env := container.env[_]
    contains(lower(env.name), "password")
    env.value  # Has a literal value, not a secretKeyRef
    msg := sprintf("Container %v has password in env var %v as plaintext",
                   [container.name, env.name])
}

# Services must not be exposed with LoadBalancer type unless explicitly tagged
deny[msg] if {
    input.kind == "Service"
    input.spec.type == "LoadBalancer"
    not input.metadata.annotations["compliance/public-exposure-approved"]
    msg := sprintf("Service %v is LoadBalancer type but lacks public-exposure approval annotation",
                   [input.metadata.name])
}
# policies/terraform-pci.rego
package terraform.pci

import future.keywords.in

# S3 buckets storing card data must have encryption enabled
deny[msg] if {
    resource := input.resource.aws_s3_bucket[name]
    contains(name, "payment")
    not resource.server_side_encryption_configuration
    msg := sprintf("S3 bucket %v stores payment data but lacks server-side encryption", [name])
}

# RDS instances in CDE must have encryption at rest
deny[msg] if {
    resource := input.resource.aws_db_instance[name]
    not resource.storage_encrypted
    msg := sprintf("RDS instance %v is not encrypted at rest — required for CDE", [name])
}

# Security groups must not allow 0.0.0.0/0 inbound on database ports
deny[msg] if {
    resource := input.resource.aws_security_group_rule[name]
    resource.type == "ingress"
    resource.cidr_blocks[_] == "0.0.0.0/0"
    resource.from_port <= 5432
    resource.to_port >= 5432
    msg := sprintf("Security group rule %v allows public inbound on PostgreSQL port", [name])
}

Run Conftest as part of your CI:

# .github/workflows/policy-check.yml
name: Policy Compliance Check

on: [pull_request]

jobs:
  policy-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install Conftest
        run: |
          wget https://github.com/open-policy-agent/conftest/releases/latest/download/conftest_Linux_x86_64.tar.gz
          tar xzf conftest_Linux_x86_64.tar.gz
          sudo mv conftest /usr/local/bin/

      - name: Check Kubernetes manifests
        run: |
          conftest test k8s/ \
            --policy policies/kubernetes-security.rego \
            --output table

      - name: Check Terraform plan
        run: |
          terraform plan -out=tfplan.binary
          terraform show -json tfplan.binary > tfplan.json
          conftest test tfplan.json \
            --policy policies/terraform-pci.rego \
            --output table

      - name: Check API manifest
        run: |
          python scripts/generate_api_manifest.py
          conftest test /tmp/api-manifest.json \
            --policy policies/api-security.rego \
            --output table

Building a Multi-Regulation Policy Library

The real power of policy-as-code is that you can tag policies with which regulation they satisfy, and generate compliance reports automatically.

# policies/multi-regulation.rego
package compliance

import future.keywords.in

# Metadata: which regulations each control satisfies
control_metadata := {
    "encryption_at_rest": {
        "regulations": ["PCI-DSS-3.5", "HIPAA-164.312e", "SOC2-CC6.7"],
        "severity": "critical"
    },
    "access_logging": {
        "regulations": ["SOC2-CC7.2", "HIPAA-164.312b", "GDPR-Art30"],
        "severity": "high"
    },
    "mfa_for_privileged_access": {
        "regulations": ["PCI-DSS-8.4.2", "SOC2-CC6.1"],
        "severity": "critical"
    },
    "data_retention_policy": {
        "regulations": ["GDPR-Art5", "HIPAA-164.530j"],
        "severity": "high"
    },
}

# Encryption at rest control
violations["encryption_at_rest"] if {
    db := input.databases[_]
    not db.encrypted_at_rest
}

# Access logging control
violations["access_logging"] if {
    service := input.services[_]
    not service.audit_logging_enabled
}

# MFA control
violations["mfa_for_privileged_access"] if {
    user := input.privileged_users[_]
    not user.mfa_enabled
}

# Generate a compliance report
compliance_report := {
    "violations": [
        {
            "control": control,
            "regulations_affected": control_metadata[control].regulations,
            "severity": control_metadata[control].severity
        }
        |
        control := violations[_]
        control_metadata[control]
    ],
    "passing_controls": [
        control
        |
        control := control_metadata[_]
        not violations[control]
    ]
}

Writing Custom Policy Assertions

For business logic that doesn't map cleanly to OPA, write your policies as pytest fixtures that other tests can import:

# tests/compliance/policies.py
"""
Reusable compliance policy assertions.
Import these in any test that needs to verify compliance behavior.
"""
import re
import requests
from functools import wraps
from typing import Callable

PAN_PATTERN = re.compile(r"\b(?:4[0-9]{12}(?:[0-9]{3})?|5[1-5][0-9]{14})\b")

class CompliancePolicy:
    """
    Decorator-based policy assertions that can be applied to any test.
    """

    @staticmethod
    def no_pii_in_errors(func: Callable):
        """Apply to any test that makes API calls — verifies errors don't leak PII."""
        @wraps(func)
        def wrapper(*args, **kwargs):
            original_get = requests.get
            original_post = requests.post
            captured_responses = []

            def capturing_get(url, **kw):
                resp = original_get(url, **kw)
                if resp.status_code >= 400:
                    captured_responses.append(resp)
                return resp

            requests.get = capturing_get
            try:
                result = func(*args, **kwargs)
            finally:
                requests.get = original_get

            pii_patterns = [PAN_PATTERN, re.compile(r"\b\d{3}-\d{2}-\d{4}\b")]
            for resp in captured_responses:
                for pattern in pii_patterns:
                    matches = pattern.findall(resp.text)
                    assert not matches, \
                        f"PII found in error response from {resp.url}: {matches}"

            return result
        return wrapper

    @staticmethod
    def assert_gdpr_deletion_complete(base_url: str, user_id: str, admin_token: str):
        """
        After a deletion request, verify the user's data is gone from primary storage.
        Call this in any test that exercises the deletion flow.
        """
        resp = requests.get(
            f"{base_url}/api/users/{user_id}",
            headers={"Authorization": f"Bearer {admin_token}"}
        )
        assert resp.status_code in (404, 410), \
            f"User {user_id} still accessible after deletion request"

    @staticmethod
    def assert_phi_access_logged(admin_token: str, patient_id: str, since_timestamp: int):
        """
        Verify that a PHI access to the given patient was logged.
        Call after any test that reads patient data.
        """
        audit_resp = requests.get(
            f"{BASE_URL}/api/admin/audit-log",
            params={"resource_id": patient_id, "from": since_timestamp},
            headers={"Authorization": f"Bearer {admin_token}"}
        )
        assert len(audit_resp.json()["events"]) >= 1, \
            f"PHI access to patient {patient_id} not recorded in audit log"


# Usage example:
class TestPaymentFlow:
    @CompliancePolicy.no_pii_in_errors
    def test_payment_with_invalid_card(self):
        """Test that invalid card errors do not expose PAN in response."""
        resp = requests.post(f"{BASE_URL}/api/payments", json={
            "card": {"number": "4111111111111111", "exp_month": 1, "exp_year": 2020},
            "amount": 100
        })
        assert resp.status_code == 422
        # The decorator checks that PAN is not in any error response

Generating Compliance Reports from CI

Wire everything together to produce a compliance dashboard:

# scripts/compliance_report.py
"""
Run all policy checks and generate a unified compliance report.
"""
import subprocess
import json
from datetime import datetime

def run_opa_check(policy_file: str, input_file: str, query: str) -> dict:
    result = subprocess.run(
        ["opa", "eval", "-i", input_file, "-d", policy_file,
         "--format", "json", query],
        capture_output=True, text=True
    )
    return json.loads(result.stdout)

def generate_report():
    report = {
        "generated_at": datetime.utcnow().isoformat(),
        "regulations": {}
    }

    checks = [
        ("GDPR", "policies/gdpr.rego", "/tmp/api-manifest.json", "data.gdpr.deny"),
        ("SOC2", "policies/soc2.rego", "/tmp/api-manifest.json", "data.soc2.deny"),
        ("PCI-DSS", "policies/pci.rego", "/tmp/api-manifest.json", "data.pci.deny"),
        ("HIPAA", "policies/hipaa.rego", "/tmp/api-manifest.json", "data.hipaa.deny"),
    ]

    for regulation, policy, manifest, query in checks:
        result = run_opa_check(policy, manifest, query)
        violations = result.get("result", [{}])[0].get("expressions", [{}])[0].get("value", [])
        report["regulations"][regulation] = {
            "violations": violations,
            "status": "FAIL" if violations else "PASS"
        }

    with open("compliance-report.json", "w") as f:
        json.dump(report, f, indent=2)

    # Summary
    for reg, data in report["regulations"].items():
        status = data["status"]
        count = len(data["violations"])
        print(f"{reg}: {status} ({count} violations)")

    any_fail = any(d["status"] == "FAIL" for d in report["regulations"].values())
    return 1 if any_fail else 0

if __name__ == "__main__":
    import sys
    sys.exit(generate_report())

The Policy-as-Code Maturity Model

Teams typically evolve through three stages:

Stage 1 — Reactive: Compliance tests exist but run manually before audits. Violations are found late and fixed under pressure.

Stage 2 — Gate: Compliance policies run in CI and block merges. Violations are found at PR time. This is the minimum viable state.

Stage 3 — Proactive: Policies are co-located with the code they govern, tagged with regulation references, and generate machine-readable evidence artifacts. Auditors receive a report generated by the same CI system that prevented regressions.

Most organizations are at Stage 1. Getting to Stage 2 requires a day of work per regulation. Getting to Stage 3 requires building the evidence generation pipeline described here. The return is that your next audit is a query against a database, not a three-month evidence-gathering exercise.

Start with the regulation that matters most to your business. Pick the five highest-risk controls. Write OPA policies for them. Wire them into CI. That is a week of work that compounds forever.

Read more

Start now free