zmievsa
commented
1 day ago Thanks for reporting this! Can I ask you to provide a code snippet that allows me to reproduce the bug?
Open swoKorbi opened 1 day ago
zmievsa
commented
1 day ago Thanks for reporting this! Can I ask you to provide a code snippet that allows me to reproduce the bug?
swoKorbi
commented
1 day ago Thanks for the quick answer:
Here is a minimal working example:
import cadwyn
import fastapi
import pydantic
import fastapi.security
import datetime
import jwt
import typing
import uvicorn
class JWTToken(pydantic.BaseModel):
access_token: str
token_type: str
class User(pydantic.BaseModel):
name: str
versions = cadwyn.VersionBundle(
cadwyn.HeadVersion(),
cadwyn.Version("2024-11-20"),
)
oauth_router = cadwyn.VersionedAPIRouter(prefix="")
oauth2_scheme = fastapi.security.OAuth2PasswordBearer(tokenUrl="/auth/login")
@oauth_router.post("/login")
async def login(
form_data: fastapi.security.OAuth2PasswordRequestForm = fastapi.Depends(),
) -> JWTToken:
user = form_data.username == 'username' and form_data.password == 'password'
if not user:
raise fastapi.HTTPException(
status_code=400, detail="Invalid authentication credentials"
)
timezone = datetime.timezone(datetime.timedelta(hours=2))
expire = datetime.datetime.now(tz=timezone) + datetime.timedelta(
minutes=60
)
data_to_encode = {"sub": form_data.username}
data_to_encode.update({"exp": expire})
access_token = jwt.encode(data_to_encode, 'SOME_SECRET_KEY', algorithm='HS256')
return JWTToken(access_token=access_token, token_type="bearer")
async def get_current_user(token: str) -> typing.Optional[User]:
payload = jwt.decode(token, 'SOME_SECRET_KEY', algorithms=['HS256'])
username: str = payload.get("sub")
return User(name=username)
async def get_current_active_user(token: str = fastapi.Depends(oauth2_scheme)) -> User:
user = await get_current_user(token)
if user is None:
raise fastapi.HTTPException(
status_code=400, detail="Invalid authentication credentials"
)
return user
@oauth_router.get("/verify")
def verify(_=fastapi.Depends(get_current_active_user)) -> bool:
return True
app = cadwyn.Cadwyn(
versions=versions,
)
app.generate_and_include_versioned_routers(oauth_router)
if __name__ == '__main__':
uvicorn.run(app, host="0.0.0.0", port=8003)
Nice example! I'll try to handle this today!
zmievsa
commented
21 hours ago FastAPI generates an openapi.json that is used for rendering the swagger docs. However, it doesn't seem like swagger supports sending custom headers such as X-API-VERSION to authentication urls. https://swagger.io/docs/specification/v3_0/authentication/
still digging
swoKorbi
commented
8 hours ago I was also playing around with openapi, but couldn't find anything that would allow me to add a custom header to the OAuth URL.
The problem definitely is in the definition of the OAuth Sheme Python oauth2_scheme = fastapi.security.OAuth2PasswordBearer(tokenUrl="/login") as no version for this tokenUrl can be passed with it, and therefore, the request generated in Swagger through the Authorize button can not find the endpoint as no verison is defined.
I was able to get a dirty workaround by defining a login endpoint outside the versioned router and injecting the latest API version, but that is not really a satisfiying option.
import cadwyn
import fastapi
import pydantic
import fastapi.security
import datetime
import jwt
import typing
import uvicorn
import requests
class JWTToken(pydantic.BaseModel):
access_token: str
token_type: str
class User(pydantic.BaseModel):
name: str
versions = cadwyn.VersionBundle(
cadwyn.HeadVersion(),
cadwyn.Version("2024-11-20"),
)
oauth_router = cadwyn.VersionedAPIRouter(prefix="")
oauth2_scheme = fastapi.security.OAuth2PasswordBearer(tokenUrl="/login")
@oauth_router.post("/authorize")
async def authorize(
form_data: fastapi.security.OAuth2PasswordRequestForm = fastapi.Depends(),
) -> JWTToken:
user = form_data.username == "username" and form_data.password == "password"
if not user:
raise fastapi.HTTPException(
status_code=400, detail="Invalid authentication credentials"
)
timezone = datetime.timezone(datetime.timedelta(hours=2))
expire = datetime.datetime.now(tz=timezone) + datetime.timedelta(minutes=60)
data_to_encode = {"sub": form_data.username}
data_to_encode.update({"exp": expire})
access_token = jwt.encode(data_to_encode, "SOME_SECRET_KEY", algorithm="HS256")
return JWTToken(access_token=access_token, token_type="bearer")
async def get_current_user(token: str) -> typing.Optional[User]:
payload = jwt.decode(token, "SOME_SECRET_KEY", algorithms=["HS256"])
username: str = payload.get("sub")
return User(name=username)
async def get_current_active_user(token: str = fastapi.Depends(oauth2_scheme)) -> User:
user = await get_current_user(token)
if user is None:
raise fastapi.HTTPException(
status_code=400, detail="Invalid authentication credentials"
)
return user
@oauth_router.get("/verify")
def verify(_=fastapi.Depends(get_current_active_user)) -> bool:
return True
app = cadwyn.Cadwyn(
versions=versions,
)
app.generate_and_include_versioned_routers(oauth_router)
@app.post("/login")
def login(
form_data: fastapi.security.OAuth2PasswordRequestForm = fastapi.Depends(),
) -> JWTToken:
url = 'http://localhost:8003/authorize'
headers = {
'accept': 'application/json',
'x-api-version': '2024-11-20',
'Content-Type': 'application/x-www-form-urlencoded'
}
data = {
'grant_type': 'password',
'username': form_data.username,
'password': form_data.password,
'scope': '',
'client_id': 'string',
'client_secret': 'string'
}
response = requests.post(url, headers=headers, data=data).json()
return JWTToken(**response)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8003)
I can successfully authorize through swagger with this example. But if any changes are made to the login endpoint it will not be versioned.
zmievsa
commented
4 hours ago There is another workaround:
openapi: 3.1.0
info:
title: test
version: 1.0.0
servers:
- url: https://httpbin.org
components:
securitySchemes:
ApiVersionHeader:
type: apiKey
in: header
name: X-API-VERSION
description: Specify the API version, e.g. `2024-01-31`
OAuth2:
type: oauth2
flows:
implicit:
authorizationUrl: https://api.example.com/oauth2/authorize
scopes: {}
security:
- OAuth2: []
ApiVersionHeader: []
paths:
/anything:
get:
responses: {}Essentially using API version header as an API key hence making us send it to the auth endpoint. It's a hack though and implementing it in Cadwyn would not be super easy.
zmievsa
commented
1 minute ago @swoKorbi your example actually looks quite nice! I suggest using a redirectresponse instead. I think, it will be even more stable and easy to use.
But if any changes are made to the login endpoint it will not be versioned. No worries. Just add a unit test (or a few) that make sure these endpoints are compatible and you'll be all set. I really like your approach and will probably add it to the docs as the accepted solution.
Describe the bug I do have OAuth2 authentication implemented in my API. After migrating to cadwyn versioning, the authorization through the Authorize button at the top of the swagger UI does not work anymore. the API response I get is 404 NotFound on the /auth/login endpoint. If I use the endpoint manually from the Swagger docs, I can sucessfully authorize against the endpoint and get back my JWT token.
I suspect, the endpoint doesn't get the proper API version header to use the correct authentication endpoint
To Reproduce Steps to reproduce the behavior:
Tried to login via Swagger Authorize button. I can log those header end get following API response:
Manual login via endpoint from swagger UI:
Expected behavior The Authroize button in the swagger UI should send the appropriate x-api-version header to use OAuth2 authentication