126 lines
7.2 KiB
Python
126 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Regression tests for stateful comparison semantics; these are not PHP parity tests."""
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
import urllib.parse
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
TOOL = Path(__file__).resolve().parents[1] / 'tools/differential_http.py'
|
|
spec = importlib.util.spec_from_file_location('differential_http', TOOL)
|
|
harness = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(harness)
|
|
|
|
class HarnessTests(unittest.TestCase):
|
|
def test_form_transport_preserves_repeated_values_and_reserved_characters(self):
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *_args): pass
|
|
def do_POST(self):
|
|
body = self.rfile.read(int(self.headers['Content-Length']))
|
|
values = urllib.parse.parse_qs(body.decode(), keep_blank_values=True)
|
|
self.send_response(200)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps({'type': self.headers['Content-Type'], 'values': values}).encode())
|
|
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
base = f'http://127.0.0.1:{server.server_port}'
|
|
case = {'path': '/oauth/token', 'method': 'POST', 'form': {'scope': ['a+b', 'c d'], 'empty': '', 'code': 'x&y=z'}}
|
|
result = harness.fetch(base, case, 'candidate', 2)
|
|
self.assertEqual(result['status'], 200)
|
|
self.assertEqual(json.loads(result['body']), {'type': 'application/x-www-form-urlencoded', 'values': {'scope': ['a+b', 'c d'], 'empty': [''], 'code': ['x&y=z']}})
|
|
with self.assertRaises(ValueError):
|
|
harness.fetch(base, dict(case, json={}), 'candidate', 2)
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
def test_typed_captures_and_pointer_escaping(self):
|
|
variables = {}
|
|
harness.capture_values({'body': b'{"a/b":{"~key":[42]}}'}, {'item': '/a~1b/~0key/0'}, variables)
|
|
self.assertEqual(harness.resolve({'path': '/items/{{item}}', 'json': ['{{item}}']}, variables), {'path': '/items/42', 'json': [42]})
|
|
with self.assertRaises(KeyError): harness.resolve('{{missing}}', variables)
|
|
with self.assertRaises(ValueError): harness.validate_path('//remote')
|
|
|
|
def test_json_url_normalization_after_unescaping(self):
|
|
response = {'body': b'{"url":"http:\\/\\/localhost:1234","nested":["http:\\/\\/localhost:1234/path"]}', 'content_type': 'application/json'}
|
|
self.assertEqual(harness.comparable(response, 'http://localhost:1234', {}), {'url': '<BASE>', 'nested': ['<BASE>/path']})
|
|
|
|
def test_json_assertions_are_independent_of_equality(self):
|
|
response = {'body': b'{"count":0}'}
|
|
self.assertFalse(harness.assert_json_response(response, {'expected_json': {'/count':1}}))
|
|
self.assertFalse(harness.assert_json_response(response, {'absent_json_pointers':['/count']}))
|
|
self.assertTrue(harness.assert_json_response(response, {'absent_json_pointers':['/secret']}))
|
|
|
|
def test_csrf_shape_rejects_json_or_html_documents(self):
|
|
self.assertTrue(harness.is_text_token({'status':200,'content_type':'text/html','body':b'A'*40}))
|
|
self.assertFalse(harness.is_text_token({'status':200,'content_type':'application/json','body':b'A'*40}))
|
|
self.assertFalse(harness.is_text_token({'status':200,'content_type':'text/html','body':b'<html>'+b'A'*40+b'</html>'}))
|
|
|
|
def test_cookie_rotation_capture_and_state_effects(self):
|
|
servers = []
|
|
for fixture_id in (14, 37):
|
|
class Handler(BaseHTTPRequestHandler):
|
|
item_id = fixture_id
|
|
saved = False
|
|
def log_message(self, *args): pass
|
|
def do_GET(self):
|
|
if self.path == '/session':
|
|
self.send_response(200)
|
|
self.send_header('Set-Cookie', 'session=fixture; Path=/')
|
|
self.send_header('Set-Cookie', 'XSRF-TOKEN=encoded%2Btoken; Path=/')
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.wfile.write(b'{}')
|
|
elif self.path == f'/items/{self.item_id}' and self.saved:
|
|
self.reply(200, {'value': 'saved'})
|
|
else: self.reply(404, {})
|
|
def do_POST(self):
|
|
body = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
|
|
if self.headers.get('Cookie', '').find('session=fixture') < 0 or self.headers.get('X-XSRF-TOKEN') != 'encoded+token' or self.headers.get('X-Compat') != 'fixture':
|
|
self.reply(419, {})
|
|
elif body == {'value': 'saved'}:
|
|
type(self).saved = True
|
|
self.reply(201, {'id': self.item_id})
|
|
else: self.reply(422, {})
|
|
def reply(self, status, payload):
|
|
self.send_response(status)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(payload).encode())
|
|
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
servers.append(server)
|
|
try:
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
cases = Path(folder) / 'cases.json'
|
|
cases.write_text(json.dumps([
|
|
{'path': '/session', 'expected_status': 200},
|
|
{'method': 'POST', 'path': '/items', 'headers': {'X-Compat': 'fixture'}, 'json': {'value': 'saved'}, 'headers_from_cookies': {'X-XSRF-TOKEN': 'XSRF-TOKEN'}, 'capture_json': {'id': '/id'}, 'ignore_json_pointers': ['/id'], 'expected_status': 201},
|
|
{'path': '/items/{{id}}', 'expected_status': 200},
|
|
]))
|
|
command = [sys.executable, str(TOOL), '--reference', f'http://127.0.0.1:{servers[0].server_port}', '--candidate', f'http://127.0.0.1:{servers[1].server_port}', '--cases', str(cases), '--allow-side-effects', '--sessions']
|
|
result = subprocess.run(command, capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual([row['result'] for row in map(json.loads, result.stdout.splitlines())], ['equal'] * 3)
|
|
self.assertNotIn('encoded', result.stdout)
|
|
cases.write_text(json.dumps([{'path': '/session', 'expected_status': 201}]))
|
|
mismatch = subprocess.run(command, capture_output=True, text=True)
|
|
self.assertEqual(mismatch.returncode, 1)
|
|
self.assertFalse(json.loads(mismatch.stdout)['checks']['expected_status'])
|
|
finally:
|
|
for server in servers:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
if __name__ == '__main__': unittest.main()
|