From b4af7f150d0d1a71af0df189df6c0b7c07bfd33f Mon Sep 17 00:00:00 2001 From: nik012003 Date: Tue, 2 Aug 2022 12:54:28 +0200 Subject: [PATCH] Proxy Regex tests complete --- tests/px_test.py | 248 ++++++++++++++++++++++++++++++++++++++ tests/run_tests.sh | 4 +- tests/utils/firegexapi.py | 20 ++- tests/utils/tcpserver.py | 5 +- 4 files changed, 268 insertions(+), 9 deletions(-) create mode 100755 tests/px_test.py diff --git a/tests/px_test.py b/tests/px_test.py new file mode 100755 index 0000000..22ae91f --- /dev/null +++ b/tests/px_test.py @@ -0,0 +1,248 @@ +#!/usr/bin/env python3 +from utils.colors import * +from utils.firegexapi import * +from utils.tcpserver import TcpServer +import argparse, secrets, base64,time,random + +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) +args = parser.parse_args() +sep() +puts(f"Testing will start on ", color=colors.cyan, end="") +puts(f"{args.address}", color=colors.yellow) + +#Create and start server +server = TcpServer(args.port,ipv6=False) +server.start() +time.sleep(0.5) + +firegex = FiregexAPI(args.address) + +#Login +if (firegex.login(args.password)): puts(f"Sucessfully logged in ✔", color=colors.green) +else: puts(f"Test Failed: Unknown response or wrong passowrd ✗", color=colors.red); exit(1) + +def exit_test(code): + if service_id: + server.stop() + if(firegex.px_delete_service(service_id)): puts(f"Sucessfully deleted service ✔", color=colors.green) + else: puts(f"Test Failed: Coulnd't deleted serivce ✗", color=colors.red); exit_test(1) + exit(code) + +#Create service +service_id = firegex.px_add_service(args.service_name, args.port, 6140) +if service_id: puts(f"Sucessfully created service {service_id} ✔", color=colors.green) +else: puts(f"Test Failed: Failed to create service ✗", color=colors.red); exit(1) + +if(firegex.px_start_service(service_id)): puts(f"Sucessfully started service ✔", color=colors.green) +else: puts(f"Test Failed: Failed to start service ✗", color=colors.red); exit_test(1) + +#Check if service is in wait mode +if(firegex.px_get_service(service_id)["status"] == "wait"): puts(f"Sucessfully started service in WAIT mode ✔", color=colors.green) +else: puts(f"Test Failed: Service not in WAIT mode ✗", color=colors.red); exit_test(1) + +#Get inernal_port +internal_port = firegex.px_get_service(service_id)["internal_port"] +if (internal_port): puts(f"Sucessfully got internal port {internal_port} ✔", color=colors.green) +else: puts(f"Test Failed: Coundn't get internal_port ✗", color=colors.red); exit_test(1) + +server.stop() +server = TcpServer(internal_port,ipv6=False, proxy_port=args.port) +server.start() +time.sleep(1) + +if(firegex.px_get_service(service_id)["status"] == "active"): puts(f"Service went in ACTIVE mode ✔", color=colors.green) +else: puts(f"Test Failed: Service not in ACTIVE mode ✗", color=colors.red); exit_test(1) + +if server.sendCheckData(secrets.token_bytes(432)): + puts(f"Successfully tested first proxy with no regex ✔", color=colors.green) +else: + puts(f"Test Failed: Data was corrupted ", color=colors.red); exit_test(1) + +#Add new regex +secret = bytes(secrets.token_hex(16).encode()) +regex = base64.b64encode(secret).decode() +if(firegex.px_add_regex(service_id,regex,"B",active=True,is_blacklist=True,is_case_sensitive=True)): + puts(f"Sucessfully added regex {str(secret)} ✔", color=colors.green) +else: puts(f"Test Failed: Coulnd't add the regex {str(secret)} ✗", color=colors.red); exit_test(1) + +#Check if regex is present in the service +n_blocked = 0 + +def checkRegex(regex, should_work=True, upper=False): + if should_work: + global n_blocked + for r in firegex.px_get_service_regexes(service_id): + if r["regex"] == regex: + #Test the regex + s = base64.b64decode(regex).upper() if upper else base64.b64decode(regex) + if not server.sendCheckData(secrets.token_bytes(200) + s + secrets.token_bytes(200)): + puts(f"The malicious request was successfully blocked ✔", color=colors.green) + n_blocked += 1 + if firegex.px_get_regex(r["id"])["n_packets"] == n_blocked: + puts(f"The packed was reported as blocked ✔", color=colors.green) + else: + puts(f"Test Failed: The packed wasn't reported as blocked ✗", color=colors.red); exit_test(1) + else: + puts(f"Test Failed: The request wasn't blocked ✗", color=colors.red);exit_test(1) + return + puts(f"Test Failed: The regex wasn't found ✗", color=colors.red); exit_test(1) + else: + if server.sendCheckData(secrets.token_bytes(200) + base64.b64decode(regex) + secrets.token_bytes(200)): + puts(f"The request wasn't blocked ✔", color=colors.green) + else: + puts(f"Test Failed: The request was blocked when it shouldn't have", color=colors.red) + +checkRegex(regex) + +#Pause the proxy +if(firegex.px_pause_service(service_id)): puts(f"Sucessfully paused service with id {service_id} ✔", color=colors.green) +else: puts(f"Test Failed: Coulnd't pause the service ✗", color=colors.red); exit_test(1) + +#Check if it's actually paused +checkRegex(regex,should_work=False) + +#Start firewall +if(firegex.px_start_service(service_id)): puts(f"Sucessfully started service with id {service_id} ✔", color=colors.green) +else: puts(f"Test Failed: Coulnd't start the service ✗", color=colors.red); exit_test(1) + +checkRegex(regex) + +#Stop firewall +if(firegex.px_stop_service(service_id)): puts(f"Sucessfully stopped service with id {service_id} ✔", color=colors.green) +else: puts(f"Test Failed: Coulnd't stop the service ✗", color=colors.red); exit_test(1) + +try: + checkRegex(regex) + puts(f"Test Failed: The service was still active ✗", color=colors.red); exit_test(1) +except Exception: + puts(f"Service was correctly stopped ✔", color=colors.green) + +#Start firewall in pause +if(firegex.px_pause_service(service_id)): puts(f"Sucessfully started service in pause mode with id {service_id} ✔", color=colors.green) +else: puts(f"Test Failed: Coulnd't start the service ✗", color=colors.red); exit_test(1) + +time.sleep(0.5) +#Check if it's actually paused +checkRegex(regex,should_work=False) + +#Start firewall +if(firegex.px_start_service(service_id)): puts(f"Sucessfully started service with id {service_id} ✔", color=colors.green) +else: puts(f"Test Failed: Coulnd't start the service ✗", color=colors.red); exit_test(1) + +checkRegex(regex) + +#Disable regex +for r in firegex.px_get_service_regexes(service_id): + if r["regex"] == regex: + if(firegex.px_disable_regex(r["id"])): + puts(f"Sucessfully disabled regex with id {r['id']} ✔", color=colors.green) + else: + puts(f"Test Failed: Coulnd't disable the regex ✗", color=colors.red); exit_test(1) + break + +#Check if it's actually disabled +checkRegex(regex,should_work=False) + +#Enable regex +for r in firegex.px_get_service_regexes(service_id): + if r["regex"] == regex: + if(firegex.px_enable_regex(r["id"])): + puts(f"Sucessfully enabled regex with id {r['id']} ✔", color=colors.green) + else: + puts(f"Test Failed: Coulnd't enable the regex ✗", color=colors.red); exit_test(1) + break + +checkRegex(regex) + +#Delete regex +n_blocked = 0 +for r in firegex.px_get_service_regexes(service_id): + if r["regex"] == regex: + if(firegex.px_delete_regex(r["id"])): + puts(f"Sucessfully deleted regex with id {r['id']} ✔", color=colors.green) + else: + puts(f"Test Failed: Coulnd't delete the regex ✗", color=colors.red); exit_test(1) + break + +#Check if it's actually deleted +checkRegex(regex,should_work=False) + +#Add case insensitive regex +if(firegex.px_add_regex(service_id,regex,"B",active=True,is_blacklist=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); exit_test(1) + +checkRegex(regex,upper=True) +checkRegex(regex) + +#Delete regex +n_blocked = 0 +for r in firegex.px_get_service_regexes(service_id): + if r["regex"] == regex: + if(firegex.px_delete_regex(r["id"])): + puts(f"Sucessfully deleted regex with id {r['id']} ✔", color=colors.green) + else: + puts(f"Test Failed: Coulnd't delete the regex ✗", color=colors.red); exit_test(1) + break + +#Add whitelist regex +if(firegex.px_add_regex(service_id,regex,"B",active=True,is_blacklist=False,is_case_sensitive=True)): + puts(f"Sucessfully added case whitelist regex {str(secret)} ✔", color=colors.green) +else: puts(f"Test Failed: Coulnd't add the case whiteblist regex {str(secret)} ✗", color=colors.red); exit_test(1) + +checkRegex(regex,should_work=False) +checkRegex(regex,upper=True) #Dirty way to test the whitelist :p + +#Delete regex +n_blocked = 0 +for r in firegex.px_get_service_regexes(service_id): + if r["regex"] == regex: + if(firegex.px_delete_regex(r["id"])): + puts(f"Sucessfully deleted regex with id {r['id']} ✔", color=colors.green) + else: + puts(f"Test Failed: Coulnd't delete the regex ✗", color=colors.red); exit_test(1) + break + +#Rename service +if(firegex.px_rename_service(service_id,f"{args.service_name}2")): puts(f"Sucessfully renamed service to {args.service_name}2 ✔", color=colors.green) +else: puts(f"Test Failed: Coulnd't rename service ✗", color=colors.red); exit_test(1) + +#Check if service was renamed correctly +found = False +for services in firegex.px_get_services(): + if services["name"] == f"{args.service_name}2": + puts(f"Checked that service was renamed correctly ✔", color=colors.green) + found = True + break + +if not found: + puts(f"Test Failed: Service wasn't renamed correctly ✗", color=colors.red); exit_test(1) + exit(1) + +#Change service port +new_internal_port = random.randrange(6000,9000) +if(firegex.px_change_service_port(service_id,internalPort=new_internal_port)): + puts(f"Sucessfully changed internal_port to {new_internal_port} ✔", color=colors.green) +else: + puts(f"Test Failed: Coulnd't change intenral port ✗", color=colors.red); exit_test(1) + +#Get inernal_port +internal_port = firegex.px_get_service(service_id)["internal_port"] +if (internal_port == new_internal_port): puts(f"Sucessfully got internal port {internal_port} ✔", color=colors.green) +else: puts(f"Test Failed: Coundn't get internal_port or port changed incorrectly ✗", color=colors.red); exit_test(1) + +if(firegex.px_regen_service_port(service_id)): + puts(f"Sucessfully changed internal_port to {new_internal_port} ✔", color=colors.green) +else: + puts(f"Test Failed: Coulnd't change internal port ✗", color=colors.red); exit_test(1) + +#Get regenerated inernal_port +new_internal_port = firegex.px_get_service(service_id)["internal_port"] +if (internal_port != new_internal_port): puts(f"Sucessfully got regenerated port {new_internal_port} ✔", color=colors.green) +else: puts(f"Test Failed: Coundn't get internal port, or it was the same as previous ✗", color=colors.red); exit_test(1) + +exit_test(0) \ No newline at end of file diff --git a/tests/run_tests.sh b/tests/run_tests.sh index 4dc9707..f210baf 100755 --- a/tests/run_tests.sh +++ b/tests/run_tests.sh @@ -8,4 +8,6 @@ python3 nf_test.py -p testpassword -m tcp -6 echo "Running Netfilter Regex UDP ipv4" python3 nf_test.py -p testpassword -m udp echo "Running Netfilter Regex UDP ipv6" -python3 nf_test.py -p testpassword -m udp -6 \ No newline at end of file +python3 nf_test.py -p testpassword -m udp -6 +echo "Running Proxy Regex" +python3 px_test.py -p testpassword \ No newline at end of file diff --git a/tests/utils/firegexapi.py b/tests/utils/firegexapi.py index d54389e..928cdbb 100644 --- a/tests/utils/firegexapi.py +++ b/tests/utils/firegexapi.py @@ -159,11 +159,15 @@ class FiregexAPI: req = self.s.get(f"{self.address}api/regexproxy/service/{service_id}/delete") return verify(req) - def px_change_service_port(self,service_id, port, internalPort): + def px_regen_service_port(self,service_id): + req = self.s.get(f"{self.address}api/regexproxy/service/{service_id}/regen-port") + return verify(req) + + def px_change_service_port(self,service_id, port=None, internalPort=None): payload = {} if port: payload["port"] = port if internalPort: payload["internalPort"] = internalPort - req = self.s.post(f"{self.address}api/regexproxy/service/{service_id}/start", json=payload) + req = self.s.post(f"{self.address}api/regexproxy/service/{service_id}/change-ports", json=payload) return req.json() if verify(req) else False def px_get_service_regexes(self,service_id): @@ -195,7 +199,11 @@ class FiregexAPI: req = self.s.post(f"{self.address}api/regexproxy/service/{service_id}/rename" , json={"name":newname}) return verify(req) - def px_add_service(self, name: str, port: int, internalPort: [int,None]): - req = self.s.post(f"{self.address}api/regexproxy/services/add" , - json={"name":name,"port":port, "internalPort": internalPort}) - return req.json()["service_id"] if verify(req) else False \ No newline at end of file + def px_add_service(self, name: str, port: int, internalPort = None): + payload = {} + payload["name"] = name + payload["port"] = port + if internalPort: + payload["internalPort"] = internalPort + req = self.s.post(f"{self.address}api/regexproxy/services/add" , json=payload) + return req.json()["id"] if verify(req) else False \ No newline at end of file diff --git a/tests/utils/tcpserver.py b/tests/utils/tcpserver.py index 6396d4e..0cfd37d 100644 --- a/tests/utils/tcpserver.py +++ b/tests/utils/tcpserver.py @@ -2,7 +2,7 @@ from multiprocessing import Process import socket class TcpServer: - def __init__(self,port,ipv6): + def __init__(self,port,ipv6,proxy_port=None): def _startServer(port): sock = socket.socket(socket.AF_INET6 if ipv6 else socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) @@ -13,6 +13,7 @@ class TcpServer: buf = connection.recv(4096) connection.send(buf) connection.close() + self.proxy_port = proxy_port self.ipv6 = ipv6 self.port = port self.server = Process(target=_startServer,args=[port]) @@ -25,7 +26,7 @@ class TcpServer: def sendCheckData(self,data): s = socket.socket(socket.AF_INET6 if self.ipv6 else socket.AF_INET, socket.SOCK_STREAM) - s.connect(('::1' if self.ipv6 else '127.0.0.1', self.port)) + s.connect(('::1' if self.ipv6 else '127.0.0.1', self.proxy_port if self.proxy_port else self.port)) s.sendall(data) received_data = s.recv(4096) s.close()