From 2fba971aeba855e6baf73671a44b5f80ca189a77 Mon Sep 17 00:00:00 2001 From: max Date: Thu, 24 Sep 2026 16:42:10 +0200 Subject: [PATCH] feat(licensing): add Master License Authority program with CSR signing, exact seat quotas, and mTLS renewals --- src/License_Authority.py | 635 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 635 insertions(+) create mode 100644 src/License_Authority.py diff --git a/src/License_Authority.py b/src/License_Authority.py new file mode 100644 index 0000000..76facb6 --- /dev/null +++ b/src/License_Authority.py @@ -0,0 +1,635 @@ +#!/usr/bin/env python3 +""" +LOGAR Master License Authority (logar.eibl.tech) + +Authoritative PKI service and CLI utility responsible for: +1. Maintaining the Master Root CA. +2. Processing intermediate CA CSRs (or .lic request files) and issuing signed + subordinate CA certificates embedding exact seat quotas via custom enterprise X.509 OIDs. +3. Providing an automated mTLS renewal endpoint (/api/license/renew) for active servers. +4. Serving an administrative WebUI and REST API to manage and revoke server licenses. +""" + +import os +import sys +import json +import sqlite3 +import argparse +import datetime +from typing import Tuple, Optional, Dict, Any + +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 +SRC_DIR = os.path.dirname(os.path.abspath(__file__)) +if SRC_DIR not in sys.path: + sys.path.insert(0, SRC_DIR) + +import server_enrollment as enrollment + +try: + from fastapi import FastAPI, HTTPException, Request, Header + from fastapi.responses import HTMLResponse, JSONResponse + import uvicorn + FASTAPI_AVAILABLE = True +except ImportError: + FASTAPI_AVAILABLE = False + + +DEFAULT_DB = "license_authority.db" +DEFAULT_MASTER_CERT_DIR = "master_ca_certs" + + +def init_authority_db(db_path: str = DEFAULT_DB): + """Initializes the SQLite schema for tracking issued server licenses and revocations.""" + conn = sqlite3.connect(db_path) + c = conn.cursor() + c.execute(""" + CREATE TABLE IF NOT EXISTS licenses ( + license_id TEXT PRIMARY KEY, + server_fingerprint TEXT, + hostname TEXT, + seats INTEGER NOT NULL, + issued_at TEXT NOT NULL, + expires_at TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + renewal_count INTEGER NOT NULL DEFAULT 0, + cert_pem TEXT NOT NULL, + csr_pem TEXT, + client_cert_fingerprint TEXT + ) + """) + conn.commit() + conn.close() + + +def generate_master_root_ca( + cert_dir: str = DEFAULT_MASTER_CERT_DIR, + common_name: str = "LOGAR-Master-Root-CA", + organization: str = "LOGAR Global Licensing", + country: str = "AT" +) -> Tuple[str, str]: + """Generates the Master Root CA certificate and private key if not already existing.""" + os.makedirs(cert_dir, exist_ok=True) + cert_path = os.path.join(cert_dir, "master_root_ca.crt") + key_path = os.path.join(cert_dir, "master_root_ca.key") + + if os.path.exists(cert_path) and os.path.exists(key_path): + with open(cert_path, "r", encoding="utf-8") as f: + cert_pem = f.read() + with open(key_path, "r", encoding="utf-8") as f: + key_pem = f.read() + return cert_pem, key_pem + + ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096) + subject = issuer = x509.Name([ + x509.NameAttribute(NameOID.COUNTRY_NAME, country), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, organization), + x509.NameAttribute(NameOID.COMMON_NAME, common_name), + ]) + + now = datetime.datetime.now(datetime.timezone.utc) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(ca_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(minutes=5)) + .not_valid_after(now + datetime.timedelta(days=3650)) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, + key_encipherment=False, + key_cert_sign=True, + crl_sign=True, + content_commitment=False, + data_encipherment=False, + key_agreement=False, + encipher_only=False, + decipher_only=False + ), + critical=True + ) + .add_extension( + x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), + critical=False + ) + .sign(ca_key, hashes.SHA256()) + ) + + cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode("utf-8") + key_pem = ca_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption() + ).decode("utf-8") + + with open(cert_path, "w", encoding="utf-8") as f: + f.write(cert_pem) + with open(key_path, "w", encoding="utf-8") as f: + f.write(key_pem) + + return cert_pem, key_pem + + +def issue_license_from_csr( + csr_pem: str, + seats: int, + days_valid: int = 365, + license_id: Optional[str] = None, + server_fingerprint: str = "", + hostname: str = "", + db_path: str = DEFAULT_DB, + cert_dir: str = DEFAULT_MASTER_CERT_DIR, + license_server_url: str = "https://logar.eibl.tech" +) -> Dict[str, Any]: + """Issues a subordinate CA certificate with the requested number of seats and registers it in the DB.""" + init_authority_db(db_path) + master_cert_pem, master_key_pem = generate_master_root_ca(cert_dir) + + if not license_id: + csr_obj = x509.load_pem_x509_csr(csr_pem.encode("utf-8")) + der_pub = csr_obj.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo + ) + license_id = f"LIC-{der_pub.hex()[:12].upper()}" + + ca_cert_pem = enrollment.sign_ca_csr_with_seats( + csr_pem=csr_pem, + master_ca_cert_pem=master_cert_pem, + master_ca_key_pem=master_key_pem, + seats=seats, + days_valid=days_valid, + license_id=license_id, + license_server_url=license_server_url + ) + + info = enrollment.extract_license_info_from_cert(ca_cert_pem) + issued_at = info["not_valid_before"] + expires_at = info["not_valid_after"] + fp = info["fingerprint"] + + conn = sqlite3.connect(db_path) + c = conn.cursor() + c.execute(""" + INSERT INTO licenses (license_id, server_fingerprint, hostname, seats, issued_at, expires_at, status, renewal_count, cert_pem, csr_pem, client_cert_fingerprint) + VALUES (?, ?, ?, ?, ?, ?, 'active', 0, ?, ?, ?) + ON CONFLICT(license_id) DO UPDATE SET + seats = excluded.seats, + expires_at = excluded.expires_at, + status = 'active', + cert_pem = excluded.cert_pem, + client_cert_fingerprint = excluded.client_cert_fingerprint + """, (license_id, server_fingerprint, hostname, seats, issued_at, expires_at, ca_cert_pem, csr_pem, fp)) + conn.commit() + conn.close() + + return { + "status": "success", + "license_id": license_id, + "seats": seats, + "issued_at": issued_at, + "expires_at": expires_at, + "fingerprint": fp, + "ca_cert": ca_cert_pem, + "master_root_ca": master_cert_pem, + "license_server": license_server_url + } + + +def renew_license_cert( + current_cert_pem: str, + days_valid: int = 365, + db_path: str = DEFAULT_DB, + cert_dir: str = DEFAULT_MASTER_CERT_DIR, + license_server_url: str = "https://logar.eibl.tech" +) -> Dict[str, Any]: + """ + Validates an existing intermediate CA cert presenting for mTLS renewal, + verifies that the license is active, and issues a renewed certificate preserving seat quota. + """ + init_authority_db(db_path) + master_cert_pem, master_key_pem = generate_master_root_ca(cert_dir) + master_cert = x509.load_pem_x509_certificate(master_cert_pem.encode("utf-8")) + + current_cert = x509.load_pem_x509_certificate(current_cert_pem.encode("utf-8")) + + # Extract license info + info = enrollment.extract_license_info_from_cert(current_cert_pem) + license_id = info.get("license_id") + seats = info.get("seats") or 10 + + # Verify license record in database + conn = sqlite3.connect(db_path) + c = conn.cursor() + c.execute("SELECT status, renewal_count, csr_pem, seats FROM licenses WHERE license_id = ?", (license_id,)) + row = c.fetchone() + if row: + status, renewals, csr_pem, db_seats = row + if status == "revoked": + conn.close() + raise ValueError(f"License {license_id} is REVOKED and cannot be renewed.") + seats = db_seats or seats + else: + # Auto-enroll license if missing from DB (signed by our master CA) + renewals = 0 + csr_pem = None + + # Build renewed certificate using the same public key and subject + master_key = serialization.load_pem_private_key(master_key_pem.encode("utf-8"), password=None) + now = datetime.datetime.now(datetime.timezone.utc) + + builder = ( + x509.CertificateBuilder() + .subject_name(current_cert.subject) + .issuer_name(master_cert.subject) + .public_key(current_cert.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(minutes=5)) + .not_valid_after(now + datetime.timedelta(days=days_valid)) + .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, + key_encipherment=False, + key_cert_sign=True, + crl_sign=True, + content_commitment=False, + data_encipherment=False, + key_agreement=False, + encipher_only=False, + decipher_only=False + ), + critical=True + ) + .add_extension( + x509.SubjectKeyIdentifier.from_public_key(current_cert.public_key()), + critical=False + ) + .add_extension( + x509.AuthorityKeyIdentifier.from_issuer_public_key(master_key.public_key()), + critical=False + ) + .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, license_server_url.encode("utf-8")), + critical=False + ) + ) + + new_cert = builder.sign(master_key, hashes.SHA256()) + new_pem = new_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8") + new_info = enrollment.extract_license_info_from_cert(new_pem) + new_fp = new_info["fingerprint"] + + c.execute(""" + UPDATE licenses + SET renewal_count = renewal_count + 1, + expires_at = ?, + cert_pem = ?, + client_cert_fingerprint = ? + WHERE license_id = ? + """, (new_info["not_valid_after"], new_pem, new_fp, license_id)) + conn.commit() + conn.close() + + return { + "status": "renewed", + "license_id": license_id, + "seats": seats, + "ca_cert": new_pem, + "expires_at": new_info["not_valid_after"], + "fingerprint": new_fp + } + + +def revoke_license(license_id: str, db_path: str = DEFAULT_DB) -> bool: + """Marks a license as revoked in the database.""" + init_authority_db(db_path) + conn = sqlite3.connect(db_path) + c = conn.cursor() + c.execute("UPDATE licenses SET status = 'revoked' WHERE license_id = ?", (license_id,)) + changed = c.rowcount > 0 + conn.commit() + conn.close() + return changed + + +def list_licenses(db_path: str = DEFAULT_DB) -> list: + """Lists all registered licenses.""" + init_authority_db(db_path) + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + c = conn.cursor() + c.execute("SELECT license_id, server_fingerprint, hostname, seats, issued_at, expires_at, status, renewal_count, client_cert_fingerprint FROM licenses ORDER BY issued_at DESC") + rows = [dict(r) for r in c.fetchall()] + conn.close() + return rows + + +# --- FastAPI Web Service (for deployment at logar.eibl.tech) --- +if FASTAPI_AVAILABLE: + app = FastAPI(title="LOGAR License Authority", version="2.0.1") + + @app.post("/api/license/issue") + async def api_issue(request: Request): + try: + body = await request.json() + except Exception: + raise HTTPException(status_code=400, detail="Invalid JSON body") + + csr_pem = body.get("csr") + seats = int(body.get("seats", 10)) + server_fp = body.get("server_fingerprint", "") + hostname = body.get("hostname", "") + lic_id = body.get("license_id") + + if not csr_pem: + raise HTTPException(status_code=400, detail="Missing 'csr' in request") + + try: + result = issue_license_from_csr( + csr_pem=csr_pem, + seats=seats, + license_id=lic_id, + server_fingerprint=server_fp, + hostname=hostname + ) + return result + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + @app.post("/api/license/renew") + async def api_renew(request: Request): + try: + body = await request.json() + except Exception: + raise HTTPException(status_code=400, detail="Invalid JSON body") + + current_cert = body.get("current_cert") + if not current_cert: + raise HTTPException(status_code=400, detail="Missing 'current_cert' in request") + + try: + result = renew_license_cert(current_cert) + return result + except ValueError as ve: + raise HTTPException(status_code=403, detail=str(ve)) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + @app.get("/api/licenses") + def api_list(): + return {"licenses": list_licenses()} + + @app.post("/api/license/revoke") + async def api_revoke(request: Request): + body = await request.json() + lic_id = body.get("license_id") + if not lic_id: + raise HTTPException(status_code=400, detail="Missing 'license_id'") + ok = revoke_license(lic_id) + if not ok: + raise HTTPException(status_code=404, detail="License not found") + return {"status": "revoked", "license_id": lic_id} + + @app.get("/", response_class=HTMLResponse) + def web_dashboard(): + licenses = list_licenses() + rows = [] + for l in licenses: + badge_cls = "badge-active" if l["status"] == "active" else "badge-revoked" + rows.append(f""" + + {l['license_id']} + {l['seats']} + {l['status'].upper()} + {l['hostname'] or 'N/A'} + {l['client_cert_fingerprint'][:16]}... + {l['expires_at'][:10]} + {l['renewal_count']} + + + + + """) + rows_html = "".join(rows) if rows else "No licenses issued yet." + + html_tpl = """ + + + LOGAR Master License Authority (logar.eibl.tech) + + + + +
+

