File size: 4,446 Bytes
21c7197 7b32412 0908a41 7b32412 21c7197 0908a41 5a6dcda 0908a41 5a6dcda 21c7197 a1cf584 0908a41 21c7197 0908a41 21c7197 0908a41 21c7197 a1cf584 21c7197 a1cf584 21c7197 0908a41 21c7197 3cf0931 0908a41 a1cf584 3cf0931 0908a41 21c7197 0908a41 21c7197 0908a41 21c7197 0908a41 3cf0931 0908a41 21c7197 3cf0931 0908a41 21c7197 0908a41 21c7197 0908a41 21c7197 0908a41 21c7197 0908a41 21c7197 3cf0931 21c7197 0908a41 21c7197 0908a41 21c7197 0908a41 21c7197 3cf0931 21c7197 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 |
"""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)
|