diff --git a/hippolyzer/apps/proxy.py b/hippolyzer/apps/proxy.py index d1e4cae..260322e 100644 --- a/hippolyzer/apps/proxy.py +++ b/hippolyzer/apps/proxy.py @@ -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__": diff --git a/hippolyzer/apps/proxy_gui.py b/hippolyzer/apps/proxy_gui.py index 8ea40fc..71f3e05 100644 --- a/hippolyzer/apps/proxy_gui.py +++ b/hippolyzer/apps/proxy_gui.py @@ -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, diff --git a/hippolyzer/lib/base/settings.py b/hippolyzer/lib/base/settings.py index 156b937..9d751b1 100644 --- a/hippolyzer/lib/base/settings.py +++ b/hippolyzer/lib/base/settings.py @@ -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 diff --git a/hippolyzer/lib/client/object_manager.py b/hippolyzer/lib/client/object_manager.py index 4020522..56bd3a5 100644 --- a/hippolyzer/lib/client/object_manager.py +++ b/hippolyzer/lib/client/object_manager.py @@ -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", diff --git a/hippolyzer/lib/proxy/caps_client.py b/hippolyzer/lib/proxy/caps_client.py index bd70cae..db88e8a 100644 --- a/hippolyzer/lib/proxy/caps_client.py +++ b/hippolyzer/lib/proxy/caps_client.py @@ -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 diff --git a/hippolyzer/lib/proxy/lludp_proxy.py b/hippolyzer/lib/proxy/lludp_proxy.py index 065046b..bc77630 100644 --- a/hippolyzer/lib/proxy/lludp_proxy.py +++ b/hippolyzer/lib/proxy/lludp_proxy.py @@ -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: diff --git a/hippolyzer/lib/proxy/object_manager.py b/hippolyzer/lib/proxy/object_manager.py index c378a24..01d0f4e 100644 --- a/hippolyzer/lib/proxy/object_manager.py +++ b/hippolyzer/lib/proxy/object_manager.py @@ -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 diff --git a/hippolyzer/lib/proxy/region.py b/hippolyzer/lib/proxy/region.py index d74ed23..ff37cb6 100644 --- a/hippolyzer/lib/proxy/region.py +++ b/hippolyzer/lib/proxy/region.py @@ -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() diff --git a/hippolyzer/lib/proxy/sessions.py b/hippolyzer/lib/proxy/sessions.py index c6065b5..f3b3a29 100644 --- a/hippolyzer/lib/proxy/sessions.py +++ b/hippolyzer/lib/proxy/sessions.py @@ -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) diff --git a/hippolyzer/lib/proxy/settings.py b/hippolyzer/lib/proxy/settings.py new file mode 100644 index 0000000..4ae166f --- /dev/null +++ b/hippolyzer/lib/proxy/settings.py @@ -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) diff --git a/tests/base/test_message_wrapper.py b/tests/base/test_message_wrapper.py index f9a3003..0085719 100644 --- a/tests/base/test_message_wrapper.py +++ b/tests/base/test_message_wrapper.py @@ -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): diff --git a/tests/base/test_settings.py b/tests/base/test_settings.py index ff175cf..738875a 100644 --- a/tests/base/test_settings.py +++ b/tests/base/test_settings.py @@ -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) diff --git a/tests/proxy/__init__.py b/tests/proxy/__init__.py index f1a7faf..f6b9d17 100644 --- a/tests/proxy/__init__.py +++ b/tests/proxy/__init__.py @@ -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(), diff --git a/tests/proxy/test_message_filter.py b/tests/proxy/test_message_filter.py index bf513fe..4ec2887 100644 --- a/tests/proxy/test_message_filter.py +++ b/tests/proxy/test_message_filter.py @@ -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",