Add headless mode

Dockerfile

@@ -0,0 +1,40 @@
+FROM almalinux:9-minimal as base
+
+USER root
+
+RUN microdnf install \
+ --assumeyes \
+ --setopt=install_weak_deps=0 \
+ --setopt=tsflags=nodocs \
+ python3.12 \
+ && microdnf clean all
+
+FROM base as build
+
+COPY requirements.txt /requirements.txt
+
+RUN microdnf install \
+ --assumeyes \
+ --setopt=install_weak_deps=0 \
+ --setopt=tsflags=nodocs \
+ python3.12-pip \
+&& microdnf clean all \
+&& python3.12 -m venv /venv \
+&& /venv/bin/python -m ensurepip --upgrade \
+&& /venv/bin/pip install --upgrade setuptools \
+&& /venv/bin/pip install \
+ --disable-pip-version-check \
+ -r /requirements.txt \
+&& rm /venv/bin/pip*
+
+FROM base
+
+WORKDIR /opt/app
+
+COPY . /opt/app/
+COPY --from=build /venv /venv
+
+EXPOSE 80
+
+ENTRYPOINT [ "/venv/bin/python", "nagstamon.py" ]
+

Nagstamon/headless/__init__.py

@@ -0,0 +1,97 @@
+import os
+import secrets
+import sys
+import uvicorn
+from typing import Annotated
+from fastapi import Depends, FastAPI, APIRouter, HTTPException, status
+from fastapi.security import HTTPBasic, HTTPBasicCredentials
+
+
+from Nagstamon.Config import (conf)
+from Nagstamon.Servers import (servers)
+
+
+class HeadlessMode:
+ security = HTTPBasic()
+ ENVVAR_NAME_ADDRESS = 'NAGSTAMON_HEADLESS_ADDRESS'
+ ENVVAR_NAME_PORT = 'NAGSTAMON_HEADLESS_PORT'
+ ENVVAR_NAME_BASICAUTH_USER = 'NAGSTAMON_HEADLESS_BASICAUTH_USER'
+ ENVVAR_NAME_BASICAUTH_PASSWORD = 'NAGSTAMON_HEADLESS_BASICAUTH_PASSWORD'
+
+ address = '0.0.0.0'
+ port = 80
+ basicauth_user = None
+ basicauth_password = None
+
+ """
+ initialize headless rest api server
+ """
+ def __init__(self, argv=None):
+
+ if not (self.ENVVAR_NAME_BASICAUTH_USER in os.environ and self.ENVVAR_NAME_BASICAUTH_PASSWORD in os.environ):
+ print(f"For headless mode the following envvars must be set for basic auth: {self.ENVVAR_NAME_BASICAUTH_USER}, {self.ENVVAR_NAME_BASICAUTH_PASSWORD}")
+ sys.exit(1)
+ else:
+ self.basicauth_user = os.environ[self.ENVVAR_NAME_BASICAUTH_USER]
+ self.basicauth_password = os.environ[self.ENVVAR_NAME_BASICAUTH_PASSWORD]
+
+ self.address = os.getenv(self.ENVVAR_NAME_ADDRESS, self.address)
+ self.port = int(os.getenv(self.ENVVAR_NAME_PORT, self.port))
+
+ self.check_servers()
+
+ self.api = FastAPI()
+
+ self.router = APIRouter()
+ self.router.add_api_route("/hosts", self.get_hosts, methods=["GET"], dependencies=[Depends(self.check_user)])
+ self.api.include_router(self.router)
+
+ uvicorn.run(self.api, host=self.address, port=self.port)
+
+ def check_user(self, credentials: Annotated[HTTPBasicCredentials, Depends(security)]):
+ current_username_bytes = credentials.username.encode("utf8")
+ correct_username_bytes = str.encode(self.basicauth_user, "utf-8")
+ is_correct_username = secrets.compare_digest(
+ current_username_bytes, correct_username_bytes
+ )
+ current_password_bytes = credentials.password.encode("utf8")
+ correct_password_bytes = str.encode(self.basicauth_password, "utf-8")
+ is_correct_password = secrets.compare_digest(
+ current_password_bytes, correct_password_bytes
+ )
+ if not (is_correct_username and is_correct_password):
+ raise HTTPException(
+ status_code=status.HTTP_401_UNAUTHORIZED,
+ detail="Incorrect username or password",
+ headers={"WWW-Authenticate": "Basic"},
+ )
+ return credentials.username
+
+ def check_servers(self):
+ """
+ check if there are any servers configured and enabled
+ """
+ # no server is configured
+ if len(servers) == 0:
+ print('no_server')
+ sys.exit(1)
+ # no server is enabled
+ elif len([x for x in conf.servers.values() if x.enabled is True]) == 0:
+ print('no_server_enabled')
+ sys.exit(1)
+
+ # TODOs
+ # * add caching
+ # * don't authenticate on each request
+ def get_hosts(self):
+ all_hosts = []
+ for server in servers.values():
+ if server.enabled:
+ server.init_config()
+ status = server.GetStatus()
+ all_hosts.append(server.hosts)
+
+ return all_hosts
+
+
+APP = HeadlessMode(sys.argv)