306 lines
12 KiB
Python
306 lines
12 KiB
Python
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_activated_lic_package_import_and_renewal_seat_increase(self):
|
|
"""Verifies that an activated .lic package can be imported, saving upstream_ca, and renewal expands seats."""
|
|
# 1. Generate CSR on server side
|
|
csr_pem, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir)
|
|
|
|
# 2. Authority signs it with 20 seats and produces activated .lic
|
|
issued = authority.issue_license_from_csr(
|
|
csr_pem=csr_pem,
|
|
seats=20,
|
|
server_fingerprint="FP_TEST_HUB",
|
|
hostname="hub.enterprise.local",
|
|
db_path=self.db_path,
|
|
cert_dir=self.master_ca_dir
|
|
)
|
|
self.assertIn("lic_package", issued)
|
|
self.assertIn("lic_json", issued)
|
|
|
|
# Write .lic file
|
|
lic_file = os.path.join(self.test_dir, "test_activated.lic")
|
|
with open(lic_file, "w", encoding="utf-8") as f:
|
|
f.write(issued["lic_json"])
|
|
|
|
# 3. Server imports .lic package
|
|
import_res = enrollment.import_license_package(lic_file, cert_dir=self.server_cert_dir)
|
|
self.assertEqual(import_res["seats"], 20)
|
|
self.assertEqual(import_res["license_id"], issued["license_id"])
|
|
self.assertTrue(os.path.exists(os.path.join(self.server_cert_dir, "ca.crt")))
|
|
self.assertTrue(os.path.exists(os.path.join(self.server_cert_dir, "upstream_ca.crt")))
|
|
|
|
# 4. Initialize server state and verify summary
|
|
Server.init_db(self.server_db, max_seats=20)
|
|
Server.SERVER_STATE["config"] = {
|
|
"db_path": self.server_db,
|
|
"cert_dir": self.server_cert_dir,
|
|
"max_seats": 20,
|
|
"tcp_host": "127.0.0.1",
|
|
"tcp_port": 9443,
|
|
"hermes_host": "127.0.0.1",
|
|
"hermes_port": 8443,
|
|
"server_fingerprint": "FP_TEST_HUB",
|
|
"server_name": "Test-Hub",
|
|
"license_server_url": "https://logar.eibl.tech"
|
|
}
|
|
summary = Server.get_license_summary()
|
|
self.assertEqual(summary["max_seats"], 20)
|
|
|
|
# 5. Authority increments seats by +15 on renewal (total 35)
|
|
with open(os.path.join(self.server_cert_dir, "ca.crt"), "r", encoding="utf-8") as f:
|
|
current_ca_pem = f.read()
|
|
|
|
renewed = authority.renew_license_cert(
|
|
current_cert_pem=current_ca_pem,
|
|
seat_increment=15,
|
|
db_path=self.db_path,
|
|
cert_dir=self.master_ca_dir
|
|
)
|
|
self.assertEqual(renewed["seats"], 35)
|
|
|
|
# 6. Server processes the renewed certificate and saves it to 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(renewed["ca_cert"])
|
|
|
|
new_info = enrollment.extract_license_info_from_cert(renewed["ca_cert"])
|
|
self.assertEqual(new_info["seats"], 35)
|
|
|
|
# Server updates DB and config
|
|
Server.SERVER_STATE["config"]["max_seats"] = new_info["seats"]
|
|
conn = sqlite3.connect(self.server_db)
|
|
conn.execute("UPDATE license_config SET max_seats = ? WHERE id = 1", (new_info["seats"],))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
summary_renewed = Server.get_license_summary()
|
|
self.assertEqual(summary_renewed["max_seats"], 35)
|
|
|
|
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 License (.lic / ca.crt)", html)
|
|
self.assertIn("Sync & Renew via mTLS", html.replace("&", "&"))
|
|
self.assertIn("logar.eibl.tech", html)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|