RBAC Drift Detection & Compliance Sync

Python scripts for async batch privilege scraping

This page gives you copy-ready Python that scrapes the privilege state of many databases concurrently — without exhausting connection pools, blocking DDL, or emitting non-deterministic output that shows up downstream as phantom drift.

When to reach for async batch scraping

Async batching is the right tool in a narrow but common set of conditions. Use it when all three hold; pick something simpler otherwise.

  • Use it when you must enumerate grants across tens or hundreds of database instances on a schedule, and a serial synchronous sweep already causes scan-induced contention (saturated pools, catalog-lock waits, application query timeouts).
  • Use it when the extraction account is strictly read-only and every batch must be reproducible — a repeated run against an unchanged catalog has to produce byte-identical output so the diff engine sees zero drift.
  • Skip it when you are scraping a single small instance, or when downstream processing is the bottleneck rather than acquisition — in those cases the extra concurrency machinery buys nothing, and a plain psycopg loop is easier to audit.

This page is the concrete implementation of one technique within Async Privilege Batching; the acquisition rows it produces flow into Cross-Environment Privilege Extraction & Parsing for normalization.

Step-by-step implementation

The scraper has five moving parts: a read-only extraction role, a chunk planner, a bounded-concurrency fetcher, a retry wrapper for transient failures, and a deterministic payload emitter. Build them in order and verify each before moving on.

Bounded-concurrency async privilege scraper pipeline A chunk planner reads each environment's schema list and enqueues one fetch task per schema. All queued tasks funnel through an asyncio.Semaphore that admits at most three concurrent catalog reads, deliberately kept below the connection pool's ceiling of twenty so application traffic is never starved. Each admitted worker runs the catalog query inside a read-only, repeatable_read transaction wrapped in with_retry — up to three attempts with exponential backoff and jitter for transient errors, while permanent errors fall through. Every successful worker emits a payload whose sha256 checksum is computed over its sorted grant rows only. The payloads are collected, sorted deterministically by schema, and handed to the drift diff engine, which compares checksums across dated snapshots. Workers that fail permanently are logged with return_exceptions=True so a single dead host never aborts the whole sweep. acquire permit checksummed payload Chunk planner cheap schema-list query Queued tasks 1 schema = 1 fetch task app billing analytics reporting audit ⋯ + N more schemas asyncio .Semaphore max = 3 permit 1 permit 2 permit 3 ≤ 3 concurrent catalog reads below pool ceiling (max_size = 20) Worker · read-only fetch repeatable_read · readonly txn provably non-mutating snapshot with_retry ×3 · backoff + jitter Worker · read-only fetch repeatable_read · readonly txn provably non-mutating snapshot with_retry ×3 · backoff + jitter Worker · read-only fetch repeatable_read · readonly txn provably non-mutating snapshot with_retry ×3 · backoff + jitter Collect + sort sorted(key=schema) sha256 over rows only Drift diff engine compare dated checksums Permanent failures land in results via return_exceptions=True — logged and quarantined, the sweep continues.

1. Provision a read-only extraction role

The safety argument for the whole pipeline rests on the scraping principal holding no DDL and no GRANT. Create a dedicated role with catalog read access only. On PostgreSQL:

CREATE ROLE drift_scraper LOGIN PASSWORD 'REDACTED';
GRANT pg_read_all_stats TO drift_scraper;   -- catalog + stats visibility
GRANT pg_read_all_data  TO drift_scraper;   -- widens object-grant visibility (PG 14+)

On MySQL 8.0+, where the role graph lives in mysql.role_edges:

CREATE USER 'drift_scraper'@'%' IDENTIFIED BY 'REDACTED';
GRANT SELECT ON mysql.role_edges TO 'drift_scraper'@'%';
GRANT SELECT ON `information_schema`.* TO 'drift_scraper'@'%';

Verify the role cannot mutate state:

-- PostgreSQL: expect zero rows; the scraper must own no grantable privileges.
SELECT grantee, table_schema, privilege_type
FROM information_schema.role_table_grants
WHERE grantee = 'drift_scraper' AND is_grantable = 'YES';

2. Partition each environment into non-overlapping chunks

A chunk is the smallest unit the concurrency limiter schedules. Chunking by schema keeps each query bounded, lets a single failed chunk retry without re-reading the whole catalog, and makes batch identity stable. The planner reads only the schema list — a cheap query — and skips transient namespaces so they never pollute a canonical baseline:

SELECT n.nspname AS schema_name
FROM pg_catalog.pg_namespace n
WHERE n.nspname NOT LIKE 'pg_temp_%'
  AND n.nspname NOT LIKE 'pg_toast_temp_%'
  AND n.nspname NOT IN ('pg_catalog', 'information_schema')
ORDER BY n.nspname;

