Skip to lesson
Exit
Capstone: Mock Loop & Self-Exam1 / 2

2 min lesson

The safeguards they grade

Work through the cases in "The safeguards they grade", pairing each signal with the move that fits.

Step 1 of 2

The safeguards they gradeBe ready to defend every one

Idempotent + re-runnable

Running it twice changes nothing the second time.

Already-deactivated users are a no-op, not an error.

State lives in the systems, not in a local file you have to keep.

Dry-run by default

Default mode prints what it would do and touches nothing.

A --apply flag is required to make a change.

Output is a diff a human can review before committing.

Blast-radius guard

Abort when the diff would deactivate more than a set fraction of the directory.

A big diff usually means a bad export, not a mass of leavers.

Surface the anomaly for a human to check before any change lands.

Pagination + rate limits

Okta returns a Link header with rel="next"; other APIs hand back a page token. Follow whichever the API gives.

Honor 429s with backoff and respect Retry-After.

Never assume page one is the whole directory.

Secrets + logging

Token from env or a secret manager, never hard-coded.

Structured logs of every decision, with the reason.

Logs are your audit evidence for a SOC 2 access review.

The two mistakes have lopsided costs. A missed deactivation leaves a stale account for a day until the next run catches it. A false deactivation locks an active employee out of everything mid-sprint. So your default has to fail toward not acting: dry-run unless told otherwise and a guardrail that refuses to deactivate more than a sane fraction of the directory in one pass.

Learn more

Full explanation

Reconcile IdP against HRIS truth

Reconcile IdP against HRIS truth: idempotent, dry-run default, paginated, protected accounts, with a blast-radius guardts
import os, sys, time, logging, requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("jml-reconcile")

OKTA_URL = os.environ["OKTA_URL"]            # never hard-code
TOKEN = os.environ["OKTA_API_TOKEN"]         # from env / secret manager
HEADERS = {"Authorization": f"SSWS {TOKEN}", "Accept": "application/json"}
MAX_DEACTIVATE_RATIO = 0.10                  # refuse a runaway purge
PROTECTED = {"svc-sync@example.com", "breakglass@example.com"}   # service accounts, never deactivated

def call(method, url):
    while True:
        r = requests.request(method, url, headers=HEADERS, timeout=30)
        if r.status_code != 429:
            return r
        wait = int(r.headers.get("Retry-After", "5"))   # respect rate limits on reads and writes
        log.warning("rate limited, sleeping %ss", wait); time.sleep(wait)

def get_paged(url):
    while url:
        r = call("GET", url)
        r.raise_for_status()
        yield from r.json()
        url = r.links.get("next", {}).get("url")   # follow pagination

def active_hris_emails(path):
    import csv
    with open(path) as f:
        return {row["work_email"].lower() for row in csv.DictReader(f)
                if row["status"].lower() == "active"}

def reconcile(hris_path, apply=False):
    keep = active_hris_emails(hris_path) | PROTECTED
    idp_active = [u for u in get_paged(f"{OKTA_URL}/api/v1/users?filter=status eq \"ACTIVE\"")]
    orphans = [u for u in idp_active if u["profile"]["email"].lower() not in keep]

    if len(orphans) > len(idp_active) * MAX_DEACTIVATE_RATIO:
        log.error("guard tripped: %d/%d would deactivate, aborting; check the HRIS export",
                  len(orphans), len(idp_active)); sys.exit(2)

    for u in orphans:
        email = u["profile"]["email"]
        if not apply:
            log.info("DRY-RUN would deactivate %s (no active HRIS record)", email); continue
        resp = call("POST", f"{OKTA_URL}/api/v1/users/{u['id']}/lifecycle/deactivate")
        if resp.status_code in (200, 404):   # 404 = already gone => idempotent no-op
            log.info("deactivated %s", email)
        else:
            log.error("FAILED %s: %s %s", email, resp.status_code, resp.text)

    log.info("done: %d orphans, apply=%s", len(orphans), apply)

if __name__ == "__main__":
    reconcile(sys.argv[1], apply="--apply" in sys.argv)