Build an Evidence Pipeline, Step by Step

Brief

You own control IAM-02: every active workforce account has multi-factor authentication enrolled. Today the evidence is a screenshot taken the week before the audit. By the end of this room you will have a pipeline that collects the population from the source every night, evaluates it with policy code you wrote, stores the result with a hash, and alerts the owner when it fails.

Each step below is one build stage. Type the code, compare your output with the expected output, and read the common failure before moving on. The same skeleton then extends to cloud storage encryption, change approvals, pipeline runs and leaver access.

Step 1: Scope the control and write the evidence spec

Before writing a collector, write down exactly what the evidence must prove. If you cannot express the pass rule in one sentence, you are not ready to automate it. The spec lives in the repository next to the code, so a change to the rule is reviewed like any other change.

1. Create a repository folder evidence/ with sub-folders collectors/, policies/, tests/ and specs/.
2. Write specs/iam-02.yaml with the control id, the source system, the population, the field that decides pass or fail, the frequency and the owner.
3. State the empty-population rule explicitly: zero records is a failure, not a pass.
4. Get the control owner to approve the spec in a pull request before you write any code.

control_id: IAM-02
statement: Every active workforce account has MFA enrolled.
source: identity
population: users where status == "active"
decision_field: mfa_enabled
pass_rule: every user in population has mfa_enabled == true
on_empty_population: fail
frequency: daily 02:00 UTC
owner: iam-team
retention_days: 400

Common failure: the spec says “all users” but the auditor's population is “active users”. Agree the population with the owner first; it decides every count you will report.

Step 2: Collector: identity provider export

The collector has one job: get the complete population from the source and save it untouched. It does not decide pass or fail. Use a read-only API token scoped to user listing, read it from the environment, and never write it to logs.

Most identity APIs return results in pages. Your loop must follow the next-page cursor until there is none, retry transient errors with back-off, and then compare the number it collected with the total the API reports. That comparison is the completeness check, and it is the line most home-grown collectors forget.

1. Read the base URL and token from environment variables.
2. Request page one, store each page's items, follow the cursor field until it is empty.
3. Retry 429 and 5xx responses up to five times with exponential back-off; fail on anything else.
4. Compare collected count with the reported total; raise an error if they differ.
5. Write the raw JSON to raw/identity/<run-id>.json and compute its SHA-256.

import hashlib, json, os, time, uuid
import urllib.request

BASE = os.environ["IDP_BASE_URL"]
TOKEN = os.environ["IDP_READ_TOKEN"]   # read-only, never logged

def get(url, attempt=0):
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {TOKEN}"})
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return json.load(r)
    except urllib.error.HTTPError as e:
        if e.code in (429, 500, 502, 503) and attempt < 5:
            time.sleep(2 ** attempt)
            return get(url, attempt + 1)
        raise

def collect():
    users, url, total = [], f"{BASE}/users?limit=200", None
    while url:
        page = get(url)
        users.extend(page["items"])
        total = page["total"]
        cursor = page.get("next_cursor")          # read the field the API actually returns
        url = f"{BASE}/users?limit=200&cursor={cursor}" if cursor else None
    if len(users) != total:
        raise RuntimeError(f"incomplete population: got {len(users)} of {total}")
    return users

if __name__ == "__main__":
    run_id = uuid.uuid4().hex[:8]
    collected_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    raw = json.dumps({"collected_at": collected_at, "users": collect()}, sort_keys=True)
    os.makedirs("raw/identity", exist_ok=True)
    path = f"raw/identity/{run_id}.json"
    open(path, "w").write(raw)
    print(path, "sha256:" + hashlib.sha256(raw.encode()).hexdigest())

raw/identity/3fa91c0e.json sha256:9f2c41d07b...
# and on a broken cursor:
RuntimeError: incomplete population: got 200 of 1340

Common failure: the loop reads `next` while the API returns `next_cursor`, so it stops after page one and reports 200 users as the full population. The completeness check turns that silent truncation into a red run.

Step 3: Collector: cloud configuration

The second collector pulls storage and logging settings from your cloud provider's configuration API. The shape is the same as Step 2: authenticate with a read-only role, paginate, retry throttling, save raw output, hash it. Collect every region and every account in scope; a collector that only reads the default region produces a population that looks complete and is not.

