python FastAPI如何将Authorization Header添加到下一个请求

tkqqtvp1  于 6个月前  发布在  Python
关注(0)|答案(2)|浏览(102)

我一直在研究一个API的身份验证,我被困在一个简单的东西上(我希望如此)。我有一个login端点:

@router.get("/login")
def login(request: Request):
    return templates.TemplateResponse("auth/login.html", {"request": request})

@router.post("/login")
def login(
    request: Request,
    username: str = Form(...),
    password: str = Form(...),

):
    errors = []
    user_db = pd.read_excel('users.xlsx')
    user = authenticate_user(user_db=user_db, username=username, password=password)

    access_token = create_access_token(data={"sub": username})

    response.set_cookie(
        key="access_token", value=f"Bearer {access_token}", httponly=True
    )
    response.headers['Authorization'] = f"Bearer {access_token}"
    return response

字符串
这可以工作并产生一个令牌,但我有另一个端点需要身份验证,但当我输入时它没有获得“Authorization”头,这使得身份验证失败:

@router.get("/register")
def register(request: Request,
             access_token: Annotated[Union[str, None], Cookie()] = None
             ):
    # print(access_token)
    #print(request.headers['Authorization'])
    return templates.TemplateResponse("auth/register.html", {"request": request})

@router.post("/register")
def register(
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)], # This here fails
    username: str = Form(...),
    password: str = Form(...),
    password_confirm: str = Form(...),

):

    print(current_user)

    try:
        user = UserCreate(username=username, password=password, password_confirm=password_confirm,
                          superuser = False)
        if current_user.superuser:
            message = crear_usuario(user=user)
            return JSONResponse(content={'message': message})
        else:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="User should be a superuser",
                headers={"WWW-Authenticate": "Bearer"},
            )
    except ValidationError as e:
        errors_list = json.loads(e.json())
        for item in errors_list:
            errors.append(item.get("loc")[0] + ": " + item.get("msg"))
        return templates.TemplateResponse(
            "auth/register.html", {"request": request, "errors": errors}
        )


我设法以编程方式发送令牌,将令牌添加到Python中的头中,但我需要这对普通用户起作用。Swagger的UI中的“Authorize”按钮有什么不同,然后让我将Authorization头与我的请求一起发送?

fjnneemd

fjnneemd1#

在您的login端点中,您应该返回的令牌不是headers,而是body,并且您不需要设置cookie,因为您使用Bearer身份验证方案:

@router.post("/login")
def login(
    request: Request,
    username: str = Form(...),
    password: str = Form(...),

):
    errors = []
    user_db = pd.read_excel('users.xlsx')
    user = authenticate_user(user_db=user_db, username=username, password=password)

    access_token = create_access_token(data={"sub": username})

    return {"access_token": access_token, "token_type": "bearer"}

字符串
而且,要让Swagger的绿色Authorize按钮工作,您应该创建OAuth2PasswordBearer类的示例:

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")


然后使用这个oauth2_scheme作为依赖项来获取get_current_user中的令牌:

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    ...

7rfyedvj

7rfyedvj2#

最后,我做到了这一点:

@router.post("/register")
def register(
    request: Request,

    username: str = Form(...),
    password: str = Form(...),
    password_confirm: str = Form(...),
    superuser: bool = Form(False),
    access_token: Annotated[Union[str, None], Cookie()] = None

):

    token = access_token

    _, token = get_authorization_scheme_param(token)
    current_user = get_current_user(token)

    errors = []
    try:
        user = UserCreate(username=username, password=password, password_confirm=password_confirm,
                          superuser=superuser)
        if current_user.superuser:
            message = crear_usuario(user=user)
            return JSONResponse(content={'message': message})
        else:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Solo un superusuario puede crear usuarios",
                headers={"WWW-Authenticate": "Bearer"},
            )
    except ValidationError as e:
        errors_list = json.loads(e.json())
        for item in errors_list:
            errors.append(item.get("loc")[0] + ": " + item.get("msg"))
        return templates.TemplateResponse(
            "auth/register.html", {"request": request, "errors": errors}
        )

字符串
它与Swagger的UI不兼容,但我将不得不支付这笔费用

相关问题