APRENDER. PRATICAR. EVOLUIR.Desenhar uma matriz que o servidor consegue aplicar
Começa com três perguntas concretas: quem pode executar a função, sobre que objetos e com que propriedades? No exemplo fictício, um analista lê instruções do seu tenant, enquanto um gestor pode alterar o montante dentro de limites acordados. Nenhum deles altera a aprovação através desse fluxo. Escreve estes casos antes de discutir nomes de papéis ou botões. Usa contexto de identidade estabelecido pelo servidor e associa a consulta ao tenant autorizado. Um campo enviado pelo cliente não pode escolher outro âmbito. Conhecer um ID também não concede acesso ao respetivo objeto.
Limitar entradas e representações
Um schema de escrita explícito reduz a superfície de atribuição indevida. A operação do laboratório aceita exatamente amount, exige um inteiro entre zero e dez mil e rejeita approved, tenant e propriedades adicionais. O uso de type(value) is int é intencional: em Python, um booleano pode passar num teste demasiado permissivo de inteiro. A leitura devolve apenas id, amount e version. O registo interno contém mais campos, mas a serialização não os copia automaticamente. Documentar readOnly pode ajudar consumidores; a prova de proteção está no comportamento do servidor e no estado persistido.
Avaliar lotes como operações completas
No contrato deste exercício, um lote é tudo ou nada. Se A1 é permitido e B1 pertence a outro tenant, nenhuma alteração pode ficar confirmada. O código verifica todos os objetos dentro da transação antes de os atualizar. Isto também evita comunicar sucesso integral depois de ignorar um item proibido. A quantidade máxima de dois objetos é um limite didático, não uma recomendação universal de capacidade. Na produção, define tamanho do corpo, operações por lote, tempo e recursos consumidos segundo os requisitos. Um limite de pedidos por minuto não limita o custo de um único pedido sem restrições.
Executar a matriz no laboratório
Guarda o código integral desta aula como run.py e executa python3 run.py --output evidence.json numa pasta temporária com Python 3.13. O programa cria dados fictícios e inicia HTTP apenas em 127.0.0.1 numa porta livre. X-Lab-Actor seleciona identidades sintéticas conhecidas: não autentica utilizadores. Observa leitura autorizada, escrita de gestor, recusa de analista, objeto de outro tenant, campo protegido, booleano, montante excessivo e lote misto. Depois dos casos negativos, o programa consulta o estado para demonstrar ausência de alterações. O ficheiro temporário não contém dados do catálogo dr.pt.
Investigar um caminho alternativo
Num caso fictício de fundos, a leitura individual funciona corretamente, mas uma nova exportação omite a condição de tenant. O incidente pode ficar escondido se a aceitação usar sempre um gestor global. Suspende o caminho afetado, preserva evidência com acesso limitado e determina os objetos e destinatários envolvidos. Revê a query da exportação, a projeção de resposta e as permissões da função. A partilha do mesmo middleware de autenticação não demonstra autorização equivalente. Para reabrir, combina um caso legítimo com identidades de outro tenant e papéis sem concessão, confirmando também os efeitos persistidos.
Entregar critérios observáveis a APS
A evidência útil liga principal, ação, objeto, propriedades, resultado esperado e resultado observado. Mantém dados e credenciais reais fora do laboratório. Para a passagem a RUN, inclui responsáveis pelas políticas, casos de revogação, caminhos de exportação e uma forma de diagnosticar recusas sem revelar segredos. O exercício local não cobre um fornecedor de identidade, TLS, limites de concorrência global ou resistência a clientes hostis. A chave pública de demonstração e os headers sintéticos nunca devem ser promovidos a credenciais reais. Usa o laboratório para aprender e preparar critérios para um ambiente autorizado.
"""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 pertence ao tenant A. Um gestor de A altera amount de 100 para 150; um lote A1+B1 é recusado e A1 permanece 150.
Armadilhas comuns
Confiar no botão escondido, aceitar tenant do corpo, ligar qualquer campo ao modelo ou testar apenas com um administrador global.
Tópicos relacionados: Pedidos e resultados · Cache e paginação · Alterações concorrentes e recuperação verificável
A autorização tem de limitar a função, o objeto e as propriedades em cada caminho que lê ou altera dados.
Referência: API1:2023 Broken Object Level Authorization · HTTP semantics RFC9110; OpenAPI3.2.1; selected primary standards and provider contracts consulted2026-09-30