From 5a303808a34beb8553133d73904aa5d8410457a4 Mon Sep 17 00:00:00 2001 From: max Date: Thu, 24 Sep 2026 17:17:35 +0200 Subject: [PATCH] refactor(licensing): delegate src/License_Authority.py to LOGAR-LICENSING submodule --- src/License_Authority.py | 648 ++------------------------------------- 1 file changed, 23 insertions(+), 625 deletions(-) diff --git a/src/License_Authority.py b/src/License_Authority.py index 76facb6..52f4d39 100644 --- a/src/License_Authority.py +++ b/src/License_Authority.py @@ -1,635 +1,33 @@ #!/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. +LOGAR Master License Authority Proxy +Delegates directly to the standalone submodule: LOGAR-LICENSING/License_Authority.py +Repo: https://gitea.eibl.tech/me0nline/LOGAR-LICENSING.git """ 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() +# Prefer the submodule implementation +ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +SUBMODULE_DIR = os.path.join(ROOT_DIR, "LOGAR-LICENSING") +if SUBMODULE_DIR not in sys.path: + sys.path.insert(0, SUBMODULE_DIR) + +from License_Authority import ( + init_authority_db, + extract_license_info_from_cert, + generate_master_root_ca, + sign_ca_csr_with_seats, + issue_license_from_csr, + renew_license_cert, + revoke_license, + list_licenses, + main, + OID_LOGAR_SEATS, + OID_LOGAR_LICENSE_ID, + OID_LOGAR_ISSUER_URL +) if __name__ == "__main__": main()