sec: enforce admin authorization across all management endpoints and randomize session secret

src/Server.py

@@ -43,7 +43,7 @@ EVALUATION_WINDOW_HOURS = 12
 RUN_THRESHOLD = 1
 
 SESSION_COOKIE_NAME = "logar_server_session"
-SESSION_SECRET = os.environ.get("LOGAR_SESSION_SECRET", "logar_server_hub_session_secret_2026_!#*")
+SESSION_SECRET = os.environ.get("LOGAR_SESSION_SECRET") or secrets.token_hex(32)
 DEFAULT_ADMIN_USER = os.environ.get("LOGAR_ADMIN_USER", "admin")
 DEFAULT_ADMIN_PASS = os.environ.get("LOGAR_ADMIN_PASS", "admin_logar_2026!")
 
@@ -152,11 +152,26 @@ def get_current_user_optional(request: Request) -> Optional[str]:
 
 def require_admin_auth(request: Request) -> str:
     """Enforces administrator authentication for management endpoints."""
+    # 1. Check if logged in admin user via session cookie or Bearer session token
     user = get_current_user_optional(request)
-    if not user:
-        raise HTTPException(status_code=401, detail="Authentication required. Please login.")
+    if user:
         return user
 
+    # 2. Check Bearer token or token query param matching config auth_token
+    token = None
+    auth_header = request.headers.get("Authorization")
+    if auth_header and auth_header.startswith("Bearer "):
+        token = auth_header.split(" ", 1)[1].strip()
+    if not token:
+        token = request.query_params.get("token")
+
+    if token:
+        cfg_token = SERVER_STATE.get("config", {}).get("auth_token")
+        if cfg_token and secrets.compare_digest(str(token), str(cfg_token)):
+            return "config_auth_token"
+
+    raise HTTPException(status_code=401, detail="Authentication required. Please login or provide admin token.")
+
 
 def verify_user_credentials(username: str, password: str, db_path: str = DEFAULT_DB_FILE) -> bool:
     """Checks user credentials against SQLite users table."""
@@ -332,6 +347,11 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
             config["license_server_url"] = "https://logar.eibl.tech"
         if "admin_user" not in config:
             config["admin_user"] = DEFAULT_ADMIN_USER
+        if "session_secret" not in config:
+            config["session_secret"] = secrets.token_hex(32)
+        global SESSION_SECRET
+        if not os.environ.get("LOGAR_SESSION_SECRET"):
+            SESSION_SECRET = config["session_secret"]
         return config
 
     print(f"[!] Config '{config_path}' not found. Initializing first-run configuration...")
@@ -339,6 +359,7 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
     private_key, public_key, fingerprint = generate_server_keypair(server_name)
     auth_token = secrets.token_hex(24)
     enrollment_secret = secrets.token_hex(24)
+    session_sec = secrets.token_hex(32)
 
     config = {
         "server_name": server_name,
@@ -349,6 +370,7 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
         "admin_user": DEFAULT_ADMIN_USER,
         "auth_token": auth_token,
         "enrollment_secret": enrollment_secret,
+        "session_secret": session_sec,
         "max_seats": 10,
         "license_server_url": "https://logar.eibl.tech",
         "cert_dir": "certs",
@@ -360,6 +382,8 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
         "public_key": public_key,
         "private_key": private_key
     }
+    if not os.environ.get("LOGAR_SESSION_SECRET"):
+        SESSION_SECRET = session_sec
 
     with open(config_path, "w", encoding="utf-8") as f:
         json.dump(config, f, indent=2)
@@ -1181,8 +1205,8 @@ def enroll_client(req: ClientEnrollRequest):
     }
 
 
-@app.get("/api/client/export-profile")
-@app.post("/api/client/export-profile")
+@app.get("/api/client/export-profile", dependencies=[Depends(require_admin_auth)])
+@app.post("/api/client/export-profile", dependencies=[Depends(require_admin_auth)])
 def export_client_profile(
     server_host: Optional[str] = None,
     server_port: Optional[int] = None,
@@ -1213,7 +1237,7 @@ def export_client_profile(
     )
 
 
-@app.get("/api/clients")
+@app.get("/api/clients", dependencies=[Depends(require_admin_auth)])
 def list_clients():
     """Returns all registered clients, seat usage, and activity timestamps."""
     db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
@@ -1248,8 +1272,8 @@ def list_clients():
     }
 
 
