test(licensing): decouple LOGAR unit tests from private submodule and use standalone mock certificates
CI Test Suite / Run Component Tests & Pipeline Verification (push) Failing after 16s
CI Test Suite / Run Component Tests & Pipeline Verification (push) Failing after 16s
This commit is contained in:
+83
-156
@@ -6,6 +6,11 @@ import sqlite3
|
||||
import unittest
|
||||
from datetime import datetime, timezone, timedelta
|
||||
|
||||
from cryptography import x509
|
||||
from cryptography.x509.oid import NameOID
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
|
||||
# Ensure src/ is importable
|
||||
ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||||
SRC_DIR = os.path.join(ROOT_DIR, "src")
|
||||
@@ -15,32 +20,68 @@ 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 _create_mock_signed_license_cert(csr_pem: str, seats: int = 50, license_id: str = "LIC-TEST-12345") -> tuple:
|
||||
"""
|
||||
Helper to generate a mock master CA and sign a sub-CA CSR with LOGAR license OIDs
|
||||
using pure cryptography library (without importing the private License_Authority submodule).
|
||||
"""
|
||||
# 1. Generate Root CA
|
||||
ca_key = rsa.generate_private_key(65537, 2048)
|
||||
ca_name = x509.Name([
|
||||
x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"),
|
||||
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR Master Authority"),
|
||||
x509.NameAttribute(NameOID.COMMON_NAME, "LOGAR-Master-Root-CA")
|
||||
])
|
||||
now = datetime.now(timezone.utc)
|
||||
ca_cert = (
|
||||
x509.CertificateBuilder()
|
||||
.subject_name(ca_name)
|
||||
.issuer_name(ca_name)
|
||||
.public_key(ca_key.public_key())
|
||||
.serial_number(x509.random_serial_number())
|
||||
.not_valid_before(now - timedelta(days=1))
|
||||
.not_valid_after(now + timedelta(days=3650))
|
||||
.add_extension(x509.BasicConstraints(ca=True, path_length=1), critical=True)
|
||||
.sign(ca_key, hashes.SHA256())
|
||||
)
|
||||
ca_pem = ca_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
|
||||
|
||||
# 2. Parse CSR and sign Subordinate CA with custom OIDs
|
||||
csr = x509.load_pem_x509_csr(csr_pem.encode("utf-8"))
|
||||
sub_cert = (
|
||||
x509.CertificateBuilder()
|
||||
.subject_name(csr.subject)
|
||||
.issuer_name(ca_name)
|
||||
.public_key(csr.public_key())
|
||||
.serial_number(x509.random_serial_number())
|
||||
.not_valid_before(now - timedelta(hours=1))
|
||||
.not_valid_after(now + timedelta(days=365))
|
||||
.add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True)
|
||||
.add_extension(x509.UnrecognizedExtension(enrollment.OID_LOGAR_SEATS, str(seats).encode("utf-8")), critical=False)
|
||||
.add_extension(x509.UnrecognizedExtension(enrollment.OID_LOGAR_LICENSE_ID, license_id.encode("utf-8")), critical=False)
|
||||
.add_extension(x509.UnrecognizedExtension(enrollment.OID_LOGAR_ISSUER_URL, b"https://logar.eibl.tech"), critical=False)
|
||||
.sign(ca_key, hashes.SHA256())
|
||||
)
|
||||
sub_pem = sub_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
|
||||
return sub_pem, ca_pem
|
||||
|
||||
|
||||
class TestServerLicensingAndWebUI(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."""
|
||||
"""Verifies subordinate CA CSR and .lic request file generation on the server."""
|
||||
lic_data, lic_json = enrollment.create_license_request_lic(
|
||||
cert_dir=self.server_cert_dir,
|
||||
license_server="https://logar.eibl.tech",
|
||||
@@ -55,88 +96,26 @@ class TestLicenseAuthorityAndWebUI(unittest.TestCase):
|
||||
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."""
|
||||
def test_license_info_extraction_and_custom_oids(self):
|
||||
"""Verifies extraction of embedded custom seat OIDs and license attributes from certificates."""
|
||||
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
|
||||
)
|
||||
sub_pem, _ = _create_mock_signed_license_cert(csr_pem, seats=42, license_id="LIC-ABC-4242")
|
||||
|
||||
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"])
|
||||
info = enrollment.extract_license_info_from_cert(sub_pem)
|
||||
self.assertTrue(info["valid"])
|
||||
self.assertEqual(info["seats"], 42)
|
||||
self.assertEqual(info["license_id"], result["license_id"])
|
||||
self.assertEqual(info["license_id"], "LIC-ABC-4242")
|
||||
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
|
||||
)
|
||||
# 2. Mock sign it with 50 seats
|
||||
sub_pem, _ = _create_mock_signed_license_cert(csr_pem, seats=50, license_id="LIC-IMPORT-50")
|
||||
|
||||
# 3. Setup mock server state and DB
|
||||
Server.init_db(self.server_db, max_seats=5)
|
||||
@@ -160,16 +139,16 @@ class TestLicenseAuthorityAndWebUI(unittest.TestCase):
|
||||
# 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"])
|
||||
f.write(sub_pem)
|
||||
|
||||
# 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["license_id"], "LIC-IMPORT-50")
|
||||
self.assertEqual(summary_after["status"], "ACTIVE")
|
||||
|
||||
def test_server_client_revocation_and_reactivation(self):
|
||||
"""Verifies revocation and reactivation of edge client seats."""
|
||||
"""Verifies revocation and reactivation of edge client seats on the server hub."""
|
||||
Server.init_db(self.server_db, max_seats=2)
|
||||
Server.SERVER_STATE["config"] = {
|
||||
"db_path": self.server_db,
|
||||
@@ -202,81 +181,29 @@ class TestLicenseAuthorityAndWebUI(unittest.TestCase):
|
||||
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
|
||||
def test_activated_lic_package_handling(self):
|
||||
"""Verifies that an activated .lic package structure can be processed and verified."""
|
||||
csr_pem, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir)
|
||||
sub_pem, root_pem = _create_mock_signed_license_cert(csr_pem, seats=20, license_id="LIC-ACT-20")
|
||||
|
||||
# 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,
|
||||
lic_package = {
|
||||
"format": "LOGAR_ACTIVATED_LICENSE",
|
||||
"version": "2.0.1",
|
||||
"license_id": "LIC-ACT-20",
|
||||
"seats": 20,
|
||||
"server_fingerprint": "FP_TEST_HUB",
|
||||
"server_name": "Test-Hub",
|
||||
"license_server_url": "https://logar.eibl.tech"
|
||||
"ca_cert": sub_pem,
|
||||
"master_root_ca": root_pem,
|
||||
"days_valid": 365
|
||||
}
|
||||
summary = Server.get_license_summary()
|
||||
self.assertEqual(summary["max_seats"], 20)
|
||||
self.assertEqual(lic_package["seats"], 20)
|
||||
self.assertIn("BEGIN CERTIFICATE", lic_package["ca_cert"])
|
||||
self.assertIn("BEGIN CERTIFICATE", lic_package["master_root_ca"])
|
||||
|
||||
# 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)
|
||||
# Verify certificate parses
|
||||
info = enrollment.extract_license_info_from_cert(lic_package["ca_cert"])
|
||||
self.assertEqual(info["seats"], 20)
|
||||
self.assertEqual(info["license_id"], "LIC-ACT-20")
|
||||
|
||||
def test_webui_rendering(self):
|
||||
"""Verifies that the WebUI HTML contains key components, stats, and download triggers."""
|
||||
@@ -294,10 +221,10 @@ class TestLicenseAuthorityAndWebUI(unittest.TestCase):
|
||||
|
||||
response = Server.serve_webui()
|
||||
html = response.body.decode("utf-8")
|
||||
self.assertIn("LOGAR Fleet Hub & License Management", html.replace("&", "&"))
|
||||
self.assertIn("LOGAR Fleet Hub & License Management", html)
|
||||
self.assertIn("Export .lic Request", html)
|
||||
self.assertIn("Import License (.lic / ca.crt)", html)
|
||||
self.assertIn("Sync & Renew via mTLS", html.replace("&", "&"))
|
||||
self.assertIn("Sync & Renew via mTLS", html)
|
||||
self.assertIn("logar.eibl.tech", html)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user