LEARN. PRACTICE. GROW.Design a matrix the server can enforce
Start with three concrete questions: who may invoke the function, on which objects, and with which properties? In the fictional example, an analyst reads instructions in their tenant while a manager may change amounts within agreed bounds. Neither changes approval through that workflow. Write these cases before discussing role names or buttons. Use identity context established by the server and bind the query to the authorized tenant. A client-supplied field cannot select another scope. Knowing an ID also does not grant access to the corresponding object.
Constrain inputs and representations
An explicit write schema reduces the surface for unauthorized assignment. The lab operation accepts exactly amount, requires an integer from zero through ten thousand, and rejects approved, tenant, and additional properties. Using type(value) is int is intentional: in Python, a boolean can pass an overly permissive integer check. Reads return only id, amount, and version. The internal record contains more fields, but serialization does not copy them automatically. Documenting readOnly may help consumers; evidence of protection comes from server behavior and persisted state.
Evaluate batches as complete operations
In this exercise’s contract, a batch is all or nothing. If A1 is allowed and B1 belongs to another tenant, neither change may be committed. The code checks every object inside the transaction before updating them. This also avoids reporting full success after silently skipping a forbidden item. The maximum of two objects is a teaching limit rather than a universal capacity recommendation. In production, define body size, operations per batch, time, and resource consumption from requirements. A requests-per-minute limit does not constrain the cost of one unrestricted request.
Run the matrix in the laboratory
Save this lesson’s full code as run.py and execute python3 run.py --output evidence.json in a temporary directory with Python 3.13. The program creates fictional data and starts HTTP only on 127.0.0.1 at an available port. X-Lab-Actor selects known synthetic identities; it does not authenticate users. Observe authorized reading, manager writing, analyst rejection, another tenant’s object, a protected field, a boolean, an excessive amount, and a mixed batch. After negative cases, the program queries state to demonstrate absence of changes. The temporary file contains no BigSavant catalogue data.
Investigate an alternate path
In a fictional funds case, individual reads work correctly but a new export omits the tenant condition. The incident can remain hidden if acceptance always uses a global manager. Suspend the affected path, preserve evidence with restricted access, and determine involved objects and recipients. Review the export query, response projection, and function permissions. Sharing authentication middleware does not establish equivalent authorization. Before reopening, combine a legitimate case with identities from another tenant and roles without grants, also checking persisted effects. Assign responsibility for reviewing the extent of prior exposure.
Give APS observable acceptance criteria
Useful evidence connects principal, action, object, properties, expected outcome, and observed outcome. Keep real data and credentials outside the lab. For RUN handover, include policy owners, revocation cases, export paths, and a way to diagnose denials without revealing secrets. The local exercise does not cover an identity provider, TLS, global concurrency limits, or resilience against hostile clients. The public demonstration key and synthetic headers must never be promoted into real credentials. Use the laboratory to learn and prepare criteria for an authorized environment with appropriate operational controls.
"""Original loopback authorization/webhook laboratory. Reference: Python 3.13.1.
Run: python3 run.py --output evidence.json
Uses a temporary SQLite database and TWO child HTTP server processes.
X-Lab-Actor is synthetic identity, not authentication. Signature format and
version ordering are a fictional contract, not a Stripe SDK implementation.
No TLS, external API, actual payment, distributed transaction or power failure.
"""
import argparse
from concurrent.futures import ThreadPoolExecutor
from contextlib import closing
import hashlib
import hmac
import http.client
import json
from pathlib import Path
import platform
import sqlite3
import subprocess
import sys
import tempfile
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
SECRET = b'public-fixture-key-not-a-production-secret'
NOW = 1000
ACTORS = {'a-viewer': ('A', 'viewer'), 'a-manager': ('A', 'manager'), 'b-manager': ('B', 'manager')}
def encode(value):
return json.dumps(value, sort_keys=True, separators=(',', ':')).encode
def signature(body, timestamp=NOW, secret=SECRET):
return hmac.new(secret, str(timestamp).encode + b'.' + body, hashlib.sha256).hexdigest
def connect(database):
db = sqlite3.connect(database, timeout=5, isolation_level=None)
db.row_factory = sqlite3.Row
return db
def initialize(database):
with closing(connect(database)) as db:
db.executescript('''
CREATE TABLE IF NOT EXISTS objects(id TEXT PRIMARY KEY, tenant TEXT, amount INTEGER, version INTEGER, approved INTEGER, internal_note TEXT);
CREATE TABLE IF NOT EXISTS inbox(event_id TEXT PRIMARY KEY, body_hash TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS outbox(event_id TEXT PRIMARY KEY, resource_id TEXT NOT NULL, version INTEGER NOT NULL);
INSERT OR IGNORE INTO objects VALUES('A1','A',100,1,0,'fictional-internal-note');
INSERT OR IGNORE INTO objects VALUES('B1','B',900,1,0,'fictional-other-note');
''')
def handler_for(database):
class Handler(BaseHTTPRequestHandler):
protocol_version = 'HTTP/1.1'
def log_message(self, *args):
pass
def reply(self, status, value):
body = encode(value)
self.send_response(status)
self.send_header('Content-Type', 'application/json')
self.send_header('Content-Length', str(len(body)))
self.send_header('Cache-Control', 'no-store')
self.end_headers
self.wfile.write(body)
def context(self):
return ACTORS.get(self.headers.get('X-Lab-Actor', ''))
def body(self):
try:
length = int(self.headers.get('Content-Length', '0'))
except ValueError:
self.reply(400, {'error': 'invalid-length'})
return None
if length < 0 or length > 4096:
self.close_connection = True
self.reply(413, {'error': 'body-limit'})
return None
return self.rfile.read(length)
def do_GET(self):
actor = self.context
if actor is None:
return self.reply(401, {'error': 'synthetic-identity-missing'})
if not self.path.startswith('/objects/'):
return self.reply(404, {'error': 'not-found'})
with closing(connect(database)) as db:
row = db.execute('SELECT * FROM objects WHERE id=? AND tenant=?', (self.path.split('/')[-1], actor[0])).fetchone
if row is None:
return self.reply(404, {'error': 'not-found'})
return self.reply(200, {k: row[k] for k in ('id', 'amount', 'version')})
def do_POST(self):
raw = self.body
if raw is None:
return
if self.path == '/webhook':
return self.webhook(raw)
actor = self.context
if actor is None:
return self.reply(401, {'error': 'synthetic-identity-missing'})
if actor[1]!= 'manager':
return self.reply(403, {'error': 'function-denied'})
try:
payload = json.loads(raw)
except (ValueError, UnicodeError):
return self.reply(400, {'error': 'invalid-json'})
if not isinstance(payload, dict):
return self.reply(400, {'error': 'object-required'})
if self.path.startswith('/objects/'):
if set(payload)!= {'amount'} or type(payload['amount']) is not int or not 0 <= payload['amount'] <= 10000:
return self.reply(400, {'error': 'write-schema'})
ids = [self.path.split('/')[-1]]
elif self.path == '/bulk':
if set(payload)!= {'ids', 'amount'} or type(payload['amount']) is not int or not 0 <= payload['amount'] <= 10000 or not isinstance(payload['ids'], list) or not 1 <= len(payload['ids']) <= 2 or any(not isinstance(x, str) for x in payload['ids']):
return self.reply(400, {'error': 'bulk-schema'})
ids = payload['ids']
else:
return self.reply(404, {'error': 'not-found'})
db = connect(database)
try:
db.execute('BEGIN IMMEDIATE')
for id_ in ids:
if db.execute('SELECT 1 FROM objects WHERE id=? AND tenant=?', (id_, actor[0])).fetchone is None:
db.rollback
return self.reply(404, {'error': 'not-found'})
for id_ in ids:
db.execute('UPDATE objects SET amount=? WHERE id=?', (payload['amount'], id_))
db.commit
return self.reply(200, {'updated': len(ids)})
finally:
db.close
def webhook(self, raw):
try:
timestamp = int(self.headers.get('X-Lab-Time', ''))
supplied = self.headers.get('X-Lab-Signature', '')
# Explicit fixture window, not a statement of a provider default.
valid = abs(NOW - timestamp) <= 300 and hmac.compare_digest(signature(raw, timestamp), supplied)
except (ValueError, TypeError):
valid = False
if not valid:
return self.reply(400, {'error': 'signature-or-time'})
try:
event = json.loads(raw)
except (ValueError, UnicodeError):
return self.reply(400, {'error': 'invalid-json'})
required = {'event_id', 'tenant', 'resource_id', 'version', 'amount'}
if not isinstance(event, dict) or set(event)!= required or any(not isinstance(event[k], str) or not event[k] for k in ('event_id', 'tenant', 'resource_id')) or type(event['version']) is not int or event['version'] < 1 or type(event['amount']) is not int or not 0 <= event['amount'] <= 10000:
return self.reply(400, {'error': 'event-schema'})
if event['tenant']!= 'A':
return self.reply(403, {'error': 'producer-scope'})
digest = hashlib.sha256(raw).hexdigest
db = connect(database)
try:
db.execute('BEGIN IMMEDIATE')
row = db.execute('SELECT * FROM objects WHERE id=? AND tenant=?', (event['resource_id'], event['tenant'])).fetchone
if row is None:
db.rollback
return self.reply(404, {'error': 'not-found'})
prior = db.execute('SELECT body_hash FROM inbox WHERE event_id=?', (event['event_id'],)).fetchone
if prior is not None:
db.rollback
return self.reply(200 if prior['body_hash'] == digest else 409, {'result': 'duplicate' if prior['body_hash'] == digest else 'id-conflict'})
db.execute('INSERT INTO inbox VALUES(?,?)', (event['event_id'], digest))
advanced = event['version'] > row['version']
if advanced:
db.execute('UPDATE objects SET amount=?,version=? WHERE id=?', (event['amount'], event['version'], event['resource_id']))
db.execute('INSERT INTO outbox VALUES(?,?,?)', (event['event_id'], event['resource_id'], event['version']))
# Deliberate test-only fault injection, before the shared commit.
if self.headers.get('X-Lab-Fail') == 'before-commit':
db.rollback
return self.reply(503, {'error': 'injected-before-commit'})
db.commit
return self.reply(200, {'result': 'accepted' if advanced else 'recorded-stale'})
finally:
db.close
return Handler
def serve(database, ready):
initialize(database)
server = ThreadingHTTPServer(('127.0.0.1', 0), handler_for(database))
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start
Path(ready).write_text(json.dumps({'port': server.server_port}))
try:
sys.stdin.readline
finally:
server.shutdown
server.server_close
thread.join(timeout=3)
def run:
checks = []
child = None
with tempfile.TemporaryDirectory(prefix='dr-rest-webhooks-') as tmp:
database = Path(tmp) / 'fixture.sqlite'
def start(index):
ready = Path(tmp) / f'ready-{index}.json'
process = subprocess.Popen([sys.executable, str(Path(__file__).resolve), '--serve', str(database), '--ready', str(ready)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
for _ in range(100):
if ready.exists:
return process, json.loads(ready.read_text)['port']
if process.poll is not None:
raise RuntimeError(process.communicate[1])
time.sleep(.02)
process.terminate
process.communicate(timeout=5)
raise TimeoutError('fixture readiness')
def stop(process):
_, err = process.communicate('\n', timeout=5)
if process.returncode!= 0:
raise RuntimeError(err)
def request(method, path, payload=None, headers=None, raw=None):
client = http.client.HTTPConnection('127.0.0.1', port, timeout=5)
try:
data = raw if raw is not None else (encode(payload) if payload is not None else None)
client.request(method, path, body=data, headers=headers or {})
response = client.getresponse
return response.status, json.loads(response.read)
finally:
client.close
def actor_request(method, id_, actor, payload=None):
return request(method, '/objects/' + id_, payload, {'X-Lab-Actor': actor})
def webhook(event, timestamp=NOW, **extra):
raw = encode(event)
return request('POST', '/webhook', headers={'X-Lab-Time': str(timestamp), 'X-Lab-Signature': signature(raw, timestamp), **extra}, raw=raw)
def check(name, actual, expected):
if actual!= expected:
raise AssertionError((name, actual, expected))
checks.append({'name': name, 'actual': actual, 'expected': expected, 'passed': True})
def state:
with closing(connect(database)) as db:
return {'amount': db.execute("SELECT amount FROM objects WHERE id='A1'").fetchone[0], 'version': db.execute("SELECT version FROM objects WHERE id='A1'").fetchone[0], 'inbox': db.execute('SELECT count(*) FROM inbox').fetchone[0], 'outbox': db.execute('SELECT count(*) FROM outbox').fetchone[0]}
try:
child, port = start(1)
check('missing synthetic identity', request('GET', '/objects/A1')[0], 401)
status, body = actor_request('GET', 'A1', 'a-viewer')
check('authorized read and response projection', [status, sorted(body)], [200, ['amount', 'id', 'version']])
check('other tenant object denied', actor_request('GET', 'B1', 'a-manager')[0], 404)
check('viewer cannot invoke write', actor_request('POST', 'A1', 'a-viewer', {'amount': 200})[0], 403)
check('manager cannot change another tenant', actor_request('POST', 'B1', 'a-manager', {'amount': 200})[0], 404)
check('protected property rejected', actor_request('POST', 'A1', 'a-manager', {'amount': 200, 'approved': True})[0], 400)
check('client tenant override rejected', actor_request('POST', 'A1', 'a-manager', {'amount': 200, 'tenant': 'B'})[0], 400)
check('boolean is not an integer amount', actor_request('POST', 'A1', 'a-manager', {'amount': True})[0], 400)
check('out of bounds amount rejected', actor_request('POST', 'A1', 'a-manager', {'amount': 10001})[0], 400)
check('negative cases left amount unchanged', state['amount'], 100)
check('approved scalar update', actor_request('POST', 'A1', 'a-manager', {'amount': 150})[0], 200)
check('mixed-tenant bulk rejected', request('POST', '/bulk', {'ids': ['A1', 'B1'], 'amount': 999}, {'X-Lab-Actor': 'a-manager'})[0], 404)
check('bulk rejection leaves first item unchanged', state['amount'], 150)
check('body limit enforced', request('POST', '/webhook', raw=b'x' * 4097)[0], 413)
event = {'event_id': 'E1', 'tenant': 'A', 'resource_id': 'A1', 'version': 2, 'amount': 250}
raw = encode(event)
check('unsigned event rejected', request('POST', '/webhook', raw=raw)[0], 400)
check('wrong key rejected', request('POST', '/webhook', headers={'X-Lab-Time': '1000', 'X-Lab-Signature': signature(raw, secret=b'wrong')}, raw=raw)[0], 400)
check('raw byte change invalidates signature', request('POST', '/webhook', headers={'X-Lab-Time': '1000', 'X-Lab-Signature': signature(raw)}, raw=raw + b' ')[0], 400)
check('old signed envelope rejected', webhook(event, 699)[0], 400)
check('signed producer outside allowed tenant', webhook({**event, 'tenant': 'B', 'resource_id': 'B1'})[0], 403)
check('no untrusted event reached inbox', state['inbox'], 0)
check('accepted event commits all local effects', [webhook(event), state], [(200, {'result': 'accepted'}), {'amount': 250, 'version': 2, 'inbox': 1, 'outbox': 1}])
check('repeat event is deduplicated', [webhook(event, 1001), state['outbox']], [(200, {'result': 'duplicate'}), 1])
check('same ID changed body conflicts', [webhook({**event, 'amount': 251})[0], state['amount']], [409, 250])
older = {**event, 'event_id': 'E0', 'version': 1, 'amount': 100}
check('out of order event does not regress state', [webhook(older), state], [(200, {'result': 'recorded-stale'}), {'amount': 250, 'version': 2, 'inbox': 2, 'outbox': 1}])
future = {**event, 'event_id': 'E2', 'version': 3, 'amount': 350}
before = state
check('fault before commit is reported', webhook(future, **{'X-Lab-Fail': 'before-commit'})[0], 503)
check('fault rolled back inbox state and outbox', state, before)
check('retry after rollback can be accepted', [webhook(future)[0], state], [200, {'amount': 350, 'version': 3, 'inbox': 3, 'outbox': 2}])
concurrent = {**event, 'event_id': 'E3', 'version': 4, 'amount': 450}
with ThreadPoolExecutor(max_workers=2) as workers:
outcomes = list(workers.map(lambda _: webhook(concurrent), [0, 1]))
check('two concurrent deliveries one accepted one duplicate', sorted(v[1]['result'] for v in outcomes), ['accepted', 'duplicate'])
check('concurrent local effects occur once', state, {'amount': 450, 'version': 4, 'inbox': 4, 'outbox': 3})
stop(child)
child = None
child, port = start(2)
check('new process sees committed state', state, {'amount': 450, 'version': 4, 'inbox': 4, 'outbox': 3})
check('new HTTP process deduplicates replay', [webhook(concurrent)[1]['result'], state['outbox']], ['duplicate', 3])
with closing(connect(database)) as db:
check('SQLite structural integrity', db.execute('PRAGMA integrity_check').fetchone[0], 'ok')
return {'runtime': platform.python_version, 'sqliteVersion': sqlite3.sqlite_version, 'transport': 'Actual loopback HTTP/1.1 with two successive child server processes.', 'scope': 'Synthetic identity and clock; fictional HMAC contract; committed local SQLite effects only. Outbox rows were not sent to any external system. Controlled rollback and clean restart are not power-loss or distributed-delivery tests.', 'passed': len(checks), 'checks': checks, 'runnerSha256': hashlib.sha256(Path(__file__).read_bytes).hexdigest}
finally:
if child is not None:
stop(child)
if __name__ == '__main__':
parser = argparse.ArgumentParser
parser.add_argument('--output')
parser.add_argument('--serve')
parser.add_argument('--ready')
args = parser.parse_args
if args.serve:
serve(args.serve, args.ready)
else:
result = json.dumps(run, indent=2) + '\n'
if args.output:
Path(args.output).write_text(result)
else:
print(result, end='')
A1 belongs to tenant A. An A manager changes amount from 100 to 150; an A1+B1 batch is rejected and A1 remains 150.
Common pitfalls
Trusting a hidden button, accepting tenant from the body, binding arbitrary fields to the model, or checking only with a global administrator.
Related topics: Requests and outcomes · Caching and pagination · Concurrent changes and verifiable recovery
Authorization must constrain the function, object, and properties on every path that reads or changes data.
Reference: API1:2023 Broken Object Level Authorization · HTTP semantics RFC9110; OpenAPI3.2.1; selected primary standards and provider contracts consulted2026-09-30