import base64 import binascii import uvicorn from fastapi import FastAPI import casbin from core.config import settings from starlette.middleware.cors import CORSMiddleware from starlette.authentication import AuthenticationBackend, AuthenticationError, SimpleUser, AuthCredentials from starlette.middleware.authentication import AuthenticationMiddleware from db import connect_to_mongo, close_mongo_connection app = FastAPI(title=settings.PROJECT_NAME) if settings.BACKEND_CORS_ORIGINS: app.add_middleware( CORSMiddleware, allow_origins=['*'], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.add_event_handler("startup", connect_to_mongo) app.add_event_handler("shutdown", close_mongo_connection) class BasicAuth(AuthenticationBackend): async def authenticate(self, request): if "Authorization" not in request.headers: return None auth = request.headers["Authorization"] try: scheme, credentials = auth.split() decoded = base64.b64decode(credentials).decode("ascii") except (ValueError, UnicodeDecodeError, binascii.Error): raise AuthenticationError("Invalid basic auth credentials") username, _, password = decoded.partition(":") return AuthCredentials(["authenticated"]), SimpleUser(username) # enforcer = casbin.Enforcer('rbac_model.conf', 'rbac_policy.csv') # app.add_middleware(CasbinMiddleware, enforcer=enforcer) # app.add_middleware(AuthenticationMiddleware, backend=BasicAuth()) from api.api_v1.api import api_router app.include_router(api_router, prefix=settings.API_V1_STR) if __name__ == '__main__': uvicorn.run(app='main:app', host="0.0.0.0", port=8889, reload=True, debug=True)