Search code examples
pythonpython-3.xfastapiendpoint

Endpoint security in FastAPI


I have an endpoint in FastAPI that receives json data and file from client. My main goal is to secure this endpoint since receiving files without validation could be risky. I have to make validation before uploading file and tried dependency injection. This is not a typical user authorization case since the main goal is to validate the data received from already authorized client.

Note: I cannot make the validation in client-side code so do not consider that as an option please.

My current program gives uuid to proper clients and checks the uuid using Depends function. It works for now but i am not certain that if the file is still uploaded before depends or not. This is the code(the json data contains string and dict. these are extracted from the fields below):

async def validate_uuid(uuid: str):

    if uuid == "a valid uuid":
        print("success")
        return uuid
    else:
        print("fail")
        raise HTTPException(status_code=400, detail="UUID invalid")

@app.post("/file")
async def file_upload(
     request: Request, response: Response, file: UploadFile = File(...), check_uuid: str = Depends(validate_uuid), dict_data: dict
):
#do smth

Solution

  • I tried to implement a middleware to my application but it was stuck when i try to run it. After my researches i found out that structure of BaseHTTPMiddleware cause this problem. Only solution is to write your own middleware from scracth. This code below is my approach:

    class DLPMiddleware(BaseHTTPMiddleware):
        def __init__(self, app):
            super().__init__(app)
    
        async def set_body(self, request: Request, call_next):
            valid_uuid_list = # your uuid list wil be defined here
            
            if request.url.path == "/file":
                raw_data = await request.form()
                uuid = raw_data.get("uuid")
                if uuid in valid_uuid_list:
                    response = await call_next(request)
                    return response
                else:
                    return Response(content="Unauthorized UUID", status_code=401)
            else:
                response = await call_next(request)
                return response