How to integrate oauth2 with fastapi?

Question

I am trying to integrate oauth2 with fastapi running with mock oidc-server authentication. I went through the documentation but not able to make out what fits where. This is a snippet from two files - main.py

from authlib.integrations.starlette_client import OAuth
oauth = OAuth()

CONF_URL = "https://localhost:8080/.well-known/openid-configuration"
oauth.register(
    name="cad",
    server_metadata_url=CONF_URL,
    client_id=settings.CLIENT_ID,
    client_secret=settings.CLIENT_SECRET,
    client_kwargs={"scope": "openid email profile authorization_group"},
)


@app.middleware("http")
async def authorize(request: Request, call_next):
    if not (request.scope["path"].startswith("/login") or request.scope["path"].startswith("/auth")):
        if not is_session_okay(request.session):
            return RedirectResponse(url="/login")
    return await call_next(request)


@app.get("/login")
async def login(request: Request):
    redirect_uri = request.url_for("auth")
    return await oauth.cad.authorize_redirect(request, redirect_uri)


@app.get("/auth")
async def auth(request: Request):
    try:
        token = await oauth.cad.authorize_access_token(request)
    except OAuthError as error:
        return HTMLResponse(f"{error.error}")
    user = await oauth.cad.parse_id_token(request, token)
    request.session["user"] = dict(user)
    request.session["session_expiry"] = str(datetime.datetime.utcnow() + 
                                        datetime.timedelta(hours=48))
    return {"access_token": create_token(user['sub'), "token_type": "bearer"}

& jwt.py

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/auth', auto_error=False)

# Error
CREDENTIALS_EXCEPTION = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail='Could not validate credentials',
    headers={'WWW-Authenticate': 'Bearer'},
)


# Create token internal function
def create_access_token(*, data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({'exp': expire})
    encoded_jwt = jwt.encode(to_encode, API_SECRET_KEY, algorithm=API_ALGORITHM)
    return encoded_jwt


def create_refresh_token(email):
    expires = timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES)
    return create_access_token(data={'sub': email}, expires_delta=expires)


def create_token(id):
    access_token_expires = timedelta(minutes=API_ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(data={'sub': id}, expires_delta=access_token_expires)
    return access_token

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = decode_token(token)
        id: str = payload.get('sub')
        if email is None:
            raise CREDENTIALS_EXCEPTION
    except jwt.JWTError:
        raise CREDENTIALS_EXCEPTION

    raise id

I have tried integrating create_token in the "auth" endpoint and adding Depends(get_current_user) parameter in get api . I get the authorize button in swagger, but authorization doesn't happen with client id & client secret, nor with user-name & passwd.

How to integrate oauth2 with fastapi?

0 Answers