me0nline
303744a3f8
CI Test Suite / Run Component Tests & Pipeline Verification (push) Failing after 1m35s
101 lines
3.9 KiB
Python
101 lines
3.9 KiB
Python
"""
|
|
Comprehensive unit tests for dynamic PKI, Sub-CA issuance, offline CSR activation, and LicenseAuthority.
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import sqlite3
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from src.hub import server_enrollment as enrollment
|
|
from src.hub.pki_manager import PKIManager
|
|
|
|
|
|
class TestHubEnrollmentPKIDeep(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.cert_dir = os.path.join(self.temp_dir, "certs")
|
|
os.makedirs(self.cert_dir, exist_ok=True)
|
|
self.db_path = os.path.join(self.temp_dir, "logar_test.db")
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.temp_dir, ignore_errors=True)
|
|
|
|
def test_pki_lifecycle_and_extensions(self):
|
|
"""Tests CA generation, Server TLS cert generation, client cert issuance, and license OID parsing."""
|
|
# 1. CA Generation
|
|
ca_cert, ca_key, ca_pem, ca_key_pem = enrollment.generate_ca_if_needed(cert_dir=self.cert_dir)
|
|
self.assertIsNotNone(ca_cert)
|
|
self.assertTrue(os.path.exists(os.path.join(self.cert_dir, "ca.crt")))
|
|
self.assertTrue(os.path.exists(os.path.join(self.cert_dir, "ca.key")))
|
|
|
|
# 2. Server TLS Certificate Generation
|
|
srv_cert, srv_key, srv_pem, srv_key_pem = enrollment.generate_server_cert_if_needed(
|
|
ca_cert=ca_cert,
|
|
ca_key=ca_key,
|
|
hostnames=["127.0.0.1", "localhost"],
|
|
cert_dir=self.cert_dir
|
|
)
|
|
self.assertIsNotNone(srv_cert)
|
|
self.assertTrue(os.path.exists(os.path.join(self.cert_dir, "server.crt")))
|
|
self.assertTrue(os.path.exists(os.path.join(self.cert_dir, "server.key")))
|
|
|
|
# 3. Client Certificate Issuance
|
|
cl_pem, cl_key_pem = enrollment.issue_client_cert(
|
|
client_id="forwarder-node-01",
|
|
ca_cert=ca_cert,
|
|
ca_key=ca_key,
|
|
days_valid=365
|
|
)
|
|
self.assertIn("BEGIN CERTIFICATE", cl_pem)
|
|
self.assertIn("BEGIN RSA PRIVATE KEY", cl_key_pem)
|
|
|
|
# 4. Extract license info
|
|
info = enrollment.extract_license_info_from_cert(ca_pem)
|
|
self.assertIn("is_self_signed", info)
|
|
|
|
# 5. Check cert expiry helper
|
|
self.assertFalse(enrollment.is_cert_expiring_soon(ca_pem, threshold_days=30))
|
|
|
|
# 6. Initialize mTLS Server context
|
|
ctx = PKIManager.init_mtls_server_context(cert_dir=self.cert_dir)
|
|
self.assertIsNotNone(ctx)
|
|
|
|
# 7. Reload mTLS context
|
|
PKIManager.reload_mtls_context(ctx, cert_dir=self.cert_dir)
|
|
|
|
def test_offline_csr_and_activation(self):
|
|
"""Tests export of offline license request CSR and activation bundle unpack."""
|
|
csr_pem, csr_key_pem = enrollment.generate_ca_csr(cert_dir=self.cert_dir)
|
|
self.assertIn("BEGIN CERTIFICATE REQUEST", csr_pem)
|
|
|
|
# Sign the CSR locally as Sub-CA for test
|
|
ca_cert, ca_key, ca_pem, ca_key_pem = enrollment.generate_ca_if_needed(cert_dir=self.cert_dir)
|
|
sub_ca_pem = enrollment.sign_ca_csr_with_seats(
|
|
csr_pem=csr_pem,
|
|
master_ca_cert_pem=ca_pem,
|
|
master_ca_key_pem=ca_key_pem,
|
|
seats=25,
|
|
days_valid=365,
|
|
license_id="LIC-TEST-2026",
|
|
features=["postgres", "realtime_alerting"]
|
|
)
|
|
self.assertIn("BEGIN CERTIFICATE", sub_ca_pem)
|
|
|
|
# Verify signed Sub-CA contains license extensions
|
|
info = enrollment.extract_license_info_from_cert(sub_ca_pem)
|
|
self.assertEqual(info.get("seats"), 25)
|
|
self.assertEqual(info.get("license_id"), "LIC-TEST-2026")
|
|
self.assertIn("postgres", info.get("features", []))
|
|
|
|
# Test create_license_request_lic
|
|
lic_data, lic_json = enrollment.create_license_request_lic(cert_dir=self.cert_dir)
|
|
self.assertEqual(lic_data["format"], "LOGAR_LICENSE_REQUEST")
|
|
self.assertIn("csr", lic_data)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|