86 lines
2.1 KiB
Python
86 lines
2.1 KiB
Python
|
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")
|