"""Server that will listen for GET and POST requests from the client.""" import time from typing import List from fastapi import FastAPI, File, Form, UploadFile from fastapi.responses import JSONResponse, Response from common import ( ENCRYPTED_OUTPUT_NAME, ENCRYPTED_QUERY_NAME, ENCRYPTED_REFERENCE_NAME, MATCHERS_PATH, SERVER_TMP_PATH, AVAILABLE_MATCHERS, ) from client_server_interface import FHEServer # Load the server objects related to all currently available matchers once and for all FHE_SERVERS = { matcher: FHEServer(MATCHERS_PATH / f"{matcher}/deployment") for matcher in AVAILABLE_MATCHERS } def get_server_file_path(name, user_id, matcher_name): """Get the correct temporary file path for the server. Args: name (str): The desired file name. user_id (int): The current user's ID. matcher_name (str): The matcher chosen by the user Returns: pathlib.Path: The file path. """ return SERVER_TMP_PATH / f"{name}_{matcher_name}_{user_id}" # Initialize an instance of FastAPI app = FastAPI() # Define the default route @app.get("/") def root(): return {"message": "Welcome to Your biometric image matching!"} @app.post("/send_input") def send_input( user_id: str = Form(), matcher: str = Form(), files: List[UploadFile] = File(), ): """Send the inputs to the server.""" # Retrieve the encrypted input image and the evaluation key paths encrypted_query_image_path = get_server_file_path( ENCRYPTED_QUERY_NAME, user_id, matcher ) encrypted_reference_image_path = get_server_file_path( ENCRYPTED_REFERENCE_NAME, user_id, matcher ) evaluation_key_path = get_server_file_path("evaluation_key", user_id, matcher) # Write the files using the above paths with encrypted_query_image_path.open( "wb" ) as encrypted_query_image_file, encrypted_reference_image_path.open( "wb" ) as encrypted_reference_image_file, evaluation_key_path.open( "wb" ) as evaluation_key: encrypted_query_image_file.write(files[0].file.read()) encrypted_reference_image_file.write(files[1].file.read()) evaluation_key.write(files[2].file.read()) @app.post("/run_fhe") def run_fhe( user_id: str = Form(), matcher: str = Form(), ): """Execute the matcher on the encrypted input images using FHE.""" # Retrieve the encrypted input image and the evaluation key paths encrypted_query_image_path = get_server_file_path( "encrypted_query_image", user_id, matcher ) encrypted_reference_image_path = get_server_file_path( "encrypted_reference_image", user_id, matcher ) evaluation_key_path = get_server_file_path("evaluation_key", user_id, matcher) # Read the files using the above paths with encrypted_query_image_path.open( "rb" ) as encrypted_query_image_file, encrypted_reference_image_path.open( "rb" ) as encrypted_reference_image_file, evaluation_key_path.open( "rb" ) as evaluation_key_file: encrypted_query_image = encrypted_query_image_file.read() encrypted_reference_image = encrypted_reference_image_file.read() evaluation_key = evaluation_key_file.read() # Load the FHE server related to the chosen matcher fhe_server = FHE_SERVERS[matcher] # Run the FHE execution start = time.time() encrypted_output = fhe_server.run( encrypted_query_image, encrypted_reference_image, evaluation_key ) fhe_execution_time = round(time.time() - start, 2) # Retrieve the encrypted output path encrypted_output_path = get_server_file_path( ENCRYPTED_OUTPUT_NAME, user_id, matcher ) # Write the file using the above path with encrypted_output_path.open("wb") as encrypted_output: encrypted_output.write(encrypted_output) return JSONResponse(content=fhe_execution_time) @app.post("/get_output") def get_output( user_id: str = Form(), matcher: str = Form(), ): """Retrieve the encrypted output.""" # Retrieve the encrypted output path encrypted_output_path = get_server_file_path( ENCRYPTED_OUTPUT_NAME, user_id, matcher ) # Read the file using the above path with encrypted_output_path.open("rb") as encrypted_output_file: encrypted_output = encrypted_output_file.read() return Response(encrypted_output)