python - Fastapi 的 ASGI 应用程序中的异常 - 错误

Question

下面是我的 fastapi 身份验证代码。已经面临这个问题几个小时了。

class AuthError(Exception):
    def __init__(self, error, status_code):
        self.error = error
        self.status_code = status_code



app = FastAPI()


@app.middleware("http")
async def authenticate_request(request: Request, call_next):

        requires_auth(request)
        await call_next(request)


def get_token_auth_header(request):
        """Obtains the Access Token from the Authorization Header
        """


        auth = request.headers.get("Authorization", None)
        print(auth)
        if not auth:
            raise AuthError({"code": "authorization_header_missing",
                             "description":
                                 "Authorization header is expected"}, 401)

        parts = auth.split()
        print(parts)
        if parts[0].lower() != "bearer":
            raise AuthError({"code": "invalid_header",
                             "description":
                                 "Authorization header must start with"
                                 " Bearer"}, 401)
        elif len(parts) == 1:
            raise AuthError({"code": "invalid_header",
                             "description": "Token not found"}, 401)
        elif len(parts) > 2:
            raise AuthError({"code": "invalid_header",
                             "description":
                                 "Authorization header must be"
                                 " Bearer token"}, 401)

        token = parts[1]

def requires_auth(request):
    
    token = get_token_auth_header(request)
    print(token)
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    jsonurl = urlopen(url="https://" + AUTH0_DOMAIN + "/.well-known/jwks.json", context= ctx)
    jwks = json.loads(jsonurl.read())
   
    unverified_header = jwt.get_unverified_header(token)
    
    rsa_key = {}
    for key in jwks["keys"]:
        if key["kid"] == unverified_header["kid"]:
            rsa_key = {
                "kty": key["kty"],
                "kid": key["kid"],
                "use": key["use"],
                "n": key["n"],
                "e": key["e"]
            }
        if rsa_key:
          
            try:
                payload = jwt.decode(
                    token,
                    rsa_key,
                    algorithms=ALGORITHMS,
                    audience=API_AUDIENCE,
                    issuer="https://" + AUTH0_DOMAIN + "/"
                )

            except jwt.ExpiredSignatureError:
                raise AuthError({"code": "token_expired",
                                "description": "token is expired"}, 401)
            except jwt.PyJWTError:
                raise AuthError({"code": "invalid_claims",
                                "description":
                                    "incorrect claims,"
                                    "please check the audience and issuer"}, 401)
            except Exception:
                raise AuthError({"code": "invalid_header",
                                "description":
                                    "Unable to parse authentication"
                                    " token."}, 401)
            _request_ctx_stack.top.current_user = payload
          
        raise AuthError({"code": "invalid_header",
                        "description": "Unable to find appropriate key"}, 401)


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=80)

为 fastapi 运行此代码时出现以下错误

信息：127.0.0.1:56994 -“GET /user HTTP/1.1”500 内部服务器错误错误：ASGI 应用程序中的异常

有人遇到过这个问题吗？任何帮助将不胜感激。

标签： pythonpython-3.xfastapiuvicorn