97 lines
2.2 KiB
Python
97 lines
2.2 KiB
Python
import asyncio
|
|
from contextlib import asynccontextmanager
|
|
|
|
from authx import AuthXConfig, AuthX, RequestToken
|
|
from fastapi import Response, FastAPI, Request, HTTPException, Depends
|
|
from fastapi.responses import RedirectResponse
|
|
import uvicorn
|
|
|
|
from src.database.db import Database
|
|
from src.database.user import User, UserLogin
|
|
|
|
### Settings
|
|
# TODO: Create .env
|
|
|
|
PORT = 7535
|
|
ADMIN_NAME = 'admin'
|
|
ADMIN_PASSWORD = 'admin'
|
|
DATABASE_USER = 'ADMIN'
|
|
DATABASE_PASS = '123123'
|
|
DATABASE_HOST = '127.0.0.1'
|
|
DATABASE_NAME = 'sonoma-db'
|
|
SECRET = 'SECRET'
|
|
|
|
###
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
await db.connect()
|
|
await db.init()
|
|
yield
|
|
await db.disconnect()
|
|
|
|
db = Database(
|
|
ADMIN_NAME,
|
|
ADMIN_PASSWORD,
|
|
DATABASE_USER,
|
|
DATABASE_PASS,
|
|
DATABASE_HOST,
|
|
DATABASE_NAME
|
|
)
|
|
|
|
app = FastAPI(title='sclient-main-server', lifespan=lifespan)
|
|
|
|
config = AuthXConfig()
|
|
config.JWT_SECRET_KEY = SECRET
|
|
config.JWT_ACCESS_COOKIE_NAME = "sclient_access_token"
|
|
config.JWT_TOKEN_LOCATION = ["cookies"]
|
|
|
|
security = AuthX(config=config)
|
|
security.handle_errors(app)
|
|
|
|
|
|
class App:
|
|
|
|
@app.get('/')
|
|
async def docs(self: Request):
|
|
return RedirectResponse(f'{self.url}docs')
|
|
|
|
### Auth
|
|
|
|
@app.post('/login')
|
|
async def login(self: Request, credentials: UserLogin, response: Response):
|
|
user = await db.get_user(credentials.username)
|
|
|
|
if user is not None:
|
|
if user[1] == credentials.password:
|
|
token = security.create_access_token(uid=credentials.username)
|
|
response.set_cookie(config.JWT_ACCESS_COOKIE_NAME, token)
|
|
return {
|
|
"access_token": token
|
|
}
|
|
raise HTTPException(
|
|
401,
|
|
detail='Incorrect username or password'
|
|
)
|
|
###
|
|
|
|
### Protected
|
|
|
|
@app.get('/protected/auth', dependencies=[Depends(security.access_token_required)])
|
|
async def auth(self: Request):
|
|
try:
|
|
return {"message": "Hello world !"}
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
401,
|
|
detail={"message": str(e)}
|
|
) from e
|
|
|
|
###
|
|
|
|
|
|
def main():
|
|
uvicorn.run(app, host='0.0.0.0', port=PORT)
|
|
|
|
if __name__ == '__main__':
|
|
main() |