#!/usr/bin/env python3 """Local PostgreSQL/PostgREST RLS evidence. Python 3 stdlib and Docker only. SPDX-License-Identifier: MIT """ import base64 import hashlib import hmac import json from pathlib import Path import secrets import subprocess import sys import time from datetime import datetime, timezone from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen ROOT = Path(__file__).resolve().parent IMAGES = { 'postgres': 'postgres:18.6-trixie@sha256:86c951e05bf56c93d95d397747fb8820ac76cc3bedb78f43abd83eedbe3666ae', 'postgrest': 'postgrest/postgrest:v14.14@sha256:d2009b5c9deffc210c8a5592698472fede14fd9f6ca89823c8474ca54d58c012', } ORG_A = 'a0000000-0000-0000-0000-000000000001' ORG_B = 'b0000000-0000-0000-0000-000000000001' USERS = { 'employee_a': '10000000-0000-0000-0000-000000000001', 'partner_a': '10000000-0000-0000-0000-000000000002', 'employee_b': '20000000-0000-0000-0000-000000000001', 'partner_b': '20000000-0000-0000-0000-000000000002', 'outsider': '30000000-0000-0000-0000-000000000001', } def docker(*args, data=None, check=True): # Never echo command lines: some contain ephemeral fixture credentials. result = subprocess.run(['docker', *args], input=data, text=True, capture_output=True, timeout=90) if check and result.returncode: raise RuntimeError(f'Docker {args[0]} failed: {result.stderr.strip()}') return result def wait_ready(probe): # Readiness polling is setup only. No assertion is retried or delayed. deadline = time.monotonic() + 60 while not probe(): if time.monotonic() >= deadline: raise RuntimeError('Fixture did not become ready within the startup guard') time.sleep(0.1) def b64(value): return base64.urlsafe_b64encode(value).decode().rstrip('=') def token(secret, user, **extra): claims = {'sub': USERS[user], 'role': 'authenticated', 'exp': int(time.time()) + 3600, **extra} message = b64(b'{"alg":"HS256","typ":"JWT"}') + '.' + b64(json.dumps(claims).encode()) return message + '.' + b64(hmac.new(secret.encode(), message.encode(), hashlib.sha256).digest()) def main(): if not __debug__: raise SystemExit('Run without -O or PYTHONOPTIMIZE: assertions must stay enabled') if len(sys.argv) > 1: raise SystemExit('Usage: python3 run.py > results.json (no remote target is accepted)') for image in IMAGES.values(): if docker('image', 'inspect', image, check=False).returncode: raise SystemExit(f'Image missing; pull this pinned public image first: {image}') name = 'obhut-rls-example-' + secrets.token_hex(6) network = None containers = [] checks = [] secret, password = secrets.token_hex(32), secrets.token_hex(24) fixture = (ROOT / 'fixture.sql').read_text() try: network = docker('network', 'create', name).stdout.strip() pg = docker('run', '--detach', '--rm', '--network', network, '--network-alias', 'database', '--name', name + '-db', '--tmpfs', '/var/lib/postgresql', '--env', 'POSTGRES_DB=rls_fixture', '--env', 'POSTGRES_PASSWORD=' + password, IMAGES['postgres']).stdout.strip() containers.append(pg) wait_ready(lambda: docker('exec', pg, 'pg_isready', '-h', '127.0.0.1', '-U', 'postgres', '-d', 'rls_fixture', check=False).returncode == 0) docker('exec', '-i', pg, 'psql', '-X', '-v', 'ON_ERROR_STOP=1', '-v', 'authenticator_password=' + password, '-U', 'postgres', '-d', 'rls_fixture', data=fixture) rest = docker('run', '--detach', '--rm', '--network', network, '--name', name + '-api', '--publish', '127.0.0.1::3000', '--env', f'PGRST_DB_URI=postgres://authenticator:{password}@database:5432/rls_fixture', '--env', 'PGRST_DB_SCHEMAS=api', '--env', 'PGRST_DB_ANON_ROLE=anon', '--env', 'PGRST_DB_CONFIG=false', '--env', 'PGRST_JWT_SECRET=' + secret, IMAGES['postgrest']).stdout.strip() containers.append(rest) address = docker('port', rest, '3000/tcp').stdout.strip() if not address.startswith('127.0.0.1:') or '\n' in address: raise RuntimeError('Fixture API must bind to a single loopback address') base = 'http://' + address tokens = {user: token(secret, user) for user in USERS} def request(method, path, who=None, body=None, headers=None): fields = {'Content-Type': 'application/json', 'Prefer': 'return=representation'} if who: fields['Authorization'] = 'Bearer ' + tokens.get(who, who) fields.update(headers or {}) req = Request(base + path, data=None if body is None else json.dumps(body).encode(), headers=fields, method=method) try: with urlopen(req, timeout=5) as response: return response.status, json.loads(response.read() or b'null') except HTTPError as error: return error.code, json.loads(error.read() or b'null') def ready(): try: return request('GET', '/rpc/request_context')[0] == 200 except (URLError, TimeoutError, ConnectionError): return False wait_ready(ready) def expect(label, method, path, who, status, expected=None, body=None, headers=None, code=None): actual_status, actual = request(method, path, who, body, headers) assert actual_status == status, (label, actual_status, actual) if expected is not None: assert actual == expected, (label, actual) if code is not None: assert actual.get('code') == code, (label, actual) checks.append({'case': label, 'http_status': actual_status, 'observed': actual}) return actual for who in (None, 'employee_a', 'partner_a', 'employee_b', 'partner_b'): context = expect(f'request role: {who or "anon"}', 'GET', '/rpc/request_context', who, 200) assert context['db_role'] == ('authenticated' if who else 'anon') assert context['subject'] == (USERS[who] if who else None) assert not context['superuser'] and not context['bypassrls'] assert context['table_owner'] != context['db_role'] assert context['rls_enabled'] and context['rls_forced'] and context['rls_active'] assert not context['can_change_org'] and not context['can_change_membership'] for who, ids in [('employee_a', ['a1', 'a2']), ('partner_a', ['a1']), ('employee_b', ['b1']), ('partner_b', ['b1']), ('outsider', [])]: expect('visible contacts: ' + who, 'GET', '/contacts?select=id&order=id', who, 200, [{'id': item} for item in ids]) expect('anonymous read denied by grants', 'GET', '/contacts', None, 401, code='42501') expect('invalid signature denied', 'GET', '/contacts', token(secrets.token_hex(32), 'employee_a'), 401) expect('expired JWT denied', 'GET', '/contacts', token(secret, 'employee_a', exp=1), 401) expect('privileged role cannot be assumed', 'GET', '/contacts', token(secret, 'employee_a', role='postgres'), 403, code='42501') expect('foreign organization read filtered', 'GET', '/contacts?id=eq.b1', 'employee_a', 200, []) expect('unassigned same-org contact filtered', 'GET', '/contacts?id=eq.a2', 'partner_a', 200, []) expect('organization header grants no rights', 'GET', '/contacts?id=eq.b1', 'employee_a', 200, [], headers={'X-Org-Id': ORG_B}) expect('user metadata grants no employee role', 'GET', '/contacts?select=id&order=id', token(secret, 'partner_a', user_metadata={'crm_role': 'employee', 'org_id': ORG_B}), 200, [{'id': 'a1'}]) expect('private schema is not exposed', 'GET', '/memberships', 'employee_a', 406, headers={'Accept-Profile': 'private'}, code='PGRST106') expect('employee update allowed', 'PATCH', '/contacts?id=eq.a1&select=id,name', 'employee_a', 200, [{'id': 'a1', 'name': 'Alpha updated'}], {'name': 'Alpha updated'}) expect('employee update persisted', 'GET', '/contacts?id=eq.a1&select=name', 'employee_a', 200, [{'name': 'Alpha updated'}]) expect('partner update filtered', 'PATCH', '/contacts?id=eq.a1', 'partner_a', 200, [], {'name': 'forbidden'}) expect('partner update left state unchanged', 'GET', '/contacts?id=eq.a1&select=name', 'employee_a', 200, [{'name': 'Alpha updated'}]) expect('cross-org update filtered', 'PATCH', '/contacts?id=eq.b1', 'employee_a', 200, [], {'name': 'forbidden'}) expect('cross-org update left state unchanged', 'GET', '/contacts?id=eq.b1&select=name', 'employee_b', 200, [{'name': 'Beta assigned'}]) for label, who, org in [('anonymous insert', None, ORG_A), ('partner insert', 'partner_a', ORG_A), ('cross-org insert', 'employee_a', ORG_B)]: expect(label + ' denied', 'POST', '/contacts', who, 401 if who is None else 403, body={'id': 'forbidden', 'org_id': org, 'name': 'forbidden'}, code='42501') for who in ('employee_a', 'employee_b'): expect('denied inserts absent: ' + who, 'GET', '/contacts?id=eq.forbidden', who, 200, []) expect('own-org insert allowed', 'POST', '/contacts?select=id', 'employee_a', 201, [{'id': 'a3'}], {'id': 'a3', 'org_id': ORG_A, 'name': 'Alpha new'}) expect('own-org insert persisted', 'GET', '/contacts?id=eq.a3&select=name', 'employee_a', 200, [{'name': 'Alpha new'}]) expect('tenant reassignment denied by column grant', 'PATCH', '/contacts?id=eq.a1', 'employee_a', 403, body={'org_id': ORG_B}, code='42501') expect('partner reassignment denied by column grant', 'PATCH', '/contacts?id=eq.a1', 'employee_a', 403, body={'partner_id': USERS['partner_b']}, code='42501') expect('tenant unchanged after reassignment attempt', 'GET', '/contacts?id=eq.a1&select=org_id,partner_id', 'employee_a', 200, [{'org_id': ORG_A, 'partner_id': USERS['partner_a']}]) expect('partner delete filtered', 'DELETE', '/contacts?id=eq.a1', 'partner_a', 200, []) expect('partner delete left contact present', 'GET', '/contacts?id=eq.a1&select=id', 'employee_a', 200, [{'id': 'a1'}]) expect('cross-org delete filtered', 'DELETE', '/contacts?id=eq.b1', 'employee_a', 200, []) expect('cross-org delete left contact present', 'GET', '/contacts?id=eq.b1&select=id', 'employee_b', 200, [{'id': 'b1'}]) expect('own-org delete allowed', 'DELETE', '/contacts?id=eq.a3&select=id', 'employee_a', 200, [{'id': 'a3'}]) expect('own-org delete persisted', 'GET', '/contacts?id=eq.a3', 'employee_a', 200, []) # Administrative fixture mutation models membership withdrawal, not a client capability. docker('exec', '-i', pg, 'psql', '-X', '-v', 'ON_ERROR_STOP=1', '-U', 'postgres', '-d', 'rls_fixture', data=f"delete from private.memberships where user_id = '{USERS['partner_a']}';") expect('withdrawn membership hides rows with unchanged JWT', 'GET', '/contacts', 'partner_a', 200, []) expect('employee still reads after partner withdrawal', 'GET', '/contacts?select=id&order=id', 'employee_a', 200, [{'id': 'a1'}, {'id': 'a2'}]) versions = { 'postgres': docker('exec', pg, 'psql', '-X', '-At', '-U', 'postgres', '-d', 'rls_fixture', '-c', 'show server_version;').stdout.strip(), 'postgrest': docker('exec', rest, 'postgrest', '--version').stdout.strip(), 'docker_server': docker('version', '--format', '{{.Server.Version}}').stdout.strip(), } result = {'executed_at': datetime.now(timezone.utc).isoformat(), 'scope': 'Local PostgreSQL + PostgREST; not hosted Supabase', 'images': IMAGES, 'versions': versions, 'source_sha256': {name: hashlib.sha256((ROOT / name).read_bytes()).hexdigest() for name in ('fixture.sql', 'run.py')}, 'passed': len(checks), 'checks': checks} finally: for container in reversed(containers): docker('rm', '--force', container, check=False) if network: docker('network', 'rm', network, check=False) print(json.dumps(result, indent=2)) if __name__ == '__main__': main()