The exact shape of the per-chunk query — keyset pagination, index-friendly predicates, avoiding OFFSET — is governed by System Catalog Query Optimization; the scraper simply dispatches whatever tuned query that work produces.

3. Fetch chunks with bounded concurrency

asyncio.Semaphore caps how many catalog reads run at once, kept below the pool ceiling so application traffic is never starved. Each fetch runs inside a repeatable_read, readonly transaction so the snapshot is internally consistent and provably non-mutating:

import asyncio
import asyncpg
import hashlib
import json
from datetime import datetime, timezone

CATALOG_QUERY = """
    SELECT grantee, table_schema, table_name, privilege_type, is_grantable
    FROM information_schema.role_table_grants
    WHERE table_schema = $1
      AND table_schema NOT IN ('pg_catalog', 'information_schema')
    ORDER BY grantee, table_name, privilege_type
"""

async def fetch_schema_grants(pool: asyncpg.Pool, schema: str, batch_id: str) -> dict:
    """Fetch all grants for one schema inside a read-only transaction."""
    async with pool.acquire() as conn:
        async with conn.transaction(isolation="repeatable_read", readonly=True):
            rows = await conn.fetch(CATALOG_QUERY, schema)
    records = [dict(r) for r in rows]
    return {
        "batch_id": batch_id,
        "schema": schema,
        "extracted_at": datetime.now(timezone.utc).isoformat(),
        "grants": records,
        "checksum": hashlib.sha256(
            json.dumps(records, sort_keys=True, default=str).encode()
        ).hexdigest(),
    }

async def scrape_privileges(dsn: str, schemas: list[str], max_concurrency: int = 5) -> list[dict]:
    """Scrape grants for every schema with bounded concurrency.
    Returns a deterministically ordered list of batch payloads."""
    semaphore = asyncio.Semaphore(max_concurrency)
    async with asyncpg.create_pool(dsn, min_size=2, max_size=max_concurrency + 2) as pool:
        async def bounded_fetch(schema: str) -> dict:
            batch_id = hashlib.sha256(
                f"{schema}:{datetime.now(timezone.utc).date()}".encode()
            ).hexdigest()[:16]
            async with semaphore:
                return await fetch_schema_grants(pool, schema, batch_id)

        tasks = [asyncio.create_task(bounded_fetch(s)) for s in schemas]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    payloads = []
    for schema, result in zip(schemas, results):
        if isinstance(result, Exception):
            print(f"ERROR schema={schema}: {result}")  # log, don't abort the run
        else:
            payloads.append(result)
    return sorted(payloads, key=lambda p: p["schema"])

Run it and confirm concurrency stays capped:

payloads = asyncio.run(scrape_privileges(dsn, ["app", "billing", "analytics"], max_concurrency=3))
# Expected: len(payloads) == number of reachable schemas; never more than
# max_concurrency connections active at once (watch pg_stat_activity).

4. Absorb transient failures with bounded retry

Catalog locks, connection resets, and network timeouts are transient and should retry with exponential backoff plus jitter; invalid credentials or missing views are permanent and must fail fast. Wrap only the fetch coroutine, and cap attempts so a dead host never blocks the event loop indefinitely:

import asyncio
import random

async def with_retry(coro_fn, max_attempts: int = 3, base_delay: float = 1.0):
    """Retry a coroutine with exponential backoff and jitter on transient errors."""
    for attempt in range(1, max_attempts + 1):
        try:
            return await coro_fn()
        except (asyncio.TimeoutError, OSError) as exc:   # transient class only
            if attempt == max_attempts:
                raise
            delay = base_delay * (2 ** (attempt - 1)) + random.uniform(0, 0.5)
            await asyncio.sleep(delay)

To use it, wrap the fetch call inside bounded_fetch: return await with_retry(lambda: fetch_schema_grants(pool, schema, batch_id)). Permanent errors (an asyncpg.InvalidPasswordError, a missing information_schema view) fall straight through and land in the results list as exceptions for quarantine.

5. Emit a deterministic, checksummed payload for diffing

Each payload already carries a sha256 checksum over its sorted grant rows. That checksum is the contract with the diff engine: identical catalog state yields an identical checksum, so a no-op run is provably a no-op. Verify determinism by scraping the same schema twice:

a = asyncio.run(scrape_privileges(dsn, ["billing"], 1))[0]["checksum"]
b = asyncio.run(scrape_privileges(dsn, ["billing"], 1))[0]["checksum"]
assert a == b, "non-deterministic extraction — check ORDER BY and timestamp keys"

From here the batches leave the acquisition stage: vendor-specific grant strings are translated by Cross-DB Parser Adapters, and structural integrity is enforced by Schema Validation Pipelines before anything is persisted.

Worked example: three PostgreSQL 15 environments, one nightly sweep

