0xb00b5/firegex-traffic-viewer
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature: HttpFullRequest and HttpFullResponse implementation

Browse Source
This commit is contained in:
Domingo Dirutigliano
2025-06-22 17:42:21 +02:00
parent 4a9fd48826
commit 6853960b6d
5 changed files with 1233 additions and 494 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
fgex-lib/README.md
View File

@@ -50,7 +50,7 @@ def my_filter(raw_packet: RawPacket): #Logging filter
## Data handlers ## Data handlers
### RawPacket ### RawPacket
```python ```python
from firegex.nfproxy import RawPacket from firegex.nfproxy import RawPacket
``` ```
@@ -111,6 +111,12 @@ from firegex.nfproxy import HttpRequestHeader
``` ```
This handler will be called only when the request headers are complete. It will receive a HttpRequestHeader object with the same properties as HttpRequest. This handler will be called only when the request headers are complete. It will receive a HttpRequestHeader object with the same properties as HttpRequest.
### HttpFullRequest
```python
from firegex.nfproxy import HttpFullRequest
```
This handler will be called only when the request is complete. It will receive a HttpFullRequest object with the same properties as HttpRequest.
### HttpResponse ### HttpResponse
```python ```python
from firegex.nfproxy import HttpResponse from firegex.nfproxy import HttpResponse
@@ -138,3 +144,8 @@ from firegex.nfproxy import HttpResponseHeader
``` ```
This handler will be called only when the response headers are complete. It will receive a HttpResponseHeader object with the same properties as HttpResponse. This handler will be called only when the response headers are complete. It will receive a HttpResponseHeader object with the same properties as HttpResponse.
### HttpFullResponse
```python
from firegex.nfproxy import HttpFullResponse
```
This handler will be called only when the response is complete. It will receive a HttpFullResponse object with the same properties as HttpResponse.

37
fgex-lib/firegex/nfproxy/models/__init__.py
View File

@@ -1,5 +1,17 @@
from firegex.nfproxy.models.tcp import TCPInputStream, TCPOutputStream, TCPClientStream, TCPServerStream from firegex.nfproxy.models.tcp import (
from firegex.nfproxy.models.http import HttpRequest, HttpResponse, HttpRequestHeader, HttpResponseHeader TCPInputStream,
TCPOutputStream,
TCPClientStream,
TCPServerStream,
)
from firegex.nfproxy.models.http import (
HttpRequest,
HttpResponse,
HttpRequestHeader,
HttpResponseHeader,
HttpFullRequest,
HttpFullResponse,
)
from firegex.nfproxy.internals.data import RawPacket from firegex.nfproxy.internals.data import RawPacket
from enum import Enum from enum import Enum
@@ -17,15 +29,28 @@ type_annotations_associations = {
HttpResponse: HttpResponse._fetch_packet, HttpResponse: HttpResponse._fetch_packet,
HttpRequestHeader: HttpRequestHeader._fetch_packet, HttpRequestHeader: HttpRequestHeader._fetch_packet,
HttpResponseHeader: HttpResponseHeader._fetch_packet, HttpResponseHeader: HttpResponseHeader._fetch_packet,
} HttpFullRequest: HttpFullRequest._fetch_packet,
HttpFullResponse: HttpFullResponse._fetch_packet,
},
} }
class Protocols(Enum): class Protocols(Enum):
TCP = "tcp" TCP = "tcp"
HTTP = "http" HTTP = "http"
__all__ = [ __all__ = [
"RawPacket", "RawPacket",
"TCPInputStream", "TCPOutputStream", "TCPClientStream", "TCPServerStream", "TCPInputStream",
"HttpRequest", "HttpResponse", "HttpRequestHeader", "HttpResponseHeader", "Protocols" "TCPOutputStream",
] "TCPClientStream",
"TCPServerStream",
"HttpRequest",
"HttpResponse",
"HttpRequestHeader",
"HttpResponseHeader",
"HttpFullRequest",
"HttpFullResponse",
"Protocols",
]

44
fgex-lib/firegex/nfproxy/models/http.py
View File