1. Enumerate in-scope accounts and regions from a config file, not from memory.
2. For each, list storage buckets and fetch encryption and public-access settings.
3. Record which account and region every item came from.
4. Fail the run if any account or region returns an authorisation error rather than skipping it.

SCOPE = json.load(open("specs/scope.json"))   # {"accounts": [...], "regions": [...]}

buckets = []
for account in SCOPE["accounts"]:
    for region in SCOPE["regions"]:
        page_token = None
        while True:
            resp = list_buckets(account, region, page_token)   # read-only role
            for b in resp["buckets"]:
                buckets.append({
                    "id": b["name"], "account": account, "region": region,
                    "encrypted": b.get("encryption", {}).get("enabled"),
                    "public": b.get("public_access", {}).get("allowed"),
                })
            page_token = resp.get("next_page_token")
            if not page_token:
                break
print(f"collected {len(buckets)} buckets across "
      f"{len(SCOPE['accounts'])} accounts x {len(SCOPE['regions'])} regions")

collected 312 buckets across 4 accounts x 3 regions

Common failure: catching every exception and continuing. An account the role cannot read disappears from the population and the control looks greener. Skipped scope must fail the run.

Step 4: Collector: ticketing and change records

Change-management controls need two populations joined together: what was deployed, and what was approved. Pull change tickets for the period from the ticketing API, pull deployments from the CI system, and join them on the change id carried in the deployment metadata. The interesting output is the set of deployments with no matching approved ticket.

1. Query tickets by updated date covering the full period, not by created date, so late approvals are included.
2. Keep only tickets whose status is approved and whose approver is not the requester.
3. Join deployments to tickets with a left join so unmatched deployments survive as findings.

approved = {t["change_id"]: t for t in tickets
            if t["status"] == "approved" and t["approver"] != t["requester"]}

unmatched = [d for d in deployments if d.get("change_id") not in approved]
print(f"{len(deployments)} deployments, {len(unmatched)} without an approved change")

418 deployments, 3 without an approved change

Common failure: an inner join. Deployments with no ticket vanish from the result and the report shows 415 of 415 approved.

Step 5: Collectors: CI runs and the HR roster

Two more sources complete most access and change controls. Pipeline run records prove that required checks ran and were not skipped. The HR roster is the authoritative list of who should have access: compare leavers against the identity export to find accounts that are still active after termination.

1. Export pipeline runs for the period with job name, conclusion and commit; treat skipped and cancelled as not passed.
2. Load the HR leaver list as a CSV export with employee id and termination date.
3. Join leavers to identity accounts on employee id, never on display name.
4. Flag accounts still active more than 24 hours after the termination date.

from datetime import datetime, timedelta

active = {u["employee_id"]: u for u in users if u["status"] == "active"}
late = []
for row in csv.DictReader(open("raw/hr/leavers.csv")):
    term = datetime.fromisoformat(row["termination_date"])
    if row["employee_id"] in active and collected_at - term > timedelta(hours=24):
        late.append(row["employee_id"])
print(f"{len(late)} leavers still active after 24h")

2 leavers still active after 24h

Step 6: Normalise every source into one evidence record

Each source has its own shape. Policies should not care. Convert every raw export into the same record so one evaluator, one store and one report work for all controls. A record describes one subject at one moment: which control, which source, when it was collected, who or what it is about, the observed value, and the hash of the raw file it came from.

collected_at is copied from the collection step. It is never the time the report runs. raw_hash links the record back to the untouched file, so an auditor can verify nothing was edited in between.

def to_records(control_id, source, raw, raw_hash, items, subject_key, value_key):
    return [{
        "control_id": control_id,
        "source": source,
        "collected_at": raw["collected_at"],      # from collection, not now()
        "subject": f"{source}:{item[subject_key]}",
        "value": item.get(value_key),             # keep None when the field is missing
        "raw_hash": raw_hash,
    } for item in items]

{"control_id": "IAM-02", "source": "identity", "collected_at": "2026-10-24T02:00:07Z",
 "subject": "identity:bob", "value": false, "raw_hash": "sha256:9f2c41d07b..."}

