diff --git a/tests/test_license_authority.py b/tests/test_license_authority.py new file mode 100644 index 0000000..746ae3e --- /dev/null +++ b/tests/test_license_authority.py @@ -0,0 +1,229 @@ +import os +import sys +import json +import shutil +import sqlite3 +import unittest +from datetime import datetime, timezone, timedelta + +# Ensure src/ is importable +ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +SRC_DIR = os.path.join(ROOT_DIR, "src") +if ROOT_DIR not in sys.path: + sys.path.insert(0, ROOT_DIR) +if SRC_DIR not in sys.path: + sys.path.insert(0, SRC_DIR) + +import server_enrollment as enrollment +import License_Authority as authority +import Server + + +class TestLicenseAuthorityAndWebUI(unittest.TestCase): + def setUp(self): + self.test_dir = os.path.join(ROOT_DIR, "test_lic_tmp") + os.makedirs(self.test_dir, exist_ok=True) + self.master_ca_dir = os.path.join(self.test_dir, "master_ca") + self.server_cert_dir = os.path.join(self.test_dir, "server_certs") + self.db_path = os.path.join(self.test_dir, "license_auth.db") + self.server_db = os.path.join(self.test_dir, "server_state.db") + + def tearDown(self): + if os.path.exists(self.test_dir): + shutil.rmtree(self.test_dir, ignore_errors=True) + + def test_master_root_ca_generation(self): + """Verifies Master Root CA creation and structure.""" + cert_pem, key_pem = authority.generate_master_root_ca(cert_dir=self.master_ca_dir) + self.assertIn("BEGIN CERTIFICATE", cert_pem) + self.assertIn("BEGIN RSA PRIVATE KEY", key_pem) + self.assertTrue(os.path.exists(os.path.join(self.master_ca_dir, "master_root_ca.crt"))) + + def test_ca_csr_and_lic_request_generation(self): + """Verifies subordinate CA CSR and .lic request file generation.""" + lic_data, lic_json = enrollment.create_license_request_lic( + cert_dir=self.server_cert_dir, + license_server="https://logar.eibl.tech", + server_fingerprint="AABBCCDDEEFF00112233", + hostname="test-hub.corp.internal", + requested_seats=30 + ) + self.assertEqual(lic_data["format"], "LOGAR_LICENSE_REQUEST") + self.assertEqual(lic_data["license_server"], "https://logar.eibl.tech") + self.assertEqual(lic_data["requested_seats"], 30) + self.assertIn("BEGIN CERTIFICATE REQUEST", lic_data["csr"]) + self.assertTrue(os.path.exists(os.path.join(self.server_cert_dir, "ca.csr"))) + self.assertTrue(os.path.exists(os.path.join(self.server_cert_dir, "logar_server.lic"))) + + def test_authority_issue_sub_ca_with_exact_seats(self): + """Verifies that License Authority signs CSR, embeds exact seat quota via custom OID, and verifies in DB.""" + csr_pem, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir) + result = authority.issue_license_from_csr( + csr_pem=csr_pem, + seats=42, + days_valid=365, + server_fingerprint="FP123456", + hostname="hub-node-1", + db_path=self.db_path, + cert_dir=self.master_ca_dir + ) + + self.assertEqual(result["status"], "success") + self.assertEqual(result["seats"], 42) + self.assertIn("BEGIN CERTIFICATE", result["ca_cert"]) + + # Decode certificate to verify seat quota extraction + info = enrollment.extract_license_info_from_cert(result["ca_cert"]) + self.assertTrue(info["valid"]) + self.assertEqual(info["seats"], 42) + self.assertEqual(info["license_id"], result["license_id"]) + self.assertEqual(info["issuer_url"], "https://logar.eibl.tech") + self.assertFalse(info["is_expired"]) + self.assertTrue(info["is_ca"]) + + # Check authority database + lics = authority.list_licenses(db_path=self.db_path) + self.assertEqual(len(lics), 1) + self.assertEqual(lics[0]["license_id"], result["license_id"]) + self.assertEqual(lics[0]["seats"], 42) + self.assertEqual(lics[0]["status"], "active") + + def test_authority_renewal_and_revocation(self): + """Verifies that an active license cert can be renewed, and revoked licenses are rejected.""" + csr_pem, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir) + initial = authority.issue_license_from_csr( + csr_pem=csr_pem, + seats=15, + db_path=self.db_path, + cert_dir=self.master_ca_dir + ) + lic_id = initial["license_id"] + + # Successful renewal + renewed = authority.renew_license_cert( + current_cert_pem=initial["ca_cert"], + days_valid=365, + db_path=self.db_path, + cert_dir=self.master_ca_dir + ) + self.assertEqual(renewed["status"], "renewed") + self.assertEqual(renewed["seats"], 15) + + # Check renewal count in DB + lics = authority.list_licenses(db_path=self.db_path) + self.assertEqual(lics[0]["renewal_count"], 1) + + # Revoke the license + revoked = authority.revoke_license(lic_id, db_path=self.db_path) + self.assertTrue(revoked) + + # Renewal attempt after revocation must fail + with self.assertRaises(ValueError): + authority.renew_license_cert( + current_cert_pem=renewed["ca_cert"], + db_path=self.db_path, + cert_dir=self.master_ca_dir + ) + + def test_server_license_import_and_seat_update(self): + """Verifies that importing a signed CA cert updates the server's seat quota in DB and memory.""" + # 1. Generate CSR on server side + csr_pem, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir) + + # 2. Authority signs it with 50 seats + signed = authority.issue_license_from_csr( + csr_pem=csr_pem, + seats=50, + db_path=self.db_path, + cert_dir=self.master_ca_dir + ) + + # 3. Setup mock server state and DB + 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, + "tcp_host": "127.0.0.1", + "tcp_port": 9443, + "hermes_host": "127.0.0.1", + "hermes_port": 8443, + "server_fingerprint": "MOCK_FP", + "server_name": "Test-Hub", + "license_server_url": "https://logar.eibl.tech" + } + + # 4. Extract summary before import + summary_before = Server.get_license_summary() + self.assertEqual(summary_before["max_seats"], 5) + + # 5. Save the signed CA certificate to cert_dir/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(signed["ca_cert"]) + + # 6. Extract summary after import + summary_after = Server.get_license_summary() + self.assertEqual(summary_after["max_seats"], 50) + self.assertEqual(summary_after["license_id"], signed["license_id"]) + self.assertEqual(summary_after["status"], "ACTIVE") + + def test_server_client_revocation_and_reactivation(self): + """Verifies revocation and reactivation of edge client seats.""" + Server.init_db(self.server_db, max_seats=2) + Server.SERVER_STATE["config"] = { + "db_path": self.server_db, + "cert_dir": self.server_cert_dir, + "max_seats": 2 + } + + # Insert active clients + conn = sqlite3.connect(self.server_db) + conn.execute("INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status) VALUES ('c1', 'win1', 'windows', 'FP1', 'active')") + conn.execute("INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status) VALUES ('c2', 'lin1', 'linux', 'FP2', 'active')") + conn.commit() + conn.close() + + # Revoke c1 + req = Server.ClientActionRequest(client_id="c1") + res = Server.revoke_client(req) + self.assertEqual(res["status"], "revoked") + + summary = Server.get_license_summary() + self.assertEqual(summary["active_seats"], 1) + self.assertEqual(summary["revoked_seats"], 1) + self.assertEqual(summary["available_seats"], 1) + + # Reactivate c1 + res_react = Server.reactivate_client(req) + self.assertEqual(res_react["status"], "active") + + summary2 = Server.get_license_summary() + self.assertEqual(summary2["active_seats"], 2) + self.assertEqual(summary2["available_seats"], 0) + + 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) + Server.SERVER_STATE["config"] = { + "db_path": self.server_db, + "cert_dir": self.server_cert_dir, + "max_seats": 10, + "tcp_port": 9443, + "hermes_port": 8443, + "server_name": "Test-Cloud-Hub", + "server_fingerprint": "1234567890ABCDEF", + "license_server_url": "https://logar.eibl.tech" + } + + response = Server.serve_webui() + html = response.body.decode("utf-8") + self.assertIn("LOGAR Fleet Hub & License Management", html.replace("&", "&")) + self.assertIn("Export .lic Request", html) + self.assertIn("Import & Activate Signed CA Certificate", html.replace("&", "&")) + self.assertIn("Renew via mTLS Now", html) + self.assertIn("logar.eibl.tech", html) + + +if __name__ == "__main__": + unittest.main()