55 lines
2.0 KiB
Python
55 lines
2.0 KiB
Python
from urllib.request import Request
|
|
|
|
import httpx
|
|
from fastapi import HTTPException, Depends, Security
|
|
from fastapi.security import SecurityScopes
|
|
from httpx import ConnectError
|
|
|
|
from .config import AUTH_SERVER_URL
|
|
from .model import UserModel, ValidateTokenInput
|
|
from .schema import oauth2
|
|
|
|
|
|
async def validate_token(_input: ValidateTokenInput):
|
|
try:
|
|
client = httpx.AsyncClient()
|
|
headers = {'Content-Type': 'application/json', }
|
|
result = await client.request('POST', AUTH_SERVER_URL, json=_input.dict(), headers=headers)
|
|
export = result.text
|
|
if 'application/json' in result.headers['content-type']:
|
|
export = result.json()
|
|
if result.status_code > 299:
|
|
raise HTTPException(status_code=result.status_code, detail=export)
|
|
return export
|
|
except ConnectError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
except HTTPException as e:
|
|
raise e
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
async def get_current_user(security_scopes: SecurityScopes, token: str = Depends(oauth2)):
|
|
if security_scopes.scopes:
|
|
authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
|
|
else:
|
|
authenticate_value = f"Bearer"
|
|
credentials_exception = HTTPException(
|
|
status_code=401,
|
|
detail="Could not validate credentials a",
|
|
headers={"WWW-Authenticate": authenticate_value},
|
|
)
|
|
user = await validate_token(ValidateTokenInput(token=token, scopes=security_scopes.scopes))
|
|
if user is None:
|
|
raise credentials_exception
|
|
user.scopes = security_scopes.scopes if isinstance(
|
|
security_scopes.scopes, list) else security_scopes.scopes
|
|
return user
|
|
|
|
|
|
async def get_current_active_user(request: Request, current_user: UserModel = Security(get_current_user, scopes=["account"]),):
|
|
if current_user.disabled:
|
|
raise HTTPException(status_code=400, detail="Inactive user")
|
|
request.state.user = UserModel(**current_user.dict())
|
|
return current_user
|