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

using brotli 1.2 from pypi, fixed tests to py 3.14, removed from experimental pyproxy

Browse Source
This commit is contained in:
Domingo Dirutigliano
2025-11-11 23:30:48 +01:00
parent 49c6c14fe5
commit 16f96aa6f6
8 changed files with 1611 additions and 1311 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/requirements.txt
View File

@@ -4,5 +4,5 @@ uvicorn[standard]
psutil
python-jose[cryptography]
python-socketio
git+https://github.com/google/brotli.git
brotli
#git+https://salsa.debian.org/pkg-netfilter-team/pkg-nftables#egg=nftables&subdirectory=py

6
frontend/src/components/NavBar/index.tsx
View File

@@ -39,12 +39,12 @@ export default function NavBar() {
<NavBarButton navigate="nfregex" closeNav={closeNav} name="Netfilter Regex" color="grape" icon={<BsRegex size={19} />} />
<NavBarButton navigate="firewall" closeNav={closeNav} name="Firewall Rules" color="red" icon={<PiWallLight size={19} />} />
<NavBarButton navigate="porthijack" closeNav={closeNav} name="Hijack Port to Proxy" color="blue" icon={<GrDirections size={19} />} />
<Box px="xs" mt="lg">
<NavBarButton navigate="nfproxy" closeNav={closeNav} name="Netfilter Proxy" color="lime" icon={<TbPlugConnected size={19} />} />
{/* <Box px="xs" mt="lg">
<Title order={5}>Experimental Features 🧪</Title>
</Box>
<Text></Text>
<Divider my="xs" />
<NavBarButton navigate="nfproxy" closeNav={closeNav} name="Netfilter Proxy" color="lime" icon={<TbPlugConnected size={19} />} />
<Divider my="xs" /> */}
</Box>
</AppShell.Navbar>

76
tests/api_test.py
View File

@@ -4,9 +4,19 @@ from utils.firegexapi import FiregexAPI
import argparse
import secrets
if __name__ == "__main__":
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("--password", "-p", type=str, required=True, help='Firegex password')
parser.add_argument(
"--address",
"-a",
type=str,
required=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"
)
args = parser.parse_args()
sep()
puts("Testing will start on ", color=colors.cyan, end="")
@@ -16,19 +26,22 @@ firegex = FiregexAPI(args.address)
# Connect to Firegex
if firegex.status()["status"] == "init":
if (firegex.set_password(args.password)):
if firegex.set_password(args.password):
puts(f"Sucessfully set password to {args.password} ✔", color=colors.green)
else:
puts("Test Failed: Unknown response or password already put ✗", color=colors.red)
puts(
"Test Failed: Unknown response or password already put ✗",
color=colors.red,
)
exit(1)
else:
if (firegex.login(args.password)):
if firegex.login(args.password):
puts("Sucessfully logged in ✔", color=colors.green)
else:
puts("Test Failed: Unknown response or wrong passowrd ✗", color=colors.red)
exit(1)
if(firegex.status()["loggined"]):
if firegex.status()["loggined"]:
puts("Correctly received status ✔", color=colors.green)
else:
puts("Test Failed: Unknown response or not logged in✗", color=colors.red)
@@ -36,60 +49,81 @@ else:
# Prepare second instance
firegex2 = FiregexAPI(args.address)
if (firegex2.login(args.password)):
if firegex2.login(args.password):
puts("Sucessfully logged in on second instance ✔", color=colors.green)
else:
puts("Test Failed: Unknown response or wrong passowrd on second instance ✗", color=colors.red)
puts(
"Test Failed: Unknown response or wrong passowrd on second instance ✗",
color=colors.red,
)
exit(1)
if(firegex2.status()["loggined"]):
if firegex2.status()["loggined"]:
puts("Correctly received status on second instance✔", color=colors.green)
else:
puts("Test Failed: Unknown response or not logged in on second instance✗", color=colors.red)
puts(
"Test Failed: Unknown response or not logged in on second instance✗",
color=colors.red,
)
exit(1)
# Change password
new_password = secrets.token_hex(10)
if (firegex.change_password(new_password,expire=True)):
if firegex.change_password(new_password, expire=True):
puts(f"Sucessfully changed password to {new_password} ✔", color=colors.green)
else:
puts("Test Failed: Coundl't change the password ✗", color=colors.red)
exit(1)
# Check if we are still logged in
if(firegex.status()["loggined"]):
if firegex.status()["loggined"]:
puts("Correctly received status after password change ✔", color=colors.green)
else:
puts("Test Failed: Unknown response or not logged after password change ✗", color=colors.red)
puts(
"Test Failed: Unknown response or not logged after password change ✗",
color=colors.red,
)
exit(1)
# Check if second session expired and relog
if(not firegex2.status()["loggined"]):
if not firegex2.status()["loggined"]:
puts("Second instance was expired currectly ✔", color=colors.green)
else:
puts("Test Failed: Still logged in on second instance, expire expected ✗", color=colors.red)
puts(
"Test Failed: Still logged in on second instance, expire expected ✗",
color=colors.red,
)
exit(1)
if (firegex2.login(new_password)):
if firegex2.login(new_password):
puts("Sucessfully logged in on second instance ✔", color=colors.green)
else:
puts("Test Failed: Unknown response or wrong passowrd on second instance ✗", color=colors.red)
puts(
"Test Failed: Unknown response or wrong passowrd on second instance ✗",
color=colors.red,
)
exit(1)
# Change it back
if (firegex.change_password(args.password,expire=False)):
if firegex.change_password(args.password, expire=False):
puts("Sucessfully restored the password ✔", color=colors.green)
else:
puts("Test Failed: Coundl't change the password ✗", color=colors.red)
exit(1)
# Check if we are still logged in
if(firegex2.status()["loggined"]):
if firegex2.status()["loggined"]:
puts("Correctly received status after password change ✔", color=colors.green)
else:
puts("Test Failed: Unknown response or not logged after password change ✗", color=colors.red)
puts(
"Test Failed: Unknown response or not logged after password change ✗",
color=colors.red,
)
exit(1)
puts("List of available interfaces:", color=colors.yellow)
for interface in firegex.get_interfaces():
puts("name: {}, address: {}".format(interface["name"], interface["addr"]), color=colors.yellow)
puts(
"name: {}, address: {}".format(interface["name"], interface["addr"]),
color=colors.yellow,
)

125
tests/nfproxy_test.py
View File

@@ -5,7 +5,9 @@ from utils.tcpserver import TcpServer
import argparse
import secrets
import time
import textwrap
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument(
"--address",
@@ -15,7 +17,9 @@ parser.add_argument(
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(
"--password", "-p", type=str, required=True, help="Firegex password"
)
parser.add_argument(
"--service_name",
"-n",
@@ -32,7 +36,9 @@ parser.add_argument(
help="Port of the test service",
default=1337,
)
parser.add_argument("--ipv6", "-6", action="store_true", help="Test Ipv6", default=False)
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
)
@@ -68,19 +74,15 @@ else:
puts("Test Failed: Failed to create service ✗", color=colors.red)
exit(1)
def exit_test(code):
if service_id:
server.stop()
"""
if firegex.nfproxy_delete_service(service_id):
puts("Sucessfully deleted service ✔", color=colors.green)
else:
puts("Test Failed: Coulnd't delete serivce ✗", color=colors.red)
"""
exit(code)
if firegex.nfproxy_start_service(service_id):
puts("Sucessfully started service ✔", color=colors.green)
else:
@@ -91,7 +93,9 @@ server.start()
time.sleep(0.5)
try:
if server.sendCheckData(secrets.token_bytes(432)):
puts("Successfully tested first proxy with no filters ✔", color=colors.green)
puts(
"Successfully tested first proxy with no filters ✔", color=colors.green
)
else:
puts("Test Failed: Data was corrupted ", color=colors.red)
exit_test(1)
@@ -99,7 +103,7 @@ except Exception:
puts("Test Failed: Couldn't send data to the server ", color=colors.red)
exit_test(1)
BASE_FILTER_VERDICT_TEST = """
BASE_FILTER_VERDICT_TEST = textwrap.dedent("""
from firegex.nfproxy.models import RawPacket
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -108,11 +112,10 @@ def verdict_test(packet:RawPacket):
if b"%%TEST%%" in packet.data:
packet.l4_data = packet.l4_data.replace(b"%%TEST%%", b"%%MANGLE%%")
return %%ACTION%%
"""
""")
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)
@@ -120,12 +123,10 @@ def get_vedict_test(to_match: str, action: str, mangle_to: str = "REDACTED"):
.replace("%%MANGLE%%", mangle_to)
)
# Check if filter is present in the service
n_blocked = 0
n_mangled = 0
def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
if mangle_with:
if should_work:
@@ -176,7 +177,8 @@ def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
exit_test(1)
else:
puts(
"Test Failed: The request wasn't mangled ✗", color=colors.red
"Test Failed: The request wasn't mangled ✗",
color=colors.red,
)
exit_test(1)
server.close_client()
@@ -200,7 +202,9 @@ def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
if r["name"] == filter_name:
# Test the filter
if not server.sendCheckData(
secrets.token_bytes(40) + match_bytes + secrets.token_bytes(40)
secrets.token_bytes(40)
+ match_bytes
+ secrets.token_bytes(40)
):
puts(
"The malicious request was successfully blocked ✔",
@@ -226,7 +230,8 @@ def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
exit_test(1)
else:
puts(
"Test Failed: The request wasn't blocked ✗", color=colors.red
"Test Failed: The request wasn't blocked ✗",
color=colors.red,
)
exit_test(1)
return
@@ -244,7 +249,6 @@ def checkFilter(match_bytes, filter_name, should_work=True, mangle_with=None):
)
exit_test(1)
# Add new filter
secret = bytes(secrets.token_hex(16).encode())
@@ -280,7 +284,10 @@ checkFilter(secret, BASE_FILTER_VERDICT_NAME)
# Disable filter
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:
puts("Test Failed: Coulnd't disable the filter ✗", color=colors.red)
exit_test(1)
@@ -290,25 +297,27 @@ checkFilter(secret, BASE_FILTER_VERDICT_NAME, should_work=False)
# Enable filter
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:
puts("Test Failed: Coulnd't enable the regex ✗", color=colors.red)
exit_test(1)
checkFilter(secret, BASE_FILTER_VERDICT_NAME)
def remove_filters():
global n_blocked, n_mangled
server.stop()
server.start()
time.sleep(0.5) # Wait for server to fully start
if not firegex.nfproxy_set_code(service_id, ""):
puts("Test Failed: Couldn't remove the filter ✗", color=colors.red)
exit_test(1)
n_blocked = 0
n_mangled = 0
remove_filters()
# Check if it's actually deleted
@@ -317,7 +326,8 @@ checkFilter(secret, BASE_FILTER_VERDICT_NAME, should_work=False)
# Check if DROP works
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
f"Sucessfully added filter for {str(secret)} in DROP mode ✔",
color=colors.green,
)
else:
puts(f"Test Failed: Couldn't add the filter {str(secret)} ✗", color=colors.red)
@@ -364,7 +374,7 @@ checkFilter(secret, BASE_FILTER_VERDICT_NAME, mangle_with=mangle_result)
remove_filters()
secret = b"8331ee1bf75893dd7fa3d34f29bac7fc8935aa3ef6c565fe8b395ef7f485"
TCP_INPUT_STREAM_TEST = f"""
TCP_INPUT_STREAM_TEST = textwrap.dedent(f"""
from firegex.nfproxy.models import TCPInputStream
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -373,7 +383,7 @@ def data_type_test(packet:TCPInputStream):
if {repr(secret)} in packet.data:
return REJECT
"""
""")
if firegex.nfproxy_set_code(service_id, TCP_INPUT_STREAM_TEST):
puts(
@@ -403,7 +413,7 @@ server.close_client()
remove_filters()
secret = b"8331ee1bf75893dd7fa3d34f29bac7fc8935aa3ef6c565fe8b395ef7f485"
TCP_OUTPUT_STREAM_TEST = f"""
TCP_OUTPUT_STREAM_TEST = textwrap.dedent(f"""
from firegex.nfproxy.models import TCPOutputStream
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -412,7 +422,7 @@ def data_type_test(packet:TCPOutputStream):
if {repr(secret)} in packet.data:
return REJECT
"""
""")
if firegex.nfproxy_set_code(service_id, TCP_OUTPUT_STREAM_TEST):
puts(
@@ -443,21 +453,21 @@ remove_filters()
secret = b"8331ee1bf75893dd7fa3d34f29bac7fc8935aa3ef6c565fe8b395ef7f485"
REQUEST_HEADER_TEST = f"""POST / HTTP/1.1
REQUEST_HEADER_TEST = textwrap.dedent(f"""POST / HTTP/1.1
Host: localhost
X-TeSt: {secret.decode()}
Content-Length: 15
A Friendly Body""".replace("\n", "\r\n")
A Friendly Body""").replace("\n", "\r\n")
REQUEST_BODY_TEST = f"""POST / HTTP/1.1
REQUEST_BODY_TEST = textwrap.dedent(f"""POST / HTTP/1.1
Host: localhost
X-TeSt: NotTheSecret
Content-Length: {len(secret.decode())}
{secret.decode()}""".replace("\n", "\r\n")
{secret.decode()}""").replace("\n", "\r\n")
HTTP_REQUEST_STREAM_TEST = f"""
HTTP_REQUEST_STREAM_TEST = textwrap.dedent(f"""
from firegex.nfproxy.models import HttpRequest
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -469,7 +479,7 @@ def data_type_test(req:HttpRequest):
if {repr(secret)} in req.body:
return REJECT
"""
""")
if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_STREAM_TEST):
puts(
@@ -512,7 +522,7 @@ server.close_client()
remove_filters()
HTTP_FULL_REQUEST_STREAM_TEST = f"""
HTTP_FULL_REQUEST_STREAM_TEST = textwrap.dedent(f"""
from firegex.nfproxy.models import HttpFullRequest
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -527,7 +537,7 @@ def data_type_test(req:HttpFullRequest):
if {repr(secret)} in req.body:
return REJECT
"""
""")
if firegex.nfproxy_set_code(service_id, HTTP_FULL_REQUEST_STREAM_TEST):
puts(
@@ -570,8 +580,7 @@ server.close_client()
remove_filters()
HTTP_REQUEST_HEADER_STREAM_TEST = f"""
HTTP_REQUEST_HEADER_STREAM_TEST = textwrap.dedent(f"""
from firegex.nfproxy.models import HttpRequestHeader
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -580,7 +589,7 @@ def data_type_test(req:HttpRequestHeader):
if {repr(secret.decode())} in req.get_header("x-test"):
return REJECT
"""
""")
if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_HEADER_STREAM_TEST):
puts(
@@ -610,21 +619,21 @@ remove_filters()
secret = b"8331ee1bf75893dd7fa3d34f29bac7fc8935aa3ef6c565fe8b395ef7f485"
RESPONSE_HEADER_TEST = f"""HTTP/1.1 200 OK
RESPONSE_HEADER_TEST = textwrap.dedent(f"""HTTP/1.1 200 OK
Host: localhost
X-TeSt: {secret.decode()}
Content-Length: 15
A Friendly Body""".replace("\n", "\r\n")
A Friendly Body""").replace("\n", "\r\n")
RESPONSE_BODY_TEST = f"""HTTP/1.1 200 OK
RESPONSE_BODY_TEST = textwrap.dedent(f"""HTTP/1.1 200 OK
Host: localhost
X-TeSt: NotTheSecret
Content-Length: {len(secret.decode())}
{secret.decode()}""".replace("\n", "\r\n")
{secret.decode()}""").replace("\n", "\r\n")
HTTP_RESPONSE_STREAM_TEST = f"""
HTTP_RESPONSE_STREAM_TEST = textwrap.dedent(f"""
from firegex.nfproxy.models import HttpResponse
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -636,7 +645,7 @@ def data_type_test(req:HttpResponse):
if {repr(secret)} in req.body:
return REJECT
"""
""")
if firegex.nfproxy_set_code(service_id, HTTP_RESPONSE_STREAM_TEST):
puts(
@@ -679,8 +688,7 @@ server.close_client()
remove_filters()
HTTP_FULL_RESPONSE_STREAM_TEST = f"""
HTTP_FULL_RESPONSE_STREAM_TEST = textwrap.dedent(f"""
from firegex.nfproxy.models import HttpFullResponse
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -695,7 +703,7 @@ def data_type_test(req:HttpFullResponse):
if {repr(secret)} in req.body:
return REJECT
"""
""")
if firegex.nfproxy_set_code(service_id, HTTP_FULL_RESPONSE_STREAM_TEST):
puts(
@@ -738,8 +746,7 @@ server.close_client()
remove_filters()
HTTP_RESPONSE_HEADER_STREAM_TEST = f"""
HTTP_RESPONSE_HEADER_STREAM_TEST = textwrap.dedent(f"""
from firegex.nfproxy.models import HttpResponseHeader
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -748,7 +755,7 @@ def data_type_test(req:HttpResponseHeader):
if {repr(secret.decode())} in req.get_header("x-test"):
return REJECT
"""
""")
if firegex.nfproxy_set_code(service_id, HTTP_RESPONSE_HEADER_STREAM_TEST):
puts(
@@ -781,7 +788,7 @@ remove_filters()
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"
HTTP_REQUEST_WS_PARSING_TEST = """
HTTP_REQUEST_WS_PARSING_TEST = textwrap.dedent("""
from firegex.nfproxy.models import HttpRequest, HttpResponse
from firegex.nfproxy import pyfilter, ACCEPT, UNSTABLE_MANGLE, DROP, REJECT
@@ -793,7 +800,7 @@ def data_type_test(req:HttpRequest):
def data_type_test(req:HttpResponse):
print(req)
"""
""")
if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_WS_PARSING_TEST):
puts(
@@ -801,7 +808,9 @@ if firegex.nfproxy_set_code(service_id, HTTP_REQUEST_WS_PARSING_TEST):
color=colors.green,
)
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)
server.connect_client()
@@ -823,7 +832,9 @@ remove_filters()
# Rename service
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:
puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
exit_test(1)
@@ -838,7 +849,9 @@ else:
# Rename back service
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:
puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
exit_test(1)
@@ -850,13 +863,15 @@ if firegex.nfproxy_settings_service(
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 ("::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)
else:
puts(
"Test Failed: Service settings weren't updated correctly ✗", color=colors.red
"Test Failed: Service settings weren't updated correctly ✗",
color=colors.red,
)
exit_test(1)
else:

215
tests/nfregex_test.py
View File

@@ -8,13 +8,47 @@ import secrets
import base64
import time
if __name__ == "__main__":
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("--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("--proto", "-m" , type=str, required=False, choices=["tcp","udp"], help='Select the protocol', default="tcp")
parser.add_argument(
"--address",
"-a",
type=str,
required=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(
"--proto",
"-m",
type=str,
required=False,
choices=["tcp", "udp"],
help="Select the protocol",
default="tcp",
)
args = parser.parse_args()
sep()
@@ -24,19 +58,21 @@ puts(f"{args.address}", color=colors.yellow)
firegex = FiregexAPI(args.address)
# Login
if (firegex.login(args.password)):
if firegex.login(args.password):
puts("Sucessfully logged in ✔", color=colors.green)
else:
puts("Test Failed: Unknown response or wrong passowrd ✗", color=colors.red)
exit(1)
# Create server
server = (TcpServer if args.proto == "tcp" else UdpServer)(args.port,ipv6=args.ipv6)
server = (TcpServer if args.proto == "tcp" else UdpServer)(
args.port, ipv6=args.ipv6
)
def exit_test(code):
if service_id:
server.stop()
if(firegex.nfregex_delete_service(service_id)):
if firegex.nfregex_delete_service(service_id):
puts("Sucessfully deleted service ✔", color=colors.green)
else:
puts("Test Failed: Coulnd't delete serivce ✗", color=colors.red)
@@ -45,17 +81,19 @@ def exit_test(code):
srvs = firegex.nfregex_get_services()
for ele in srvs:
if ele['name'] == args.service_name:
firegex.nfregex_delete_service(ele['service_id'])
if ele["name"] == args.service_name:
firegex.nfregex_delete_service(ele["service_id"])
service_id = firegex.nfregex_add_service(args.service_name, args.port, args.proto , "::1" if args.ipv6 else "127.0.0.1" )
service_id = firegex.nfregex_add_service(
args.service_name, args.port, args.proto, "::1" if args.ipv6 else "127.0.0.1"
)
if service_id:
puts(f"Sucessfully created service {service_id} ✔", color=colors.green)
else:
puts("Test Failed: Failed to create service ✗", color=colors.red)
exit(1)
if(firegex.nfregex_start_service(service_id)):
if firegex.nfregex_start_service(service_id):
puts("Sucessfully started service ✔", color=colors.green)
else:
puts("Test Failed: Failed to start service ✗", color=colors.red)
@@ -75,13 +113,14 @@ except Exception:
# Add new regex
secret = bytes(secrets.token_hex(16).encode())
if firegex.nfregex_add_regex(service_id,secret,"B",active=True,is_case_sensitive=True):
if firegex.nfregex_add_regex(
service_id, secret, "B", active=True, is_case_sensitive=True
):
puts(f"Sucessfully added regex {str(secret)} ✔", color=colors.green)
else:
puts(f"Test Failed: Couldn't add the regex {str(secret)} ✗", color=colors.red)
exit_test(1)
# Check if regex is present in the service
n_blocked = 0
@@ -97,42 +136,84 @@ def checkRegex(regex, should_work=True, upper=False, deleted=False):
if r["regex"] == secret:
# Test the regex
s = regex.upper() if upper else regex
if not server.sendCheckData(secrets.token_bytes(40) + s + secrets.token_bytes(40)):
puts("The malicious request was successfully blocked ✔", color=colors.green)
if not server.sendCheckData(
secrets.token_bytes(40) + s + secrets.token_bytes(40)
):
puts(
"The malicious request was successfully blocked ✔",
color=colors.green,
)
n_blocked += 1
time.sleep(1)
if firegex.nfregex_get_regex(r["id"])["n_packets"] == n_blocked:
puts("The packet was reported as blocked in the API ✔", color=colors.green)
puts(
"The packet was reported as blocked in the API ✔",
color=colors.green,
)
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)
if getMetric("firegex_blocked_packets", secret.decode()) == n_blocked:
puts("The packet was reported as blocked in the metrics ✔", color=colors.green)
if (
getMetric("firegex_blocked_packets", secret.decode())
== n_blocked
):
puts(
"The packet was reported as blocked in the metrics ✔",
color=colors.green,
)
else:
puts("Test Failed: The packet wasn't reported as blocked in the metrics ✗", color=colors.red)
puts(
"Test Failed: The packet wasn't reported as blocked in the metrics ✗",
color=colors.red,
)
exit_test(1)
if getMetric("firegex_active", secret.decode()) == 1:
puts("The regex was reported as active in the metrics ✔", color=colors.green)
puts(
"The regex was reported as active in the metrics ✔",
color=colors.green,
)
else:
puts("Test Failed: The regex wasn't reported as active in the metrics ✗", color=colors.red)
puts(
"Test Failed: The regex wasn't reported as active in the metrics ✗",
color=colors.red,
)
exit_test(1)
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)
return
puts("Test Failed: The regex wasn't found ✗", color=colors.red)
exit_test(1)
else:
if server.sendCheckData(secrets.token_bytes(40) + base64.b64decode(regex) + secrets.token_bytes(40)):
if server.sendCheckData(
secrets.token_bytes(40)
+ base64.b64decode(regex)
+ secrets.token_bytes(40)
):
puts("The request wasn't blocked ✔", color=colors.green)
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)
if not deleted:
if getMetric("firegex_active", secret.decode()) == 0:
puts("The regex was reported as inactive in the metrics ✔", color=colors.green)
puts(
"The regex was reported as inactive in the metrics ✔",
color=colors.green,
)
else:
puts("Test Failed: The regex wasn't reported as inactive in the metrics ✗", color=colors.red)
puts(
"Test Failed: The regex wasn't reported as inactive in the metrics ✗",
color=colors.red,
)
exit_test(1)
def clear_regexes():
@@ -140,8 +221,11 @@ def clear_regexes():
n_blocked = 0
for r in firegex.nfregex_get_service_regexes(service_id):
if r["regex"] == secret:
if(firegex.nfregex_delete_regex(r["id"])):
puts(f"Sucessfully deleted regex with id {r['id']} ✔", color=colors.green)
if firegex.nfregex_delete_regex(r["id"]):
puts(
f"Sucessfully deleted regex with id {r['id']} ✔",
color=colors.green,
)
else:
puts("Test Failed: Coulnd't delete the regex ✗", color=colors.red)
exit_test(1)
@@ -149,13 +233,16 @@ def clear_regexes():
if f'regex="{secret.decode()}"' not in firegex.nfregex_get_metrics():
puts("No regex metrics after deletion ✔", color=colors.green)
else:
puts("Test Failed: Metrics found after deleting the regex ✗", color=colors.red)
puts(
"Test Failed: Metrics found after deleting the regex ✗",
color=colors.red,
)
exit_test(1)
checkRegex(secret)
# Pause the proxy
if(firegex.nfregex_stop_service(service_id)):
if firegex.nfregex_stop_service(service_id):
puts(f"Sucessfully paused service with id {service_id} ✔", color=colors.green)
else:
puts("Test Failed: Coulnd't pause the service ✗", color=colors.red)
@@ -165,7 +252,7 @@ else:
checkRegex(secret, should_work=False)
# Start firewall
if(firegex.nfregex_start_service(service_id)):
if firegex.nfregex_start_service(service_id):
puts(f"Sucessfully started service with id {service_id} ✔", color=colors.green)
else:
puts("Test Failed: Coulnd't start the service ✗", color=colors.red)
@@ -176,8 +263,11 @@ checkRegex(secret)
# Disable regex
for r in firegex.nfregex_get_service_regexes(service_id):
if r["regex"] == secret:
if(firegex.nfregex_disable_regex(r["id"])):
puts(f"Sucessfully disabled regex with id {r['id']} ✔", color=colors.green)
if firegex.nfregex_disable_regex(r["id"]):
puts(
f"Sucessfully disabled regex with id {r['id']} ✔",
color=colors.green,
)
else:
puts("Test Failed: Coulnd't disable the regex ✗", color=colors.red)
exit_test(1)
@@ -189,8 +279,10 @@ checkRegex(secret,should_work=False)
# Enable regex
for r in firegex.nfregex_get_service_regexes(service_id):
if r["regex"] == secret:
if(firegex.nfregex_enable_regex(r["id"])):
puts(f"Sucessfully enabled regex with id {r['id']} ✔", color=colors.green)
if firegex.nfregex_enable_regex(r["id"]):
puts(
f"Sucessfully enabled regex with id {r['id']} ✔", color=colors.green
)
else:
puts("Test Failed: Coulnd't enable the regex ✗", color=colors.red)
exit_test(1)
@@ -205,10 +297,18 @@ clear_regexes()
checkRegex(secret, should_work=False, deleted=True)
# Add case insensitive regex
if(firegex.nfregex_add_regex(service_id,secret,"B",active=True, is_case_sensitive=False)):
puts(f"Sucessfully added case insensitive regex {str(secret)} ✔", color=colors.green)
if firegex.nfregex_add_regex(
service_id, secret, "B", active=True, is_case_sensitive=False
):
puts(
f"Sucessfully added case insensitive regex {str(secret)} ✔",
color=colors.green,
)
else:
puts(f"Test Failed: Coulnd't add the case insensitive regex {str(secret)} ✗", color=colors.red)
puts(
f"Test Failed: Coulnd't add the case insensitive regex {str(secret)} ✗",
color=colors.red,
)
exit_test(1)
checkRegex(secret, upper=True)
@@ -217,8 +317,10 @@ checkRegex(secret)
clear_regexes()
# Rename service
if(firegex.nfregex_rename_service(service_id,f"{args.service_name}2")):
puts(f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green)
if firegex.nfregex_rename_service(service_id, f"{args.service_name}2"):
puts(
f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green
)
else:
puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
exit_test(1)
@@ -232,20 +334,37 @@ else:
exit_test(1)
# Rename back service
if(firegex.nfregex_rename_service(service_id,f"{args.service_name}")):
puts(f"Sucessfully renamed service to {args.service_name} ✔", color=colors.green)
if firegex.nfregex_rename_service(service_id, f"{args.service_name}"):
puts(
f"Sucessfully renamed service to {args.service_name} ✔", color=colors.green
)
else:
puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
exit_test(1)
# Change settings
opposite_proto = "udp" if args.proto == "tcp" else "tcp"
if(firegex.nfregex_settings_service(service_id, 1338, opposite_proto, "::dead:beef" if args.ipv6 else "123.123.123.123", True)):
if firegex.nfregex_settings_service(
service_id,
1338,
opposite_proto,
"::dead:beef" if args.ipv6 else "123.123.123.123",
True,
):
srv_updated = firegex.nfregex_get_service(service_id)
if srv_updated["port"] == 1338 and srv_updated["proto"] == opposite_proto 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 srv_updated["proto"] == opposite_proto
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)
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)
else:
puts("Test Failed: Coulnd't change service settings ✗", color=colors.red)

