2608a02d5c
Without this we end up in weird cases where the viewer gets a cache hit and never request the object data, creating link heirarchies where the viewer knows about all the prims but Hippolyzer only knows some of them and orphans them. Since we don't know what viewer the user is using, we scan around the disk for object caches and try to use those. 99% of the time the connection will be coming from localhost so this is fine. Fixes #11
112 lines
4.9 KiB
Python
112 lines
4.9 KiB
Python
import unittest
|
|
|
|
from mitmproxy.test import tflow, tutils
|
|
|
|
from hippolyzer.lib.base.datatypes import UUID
|
|
from hippolyzer.lib.proxy.http_flow import HippoHTTPFlow
|
|
from hippolyzer.lib.proxy.message_logger import HTTPMessageLogEntry
|
|
from hippolyzer.lib.proxy.sessions import SessionManager
|
|
|
|
|
|
class TestHTTPFlows(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session_manager = SessionManager()
|
|
self.session = self.session = self.session_manager.create_session({
|
|
"session_id": UUID.random(),
|
|
"secure_session_id": UUID.random(),
|
|
"agent_id": UUID.random(),
|
|
"circuit_code": 0,
|
|
"sim_ip": "127.0.0.1",
|
|
"region_x": 1,
|
|
"region_y": 2,
|
|
"sim_port": "1",
|
|
"seed_capability": "https://test.localhost:4/foo",
|
|
})
|
|
self.region = self.session.register_region(
|
|
("127.0.0.1", 2),
|
|
"https://test.localhost:4/foo",
|
|
handle=90,
|
|
)
|
|
self.region.update_caps({
|
|
"FakeCap": "http://example.com",
|
|
"ViewerAsset": "http://assets.example.com",
|
|
})
|
|
|
|
def test_request_formatting(self):
|
|
req = tutils.treq(host="example.com", port=80)
|
|
fake_flow = tflow.tflow(req=req, resp=tutils.tresp())
|
|
flow = HippoHTTPFlow.from_state(fake_flow.get_state(), self.session_manager)
|
|
# Make sure cap resolution works correctly
|
|
flow.cap_data = self.session_manager.resolve_cap(flow.request.url)
|
|
entry = HTTPMessageLogEntry(flow)
|
|
self.assertEqual(entry.request(beautify=True), """GET [[FakeCap]]/path HTTP/1.1\r
|
|
# http://example.com/path\r
|
|
header: qvalue\r
|
|
content-length: 7\r
|
|
\r
|
|
content""")
|
|
|
|
def test_binary_request_formatting(self):
|
|
req = tutils.treq(host="example.com", port=80)
|
|
fake_flow = tflow.tflow(req=req, resp=tutils.tresp())
|
|
flow = HippoHTTPFlow.from_state(fake_flow.get_state(), self.session_manager)
|
|
# This should trigger the escaped body path without changing content-length
|
|
flow.request.content = b"c\x00ntent"
|
|
entry = HTTPMessageLogEntry(flow)
|
|
self.assertEqual(entry.request(beautify=True), """GET http://example.com/path HTTP/1.1\r
|
|
header: qvalue\r
|
|
content-length: 7\r
|
|
X-Hippo-Escaped-Body: 1\r
|
|
\r
|
|
c\\x00ntent""")
|
|
|
|
def test_llsd_response_formatting(self):
|
|
fake_flow = tflow.tflow(req=tutils.treq(), resp=tutils.tresp())
|
|
flow = HippoHTTPFlow.from_state(fake_flow.get_state(), self.session_manager)
|
|
# Half the time LLSD is sent with a random Content-Type and no PI indicating
|
|
# what flavor of LLSD it is. Make sure the sniffing works correctly.
|
|
flow.response.content = b"<llsd><integer>1</integer></llsd>"
|
|
entry = HTTPMessageLogEntry(flow)
|
|
self.assertEqual(entry.response(beautify=True), """HTTP/1.1 200 OK\r
|
|
header-response: svalue\r
|
|
content-length: 33\r
|
|
\r
|
|
<?xml version="1.0" ?>
|
|
<llsd>
|
|
<integer>1</integer>
|
|
</llsd>
|
|
""")
|
|
|
|
def test_flow_state_serde(self):
|
|
fake_flow = tflow.tflow(req=tutils.treq(host="example.com"), resp=tutils.tresp())
|
|
flow = HippoHTTPFlow.from_state(fake_flow.get_state(), self.session_manager)
|
|
# Make sure cap resolution works correctly
|
|
flow.cap_data = self.session_manager.resolve_cap(flow.request.url)
|
|
flow_state = flow.get_state()
|
|
new_flow = HippoHTTPFlow.from_state(flow_state, self.session_manager)
|
|
self.assertIs(self.session, new_flow.cap_data.session())
|
|
|
|
def test_http_asset_repo(self):
|
|
asset_repo = self.session_manager.asset_repo
|
|
asset_id = asset_repo.create_asset(b"foobar", one_shot=True)
|
|
req = tutils.treq(host="assets.example.com", path=f"/?animatn_id={asset_id}")
|
|
fake_flow = tflow.tflow(req=req)
|
|
flow = HippoHTTPFlow.from_state(fake_flow.get_state(), self.session_manager)
|
|
# Have to resolve cap data so the asset repo knows this is an asset server cap
|
|
flow.cap_data = self.session_manager.resolve_cap(flow.request.url)
|
|
self.assertTrue(asset_repo.try_serve_asset(flow))
|
|
self.assertEqual(b"foobar", flow.response.content)
|
|
|
|
def test_temporary_cap_resolution(self):
|
|
self.region.register_temporary_cap("TempExample", "http://not.example.com")
|
|
self.region.register_temporary_cap("TempExample", "http://not2.example.com")
|
|
# Resolving the cap should consume it
|
|
cap_data = self.session_manager.resolve_cap("http://not.example.com")
|
|
self.assertEqual(cap_data.cap_name, "TempExample")
|
|
# A CapData object should always be returned, but the cap_name field will be None
|
|
new_cap_data = self.session_manager.resolve_cap("http://not.example.com")
|
|
self.assertIsNone(new_cap_data.cap_name)
|
|
# The second temp cap with the same name should still be in there
|
|
cap_data = self.session_manager.resolve_cap("http://not2.example.com")
|
|
self.assertEqual(cap_data.cap_name, "TempExample")
|