From 6db8c56290c5b69868a3bf9d9bea724613c615a0 Mon Sep 17 00:00:00 2001 From: TiagoRG Date: Mon, 22 May 2023 20:23:57 +0100 Subject: [PATCH] [LABI] Added directory with mid-semester project --- 1ano/2semestre/labi/labi2023-ap-g6/README.md | 11 + .../labi2023-ap-g6/client_server/client.py | 394 ++++++++++++++++++ .../client_server/common_comm.py | 82 ++++ .../labi2023-ap-g6/client_server/server.py | 384 +++++++++++++++++ .../client_server/test_client.py | 344 +++++++++++++++ .../client_server/test_server.py | 39 ++ .../client_server/test_units.py | 50 +++ 1ano/2semestre/labi/labi2023-ap-g6/readme.txt | 2 + 8 files changed, 1306 insertions(+) create mode 100644 1ano/2semestre/labi/labi2023-ap-g6/README.md create mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/client.py create mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/common_comm.py create mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/server.py create mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/test_client.py create mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/test_server.py create mode 100644 1ano/2semestre/labi/labi2023-ap-g6/client_server/test_units.py create mode 100644 1ano/2semestre/labi/labi2023-ap-g6/readme.txt diff --git a/1ano/2semestre/labi/labi2023-ap-g6/README.md b/1ano/2semestre/labi/labi2023-ap-g6/README.md new file mode 100644 index 0000000..efb8c23 --- /dev/null +++ b/1ano/2semestre/labi/labi2023-ap-g6/README.md @@ -0,0 +1,11 @@ +# labiaprof +Trabalho de aprofundamento de laboratórios de informática + +## Grupo +### Rúben Gomes +* 113435 +* rlcg@ua.pt + +### Tiago Garcia +* 114184 +* tiago.rgarcia@ua.pt diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/client.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/client.py new file mode 100644 index 0000000..f50c2f4 --- /dev/null +++ b/1ano/2semestre/labi/labi2023-ap-g6/client_server/client.py @@ -0,0 +1,394 @@ +#!/usr/bin/python3 +import os +import re +import sys +import socket +import json +import base64 +from common_comm import send_dict, recv_dict, sendrecv_dict + +from Crypto.Cipher import AES +from Crypto.Hash import SHA256 + + +class Tcolors: + HEADER = '\033[95m' + OKBLUE = '\033[94m' + OKCYAN = '\033[96m' + OKGREEN = '\033[92m' + WARNING = '\033[93m' + FAIL = '\033[91m' + ENDC = '\033[0m' + BOLD = '\033[1m' + UNDERLINE = '\033[4m' + + +# Function to encript values for sending in json format +# return int data encrypted in a 16 bytes binary string coded in base64 +def encrypt_intvalue(cipherkey, data): + key = base64.b64decode(cipherkey) + cipher = AES.new(key, AES.MODE_ECB) + data = cipher.encrypt(bytes("%16d" % data, "utf8")) + return str(base64.b64encode(data), "utf8") + + +# Function to decript values received in json format +# return int data decrypted from a 16 bytes binary strings coded in base64 +def decrypt_intvalue(cipherkey, data_arg): + cipher = AES.new(base64.b64decode(cipherkey), AES.MODE_ECB) + data = base64.b64decode(data_arg) + data = cipher.decrypt(data) + return int(str(data, "utf8")) + + +# verify if response from server is valid or is an error message and act accordingly - já está implementada +def validate_response(client_sock, response): + if not response["status"]: + print(f"{Tcolors.FAIL}Error: {response['error']}{Tcolors.ENDC}") + client_sock.close() + sys.exit(3) + + +# process QUIT operation +def quit_action(client_sock, has_started): + print(f"{Tcolors.ENDC}Quitting...") + if has_started: + senddata = {"op": "QUIT"} + recvdata = sendrecv_dict(client_sock, senddata) + + try: + # status = False + if not recvdata["status"]: + print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: {recvdata['error']}{Tcolors.ENDC}") + return + except TypeError: + print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: an error occurred with the server.{Tcolors.ENDC}") + print(f"{Tcolors.ENDC}{Tcolors.WARNING}Client not removed from server, quitting...{Tcolors.ENDC}") + client_sock.close() + exit(1) + + # status = True + print(f"{Tcolors.OKGREEN}Client quit with success") + client_sock.close() + exit(0) + + +# Outcomming message structure: +# { op = "START", client_id, [cipher] } +# { op = "QUIT" } +# { op = "NUMBER", number } +# { op = "STOP", [shasum] } +# { op = "GUESS", choice } +# +# Incomming message structure: +# { op = "START", status } +# { op = "QUIT" , status } +# { op = "NUMBER", status } +# { op = "STOP", status, value } +# { op = "GUESS", status, result } + +# +# Suport for executing the client pretended behaviour +# + +# returns a valid number +def returnValidNum(): + while 1: + try: + num = int(input(f"\n{Tcolors.ENDC}{Tcolors.BOLD}Number > {Tcolors.UNDERLINE}")) + except ValueError: + print(f"{Tcolors.ENDC}{Tcolors.WARNING}Invalid input{Tcolors.ENDC}") + continue + break + return num + + +# verify if port is valid +def verify_port(port): + # verify if port is a number + if not port.isdigit(): + return {"status": False, "error": "Port must be an integer"} + # verify if port is between 1024 and 65535 + if not (1024 <= int(port) <= 65535): + return {"status": False, "error": "Port number must be between 1024 and 65535"} + return {"status": True, "port": int(port)} + + +# verify if hostname is valid +def verify_hostname(hostname): + if hostname == "localhost": + return {"status": True} + if not (re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', hostname) and all(0 <= int(n) < 256 for n in hostname.split('.'))): + return {"status": False, "error": "Invalid DNS address"} + return {"status": True} + + +def run_client(client_sock, client_id): + # Print the welcome message + print(f"{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC}\n") + + # client runtime global variables + has_stopped = False + has_started = False + cipherkey = None + numbers = [] + + while 1: + option = input(f"{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}") + + # start option + if option.upper() == "START": + if has_started: + print(f"{Tcolors.ENDC}{Tcolors.WARNING}Client already started\n{Tcolors.ENDC}") + continue + + while 1: + # ask user if cipher is needed + choice = input(f"\n{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}") + if choice.upper() == "Y": + # create cipher key for server + cipherkey = base64.b64encode(os.urandom(16)).decode() + break + elif choice.upper() == "N": + # do nothing since cipher will be None + break + else: + # loop if invalid option + print(f"{Tcolors.ENDC}{Tcolors.WARNING}Invalid input{Tcolors.ENDC}") + continue + + # send dict and receive response + senddata = {"op": "START", "client_id": client_id, "cipher": cipherkey} + recvdata = sendrecv_dict(client_sock, senddata) + + try: + # status = False + if not recvdata["status"]: + print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: {recvdata['error']}{Tcolors.ENDC}") + print(f"{Tcolors.ENDC}{Tcolors.WARNING}Client not added, quitting...{Tcolors.ENDC}") + client_sock.close() + exit(1) + except TypeError: + print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: an error occurred with the server, try again later{Tcolors.ENDC}") + print(f"{Tcolors.ENDC}{Tcolors.WARNING}Client not added, quitting...{Tcolors.ENDC}") + client_sock.close() + exit(1) + + # status = True + has_started = True + print(f"{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC}\n") + + elif option.upper() == "QUIT": + quit_action(client_sock, has_started) + continue + + elif option.upper() == "NUMBER": + if not has_started: + print(f"{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC}") + continue + + if has_stopped: + print(f"{Tcolors.ENDC}{Tcolors.WARNING}You can't add more numbers\n{Tcolors.ENDC}") + continue + # verify if number is int + num = returnValidNum() + + # encrypt the number is a cipher is being used + if cipherkey is not None: + encrypted_num = encrypt_intvalue(cipherkey, num) + else: + encrypted_num = num + + # send dict and receive response + senddata = {"op": "NUMBER", "number": encrypted_num} + recvdata = sendrecv_dict(client_sock, senddata) + + try: + # status = False + if not recvdata["status"]: + print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: {recvdata['error']}{Tcolors.ENDC}") + client_sock.close() + continue + # status = True + numbers.append(num) + print(f"{Tcolors.ENDC}{Tcolors.OKGREEN}Number added with success{Tcolors.ENDC}\n") + except TypeError: + print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: an error occurred with the server\nSocket has been closed, try to start again{Tcolors.ENDC}") + client_sock.close() + has_started = False + cipherkey = None + numbers = [] + continue + + elif option.upper() == "STOP": + # check if client has started the game + if not has_started: + print(f"{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC}") + continue + # check if client has stopped adding numbers + if has_stopped: + print(f"{Tcolors.ENDC}{Tcolors.WARNING}You can't stop the game again\n{Tcolors.ENDC}") + continue + + # creates the synthesis for the number list + hasher = SHA256.new() + for number in numbers: + hasher.update(bytes(str(number), "utf8")) + + # send dict and receive response + senddata = {"op": "STOP", "shasum": hasher.hexdigest()} + recvdata = sendrecv_dict(client_sock, senddata) + + try: + # status = False + if not recvdata["status"]: + print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: {recvdata['error']}{Tcolors.ENDC}") + continue + except TypeError: + print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: an error occurred with the server\nSocket has been closed, try to start again{Tcolors.ENDC}") + client_sock.close() + has_started = False + cipherkey = None + numbers = [] + continue + + # decipher data if using encryption + data = recvdata["value"] + if cipherkey is not None: + data = decrypt_intvalue(cipherkey, data) + has_stopped = True + # status = True + print(f"{Tcolors.ENDC}{Tcolors.OKGREEN}\nChosen number: {Tcolors.UNDERLINE}{data}{Tcolors.ENDC}\n") + + elif option.upper() == "GUESS": + # check if client has started the game + if not has_started: + print(f"{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC}") + continue + # check if client has stopped adding numbers + if not has_stopped: + print(f"{Tcolors.ENDC}{Tcolors.WARNING}You can't guess before stopping the game\n{Tcolors.ENDC}") + continue + + # print the possible choices + print(f""" +{Tcolors.ENDC}Choose one of the following options: +1 - first +2 - last +3 - min +4 - max +5 - median +6 - min, first +7 - max, first +8 - min, last +9 - max, last +10 - median, first +11 - median, last +""") + while True: + try: + choice_num = int(input(f"{Tcolors.BOLD}\n> {Tcolors.UNDERLINE}")) + if choice_num == 1: + choice = ["first"] + elif choice_num == 2: + choice = ["last"] + elif choice_num == 3: + choice = ["min"] + elif choice_num == 4: + choice = ["max"] + elif choice_num == 5: + choice = ["median"] + elif choice_num == 6: + choice = ["min", "first"] + elif choice_num == 7: + choice = ["max", "first"] + elif choice_num == 8: + choice = ["min", "last"] + elif choice_num == 9: + choice = ["max", "last"] + elif choice_num == 10: + choice = ["median", "first"] + elif choice_num == 11: + choice = ["median", "last"] + else: + print(f"{Tcolors.ENDC}{Tcolors.WARNING}Invalid input{Tcolors.ENDC}") + continue + break + except ValueError: + print(f"{Tcolors.ENDC}{Tcolors.WARNING}Invalid input{Tcolors.ENDC}") + continue + + # send dict and receive response + senddata = {"op": "GUESS", "choice": choice} + recvdata = sendrecv_dict(client_sock, senddata) + + try: + # status = False + if not recvdata["status"]: + print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: {recvdata['error']}{Tcolors.ENDC}") + continue + except TypeError: + print(f"{Tcolors.ENDC}{Tcolors.FAIL}Error: an error occurred with the server\nSocket has been closed, try to start again{Tcolors.ENDC}") + client_sock.close() + has_started = False + cipherkey = None + numbers = [] + continue + + # status = True + print(f"\n\n{Tcolors.ENDC}{Tcolors.BOLD}{Tcolors.OKBLUE}{'='*15}\n\n{Tcolors.UNDERLINE}{Tcolors.OKCYAN}" + + ("You are right!" if recvdata["result"] else "You are wrong!") + + f"{Tcolors.ENDC}{Tcolors.BOLD}{Tcolors.OKBLUE}\n\n{'='*15}{Tcolors.ENDC}\n\n") + quit_action(client_sock, has_started) + else: + print(f"{Tcolors.ENDC}{Tcolors.WARNING}Invalid option!\n{Tcolors.ENDC}") + + return None + + +def main(): + # validate the number of arguments and eventually print error message and exit with error + # verify type of arguments and eventually print error message and exit with error + if len(sys.argv) not in [3, 4]: + print(f"{Tcolors.WARNING}Usage: python3 client.py client_id port DNS{Tcolors.ENDC}") + sys.exit(1) + + # check if indicated port is valid and get its value + port = sys.argv[2] + verified = verify_port(port) + if not verified["status"]: + print(f"{Tcolors.WARNING}{verified['error']}{Tcolors.ENDC}") + sys.exit(1) + port = verified["port"] + + # get the ip address of the DNS and get its value + hostname = sys.argv[3] if len(sys.argv) == 4 else socket.gethostbyname(socket.gethostname()) + verified = verify_hostname(hostname) + if not verified["status"]: + print(f"{Tcolors.WARNING}{verified['error']}{Tcolors.ENDC}") + sys.exit(1) + + # create the socket + client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + client_socket.bind(("0.0.0.0", 0)) + + # catch error message if server does not exist in those specifications + print(f"{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC}") + try: + client_socket.connect((hostname, port)) + except OSError: + print(f"{Tcolors.FAIL}Error: connection to server failed{Tcolors.ENDC}") + sys.exit(1) + + # send confirmation about the connection + print(f"{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{sys.argv[1]}\n{Tcolors.ENDC}") + + # run the client + run_client(client_socket, sys.argv[1]) + + client_socket.close() + sys.exit(0) + + +if __name__ == "__main__": + main() diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/common_comm.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/common_comm.py new file mode 100644 index 0000000..8862cf5 --- /dev/null +++ b/1ano/2semestre/labi/labi2023-ap-g6/client_server/common_comm.py @@ -0,0 +1,82 @@ +import socket +import json +import base64 + + +# +# Universal function to send a given amount of data to a TCP socket. +# It returns True or False, depending on the success of +# sending all the data to the socket. +# +def exact_send(dst, data): + try: + while len(data) != 0: + bytes_sent = dst.send(data) + data = data[bytes_sent:] + return True + except OSError: + return False + + +# +# Universal function to receive a given amount of data from a TCP socket. +# It returns None or data, depending on the success of +# receiving all the required data from the socket. +# +def exact_recv(src, count): + data = bytearray(0) + while count != 0: + new_data = src.recv(count) + + if len(new_data) == 0: return None + + data += new_data + count -= len(new_data) + + return data + + +# +# Universal function to send a dictionary message to a TCP socket. +# It actually transmits a JSON object, prefixed by its length (in network byte order). +# +# The JSON object is created from the dictionary. +# It returns True or False, depending on the success of sending the +# message to the socket. +# +def send_dict(dst, msg): + # DEBUG print ("Send: %s" % (msg)) + data = bytes(json.dumps(msg), "utf8") + prefixed_data = len(data).to_bytes(4, "big") + data + return exact_send(dst, prefixed_data) + + +# +# Universal function to receive a dictionary message from a TCP socket. +# It actually receives a JSON object, prefixed by its length (in network byte order). +# The dictionary is created from that JSON object. +# +def recv_dict(src): + prefix = exact_recv(src, 4) + + if prefix == None: return None + + length = int.from_bytes(prefix, "big") + data = exact_recv(src, length) + + if data == None: return None + + msg = json.loads(str(data, "utf8")) + # DEBUG print ("Recv: %s" % (msg)) + return msg + + +# +# Universal function to send and receive a dictionary to/from a TCP socket peer. +# It returns None upon an error. +# +def sendrecv_dict(peer, msg): + if send_dict(peer, msg): + return recv_dict(peer) + else: + return None diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/server.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/server.py new file mode 100644 index 0000000..39f0b82 --- /dev/null +++ b/1ano/2semestre/labi/labi2023-ap-g6/client_server/server.py @@ -0,0 +1,384 @@ +#!/usr/bin/python3 +import sys +import socket +import select +import json +import base64 +import csv +import random +from common_comm import send_dict, recv_dict, sendrecv_dict + +from Crypto.Cipher import AES +from Crypto.Hash import SHA256 + +# Dicionário com a informação relativa aos clientes +users = {} + + +# return the client_id of a socket or None +def find_client_id(client_sock): + for client_id in users: + if users[client_id]["socket"] == client_sock: + return client_id + return None + + +# Função para encriptar valores a enviar em formato json com codificação base64 +# return int data encrypted in a 16 bytes binary string and coded base64 +def encrypt_intvalue(client_id, data_arg): + key = base64.b64decode(users[client_id]["cipher"]) + cipher = AES.new(key, AES.MODE_ECB) + data = cipher.encrypt(bytes("%16d" % data_arg, "utf8")) + return str(base64.b64encode(data), "utf8") + + +# Função para desencriptar valores recebidos em formato json com codificação base64 +# return int data decrypted from a 16 bytes binary string and coded base64 +def decrypt_intvalue(client_id, data_arg): + key = base64.b64decode(users[client_id]["cipher"]) + cipher = AES.new(key, AES.MODE_ECB) + data = base64.b64decode(data_arg) + data = cipher.decrypt(data) + return int(str(data, "utf8")) + + +# Função auxiliar para gerar o resultado - já está implementada +# return int value and list of description strings identifying the characteristic of the value +def generate_result(list_values): + if len(list_values) % 2 == 1: + test = 4 + else: + test = 3 + + minimal = min(list_values) + maximal = max(list_values) + first = list_values[0] + last = list_values[-1] + + choice = random.randint(0, test) + if choice == 0: + if minimal == first: + return first, ["min", "first"] + elif maximal == first: + return first, ["max", "first"] + else: + return first, ["first"] + elif choice == 1: + if minimal == last: + return last, ["min", "last"] + elif maximal == last: + return last, ["max", "last"] + else: + return last, ["last"] + elif choice == 2: + if minimal == first: + return first, ["min", "first"] + elif minimal == last: + return last, ["min", "last"] + else: + return minimal, ["min"] + elif choice == 3: + if maximal == first: + return first, ["max", "first"] + elif maximal == last: + return last, ["max", "last"] + else: + return maximal, ["max"] + elif choice == 4: + list_values.sort() + median = list_values[len(list_values) // 2] + if median == first: + return first, ["median", "first"] + elif median == last: + return last, ["median", "last"] + else: + return median, ["median"] + else: + return None + + +# Incomming message structure: +# { op = "START", client_id, [cipher] } +# { op = "QUIT" } +# { op = "NUMBER", number } +# { op = "STOP", [shasum] } +# { op = "GUESS", choice } +# +# Outcomming message structure: +# { op = "START", status } +# { op = "QUIT" , status } +# { op = "NUMBER", status } +# { op = "STOP", status, value } +# { op = "GUESS", status, result } + + +# +# Suporte de descodificação da operação pretendida pelo cliente - já está implementada +# +def new_msg(client_sock): + request = recv_dict(client_sock) + # print( "Command: %s" % (str(request)) ) + + op = request["op"] + if op == "START": + response = new_client(client_sock, request) + elif op == "QUIT": # + response = quit_client(client_sock, request) + elif op == "NUMBER": # + response = number_client(client_sock, request) + elif op == "STOP": # + response = stop_client(client_sock, request) + elif op == "GUESS": # + response = guess_client(client_sock, request) + else: + response = {"op": op, "status": False, "error": "Invalid operation"} + + # print (response) + send_dict(client_sock, response) + + +# +# Suporte da criação de um novo cliente - operação START +# +# detect the client in the request +# verify the appropriate conditions for executing this operation +# process the client in the dictionary +# return response message with or without error message +def new_client(client_sock, request): + client_id = request["client_id"] + + # check if the client_id is in users + if client_id in users: + response = {"op": "START", "status": False, "error": "Client already exists"} + print("Failed to add client %s\nReason: %s" % (client_id, response["error"])) + else: + cipher = None + # verify if client wants to use cipher + if request["cipher"] is not None: + cipher = request["cipher"] + + users[client_id] = {"socket": client_sock, "cipher": cipher, "numbers": [], "has_stopped": False} + response = {"op": "START", "status": True} + print("Client %s added\n" % client_id) + return response + + +# +# Suporte da eliminação de um cliente - já está implementada +# +# obtain the client_id from his socket and delete from the dictionary +def clean_client(client_sock): + client_id = find_client_id(client_sock) + # check if the client_id is in users + if client_id is not None: + print("Client %s removed\n" % client_id) + del users[client_id] + + +# +# Suporte do pedido de desistência de um cliente - operação QUIT +# +# obtain the client_id from his socket +# verify the appropriate conditions for executing this operation +# process the report file with the QUIT result +# eliminate client from dictionary using the function clean_client +# return response message with or without error message +def quit_client(client_sock, request): + client_id = find_client_id(client_sock) + # check if the client_id is in users + if client_id is None: + response = {"op": "QUIT", "status": False, "error": "Client does not exist"} + print("Failed to remove client %s\nReason: %s" % (client_id, response["error"])) + else: + # remove client + clean_client(client_sock) + response = {"op": "QUIT", "status": True} + return response + + +# +# Suporte da criação de um ficheiro csv com o respetivo cabeçalho - já está implementada +# +def create_file(): + with open("result.csv", "w", newline="") as csvfile: + columns = ["client_id", "number_of_numbers", "guess"] + + fw = csv.DictWriter(csvfile, delimiter=",", fieldnames=columns) + fw.writeheader() + + +# +# Suporte da actualização de um ficheiro csv com a informação do cliente +# +# update report csv file with the simulation of the client +def update_file(client_id, size, guess): + with open("result.csv", "a", newline="") as csvfile: + writer = csv.DictWriter(csvfile, delimiter=',', fieldnames=["client_id", "number_of_numbers", "guess"]) + writer.writerow({"client_id": client_id, "number_of_numbers": size, "guess": guess}) + + +# +# Suporte do processamento do número de um cliente - operação NUMBER +# +# obtain the client_id from his socket +# verify the appropriate conditions for executing this operation +# return response message with or without error message +def number_client(client_sock, request): + client_id = find_client_id(client_sock) + # check if the client_id is in users + if client_id is None: + response = {"op": "NUMBER", "status": False, "error": "Client does not exist"} + print("Failed to add number to client %s\nReason: %s" % (client_id, response["error"])) + # check if client has stopped adding numbers + elif users[client_id]["has_stopped"]: + response = {"op": "NUMBER", "status": False, "error": "Client has stopped"} + print("Failed to add number to client %s\nReason: %s" % (client_id, response["error"])) + else: + num = request["number"] + # decrypt the number if a cipher is being used + if users[client_id]["cipher"] is not None: + num = decrypt_intvalue(client_id, num) + + users[client_id]["numbers"].append(num) + response = {"op": "NUMBER", "status": True} + print("Number %d added to client %s\n" % (num, client_id)) + return response + + +# +# Suporte do pedido de terminação de um cliente - operação STOP +# +# obtain the client_id from his socket +# verify the appropriate conditions for executing this operation +# randomly generate a value to return using the function generate_result +# process the report file with the result +# return response message with result or error message +def stop_client(client_sock, request): + client_id = find_client_id(client_sock) + if client_id is None: + response = {"op": "STOP", "status": False, "error": "Client does not exist"} + print("Failed to stop client %s\nReason: %s" % (client_id, response["error"])) + elif len(users[client_id]["numbers"]) < 1: + response = {"op": "STOP", "status": False, "error": "Client has not yet sent any number"} + print("Failed to stop client %s\nReason: %s" % (client_id, response["error"])) + else: + # creates the synthesis for the list + hasher = SHA256.new() + for number in users[client_id]["numbers"]: + hasher.update(bytes(str(number), "utf8")) + + # compares the synthesis of the server and the client to see if they match + if hasher.hexdigest() != request["shasum"]: + response = {"op": "STOP", "status": False, "error": "Server numbers list synthesis doesn't match with client list"} + print("Failed to stop client %s\nReason: %s" % (client_id, response["error"])) + else: + # generates the result + value, solution = generate_result(users[client_id]["numbers"]) + # encrypts the value if a cipher is being used + if users[client_id]["cipher"] is not None: + encripted_value = encrypt_intvalue(client_id, value) + else: + encripted_value = value + + update_file(client_id, len(users[client_id]["numbers"]), solution) + response = {"op": "STOP", "status": True, "value": encripted_value} + users[client_id]["solution"] = solution + users[client_id]["has_stopped"] = True + print("Client %s stopped\nChosen number: %d\nSolution: %s" % (client_id, value, solution)) + return response + + +# +# Suporte da adivinha de um cliente - operação GUESS +# +# obtain the client_id from his socket +# verify the appropriate conditions for executing this operation +# eliminate client from dictionary +# return response message with result or error message +def guess_client(client_sock, request): + client_id = find_client_id(client_sock) + if client_id is None: + response = {"op": "GUESS", "status": False, "error": "Client does not exist"} + print("Failed to guess client %s\nReason: %s" % (client_id, response["error"])) + elif not users[client_id]["has_stopped"]: + response = {"op": "GUESS", "status": False, "error": "Client has not yet stopped"} + print("Failed to guess client %s\nReason: %s" % (client_id, response["error"])) + else: + choice = request["choice"] + response = {"op": "GUESS", "status": True, "result": choice == users[client_id]["solution"]} + print("Client %s guessed %s\n" % (client_id, choice)) + return response + + +def verify_port(port): + # verify if port is a number + if not port.isdigit(): + return {"status": False, "error": "Port must be an integer"} + # verify if port is between 1024 and 65535 + if not (1024 <= int(port) <= 65535): + return {"status": False, "error": "Port number must be between 1024 and 65535"} + return {"status": True, "port": int(port)} + + +def main(): + # validate the number of arguments and eventually print error message and exit with error + # verify type of arguments and eventually print error message and exit with error + if len(sys.argv) != 2: + print("Usage: python3 %s " % sys.argv[0]) + sys.exit(1) + + # obtain the port number + port = sys.argv[1] + verified = verify_port(port) + if not verified["status"]: + print(f"{verified['error']}") + sys.exit(1) + port = verified["port"] + + # create the server socket + try: + hostname = socket.gethostbyname(socket.gethostname()) + server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_socket.bind((hostname, port)) + print(f"Server started.\nHostname: {hostname}\nPort: {port}\n-----------------------------------------------------\n") + # handle errors while creating server socket and exit with error + except OSError: + print(f"Failed to create server socket, maybe the port is already in use?\n") + sys.exit(1) + + server_socket.listen() + + clients = [] + create_file() + + while True: + try: + available = select.select([server_socket] + clients, [], [])[0] + except ValueError: + # Sockets may have been closed, check for that + for client_sock in clients: + if client_sock.fileno() == -1: + clients.remove(client_sock) # closed + continue # Reiterate select + + for client_sock in available: + # New client? + if client_sock is server_socket: + newclient, addr = server_socket.accept() + clients.append(newclient) + # Or an existing client + else: + # See if client sent a message + if len(client_sock.recv(1, socket.MSG_PEEK)) != 0: + # client socket has a message + # print ("server" + str (client_sock)) + new_msg(client_sock) + else: # Or just disconnected + clients.remove(client_sock) + clean_client(client_sock) + client_sock.close() + break # Reiterate select + + +if __name__ == "__main__": + main() diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_client.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_client.py new file mode 100644 index 0000000..d585307 --- /dev/null +++ b/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_client.py @@ -0,0 +1,344 @@ +import socket +from subprocess import Popen +from subprocess import PIPE +import pytest + + +# class for colors in terminal +class Tcolors: + HEADER = '\033[95m' + OKBLUE = '\033[94m' + OKCYAN = '\033[96m' + OKGREEN = '\033[92m' + WARNING = '\033[93m' + FAIL = '\033[91m' + ENDC = '\033[0m' + BOLD = '\033[1m' + UNDERLINE = '\033[4m' + + +def test_arguments(): + # start of testing lack of or too many sys.argv + proc = Popen("python3 client.py", stdout=PIPE, shell=True) + output = proc.stdout.read() + + assert output == f"{Tcolors.WARNING}Usage: python3 client.py client_id port DNS{Tcolors.ENDC}\n".encode("utf-8") + + proc = Popen("python3 client.py test", stdout=PIPE, shell=True) + output = proc.stdout.read().decode("utf-8") + + assert output == f"{Tcolors.WARNING}Usage: python3 client.py client_id port DNS{Tcolors.ENDC}\n" + + proc = Popen("python3 client.py test 2000 123.245.14.25 123", stdout=PIPE, shell=True) + output = proc.stdout.read().decode("utf-8") + + assert output == f"{Tcolors.WARNING}Usage: python3 client.py client_id port DNS{Tcolors.ENDC}\n" + # end of testing lack of or too many sys.argv + + +def test_invalid_port(): + # start of testing invalid port + proc = Popen("python3 client.py test 1000", stdout=PIPE, shell=True) + output = proc.stdout.read().decode("utf-8") + + assert output == f"{Tcolors.WARNING}Port number must be between 1024 and 65535{Tcolors.ENDC}\n" + + proc = Popen("python3 client.py test 1000000", stdout=PIPE, shell=True) + output = proc.stdout.read().decode("utf-8") + + assert output == f"{Tcolors.WARNING}Port number must be between 1024 and 65535{Tcolors.ENDC}\n" + + proc = Popen("python3 client.py test test", stdout=PIPE, shell=True) + output = proc.stdout.read().decode("utf-8") + + assert output == f"{Tcolors.WARNING}Port must be an integer{Tcolors.ENDC}\n" + # end of testing invalid port + + +def test_invalid_ip(): + # start of testing invalid IP address + proc = Popen("python3 client.py test 2000 2154", stdout=PIPE, shell=True) + output = proc.stdout.read().decode("utf-8") + assert output == f"{Tcolors.WARNING}Invalid DNS address{Tcolors.ENDC}\n" + + proc = Popen("python3 client.py test 2000 256.256.256.256", stdout=PIPE, shell=True) + output = proc.stdout.read().decode("utf-8") + assert output == f"{Tcolors.WARNING}Invalid DNS address{Tcolors.ENDC}\n" + + proc = Popen("python3 client.py test 2000 255.255.str.255", stdout=PIPE, shell=True) + output = proc.stdout.read().decode("utf-8") + + assert output == f"{Tcolors.WARNING}Invalid DNS address{Tcolors.ENDC}\n" + # end of testing invalid IP address + + +def test_invalid_connection(): + # start of testing an invalid connection to server + proc = Popen("python3 client.py test 2040", stdout=PIPE, shell=True) + output = proc.stdout.read().decode("utf-8") + hostname = socket.gethostbyname(socket.gethostname()) + port = proc.args.split(" ")[3] + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.FAIL}Error: connection to server failed{Tcolors.ENDC}\n""" + # end of testing an invalid connection to server + + +def test_valid_connection(): + # start of testing a valid connection to server + server = Popen("python3 server.py 2000", stdout=PIPE, shell=True) + client_test = Popen("python3 client.py test 2000", stdout=PIPE, shell=True) + output = client_test.stdout.read().decode("utf-8") + port = client_test.args.split(" ")[3] + hostname = socket.gethostbyname(socket.gethostname()) + + client_test.terminate() + server.terminate() + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC} +{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""" + # end of testing a valid connection to server + + +def test_start_option(): + # start of testing the START option + server = Popen("python3 server.py 3000", stdout=PIPE, shell=True, close_fds=True) + client_test = Popen("python3 client.py test 3000", stdout=PIPE, stdin=PIPE, shell=True) + port = client_test.args.split(" ")[3] + hostname = socket.gethostbyname(socket.gethostname()) + + output = client_test.communicate(input=b"start\ny")[0] + + client_test.terminate() + server.terminate() + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC} +{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE} +{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8") + # end of testing the START option + + +def test_quit_option(): + # start of testing the QUIT option + server = Popen("python3 server.py 4000", stdout=PIPE, shell=True, close_fds=True) + client_test = Popen("python3 client.py test 4000", stdout=PIPE, stdin=PIPE, shell=True) + port = client_test.args.split(" ")[3] + hostname = socket.gethostbyname(socket.gethostname()) + + output = client_test.communicate(input=b"quit")[0] + + client_test.terminate() + server.terminate() + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC} +{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}Quitting... +{Tcolors.OKGREEN}Client quit with success\n""".encode("utf-8") + # end of testing the QUIT option + + +def test_number_option(): + # start of testing the NUMBER option + server = Popen("python3 server.py 5000", stdout=PIPE, shell=True, close_fds=True) + client_test = Popen("python3 client.py test 5000", stdout=PIPE, stdin=PIPE, shell=True) + port = client_test.args.split(" ")[3] + hostname = socket.gethostbyname(socket.gethostname()) + + output = client_test.communicate(input=b"start\ny\nnumber\n200")[0] + + client_test.terminate() + server.terminate() + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC} +{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE} +{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE} +{Tcolors.ENDC}{Tcolors.BOLD}Number > {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}Number added with success{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8") + # end of testing the NUMBER option + + +def test_guess_without_stopping(): + # start of testing the GUESS option without a number + server = Popen("python3 server.py 6000", stdout=PIPE, shell=True, close_fds=True) + client_test = Popen("python3 client.py test 6000", stdout=PIPE, stdin=PIPE, shell=True) + port = client_test.args.split(" ")[3] + hostname = socket.gethostbyname(socket.gethostname()) + + output = client_test.communicate(input=b"start\ny\nguess")[0] + + client_test.terminate() + server.terminate() + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC} +{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE} +{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.WARNING}You can't guess before stopping the game\n{Tcolors.ENDC} +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8") + # end of testing the GUESS option without a number + + +def test_stop_option(): + # start of testing the STOP option + server = Popen("python3 server.py 7000", stdout=PIPE, shell=True, close_fds=True) + client_test = Popen("python3 client.py test 7000", stdout=PIPE, stdin=PIPE, shell=True) + port = client_test.args.split(" ")[3] + hostname = socket.gethostbyname(socket.gethostname()) + + output = client_test.communicate(input=b"start\ny\nnumber\n200\nstop")[0] + + data = 200 + + client_test.terminate() + server.terminate() + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC} +{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE} +{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE} +{Tcolors.ENDC}{Tcolors.BOLD}Number > {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}Number added with success{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nChosen number: {Tcolors.UNDERLINE}{data}{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8") + # end of testing the STOP option + + +def test_guess_without_starting(): + # start of testing the GUESS option without starting the game + server = Popen("python3 server.py 8000", stdout=PIPE, shell=True, close_fds=True) + client_test = Popen("python3 client.py test 8000", stdout=PIPE, stdin=PIPE, shell=True) + port = client_test.args.split(" ")[3] + hostname = socket.gethostbyname(socket.gethostname()) + + output = client_test.communicate(input=b"guess")[0] + + client_test.terminate() + server.terminate() + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC} +{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC} +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8") + # end of testing the GUESS option without starting the game + + +def test_number_without_starting(): + # start of testing the NUMBER option without starting the game + server = Popen("python3 server.py 9000", stdout=PIPE, shell=True, close_fds=True) + client_test = Popen("python3 client.py test 9000", stdout=PIPE, stdin=PIPE, shell=True) + port = client_test.args.split(" ")[3] + hostname = socket.gethostbyname(socket.gethostname()) + + output = client_test.communicate(input=b"number")[0] + + client_test.terminate() + server.terminate() + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC} +{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC} +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8") + # end of testing the NUMBER option without starting the game + + +def test_stop_without_starting(): + # start of testing the STOP option without starting the game + server = Popen("python3 server.py 10000", stdout=PIPE, shell=True, close_fds=True) + client_test = Popen("python3 client.py test 10000", stdout=PIPE, stdin=PIPE, shell=True) + port = client_test.args.split(" ")[3] + hostname = socket.gethostbyname(socket.gethostname()) + + output = client_test.communicate(input=b"stop")[0] + + client_test.terminate() + server.terminate() + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC} +{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.WARNING}You must start the game first\n{Tcolors.ENDC} +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8") + # end of testing the STOP option without starting the game + + +def test_stop_without_numbers(): + # start of testing the STOP option after stopping the game + server = Popen("python3 server.py 11000", stdout=PIPE, shell=True, close_fds=True) + client_test = Popen("python3 client.py test 11000", stdout=PIPE, stdin=PIPE, shell=True) + port = client_test.args.split(" ")[3] + hostname = socket.gethostbyname(socket.gethostname()) + + output = client_test.communicate(input=b"start\ny\nstop")[0] + + client_test.terminate() + server.terminate() + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC} +{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE} +{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.FAIL}Error: Client has not yet sent any number{Tcolors.ENDC} +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8") + # end of testing the STOP option after stopping the game + + +def test_stop_after_stopping(): + # start of testing the STOP option after stopping the game + server = Popen("python3 server.py 12000", stdout=PIPE, shell=True, close_fds=True) + client_test = Popen("python3 client.py test 12000", stdout=PIPE, stdin=PIPE, shell=True) + port = client_test.args.split(" ")[3] + hostname = socket.gethostbyname(socket.gethostname()) + + output = client_test.communicate(input=b"start\ny\nnumber\n123\nstop\nstop")[0] + + client_test.terminate() + server.terminate() + + data = 123 + + assert output == f"""{Tcolors.WARNING}Connecting to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.WARNING}...{Tcolors.ENDC} +{Tcolors.OKGREEN}Connected to {Tcolors.UNDERLINE}{hostname}:{port}{Tcolors.ENDC}{Tcolors.OKGREEN} as client {Tcolors.UNDERLINE}{client_test.args.split(" ")[2]}\n{Tcolors.ENDC} +{Tcolors.OKCYAN}{Tcolors.BOLD}{Tcolors.UNDERLINE}Number characteristics guesser game!{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE} +{Tcolors.ENDC}Do you wish to use a cipher? {Tcolors.BOLD}(Y/N)\n> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nClient added with success{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE} +{Tcolors.ENDC}{Tcolors.BOLD}Number > {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}Number added with success{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.OKGREEN}\nChosen number: {Tcolors.UNDERLINE}{data}{Tcolors.ENDC} + +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}{Tcolors.ENDC}{Tcolors.WARNING}You can't stop the game again\n{Tcolors.ENDC} +{Tcolors.ENDC}\nOperation? (START, QUIT, NUMBER, STOP, GUESS)\n{Tcolors.BOLD}> {Tcolors.UNDERLINE}""".encode("utf-8") diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_server.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_server.py new file mode 100644 index 0000000..48e7fde --- /dev/null +++ b/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_server.py @@ -0,0 +1,39 @@ +import socket +from subprocess import Popen +from subprocess import PIPE +import pytest + + +def test_lack_args(): + # no args + proc = Popen("python3 server.py", stdout=PIPE, shell=True, close_fds=True) + output = proc.communicate()[0] + + assert output == "Usage: python3 server.py \n".encode("utf-8") + + +def test_too_many_args(): + # too many args + proc = Popen("python3 server.py 1234 1234", stdout=PIPE, shell=True, close_fds=True) + output = proc.communicate()[0] + + assert output == "Usage: python3 server.py \n".encode("utf-8") + + +def test_str_port(): + proc = Popen("python3 server.py abc", stdout=PIPE, shell=True, close_fds=True) + output = proc.communicate()[0] + + proc.terminate() + assert output == "Port must be an integer\n".encode("utf-8") + + +def test_port_in_use(): + Popen("python3 server.py 50000", stdout=PIPE, shell=True, close_fds=True) + proc2 = Popen("python3 server.py 50000", stdout=PIPE, shell=True, close_fds=True) + output = proc2.communicate()[0] + + proc2.terminate() + hostname = socket.gethostbyname(socket.gethostname()) + port = 50000 + assert output == f"Failed to create server socket, maybe the port is already in use?\n\n".encode("utf-8") \ No newline at end of file diff --git a/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_units.py b/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_units.py new file mode 100644 index 0000000..685b70b --- /dev/null +++ b/1ano/2semestre/labi/labi2023-ap-g6/client_server/test_units.py @@ -0,0 +1,50 @@ +import random +import socket +import subprocess +import sys +from subprocess import Popen +from subprocess import PIPE +import pytest +import client +import server + + +def test_client_verify_port(): + assert client.verify_port("0") == {'status': False, 'error': 'Port number must be between 1024 and 65535'} + assert client.verify_port("1023") == {'status': False, 'error': 'Port number must be between 1024 and 65535'} + assert client.verify_port("1024") == {'status': True, "port": 1024} + assert client.verify_port("65535") == {'status': True, "port": 65535} + assert client.verify_port("65536") == {'status': False, 'error': 'Port number must be between 1024 and 65535'} + assert client.verify_port("100000") == {'status': False, 'error': 'Port number must be between 1024 and 65535'} + assert client.verify_port("example") == {'status': False, 'error': 'Port must be an integer'} + assert client.verify_port("test") == {'status': False, 'error': 'Port must be an integer'} + + +def test_verify_hostname(): + assert client.verify_hostname("localhost") == {'status': True} + assert client.verify_hostname("123.123.123.123") == {'status': True} + assert client.verify_hostname("example.com") == {'status': False, 'error': 'Invalid DNS address'} + assert client.verify_hostname("test") == {'status': False, 'error': 'Invalid DNS address'} + assert client.verify_hostname("123.123.123") == {'status': False, 'error': 'Invalid DNS address'} + assert client.verify_hostname("413.123.123.123") == {'status': False, 'error': 'Invalid DNS address'} + assert client.verify_hostname("123.413.123.123") == {'status': False, 'error': 'Invalid DNS address'} + assert client.verify_hostname("123.123.413.123") == {'status': False, 'error': 'Invalid DNS address'} + assert client.verify_hostname("123.123.123.413") == {'status': False, 'error': 'Invalid DNS address'} + assert client.verify_hostname("-1.-1.-1.-1") == {'status': False, 'error': 'Invalid DNS address'} + assert client.verify_hostname("256.256.256.256") == {'status': False, 'error': 'Invalid DNS address'} + assert client.verify_hostname("0.0.0.0") == {'status': True} + assert client.verify_hostname("255.255.255.255") == {'status': True} + + +def test_server_verify_port(): + assert server.verify_port("0") == {'status': False, 'error': 'Port number must be between 1024 and 65535'} + assert server.verify_port("1023") == {'status': False, 'error': 'Port number must be between 1024 and 65535'} + assert server.verify_port("1024") == {'status': True, "port": 1024} + assert server.verify_port("65535") == {'status': True, "port": 65535} + assert server.verify_port("65536") == {'status': False, 'error': 'Port number must be between 1024 and 65535'} + assert server.verify_port("100000") == {'status': False, 'error': 'Port number must be between 1024 and 65535'} + assert server.verify_port("example") == {'status': False, 'error': 'Port must be an integer'} + assert server.verify_port("test") == {'status': False, 'error': 'Port must be an integer'} + + + diff --git a/1ano/2semestre/labi/labi2023-ap-g6/readme.txt b/1ano/2semestre/labi/labi2023-ap-g6/readme.txt new file mode 100644 index 0000000..f6db3db --- /dev/null +++ b/1ano/2semestre/labi/labi2023-ap-g6/readme.txt @@ -0,0 +1,2 @@ +Nome: Rúben Gomes, NMec: 113435, Email: rlcg@ua.pt, Trabalho: 50% +Nome: Tiago Garcia, NMec: 114184, Email: tiago.rgarcia@ua.pt, Trabalho: 50%