feat: aggregation server info webui with client profile export, seat management, last activity time, auth, and hermes tokens

tests/test_server.py

@@ -329,6 +329,119 @@ class TestServerComponent(unittest.TestCase):
             if os.path.exists(test_cert_dir):
                 shutil.rmtree(test_cert_dir, ignore_errors=True)
 
+    def test_user_auth_and_session(self):
+        """Tests admin password hashing, verification, session tokens, and password change."""
+        Server.init_db(self.test_db)
+        
+        # Verify default admin credentials
+        self.assertTrue(Server.verify_user_credentials("admin", "admin_logar_2026!", db_path=self.test_db))
+        self.assertFalse(Server.verify_user_credentials("admin", "wrong_password", db_path=self.test_db))
+        self.assertFalse(Server.verify_user_credentials("nonexistent", "password", db_path=self.test_db))
+
+        # Test session token generation & validation
+        token = Server.generate_session_token("admin")
+        self.assertEqual(Server.verify_session_token(token), "admin")
+        self.assertIsNone(Server.verify_session_token("invalid.token.structure"))
+
+        # Test changing password
+        Server.change_user_password("admin", "new_secure_pass123", db_path=self.test_db)
+        self.assertFalse(Server.verify_user_credentials("admin", "admin_logar_2026!", db_path=self.test_db))
+        self.assertTrue(Server.verify_user_credentials("admin", "new_secure_pass123", db_path=self.test_db))
+
+    def test_hermes_tokens_and_auth(self):
+        """Tests Hermes Bearer token creation, verification, revocation, and deletion."""
+        Server.init_db(self.test_db)
+        
+        # Test default Hermes token was created on init
+        tokens = Server.list_hermes_tokens(db_path=self.test_db)
+        self.assertEqual(len(tokens), 1)
+        self.assertEqual(tokens[0]["status"], "active")
+        default_token = tokens[0]["token"]
+        self.assertTrue(Server.verify_hermes_token(default_token, db_path=self.test_db))
+
+        # Test creating new token
+        new_tok = Server.create_hermes_token(name="Hermes Autonomous Agent 01", db_path=self.test_db)
+        self.assertIn("token", new_tok)
+        self.assertEqual(new_tok["name"], "Hermes Autonomous Agent 01")
+        self.assertTrue(Server.verify_hermes_token(new_tok["token"], db_path=self.test_db))
+
+        # Test revoking token
+        self.assertTrue(Server.revoke_hermes_token(new_tok["token"], db_path=self.test_db))
+        self.assertFalse(Server.verify_hermes_token(new_tok["token"], db_path=self.test_db))
+
+        # Test deleting token
+        self.assertTrue(Server.delete_hermes_token(new_tok["token"], db_path=self.test_db))
+        tokens_after = Server.list_hermes_tokens(db_path=self.test_db)
+        self.assertEqual(len(tokens_after), 1)
+
+    def test_client_profile_export(self):
+        """Tests client profile dictionary generation and export formatting."""
+        cfg = Server.load_or_init_config(self.test_config)
+        profile = Server.generate_client_profile_dict(
+            server_host="logar-hub.corp.internal",
+            server_port=9443,
+            client_id="worker-node-101",
+            config_dict=cfg
+        )
+        self.assertEqual(profile["server_host"], "logar-hub.corp.internal")
+        self.assertEqual(profile["server_port"], 9443)
+        self.assertEqual(profile["client_id"], "worker-node-101")
+        self.assertEqual(profile["server_fingerprint"], cfg["server_fingerprint"])
+        self.assertEqual(profile["server_public_key"], cfg["public_key"])
+        self.assertEqual(profile["enrollment_secret"], cfg["enrollment_secret"])
+        self.assertEqual(profile["format"], "LOGAR_CLIENT_PROFILE")
+
+    def test_client_seat_management_and_relative_time(self):
+        """Tests client listing, seat approval, revocation, deletion, and relative timestamp helper."""
+        Server.init_db(self.test_db, enrollment_secret="test-sec-123", max_seats=5)
+        Server.SERVER_STATE["config"] = {"db_path": self.test_db}
+
+        # Populate test clients
+        conn = sqlite3.connect(self.test_db)
+        now = datetime.now(timezone.utc)
+        five_min_ago = (now - timedelta(minutes=5)).strftime("%Y-%m-%d %H:%M:%S")
+        two_hours_ago = (now - timedelta(hours=2)).strftime("%Y-%m-%d %H:%M:%S")
+        
+        conn.execute("INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status, last_seen) VALUES ('c-win', 'WIN-PROD-01', 'windows', 'FP_W1', 'active', ?)", (five_min_ago,))
+        conn.execute("INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status, last_seen) VALUES ('c-lin', 'LIN-PROD-01', 'linux', 'FP_L1', 'active', ?)", (two_hours_ago,))
+        conn.commit()
+        conn.close()
+
+        # 1. Test listing
+        client_data = Server.list_clients()
+        self.assertEqual(client_data["active_seats"], 2)
+        self.assertEqual(client_data["max_seats"], 5)
+        self.assertEqual(len(client_data["clients"]), 2)
+        self.assertEqual(client_data["clients"][0]["last_activity_relative"], "5m ago")
+
+        # 2. Test revoking
+        rev_req = Server.ClientActionRequest(client_id="c-win")
+        rev_res = Server.revoke_client(rev_req)
+        self.assertEqual(rev_res["status"], "revoked")
+        client_data = Server.list_clients()
+        self.assertEqual(client_data["active_seats"], 1)
+
+        # 3. Test approving / reactivating
+        app_res = Server.approve_client(rev_req)
+        self.assertEqual(app_res["status"], "active")
+        client_data = Server.list_clients()
+        self.assertEqual(client_data["active_seats"], 2)
+
+        # 4. Test deleting client record
+        del_res = Server.delete_client(rev_req)
+        self.assertEqual(del_res["status"], "deleted")
+        client_data = Server.list_clients()
+        self.assertEqual(len(client_data["clients"]), 1)
+        self.assertEqual(client_data["active_seats"], 1)
+
+        # 5. Test format_relative_time edge cases
+        self.assertEqual(Server.format_relative_time(None), "Never")
+        self.assertEqual(Server.format_relative_time(""), "Never")
+        self.assertEqual(Server.format_relative_time("invalid-date-string"), "invalid-date-str")
+        just_now = now.strftime("%Y-%m-%d %H:%M:%S")
+        self.assertIn(Server.format_relative_time(just_now), ("0s ago", "1s ago", "Just now"))
+
 
 if __name__ == "__main__":
     unittest.main()
+