tegwick
e7282e493d
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e6f1-443f-7783-9920-a16b2ffc467f
50 lines
2.3 KiB
Python
50 lines
2.3 KiB
Python
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
|
|
from alert_audit import AuditTransport
|
|
|
|
|
|
class AuditTransportTests(unittest.TestCase):
|
|
def test_real_http_headers_and_redirect_refusal(self):
|
|
calls = []
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
def do_POST(self):
|
|
calls.append((self.path, self.headers['Authorization'], self.headers['Idempotency-Key']))
|
|
body = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
|
|
if body['id'] == 'redirect':
|
|
self.send_response(302)
|
|
self.send_header('Location', '/credential-leak')
|
|
self.end_headers()
|
|
return
|
|
self.send_response(202)
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps({'status': 'accepted', 'reference': 'audit:' + body['id']}).encode())
|
|
|
|
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever)
|
|
thread.start()
|
|
try:
|
|
transport = AuditTransport(f'http://127.0.0.1:{server.server_port}',
|
|
lambda: 'fixture-credential', allow_internal_http=True)
|
|
self.assertEqual(transport({'id': 'test'}), (202, {'status': 'accepted', 'reference': 'audit:test'}))
|
|
self.assertEqual(transport({'id': 'redirect'}), (302, {}))
|
|
self.assertEqual(calls, [('/v1/events', 'Bearer fixture-credential', 'test'),
|
|
('/v1/events', 'Bearer fixture-credential', 'redirect')])
|
|
finally:
|
|
server.shutdown()
|
|
thread.join()
|
|
server.server_close()
|
|
|
|
def test_public_cleartext_and_credentialed_origin_refused(self):
|
|
for origin in ('http://audit.example', 'https://user:password@audit.example',
|
|
'https://audit.example/other', 'https://audit.example?next=elsewhere'):
|
|
with self.assertRaises(ValueError):
|
|
AuditTransport(origin, lambda: 'fixture', allow_internal_http=True)
|