from os import getenv, path from typing import Annotated from docker import from_env from dotenv import load_dotenv from fastapi import Depends, FastAPI, HTTPException, Request from fastapi.security import HTTPBasic, HTTPBasicCredentials from fastapi.staticfiles import StaticFiles from starlette import status, types from uvicorn import run load_dotenv() client = from_env() security = HTTPBasic() def http_401() -> HTTPException: return HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Basic"}, ) async def check_auth( credentials: Annotated[HTTPBasicCredentials, Depends(security)], ) -> HTTPBasicCredentials: usernames = getenv("USERNAMES", "").split(",") passwords = getenv("PASSWORDS", "").split(",") if credentials.username not in usernames or credentials.password not in passwords: raise http_401() user_index = usernames.index(credentials.username) password = passwords[user_index] if credentials.password != password: raise http_401() return credentials class AuthStaticFiles(StaticFiles): async def __call__( self, scope: types.Scope, receive: types.Receive, send: types.Send ) -> None: request = Request(scope, receive) credentials = await security(request) if not credentials: raise http_401() await check_auth(credentials) await super().__call__(scope, receive, send) app = FastAPI(dependencies=[Depends(check_auth)]) @app.get("/api/containers") def get_containers( credentials: Annotated[HTTPBasicCredentials, Depends(security)], ) -> list[str]: if credentials.username == "admin": return [container.name for container in client.containers.list()] return [ container.name for container in client.containers.list( filters={"label": f"owner={credentials.username}"} ) ] app.mount( "/", AuthStaticFiles( directory=f"{path.dirname(path.realpath(__file__))}/dist", html=True ), name="static", ) def launch(): run(app, host="0.0.0.0")