-@app.post("/api/clients/approve")
-@app.post("/api/clients/reactivate")
+@app.post("/api/clients/approve", dependencies=[Depends(require_admin_auth)])
+@app.post("/api/clients/reactivate", dependencies=[Depends(require_admin_auth)])
 def approve_client(action: ClientActionRequest):
     """Approves / reactivates a client certificate if within seat quota."""
     db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
@@ -1281,7 +1305,7 @@ reactivate_client = approve_client
 
 
 
-@app.post("/api/clients/revoke")
+@app.post("/api/clients/revoke", dependencies=[Depends(require_admin_auth)])
 def revoke_client(action: ClientActionRequest):
     """Revokes a client certificate/license seat. Revoked clients are immediately rejected during mTLS handshakes."""
     db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
@@ -1298,8 +1322,8 @@ def revoke_client(action: ClientActionRequest):
     return {"status": "revoked", "client_id": action.client_id}
 
 
-@app.post("/api/clients/delete")
-@app.post("/api/clients/remove")
+@app.post("/api/clients/delete", dependencies=[Depends(require_admin_auth)])
+@app.post("/api/clients/remove", dependencies=[Depends(require_admin_auth)])
 def delete_client(action: ClientActionRequest):
     """Permanently removes a client forwarder record from SQLite, immediately freeing its seat."""
     db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
@@ -1503,7 +1527,7 @@ def api_license_status():
     return get_license_summary()
 
 
-@app.get("/api/license/export-lic")
+@app.get("/api/license/export-lic", dependencies=[Depends(require_admin_auth)])
 def export_lic_file():
     """Generates and downloads the logar_server.lic file containing CA CSR and authority coordinates."""
     config = SERVER_STATE.get("config", {})
@@ -1515,7 +1539,7 @@ def export_lic_file():
     )
 
 
-@app.get("/api/license/csr")
+@app.get("/api/license/csr", dependencies=[Depends(require_admin_auth)])
 def get_ca_csr():
     """Returns raw PEM Certificate Signing Request (CSR) for subordinate CA."""
     config = SERVER_STATE.get("config", {})
@@ -1529,7 +1553,7 @@ def get_ca_csr():
     return {"csr": csr_pem}
 
 
-@app.post("/api/license/import")
+@app.post("/api/license/import", dependencies=[Depends(require_admin_auth)])
 async def import_signed_license(req: Request):
     """Imports and activates a signed CA certificate or .lic file."""
     try:
@@ -1596,8 +1620,8 @@ async def import_signed_license(req: Request):
     }
 
 
-@app.post("/api/license/renew")
-@app.post("/api/license/sync")
+@app.post("/api/license/renew", dependencies=[Depends(require_admin_auth)])
+@app.post("/api/license/sync", dependencies=[Depends(require_admin_auth)])
 async def trigger_license_renewal(request: Request):
     """Explicitly triggers automated mTLS sync and renewal with logar.eibl.tech."""
     requested_seats = None

tests/test_pipeline.py

@@ -109,7 +109,9 @@ def run_tests():
     print(f"[OK] Client enrolled successfully. Certificates stored in {cert_dir}")
 
     # 2c. Verify client shows in /api/clients
-    clients_req = urllib.request.Request(f"http://{HERMES_HOST}:{HERMES_PORT}/api/clients")
+    auth_tok = server_conf.get("auth_token") or client_conf.get("auth_token")
+    client_headers = {"Authorization": f"Bearer {auth_tok}"} if auth_tok else {}
+    clients_req = urllib.request.Request(f"http://{HERMES_HOST}:{HERMES_PORT}/api/clients", headers=client_headers)
     with urllib.request.urlopen(clients_req, timeout=5) as resp:
         clients_data = json.loads(resp.read().decode("utf-8"))
         assert clients_data["active_seats"] >= 1