Harden PKI license verification, database seat reconciliation, and binary packaging

tests/test_license_authority.py

@@ -205,6 +205,94 @@ class TestServerLicensingAndWebUI(unittest.TestCase):
         self.assertEqual(info["seats"], 20)
         self.assertEqual(info["license_id"], "LIC-ACT-20")
 
+    def test_signature_verification_and_anti_forgery(self):
+        """Verifies cryptographic signature validation against master root CA and detection of forged certificates."""
+        csr_pem, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir)
+        sub_pem, root_pem = _create_mock_signed_license_cert(csr_pem, seats=30, license_id="LIC-VALID-30")
+
+        # 1. Valid signature against genuine root CA
+        self.assertTrue(enrollment.verify_certificate_signature(sub_pem, root_pem))
+        info_valid = enrollment.extract_license_info_from_cert(sub_pem, upstream_ca_pem=root_pem)
+        self.assertTrue(info_valid["signature_verified"])
+        self.assertEqual(info_valid["seats"], 30)
+
+        # 2. Fake / Rogue Root CA trying to verify the same cert
+        fake_csr, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir)
+        _, fake_root_pem = _create_mock_signed_license_cert(fake_csr, seats=999, license_id="LIC-FAKE")
+        self.assertFalse(enrollment.verify_certificate_signature(sub_pem, fake_root_pem))
+        info_invalid = enrollment.extract_license_info_from_cert(sub_pem, upstream_ca_pem=fake_root_pem)
+        self.assertFalse(info_invalid["signature_verified"])
+
+    def test_database_tampering_seat_reconciliation(self):
+        """Verifies that direct DB manipulation of max_seats is reconciled against the certified X.509 quota."""
+        from license_manager import LicenseManager
+
+        csr_pem, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir)
+        sub_pem, root_pem = _create_mock_signed_license_cert(csr_pem, seats=25, license_id="LIC-SEAT-25")
+
+        # Save certified ca.crt and upstream_ca.crt
+        ca_path = os.path.join(self.server_cert_dir, "ca.crt")
+        with open(ca_path, "w", encoding="utf-8") as f:
+            f.write(sub_pem)
+        with open(os.path.join(self.server_cert_dir, "upstream_ca.crt"), "w", encoding="utf-8") as f:
+            f.write(root_pem)
+
+        Server.init_db(self.server_db, max_seats=25)
+        server_state = {
+            "config": {
+                "db_path": self.server_db,
+                "cert_dir": self.server_cert_dir,
+                "max_seats": 25
+            }
+        }
+
+        # Attacker attempts direct SQLite manipulation: SET max_seats = 99999
+        conn = sqlite3.connect(self.server_db)
+        conn.execute("UPDATE license_config SET max_seats = 99999 WHERE id = 1")
+        conn.commit()
+        conn.close()
+
+        # LicenseManager inspects state and auto-reconciles max_seats to certified 25
+        summary = LicenseManager.get_license_summary(server_state=server_state, db_path=self.server_db)
+        self.assertEqual(summary["max_seats"], 25)
+        self.assertTrue(summary["signature_verified"])
+
+        # Check DB was reconciled
+        conn = sqlite3.connect(self.server_db)
+        row = conn.execute("SELECT max_seats FROM license_config WHERE id = 1").fetchone()
+        conn.close()
+        self.assertEqual(row[0], 25)
+
+    def test_import_license_signature_validation(self):
+        """Verifies that /api/license/import rejects certificates not signed by the upstream authority."""
+        from fastapi import HTTPException
+        csr_pem, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir)
+        sub_pem, root_pem = _create_mock_signed_license_cert(csr_pem, seats=50, license_id="LIC-MOCK-50")
+
+        # Generate mismatched root CA
+        fake_csr, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir)
+        _, rogue_root_pem = _create_mock_signed_license_cert(fake_csr, seats=10, license_id="LIC-ROGUE")
+
+        Server.init_db(self.server_db, max_seats=5)
+        Server.SERVER_STATE["config"] = {
+            "db_path": self.server_db,
+            "cert_dir": self.server_cert_dir,
+            "max_seats": 5
+        }
+
+        # Attempt to import sub_pem with mismatched upstream_ca -> must raise HTTPException 400
+        req_bad = Server.LicenseImportRequest(ca_cert=sub_pem, upstream_ca=rogue_root_pem)
+        with self.assertRaises(HTTPException) as ctx:
+            Server.import_signed_license(req_bad, auth_user="admin")
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertIn("Cryptographic verification failed", ctx.exception.detail)
+
+        # Import sub_pem with valid matching upstream_ca -> must succeed
+        req_good = Server.LicenseImportRequest(ca_cert=sub_pem, upstream_ca=root_pem)
+        res = Server.import_signed_license(req_good, auth_user="admin")
+        self.assertEqual(res["status"], "success")
+        self.assertEqual(res["seats"], 50)
+
     def test_webui_rendering(self):
         """Verifies that the WebUI HTML contains key components, stats, and download triggers."""
         Server.init_db(self.server_db, max_seats=10)
@@ -231,3 +319,4 @@ class TestServerLicensingAndWebUI(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+