Fix Gitea issues #1 and #2, add firewall automation, forwarder telemetry, and dynamic key rotation

tests/test_server.py

@@ -882,6 +882,110 @@ class TestServerComponent(unittest.TestCase):
         # Ingestion cache mode is direct
         self.assertEqual(Server.SERVER_STATE["config"]["cache_mode"], "direct")
 
+    def test_reset_enrollment_secret_and_invalidation(self):
+        """Tests that rotating enrollment secret invalidates the previous secret and updates DB & config."""
+        init_sec = "initial-secret-123456"
+        Server.init_db(self.test_db, enrollment_secret=init_sec, max_seats=5)
+        Server.SERVER_STATE["config"] = {
+            "db_path": self.test_db,
+            "enrollment_secret": init_sec,
+            "config_file_path": self.test_config
+        }
+        with open(self.test_config, "w", encoding="utf-8") as f:
+            json.dump({"enrollment_secret": init_sec, "db_path": self.test_db}, f)
+
+        # 1. Rotate enrollment secret
+        new_sec = Server.rotate_enrollment_secret()
+        self.assertNotEqual(new_sec, init_sec)
+        self.assertEqual(Server.SERVER_STATE["config"]["enrollment_secret"], new_sec)
+
+        # 2. Check DB
+        conn = sqlite3.connect(self.test_db)
+        c = conn.cursor()
+        c.execute("SELECT enrollment_secret FROM license_config WHERE id = 1")
+        db_sec = c.fetchone()[0]
+        conn.close()
+        self.assertEqual(db_sec, new_sec)
+
+        # 3. Check config file
+        with open(self.test_config, "r", encoding="utf-8") as f:
+            cfg_file = json.load(f)
+        self.assertEqual(cfg_file["enrollment_secret"], new_sec)
+
+        # 4. Check export_client_profile automatically generates a new secret
+        resp = Server.export_client_profile(server_host="10.0.0.5", server_port=9443)
+        exported_profile = json.loads(resp.body.decode("utf-8"))
+        self.assertIn("enrollment_secret", exported_profile)
+        self.assertNotEqual(exported_profile["enrollment_secret"], new_sec)
+        self.assertEqual(exported_profile["enrollment_secret"], Server.SERVER_STATE["config"]["enrollment_secret"])
+
+        # 5. Check reset API endpoint
+        reset_res = Server.reset_server_enrollment_secret()
+        self.assertEqual(reset_res["status"], "success")
+        self.assertIn("enrollment_secret", reset_res)
+        self.assertEqual(reset_res["enrollment_secret"], Server.SERVER_STATE["config"]["enrollment_secret"])
+
+    def test_hermes_offline_node_detection_and_metrics(self):
+        """Tests that Hermes reporting detects silent forwarders (>15m) and includes host telemetry metrics."""
+        Server.init_db(self.test_db)
+        Server.SERVER_STATE["config"] = {
+            "db_path": self.test_db,
+            "evaluation_window_hours": 12
+        }
+        conn = sqlite3.connect(self.test_db)
+        now = datetime.now(timezone.utc)
+        thirty_min_ago = (now - timedelta(minutes=30)).strftime("%Y-%m-%d %H:%M:%S")
+        two_min_ago = (now - timedelta(minutes=2)).strftime("%Y-%m-%d %H:%M:%S")
+
+        metrics_node1 = json.dumps({"cpu_percent": 15.5, "memory_percent": 60.0, "disk_percent_free": 45.0})
+        metrics_node2 = json.dumps({"cpu_percent": 88.0, "memory_percent": 95.0, "disk_percent_free": 4.5})
+
+        conn.execute("""
+            INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status, last_seen, metrics_json, cpu_percent, memory_percent, disk_percent_free)
+            VALUES ('node-online', 'srv-online.corp', 'windows', 'FP1', 'active', ?, ?, 15.5, 60.0, 45.0)
+        """, (two_min_ago, metrics_node1))
+
+        conn.execute("""
+            INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status, last_seen, metrics_json, cpu_percent, memory_percent, disk_percent_free)
+            VALUES ('node-dead', 'srv-dead.corp', 'linux', 'FP2', 'active', ?, ?, 88.0, 95.0, 4.5)
+        """, (thirty_min_ago, metrics_node2))
+
+        # Also insert a verified operational error for the online node
+        conn.execute("""
+            INSERT INTO active_issues (fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status, last_run_id)
+            VALUES ('default:srv-online.corp:AppCrash', 'default', 'srv-online.corp', 'AppCrash', 'ERROR', 'Segmentation fault', 'windows', ?, ?, 2, 'VERIFIED', 'run-1')
+        """, (two_min_ago, two_min_ago))
+
+        conn.commit()
+        conn.close()
+
+        report = Server.get_verified_anomalies_for_hermes()
+        self.assertGreaterEqual(len(report), 2)
+
+        # Verify offline node alert
+        offline_alert = next((r for r in report if r.get("status") == "NODE_OFFLINE"), None)
+        self.assertIsNotNone(offline_alert)
+        self.assertEqual(offline_alert["server"], "srv-dead.corp")
+        self.assertEqual(offline_alert["severity"], "CRITICAL")
+        self.assertIn("has gone silent", offline_alert["message"])
+        self.assertEqual(offline_alert["metrics"]["disk_percent_free"], 4.5)
+
+        # Verify online error has attached telemetry
+        online_err = next((r for r in report if r.get("signature") == "AppCrash"), None)
+        self.assertIsNotNone(online_err)
+        self.assertEqual(online_err["metrics"]["cpu_percent"], 15.5)
+
+    def test_routine_service_noise_filtering(self):
+        """Tests that benign routine background service start/stop cycles are filtered from anomaly spam."""
+        self.assertTrue(Server.is_routine_service_noise("sppsvc", "The Software Protection service has entered the stopped state.", "INFO"))
+        self.assertTrue(Server.is_routine_service_noise("edgeupdate", "The Microsoft Edge Update service entered the running state.", "INFO"))
+        self.assertTrue(Server.is_routine_service_noise("Service Control Manager", "The Windows Update service entered the stopped state.", "INFO"))
+        self.assertTrue(Server.is_routine_service_noise("systemd", "Service started successfully.", "INFO"))
+
+        # Errors or warnings must NOT be filtered
+        self.assertFalse(Server.is_routine_service_noise("sppsvc", "Software Protection activation failed with error 0xC004F074", "ERROR"))
+        self.assertFalse(Server.is_routine_service_noise("Database", "Connection pool exhausted", "WARNING"))
+
 
 if __name__ == "__main__":
     unittest.main()