#!/usr/bin/env python3 """Original PostgreSQL 17 row-policy exercise; Python 3.11+ and local Docker. Uses one cached postgres:17 image by its resolved image ID. No image pull, host port, application connection string or existing database is used. Synthetic LOGIN roles use the disposable container's local trust fixture; this is not a password/token verification exercise. Only the uniquely named owned container is removed. Keep its image identity, actual SQL, role attributes and results. Optional --session-reuse observes context lifetime on one physical connection; logical requests are fixtures, not an instantiated driver pool or verified users. """ import argparse import json import re import shutil import subprocess import sys import time import uuid IMAGE = "postgres:17" ROLES = ("tenant_alpha", "tenant_beta", "lab_owner", "lab_bypass") SETUP = """ CREATE ROLE lab_owner LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS; CREATE ROLE tenant_alpha LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS; CREATE ROLE tenant_beta LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS; CREATE ROLE lab_bypass LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE BYPASSRLS; CREATE SCHEMA lab AUTHORIZATION lab_owner; SET ROLE lab_owner; CREATE TABLE lab.documents (id integer PRIMARY KEY, tenant_id name NOT NULL, title text NOT NULL); INSERT INTO lab.documents VALUES (101, 'tenant_alpha', 'Alpha procedure'), (202, 'tenant_beta', 'Beta procedure'); GRANT USAGE ON SCHEMA lab TO tenant_alpha, tenant_beta, lab_bypass; GRANT SELECT, INSERT ON lab.documents TO tenant_alpha, tenant_beta, lab_bypass; RESET ROLE; """ POLICY = """CREATE POLICY tenant_scope ON lab.documents USING (tenant_id = current_user) WITH CHECK (tenant_id = current_user);""" ROWS = "SELECT json_build_object('session_user', session_user, 'current_user', current_user, 'rows', coalesce((SELECT json_agg(d ORDER BY d.id) FROM (SELECT id, tenant_id, title FROM lab.documents{where}) d), '[]'::json));" def command(args, *, timeout=30, input_text=None): return subprocess.run(args, input=input_text, capture_output=True, text=True, timeout=timeout, check=False) class Lab: def __init__(self): self.name = "fde-row-security-" + uuid.uuid4().hex[:16] self.created = False self.container_id = None self.image = None def sql(self, role, sql): if role not in ROLES + ("postgres",): raise ValueError("Unknown synthetic role") result = command(["docker", "exec", self.container_id, "psql", "-X", "-w", "-qAt", "--username", role, "--dbname", "postgres", "--set", "ON_ERROR_STOP=1", "--set", "VERBOSITY=sqlstate", "--command", sql]) error = re.search(r"(?:ERROR|FATAL):\s+([0-9A-Z]{5})\b", result.stderr) return {"role": role, "sql": sql, "exit_code": result.returncode, "sqlstate": error.group(1) if error else None, "output": result.stdout.strip(), "diagnostic": result.stderr.strip()} def admin(self, sql): result = self.sql("postgres", sql) if result["exit_code"]: raise RuntimeError("Scratch database setup failed: " + result["diagnostic"]) return result def read(self, role, where=""): result = self.sql(role, ROWS.format(where=where)) if result["exit_code"]: raise RuntimeError("Scratch read failed: " + result["diagnostic"]) observed = json.loads(result.pop("output")) result["rows"] = observed.pop("rows") result["identity"] = observed return result def start(self): if not shutil.which("docker"): raise RuntimeError("Install and start your approved local Docker runtime first") image = command(["docker", "image", "inspect", IMAGE]) if image.returncode: raise RuntimeError("Cache an approved postgres:17 image before running; this lab never pulls an image") self.image = json.loads(image.stdout)[0] result = command(["docker", "run", "--detach", "--rm", "--pull=never", "--network", "none", "--name", self.name, "--memory", "384m", "--cpus", "1", "--pids-limit", "96", "--tmpfs", "/var/lib/postgresql/data:rw,size=192m", "--env", "POSTGRES_HOST_AUTH_METHOD=trust", self.image["Id"]]) if result.returncode: raise RuntimeError("Could not start the isolated scratch PostgreSQL container") self.container_id = result.stdout.strip() self.created = True # A fixed readiness bound; no unbounded retry or changes to another container. for _ in range(60): ready = command(["docker", "exec", self.container_id, "pg_isready", "-h", "127.0.0.1", "-U", "postgres"], timeout=5) if ready.returncode == 0: # TCP readiness excludes the entrypoint's socket-only bootstrap server. check = self.sql("postgres", "SELECT current_setting('server_version_num');") if check["exit_code"] == 0: version = int(check["output"]) if not 170000 <= version < 180000: raise RuntimeError("This exercise requires PostgreSQL 17") return time.sleep(0.25) raise RuntimeError("Scratch PostgreSQL did not become ready within the bounded startup window") def remove(self): if not self.created: return True result = command(["docker", "rm", "--force", self.container_id]) if result.returncode: # --rm may already have removed a stopped owned container. result = command(["docker", "container", "ls", "--all", "--quiet", "--filter", "id=" + self.container_id]) return result.returncode == 0 and not result.stdout.strip() return True def run(requested_tenant): lab = Lab() report = None try: lab.start() lab.admin(SETUP) version = lab.admin("SELECT version();")["output"] inspected = command(["docker", "container", "inspect", lab.container_id]) if inspected.returncode: raise RuntimeError("Could not inspect the owned scratch container") container = json.loads(inspected.stdout)[0] attributes = lab.admin("""SELECT json_agg(r ORDER BY r.rolname) FROM (SELECT rolname, rolsuper, rolbypassrls, rolcreaterole, rolcreatedb FROM pg_roles WHERE rolname IN ('tenant_alpha','tenant_beta','lab_owner','lab_bypass')) r;""") cases = [] cases.append({"case": "ordinary-grant", "changes": [], "reads": [lab.read("tenant_alpha")]}) enable = "ALTER TABLE lab.documents ENABLE ROW LEVEL SECURITY;" lab.admin(enable) cases.append({"case": "default-deny", "changes": [enable], "reads": [lab.read("tenant_alpha")]}) lab.admin(POLICY) cases.append({"case": "scoped-roles", "changes": [POLICY], "reads": [lab.read("tenant_alpha"), lab.read("tenant_beta")]}) # Choice is an argparse enum, never an untrusted SQL fragment. It changes # the requested row filter, not the connection identity or policy. cases.append({"case": "requested-tenant", "requested_tenant": requested_tenant, "changes": [], "reads": [lab.read("tenant_alpha", " WHERE tenant_id = '" + requested_tenant + "'")]}) attempt = lab.sql("tenant_alpha", "INSERT INTO lab.documents VALUES (303, 'tenant_beta', 'Wrong-workspace write');") cases.append({"case": "write-check", "changes": [], "attempt": attempt, "reads": [lab.read("postgres")]}) cases.append({"case": "owner-bypass", "changes": [], "reads": [lab.read("lab_owner")]}) force = "ALTER TABLE lab.documents FORCE ROW LEVEL SECURITY;" lab.admin(force) cases.append({"case": "force-and-bypass", "changes": [force], "reads": [lab.read("lab_owner"), lab.read("lab_bypass")]}) broad = "CREATE POLICY extra_read ON lab.documents FOR SELECT TO tenant_alpha USING (true);" lab.admin(broad) cases.append({"case": "permissive-or", "changes": [broad], "reads": [lab.read("tenant_alpha"), lab.read("tenant_beta")]}) drop = "DROP POLICY extra_read ON lab.documents;" lab.admin(drop) cases.append({"case": "policy-restored", "changes": [drop], "reads": [lab.read("tenant_alpha")]}) attempt = lab.sql("tenant_alpha", "SET ROLE tenant_beta;") identity = lab.sql("tenant_alpha", "SELECT json_build_object('session_user', session_user, 'current_user', current_user);") if identity["exit_code"]: raise RuntimeError("Could not inspect the synthetic identity") identity["identity"] = json.loads(identity.pop("output")) cases.append({"case": "role-switch-denied", "changes": [], "attempt": attempt, "fresh_connection_identity": identity}) table = lab.admin("SELECT json_build_object('owner', pg_get_userbyid(relowner), 'row_security', relrowsecurity, 'force', relforcerowsecurity) FROM pg_class WHERE oid = 'lab.documents'::regclass;") report = {"scope": "Original synthetic PostgreSQL role-policy observations, not application authentication or production isolation acceptance.", "postgres_version": version, "image_id": lab.image["Id"], "image_repo_digests": lab.image.get("RepoDigests", []), "container_id": lab.container_id, "network_mode": container["HostConfig"]["NetworkMode"], "host_ports_published": bool(container["HostConfig"].get("PortBindings")), "final_table_control": json.loads(table["output"]), "identity_fixture": "Local trust login per explicitly selected role; no password or token verified.", "requested_tenant": requested_tenant, "setup_sql": SETUP, "role_attributes": json.loads(attributes["output"]), "cases": cases} finally: removed = lab.remove() if report is not None: report["owned_container_removed"] = removed if not removed: raise RuntimeError("Owned scratch container cleanup was not confirmed: " + lab.name) return report SESSION_SETUP = """ CREATE ROLE lab_owner NOLOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS; CREATE ROLE app_runtime LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS; CREATE SCHEMA lab AUTHORIZATION lab_owner; SET ROLE lab_owner; CREATE TABLE lab.documents (id integer PRIMARY KEY, tenant_id text NOT NULL); INSERT INTO lab.documents VALUES (101, 'tenant_alpha'), (202, 'tenant_beta'); ALTER TABLE lab.documents ENABLE ROW LEVEL SECURITY; ALTER TABLE lab.documents FORCE ROW LEVEL SECURITY; CREATE POLICY request_scope ON lab.documents FOR SELECT USING (tenant_id = nullif(current_setting('app.tenant', true), '')); GRANT USAGE ON SCHEMA lab TO app_runtime; GRANT SELECT ON lab.documents TO app_runtime; RESET ROLE; """ def session_reuse_steps(baseline_tenant): """Authored logical requests; observations come from one physical backend.""" baseline = "" if baseline_tenant == "unset" else "tenant_alpha" steps = [] def observe(case, step, commands, request_tenant=None): query = """SELECT json_build_object( 'case', '%s', 'step', '%s', 'backend_pid', pg_backend_pid(), 'session_user', session_user, 'current_user', current_user, 'context', nullif(current_setting('app.tenant', true), ''), 'rows', coalesce((SELECT json_agg(d ORDER BY d.id) FROM (SELECT id, tenant_id FROM lab.documents) d), '[]'::json));""" % (case, step) steps.append({"case": case, "step": step, "supplied_request_tenant": request_tenant, "commands": commands, "observation_sql": query}) observe("sticky-session", "sticky-alpha", "SET app.tenant = 'tenant_alpha';", "tenant_alpha") observe("sticky-session", "sticky-reused", "-- Next logical request omits its context setup.", "tenant_beta") observe("sticky-session", "sticky-cleared", "SET app.tenant = '';", "tenant_beta") observe("commit-restoration", "baseline-before", "SET app.tenant = '" + baseline + "';") observe("commit-restoration", "commit-local-beta", "BEGIN; SET LOCAL app.tenant = 'tenant_beta';", "tenant_beta") observe("commit-restoration", "after-commit", "COMMIT;") observe("rollback-restoration", "rollback-local-beta", "BEGIN; SET LOCAL app.tenant = 'tenant_beta';", "tenant_beta") observe("rollback-restoration", "after-rollback", "ROLLBACK;") observe("outside-transaction", "outside-local-ignored", "SET app.tenant = ''; SET LOCAL app.tenant = 'tenant_alpha';", "tenant_alpha") observe("caller-chosen-context", "caller-chooses-beta", "BEGIN; SET LOCAL app.tenant = 'tenant_beta';", "tenant_alpha") observe("caller-chosen-context", "after-caller-rollback", "ROLLBACK;") observe("clean-request-boundaries", "clean-alpha", "SET app.tenant = ''; BEGIN; SET LOCAL app.tenant = 'tenant_alpha';", "tenant_alpha") observe("clean-request-boundaries", "after-clean-commit", "COMMIT;") observe("clean-request-boundaries", "clean-beta", "BEGIN; SET LOCAL app.tenant = 'tenant_beta';", "tenant_beta") observe("clean-request-boundaries", "after-clean-rollback", "ROLLBACK;") return steps def run_session_reuse(baseline_tenant): lab = Lab() report = None try: lab.start() lab.admin(SESSION_SETUP) version = lab.admin("SELECT version();")["output"] inspected = command(["docker", "container", "inspect", lab.container_id]) if inspected.returncode: raise RuntimeError("Could not inspect the owned scratch container") container = json.loads(inspected.stdout)[0] attributes = lab.admin("SELECT row_to_json(r) FROM (SELECT rolname, rolsuper, rolbypassrls, rolcreaterole, rolcreatedb FROM pg_roles WHERE rolname = 'app_runtime') r;") table = lab.admin("SELECT json_build_object('owner', pg_get_userbyid(relowner), 'row_security', relrowsecurity, 'force', relforcerowsecurity) FROM pg_class WHERE oid = 'lab.documents'::regclass;") steps = session_reuse_steps(baseline_tenant) script = "\n".join(step["commands"] + "\n" + step["observation_sql"] for step in steps) # One psql process and one PostgreSQL connection for the complete script. result = command(["docker", "exec", "-i", lab.container_id, "psql", "-X", "-w", "-qAt", "--username", "app_runtime", "--dbname", "postgres", "--set", "ON_ERROR_STOP=1", "--set", "VERBOSITY=default"], input_text=script) if result.returncode: raise RuntimeError("Session-context experiment did not complete: " + result.stderr.strip()) observations = [json.loads(line) for line in result.stdout.splitlines() if line.strip()] if len(observations) != len(steps) or [row["step"] for row in observations] != [step["step"] for step in steps]: raise RuntimeError("Session-context observations did not match the requested sequence") report = {"scope": "Original synthetic PostgreSQL session-context observations. One physical connection, not a driver pool or authenticated application.", "mode": "session-reuse", "baseline_tenant": baseline_tenant, "postgres_version": version, "image_id": lab.image["Id"], "image_repo_digests": lab.image.get("RepoDigests", []), "container_id": lab.container_id, "network_mode": container["HostConfig"]["NetworkMode"], "host_ports_published": bool(container["HostConfig"].get("PortBindings")), "identity_fixture": "Local trust supplies app_runtime. Request labels are authored fixtures, not verified caller identities.", "runtime_attributes": json.loads(attributes["output"]), "table_control": json.loads(table["output"]), "setup_sql": SESSION_SETUP, "steps": steps, "sql_script": script, "observations": observations, "same_backend_connection": len({row["backend_pid"] for row in observations}) == 1, "diagnostic": result.stderr.strip(), "script_exit_code": result.returncode} finally: removed = lab.remove() if report is not None: report["owned_container_removed"] = removed if not removed: raise RuntimeError("Owned scratch container cleanup was not confirmed: " + lab.name) return report def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--requested-tenant", choices=("tenant_alpha", "tenant_beta")) parser.add_argument("--session-reuse", action="store_true", help="Observe transaction and session context on one physical PostgreSQL connection") parser.add_argument("--baseline-tenant", choices=("unset", "tenant_alpha"), help="Initial session baseline for --session-reuse only") args = parser.parse_args() if args.baseline_tenant is not None and not args.session_reuse: parser.error("--baseline-tenant requires --session-reuse") if args.session_reuse and args.requested_tenant is not None: parser.error("--requested-tenant belongs to the original role-policy mode") try: report = run_session_reuse(args.baseline_tenant or "unset") if args.session_reuse else run(args.requested_tenant or "tenant_beta") print(json.dumps(report, indent=2)) except (RuntimeError, subprocess.TimeoutExpired, OSError, ValueError) as error: print("Row-policy lab could not complete: " + str(error), file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())