🛡️ LOGAR Master License Authority

+

Authoritative PKI for issuing seat-bearing Intermediate CA certificates and automatic mTLS renewals.

+ +
+

Issue Subordinate CA Certificate from CSR / .lic

+
+ + +
+
+ + +
+ + +
+ +
+

Active & Issued Server Licenses

+ + + + + + + + + + + + + + + __ROWS_HTML__ + +
License IDSeatsStatusHostCert FingerprintExpiresRenewalsActions
+
+
+ + + +""" + return html_tpl.replace("__ROWS_HTML__", rows_html) + + + +def main(): + parser = argparse.ArgumentParser(description="LOGAR Master License Authority (logar.eibl.tech)") + parser.add_argument("--init-master-ca", action="store_true", help="Initialize the Master Root CA") + parser.add_argument("--sign-lic", type=str, help="Path to .lic file or CSR PEM file to sign") + parser.add_argument("--seats", type=int, default=25, help="Number of seats to grant (default: 25)") + parser.add_argument("--out", type=str, default="ca.crt", help="Output path for signed CA cert") + parser.add_argument("--revoke", type=str, help="License ID to revoke") + parser.add_argument("--list", action="store_true", help="List all issued licenses") + parser.add_argument("--serve", action="store_true", help="Run the FastAPI / mTLS renewal service") + parser.add_argument("--host", type=str, default="0.0.0.0", help="Service host (default: 0.0.0.0)") + parser.add_argument("--port", type=int, default=8080, help="Service port (default: 8080)") + + args = parser.parse_args() + + if args.init_master_ca: + generate_master_root_ca() + print(f"[+] Master Root CA initialized in {DEFAULT_MASTER_CERT_DIR}/") + return + + if args.list: + lics = list_licenses() + print(f"\n{'License ID':<20} {'Seats':<8} {'Status':<10} {'Host':<20} {'Expires':<12} {'Renewals':<8}") + print("-" * 80) + for l in lics: + print(f"{l['license_id']:<20} {l['seats']:<8} {l['status']:<10} {l['hostname'] or 'N/A':<20} {l['expires_at'][:10]:<12} {l['renewal_count']:<8}") + return + + if args.revoke: + if revoke_license(args.revoke): + print(f"[+] License {args.revoke} revoked successfully.") + else: + print(f"[-] License {args.revoke} not found.") + return + + if args.sign_lic: + with open(args.sign_lic, "r", encoding="utf-8") as f: + content = f.read() + + csr_pem = content + server_fp = "" + hostname = "" + if content.strip().startswith("{"): + data = json.loads(content) + csr_pem = data["csr"] + server_fp = data.get("server_fingerprint", "") + hostname = data.get("hostname", "") + + res = issue_license_from_csr( + csr_pem=csr_pem, + seats=args.seats, + server_fingerprint=server_fp, + hostname=hostname + ) + with open(args.out, "w", encoding="utf-8") as f: + f.write(res["ca_cert"]) + + print(f"[+] Signed CA certificate issued successfully!") + print(f" - License ID: {res['license_id']}") + print(f" - Granted Seats: {res['seats']}") + print(f" - Expires: {res['expires_at']}") + print(f" - Saved to: {args.out}") + return + + if args.serve: + if not FASTAPI_AVAILABLE: + print("[-] FastAPI/uvicorn not installed.") + sys.exit(1) + generate_master_root_ca() + print(f"[*] Starting LOGAR License Authority at http://{args.host}:{args.port}") + uvicorn.run(app, host=args.host, port=args.port) + return + + parser.print_help() + + +if __name__ == "__main__": + main()