Common failure: defaulting missing values to false or true during normalisation. Keep missing as missing; the policy must see it and decide.

Step 7: Writing policy code: the syntax

The evaluator is written in Rego, a declarative policy language widely used for policy as code. You do not need to know any other programming language to read it; you need to learn about ten constructs. Each one is shown below against the identity export.

A file starts with a package, which names the policy and becomes its address when you query it (data.evidence.iam.deny). `import rego.v1` turns on the current syntax, where rules are written `name contains value if { ... }`. The document being checked always arrives as `input`; you reach into it with dots, so `input.users` is the users array from the collector.

A rule has a head and a body. The head says what the rule produces; the body, inside braces, is a list of expressions. `deny contains msg if { ... }` means: add msg to the set called deny whenever every expression in the body is true. A rule can also produce a single value: `default compliant := false` sets a fallback, and `compliant if { count(deny) == 0 }` overrides it when the body holds.

Iteration is written `some u in input.users`: the body is tried once for every element, with u bound to that element. Assignment uses `:=` and creates a local variable; comparison uses `==`, `!=`, `<`, `>`. Strings are built with `sprintf("%s has no MFA", [u.id])`: the arguments go in a list. `count(x)` returns the length of an array, set or object.

Comprehensions build collections inline: `active := [u | some u in input.users; u.status == "active"]` is the list of active users. Set comprehensions use braces: `{u.id | some u in input.users}`. Helper rules with arguments are functions: `is_admin(u) if { "admin" in u.roles }`. Comments start with #. Name rules for what they mean (deny, violations, in_scope), not how they are computed.

package evidence.iam                 # address: data.evidence.iam

import rego.v1                       # current syntax: contains / if / in

# helper: the population the spec defined
active_users := [u | some u in input.users; u.status == "active"]

# function-style helper rule
is_admin(u) if "admin" in u.roles

# the set of violations, each with a reason
deny contains msg if {
	some u in active_users           # iterate
	not u.mfa_enabled                # condition (see semantics)
	msg := sprintf("%s: no MFA enrolled", [u.id])
}

# single-value rule with a default
default compliant := false

compliant if {
	count(active_users) > 0
	count(deny) == 0
}

opa eval -d policies/ -i normalised/identity.json "data.evidence.iam.deny" --format pretty

[
  "bob: no MFA enrolled",
  "cara: no MFA enrolled"
]

Step 8: Writing policy code: the semantics

Rego looks like an if-statement language and is not one. A rule body is a query: the engine searches for every combination of variable bindings that makes all expressions true, and the head is produced once per combination. There is no else, no return and no loop counter. Get these five rules into your head and most policy bugs disappear.

1. Lines in a body are ANDed. If any expression is false or undefined, that attempt produces nothing.
2. Two rule bodies with the same head are ORed. To deny for reason A or reason B, write two deny rules, never one rule with both lines.
3. A missing field is undefined, not false. `u.mfa_enabled == false` is undefined when the field is absent, so the body fails and the user is silently not denied.
4. `not expr` is true when expr is false or undefined. That is why `not u.mfa_enabled` denies both bob (false) and cara (missing). Use it whenever absence must count as failure.
5. An empty set is not an error. If input.users is empty, deny is empty and a naive compliant rule passes. Add an explicit rule that denies when count(input.users) == 0: the denominator check.

# WRONG: cara has no mfa_enabled field -> undefined -> not denied
deny contains msg if {
	some u in input.users
	u.mfa_enabled == false
	msg := sprintf("%s: no MFA", [u.id])
}

# RIGHT: false OR missing both deny
deny contains msg if {
	some u in input.users
	not u.mfa_enabled
	msg := sprintf("%s: no MFA", [u.id])
}

# RIGHT: refuse to pass on zero records (a second body = OR)
deny contains msg if {
	count(input.users) == 0
	msg := "identity export is empty: refusing to pass"
}

wrong policy  -> ["bob: no MFA"]
right policy  -> ["bob: no MFA", "cara: no MFA"]
empty input   -> ["identity export is empty: refusing to pass"]

