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)