@@ -73,6 +73,7 @@ class InternalCallbackHandler:
messages: deque[InternalHTTPMessage] = deque() messages: deque[InternalHTTPMessage] = deque()
_ws_extentions = None _ws_extentions = None
_ws_raised_error = False _ws_raised_error = False
release_message_headers = True
def reset_data(self): def reset_data(self):
self.msg = InternalHTTPMessage() self.msg = InternalHTTPMessage()
@@ -604,6 +605,7 @@ class InternalBasicHttpMetaClass:
if ( if (
not internal_data.call_mem["headers_were_set"] not internal_data.call_mem["headers_were_set"]
and parser.msg.headers_complete and parser.msg.headers_complete
and parser.release_message_headers
): ):
messages_tosend.append( messages_tosend.append(
parser.msg parser.msg
@@ -641,7 +643,7 @@ class HttpRequest(InternalBasicHttpMetaClass):
@staticmethod @staticmethod
def _parser_class() -> str: def _parser_class() -> str:
return "full_http" return "http_module"
def __repr__(self): def __repr__(self):
return f"<HttpRequest method={self.method} url={self.url} headers={self.headers} body=[{0 if not self.body else len(self.body)} bytes] http_version={self.http_version} keep_alive={self.keep_alive} should_upgrade={self.should_upgrade} headers_complete={self.headers_complete} message_complete={self.message_complete} content_length={self.content_length} stream={self.stream} ws_stream={self.ws_stream}>" return f"<HttpRequest method={self.method} url={self.url} headers={self.headers} body=[{0 if not self.body else len(self.body)} bytes] http_version={self.http_version} keep_alive={self.keep_alive} should_upgrade={self.should_upgrade} headers_complete={self.headers_complete} message_complete={self.message_complete} content_length={self.content_length} stream={self.stream} ws_stream={self.ws_stream}>"
@@ -664,12 +666,46 @@ class HttpResponse(InternalBasicHttpMetaClass):
@staticmethod @staticmethod
def _parser_class() -> str: def _parser_class() -> str:
return "full_http" return "http_module"
def __repr__(self): def __repr__(self):
return f"<HttpResponse status_code={self.status_code} url={self.url} headers={self.headers} body=[{0 if not self.body else len(self.body)} bytes] http_version={self.http_version} keep_alive={self.keep_alive} should_upgrade={self.should_upgrade} headers_complete={self.headers_complete} message_complete={self.message_complete} content_length={self.content_length} stream={self.stream} ws_stream={self.ws_stream}>" return f"<HttpResponse status_code={self.status_code} url={self.url} headers={self.headers} body=[{0 if not self.body else len(self.body)} bytes] http_version={self.http_version} keep_alive={self.keep_alive} should_upgrade={self.should_upgrade} headers_complete={self.headers_complete} message_complete={self.message_complete} content_length={self.content_length} stream={self.stream} ws_stream={self.ws_stream}>"
class HttpFullRequest(HttpRequest):
"""
HTTP Request handler
This data handler will be called when the request data is complete
"""
def _contructor_hook(self):
self._parser.release_message_headers = False
@staticmethod
def _parser_class() -> str:
return "http_full"
def __repr__(self):
return f"<HttpFullRequest method={self.method} url={self.url} headers={self.headers} body=[{0 if not self.body else len(self.body)} bytes] http_version={self.http_version} keep_alive={self.keep_alive} should_upgrade={self.should_upgrade} headers_complete={self.headers_complete} message_complete={self.message_complete} content_length={self.content_length} stream={self.stream} ws_stream={self.ws_stream}>"
class HttpFullResponse(HttpResponse):
"""
HTTP Response handler
This data handler will be called when the response data is complete
"""
def _contructor_hook(self):
self._parser.release_message_headers = False
@staticmethod
def _parser_class() -> str:
return "http_full"
def __repr__(self):
return f"<HttpFullResponse status_code={self.status_code} url={self.url} headers={self.headers} body=[{0 if not self.body else len(self.body)} bytes] http_version={self.http_version} keep_alive={self.keep_alive} should_upgrade={self.should_upgrade} headers_complete={self.headers_complete} message_complete={self.message_complete} content_length={self.content_length} stream={self.stream} ws_stream={self.ws_stream}>"
class HttpRequestHeader(HttpRequest): class HttpRequestHeader(HttpRequest):
""" """
HTTP Request Header handler HTTP Request Header handler
@@ -681,7 +717,7 @@ class HttpRequestHeader(HttpRequest):
@staticmethod @staticmethod
def _parser_class() -> str: def _parser_class() -> str:
return "header_http" return "http_header"
class HttpResponseHeader(HttpResponse): class HttpResponseHeader(HttpResponse):
@@ -695,4 +731,4 @@ class HttpResponseHeader(HttpResponse):
@staticmethod @staticmethod
def _parser_class() -> str: def _parser_class() -> str:
return "header_http" return "http_header"

1146
frontend/src/components/NFProxy/NFProxyDocs.tsx
View File

File diff suppressed because it is too large Load Diff

487
tests/nfproxy_test.py
View File

@@ -7,12 +7,35 @@ import secrets
import time import time
parser = argparse.ArgumentParser() parser = argparse.ArgumentParser()
parser.add_argument("--address", "-a", type=str , required=False, help='Address of firegex backend', default="http://127.0.0.1:4444/") parser.add_argument(
parser.add_argument("--password", "-p", type=str, required=True, help='Firegex password') "--address",
parser.add_argument("--service_name", "-n", type=str , required=False, help='Name of the test service', default="Test Service") "-a",
parser.add_argument("--port", "-P", type=int , required=False, help='Port of the test service', default=1337) type=str,
parser.add_argument("--ipv6", "-6" , action="store_true", help='Test Ipv6', default=False) required=False,
parser.add_argument("--verbose", "-V" , action="store_true", help='Verbose output', default=False) help="Address of firegex backend",
default="http://127.0.0.1:4444/",
)
parser.add_argument("--password", "-p", type=str, required=True, help="Firegex password")
parser.add_argument(
"--service_name",
"-n",
type=str,
required=False,
help="Name of the test service",
default="Test Service",
)
parser.add_argument(
"--port",
"-P",
type=int,
required=False,
help="Port of the test service",
default=1337,
)
parser.add_argument("--ipv6", "-6", action="store_true", help="Test Ipv6", default=False)
parser.add_argument(
"--verbose", "-V", action="store_true", help="Verbose output", default=False
)
args = parser.parse_args() args = parser.parse_args()
sep() sep()
@@ -21,28 +44,31 @@ puts(f"{args.address}", color=colors.yellow)
firegex = FiregexAPI(args.address) firegex = FiregexAPI(args.address)
#Login # Login
if (firegex.login(args.password)): if firegex.login(args.password):
puts("Sucessfully logged in ✔", color=colors.green) puts("Sucessfully logged in ✔", color=colors.green)
else: else:
puts("Test Failed: Unknown response or wrong passowrd ✗", color=colors.red) puts("Test Failed: Unknown response or wrong passowrd ✗", color=colors.red)
exit(1) exit(1)
#Create server # Create server
server = TcpServer(args.port,ipv6=args.ipv6, verbose=args.verbose) server = TcpServer(args.port, ipv6=args.ipv6, verbose=args.verbose)
srvs = firegex.nfproxy_get_services() srvs = firegex.nfproxy_get_services()
for ele in srvs: for ele in srvs:
if ele['name'] == args.service_name: if ele["name"] == args.service_name:
firegex.nfproxy_delete_service(ele['service_id']) firegex.nfproxy_delete_service(ele["service_id"])
service_id = firegex.nfproxy_add_service(args.service_name, args.port, "http" , "::1" if args.ipv6 else "127.0.0.1" ) service_id = firegex.nfproxy_add_service(
args.service_name, args.port, "http", "::1" if args.ipv6 else "127.0.0.1"
)
if service_id: if service_id:
puts(f"Sucessfully created service {service_id} ✔", color=colors.green) puts(f"Sucessfully created service {service_id} ✔", color=colors.green)
else: else:
puts("Test Failed: Failed to create service ✗", color=colors.red) puts("Test Failed: Failed to create service ✗", color=colors.red)
exit(1) exit(1)
def exit_test(code): def exit_test(code):
if service_id: if service_id:
server.stop() server.stop()
@@ -50,11 +76,12 @@ def exit_test(code):
if firegex.nfproxy_delete_service(service_id): if firegex.nfproxy_delete_service(service_id):
puts("Sucessfully deleted service ✔", color=colors.green) puts("Sucessfully deleted service ✔", color=colors.green)
else: else:
puts("Test Failed: Coulnd't delete serivce ✗", color=colors.red) puts("Test Failed: Coulnd't delete serivce ✗", color=colors.red)
""" """
exit(code) exit(code)
if(firegex.nfproxy_start_service(service_id)):
if firegex.nfproxy_start_service(service_id):
puts("Sucessfully started service ✔", color=colors.green) puts("Sucessfully started service ✔", color=colors.green)
else: else:
puts("Test Failed: Failed to start service ✗", color=colors.red) puts("Test Failed: Failed to start service ✗", color=colors.red)
@@ -85,21 +112,27 @@ def verdict_test(packet:RawPacket):
BASE_FILTER_VERDICT_NAME = "verdict_test" BASE_FILTER_VERDICT_NAME = "verdict_test"
def get_vedict_test(to_match:str, action:str, mangle_to:str="REDACTED"):
return BASE_FILTER_VERDICT_TEST.replace("%%TEST%%", to_match).replace("%%ACTION%%", action).replace("%%MANGLE%%", mangle_to) def get_vedict_test(to_match: str, action: str, mangle_to: str = "REDACTED"):
return (
BASE_FILTER_VERDICT_TEST.replace("%%TEST%%", to_match)
.replace("%%ACTION%%", action)
.replace("%%MANGLE%%", mangle_to)
)
#Check if filter is present in the service # Check if filter is present in the service
n_blocked = 0 n_blocked = 0
n_mangled = 0 n_mangled = 0
def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None): def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
if mangle_with: if mangle_with:
if should_work: if should_work:
global n_mangled global n_mangled
for r in firegex.nfproxy_get_service_pyfilters(service_id): for r in firegex.nfproxy_get_service_pyfilters(service_id):
if r["name"] == filter_name: if r["name"] == filter_name:
#Test the filter # Test the filter
pre_packet = secrets.token_bytes(40) pre_packet = secrets.token_bytes(40)
post_packet = secrets.token_bytes(40) post_packet = secrets.token_bytes(40)
server.connect_client() server.connect_client()
@@ -107,82 +140,136 @@ def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
real_response = server.recv_packet() real_response = server.recv_packet()
expected_response = pre_packet + mangle_with + post_packet expected_response = pre_packet + mangle_with + post_packet
if real_response == expected_response: if real_response == expected_response:
puts("The malicious request was successfully mangled ✔", color=colors.green) puts(
"The malicious request was successfully mangled ✔",
color=colors.green,
)
n_mangled += 1 n_mangled += 1
time.sleep(1) time.sleep(1)
if firegex.nfproxy_get_pyfilter(service_id, filter_name)["edited_packets"] == n_mangled: if (
puts("The packet was reported as mangled in the API ✔", color=colors.green) firegex.nfproxy_get_pyfilter(service_id, filter_name)[
"edited_packets"
]
== n_mangled
):
puts(
"The packet was reported as mangled in the API ✔",
color=colors.green,
)
else: else:
puts("Test Failed: The packet wasn't reported as mangled in the API ✗", color=colors.red) puts(
"Test Failed: The packet wasn't reported as mangled in the API ✗",
color=colors.red,
)
exit_test(1) exit_test(1)
server.send_packet(pre_packet) server.send_packet(pre_packet)
if server.recv_packet() == pre_packet: if server.recv_packet() == pre_packet:
puts("Is able to communicate after mangle ✔", color=colors.green) puts(
"Is able to communicate after mangle ✔",
color=colors.green,
)
else: else:
puts("Test Failed: Couldn't communicate after mangle ✗", color=colors.red) puts(
"Test Failed: Couldn't communicate after mangle ✗",
color=colors.red,
)
exit_test(1) exit_test(1)
else: else:
puts("Test Failed: The request wasn't mangled ✗", color=colors.red) puts(
"Test Failed: The request wasn't mangled ✗", color=colors.red
)
exit_test(1) exit_test(1)
server.close_client() server.close_client()
return return
puts("Test Failed: The filter wasn't found ✗", color=colors.red) puts("Test Failed: The filter wasn't found ✗", color=colors.red)
else: else:
if server.sendCheckData(secrets.token_bytes(40) + match_bytes + secrets.token_bytes(40)): if server.sendCheckData(
secrets.token_bytes(40) + match_bytes + secrets.token_bytes(40)
):
puts("The request wasn't mangled ✔", color=colors.green) puts("The request wasn't mangled ✔", color=colors.green)
else: else:
puts("Test Failed: The request was mangled when it shouldn't have", color=colors.red) puts(
"Test Failed: The request was mangled when it shouldn't have",
color=colors.red,
)
exit_test(1) exit_test(1)
else: else:
if should_work: if should_work:
global n_blocked global n_blocked
for r in firegex.nfproxy_get_service_pyfilters(service_id): for r in firegex.nfproxy_get_service_pyfilters(service_id):
if r["name"] == filter_name: if r["name"] == filter_name:
#Test the filter # Test the filter
if not server.sendCheckData(secrets.token_bytes(40) + match_bytes + secrets.token_bytes(40)): if not server.sendCheckData(
puts("The malicious request was successfully blocked ✔", color=colors.green) secrets.token_bytes(40) + match_bytes + secrets.token_bytes(40)
):
puts(
"The malicious request was successfully blocked ✔",
color=colors.green,
)
n_blocked += 1 n_blocked += 1
time.sleep(1) time.sleep(1)
if firegex.nfproxy_get_pyfilter(service_id, filter_name)["blocked_packets"] == n_blocked: if (
puts("The packet was reported as blocked in the API ✔", color=colors.green) firegex.nfproxy_get_pyfilter(service_id, filter_name)[
"blocked_packets"
]
== n_blocked
):
puts(
"The packet was reported as blocked in the API ✔",
color=colors.green,
)
else: else:
puts("Test Failed: The packet wasn't reported as blocked in the API ✗", color=colors.red) puts(
"Test Failed: The packet wasn't reported as blocked in the API ✗",
color=colors.red,
)
exit_test(1) exit_test(1)
else: else:
puts("Test Failed: The request wasn't blocked ✗", color=colors.red) puts(
"Test Failed: The request wasn't blocked ✗", color=colors.red
)
exit_test(1) exit_test(1)
return return
puts("Test Failed: The filter wasn't found ✗", color=colors.red) puts("Test Failed: The filter wasn't found ✗", color=colors.red)
exit_test(1) exit_test(1)
else: else:
if server.sendCheckData(secrets.token_bytes(40) + match_bytes + secrets.token_bytes(40)): if server.sendCheckData(
secrets.token_bytes(40) + match_bytes + secrets.token_bytes(40)
):
puts("The request wasn't blocked ✔", color=colors.green) puts("The request wasn't blocked ✔", color=colors.green)
else: else:
puts("Test Failed: The request was blocked when it shouldn't have", color=colors.red) puts(
"Test Failed: The request was blocked when it shouldn't have",
color=colors.red,
)
exit_test(1) exit_test(1)
#Add new filter
# Add new filter
secret = bytes(secrets.token_hex(16).encode()) secret = bytes(secrets.token_hex(16).encode())
if firegex.nfproxy_set_code(service_id,get_vedict_test(secret.decode(), "REJECT")): if firegex.nfproxy_set_code(service_id, get_vedict_test(secret.decode(), "REJECT")):
puts(f"Sucessfully added filter for {str(secret)} in REJECT mode ✔", color=colors.green) puts(
f"Sucessfully added filter for {str(secret)} in REJECT mode ✔",
color=colors.green,
)
else: else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red) puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1) exit_test(1)
checkFilter(secret, BASE_FILTER_VERDICT_NAME) checkFilter(secret, BASE_FILTER_VERDICT_NAME)
#Pause the proxy # Pause the proxy
if firegex.nfproxy_stop_service(service_id): if firegex.nfproxy_stop_service(service_id):
puts(f"Sucessfully paused service with id {service_id} ✔", color=colors.green) puts(f"Sucessfully paused service with id {service_id} ✔", color=colors.green)
else: else:
puts("Test Failed: Coulnd't pause the service ✗", color=colors.red) puts("Test Failed: Coulnd't pause the service ✗", color=colors.red)
exit_test(1) exit_test(1)
#Check if it's actually paused # Check if it's actually paused
checkFilter(secret, BASE_FILTER_VERDICT_NAME, should_work=False) checkFilter(secret, BASE_FILTER_VERDICT_NAME, should_work=False)
#Start firewall # Start firewall
if firegex.nfproxy_start_service(service_id): if firegex.nfproxy_start_service(service_id):
puts(f"Sucessfully started service with id {service_id} ✔", color=colors.green) puts(f"Sucessfully started service with id {service_id} ✔", color=colors.green)
else: else:
@@ -191,25 +278,26 @@ else:
checkFilter(secret, BASE_FILTER_VERDICT_NAME) checkFilter(secret, BASE_FILTER_VERDICT_NAME)
#Disable filter # Disable filter
if firegex.nfproxy_disable_pyfilter(service_id, BASE_FILTER_VERDICT_NAME): if firegex.nfproxy_disable_pyfilter(service_id, BASE_FILTER_VERDICT_NAME):
puts(f"Sucessfully disabled filter {BASE_FILTER_VERDICT_NAME} ✔", color=colors.green) puts(f"Sucessfully disabled filter {BASE_FILTER_VERDICT_NAME} ✔", color=colors.green)
else: else:
puts("Test Failed: Coulnd't disable the filter ✗", color=colors.red) puts("Test Failed: Coulnd't disable the filter ✗", color=colors.red)
exit_test(1) exit_test(1)
#Check if it's actually disabled # Check if it's actually disabled
checkFilter(secret, BASE_FILTER_VERDICT_NAME, should_work=False) checkFilter(secret, BASE_FILTER_VERDICT_NAME, should_work=False)
#Enable filter # Enable filter
if firegex.nfproxy_enable_pyfilter(service_id, BASE_FILTER_VERDICT_NAME): if firegex.nfproxy_enable_pyfilter(service_id, BASE_FILTER_VERDICT_NAME):
puts(f"Sucessfully enabled filter {BASE_FILTER_VERDICT_NAME} ✔", color=colors.green) puts(f"Sucessfully enabled filter {BASE_FILTER_VERDICT_NAME} ✔", color=colors.green)
else: else:
puts("Test Failed: Coulnd't enable the regex ✗", color=colors.red) puts("Test Failed: Coulnd't enable the regex ✗", color=colors.red)
exit_test(1) exit_test(1)
checkFilter(secret, BASE_FILTER_VERDICT_NAME) checkFilter(secret, BASE_FILTER_VERDICT_NAME)
def remove_filters(): def remove_filters():
global n_blocked, n_mangled global n_blocked, n_mangled
server.stop() server.stop()
@@ -220,14 +308,17 @@ def remove_filters():
n_blocked = 0 n_blocked = 0
n_mangled = 0 n_mangled = 0
remove_filters() remove_filters()
#Check if it's actually deleted # Check if it's actually deleted
checkFilter(secret, BASE_FILTER_VERDICT_NAME, should_work=False) checkFilter(secret, BASE_FILTER_VERDICT_NAME, should_work=False)
#Check if DROP works # Check if DROP works
if firegex.nfproxy_set_code(service_id,get_vedict_test(secret.decode(), "DROP")): if firegex.nfproxy_set_code(service_id, get_vedict_test(secret.decode(), "DROP")):
puts(f"Sucessfully added filter for {str(secret)} in DROP mode ✔", color=colors.green) puts(
f"Sucessfully added filter for {str(secret)} in DROP mode ✔", color=colors.green
)
else: else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red) puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1) exit_test(1)
@@ -236,10 +327,16 @@ checkFilter(secret, BASE_FILTER_VERDICT_NAME)
remove_filters() remove_filters()
#Check if UNSTABLE_MANGLE works # Check if UNSTABLE_MANGLE works
mangle_result = secrets.token_hex(4).encode() # Mangle to a smaller packet mangle_result = secrets.token_hex(4).encode() # Mangle to a smaller packet
if firegex.nfproxy_set_code(service_id, get_vedict_test(secret.decode(), "UNSTABLE_MANGLE", mangle_result.decode())): if firegex.nfproxy_set_code(
puts(f"Sucessfully added filter for {str(secret)} in UNSTABLE_MANGLE mode to a smaller packet size ✔", color=colors.green) service_id,
get_vedict_test(secret.decode(), "UNSTABLE_MANGLE", mangle_result.decode()),
):
puts(
f"Sucessfully added filter for {str(secret)} in UNSTABLE_MANGLE mode to a smaller packet size ✔",
color=colors.green,
)
else: else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red) puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1) exit_test(1)
@@ -248,10 +345,16 @@ checkFilter(secret, BASE_FILTER_VERDICT_NAME, mangle_with=mangle_result)
remove_filters() remove_filters()
#Check if UNSTABLE_MANGLE works # Check if UNSTABLE_MANGLE works
mangle_result = secrets.token_hex(60).encode() # Mangle to a bigger packet mangle_result = secrets.token_hex(60).encode() # Mangle to a bigger packet
if firegex.nfproxy_set_code(service_id, get_vedict_test(secret.decode(), "UNSTABLE_MANGLE", mangle_result.decode())): if firegex.nfproxy_set_code(
puts(f"Sucessfully added filter for {str(secret)} in UNSTABLE_MANGLE mode to a bigger packet size ✔", color=colors.green) service_id,
get_vedict_test(secret.decode(), "UNSTABLE_MANGLE", mangle_result.decode()),
):
puts(
f"Sucessfully added filter for {str(secret)} in UNSTABLE_MANGLE mode to a bigger packet size ✔",
color=colors.green,
)
else: else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red) puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1) exit_test(1)
@@ -273,12 +376,15 @@ def data_type_test(packet:TCPInputStream):
""" """
if firegex.nfproxy_set_code(service_id, TCP_INPUT_STREAM_TEST): if firegex.nfproxy_set_code(service_id, TCP_INPUT_STREAM_TEST):
puts(f"Sucessfully added filter for {str(secret)} for TCPInputStream ✔", color=colors.green) puts(
f"Sucessfully added filter for {str(secret)} for TCPInputStream ✔",
color=colors.green,
)
else: else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red) puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1) exit_test(1)
data_split = len(secret)//2 data_split = len(secret) // 2
server.connect_client() server.connect_client()
server.send_packet(secret[:data_split]) server.send_packet(secret[:data_split])
if server.recv_packet() == secret[:data_split]: if server.recv_packet() == secret[:data_split]:
@@ -309,12 +415,15 @@ def data_type_test(packet:TCPOutputStream):
""" """
if firegex.nfproxy_set_code(service_id, TCP_OUTPUT_STREAM_TEST): if firegex.nfproxy_set_code(service_id, TCP_OUTPUT_STREAM_TEST):
puts(f"Sucessfully added filter for {str(secret)} for TCPOutputStream ✔", color=colors.green) puts(
f"Sucessfully added filter for {str(secret)} for TCPOutputStream ✔",
color=colors.green,
)
else: else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red) puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1) exit_test(1)
data_split = len(secret)//2 data_split = len(secret) // 2
server.connect_client() server.connect_client()
server.send_packet(secret[:data_split]) server.send_packet(secret[:data_split])
if server.recv_packet() == secret[:data_split]: if server.recv_packet() == secret[:data_split]:
@@ -363,7 +472,10 @@ def data_type_test(req:HttpRequest):
""" """
if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_STREAM_TEST): if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_STREAM_TEST):
puts(f"Sucessfully added filter for {str(secret)} for HttpRequest ✔", color=colors.green) puts(
f"Sucessfully added filter for {str(secret)} for HttpRequest ✔",
color=colors.green,
)
else: else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red) puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1) exit_test(1)
@@ -371,23 +483,94 @@ else:
server.connect_client() server.connect_client()
server.send_packet(REQUEST_HEADER_TEST.encode()) server.send_packet(REQUEST_HEADER_TEST.encode())
if not server.recv_packet(): if not server.recv_packet():
puts("The malicious HTTP request with the malicious header was successfully blocked ✔", color=colors.green) puts(
"The malicious HTTP request with the malicious header was successfully blocked ✔",
color=colors.green,
)
else: else:
puts("Test Failed: The HTTP request with the malicious header wasn't blocked ✗", color=colors.red) puts(
"Test Failed: The HTTP request with the malicious header wasn't blocked ✗",
color=colors.red,
)
exit_test(1) exit_test(1)
server.close_client() server.close_client()
server.connect_client() server.connect_client()
server.send_packet(REQUEST_BODY_TEST.encode()) server.send_packet(REQUEST_BODY_TEST.encode())
if not server.recv_packet(): if not server.recv_packet():
puts("The malicious HTTP request with the malicious body was successfully blocked ✔", color=colors.green) puts(
"The malicious HTTP request with the malicious body was successfully blocked ✔",
color=colors.green,
)
else: else:
puts("Test Failed: The HTTP request with the malicious body wasn't blocked ✗", color=colors.red) puts(
"Test Failed: The HTTP request with the malicious body wasn't blocked ✗",
color=colors.red,
)
exit_test(1) exit_test(1)
server.close_client() server.close_client()
remove_filters() remove_filters()
HTTP_FULL_REQUEST_STREAM_TEST = f"""
from firegex.nfproxy.models import HttpFullRequest
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@pyfilter
def data_type_test(req:HttpFullRequest):
if not req.body:
return ACCEPT
if not req.get_header("x-test"):
return ACCEPT
if {repr(secret.decode())} in req.get_header("x-test"):
return REJECT
if {repr(secret)} in req.body:
return REJECT
"""
if firegex.nfproxy_set_code(service_id, HTTP_FULL_REQUEST_STREAM_TEST):
puts(
f"Sucessfully added filter for {str(secret)} for HttpFullRequest ✔",
color=colors.green,
)
else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1)
server.connect_client()
server.send_packet(REQUEST_HEADER_TEST.encode())
if not server.recv_packet():
puts(
"The malicious HTTP request with the malicious header was successfully blocked ✔",
color=colors.green,
)
else:
puts(
"Test Failed: The HTTP request with the malicious header wasn't blocked ✗",
color=colors.red,
)
exit_test(1)
server.close_client()
server.connect_client()
server.send_packet(REQUEST_BODY_TEST.encode())
if not server.recv_packet():
puts(
"The malicious HTTP request with the malicious body was successfully blocked ✔",
color=colors.green,
)
else:
puts(
"Test Failed: The HTTP request with the malicious body wasn't blocked ✗",
color=colors.red,
)
exit_test(1)
server.close_client()
remove_filters()
HTTP_REQUEST_HEADER_STREAM_TEST = f""" HTTP_REQUEST_HEADER_STREAM_TEST = f"""
from firegex.nfproxy.models import HttpRequestHeader from firegex.nfproxy.models import HttpRequestHeader
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -400,7 +583,10 @@ def data_type_test(req:HttpRequestHeader):
""" """
if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_HEADER_STREAM_TEST): if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_HEADER_STREAM_TEST):
puts(f"Sucessfully added filter for {str(secret)} for HttpRequestHeader ✔", color=colors.green) puts(
f"Sucessfully added filter for {str(secret)} for HttpRequestHeader ✔",
color=colors.green,
)
else: else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red) puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1) exit_test(1)
@@ -408,9 +594,15 @@ else:
server.connect_client() server.connect_client()
server.send_packet(REQUEST_HEADER_TEST.encode()) server.send_packet(REQUEST_HEADER_TEST.encode())
if not server.recv_packet(): if not server.recv_packet():
puts("The malicious HTTP request with the malicious header was successfully blocked ✔", color=colors.green) puts(
"The malicious HTTP request with the malicious header was successfully blocked ✔",
color=colors.green,
)
else: else:
puts("Test Failed: The HTTP request with the malicious header wasn't blocked ✗", color=colors.red) puts(
"Test Failed: The HTTP request with the malicious header wasn't blocked ✗",
color=colors.red,
)
exit_test(1) exit_test(1)
server.close_client() server.close_client()
@@ -447,7 +639,10 @@ def data_type_test(req:HttpResponse):
""" """
if firegex.nfproxy_set_code(service_id, HTTP_RESPONSE_STREAM_TEST): if firegex.nfproxy_set_code(service_id, HTTP_RESPONSE_STREAM_TEST):
puts(f"Sucessfully added filter for {str(secret)} for HttpResponse ✔", color=colors.green) puts(
f"Sucessfully added filter for {str(secret)} for HttpResponse ✔",
color=colors.green,
)
else: else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red) puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1) exit_test(1)
@@ -455,23 +650,95 @@ else:
server.connect_client() server.connect_client()
server.send_packet(RESPONSE_HEADER_TEST.encode()) server.send_packet(RESPONSE_HEADER_TEST.encode())
if not server.recv_packet(): if not server.recv_packet():
puts("The malicious HTTP request with the malicious header was successfully blocked ✔", color=colors.green) puts(
"The malicious HTTP request with the malicious header was successfully blocked ✔",
color=colors.green,
)
else: else:
puts("Test Failed: The HTTP request with the malicious header wasn't blocked ✗", color=colors.red) puts(
"Test Failed: The HTTP request with the malicious header wasn't blocked ✗",
color=colors.red,
)
exit_test(1) exit_test(1)
server.close_client() server.close_client()
server.connect_client() server.connect_client()
server.send_packet(RESPONSE_BODY_TEST.encode()) server.send_packet(RESPONSE_BODY_TEST.encode())
if not server.recv_packet(): if not server.recv_packet():
puts("The malicious HTTP request with the malicious body was successfully blocked ✔", color=colors.green) puts(
"The malicious HTTP request with the malicious body was successfully blocked ✔",
color=colors.green,
)
else: else:
puts("Test Failed: The HTTP request with the malicious body wasn't blocked ✗", color=colors.red) puts(
"Test Failed: The HTTP request with the malicious body wasn't blocked ✗",
color=colors.red,
)
exit_test(1) exit_test(1)
server.close_client() server.close_client()
remove_filters() remove_filters()
HTTP_FULL_RESPONSE_STREAM_TEST = f"""
from firegex.nfproxy.models import HttpFullResponse
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@pyfilter
def data_type_test(req:HttpFullResponse):
if not req.body:
return ACCEPT
if not req.get_header("x-test"):
return ACCEPT
if {repr(secret.decode())} in req.get_header("x-test"):
return REJECT
if {repr(secret)} in req.body:
return REJECT
"""
if firegex.nfproxy_set_code(service_id, HTTP_FULL_RESPONSE_STREAM_TEST):
puts(
f"Sucessfully added filter for {str(secret)} for HttpFullResponse ✔",
color=colors.green,
)
else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1)
server.connect_client()
server.send_packet(RESPONSE_HEADER_TEST.encode())
if not server.recv_packet():
puts(
"The malicious HTTP request with the malicious header was successfully blocked ✔",
color=colors.green,
)
else:
puts(
"Test Failed: The HTTP request with the malicious header wasn't blocked ✗",
color=colors.red,
)
exit_test(1)
server.close_client()
server.connect_client()
server.send_packet(RESPONSE_BODY_TEST.encode())
if not server.recv_packet():
puts(
"The malicious HTTP request with the malicious body was successfully blocked ✔",
color=colors.green,
)
else:
puts(
"Test Failed: The HTTP request with the malicious body wasn't blocked ✗",
color=colors.red,
)
exit_test(1)
server.close_client()
remove_filters()
HTTP_RESPONSE_HEADER_STREAM_TEST = f""" HTTP_RESPONSE_HEADER_STREAM_TEST = f"""
from firegex.nfproxy.models import HttpResponseHeader from firegex.nfproxy.models import HttpResponseHeader
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -484,7 +751,10 @@ def data_type_test(req:HttpResponseHeader):
""" """
if firegex.nfproxy_set_code(service_id, HTTP_RESPONSE_HEADER_STREAM_TEST): if firegex.nfproxy_set_code(service_id, HTTP_RESPONSE_HEADER_STREAM_TEST):
puts(f"Sucessfully added filter for {str(secret)} for HttpResponseHeader ✔", color=colors.green) puts(
f"Sucessfully added filter for {str(secret)} for HttpResponseHeader ✔",
color=colors.green,
)
else: else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red) puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
exit_test(1) exit_test(1)
@@ -492,18 +762,24 @@ else:
server.connect_client() server.connect_client()
server.send_packet(RESPONSE_HEADER_TEST.encode()) server.send_packet(RESPONSE_HEADER_TEST.encode())
if not server.recv_packet(): if not server.recv_packet():
puts("The malicious HTTP request with the malicious header was successfully blocked ✔", color=colors.green) puts(
"The malicious HTTP request with the malicious header was successfully blocked ✔",
color=colors.green,
)
else: else:
puts("Test Failed: The HTTP request with the malicious header wasn't blocked ✗", color=colors.red) puts(
"Test Failed: The HTTP request with the malicious header wasn't blocked ✗",
color=colors.red,
)
exit_test(1) exit_test(1)
server.close_client() server.close_client()
remove_filters() remove_filters()
#Simulating requests is more complex due to websocket extensions handshake # 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' 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"
WS_RESPONSE_PARSING_TEST = b'HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: eGnJqUSoSKE3wOfKD2M3G82RsS8=\r\nSec-WebSocket-Extensions: permessage-deflate\r\ndate: Sat, 15 Mar 2025 12:04:19 GMT\r\nserver: uvicorn\r\n\r\n\xc1_2\xa8V*\xceLQ\xb2Rr1\xb4\xc8\xf6r\x0c\xf3\xaf\xd25\xf7\x8e\xf4\xb3LsttrW\xd2Q*-H/JLI-V\xb2\x8a\x8e\xd5Q*\xc8\xccK\x0f\xc9\xccM\xcd/-Q\xb222\x00\x02\x88\x98g^IjQYb\x0eP\xd0\x14,\x98\x9bX\x11\x90X\x99\x93\x9f\x084\xda\xd0\x00\x0cj\x01\x00\xc1\x1b21\x80\xd9e\xe1n\x19\x9e\xe3RP\x9a[Z\x99\x93j\xea\x15\x00\xb4\xcbC\xa9\x16\x00' WS_RESPONSE_PARSING_TEST = b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: eGnJqUSoSKE3wOfKD2M3G82RsS8=\r\nSec-WebSocket-Extensions: permessage-deflate\r\ndate: Sat, 15 Mar 2025 12:04:19 GMT\r\nserver: uvicorn\r\n\r\n\xc1_2\xa8V*\xceLQ\xb2Rr1\xb4\xc8\xf6r\x0c\xf3\xaf\xd25\xf7\x8e\xf4\xb3LsttrW\xd2Q*-H/JLI-V\xb2\x8a\x8e\xd5Q*\xc8\xccK\x0f\xc9\xccM\xcd/-Q\xb222\x00\x02\x88\x98g^IjQYb\x0eP\xd0\x14,\x98\x9bX\x11\x90X\x99\x93\x9f\x084\xda\xd0\x00\x0cj\x01\x00\xc1\x1b21\x80\xd9e\xe1n\x19\x9e\xe3RP\x9a[Z\x99\x93j\xea\x15\x00\xb4\xcbC\xa9\x16\x00"
HTTP_REQUEST_WS_PARSING_TEST = """ HTTP_REQUEST_WS_PARSING_TEST = """
from firegex.nfproxy.models import HttpRequest, HttpResponse from firegex.nfproxy.models import HttpRequest, HttpResponse
@@ -520,7 +796,10 @@ def data_type_test(req:HttpResponse):
""" """
if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_WS_PARSING_TEST): if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_WS_PARSING_TEST):
puts("Sucessfully added filter websocket parsing with HttpRequest and HttpResponse ✔", color=colors.green) puts(
"Sucessfully added filter websocket parsing with HttpRequest and HttpResponse ✔",
color=colors.green,
)
else: else:
puts("Test Failed: Couldn't add the websocket parsing filter ✗", color=colors.red) puts("Test Failed: Couldn't add the websocket parsing filter ✗", color=colors.red)
exit_test(1) exit_test(1)
@@ -528,43 +807,57 @@ else:
server.connect_client() server.connect_client()
server.send_packet(WS_REQUEST_PARSING_TEST, server_reply=WS_RESPONSE_PARSING_TEST) server.send_packet(WS_REQUEST_PARSING_TEST, server_reply=WS_RESPONSE_PARSING_TEST)
if server.recv_packet(): if server.recv_packet():
puts("The HTTP websocket upgrade request was successfully parsed ✔", color=colors.green) puts(
"The HTTP websocket upgrade request was successfully parsed ✔",
color=colors.green,
)
else: else:
puts("Test Failed: The HTTP websocket upgrade request wasn't parsed (an error occurred) ✗", color=colors.red) puts(
"Test Failed: The HTTP websocket upgrade request wasn't parsed (an error occurred) ✗",
color=colors.red,
)
exit_test(1) exit_test(1)
server.close_client() server.close_client()
remove_filters() remove_filters()
#Rename service # Rename service
if firegex.nfproxy_rename_service(service_id,f"{args.service_name}2"): if firegex.nfproxy_rename_service(service_id, f"{args.service_name}2"):
puts(f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green) puts(f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green)
else: else:
puts("Test Failed: Coulnd't rename service ✗", color=colors.red) puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
exit_test(1) exit_test(1)
#Check if service was renamed correctly # Check if service was renamed correctly
service = firegex.nfproxy_get_service(service_id) service = firegex.nfproxy_get_service(service_id)
if service["name"] == f"{args.service_name}2": if service["name"] == f"{args.service_name}2":
puts("Checked that service was renamed correctly ✔", color=colors.green) puts("Checked that service was renamed correctly ✔", color=colors.green)
else: else:
puts("Test Failed: Service wasn't renamed correctly ✗", color=colors.red) puts("Test Failed: Service wasn't renamed correctly ✗", color=colors.red)
exit_test(1) exit_test(1)
#Rename back service # Rename back service
if(firegex.nfproxy_rename_service(service_id,f"{args.service_name}")): if firegex.nfproxy_rename_service(service_id, f"{args.service_name}"):
puts(f"Sucessfully renamed service to {args.service_name} ✔", color=colors.green) puts(f"Sucessfully renamed service to {args.service_name} ✔", color=colors.green)
else: else:
puts("Test Failed: Coulnd't rename service ✗", color=colors.red) puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
exit_test(1) exit_test(1)
#Change settings # Change settings
if(firegex.nfproxy_settings_service(service_id, 1338, "::dead:beef" if args.ipv6 else "123.123.123.123", True)): if firegex.nfproxy_settings_service(
service_id, 1338, "::dead:beef" if args.ipv6 else "123.123.123.123", True
):
srv_updated = firegex.nfproxy_get_service(service_id) srv_updated = firegex.nfproxy_get_service(service_id)
if srv_updated["port"] == 1338 and ("::dead:beef" if args.ipv6 else "123.123.123.123") in srv_updated["ip_int"] and srv_updated["fail_open"]: if (
srv_updated["port"] == 1338
and ("::dead:beef" if args.ipv6 else "123.123.123.123") in srv_updated["ip_int"]
and srv_updated["fail_open"]
):
puts("Sucessfully changed service settings ✔", color=colors.green) puts("Sucessfully changed service settings ✔", color=colors.green)
else: else:
puts("Test Failed: Service settings weren't updated correctly ✗", color=colors.red) puts(
"Test Failed: Service settings weren't updated correctly ✗", color=colors.red
)
exit_test(1) exit_test(1)
else: else:
puts("Test Failed: Coulnd't change service settings ✗", color=colors.red) puts("Test Failed: Coulnd't change service settings ✗", color=colors.red)