Rule of thumb: every policy needs three tests: one input that passes, one that fails, and one with the decision field missing. If the missing-field test passes silently, your policy is wrong.

Step 9: Evaluate the pipeline and test the policies

Wire the evaluator to the normalised records and make the policies prove they can fail. Policy tests live beside the policy, are named test_*, and run in CI before the pipeline is allowed to use the policy.

package evidence.iam_test

import rego.v1
import data.evidence.iam

test_enrolled_user_passes if {
	count(iam.deny) == 0 with input as {"users": [{"id": "a", "status": "active", "mfa_enabled": true}]}
}

test_false_is_denied if {
	"b: no MFA enrolled" in iam.deny with input as {"users": [{"id": "b", "status": "active", "mfa_enabled": false}]}
}

test_missing_field_is_denied if {
	"c: no MFA enrolled" in iam.deny with input as {"users": [{"id": "c", "status": "active"}]}
}

test_empty_population_is_not_compliant if {
	not iam.compliant with input as {"users": []}
}

opa test policies/ -v

PASS: 4/4

Step 10: Store with custody

Evidence that can be edited after the fact is an opinion. Write the raw file, the normalised records and the evaluation result to append-only storage: object storage with a write-once retention lock, or a table the pipeline identity can insert into but not update or delete. Store a manifest per run listing each file and its SHA-256, and sign or hash the manifest itself.

1. Write raw/, records/ and results/ for the run under one run id.
2. Generate manifest.json with file paths, SHA-256 hashes, collected_at and the policy commit hash.
3. Upload to a bucket with retention lock set to the spec's retention_days (400 covers a 12-month audit period plus fieldwork).
4. Give the pipeline identity write-only access; nobody, including admins, gets delete during retention.

manifest = {
    "run_id": run_id,
    "control_id": "IAM-02",
    "collected_at": collected_at,
    "policy_commit": os.environ["GIT_SHA"],
    "files": {p: sha256_file(p) for p in sorted(glob.glob(f"out/{run_id}/**/*", recursive=True))},
}
open(f"out/{run_id}/manifest.json", "w").write(json.dumps(manifest, sort_keys=True, indent=2))

{"run_id": "3fa91c0e", "control_id": "IAM-02", "collected_at": "2026-10-24T02:00:07Z",
 "policy_commit": "a41be7c", "files": {"out/3fa91c0e/raw/identity.json": "9f2c41...", ...}}

Common failure: overwriting latest.json every night. The auditor asks for March and there is only October.

Step 11: Schedule and run it in CI

Run the whole pipeline as a scheduled CI job so every run is logged, versioned and reproducible. Secrets come from the CI secret store, the job fails loudly, and a re-run for the same date produces the same run, not a duplicate.

schedule: "0 2 * * *"          # daily 02:00 UTC
jobs:
  iam-02:
    steps:
      - run: opa test policies/ -v                 # policies must pass their own tests
      - run: python collectors/identity.py         # raises on incomplete population
        env: { IDP_READ_TOKEN: ${{ secrets.IDP_READ_TOKEN }} }
      - run: python normalise.py && opa eval -d policies/ -i out/records.json data.evidence.iam > out/result.json
      - run: python store.py                        # append-only, run id = date + control
      - run: python report.py --fail-on-deny        # non-zero exit when deny is not empty
    on_failure:
      notify: iam-team

Common failure: `|| true` on the evaluate step so the job is always green. A failing control must produce a failing run and a notification to the owner.

Step 12: Report and alert

The report is what the auditor and the control owner read. Per control and per period it shows: how many runs were expected, how many ran, population size per run, pass or fail per run, and the open violations with age. Gaps in the schedule are findings in their own right: a missing night is a night with no evidence.

python report.py --control IAM-02 --period 2026-10

IAM-02  period 2026-10-01..2026-10-31
runs expected 31  ran 30  missing: 2026-10-14
population  min 1318  max 1340
result      28 pass  2 fail
open        identity:bob   no MFA enrolled   age 3d   owner iam-team

Done means: an auditor can pick any day in the period, find the run, re-hash the raw file, re-run the policy at the recorded commit and get the same answer.

Answers are graded on the server: the answer key never reaches your browser.