#!/usr/bin/env python3
"""Synthetic browser/server boundary and key-rotation exercise. Python 3.10+.
SPDX-License-Identifier: MIT
No Lovable/Supabase service or real provider is contacted. No remote target accepted.
"""
from datetime import datetime, timezone
import hashlib
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
from pathlib import Path
import platform
import secrets
import sys
import threading
from urllib.error import HTTPError
from urllib.request import build_opener, ProxyHandler, Request
def main():
if len(sys.argv) != 1:
raise SystemExit('Usage: python3 run.py > results.json; local fixture only')
# All credentials are synthetic, created only for this run and never printed.
old_key, new_key = secrets.token_urlsafe(32), secrets.token_urlsafe(32)
session_alpha, session_beta = secrets.token_urlsafe(24), secrets.token_urlsafe(24)
provider = {'accepted_keys': {old_key}, 'calls': 0}
backend = {'key': old_key}
sessions = {session_alpha: 'alpha', session_beta: 'beta'}
checks = []
public_marker = 'demo_publishable_key_not_a_real_credential'
client_js = (
f'const projectKey = "{public_marker}";\n'
'document.querySelector("output").textContent = "Synthetisches Browserbeispiel";\n'
'// Session handling is intentionally not implemented in this HTTP fixture.\n'
'// No provider key is shipped in this file.\n'
).encode()
html = b'Secrets-Fixture'
def provider_call(key):
if key not in provider['accepted_keys']:
return 401
provider['calls'] += 1
return 200
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def reply(self, status, data, content_type='application/json'):
payload = data if isinstance(data, bytes) else json.dumps(data).encode()
self.send_response(status)
self.send_header('Content-Type', content_type)
self.send_header('Cache-Control', 'no-store')
self.send_header('Content-Length', str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def do_GET(self):
if self.path == '/':
return self.reply(200, html, 'text/html; charset=utf-8')
if self.path == '/app.js':
return self.reply(200, client_js, 'text/javascript; charset=utf-8')
return self.reply(404, {'code': 'not_found'})
def do_POST(self):
if self.path != '/api/summary':
return self.reply(404, {'code': 'not_found'})
tenant = sessions.get(self.headers.get('Authorization', '').removeprefix('Bearer '))
if tenant is None:
return self.reply(401, {'code': 'authentication_required'})
lengths = self.headers.get_all('Content-Length', [])
if self.headers.get('Transfer-Encoding') or len(lengths) != 1 or not lengths[0].isdigit():
return self.reply(400, {'code': 'invalid_length'})
length = int(lengths[0])
if length > 1024:
return self.reply(413, {'code': 'payload_too_large'})
try:
body = json.loads(self.rfile.read(length))
except (ValueError, UnicodeError):
return self.reply(400, {'code': 'invalid_json'})
if not isinstance(body, dict) or set(body) != {'contact_id'}:
return self.reply(400, {'code': 'invalid_input'})
# The object-level check happens BEFORE privileged provider work.
if body['contact_id'] != tenant + '-contact':
return self.reply(403, {'code': 'contact_forbidden'})
if provider_call(backend['key']) != 200:
return self.reply(502, {'code': 'provider_unavailable'})
return self.reply(200, {'summary': 'Synthetische Zusammenfassung'})
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
base = f'http://127.0.0.1:{server.server_port}'
def request(path, token=None, body=None):
headers = {'Content-Type': 'application/json'}
if token:
headers['Authorization'] = 'Bearer ' + token
req = Request(base + path, data=None if body is None else json.dumps(body).encode(), headers=headers)
try:
response = build_opener(ProxyHandler({})).open(req, timeout=10)
except HTTPError as error:
response = error
with response:
payload = response.read()
if old_key.encode() in payload or new_key.encode() in payload:
raise RuntimeError('Privileged secret leaked in response')
return response.status, payload
def check(label, condition):
if not condition:
raise RuntimeError(label)
checks.append({'case': label, 'passed': True})
try:
check('HTML served without privileged key', request('/')[0] == 200)
status, javascript = request('/app.js')
check('public marker is visible in JavaScript', status == 200 and public_marker.encode() in javascript)
# Positive control proves that a simple byte scan can find this exact fake secret.
unsafe_sample = b'const key="' + old_key.encode() + b'";'
check('scan detects intentionally exposed synthetic secret', old_key.encode() in unsafe_sample)
check('served JavaScript contains neither provider key', old_key.encode() not in javascript and new_key.encode() not in javascript)
check('anonymous denied', request('/api/summary', body={'contact_id': 'alpha-contact'})[0] == 401)
check('public key is not a user session', request('/api/summary', public_marker, {'contact_id': 'alpha-contact'})[0] == 401)
check('foreign tenant denied', request('/api/summary', session_beta, {'contact_id': 'alpha-contact'})[0] == 403)
check('arbitrary provider URL rejected', request('/api/summary', session_alpha, {'contact_id': 'alpha-contact', 'url': 'https://example.invalid'})[0] == 400)
check('denied requests did not reach provider', provider['calls'] == 0)
check('authorized request succeeds', request('/api/summary', session_alpha, {'contact_id': 'alpha-contact'})[0] == 200)
provider['accepted_keys'].add(new_key)
check('creating a second key does not revoke the first', provider_call(old_key) == 200)
backend['key'] = new_key
check('backend works with replacement', request('/api/summary', session_alpha, {'contact_id': 'alpha-contact'})[0] == 200)
provider['accepted_keys'].remove(old_key)
check('old key fails after explicit revocation', provider_call(old_key) == 401)
check('backend remains available after revocation', request('/api/summary', session_alpha, {'contact_id': 'alpha-contact'})[0] == 200)
backend['key'] = old_key
check('stale backend fails without leaking credential', request('/api/summary', session_alpha, {'contact_id': 'alpha-contact'})[0] == 502)
print(json.dumps({
'recorded_at': datetime.now(timezone.utc).isoformat(), 'python': platform.python_version(),
'scope': 'Local HTTP example with synthetic sessions and in-memory provider stub. Not a Lovable export, Vite build, Supabase Auth test, browser execution or real provider rotation.',
'source_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
'passed': len(checks), 'checks': checks,
}, indent=2))
finally:
server.shutdown()
server.server_close()
thread.join(timeout=10)
if __name__ == '__main__':
main()