"""Focused boundaries of the additive installed adapter, not mirrored core tests.""" import http.client import json import os from pathlib import Path import socket import ssl import subprocess import sys import tempfile import threading import unittest from unittest import mock from odexa_ref import crypto from odexa_runtime.bootstrap import initialize from odexa_runtime.client import Client from odexa_runtime.config import Config from odexa_runtime.runtime import Runtime from odexa_runtime.server import Server, BodyWriter class CurrentRuntimeTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory(); self.root=Path(self.tmp.name) with socket.socket() as sock: sock.bind(('127.0.0.1',0)); self.port=sock.getsockname()[1] self.origin=f'https://127.0.0.1:{self.port}' initialize(self.root,origin=self.origin) self.runtime=Runtime(self.root) self.server=Server(self.runtime,'origin') self.thread=threading.Thread(target=self.server.serve_forever,kwargs={'poll_interval':0.02},daemon=True) self.thread.start() self.context=ssl.create_default_context(cafile=str(self.root/'ca.pem')) def tearDown(self): self.server.shutdown(); self.server.server_close(); self.thread.join(timeout=2); self.tmp.cleanup() def get(self,path,headers=None): conn=http.client.HTTPSConnection('127.0.0.1',self.port,context=self.context,timeout=5) try: conn.request('GET',path,headers=headers or {}) response=conn.getresponse(); return response.status,response.read() finally: conn.close() def test_static_only_allowlist_no_secrets_or_unprotected_asset(self): status,body=self.get('/odexa.json') self.assertEqual(status,200); self.assertEqual(crypto.strict_json(body)['origin'],self.origin) for path in ('/runtime.json','/agent.json','/keys/service.pem','/state/agreements.sqlite3','/../runtime.json','/about'): with self.subTest(path=path): self.assertNotEqual(self.get(path)[0],200) def test_host_and_framing_rejected_before_dispatch(self): self.assertEqual(self.get('/odexa.json',{'Host':'attacker.example'})[0],403) with socket.create_connection(('127.0.0.1',self.port),timeout=5) as tcp: with self.context.wrap_socket(tcp,server_hostname='127.0.0.1') as secure: secure.sendall((f'GET /odexa.json HTTP/1.1\r\nHost: 127.0.0.1:{self.port}\r\nContent-Length: 0\r\nContent-Length: 1\r\n\r\n').encode()) self.assertTrue(secure.recv(256).startswith(b'HTTP/1.1 403')) self.assertEqual(self.get('/odexa.json',{'Content-Length':'999999'})[0],403) def test_untrusted_certificate_cannot_fetch_origin(self): conn=http.client.HTTPSConnection('127.0.0.1',self.port,timeout=5) try: with self.assertRaises(ssl.SSLCertVerificationError): conn.request('GET','/odexa.json') finally: conn.close() def test_raw_request_target_aliases_rejected_before_dispatch(self): for target in ('//odexa.json','///odexa.json','//about','/./about','/%61bout','/about?','/about#fragment'): with self.subTest(target=target), mock.patch.object(self.runtime,'route',wraps=self.runtime.route) as route: with socket.create_connection(('127.0.0.1',self.port),timeout=5) as tcp: with self.context.wrap_socket(tcp,server_hostname='127.0.0.1') as secure: secure.sendall((f'GET {target} HTTP/1.1\r\nHost: 127.0.0.1:{self.port}\r\n\r\n').encode()) self.assertTrue(secure.recv(256).startswith(b'HTTP/1.1 403')) route.assert_not_called() self.assertEqual(self.get('/odexa.json')[0],200) def test_client_verifies_public_authority_without_reading_service_private_key(self): client=Client(self.root); original=client.cfg.key def selected(name): self.assertNotEqual(name,'service','agent client must not read issuer private key') return original(name) with mock.patch.object(client.cfg,'key',selected): result=client.agree(self.root/'exchange.json') self.assertEqual(result['state'],'active') self.assertEqual(client.status(self.root/'exchange.json')['state'],'active') def test_free_initialisation_imports_no_paid_runtime(self): code='''import json,sys,tempfile from odexa_runtime.bootstrap import initialize with tempfile.TemporaryDirectory() as p: initialize(p) print(json.dumps([n for n in sys.modules if n.startswith(("profiles.paid", "profiles.payment", "profiles.network_payments"))]))''' result=subprocess.run([sys.executable,'-c',code],capture_output=True,text=True,timeout=10) self.assertEqual(result.returncode,0,result.stderr); self.assertEqual(json.loads(result.stdout),[]) def test_private_configuration_and_no_synthetic_public_verifier(self): original=(self.root/'runtime.json').read_bytes() os.chmod(self.root/'runtime.json',0o644) with self.assertRaises(ValueError): Config(self.root) os.chmod(self.root/'runtime.json',0o600) with self.assertRaises(ValueError): initialize(self.root,origin=self.origin) self.assertEqual((self.root/'runtime.json').read_bytes(),original) with tempfile.TemporaryDirectory() as other: with self.assertRaises(ValueError): initialize(other,origin='https://origin.example',service_origin='https://provider.example',paid=True) with self.assertRaises(ValueError): initialize(other,origin='https://origin.example') self.assertEqual(list(Path(other).iterdir()),[]) def test_origin_withdrawal_prevents_new_assent(self): data=crypto.strict_json((self.root/'public'/'odexa-service.json').read_bytes()) data.update(revision=2,services=[]) (self.root/'public'/'odexa-service.json').write_bytes(crypto.json_bytes(data)) with self.assertRaises(ValueError): Client(self.root).agree(self.root/'denied.json') self.assertFalse((self.root/'denied.json').exists()) def test_partial_body_write_retains_confirmed_local_prefix(self): connection=mock.Mock() connection.send.side_effect=[3,2,OSError('synthetic transport failure')] writer=BodyWriter(connection) with self.assertRaises(OSError): writer.write(b'abcdefgh') self.assertEqual(writer.count,5) self.assertEqual([bytes(call.args[0]) for call in connection.send.call_args_list],[b'abcdefgh',b'defgh',b'fgh']) stalled=BodyWriter(mock.Mock(send=mock.Mock(return_value=0))) with self.assertRaises(OSError): stalled.write(b'body') self.assertEqual(stalled.count,0) if __name__=='__main__': unittest.main()