From a94854d00e344968fd8e1c29bb99f91faeed00fb Mon Sep 17 00:00:00 2001 From: Sogoss <> Date: Mon, 20 Jul 2026 15:31:56 +0200 Subject: [PATCH 1/5] feat(nfproxy): add HttpHistory support for pyfilters in HTTP requests --- fgex-lib/firegex/nfproxy/__init__.py | 5 +- fgex-lib/firegex/nfproxy/models/__init__.py | 7 + fgex-lib/firegex/nfproxy/models/http.py | 126 ++++++++++++++++- tests/test_http_history.py | 146 ++++++++++++++++++++ 4 files changed, 277 insertions(+), 7 deletions(-) create mode 100644 tests/test_http_history.py diff --git a/fgex-lib/firegex/nfproxy/__init__.py b/fgex-lib/firegex/nfproxy/__init__.py index ae5d3e8..99ad65a 100644 --- a/fgex-lib/firegex/nfproxy/__init__.py +++ b/fgex-lib/firegex/nfproxy/__init__.py @@ -35,7 +35,8 @@ def clear_pyfilter_registry(): pyfilter.registry.clear() __all__ = [ - "ACCEPT", "DROP", "REJECT", "UNSTABLE_MANGLE" + "ACCEPT", "DROP", "REJECT", "UNSTABLE_MANGLE", "Action", "FullStreamAction", "pyfilter", - "RawPacket", "TCPInputStream", "TCPOutputStream", "TCPClientStream", "TCPServerStream" + "RawPacket", "TCPInputStream", "TCPOutputStream", "TCPClientStream", "TCPServerStream", + "HttpHistory", "HttpStreamHistory" ] \ No newline at end of file diff --git a/fgex-lib/firegex/nfproxy/models/__init__.py b/fgex-lib/firegex/nfproxy/models/__init__.py index c990fcf..c5dffb2 100644 --- a/fgex-lib/firegex/nfproxy/models/__init__.py +++ b/fgex-lib/firegex/nfproxy/models/__init__.py @@ -11,6 +11,8 @@ from firegex.nfproxy.models.http import ( HttpResponseHeader, HttpFullRequest, HttpFullResponse, + HttpHistory, + HttpStreamHistory, ) from firegex.nfproxy.internals.data import RawPacket from enum import Enum @@ -31,6 +33,8 @@ type_annotations_associations = { HttpResponseHeader: HttpResponseHeader._fetch_packet, HttpFullRequest: HttpFullRequest._fetch_packet, HttpFullResponse: HttpFullResponse._fetch_packet, + HttpHistory: HttpHistory._fetch_packet, + HttpStreamHistory: HttpStreamHistory._fetch_packet, }, } @@ -52,5 +56,8 @@ __all__ = [ "HttpResponseHeader", "HttpFullRequest", "HttpFullResponse", + "HttpHistory", + "HttpStreamHistory", "Protocols", ] + diff --git a/fgex-lib/firegex/nfproxy/models/http.py b/fgex-lib/firegex/nfproxy/models/http.py index 305f694..39f3f26 100644 --- a/fgex-lib/firegex/nfproxy/models/http.py +++ b/fgex-lib/firegex/nfproxy/models/http.py @@ -48,6 +48,7 @@ class InternalHTTPMessage: ws_stream: list[Frame] = field(default_factory=list) # Decoded websocket stream upgrading_to_h2: bool = field(default=False) upgrading_to_ws: bool = field(default=False) + added_to_history: bool = field(default=False) @dataclass @@ -393,6 +394,59 @@ class InternalHttpResponse(InternalCallbackHandler, pyllhttp.Response): return False +class HttpHistory: + """ + HTTP History handler for pyfilters. + Provides access to completed previous requests and responses in the current TCP stream. + """ + + def __init__( + self, + requests: list["HttpFullRequest"] | None = None, + responses: list["HttpFullResponse"] | None = None, + ): + self._requests = list(requests) if requests is not None else [] + self._responses = list(responses) if responses is not None else [] + + @property + def requests(self) -> list["HttpFullRequest"]: + """List of previous completed HTTP requests""" + return self._requests + + @property + def responses(self) -> list["HttpFullResponse"]: + """List of previous completed HTTP responses""" + return self._responses + + @classmethod + def _fetch_packet(cls, internal_data: DataStreamCtx): + if ( + internal_data.current_pkt is None + or internal_data.current_pkt.is_tcp is False + ): + raise NotReadyToRun() + + for key in ["http_module", "http_full", "http_header"]: + obj = internal_data.call_mem.get(f"_fetched_obj_{key}") + if obj is not None: + if isinstance(obj, list) and len(obj) > 0: + return [item.history for item in obj] + elif hasattr(obj, "history"): + return obj.history + + fetch_cls = HttpRequest if internal_data.current_pkt.is_input else HttpResponse + obj = fetch_cls._fetch_packet(internal_data) + if isinstance(obj, list): + return [item.history for item in obj] + return obj.history + + def __repr__(self): + return f"" + + +HttpStreamHistory = HttpHistory + + class InternalBasicHttpMetaClass: """Internal class to handle HTTP requests and responses""" @@ -404,11 +458,19 @@ class InternalBasicHttpMetaClass: self._parser = parser self.raised_error = False self._message: InternalHTTPMessage | None = msg + self._history: HttpHistory | None = None self._contructor_hook() def _contructor_hook(self): pass + @property + def history(self) -> HttpHistory: + """HTTP History for the current stream connection""" + if self._history is None: + return HttpHistory([], []) + return self._history + @property def total_size(self) -> int: """Total size of the stream""" @@ -493,6 +555,10 @@ class InternalBasicHttpMetaClass: """Get a header from the message without caring about the case""" return self._message.lheaders.get(header.lower(), default) + @classmethod + def _should_release_message_headers(cls) -> bool: + return True + @staticmethod def _before_fetch_callable_checks(internal_data: DataStreamCtx) -> bool: raise NotImplementedError() @@ -521,6 +587,9 @@ class InternalBasicHttpMetaClass: parser: InternalHttpRequest | InternalHttpResponse = ParserType() internal_data.data_handler_context[parser_key] = parser + parser.release_message_headers = cls._should_release_message_headers() + + if not internal_data.call_mem.get( cls._parser_class(), False ): # Need to parse HTTP @@ -620,10 +689,47 @@ class InternalBasicHttpMetaClass: if messages_to_call == 0: raise NotReadyToRun() - elif messages_to_call == 1: - return cls(parser, messages_tosend[0]) - return [cls(parser, ele) for ele in messages_tosend] + max_history = int(internal_data.filter_glob.get("FGEX_MAX_HISTORY_SIZE", 100)) + req_history_deque: deque = ( + internal_data.data_handler_context.setdefault( + "http_history_requests", deque(maxlen=max_history) + ) + ) + resp_history_deque: deque = ( + internal_data.data_handler_context.setdefault( + "http_history_responses", deque(maxlen=max_history) + ) + ) + + built_instances = [] + for msg in messages_tosend: + history_snapshot = HttpHistory( + list(req_history_deque), list(resp_history_deque) + ) + instance = cls(parser, msg) + instance._history = history_snapshot + built_instances.append(instance) + + if msg.message_complete and not getattr(msg, "added_to_history", False): + msg.added_to_history = True + if internal_data.current_pkt.is_input: + full_req = HttpFullRequest(parser, msg) + full_req._history = history_snapshot + req_history_deque.append(full_req) + else: + full_resp = HttpFullResponse(parser, msg) + full_resp._history = history_snapshot + resp_history_deque.append(full_resp) + + if len(built_instances) == 1: + res = built_instances[0] + internal_data.call_mem[f"_fetched_obj_{cls._parser_class()}"] = res + return res + + internal_data.call_mem[f"_fetched_obj_{cls._parser_class()}"] = built_instances + return built_instances + class HttpRequest(InternalBasicHttpMetaClass): @@ -639,7 +745,7 @@ class HttpRequest(InternalBasicHttpMetaClass): @property def method(self) -> bytes: """Method of the request""" - return self._parser.msg.method + return self._message.method @staticmethod def _parser_class() -> str: @@ -662,7 +768,8 @@ class HttpResponse(InternalBasicHttpMetaClass): @property def status_code(self) -> int: """Status code of the response""" - return self._parser.msg.status + return self._message.status + @staticmethod def _parser_class() -> str: @@ -678,6 +785,10 @@ class HttpFullRequest(HttpRequest): This data handler will be called when the request data is complete """ + @classmethod + def _should_release_message_headers(cls) -> bool: + return False + def _contructor_hook(self): self._parser.release_message_headers = False @@ -695,6 +806,10 @@ class HttpFullResponse(HttpResponse): This data handler will be called when the response data is complete """ + @classmethod + def _should_release_message_headers(cls) -> bool: + return False + def _contructor_hook(self): self._parser.release_message_headers = False @@ -706,6 +821,7 @@ class HttpFullResponse(HttpResponse): return f"" + class HttpRequestHeader(HttpRequest): """ HTTP Request Header handler diff --git a/tests/test_http_history.py b/tests/test_http_history.py new file mode 100644 index 0000000..0f1cc9e --- /dev/null +++ b/tests/test_http_history.py @@ -0,0 +1,146 @@ +from firegex.nfproxy import pyfilter, ACCEPT, clear_pyfilter_registry +from firegex.nfproxy.models import ( + HttpRequest, + HttpResponse, + HttpFullRequest, + HttpFullResponse, + HttpHistory, + HttpStreamHistory, +) +from firegex.nfproxy.internals import compile, handle_packet + + +def create_packet_info(payload: bytes, is_input: bool): + return { + "data": payload, + "raw_packet": b"\x00" * 40 + payload, + "is_input": is_input, + "is_ipv6": False, + "is_tcp": True, + "l4_size": len(payload), + } + + +def test_http_history_model(): + req1 = None + res1 = None + history_obj = HttpHistory(requests=[req1], responses=[res1]) + assert history_obj.requests == [req1] + assert history_obj.responses == [res1] + assert "HttpHistory" in repr(history_obj) + + +def test_http_history_stream_execution(): + glob = {} + clear_pyfilter_registry() + + code = """ +from firegex.nfproxy import pyfilter, ACCEPT +from firegex.nfproxy.models import HttpFullRequest, HttpFullResponse, HttpHistory + +recorded_history = [] + +@pyfilter +def filter_req(req: HttpFullRequest, history: HttpHistory): + req_urls = [r.url for r in history.requests] + resp_statuses = [r.status_code for r in history.responses] + recorded_history.append(("req", req.url, req_urls, resp_statuses)) + return ACCEPT + +@pyfilter +def filter_resp(resp: HttpFullResponse, history: HttpHistory): + req_urls = [r.url for r in history.requests] + resp_statuses = [r.status_code for r in history.responses] + recorded_history.append(("resp", resp.status_code, req_urls, resp_statuses)) + return ACCEPT +""" + + glob["__firegex_pyfilter_enabled"] = ["filter_req", "filter_resp"] + glob["__firegex_proto"] = "http" + exec(code, glob, glob) + compile(glob) + + # 1. First Request + req1_data = b"GET /first HTTP/1.1\r\nHost: example.com\r\n\r\n" + glob["__firegex_packet_info"] = create_packet_info(req1_data, is_input=True) + handle_packet(glob) + + # 2. First Response + resp1_data = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK" + glob["__firegex_packet_info"] = create_packet_info(resp1_data, is_input=False) + handle_packet(glob) + + # 3. Second Request + req2_data = b"GET /second HTTP/1.1\r\nHost: example.com\r\n\r\n" + glob["__firegex_packet_info"] = create_packet_info(req2_data, is_input=True) + handle_packet(glob) + + # 4. Second Response + resp2_data = b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n" + glob["__firegex_packet_info"] = create_packet_info(resp2_data, is_input=False) + handle_packet(glob) + + recorded = glob["recorded_history"] + + # 1. First Request: history is empty + assert recorded[0] == ("req", "/first", [], []) + + # 2. First Response: past requests contains /first + assert recorded[1] == ("resp", "OK", ["/first"], []) + + # 3. Second Request: past requests contains [/first], past responses contains [OK] + assert recorded[2] == ("req", "/second", ["/first"], ["OK"]) + + # 4. Second Response: past requests contains [/first, /second], past responses contains [OK] + assert recorded[3] == ("resp", "Not Found", ["/first", "/second"], ["OK"]) + + +def test_http_history_attribute_on_request(): + glob = {} + clear_pyfilter_registry() + + code = """ +from firegex.nfproxy import pyfilter, ACCEPT +from firegex.nfproxy.models import HttpFullRequest, HttpFullResponse + +history_lengths = [] + +@pyfilter +def filter_req(req: HttpFullRequest): + history_lengths.append((len(req.history.requests), len(req.history.responses))) + return ACCEPT + +@pyfilter +def filter_resp(resp: HttpFullResponse): + history_lengths.append((len(resp.history.requests), len(resp.history.responses))) + return ACCEPT +""" + + glob["__firegex_pyfilter_enabled"] = ["filter_req", "filter_resp"] + glob["__firegex_proto"] = "http" + exec(code, glob, glob) + compile(glob) + + # Req 1 + glob["__firegex_packet_info"] = create_packet_info(b"GET /a HTTP/1.1\r\nHost: a\r\n\r\n", is_input=True) + handle_packet(glob) + + # Resp 1 + glob["__firegex_packet_info"] = create_packet_info(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n", is_input=False) + handle_packet(glob) + + # Req 2 + glob["__firegex_packet_info"] = create_packet_info(b"GET /b HTTP/1.1\r\nHost: b\r\n\r\n", is_input=True) + handle_packet(glob) + + lengths = glob["history_lengths"] + assert lengths[0] == (0, 0) + assert lengths[1] == (1, 0) + assert lengths[2] == (1, 1) + + +if __name__ == "__main__": + test_http_history_model() + test_http_history_stream_execution() + test_http_history_attribute_on_request() + print("All HTTP history unit tests passed successfully!") From dbc8b176b17651c843933930245f56347bcadddf Mon Sep 17 00:00:00 2001 From: Sogoss <> Date: Mon, 20 Jul 2026 16:03:29 +0200 Subject: [PATCH 2/5] fix(nfproxy): fix HttpHistory parameter ordering and state immutability - Fix parameter order sensitivity in HttpHistory._fetch_packet when history is first argument - Delegate total_size and should_upgrade to _message instead of live _parser - Sanitize FGEX_MAX_HISTORY_SIZE with max(0, int(val)) and dynamically sync deque maxlen --- fgex-lib/firegex/nfproxy/models/http.py | 38 ++++--- tests/test_http_history.py | 131 ++++++++++++++++++++++++ 2 files changed, 157 insertions(+), 12 deletions(-) diff --git a/fgex-lib/firegex/nfproxy/models/http.py b/fgex-lib/firegex/nfproxy/models/http.py index 39f3f26..6ffcd4e 100644 --- a/fgex-lib/firegex/nfproxy/models/http.py +++ b/fgex-lib/firegex/nfproxy/models/http.py @@ -426,19 +426,21 @@ class HttpHistory: ): raise NotReadyToRun() - for key in ["http_module", "http_full", "http_header"]: + for key in ("http_module", "http_full", "http_header"): obj = internal_data.call_mem.get(f"_fetched_obj_{key}") if obj is not None: - if isinstance(obj, list) and len(obj) > 0: + if isinstance(obj, list) and obj: return [item.history for item in obj] elif hasattr(obj, "history"): return obj.history - fetch_cls = HttpRequest if internal_data.current_pkt.is_input else HttpResponse - obj = fetch_cls._fetch_packet(internal_data) - if isinstance(obj, list): - return [item.history for item in obj] - return obj.history + req_history_deque = internal_data.data_handler_context.get( + "http_history_requests", deque() + ) + resp_history_deque = internal_data.data_handler_context.get( + "http_history_responses", deque() + ) + return HttpHistory(list(req_history_deque), list(resp_history_deque)) def __repr__(self): return f"" @@ -473,8 +475,8 @@ class InternalBasicHttpMetaClass: @property def total_size(self) -> int: - """Total size of the stream""" - return self._parser.total_size + """Total size of the message""" + return self._message.total_size @property def url(self) -> str | None: @@ -524,7 +526,7 @@ class InternalBasicHttpMetaClass: @property def should_upgrade(self) -> bool: """If the message should upgrade""" - return self._parser.should_upgrade + return self._message.should_upgrade @property def content_length(self) -> int | None: @@ -690,17 +692,29 @@ class InternalBasicHttpMetaClass: if messages_to_call == 0: raise NotReadyToRun() - max_history = int(internal_data.filter_glob.get("FGEX_MAX_HISTORY_SIZE", 100)) + raw_max = internal_data.filter_glob.get("FGEX_MAX_HISTORY_SIZE", 100) + try: + max_history = max(0, int(raw_max)) + except (ValueError, TypeError): + max_history = 100 + req_history_deque: deque = ( internal_data.data_handler_context.setdefault( "http_history_requests", deque(maxlen=max_history) ) ) + if req_history_deque.maxlen != max_history: + req_history_deque = deque(req_history_deque, maxlen=max_history) + internal_data.data_handler_context["http_history_requests"] = req_history_deque + resp_history_deque: deque = ( internal_data.data_handler_context.setdefault( "http_history_responses", deque(maxlen=max_history) ) ) + if resp_history_deque.maxlen != max_history: + resp_history_deque = deque(resp_history_deque, maxlen=max_history) + internal_data.data_handler_context["http_history_responses"] = resp_history_deque built_instances = [] for msg in messages_tosend: @@ -711,7 +725,7 @@ class InternalBasicHttpMetaClass: instance._history = history_snapshot built_instances.append(instance) - if msg.message_complete and not getattr(msg, "added_to_history", False): + if msg.message_complete and not msg.added_to_history: msg.added_to_history = True if internal_data.current_pkt.is_input: full_req = HttpFullRequest(parser, msg) diff --git a/tests/test_http_history.py b/tests/test_http_history.py index 0f1cc9e..e7514e5 100644 --- a/tests/test_http_history.py +++ b/tests/test_http_history.py @@ -139,8 +139,139 @@ def filter_resp(resp: HttpFullResponse): assert lengths[2] == (1, 1) +def test_http_history_invalid_max_size(): + glob = {} + clear_pyfilter_registry() + + code = """ +from firegex.nfproxy import pyfilter, ACCEPT +from firegex.nfproxy.models import HttpFullRequest + +@pyfilter +def filter_req(req: HttpFullRequest): + return ACCEPT +""" + + glob["__firegex_pyfilter_enabled"] = ["filter_req"] + glob["__firegex_proto"] = "http" + glob["FGEX_MAX_HISTORY_SIZE"] = "invalid_number" + exec(code, glob, glob) + compile(glob) + + glob["__firegex_packet_info"] = create_packet_info(b"GET /test HTTP/1.1\r\nHost: test\r\n\r\n", is_input=True) + # Should not raise ValueError, should default to 100 + handle_packet(glob) + + +def test_http_history_parameter_order(): + glob = {} + clear_pyfilter_registry() + + code = """ +from firegex.nfproxy import pyfilter, ACCEPT +from firegex.nfproxy.models import HttpFullRequest, HttpHistory + +calls_log = [] + +@pyfilter +def filter_history_first(history: HttpHistory, req: HttpFullRequest): + calls_log.append(("history_first", req.url, len(history.requests))) + return ACCEPT +""" + + glob["__firegex_pyfilter_enabled"] = ["filter_history_first"] + glob["__firegex_proto"] = "http" + exec(code, glob, glob) + compile(glob) + + glob["__firegex_packet_info"] = create_packet_info(b"GET /first HTTP/1.1\r\nHost: test\r\n\r\n", is_input=True) + handle_packet(glob) + + calls = glob["calls_log"] + # Must be called EXACTLY ONCE with 0 history elements for the first request + assert len(calls) == 1 + assert calls[0] == ("history_first", "/first", 0) + + +def test_http_history_property_immutability(): + glob = {} + clear_pyfilter_registry() + immutability_log = [] + + code = """ +from firegex.nfproxy import pyfilter, ACCEPT +from firegex.nfproxy.models import HttpFullRequest, HttpHistory + +@pyfilter +def filter_req(req: HttpFullRequest, history: HttpHistory): + if len(history.requests) > 0: + past = history.requests[0] + immutability_log.append({ + "past_size": past.total_size, + "past_url": past.url, + "curr_size": req.total_size, + "curr_url": req.url, + }) + return ACCEPT +""" + + glob["__firegex_pyfilter_enabled"] = ["filter_req"] + glob["__firegex_proto"] = "http" + glob["immutability_log"] = immutability_log + exec(code, glob, glob) + compile(glob) + + # 1. First short request + req1_data = b"GET /short HTTP/1.1\r\nHost: test\r\n\r\n" + glob["__firegex_packet_info"] = create_packet_info(req1_data, is_input=True) + handle_packet(glob) + + # 2. Second longer request + req2_data = b"POST /large_path HTTP/1.1\r\nHost: test\r\nContent-Length: 5\r\n\r\nHELLO" + glob["__firegex_packet_info"] = create_packet_info(req2_data, is_input=True) + handle_packet(glob) + + entries = glob["immutability_log"] + assert len(entries) == 1 + # Historical request 1 size must equal 14 (/short), current size 39 (/large_path) + assert entries[0]["past_size"] == 14 + assert entries[0]["past_url"] == "/short" + assert entries[0]["curr_size"] == 39 + assert entries[0]["curr_url"] == "/large_path" + + +def test_http_history_negative_max_size(): + glob = {} + clear_pyfilter_registry() + + code = """ +from firegex.nfproxy import pyfilter, ACCEPT +from firegex.nfproxy.models import HttpFullRequest + +@pyfilter +def filter_req(req: HttpFullRequest): + return ACCEPT +""" + + glob["__firegex_pyfilter_enabled"] = ["filter_req"] + glob["__firegex_proto"] = "http" + glob["FGEX_MAX_HISTORY_SIZE"] = "-5" + exec(code, glob, glob) + compile(glob) + + glob["__firegex_packet_info"] = create_packet_info(b"GET /test HTTP/1.1\r\nHost: test\r\n\r\n", is_input=True) + # Should not raise ValueError for maxlen=-5, should clamp to 0 + handle_packet(glob) + + if __name__ == "__main__": test_http_history_model() test_http_history_stream_execution() test_http_history_attribute_on_request() + test_http_history_invalid_max_size() + test_http_history_parameter_order() + test_http_history_property_immutability() + test_http_history_negative_max_size() print("All HTTP history unit tests passed successfully!") + + From c71ad7c1991c6973de5ab9a908b1b502e014813e Mon Sep 17 00:00:00 2001 From: Sogoss <> Date: Mon, 20 Jul 2026 16:14:07 +0200 Subject: [PATCH 3/5] fix(nfproxy): export HttpHistory in nfproxy and ensure list immutability in getters --- fgex-lib/firegex/nfproxy/__init__.py | 10 +++++++++- fgex-lib/firegex/nfproxy/models/http.py | 4 ++-- tests/test_http_history.py | 22 ++++++++++++++++++++++ 3 files changed, 33 insertions(+), 3 deletions(-) diff --git a/fgex-lib/firegex/nfproxy/__init__.py b/fgex-lib/firegex/nfproxy/__init__.py index 99ad65a..67fc9e2 100644 --- a/fgex-lib/firegex/nfproxy/__init__.py +++ b/fgex-lib/firegex/nfproxy/__init__.py @@ -1,5 +1,13 @@ import functools -from firegex.nfproxy.models import RawPacket, TCPInputStream, TCPOutputStream, TCPClientStream, TCPServerStream +from firegex.nfproxy.models import ( + RawPacket, + TCPInputStream, + TCPOutputStream, + TCPClientStream, + TCPServerStream, + HttpHistory, + HttpStreamHistory, +) from firegex.nfproxy.internals.models import Action, FullStreamAction ACCEPT = Action.ACCEPT diff --git a/fgex-lib/firegex/nfproxy/models/http.py b/fgex-lib/firegex/nfproxy/models/http.py index 6ffcd4e..45ec83b 100644 --- a/fgex-lib/firegex/nfproxy/models/http.py +++ b/fgex-lib/firegex/nfproxy/models/http.py @@ -411,12 +411,12 @@ class HttpHistory: @property def requests(self) -> list["HttpFullRequest"]: """List of previous completed HTTP requests""" - return self._requests + return self._requests.copy() @property def responses(self) -> list["HttpFullResponse"]: """List of previous completed HTTP responses""" - return self._responses + return self._responses.copy() @classmethod def _fetch_packet(cls, internal_data: DataStreamCtx): diff --git a/tests/test_http_history.py b/tests/test_http_history.py index e7514e5..80a9f36 100644 --- a/tests/test_http_history.py +++ b/tests/test_http_history.py @@ -264,6 +264,25 @@ def filter_req(req: HttpFullRequest): handle_packet(glob) +def test_http_history_nfproxy_import(): + from firegex.nfproxy import HttpHistory as HH1, HttpStreamHistory as HH2 + assert HH1 is HttpHistory + assert HH2 is HttpStreamHistory + + +def test_http_history_list_mutation(): + history_obj = HttpHistory(requests=["dummy_req"], responses=["dummy_resp"]) + reqs = history_obj.requests + reqs.clear() + assert len(history_obj.requests) == 1 + assert history_obj.requests == ["dummy_req"] + + resps = history_obj.responses + resps.append("extra_resp") + assert len(history_obj.responses) == 1 + assert history_obj.responses == ["dummy_resp"] + + if __name__ == "__main__": test_http_history_model() test_http_history_stream_execution() @@ -272,6 +291,9 @@ if __name__ == "__main__": test_http_history_parameter_order() test_http_history_property_immutability() test_http_history_negative_max_size() + test_http_history_nfproxy_import() + test_http_history_list_mutation() print("All HTTP history unit tests passed successfully!") + From 9e4f93455bbb7d5d29a556b2a4275ff40611e3f9 Mon Sep 17 00:00:00 2001 From: Sogoss <> Date: Tue, 21 Jul 2026 15:19:30 +0200 Subject: [PATCH 4/5] =?UTF-8?q?fix(nfproxy):=20prevent=20O(N=C2=B2)=20memo?= =?UTF-8?q?ry=20growth=20from=20nested=20history=20snapshots?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Objects archived in the history deques no longer carry their own _history snapshot, breaking the recursive chain where each historical entry held copies of all prior entries. The .history property on archived objects now returns an empty HttpHistory([], []) instead. --- fgex-lib/firegex/nfproxy/models/http.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/fgex-lib/firegex/nfproxy/models/http.py b/fgex-lib/firegex/nfproxy/models/http.py index 45ec83b..7967ef8 100644 --- a/fgex-lib/firegex/nfproxy/models/http.py +++ b/fgex-lib/firegex/nfproxy/models/http.py @@ -728,13 +728,9 @@ class InternalBasicHttpMetaClass: if msg.message_complete and not msg.added_to_history: msg.added_to_history = True if internal_data.current_pkt.is_input: - full_req = HttpFullRequest(parser, msg) - full_req._history = history_snapshot - req_history_deque.append(full_req) + req_history_deque.append(HttpFullRequest(parser, msg)) else: - full_resp = HttpFullResponse(parser, msg) - full_resp._history = history_snapshot - resp_history_deque.append(full_resp) + resp_history_deque.append(HttpFullResponse(parser, msg)) if len(built_instances) == 1: res = built_instances[0] From 53656d531cf6eba6f1903e1e04a5caa27386e19e Mon Sep 17 00:00:00 2001 From: Domingo Dirutigliano Date: Wed, 22 Jul 2026 15:10:31 +0200 Subject: [PATCH 5/5] minor fixes and integrated test --- frontend/src/App.tsx | 71 +++++--- frontend/src/js/models.ts | 3 +- frontend/src/js/store.ts | 10 + frontend/src/js/utils.tsx | 7 +- tests/nfproxy_test.py | 46 +++++ tests/stress_test_history.py | 344 +++++++++++++++++++++++++++++++++++ 6 files changed, 452 insertions(+), 29 deletions(-) create mode 100755 tests/stress_test_history.py diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 49ba445..23ee983 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -13,7 +13,7 @@ import { Firewall } from './pages/Firewall'; import { useQueryClient } from '@tanstack/react-query'; import NFProxy from './pages/NFProxy'; import ServiceDetailsNFProxy from './pages/NFProxy/ServiceDetails'; -import { useAuthStore } from './js/store'; +import { useAuthStore, useSystemStore } from './js/store'; function App() { @@ -24,40 +24,57 @@ function App() { const [loadinBtn, setLoadingBtn] = useState(false); const queryClient = useQueryClient() const { access_token } = useAuthStore() + const { setVersion } = useSystemStore() - useEffect(()=>{ - socketio.auth = { token: access_token || "" } - socketio.connect() - getStatus() - socketio.on("update", (data) => { - queryClient.invalidateQueries({ queryKey: data }) - }) - socketio.on("connect_error", (err) => { - if (access_token){ - errorNotify("Socket.Io connection failed! ",`Error message: [${err.message}]`) - } - getStatus() - }); - return () => { - socketio.off("update") - socketio.off("connect_error") - socketio.disconnect() - } - },[access_token]) - - const getStatus = () =>{ - getstatus().then( res =>{ + const getStatus = () => { + getstatus().then(res => { setSystemStatus(res) + setVersion(res.version || "unknown") + if (!res.loggined && useAuthStore.getState().getAccessToken()) { + useAuthStore.getState().clearAccessToken() + } setReqError(undefined) - }).catch(err=>{ + }).catch(err => { setReqError(err.toString()) setTimeout(getStatus, 500) - }).finally( ()=>setLoading(false) ) + }).finally(() => setLoading(false)) } - useEffect(()=>{ + useEffect(() => { getStatus() - },[]) + }, []) + + useEffect(() => { + if (access_token && systemStatus.loggined) { + socketio.auth = { token: access_token } + if (!socketio.connected) { + socketio.connect() + } + } else { + if (socketio.connected) { + socketio.disconnect() + } + } + }, [access_token, systemStatus.loggined]) + + useEffect(() => { + socketio.on("update", (data) => { + queryClient.invalidateQueries({ queryKey: data }) + }) + socketio.on("connect_error", (err) => { + if (err.message === "Unauthorized") { + useAuthStore.getState().clearAccessToken() + getStatus() + } else if (access_token && systemStatus.loggined) { + errorNotify("Socket.Io connection failed! ", `Error message: [${err.message}]`) + getStatus() + } + }); + return () => { + socketio.off("update") + socketio.off("connect_error") + } + }, [access_token, systemStatus.loggined]) const form = useForm({ initialValues: { diff --git a/frontend/src/js/models.ts b/frontend/src/js/models.ts index 19c6bb9..03bd8ee 100644 --- a/frontend/src/js/models.ts +++ b/frontend/src/js/models.ts @@ -15,7 +15,8 @@ export type LoginResponse = { export type ServerStatusResponse = { status:string, - loggined:boolean + loggined:boolean, + version?:string } export type PasswordSend = { diff --git a/frontend/src/js/store.ts b/frontend/src/js/store.ts index cea0bbb..fb0f006 100644 --- a/frontend/src/js/store.ts +++ b/frontend/src/js/store.ts @@ -73,3 +73,13 @@ export const useSession = () => { getHomeSection, }; }; + +interface SystemState { + version: string | null; + setVersion: (version: string | null) => void; +} + +export const useSystemStore = create()((set) => ({ + version: null, + setVersion: (version) => set({ version }), +})); diff --git a/frontend/src/js/utils.tsx b/frontend/src/js/utils.tsx index b02e6ce..57e6606 100644 --- a/frontend/src/js/utils.tsx +++ b/frontend/src/js/utils.tsx @@ -32,6 +32,7 @@ export const socketio = import.meta.env.DEV? io("ws://"+DEV_IP_BACKEND, { path:"/sock/socket.io", transports: ['websocket'], + autoConnect: false, auth: { token: useAuthStore.getState().getAccessToken() } @@ -39,6 +40,7 @@ export const socketio = import.meta.env.DEV? io({ path:"/sock/socket.io", transports: ['websocket'], + autoConnect: false, auth: { token: useAuthStore.getState().getAccessToken() } @@ -91,7 +93,10 @@ export async function genericapi(method:string,path:string,data:any = undefined, }, body: data? (is_form ? (new URLSearchParams(data)).toString() : JSON.stringify(data)) : undefined }).then(res => { - if(res.status === 401) window.location.reload() + if(res.status === 401) { + useAuthStore.getState().clearAccessToken(); + window.location.reload(); + } if(res.status === 406) resolve({status:"Wrong Password"}) if(!res.ok){ const errorDefault = res.statusText diff --git a/tests/nfproxy_test.py b/tests/nfproxy_test.py index 05efd6c..a4777ae 100644 --- a/tests/nfproxy_test.py +++ b/tests/nfproxy_test.py @@ -783,6 +783,52 @@ if __name__ == "__main__": remove_filters() + HTTP_HISTORY_TEST_FILTER = textwrap.dedent(""" + from firegex.nfproxy.models import HttpFullRequest, HttpHistory + from firegex.nfproxy import pyfilter, ACCEPT, REJECT + + @pyfilter + def history_filter_test(req: HttpFullRequest, history: HttpHistory): + past_urls = [r.url for r in history.requests] + if "/history_flag" in past_urls and req.url == "/history_secret": + return REJECT + return ACCEPT + """) + + if firegex.nfproxy_set_code(service_id, HTTP_HISTORY_TEST_FILTER): + puts( + "Successfully added filter for HttpHistory test ✔", + color=colors.green, + ) + else: + puts("Test Failed: Couldn't add HttpHistory filter ✗", color=colors.red) + exit_test(1) + + server.connect_client() + req1 = b"GET /history_flag HTTP/1.1\r\nHost: localhost\r\nContent-Length: 0\r\n\r\n" + resp1 = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK" + server.send_packet(req1, server_reply=resp1) + if not server.recv_packet(): + puts("Test Failed: /history_flag request was blocked unexpectedly ✗", color=colors.red) + exit_test(1) + + req2 = b"GET /history_secret HTTP/1.1\r\nHost: localhost\r\nContent-Length: 0\r\n\r\n" + server.send_packet(req2) + if not server.recv_packet(): + puts( + "The HTTP request to /history_secret was successfully blocked based on HttpHistory ✔", + color=colors.green, + ) + else: + puts( + "Test Failed: The HTTP request to /history_secret wasn't blocked by HttpHistory filter ✗", + color=colors.red, + ) + exit_test(1) + server.close_client() + + remove_filters() + # Simulating requests is more complex due to websocket extensions handshake WS_REQUEST_PARSING_TEST = b"GET /sock/?EIO=4&transport=websocket HTTP/1.1\r\nHost: localhost:8080\r\nConnection: Upgrade\r\nPragma: no-cache\r\nCache-Control: no-cache\r\nUser-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)\xac AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36\r\nUpgrade: websocket\r\nOrigin: http://localhost:8080\r\nSec-WebSocket-Version: 13\r\nAccept-Encoding: gzip, deflate, br, zstd\r\nAccept-Language: it-IT,it;q=0.9,en-US;q=0.8,en;q=0.7,zh-CN;q=0.6,zh;q=0.5\r\nCookie: cookie-consent=true; _iub_cs-86405163=%7B%22timestamp%22%3A%222024-09-12T18%3A20%3A18.627Z%22%2C%22version%22%3A%221.65.1%22%2C%22purposes%22%3A%7B%221%22%3Atrue%2C%224%22%3Atrue%7D%2C%22id%22%3A86405163%2C%22cons%22%3A%7B%22rand%22%3A%222b09e6%22%7D%7D\r\nSec-WebSocket-Key: eE01O3/ZShPKsrykACLAaA==\r\nSec-WebSocket-Extensions: permessage-deflate; client_max_window_bits\r\n\r\n\xc1\x84#\x8a\xb2\xbb\x11\xbb\xb2\xbb" diff --git a/tests/stress_test_history.py b/tests/stress_test_history.py new file mode 100755 index 0000000..711a848 --- /dev/null +++ b/tests/stress_test_history.py @@ -0,0 +1,344 @@ +#!/usr/bin/env python3 +""" +Firegex HttpHistory Real-Time Stress Test. +Connects to Firegex API, deploys an nfproxy HTTP service with HttpHistory filter enabled, +starts an HTTP 200 OK backend server and continuous HTTP client requests. +Measures real-time memory usage (Firegex cpproxy / local process) and request throughput until stopped (Ctrl+C). +""" + +import sys +import os +import time +import signal +import socket +import threading +import argparse +import traceback +from typing import Optional, List + +# Add project paths +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../fgex-lib"))) + +try: + import psutil + HAS_PSUTIL = True +except ImportError: + HAS_PSUTIL = False + +from tests.utils.colors import colors, puts, sep +from tests.utils.firegexapi import FiregexAPI + +HTTP_STRESS_FILTER_CODE = """ +from firegex.nfproxy.models import HttpFullRequest, HttpFullResponse, HttpHistory +from firegex.nfproxy import pyfilter, ACCEPT + +req_count = 0 +resp_count = 0 + +@pyfilter +def history_stress_req(req: HttpFullRequest, history: HttpHistory): + global req_count + req_count += 1 + _ = len(history.requests) + _ = len(history.responses) + return ACCEPT + +@pyfilter +def history_stress_resp(resp: HttpFullResponse, history: HttpHistory): + global resp_count + resp_count += 1 + _ = len(history.requests) + _ = len(history.responses) + return ACCEPT +""" + + +def start_http_backend_server(port: int, ipv6: bool = False): + """Starts a multithreaded HTTP backend server that returns valid HTTP 200 OK responses.""" + server_sock = socket.socket(socket.AF_INET6 if ipv6 else socket.AF_INET, socket.SOCK_STREAM) + server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server_sock.bind(("::1" if ipv6 else "127.0.0.1", port)) + server_sock.listen(128) + server_sock.settimeout(0.5) + + running = True + + def handle_client(conn): + conn.settimeout(2.0) + resp = b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 12\r\nConnection: keep-alive\r\n\r\nHello World!" + try: + while running: + buf = conn.recv(4096) + if not buf: + break + conn.sendall(resp) + except Exception: + pass + finally: + try: + conn.close() + except Exception: + pass + + def server_loop(): + while running: + try: + conn, _ = server_sock.accept() + t = threading.Thread(target=handle_client, args=(conn,), daemon=True) + t.start() + except socket.timeout: + continue + except Exception: + break + try: + server_sock.close() + except Exception: + pass + + t = threading.Thread(target=server_loop, daemon=True) + t.start() + + def stop_server(): + nonlocal running + running = False + try: + server_sock.close() + except Exception: + pass + + return stop_server + + +def find_cpproxy_process(): + """Finds cpproxy process using psutil if running.""" + if not HAS_PSUTIL: + return None + try: + for proc in psutil.process_iter(['pid', 'name', 'cmdline']): + name = proc.info['name'] or '' + cmdline = ' '.join(proc.info['cmdline'] or []) + if 'cpproxy' in name or 'cpproxy' in cmdline: + return proc + except Exception: + pass + return None + + +def get_memory_mb(proc=None) -> float: + """Returns memory usage in MB for a given process or current process.""" + try: + if proc: + return proc.memory_info().rss / (1024 * 1024) + if HAS_PSUTIL: + return psutil.Process(os.getpid()).memory_info().rss / (1024 * 1024) + except Exception: + pass + return 0.0 + + +def create_client_socket(port: int, ipv6: bool = False) -> socket.socket: + s = socket.socket(socket.AF_INET6 if ipv6 else socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(2.0) + s.connect(("::1" if ipv6 else "127.0.0.1", port)) + return s + + +def main(): + parser = argparse.ArgumentParser(description="Firegex HttpHistory Stress Test") + parser.add_argument("--address", "-a", type=str, default="http://127.0.0.1:4444/", help="Firegex API address (default: http://127.0.0.1:4444/)") + parser.add_argument("--password", "-p", type=str, required=True, help="Firegex admin password") + parser.add_argument("--service-name", "-n", type=str, default="StressTestHistory", help="Service name (default: StressTestHistory)") + parser.add_argument("--port", "-P", type=int, default=1337, help="Service port (default: 1337)") + parser.add_argument("--ipv6", "-6", action="store_true", default=False, help="Use IPv6") + parser.add_argument("--connections", "-c", type=int, default=4, help="Number of concurrent client connections (default: 4)") + args = parser.parse_args() + + sep() + puts("🔥 Firegex HttpHistory Real-Time Stress Test 🔥", color=colors.cyan, is_bold=True) + puts(f"Firegex API: {args.address}", color=colors.yellow) + puts(f"Target Port: {args.port}", color=colors.yellow) + puts(f"Connections: {args.connections}", color=colors.yellow) + puts("Press Ctrl+C at any time to stop the stress test.", color=colors.magenta, is_bold=True) + sep() + + firegex = FiregexAPI(args.address) + + # 1. Login to Firegex + puts("Logging in to Firegex API...", color=colors.cyan) + if firegex.login(args.password): + puts("Successfully logged in ✔", color=colors.green) + else: + puts("Error: Login failed! Check password or address ✗", color=colors.red) + sys.exit(1) + + # 2. Cleanup existing service if any + for srv in firegex.nfproxy_get_services(): + if srv.get("name") == args.service_name or srv.get("port") == args.port: + puts(f"Cleaning up existing service ID {srv['service_id']}...", color=colors.yellow) + firegex.nfproxy_delete_service(srv["service_id"]) + + # 3. Create service in Firegex + ip_int = "::1" if args.ipv6 else "127.0.0.1" + service_id = firegex.nfproxy_add_service(args.service_name, args.port, "http", ip_int, fail_open=True) + if not service_id: + puts("Error: Failed to create nfproxy service in Firegex ✗", color=colors.red) + sys.exit(1) + puts(f"Created nfproxy service ID: {service_id} ✔", color=colors.green) + + def cleanup_and_exit(code=0): + puts("\nCleaning up Firegex service...", color=colors.yellow) + try: + firegex.nfproxy_stop_service(service_id) + firegex.nfproxy_delete_service(service_id) + puts("Successfully cleaned up Firegex service ✔", color=colors.green) + except Exception as e: + puts(f"Error during cleanup: {e}", color=colors.red) + sys.exit(code) + + # 4. Upload HttpHistory filter code + puts("Deploying HttpHistory pyfilter code to Firegex...", color=colors.cyan) + if not firegex.nfproxy_set_code(service_id, HTTP_STRESS_FILTER_CODE): + puts("Error: Failed to upload pyfilter code ✗", color=colors.red) + cleanup_and_exit(1) + puts("Successfully deployed HttpHistory filter ✔", color=colors.green) + + # 5. Start Firegex Service + puts("Starting Firegex service...", color=colors.cyan) + if not firegex.nfproxy_start_service(service_id): + puts("Error: Failed to start Firegex service ✗", color=colors.red) + cleanup_and_exit(1) + puts("Firegex service started successfully ✔", color=colors.green) + + # 6. Start HTTP Backend Server + stop_backend_server = start_http_backend_server(args.port, ipv6=args.ipv6) + puts(f"HTTP Backend server listening on port {args.port} ✔", color=colors.green) + time.sleep(0.5) + + # Find cpproxy process if available + cpproxy_proc = find_cpproxy_process() + if cpproxy_proc: + puts(f"Detected Firegex cpproxy process (PID {cpproxy_proc.pid}) ✔", color=colors.green) + + # 7. Connect client sockets + running = True + + def sig_handler(sig, frame): + nonlocal running + running = False + + signal.signal(signal.SIGINT, sig_handler) + signal.signal(signal.SIGTERM, sig_handler) + + client_socks: List[Optional[socket.socket]] = [] + for _ in range(args.connections): + try: + client_socks.append(create_client_socket(args.port, args.ipv6)) + except Exception: + client_socks.append(None) + + puts(f"Initialized {len(client_socks)} client socket slot(s) ✔", color=colors.green) + puts("\n🚀 Spamming HTTP requests continuously. Press Ctrl+C to stop...\n", color=colors.cyan, is_bold=True) + + req_index = 0 + total_requests = 0 + start_time = time.time() + last_print_time = start_time + last_req_count = 0 + peak_script_mem = get_memory_mb() + peak_cpproxy_mem = get_memory_mb(cpproxy_proc) if cpproxy_proc else 0.0 + + try: + while running: + for i in range(len(client_socks)): + if not running: + break + + s = client_socks[i] + if s is None: + try: + s = create_client_socket(args.port, args.ipv6) + client_socks[i] = s + except Exception: + continue + + req_index += 1 + req_bytes = f"GET /stress_req_{req_index} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: keep-alive\r\nUser-Agent: FiregexStressClient\r\n\r\n".encode() + + try: + s.sendall(req_bytes) + resp = s.recv(4096) + if resp and b"200 OK" in resp: + total_requests += 1 + else: + raise ConnectionResetError("Invalid or empty HTTP response") + except Exception: + # Safely close broken socket and mark slot for reconnection + try: + s.close() + except Exception: + pass + client_socks[i] = None + + curr_time = time.time() + elapsed_print = curr_time - last_print_time + + if elapsed_print >= 1.0 or not running: + script_mem = get_memory_mb() + cpproxy_mem = get_memory_mb(cpproxy_proc) if cpproxy_proc else 0.0 + + if script_mem > peak_script_mem: + peak_script_mem = script_mem + if cpproxy_mem > peak_cpproxy_mem: + peak_cpproxy_mem = cpproxy_mem + + batch_reqs = total_requests - last_req_count + reqs_per_sec = batch_reqs / elapsed_print if elapsed_print > 0 else 0 + total_elapsed = curr_time - start_time + + mem_str = f"Script RSS: {script_mem:.1f}MB" + if cpproxy_proc: + mem_str += f" | cpproxy RSS: {cpproxy_mem:.1f}MB (Peak: {peak_cpproxy_mem:.1f}MB)" + + sys.stdout.write( + f"\r\033[K[{time.strftime('%H:%M:%S')}] Elapsed: {total_elapsed:.1f}s | " + f"Reqs: {total_requests:,} | " + f"Speed: {reqs_per_sec:,.0f} req/s | " + f"{mem_str}" + ) + sys.stdout.flush() + + last_print_time = curr_time + last_req_count = total_requests + + except KeyboardInterrupt: + pass + + end_time = time.time() + duration = end_time - start_time + + # Close client sockets + for s in client_socks: + if s: + try: + s.close() + except Exception: + pass + + stop_backend_server() + + print() # New line + sep() + puts("🛑 Stress Test Interrupted by User", color=colors.yellow, is_bold=True) + puts(f"Total Time Elapsed: {duration:.2f} seconds") + puts(f"Total HTTP Reqs: {total_requests:,}") + puts(f"Average Throughput: {total_requests / max(duration, 0.001):,.0f} req/sec") + if cpproxy_proc: + puts(f"cpproxy Peak Memory: {peak_cpproxy_mem:.2f} MB", color=colors.cyan, is_bold=True) + sep() + + cleanup_and_exit(0) + + +if __name__ == "__main__": + main()