Consider a topology of three PostgreSQL 15 instances — prod, staging, and dev — each holding the schemas app, billing, and analytics, scraped nightly by the drift_scraper role from step 1. The orchestration layer iterates environments and calls scrape_privileges per DSN with max_concurrency=3 (well under each instance’s 20-connection pool):

ENVIRONMENTS = {
    "prod":    "postgresql://drift_scraper@prod-db/appdb",
    "staging": "postgresql://drift_scraper@stg-db/appdb",
    "dev":     "postgresql://drift_scraper@dev-db/appdb",
}
SCHEMAS = ["app", "billing", "analytics"]

async def sweep() -> dict:
    out = {}
    for env, dsn in ENVIRONMENTS.items():
        out[env] = await scrape_privileges(dsn, SCHEMAS, max_concurrency=3)
    return out

snapshot = asyncio.run(sweep())

A representative snapshot["prod"][1] payload:

{
  "batch_id": "9f2c1a7b4e6d0c83",
  "schema": "billing",
  "extracted_at": "2026-07-04T02:00:07.412903+00:00",
  "grants": [
    {"grantee": "billing_ro", "table_schema": "billing",
     "table_name": "invoices", "privilege_type": "SELECT", "is_grantable": "NO"},
    {"grantee": "ci_service", "table_schema": "billing",
     "table_name": "invoices", "privilege_type": "SELECT", "is_grantable": "NO"}
  ],
  "checksum": "b1946ac92492d2347c6235b4d2611184..."
}

The nightly job persists each payload keyed by (env, schema). When the diff engine compares tonight’s checksum for prod/billing against last night’s, an unchanged catalog produces an identical hash and zero drift; the addition of a ci_service INSERT grant flips the checksum, and the row-level diff surfaces exactly that one added privilege — the granularity auditors expect from an evidence trail.

Gotchas and engine-specific notes

PostgreSQL vs MySQL role catalogs. PostgreSQL resolves object grants through information_schema.role_table_grants and role membership through pg_catalog.pg_auth_members. MySQL 8.0 stores role edges in mysql.role_edges, which does not exist before 8.0 — a scraper pointed at MySQL 5.7 must fall back to mysql.user and the *_priv columns, or it will raise a missing-table error that the retry wrapper will (correctly) treat as permanent.

Never wrap a sync driver in an executor. Calling loop.run_in_executor around psycopg2 re-introduces the thread-pool contention async is meant to remove and silently breaks backpressure. Use a genuinely async driver: asyncpg for PostgreSQL, aiomysql or asyncmy for MySQL. Any blocking call inside a coroutine stalls the whole event loop.

Timestamps must stay out of the identity key. extracted_at is provenance metadata, not part of what the checksum hashes — the checksum is computed over records only. If a timestamp leaks into the hashed set, every run drifts against the last and the diff engine drowns in false positives.

Keyset over OFFSET on large grant tables. On estates with very large role_table_grants, OFFSET-based pagination forces repeated sequential scans and lengthens catalog-lock windows. Page with WHERE oid > $last_seen_oid instead; see System Catalog Query Optimization for the full pattern.

Compliance note

Read-only async scraping is the evidence-collection mechanism for access-review controls: SOC 2 CC6.1 and CC6.3 (logical access is provisioned and reviewed against least privilege), PCI-DSS Requirement 7 (need-to-know restriction), and HIPAA §164.312(a)(1) (access control). The artifact it produces for auditors is a timestamped, checksummed per-environment privilege payload — the batch_id and sha256 fields give each snapshot a tamper-evident identity, and the diff between two dated snapshots is the periodic access-review record itself. Because the extraction principal holds no DDL or GRANT, the collection process cannot alter the state it attests to, which is the property an auditor checks first.

Frequently asked questions

Why bound concurrency with a semaphore instead of just launching every task? Unbounded asyncio.gather opens one connection per schema per host and instantly exhausts the pool, which is the exact scan-induced contention this technique exists to prevent. The asyncio.Semaphore caps active catalog reads below the pool ceiling so application queries always have headroom.

How do I run this as a dry run against production? It already is one. Every fetch executes inside a readonly=True transaction and the role from step 1 holds no write privileges, so there is no DDL or DML path. To confirm before a first run, execute the verification query in step 1 and expect zero grantable rows.

What makes the output safe to diff? Determinism. The ORDER BY in the catalog query fixes row order, json.dumps(..., sort_keys=True) fixes key order, and the timestamp is excluded from the hashed set — so an unchanged catalog yields an identical checksum on every run.

Does this work for MySQL and Oracle too? The concurrency and retry structure is engine-agnostic; only the driver and catalog query change. Swap asyncpg for aiomysql and query mysql.role_edges on MySQL 8.0+, or DBA_TAB_PRIVS / DBA_ROLE_PRIVS on Oracle. The per-dialect grant strings are reconciled downstream by the Cross-DB Parser Adapters.

Up: Cross-Environment Privilege Extraction & Parsing