Miko/Hippolyzer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add new persistent (Proxy)Settings object, use to pass down settings

Browse Source
This commit is contained in:
Salad Dais
2021-06-08 16:46:03 +00:00
parent 24abc36df2
commit 5618bcbac1
14 changed files with 149 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
hippolyzer/apps/proxy.py
View File

@@ -20,6 +20,7 @@ from hippolyzer.lib.proxy.lludp_proxy import SLSOCKS5Server
from hippolyzer.lib.base.message.message import Message
from hippolyzer.lib.proxy.region import ProxiedRegion
from hippolyzer.lib.proxy.sessions import SessionManager, Session
from hippolyzer.lib.proxy.settings import ProxySettings
LOG = logging.getLogger(__name__)
@@ -91,8 +92,8 @@ def run_http_proxy_process(proxy_host, http_proxy_port, flow_context: HTTPFlowCo
mitm_loop.run_forever()
def start_proxy(extra_addons: Optional[list] = None, extra_addon_paths: Optional[list] = None,
session_manager=None, proxy_host=None):
def start_proxy(session_manager: SessionManager, extra_addons: Optional[list] = None,
extra_addon_paths: Optional[list] = None, proxy_host=None):
extra_addons = extra_addons or []
extra_addon_paths = extra_addon_paths or []
extra_addons.append(SelectionManagerAddon())
@@ -105,12 +106,11 @@ def start_proxy(extra_addons: Optional[list] = None, extra_addon_paths: Optional
loop = asyncio.get_event_loop()
udp_proxy_port = int(os.environ.get("HIPPO_UDP_PORT", 9061))
http_proxy_port = int(os.environ.get("HIPPO_HTTP_PORT", 9062))
udp_proxy_port = session_manager.settings.SOCKS_PROXY_PORT
http_proxy_port = session_manager.settings.HTTP_PROXY_PORT
if proxy_host is None:
proxy_host = os.environ.get("HIPPO_BIND_HOST", "127.0.0.1")
proxy_host = session_manager.settings.PROXY_BIND_ADDR
session_manager = session_manager or SessionManager()
flow_context = session_manager.flow_context
session_manager.name_cache.load_viewer_caches()
@@ -186,7 +186,7 @@ def _windows_timeout_killer(pid: int):
def main():
multiprocessing.set_start_method("spawn")
start_proxy()
start_proxy(SessionManager(ProxySettings()))
if __name__ == "__main__":

56
hippolyzer/apps/proxy_gui.py
View File

@@ -1,5 +1,6 @@
import asyncio
import base64
import dataclasses
import email
import functools
import html
@@ -44,6 +45,7 @@ from hippolyzer.lib.proxy.http_proxy import create_proxy_master, HTTPFlowContext
from hippolyzer.lib.proxy.message_logger import LLUDPMessageLogEntry, AbstractMessageLogEntry
from hippolyzer.lib.proxy.region import ProxiedRegion
from hippolyzer.lib.proxy.sessions import Session, SessionManager
from hippolyzer.lib.proxy.settings import ProxySettings
from hippolyzer.lib.proxy.templates import CAP_TEMPLATES
LOG = logging.getLogger(__name__)
@@ -66,8 +68,8 @@ class GUISessionManager(SessionManager, QtCore.QObject):
regionAdded = QtCore.Signal(ProxiedRegion)
regionRemoved = QtCore.Signal(ProxiedRegion)
def __init__(self, model):
SessionManager.__init__(self)
def __init__(self, settings, model):
SessionManager.__init__(self, settings)
QtCore.QObject.__init__(self)
self.all_regions = []
self.message_logger = model
@@ -172,9 +174,9 @@ class ProxyGUI(QtWidgets.QMainWindow):
super().__init__()
loadUi(MAIN_WINDOW_UI_PATH, self)
self.settings = QtCore.QSettings("SaladDais", "hippolyzer")
self._selectedEntry: Optional[AbstractMessageLogEntry] = None
self.settings = GUIProxySettings(QtCore.QSettings("SaladDais", "hippolyzer"))
self.model = MessageLogModel(parent=self.tableView)
self.tableView.setModel(self.model)
self.model.rowsAboutToBeInserted.connect(self.beforeInsert)
@@ -191,10 +193,9 @@ class ProxyGUI(QtWidgets.QMainWindow):
self.actionManageAddons.triggered.connect(self._manageAddons)
self.actionManageFilters.triggered.connect(self._manageFilters)
self.actionOpenMessageBuilder.triggered.connect(self._openMessageBuilder)
self.actionProxyRemotelyAccessible.setChecked(
self.settings.value("RemotelyAccessible", False, type=bool))
self.actionUseViewerObjectCache.setChecked(
self.settings.value("UseViewerObjectCache", False, type=bool))
self.actionProxyRemotelyAccessible.setChecked(self.settings.REMOTELY_ACCESSIBLE)
self.actionUseViewerObjectCache.setChecked(self.settings.USE_VIEWER_OBJECT_CACHE)
self.actionProxyRemotelyAccessible.triggered.connect(self._setProxyRemotelyAccessible)
self.actionUseViewerObjectCache.triggered.connect(self._setUseViewerObjectCache)
@@ -202,7 +203,7 @@ class ProxyGUI(QtWidgets.QMainWindow):
self._populateFilterMenu()
self.toolButtonFilter.setMenu(self._filterMenu)
self.sessionManager = GUISessionManager(self.model)
self.sessionManager = GUISessionManager(self.settings, self.model)
self.interactionManager = GUIInteractionManager(self)
AddonManager.UI = self.interactionManager
@@ -223,15 +224,12 @@ class ProxyGUI(QtWidgets.QMainWindow):
self._filterMenu.clear()
_addFilterAction("Default", self.DEFAULT_FILTER)
filters = self.getFilterDict()
filters = self.settings.FILTERS
for preset_name, preset_filter in filters.items():
_addFilterAction(preset_name, preset_filter)
def getFilterDict(self):
return json.loads(str(self.settings.value("Filters", "{}")))
def setFilterDict(self, val: dict):
self.settings.setValue("Filters", json.dumps(val))
self.settings.FILTERS = val
self._populateFilterMenu()
def _manageFilters(self):
@@ -376,24 +374,23 @@ class ProxyGUI(QtWidgets.QMainWindow):
msg.exec()
def _setProxyRemotelyAccessible(self, checked: bool):
self.settings.setValue("RemotelyAccessible", checked)
self.sessionManager.settings.REMOTELY_ACCESSIBLE = checked
msg = QtWidgets.QMessageBox()
msg.setText("Remote accessibility setting changes will take effect on next run")
msg.exec()
def _setUseViewerObjectCache(self, checked: bool):
self.settings.setValue("UseViewerObjectCache", checked)
self.sessionManager.use_viewer_object_cache = checked
self.sessionManager.settings.USE_VIEWER_OBJECT_CACHE = checked
def _manageAddons(self):
dialog = AddonDialog(self)
dialog.exec_()
def getAddonList(self) -> List[str]:
return json.loads(str(self.settings.value("Addons", "[]")))
return self.sessionManager.settings.ADDON_SCRIPTS
def setAddonList(self, val: List[str]):
self.settings.setValue("Addons", json.dumps(val))
self.sessionManager.settings.ADDON_SCRIPTS = val
BANNED_HEADERS = ("content-length", "host")
@@ -719,7 +716,7 @@ class MessageBuilderWindow(QtWidgets.QMainWindow):
return val
def _sendHTTPRequest(self, method, uri, headers, body):
caps_client = ProxyCapsClient()
caps_client = ProxyCapsClient(self.sessionManager.settings)
async def _send_request():
req = caps_client.request(method, uri, headers=headers, data=body)
@@ -823,6 +820,22 @@ class FilterDialog(QtWidgets.QDialog):
self.listFilters.takeItem(idx)
class GUIProxySettings(ProxySettings):
"""Persistent settings backed by QSettings"""
def __init__(self, settings: QtCore.QSettings):
super().__init__()
self._settings_obj = settings
def get_setting(self, name: str) -> Any:
val: Any = self._settings_obj.value(name, defaultValue=dataclasses.MISSING)
if val is dataclasses.MISSING:
return val
return json.loads(val)
def set_setting(self, name: str, val: Any):
self._settings_obj.setValue(name, json.dumps(val))
def gui_main():
multiprocessing.set_start_method('spawn')
QtCore.QCoreApplication.setAttribute(QtCore.Qt.AA_ShareOpenGLContexts)
@@ -835,11 +848,8 @@ def gui_main():
timer.start(100)
signal.signal(signal.SIGINT, lambda *args: QtWidgets.QApplication.quit())
window.show()
remote_access = window.settings.value("RemotelyAccessible", False, type=bool)
use_vocache = window.settings.value("UseViewerObjectCache", False, type=bool)
window.sessionManager.use_viewer_object_cache = use_vocache
http_host = None
if remote_access:
if window.sessionManager.settings.REMOTELY_ACCESSIBLE:
http_host = "0.0.0.0"
start_proxy(
session_manager=window.sessionManager,

113
hippolyzer/lib/base/settings.py
View File

@@ -19,81 +19,48 @@ along with this program; if not, write to the Free Software Foundation,
Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
"""
from __future__ import annotations
import dataclasses
from typing import *
_T = TypeVar("_T")
class SettingDescriptor(Generic[_T]):
__slots__ = ("name", "default")
def __init__(self, default: Union[Callable[[], _T], _T]):
self.default = default
self.name: Optional[str] = None
def __set_name__(self, owner: Settings, name: str):
self.name = name
def _make_default(self) -> _T:
if callable(self.default):
return self.default()
return self.default
def __get__(self, obj: Settings, owner: Optional[Type] = None) -> _T:
val: Union[_T, dataclasses.MISSING] = obj.get_setting(self.name)
if val is dataclasses.MISSING:
val = self._make_default()
return val
def __set__(self, obj: Settings, value: _T) -> None:
obj.set_setting(self.name, value)
class Settings:
def __init__(self, quiet_logging=False, spammy_logging=False, log_tests=True):
""" some lovely configurable settings
ENABLE_DEFERRED_PACKET_PARSING: bool = SettingDescriptor(True)
These are applied application wide, and can be
overridden at any time in a specific instance
def __init__(self):
self._settings: Dict[str, Any] = {}
quiet_logging overrides spammy_logging
"""
def get_setting(self, name: str) -> Any:
return self._settings.get(name, dataclasses.MISSING)
self.quiet_logging = quiet_logging
self.spammy_logging = spammy_logging
# toggle handling udp packets
self.HANDLE_PACKETS = True
self.HANDLE_OUTGOING_PACKETS = False
# toggle parsing all/handled packets
self.ENABLE_DEFERRED_PACKET_PARSING = True
# ~~~~~~~~~~~~~~~~~~
# Logging behaviors
# ~~~~~~~~~~~~~~~~~~
# being a test tool, and an immature one at that,
# enable fine granularity in the logging, but
# make sure we can tone it down as well
self.LOG_VERBOSE = True
self.ENABLE_BYTES_TO_HEX_LOGGING = False
self.ENABLE_CAPS_LOGGING = True
self.ENABLE_CAPS_LLSD_LOGGING = False
self.ENABLE_EQ_LOGGING = True
self.ENABLE_UDP_LOGGING = True
self.ENABLE_OBJECT_LOGGING = True
self.LOG_SKIPPED_PACKETS = True
self.ENABLE_HOST_LOGGING = True
self.LOG_COROUTINE_SPAWNS = True
self.PROXY_LOGGING = False
# allow disabling logging of certain packets
self.DISABLE_SPAMMERS = True
self.UDP_SPAMMERS = ['PacketAck', 'AgentUpdate']
# toggle handling a region's event queue
self.ENABLE_REGION_EVENT_QUEUE = True
# how many seconds to wait between polling
# a region's event queue
self.REGION_EVENT_QUEUE_POLL_INTERVAL = 1
if self.spammy_logging:
self.ENABLE_BYTES_TO_HEX_LOGGING = True
self.ENABLE_CAPS_LLSD_LOGGING = True
self.DISABLE_SPAMMERS = False
# override the defaults
if self.quiet_logging:
self.LOG_VERBOSE = False
self.ENABLE_BYTES_TO_HEX_LOGGING = False
self.ENABLE_CAPS_LOGGING = False
self.ENABLE_CAPS_LLSD_LOGGING = False
self.ENABLE_EQ_LOGGING = False
self.ENABLE_UDP_LOGGING = False
self.LOG_SKIPPED_PACKETS = False
self.ENABLE_OBJECT_LOGGING = False
self.ENABLE_HOST_LOGGING = False
self.LOG_COROUTINE_SPAWNS = False
self.DISABLE_SPAMMERS = True
# ~~~~~~~~~~~~~~~~~~~~~~
# Test related settings
# ~~~~~~~~~~~~~~~~~~~~~~
if log_tests:
self.ENABLE_LOGGING_IN_TESTS = True
else:
self.ENABLE_LOGGING_IN_TESTS = False
def set_setting(self, name: str, val: Any):
self._settings[name] = val

6
hippolyzer/lib/client/object_manager.py
View File

@@ -23,6 +23,7 @@ from hippolyzer.lib.base.objects import (
normalize_object_update_compressed,
Object, handle_to_global_pos,
)
from hippolyzer.lib.base.settings import Settings
from hippolyzer.lib.client.namecache import NameCache, NameCacheEntry
from hippolyzer.lib.client.state import BaseClientSession, BaseClientRegion
from hippolyzer.lib.base.templates import PCode, ObjectStateSerializer
@@ -162,13 +163,14 @@ class ClientObjectManager:
class ClientWorldObjectManager:
"""Manages Objects for a session's whole world"""
def __init__(self, session: BaseClientSession, name_cache: Optional[NameCache]):
def __init__(self, session: BaseClientSession, settings: Settings, name_cache: Optional[NameCache]):
self._session: BaseClientSession = session
self._settings = settings
self.name_cache = name_cache or NameCache()
self._fullid_lookup: Dict[UUID, Object] = {}
self._avatars: Dict[UUID, Avatar] = {}
self._avatar_objects: Dict[UUID, Object] = {}
self._region_managers: Dict[int, ClientObjectManager] = {}
self.name_cache = name_cache or NameCache()
message_handler = self._session.message_handler
message_handler.subscribe("ObjectUpdate", self._handle_object_update)
message_handler.subscribe("ImprovedTerseObjectUpdate",

8
hippolyzer/lib/proxy/caps_client.py
View File

@@ -1,20 +1,21 @@
from __future__ import annotations
import os
import re
import sys
from typing import *
from hippolyzer.lib.base.network.caps_client import CapsClient, CAPS_DICT
from hippolyzer.lib.proxy.settings import ProxySettings
if TYPE_CHECKING:
from hippolyzer.lib.proxy.region import ProxiedRegion
class ProxyCapsClient(CapsClient):
def __init__(self, region: Optional[ProxiedRegion] = None):
def __init__(self, settings: ProxySettings, region: Optional[ProxiedRegion] = None):
super().__init__(None)
self._region = region
self._settings = settings
def _get_caps(self) -> Optional[CAPS_DICT]:
if not self._region:
@@ -28,8 +29,7 @@ class ProxyCapsClient(CapsClient):
# request came from us so we can tag the request as injected. The header will be popped
# off before passing through to the server.
headers["X-Hippo-Injected"] = "1"
# TODO: Have a setting for this
proxy_port = int(os.environ.get("HIPPO_HTTP_PORT", 9062))
proxy_port = self._settings.HTTP_PROXY_PORT
proxy = f"http://127.0.0.1:{proxy_port}"
# TODO: set up the SSLContext to validate mitmproxy's cert
ssl = ssl or False

21
hippolyzer/lib/proxy/lludp_proxy.py
View File

@@ -5,7 +5,6 @@ from typing import Optional, Tuple
from hippolyzer.lib.base.message.message_dot_xml import MessageDotXML
from hippolyzer.lib.base.message.udpdeserializer import UDPMessageDeserializer
from hippolyzer.lib.base.message.udpserializer import UDPMessageSerializer
from hippolyzer.lib.base.settings import Settings
from hippolyzer.lib.proxy.addons import AddonManager
from hippolyzer.lib.base.network.transport import UDPPacket
from hippolyzer.lib.base.message.message import Message
@@ -26,17 +25,16 @@ class SLSOCKS5Server(SOCKS5Server):
return lambda: InterceptingLLUDPProxyProtocol(source_addr, self.session_manager)
class BaseLLUDPProxyProtocol(UDPProxyProtocol):
def __init__(self, source_addr: Tuple[str, int]):
class InterceptingLLUDPProxyProtocol(UDPProxyProtocol):
def __init__(self, source_addr: Tuple[str, int], session_manager: SessionManager):
super().__init__(source_addr)
self.settings = Settings()
self.settings.ENABLE_DEFERRED_PACKET_PARSING = True
self.settings.HANDLE_PACKETS = False
self.session_manager: SessionManager = session_manager
self.serializer = UDPMessageSerializer()
self.deserializer = UDPMessageDeserializer(
settings=self.settings,
settings=self.session_manager.settings,
)
self.message_xml = MessageDotXML()
self.session: Optional[Session] = None
def _ensure_message_allowed(self, msg: Message):
if not self.message_xml.validate_udp_msg(msg.name):
@@ -45,13 +43,6 @@ class BaseLLUDPProxyProtocol(UDPProxyProtocol):
)
raise PermissionError(f"UDPBanned message {msg.name}")
class InterceptingLLUDPProxyProtocol(BaseLLUDPProxyProtocol):
def __init__(self, source_addr: Tuple[str, int], session_manager: SessionManager):
super().__init__(source_addr)
self.session_manager: SessionManager = session_manager
self.session: Optional[Session] = None
def _handle_proxied_packet(self, packet: UDPPacket):
message: Optional[Message] = None
region: Optional[ProxiedRegion] = None
@@ -125,7 +116,7 @@ class InterceptingLLUDPProxyProtocol(BaseLLUDPProxyProtocol):
if message.name == "RegionHandshake":
region.cache_id = message["RegionInfo"]["CacheID"]
self.session.objects.track_region_objects(region.handle)
if self.session_manager.use_viewer_object_cache:
if self.session_manager.settings.USE_VIEWER_OBJECT_CACHE:
try:
region.objects.load_cache()
except:

11
hippolyzer/lib/proxy/object_manager.py
View File

@@ -13,6 +13,7 @@ from hippolyzer.lib.client.object_manager import (
from hippolyzer.lib.base.objects import Object
from hippolyzer.lib.proxy.addons import AddonManager
from hippolyzer.lib.proxy.http_flow import HippoHTTPFlow
from hippolyzer.lib.proxy.settings import ProxySettings
from hippolyzer.lib.proxy.vocache import RegionViewerObjectCacheChain
if TYPE_CHECKING:
@@ -31,15 +32,15 @@ class ProxyObjectManager(ClientObjectManager):
def __init__(
self,
region: ProxiedRegion,
use_vo_cache: bool = False
may_use_vo_cache: bool = False
):
super().__init__(region)
self.use_vo_cache = use_vo_cache
self.may_use_vo_cache = may_use_vo_cache
self.cache_loaded = False
self.object_cache = RegionViewerObjectCacheChain([])
def load_cache(self):
if not self.use_vo_cache or self.cache_loaded:
if not self.may_use_vo_cache or self.cache_loaded:
return
handle = self._region.handle
if not handle:
@@ -60,8 +61,8 @@ class ProxyObjectManager(ClientObjectManager):
class ProxyWorldObjectManager(ClientWorldObjectManager):
_session: Session
def __init__(self, session: Session, name_cache: Optional[NameCache]):
super().__init__(session, name_cache)
def __init__(self, session: Session, settings: ProxySettings, name_cache: Optional[NameCache]):
super().__init__(session, settings, name_cache)
session.http_message_handler.subscribe(
"GetObjectCost",
self._handle_get_object_cost

7
hippolyzer/lib/proxy/region.py
View File

@@ -49,7 +49,7 @@ class CapsMultiDict(multidict.MultiDict[Tuple[CapType, str]]):
class ProxiedRegion(BaseClientRegion):
def __init__(self, circuit_addr, seed_cap: str, session, handle=None):
def __init__(self, circuit_addr, seed_cap: str, session: Session, handle=None):
# A client may make a Seed request twice, and may get back two (valid!) sets of
# Cap URIs. We need to be able to look up both, so MultiDict is necessary.
self.handle: Optional[int] = handle
@@ -66,8 +66,9 @@ class ProxiedRegion(BaseClientRegion):
self.message_handler: MessageHandler[Message] = MessageHandler()
self.http_message_handler: MessageHandler[HippoHTTPFlow] = MessageHandler()
self.eq_manager = EventQueueManager(self)
self.caps_client = ProxyCapsClient(proxify(self))
self.objects: ProxyObjectManager = ProxyObjectManager(self, use_vo_cache=True)
settings = session.session_manager.settings
self.caps_client = ProxyCapsClient(settings, proxify(self))
self.objects: ProxyObjectManager = ProxyObjectManager(self, may_use_vo_cache=True)
self.xfer_manager = XferManager(proxify(self), self.session().secure_session_id)
self.transfer_manager = TransferManager(proxify(self), session.agent_id, session.id)
self._recalc_caps()

11
hippolyzer/lib/proxy/sessions.py
View File

@@ -19,6 +19,7 @@ from hippolyzer.lib.proxy.http_proxy import HTTPFlowContext, is_asset_server_cap
from hippolyzer.lib.proxy.namecache import ProxyNameCache
from hippolyzer.lib.proxy.object_manager import ProxyWorldObjectManager
from hippolyzer.lib.proxy.region import ProxiedRegion, CapType
from hippolyzer.lib.proxy.settings import ProxySettings
if TYPE_CHECKING:
from hippolyzer.lib.proxy.message_logger import BaseMessageLogger
@@ -27,7 +28,7 @@ if TYPE_CHECKING:
class Session(BaseClientSession):
def __init__(self, session_id, secure_session_id, agent_id, circuit_code,
login_data=None, session_manager: Optional[SessionManager] = None):
session_manager: Optional[SessionManager], login_data=None):
self.login_data = login_data or {}
self.pending = True
self.id: UUID = session_id
@@ -43,7 +44,7 @@ class Session(BaseClientSession):
self.started_at = datetime.datetime.now()
self.message_handler: MessageHandler[Message] = MessageHandler()
self.http_message_handler: MessageHandler[HippoHTTPFlow] = MessageHandler()
self.objects = ProxyWorldObjectManager(self, session_manager.name_cache)
self.objects = ProxyWorldObjectManager(self, session_manager.settings, session_manager.name_cache)
self._main_region = None
@property
@@ -59,8 +60,8 @@ class Session(BaseClientSession):
secure_session_id=UUID(login_data["secure_session_id"]),
agent_id=UUID(login_data["agent_id"]),
circuit_code=int(login_data["circuit_code"]),
login_data=login_data,
session_manager=session_manager,
login_data=login_data,
)
appearance_service = login_data.get("agent_appearance_service")
map_image_service = login_data.get("map-server-url")
@@ -168,7 +169,8 @@ class Session(BaseClientSession):
class SessionManager:
def __init__(self):
def __init__(self, settings: ProxySettings):
self.settings: ProxySettings = settings
self.sessions: List[Session] = []
self.shutdown_signal = multiprocessing.Event()
self.flow_context = HTTPFlowContext()
@@ -176,7 +178,6 @@ class SessionManager:
self.message_logger: Optional[BaseMessageLogger] = None
self.addon_ctx: Dict[str, Any] = {}
self.name_cache = ProxyNameCache()
self.use_viewer_object_cache: bool = False
def create_session(self, login_data) -> Session:
session = Session.from_login_data(login_data, self)

32
hippolyzer/lib/proxy/settings.py Normal file
View File

@@ -0,0 +1,32 @@
import os
from typing import *
from hippolyzer.lib.base.settings import Settings, SettingDescriptor
_T = TypeVar("_T")
class EnvSettingDescriptor(SettingDescriptor):
"""A setting that prefers to pull its value from the environment"""
__slots__ = ("_env_name", "_env_callable")
def __init__(self, default: Union[Callable[[], _T], _T], env_name: str, spec: Callable[[str], _T]):
super().__init__(default)
self._env_name = env_name
self._env_callable = spec
def __get__(self, obj, owner=None) -> _T:
val = os.getenv(self._env_name)
if val is not None:
return self._env_callable(val)
return super().__get__(obj, owner)
class ProxySettings(Settings):
SOCKS_PROXY_PORT: int = EnvSettingDescriptor(9061, "HIPPO_UDP_PORT", int)
HTTP_PROXY_PORT: int = EnvSettingDescriptor(9062, "HIPPO_HTTP_PORT", int)
PROXY_BIND_ADDR: str = EnvSettingDescriptor("127.0.0.1", "HIPPO_BIND_HOST", str)
REMOTELY_ACCESSIBLE: bool = SettingDescriptor(False)
USE_VIEWER_OBJECT_CACHE: bool = SettingDescriptor(False)
ADDON_SCRIPTS: List[str] = SettingDescriptor(list)
FILTERS: Dict[str, str] = SettingDescriptor(dict)

1
tests/base/test_message_wrapper.py
View File

@@ -46,7 +46,6 @@ class TestMessage(unittest.TestCase):
self.serial = UDPMessageSerializer()
settings = Settings()
settings.ENABLE_DEFERRED_PACKET_PARSING = True
settings.HANDLE_PACKETS = False
self.deserial = UDPMessageDeserializer(settings=settings)
def test_block(self):

32
tests/base/test_settings.py
View File

@@ -32,32 +32,6 @@ class TestEvents(unittest.TestCase):
def test_base_settings(self):
settings = Settings()
self.assertEqual(settings.quiet_logging, False)
self.assertEqual(settings.HANDLE_PACKETS, True)
self.assertEqual(settings.LOG_VERBOSE, True)
self.assertEqual(settings.ENABLE_BYTES_TO_HEX_LOGGING, False)
self.assertEqual(settings.ENABLE_CAPS_LOGGING, True)
self.assertEqual(settings.ENABLE_CAPS_LLSD_LOGGING, False)
self.assertEqual(settings.ENABLE_EQ_LOGGING, True)
self.assertEqual(settings.ENABLE_UDP_LOGGING, True)
self.assertEqual(settings.ENABLE_OBJECT_LOGGING, True)
self.assertEqual(settings.LOG_SKIPPED_PACKETS, True)
self.assertEqual(settings.ENABLE_HOST_LOGGING, True)
self.assertEqual(settings.LOG_COROUTINE_SPAWNS, True)
self.assertEqual(settings.DISABLE_SPAMMERS, True)
self.assertEqual(settings.UDP_SPAMMERS, ['PacketAck', 'AgentUpdate'])
def test_quiet_settings(self):
settings = Settings(True)
self.assertEqual(settings.quiet_logging, True)
self.assertEqual(settings.HANDLE_PACKETS, True)
self.assertEqual(settings.LOG_VERBOSE, False)
self.assertEqual(settings.ENABLE_BYTES_TO_HEX_LOGGING, False)
self.assertEqual(settings.ENABLE_CAPS_LOGGING, False)
self.assertEqual(settings.ENABLE_CAPS_LLSD_LOGGING, False)
self.assertEqual(settings.ENABLE_EQ_LOGGING, False)
self.assertEqual(settings.ENABLE_UDP_LOGGING, False)
self.assertEqual(settings.ENABLE_OBJECT_LOGGING, False)
self.assertEqual(settings.LOG_SKIPPED_PACKETS, False)
self.assertEqual(settings.ENABLE_HOST_LOGGING, False)
self.assertEqual(settings.LOG_COROUTINE_SPAWNS, False)
self.assertEqual(settings.ENABLE_DEFERRED_PACKET_PARSING, True)
settings.ENABLE_DEFERRED_PACKET_PARSING = False
self.assertEqual(settings.ENABLE_DEFERRED_PACKET_PARSING, False)

3
tests/proxy/__init__.py
View File

@@ -9,6 +9,7 @@ from hippolyzer.lib.proxy.lludp_proxy import InterceptingLLUDPProxyProtocol
from hippolyzer.lib.base.message.message import Message
from hippolyzer.lib.proxy.region import ProxiedRegion
from hippolyzer.lib.proxy.sessions import SessionManager
from hippolyzer.lib.proxy.settings import ProxySettings
from hippolyzer.lib.proxy.transport import SOCKS5UDPTransport
@@ -35,7 +36,7 @@ class BaseProxyTest(unittest.IsolatedAsyncioTestCase):
self.client_addr = ("127.0.0.1", 1)
self.region_addr = ("127.0.0.1", 3)
self.circuit_code = 1234
self.session_manager = SessionManager()
self.session_manager = SessionManager(ProxySettings())
self.session = self.session_manager.create_session({
"session_id": UUID.random(),
"secure_session_id": UUID.random(),

4
tests/proxy/test_message_filter.py
View File

@@ -11,6 +11,7 @@ from hippolyzer.lib.proxy.http_proxy import SerializedCapData
from hippolyzer.lib.proxy.message_logger import LLUDPMessageLogEntry, HTTPMessageLogEntry
from hippolyzer.lib.proxy.message_filter import compile_filter
from hippolyzer.lib.proxy.sessions import SessionManager
from hippolyzer.lib.proxy.settings import ProxySettings
OBJECT_UPDATE = b'\xc0\x00\x00\x00Q\x00\x0c\x00\x01\xea\x03\x00\x02\xe6\x03\x00\x01\xbe\xff\x01\x06\xbc\x8e\x0b\x00' \
b'\x01i\x94\x8cjM"\x1bf\xec\xe4\xac1c\x93\xcbKW\x89\x98\x01\t\x03\x00\x01Q@\x88>Q@\x88>Q@\x88><\xa2D' \
@@ -111,7 +112,6 @@ class MessageFilterTests(unittest.TestCase):
def test_tagged_union_subfield(self):
settings = Settings()
settings.ENABLE_DEFERRED_PACKET_PARSING = False
settings.HANDLE_PACKETS = False
deser = UDPMessageDeserializer(settings=settings)
update_msg = deser.deserialize(OBJECT_UPDATE)
entry = LLUDPMessageLogEntry(update_msg, None, None)
@@ -119,7 +119,7 @@ class MessageFilterTests(unittest.TestCase):
self.assertTrue(self._filter_matches("ObjectUpdate.ObjectData.ObjectData.Position < (90, 43, 27)", entry))
def test_http_flow(self):
session_manager = SessionManager()
session_manager = SessionManager(ProxySettings())
fake_flow = tflow.tflow(req=tutils.treq(), resp=tutils.tresp())
fake_flow.metadata["cap_data_ser"] = SerializedCapData(
cap_name="FakeCap",