test(licensing): add tests for Master License Authority, X.509 seat quotas, and WebUI
This commit is contained in:
@@ -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()
|
||||
Reference in New Issue
Block a user