105 lines
2.7 KiB
Python
105 lines
2.7 KiB
Python
from os import getenv, path
|
|
from typing import Annotated, Any
|
|
|
|
from docker import errors, 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.get("/api/container/{container_name}")
|
|
def get_container(
|
|
container_name: str,
|
|
credentials: Annotated[HTTPBasicCredentials, Depends(security)],
|
|
) -> dict[str, Any]:
|
|
try:
|
|
container = client.containers.get(container_name)
|
|
except errors.APIError:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
|
|
|
|
if (
|
|
credentials.username != "admin"
|
|
and f"owner={credentials.username}" not in container.labels
|
|
):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
|
|
|
|
return container.attrs
|
|
|
|
|
|
app.mount(
|
|
"/",
|
|
AuthStaticFiles(
|
|
directory=f"{path.dirname(path.realpath(__file__))}/dist", html=True
|
|
),
|
|
name="static",
|
|
)
|
|
|
|
|
|
def launch():
|
|
run(app, host="0.0.0.0")
|