1 min lesson

The four constraints they grade

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

The four constraints they gradeNon-negotiable for this role

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.

Pagination + rate limits

Okta and Google paginate; you must follow next links.

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.

Notice the safety asymmetry. 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, 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

def get_paged(url):
    while url:
        r = requests.get(url, headers=HEADERS, timeout=30)
        if r.status_code == 429:             # respect rate limits
            wait = int(r.headers.get("Retry-After", "5"))
            log.warning("rate limited, sleeping %ss", wait); time.sleep(wait); continue
        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):
    truth = active_hris_emails(hris_path)
    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 truth]

    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 = requests.post(f"{OKTA_URL}/api/v1/users/{u['id']}/lifecycle/deactivate",
                             headers=HEADERS, timeout=30)
        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)

The four constraints they grade

Step 1 of 2

The four constraints they gradeNon-negotiable for this role

Reconcile IdP against HRIS truth