written new tests
This commit is contained in:
nik012003
112
tests/test.py
112
tests/test.py
@@ -2,7 +2,7 @@
|
||||
import argparse, socket,secrets, base64
|
||||
from time import sleep
|
||||
from multiprocessing import Process
|
||||
from proxyapi import ProxyAPI
|
||||
from firegexapi import FiregexAPI
|
||||
|
||||
pref = "\033["
|
||||
reset = f"{pref}0m"
|
||||
@@ -33,16 +33,16 @@ puts(f"Testing will start on ", color=colors.cyan, end="")
|
||||
puts(f"{args.address}", color=colors.yellow)
|
||||
|
||||
|
||||
proxy = ProxyAPI(args.address,args.password)
|
||||
firegex = FiregexAPI(args.address)
|
||||
service_created = False
|
||||
|
||||
#Connect to Firegex
|
||||
if (proxy.connect()): puts(f"Sucessfully logged in ✔", color=colors.green)
|
||||
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)
|
||||
|
||||
|
||||
#Create new Service
|
||||
if (proxy.create_service(args.service_name,args.service_port)):
|
||||
if (firegex.create_service(args.service_name,args.service_port)):
|
||||
puts(f"Sucessfully created service {args.service_name} with public port {args.service_port} ✔", color=colors.green)
|
||||
service_created = True
|
||||
else: puts(f"Test Failed: Couldn't create service ✗", color=colors.red); exit(1)
|
||||
@@ -50,7 +50,7 @@ else: puts(f"Test Failed: Couldn't create service ✗", color=colors.red); exit(
|
||||
#Delete the Service and exit
|
||||
def exit_test(status_code=0):
|
||||
if service_created:
|
||||
if(proxy.delete(service_id)):
|
||||
if(firegex.delete(service_id)):
|
||||
puts(f"Sucessfully delete service with id {service_id} ✔", color=colors.green)
|
||||
else:
|
||||
puts(f"Test Failed: Couldn't delete service ✗", color=colors.red); exit(1)
|
||||
@@ -59,13 +59,17 @@ def exit_test(status_code=0):
|
||||
exit(status_code)
|
||||
|
||||
#Find the Service
|
||||
service_id, internal_port = proxy.get_service_details(args.service_name)
|
||||
if (internal_port): puts(f"Sucessfully received the internal port {internal_port} ✔", color=colors.green)
|
||||
service = firegex.get_service_details(args.service_name)
|
||||
if (service):
|
||||
internal_port= service["internal_port"]
|
||||
service_id = service["id"]
|
||||
puts(f"Sucessfully received the internal port {internal_port} ✔", color=colors.green)
|
||||
else: puts(f"Test Failed: Coulnd't get the service internal port ✗", color=colors.red); exit_test(1)
|
||||
|
||||
#Start listener
|
||||
def startServer(port):
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
sock.bind(('127.0.0.1', port))
|
||||
sock.listen(8)
|
||||
while True:
|
||||
@@ -78,7 +82,7 @@ server = Process(target=startServer,args=[internal_port])
|
||||
server.start()
|
||||
|
||||
#Start firewall
|
||||
if(proxy.start(service_id)): puts(f"Sucessfully started service with id {service_id} ✔", color=colors.green)
|
||||
if(firegex.start(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)
|
||||
|
||||
#Hacky solution - wait a bit for the server to start
|
||||
@@ -101,25 +105,33 @@ else:
|
||||
#Add new regex
|
||||
secret = bytes(secrets.token_hex(16).encode())
|
||||
regex = base64.b64encode(secret).decode()
|
||||
if(proxy.add_regex(service_id,regex)): puts(f"Sucessfully added regex {secret} ✔", color=colors.green)
|
||||
else: puts(f"Test Failed: Coulnd't start the service ✗", color=colors.red); exit_test(1)
|
||||
if(firegex.add_regex(service_id,regex)): puts(f"Sucessfully added regex {secret} ✔", color=colors.green)
|
||||
else: puts(f"Test Failed: Coulnd't add the regex {secret} ✗", color=colors.red); exit_test(1)
|
||||
|
||||
#Check if regex is present in the service
|
||||
n_blocked = 0
|
||||
|
||||
def checkRegex(regex):
|
||||
for i,r in enumerate(proxy.get_service_regexes(service_id)):
|
||||
global n_blocked
|
||||
for r in firegex.get_service_regexes(service_id):
|
||||
if r["regex"] == regex:
|
||||
#Test the regex
|
||||
if not sendCheckData(secrets.token_bytes(200) + secret + secrets.token_bytes(200)):
|
||||
puts(f"The malicious request was successfully blocked ✔", color=colors.green)
|
||||
n_blocked += 1
|
||||
if firegex.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)
|
||||
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)
|
||||
puts(f"Test Failed: The regex wasn't found ✗", color=colors.red); exit_test(1)
|
||||
|
||||
checkRegex(regex)
|
||||
|
||||
#Pause the proxy
|
||||
if(proxy.pause(service_id)): puts(f"Sucessfully paused service with id {service_id} ✔", color=colors.green)
|
||||
if(firegex.pause(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
|
||||
@@ -127,20 +139,68 @@ if sendCheckData(secrets.token_bytes(200) + secret + 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)
|
||||
|
||||
#Stop the proxy
|
||||
if(proxy.stop(service_id)): puts(f"Sucessfully stopped service with id {service_id} ✔", color=colors.green)
|
||||
if(firegex.stop(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)
|
||||
sleep(100)
|
||||
|
||||
#Check if proxy is stopped
|
||||
try:
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.bind(('0.0.0.0', args.service_port))
|
||||
puts(f"I could bind to the port, so {service_id} was stopped correctly ✔", color=colors.green)
|
||||
sock.close()
|
||||
except Exception as e:
|
||||
print(e)
|
||||
puts(f"Test Failed: Coulnd't bind to port {args.service_port} ✗", color=colors.red); exit_test(1)
|
||||
#Check if proxy is stopped and check if WAIT works
|
||||
def bindForSeconds(port,delay=4):
|
||||
try:
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
sock.bind(('127.0.0.1', port))
|
||||
sock.listen()
|
||||
puts(f"The test could bind to port {port} ✔", color=colors.green)
|
||||
sleep(delay)
|
||||
sock.close()
|
||||
except Exception as e:
|
||||
puts(f"The test couldn't bind to port {port} ✗", color=colors.red)
|
||||
|
||||
bindingTest = Process(target=bindForSeconds, args=[args.service_port])
|
||||
bindingTest.start()
|
||||
sleep(1)
|
||||
|
||||
#Restart it
|
||||
if(firegex.pause(service_id)): puts(f"Sucessfully started in pause mode ✔", color=colors.green)
|
||||
else: puts(f"Test Failed: Coulnd't start the service ✗", color=colors.red); exit_test(1)
|
||||
if firegex.get_service_details(args.service_name)["status"] == "wait":
|
||||
puts(f"Service started in WAIT mode ✔", color=colors.green)
|
||||
else:
|
||||
puts(f"Service started but it's not WAIT mode ✗", color=colors.red); exit_test(1)
|
||||
|
||||
bindingTest.join()
|
||||
sleep(1)
|
||||
#Check if service started in pause mode successfully
|
||||
if firegex.get_service_details(args.service_name)["status"] == "pause":
|
||||
puts(f"Service started in PAUSE mode ✔", color=colors.green)
|
||||
else:
|
||||
puts(f"Service is not in PAUSE mode ✗", color=colors.red); exit_test(1)
|
||||
|
||||
if sendCheckData(secrets.token_bytes(200) + secret + 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)
|
||||
|
||||
#Restart it and check
|
||||
if(firegex.start(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)
|
||||
|
||||
#Delete the regex
|
||||
def removeRegex(regex):
|
||||
for r in firegex.get_service_regexes(service_id):
|
||||
if r["regex"] == regex:
|
||||
if(firegex.delete_regex(r["id"])): puts(f"Sucessfully deleted regex ✔", color=colors.green)
|
||||
else: puts(f"Test Failed: Coulnd't deleted the regex ✗", color=colors.red); exit_test(1)
|
||||
return
|
||||
puts(f"Test Failed: The regex wasn't found ✗", color=colors.red)
|
||||
|
||||
removeRegex(regex)
|
||||
|
||||
if sendCheckData(secrets.token_bytes(200) + secret + 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)
|
||||
|
||||
exit_test()
|
||||
Reference in New Issue
Block a user