compliance monitor: add healthz endpoint (#912)

Signed-off-by: Anja Strunk <[email protected]>
Signed-off-by: Matthias Büchse <[email protected]>
Co-authored-by: Matthias Büchse <[email protected]>

Author: anjastrunk

compliance-monitor/docker-compose.yml

@@ -28,6 +28,9 @@ services:
       - SCM_DB_HOST=postgres
       - SCM_DB_PORT=5432
       - SCM_DB_PASSWORD_FILE=/run/secrets/db_password
+      # pass the following two from the shell
+      - SCM_HC_USER
+      - SCM_HC_PASSWORD
       - SCM_BASE_URL=https://compliance.sovereignit.cloud/
     volumes:
       - ../Tests:/Tests

compliance-monitor/monitor.py

@@ -64,6 +64,9 @@ def __init__(self):
         self.db_host = os.getenv("SCM_DB_HOST", "localhost")
         self.db_port = os.getenv("SCM_DB_PORT", 5432)
         self.db_user = os.getenv("SCM_DB_USER", "postgres")
+        # use default value of None for security reasons (won't be matched)
+        self.hc_user = os.getenv("SCM_HC_USER", None)
+        self.hc_password = os.getenv("SCM_HC_PASSWORD", None)
         password_file_path = os.getenv("SCM_DB_PASSWORD_FILE", None)
         if password_file_path:
             with open(os.path.abspath(password_file_path), "r") as fileobj:
@@ -123,6 +126,7 @@ class ViewType(Enum):
 # do I hate these globals, but I don't see another way with these frameworks
 app = FastAPI()
 security = HTTPBasic(realm="Compliance monitor", auto_error=True)  # use False for optional login
+optional_security = HTTPBasic(realm="Compliance monitor", auto_error=False)
 settings = Settings()
 # see https://passlib.readthedocs.io/en/stable/narr/quickstart.html
 cryptctx = CryptContext(
@@ -719,6 +723,22 @@ async def post_results(
     conn.commit()
 
 
+@app.get("/healthz")
+async def get_healthz(request: Request):
+    """return compliance monitor's health status"""
+    credentials = await optional_security(request)
+    authorized = credentials and \
+        credentials.username == settings.hc_user and credentials.password == settings.hc_password
+
+    try:
+        mk_conn(settings=settings)
+    except Exception as e:
+        detail = str(e) if authorized else 'internal server error'
+        return Response(status_code=500, content=detail, media_type='text/plain')
+
+    return Response()  # empty response with status 200
+
+
 def pick_filter(results, subject, scope):
     """Jinja filter to pick scope results from `results` for given `subject` and `scope`"""
     return results.get(subject, {}).get(scope, {})