From f0c4a82fc3c8cd2cad9e69f8bb6170c8ed1ced70 Mon Sep 17 00:00:00 2001 From: max Date: Fri, 2 Oct 2026 22:24:21 +0200 Subject: [PATCH] Harden PKI license verification, database seat reconciliation, and binary packaging --- compilation/package_dist.py | 4 +- src/Server.py | 19 +++++-- src/license_manager.py | 53 ++++++++++++++++---- src/server_enrollment.py | 29 ++++++++++- tests/test_license_authority.py | 89 +++++++++++++++++++++++++++++++++ 5 files changed, 180 insertions(+), 14 deletions(-) diff --git a/compilation/package_dist.py b/compilation/package_dist.py index f0b6bab..e75b29d 100644 --- a/compilation/package_dist.py +++ b/compilation/package_dist.py @@ -60,11 +60,13 @@ def write_checksum_file(filename, digest, binary_filename): def build_pyinstaller_binary(script_path, binary_name): - print(f"[*] Compiling {binary_name} with PyInstaller...") + print(f"[*] Compiling {binary_name} with PyInstaller (Bytecode optimized & stripped)...") cmd = [ sys.executable, "-m", "PyInstaller", "--onefile", "--clean", + "--strip", + "--optimize", "2", "--distpath", DIST_DIR, "--workpath", os.path.join(BUILD_TEMP, f"work_{binary_name}"), "--specpath", os.path.join(BUILD_TEMP, f"spec_{binary_name}"), diff --git a/src/Server.py b/src/Server.py index 5d7ce39..18f3acf 100644 --- a/src/Server.py +++ b/src/Server.py @@ -1313,8 +1313,10 @@ def enroll_client(req: ClientEnrollRequest, request: Request = None): conn.close() raise HTTPException(status_code=500, detail="License configuration not initialized") - expected_secret, max_seats = row[0], row[1] - license_status = row[2] if len(row) > 2 and row[2] else ("active" if max_seats > 0 else "unlicensed") + expected_secret = row[0] + lic_summary = LicenseManager.get_license_summary(SERVER_STATE, db_path=db_path) + max_seats = lic_summary.get("max_seats", 0) + license_status = lic_summary.get("status", "UNLICENSED").lower() if max_seats <= 0 or license_status in ("unlicensed", "revoked", "expired"): conn.close() raise HTTPException(status_code=403, detail="Server is unlicensed. Import an activated .lic file via the WebUI before enrolling edge clients.") @@ -1893,9 +1895,20 @@ def import_signed_license(req: LicenseImportRequest, auth_user: str = Depends(re cert_dir = SERVER_STATE.get("config", {}).get("cert_dir", "certs") ca_file = os.path.join(cert_dir, "ca.crt") upstream_file = os.path.join(cert_dir, "upstream_ca.crt") + upstream_pem = req.upstream_ca.strip() if req.upstream_ca else None + if not upstream_pem and os.path.exists(upstream_file): + try: + with open(upstream_file, "r", encoding="utf-8") as f: + upstream_pem = f.read().strip() + except Exception: + upstream_pem = None + + if upstream_pem: + if not enrollment.verify_certificate_signature(req.ca_cert, upstream_pem): + raise HTTPException(status_code=400, detail="Cryptographic verification failed: Sub-CA certificate was not signed by the upstream Master Authority.") try: - info = enrollment.extract_license_info_from_cert(req.ca_cert) + info = enrollment.extract_license_info_from_cert(req.ca_cert, upstream_ca_pem=upstream_pem) except Exception as e: raise HTTPException(status_code=400, detail=f"Failed to parse X.509 certificate: {e}") diff --git a/src/license_manager.py b/src/license_manager.py index f103765..7f03513 100644 --- a/src/license_manager.py +++ b/src/license_manager.py @@ -42,15 +42,23 @@ class LicenseManager: def get_active_licensed_features(db_path: Optional[str] = None, server_state: Optional[Dict[str, Any]] = None) -> List[str]: """ Returns the list of all active enterprise capability identifiers - authorized either via X.509 certificate extensions or the persistent database license config. + authorized either via verified X.509 certificate extensions or the persistent database license config. """ features = set() state = server_state or {} + cert_dir = state.get("config", {}).get("cert_dir", "certs") + upstream_pem = None + upstream_file = os.path.join(cert_dir, "upstream_ca.crt") + if os.path.exists(upstream_file): + try: + with open(upstream_file, "r", encoding="utf-8") as f: + upstream_pem = f.read() + except Exception: + pass # 1. Check in-memory / active Subordinate CA certificate ca_pem = state.get("ca_cert_pem") if not ca_pem: - cert_dir = state.get("config", {}).get("cert_dir", "certs") ca_file = os.path.join(cert_dir, "ca.crt") if os.path.exists(ca_file): try: @@ -59,9 +67,11 @@ class LicenseManager: except Exception: pass + cert_is_verified = False if ca_pem: try: - info = enrollment.extract_license_info_from_cert(ca_pem) + info = enrollment.extract_license_info_from_cert(ca_pem, upstream_ca_pem=upstream_pem) + cert_is_verified = info.get("signature_verified", False) or not info.get("is_self_signed", False) for feat in info.get("features", []): if feat: features.add(str(feat).strip()) @@ -105,14 +115,23 @@ class LicenseManager: @staticmethod def get_license_summary(server_state: Optional[Dict[str, Any]] = None, db_path: Optional[str] = None) -> Dict[str, Any]: - """Inspects certs/ca.crt, database quotas, and returns comprehensive licensing status.""" + """Inspects certs/ca.crt, database quotas, verifies cryptographic signatures, and returns comprehensive licensing status.""" state = server_state or {} config = state.get("config", {}) cert_dir = config.get("cert_dir", "certs") ca_path = os.path.join(cert_dir, "ca.crt") + upstream_path = os.path.join(cert_dir, "upstream_ca.crt") target_db = db_path or config.get("db_path", LicenseManager.DEFAULT_DB_FILE) - max_seats = config.get("max_seats", 0) + upstream_pem = None + if os.path.exists(upstream_path): + try: + with open(upstream_path, "r", encoding="utf-8") as f: + upstream_pem = f.read().strip() + except Exception: + pass + + db_max_seats = config.get("max_seats", 0) license_status_db = "unlicensed" try: conn = sqlite3.connect(target_db) @@ -121,7 +140,7 @@ class LicenseManager: row = c.fetchone() if row: if row[0] is not None: - max_seats = row[0] + db_max_seats = row[0] if len(row) > 1 and row[1]: license_status_db = row[1] c.execute("SELECT COUNT(*) FROM clients WHERE status = 'active' AND cert_fingerprint != 'LOCAL-HOST-BUILTIN'") @@ -136,18 +155,33 @@ class LicenseManager: csr_path = os.path.join(cert_dir, "ca.csr") csr_exists = os.path.exists(csr_path) + cert_seats = None if os.path.exists(ca_path): try: with open(ca_path, "r", encoding="utf-8") as f: ca_pem = f.read() - info = enrollment.extract_license_info_from_cert(ca_pem) - if info.get("seats"): - max_seats = info["seats"] + info = enrollment.extract_license_info_from_cert(ca_pem, upstream_ca_pem=upstream_pem) + if info.get("seats") is not None: + cert_seats = info["seats"] except Exception: info = {"valid": False} else: info = {"valid": False, "is_self_signed": False, "is_expired": False, "days_remaining": 0} + # Anti-tampering reconciliation: If database max_seats diverged from certified quota, enforce certified quota and update DB + if cert_seats is not None: + max_seats = cert_seats + if db_max_seats != cert_seats and os.path.exists(target_db): + try: + conn = sqlite3.connect(target_db) + conn.execute("UPDATE license_config SET max_seats = ? WHERE id = 1", (cert_seats,)) + conn.commit() + conn.close() + except Exception: + pass + else: + max_seats = db_max_seats + license_id = info.get("license_id") or "UNLICENSED" issuer_url = info.get("issuer_url") or config.get("license_server_url", "https://logar.eibl.tech") @@ -176,6 +210,7 @@ class LicenseManager: "issuer_cn": info.get("issuer_cn", "N/A"), "subject_cn": info.get("subject_cn", "N/A"), "is_self_signed": info.get("is_self_signed", False), + "signature_verified": info.get("signature_verified", False), "not_valid_before": info.get("not_valid_before", "N/A"), "not_valid_after": info.get("not_valid_after", "N/A"), "days_remaining": info.get("days_remaining", 0), diff --git a/src/server_enrollment.py b/src/server_enrollment.py index bdafa8e..202af53 100644 --- a/src/server_enrollment.py +++ b/src/server_enrollment.py @@ -408,10 +408,30 @@ def check_and_renew_hub_pki( return False, False -def extract_license_info_from_cert(cert_pem: str) -> Dict[str, Any]: +def verify_certificate_signature(cert_pem: Any, issuer_ca_pem: Any) -> bool: + """ + Cryptographically verifies that cert_pem was signed by the private key corresponding to issuer_ca_pem. + """ + try: + from cryptography.hazmat.primitives.asymmetric import padding + cert = x509.load_pem_x509_certificate(cert_pem.encode("utf-8")) if isinstance(cert_pem, str) else cert_pem + issuer = x509.load_pem_x509_certificate(issuer_ca_pem.encode("utf-8")) if isinstance(issuer_ca_pem, str) else issuer_ca_pem + issuer.public_key().verify( + cert.signature, + cert.tbs_certificate_bytes, + padding.PKCS1v15(), + cert.signature_hash_algorithm + ) + return True + except Exception: + return False + + +def extract_license_info_from_cert(cert_pem: str, upstream_ca_pem: Optional[str] = None) -> Dict[str, Any]: """ Extracts embedded licensing attributes, validity, and metadata from a PEM X.509 certificate. Supports decoding seat quota and license IDs embedded via LOGAR custom enterprise OIDs. + Cryptographically verifies the signature if upstream_ca_pem is provided. """ try: cert = x509.load_pem_x509_certificate(cert_pem.encode("utf-8")) @@ -490,6 +510,12 @@ def extract_license_info_from_cert(cert_pem: str) -> Dict[str, Any]: break is_self_signed = (cert.subject == cert.issuer) + signature_verified = False + + if upstream_ca_pem: + signature_verified = verify_certificate_signature(cert, upstream_ca_pem) + elif is_self_signed: + signature_verified = verify_certificate_signature(cert, cert) return { "valid": True, @@ -501,6 +527,7 @@ def extract_license_info_from_cert(cert_pem: str) -> Dict[str, Any]: "issuer_cn": issuer_cn, "is_ca": is_ca, "is_self_signed": is_self_signed, + "signature_verified": signature_verified, "fingerprint": cert.fingerprint(hashes.SHA256()).hex().upper(), "not_valid_before": not_before.isoformat(), "not_valid_after": not_after.isoformat(), diff --git a/tests/test_license_authority.py b/tests/test_license_authority.py index 1dc9cd1..1f27501 100644 --- a/tests/test_license_authority.py +++ b/tests/test_license_authority.py @@ -205,6 +205,94 @@ class TestServerLicensingAndWebUI(unittest.TestCase): self.assertEqual(info["seats"], 20) self.assertEqual(info["license_id"], "LIC-ACT-20") + def test_signature_verification_and_anti_forgery(self): + """Verifies cryptographic signature validation against master root CA and detection of forged certificates.""" + csr_pem, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir) + sub_pem, root_pem = _create_mock_signed_license_cert(csr_pem, seats=30, license_id="LIC-VALID-30") + + # 1. Valid signature against genuine root CA + self.assertTrue(enrollment.verify_certificate_signature(sub_pem, root_pem)) + info_valid = enrollment.extract_license_info_from_cert(sub_pem, upstream_ca_pem=root_pem) + self.assertTrue(info_valid["signature_verified"]) + self.assertEqual(info_valid["seats"], 30) + + # 2. Fake / Rogue Root CA trying to verify the same cert + fake_csr, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir) + _, fake_root_pem = _create_mock_signed_license_cert(fake_csr, seats=999, license_id="LIC-FAKE") + self.assertFalse(enrollment.verify_certificate_signature(sub_pem, fake_root_pem)) + info_invalid = enrollment.extract_license_info_from_cert(sub_pem, upstream_ca_pem=fake_root_pem) + self.assertFalse(info_invalid["signature_verified"]) + + def test_database_tampering_seat_reconciliation(self): + """Verifies that direct DB manipulation of max_seats is reconciled against the certified X.509 quota.""" + from license_manager import LicenseManager + + csr_pem, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir) + sub_pem, root_pem = _create_mock_signed_license_cert(csr_pem, seats=25, license_id="LIC-SEAT-25") + + # Save certified ca.crt and upstream_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(sub_pem) + with open(os.path.join(self.server_cert_dir, "upstream_ca.crt"), "w", encoding="utf-8") as f: + f.write(root_pem) + + Server.init_db(self.server_db, max_seats=25) + server_state = { + "config": { + "db_path": self.server_db, + "cert_dir": self.server_cert_dir, + "max_seats": 25 + } + } + + # Attacker attempts direct SQLite manipulation: SET max_seats = 99999 + conn = sqlite3.connect(self.server_db) + conn.execute("UPDATE license_config SET max_seats = 99999 WHERE id = 1") + conn.commit() + conn.close() + + # LicenseManager inspects state and auto-reconciles max_seats to certified 25 + summary = LicenseManager.get_license_summary(server_state=server_state, db_path=self.server_db) + self.assertEqual(summary["max_seats"], 25) + self.assertTrue(summary["signature_verified"]) + + # Check DB was reconciled + conn = sqlite3.connect(self.server_db) + row = conn.execute("SELECT max_seats FROM license_config WHERE id = 1").fetchone() + conn.close() + self.assertEqual(row[0], 25) + + def test_import_license_signature_validation(self): + """Verifies that /api/license/import rejects certificates not signed by the upstream authority.""" + from fastapi import HTTPException + csr_pem, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir) + sub_pem, root_pem = _create_mock_signed_license_cert(csr_pem, seats=50, license_id="LIC-MOCK-50") + + # Generate mismatched root CA + fake_csr, _ = enrollment.generate_ca_csr(cert_dir=self.server_cert_dir) + _, rogue_root_pem = _create_mock_signed_license_cert(fake_csr, seats=10, license_id="LIC-ROGUE") + + 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 + } + + # Attempt to import sub_pem with mismatched upstream_ca -> must raise HTTPException 400 + req_bad = Server.LicenseImportRequest(ca_cert=sub_pem, upstream_ca=rogue_root_pem) + with self.assertRaises(HTTPException) as ctx: + Server.import_signed_license(req_bad, auth_user="admin") + self.assertEqual(ctx.exception.status_code, 400) + self.assertIn("Cryptographic verification failed", ctx.exception.detail) + + # Import sub_pem with valid matching upstream_ca -> must succeed + req_good = Server.LicenseImportRequest(ca_cert=sub_pem, upstream_ca=root_pem) + res = Server.import_signed_license(req_good, auth_user="admin") + self.assertEqual(res["status"], "success") + self.assertEqual(res["seats"], 50) + 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) @@ -231,3 +319,4 @@ class TestServerLicensingAndWebUI(unittest.TestCase): if __name__ == "__main__": unittest.main() +