87
tests/ph_test.py
View File

@@ -7,13 +7,47 @@ import argparse
import secrets
import time
if __name__ == "__main__":
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("--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("--proto", "-m" , type=str, required=False, choices=["tcp","udp"], help='Select the protocol', default="tcp")
parser.add_argument(
"--address",
"-a",
type=str,
required=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(
"--proto",
"-m",
type=str,
required=False,
choices=["tcp", "udp"],
help="Select the protocol",
default="tcp",
)
args = parser.parse_args()
sep()
@@ -23,19 +57,21 @@ puts(f"{args.address}", color=colors.yellow)
firegex = FiregexAPI(args.address)
# Login
if (firegex.login(args.password)):
if firegex.login(args.password):
puts("Sucessfully logged in ✔", color=colors.green)
else:
puts("Test Failed: Unknown response or wrong passowrd ✗", color=colors.red)
exit(1)
# Create server
server = (TcpServer if args.proto == "tcp" else UdpServer)(args.port+1,ipv6=args.ipv6,proxy_port=args.port)
server = (TcpServer if args.proto == "tcp" else UdpServer)(
args.port + 1, ipv6=args.ipv6, proxy_port=args.port
)
def exit_test(code):
if service_id:
server.stop()
if(firegex.ph_delete_service(service_id)):
if firegex.ph_delete_service(service_id):
puts("Sucessfully deleted service ✔", color=colors.green)
else:
puts("Test Failed: Coulnd't delete serivce ✗", color=colors.red)
@@ -44,18 +80,25 @@ def exit_test(code):
srvs = firegex.ph_get_services()
for ele in srvs:
if ele['name'] == args.service_name:
firegex.ph_delete_service(ele['service_id'])
if ele["name"] == args.service_name:
firegex.ph_delete_service(ele["service_id"])
# Create and start serivce
service_id = firegex.ph_add_service(args.service_name, args.port, args.port+1, args.proto , "::1" if args.ipv6 else "127.0.0.1", "::1" if args.ipv6 else "127.0.0.1")
service_id = firegex.ph_add_service(
args.service_name,
args.port,
args.port + 1,
args.proto,
"::1" if args.ipv6 else "127.0.0.1",
"::1" if args.ipv6 else "127.0.0.1",
)
if service_id:
puts(f"Sucessfully created service {service_id} ✔", color=colors.green)
else:
puts("Test Failed: Failed to create service ✗", color=colors.red)
exit(1)
if(firegex.ph_start_service(service_id)):
if firegex.ph_start_service(service_id):
puts("Sucessfully started service ✔", color=colors.green)
else:
puts("Test Failed: Failed to start service ✗", color=colors.red)
@@ -87,7 +130,7 @@ def checkData(should_work):
checkData(True)
# Pause the proxy
if(firegex.ph_stop_service(service_id)):
if firegex.ph_stop_service(service_id):
puts(f"Sucessfully paused service with id {service_id} ✔", color=colors.green)
else:
puts("Test Failed: Coulnd't pause the service ✗", color=colors.red)
@@ -96,7 +139,7 @@ else:
checkData(False)
# Start firewall
if(firegex.ph_start_service(service_id)):
if firegex.ph_start_service(service_id):
puts(f"Sucessfully started service with id {service_id} ✔", color=colors.green)
else:
puts("Test Failed: Coulnd't start the service ✗", color=colors.red)
@@ -105,7 +148,9 @@ else:
checkData(True)
# Change port
if(firegex.ph_change_destination(service_id, "::1" if args.ipv6 else "127.0.0.1", args.port+2)):
if firegex.ph_change_destination(
service_id, "::1" if args.ipv6 else "127.0.0.1", args.port + 2
):
puts("Sucessfully changed port ✔", color=colors.green)
else:
puts("Test Failed: Coulnd't change destination ✗", color=colors.red)
@@ -114,15 +159,19 @@ else:
checkData(False)
server.stop()
server = (TcpServer if args.proto == "tcp" else UdpServer)(args.port+2,ipv6=args.ipv6,proxy_port=args.port)
server = (TcpServer if args.proto == "tcp" else UdpServer)(
args.port + 2, ipv6=args.ipv6, proxy_port=args.port
)
server.start()
time.sleep(0.5)
checkData(True)
# Rename service
if(firegex.ph_rename_service(service_id,f"{args.service_name}2")):
puts(f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green)
if firegex.ph_rename_service(service_id, f"{args.service_name}2"):
puts(
f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green
)
else:
puts("Test Failed: Coulnd't rename service ✗", color=colors.red)
exit_test(1)

76
tests/utils/tcpserver.py
View File

@@ -1,7 +1,44 @@
import queue
from multiprocessing import Process, Queue
import socket
import traceback
def _start_tcp_server(port, server_queue: Queue, ipv6, verbose):
sock = socket.socket(
socket.AF_INET6 if ipv6 else socket.AF_INET, socket.SOCK_STREAM
)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(("::1" if ipv6 else "127.0.0.1", port))
sock.listen(8)
while True:
connection, address = sock.accept()
while True:
try:
buf = connection.recv(4096)
if buf == b"":
break
reply = buf # Default to echo
try:
# See if there is a custom reply, but don't block
custom_reply = server_queue.get(block=False)
reply = custom_reply
except queue.Empty:
pass # No custom reply, just echo
if verbose:
print("SERVER: ", reply)
connection.sendall(reply)
except (ConnectionResetError, BrokenPipeError):
break # Client closed connection
except Exception:
if verbose:
traceback.print_exc()
break # Exit on other errors
connection.close()
class TcpServer:
def __init__(self, port, ipv6, proxy_port=None, verbose=False):
self.proxy_port = proxy_port
@@ -12,30 +49,10 @@ class TcpServer:
self._regen_process()
def _regen_process(self):
def _startServer(port, server_queue:Queue):
sock = socket.socket(socket.AF_INET6 if self.ipv6 else socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('::1' if self.ipv6 else '127.0.0.1', port))
sock.listen(8)
while True:
connection,address = sock.accept()
while True:
try:
buf = connection.recv(4096)
if buf == b'':
break
try:
buf = server_queue.get(block=False)
except Exception:
pass
if self.verbose:
print("SERVER: ", buf)
connection.sendall(buf)
except Exception:
if self.verbose:
traceback.print_exc()
connection.close()
self.server = Process(target=_startServer,args=[self.port, self._server_data_queue])
self.server = Process(
target=_start_tcp_server,
args=[self.port, self._server_data_queue, self.ipv6, self.verbose],
)
def start(self):
self.server.start()
@@ -46,9 +63,16 @@ class TcpServer:
self._regen_process()
def connect_client(self):
self.client_sock = socket.socket(socket.AF_INET6 if self.ipv6 else socket.AF_INET, socket.SOCK_STREAM)
self.client_sock = socket.socket(
socket.AF_INET6 if self.ipv6 else socket.AF_INET, socket.SOCK_STREAM
)
self.client_sock.settimeout(1)
self.client_sock.connect(('::1' if self.ipv6 else '127.0.0.1', self.proxy_port if self.proxy_port else self.port))
self.client_sock.connect(
(
"::1" if self.ipv6 else "127.0.0.1",
self.proxy_port if self.proxy_port else self.port,
)
)
def close_client(self):
if self.client_sock:

91
tests/utils/udpserver.py
View File

@@ -1,35 +1,94 @@
from multiprocessing import Process
from multiprocessing import Process, Queue
import socket
import queue
import traceback
class UdpServer:
def __init__(self,port,ipv6, proxy_port = None):
def _startServer(port):
def _start_udp_server(port, server_queue: Queue, ipv6, verbose):
sock = socket.socket(socket.AF_INET6 if ipv6 else socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('::1' if ipv6 else '127.0.0.1', port))
sock.bind(("::1" if ipv6 else "127.0.0.1", port))
while True:
bytesAddressPair = sock.recvfrom(432)
try:
bytesAddressPair = sock.recvfrom(4096)
message = bytesAddressPair[0]
address = bytesAddressPair[1]
sock.sendto(message, address)
self.ipv6 = ipv6
reply = message # Default to echo
try:
# See if there is a custom reply, but don't block
custom_reply = server_queue.get(block=False)
reply = custom_reply
except queue.Empty:
pass # No custom reply, just echo
if verbose:
print(f"SERVER: sending {reply} to {address}")
sock.sendto(reply, address)
except Exception:
if verbose:
traceback.print_exc()
class UdpServer:
def __init__(self, port, ipv6, proxy_port=None, verbose=False):
self.port = port
self.ipv6 = ipv6
self.proxy_port = proxy_port
self.server = Process(target=_startServer,args=[port])
self.verbose = verbose
self._server_data_queue = Queue()
self._regen_process()
def _regen_process(self):
self.server = Process(
target=_start_udp_server,
args=[self.port, self._server_data_queue, self.ipv6, self.verbose],
)
def start(self):
self.server.start()
def stop(self):
self.server.terminate()
self.server.join()
self._regen_process()
def sendCheckData(self,data):
s = socket.socket(socket.AF_INET6 if self.ipv6 else socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(2)
s.sendto(data, ('::1' if self.ipv6 else '127.0.0.1', self.proxy_port if self.proxy_port else self.port))
def connect_client(self):
self.client_sock = socket.socket(
socket.AF_INET6 if self.ipv6 else socket.AF_INET, socket.SOCK_DGRAM
)
self.client_sock.settimeout(1)
self.client_sock.connect(
(
"::1" if self.ipv6 else "127.0.0.1",
self.proxy_port if self.proxy_port else self.port,
)
)
def close_client(self):
if self.client_sock:
self.client_sock.close()
def send_packet(self, packet, server_reply=None):
if self.verbose:
print("CLIENT: ", packet)
if server_reply:
self._server_data_queue.put(server_reply)
self.client_sock.sendall(packet)
def recv_packet(self):
try:
received_data = s.recvfrom(432)
except Exception:
return self.client_sock.recv(4096)
except (TimeoutError, ConnectionResetError):
if self.verbose:
traceback.print_exc()
return False
return received_data[0] == data
def sendCheckData(self, data, get_data=False):
self.connect_client()
self.send_packet(data)
received_data = self.recv_packet()
self.close_client()
if get_data:
return received_data
return received_data == data