import os import json import ssl import datetime import ipaddress import urllib.request import urllib.error import urllib.parse import http.client from typing import Tuple, List, Optional, Dict, Any from cryptography import x509 from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa # Custom Enterprise X.509 OIDs for LOGAR Licensing OID_LOGAR_SEATS = x509.ObjectIdentifier("1.3.6.1.4.1.59999.1.1") OID_LOGAR_LICENSE_ID = x509.ObjectIdentifier("1.3.6.1.4.1.59999.1.2") OID_LOGAR_ISSUER_URL = x509.ObjectIdentifier("1.3.6.1.4.1.59999.1.3") OID_LOGAR_FEATURES = x509.ObjectIdentifier("1.3.6.1.4.1.59999.1.4") def calculate_cert_fingerprint(cert_pem: str) -> str: """Computes SHA-256 fingerprint for a PEM-encoded X.509 certificate.""" cert = x509.load_pem_x509_certificate(cert_pem.encode("utf-8")) return cert.fingerprint(hashes.SHA256()).hex().upper() def is_cert_expiring_soon(cert_pem: str, threshold_days: int = 30) -> bool: """ Checks if a PEM-encoded X.509 certificate expires within `threshold_days` (or is already expired). Returns True if expiring soon or expired, False otherwise. """ try: cert = x509.load_pem_x509_certificate(cert_pem.encode("utf-8")) expiry = getattr(cert, "not_valid_after_utc", None) if expiry is None: expiry = cert.not_valid_after.replace(tzinfo=datetime.timezone.utc) now = datetime.datetime.now(datetime.timezone.utc) return expiry <= (now + datetime.timedelta(days=threshold_days)) except Exception: return True def generate_ca_if_needed( cert_dir: str = "certs", common_name: str = "LOGAR-Root-CA", force_renew: bool = False, threshold_days: int = 30 ) -> Tuple[x509.Certificate, rsa.RSAPrivateKey, str, str]: """ Loads an existing Root CA or generates a self-signed Root CA certificate and private key. If existing CA cert is expiring within threshold_days (or force_renew is True), regenerates it. Returns (ca_cert_obj, ca_key_obj, ca_cert_pem, ca_key_pem). """ os.makedirs(cert_dir, exist_ok=True) ca_cert_path = os.path.join(cert_dir, "ca.crt") ca_key_path = os.path.join(cert_dir, "ca.key") if not force_renew and os.path.exists(ca_cert_path) and os.path.exists(ca_key_path): with open(ca_cert_path, "r", encoding="utf-8") as f: ca_cert_pem = f.read() with open(ca_key_path, "r", encoding="utf-8") as f: ca_key_pem = f.read() try: ca_cert = x509.load_pem_x509_certificate(ca_cert_pem.encode("utf-8")) ca_key = serialization.load_pem_private_key(ca_key_pem.encode("utf-8"), password=None) if not is_cert_expiring_soon(ca_cert_pem, threshold_days=threshold_days): return ca_cert, ca_key, ca_cert_pem, ca_key_pem except Exception: pass # Create timestamped backup of previous CA if present if os.path.exists(ca_cert_path): try: timestamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d_%H%M%S") os.replace(ca_cert_path, f"{ca_cert_path}.{timestamp}.bak") if os.path.exists(ca_key_path): os.replace(ca_key_path, f"{ca_key_path}.{timestamp}.bak") except Exception: pass # Generate RSA 4096 private key for Root CA ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096) subject = issuer = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR"), x509.NameAttribute(NameOID.COMMON_NAME, common_name), ]) now = datetime.datetime.now(datetime.timezone.utc) ca_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()) ) ca_cert_pem = ca_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8") ca_key_pem = ca_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption() ).decode("utf-8") with open(ca_cert_path, "w", encoding="utf-8") as f: f.write(ca_cert_pem) with open(ca_key_path, "w", encoding="utf-8") as f: f.write(ca_key_pem) try: os.chmod(ca_key_path, 0o600) except Exception: pass return ca_cert, ca_key, ca_cert_pem, ca_key_pem def generate_server_cert_if_needed( ca_cert: x509.Certificate, ca_key: rsa.RSAPrivateKey, hostnames: Optional[List[str]] = None, cert_dir: str = "certs", days_valid: int = 825, force_renew: bool = False, threshold_days: int = 30 ) -> Tuple[x509.Certificate, rsa.RSAPrivateKey, str, str]: """ Loads an existing server certificate or generates a new server TLS certificate signed by the Root CA. If existing server cert is expiring within threshold_days (or force_renew is True), regenerates it. Includes SANs for localhost, 127.0.0.1, and specified hostnames. """ os.makedirs(cert_dir, exist_ok=True) server_cert_path = os.path.join(cert_dir, "server.crt") server_key_path = os.path.join(cert_dir, "server.key") if not force_renew and os.path.exists(server_cert_path) and os.path.exists(server_key_path): with open(server_cert_path, "r", encoding="utf-8") as f: server_cert_pem = f.read() with open(server_key_path, "r", encoding="utf-8") as f: server_key_pem = f.read() try: srv_cert = x509.load_pem_x509_certificate(server_cert_pem.encode("utf-8")) srv_key = serialization.load_pem_private_key(server_key_pem.encode("utf-8"), password=None) if not is_cert_expiring_soon(server_cert_pem, threshold_days=threshold_days): return srv_cert, srv_key, server_cert_pem, server_key_pem except Exception: pass # Create timestamped backup of previous server cert if present if os.path.exists(server_cert_path): try: timestamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d_%H%M%S") os.replace(server_cert_path, f"{server_cert_path}.{timestamp}.bak") if os.path.exists(server_key_path): os.replace(server_key_path, f"{server_key_path}.{timestamp}.bak") except Exception: pass server_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) subject = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR"), x509.NameAttribute(NameOID.COMMON_NAME, "LOGAR-Server-Hub"), ]) san_list = [ x509.DNSName("localhost"), x509.DNSName("LOGAR-Server-Hub"), x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")), x509.IPAddress(ipaddress.IPv6Address("::1")), ] if hostnames: for host in hostnames: if not host: continue try: ip_obj = ipaddress.ip_address(host) san_list.append(x509.IPAddress(ip_obj)) except ValueError: san_list.append(x509.DNSName(host)) now = datetime.datetime.now(datetime.timezone.utc) server_cert = ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(ca_cert.subject) .public_key(server_key.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=False, path_length=None), critical=True) .add_extension( x509.KeyUsage( digital_signature=True, key_encipherment=True, key_cert_sign=False, crl_sign=False, content_commitment=False, data_encipherment=False, key_agreement=False, encipher_only=False, decipher_only=False ), critical=True ) .add_extension( x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=False ) .add_extension( x509.SubjectKeyIdentifier.from_public_key(server_key.public_key()), critical=False ) .add_extension( x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False ) .add_extension(x509.SubjectAlternativeName(san_list), critical=False) .sign(ca_key, hashes.SHA256()) ) server_cert_pem = server_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8") server_key_pem = server_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption() ).decode("utf-8") # Bundle subordinate CA and upstream CA into server.crt for complete TLS handshake chain full_server_cert_pem = server_cert_pem ca_cert_path = os.path.join(cert_dir, "ca.crt") if os.path.exists(ca_cert_path): with open(ca_cert_path, "r", encoding="utf-8") as f: ca_pem = f.read().strip() if ca_pem and ca_pem not in full_server_cert_pem: full_server_cert_pem = full_server_cert_pem.strip() + "\n" + ca_pem upstream_ca_path = os.path.join(cert_dir, "upstream_ca.crt") if os.path.exists(upstream_ca_path): with open(upstream_ca_path, "r", encoding="utf-8") as f: upstream_pem = f.read().strip() if upstream_pem and upstream_pem not in full_server_cert_pem: full_server_cert_pem = full_server_cert_pem.strip() + "\n" + upstream_pem with open(server_cert_path, "w", encoding="utf-8") as f: f.write(full_server_cert_pem) with open(server_key_path, "w", encoding="utf-8") as f: f.write(server_key_pem) try: os.chmod(server_key_path, 0o600) except Exception: pass return server_cert, server_key, server_cert_pem, server_key_pem def issue_client_cert( client_id: str, ca_cert: x509.Certificate, ca_key: rsa.RSAPrivateKey, days_valid: int = 365 ) -> Tuple[str, str]: """ Generates a 2048-bit RSA private key and signs an X.509 client certificate with Common Name set to client_id. Returns (cert_pem, key_pem). """ client_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) subject = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR"), x509.NameAttribute(NameOID.COMMON_NAME, client_id), ]) now = datetime.datetime.now(datetime.timezone.utc) cert = ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(ca_cert.subject) .public_key(client_key.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=False, path_length=None), critical=True) .add_extension( x509.KeyUsage( digital_signature=True, key_encipherment=True, key_cert_sign=False, crl_sign=False, content_commitment=False, data_encipherment=False, key_agreement=False, encipher_only=False, decipher_only=False ), critical=True ) .add_extension( x509.ExtendedKeyUsage([ExtendedKeyUsageOID.CLIENT_AUTH]), critical=False ) .add_extension( x509.SubjectKeyIdentifier.from_public_key(client_key.public_key()), critical=False ) .add_extension( x509.AuthorityKeyIdentifier.from_issuer_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 = client_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption() ).decode("utf-8") return cert_pem, key_pem def check_and_renew_hub_pki( cert_dir: str = "certs", hostnames: Optional[List[str]] = None, threshold_days: int = 30 ) -> Tuple[bool, bool]: """ Evaluates expiration status of Root CA and Server TLS certificates. If CA certificate is expiring within threshold_days (or missing): - Regenerates Root CA. - Automatically regenerates Server TLS certificate (since CA issuer changed). - Returns (ca_renewed=True, server_renewed=True) Else if Server TLS certificate is expiring within threshold_days (or missing): - Regenerates Server TLS certificate signed by existing Root CA. - Returns (ca_renewed=False, server_renewed=True) Otherwise: - Returns (False, False) """ os.makedirs(cert_dir, exist_ok=True) ca_cert_path = os.path.join(cert_dir, "ca.crt") server_cert_path = os.path.join(cert_dir, "server.crt") renew_ca = False renew_server = False if not os.path.exists(ca_cert_path): renew_ca = True else: try: with open(ca_cert_path, "r", encoding="utf-8") as f: ca_pem = f.read() if is_cert_expiring_soon(ca_pem, threshold_days=threshold_days): renew_ca = True except Exception: renew_ca = True if renew_ca: ca_cert, ca_key, _, _ = generate_ca_if_needed(cert_dir=cert_dir, force_renew=True) generate_server_cert_if_needed(ca_cert, ca_key, hostnames=hostnames, cert_dir=cert_dir, force_renew=True) return True, True if not os.path.exists(server_cert_path): renew_server = True else: try: with open(server_cert_path, "r", encoding="utf-8") as f: srv_pem = f.read() if is_cert_expiring_soon(srv_pem, threshold_days=threshold_days): renew_server = True except Exception: renew_server = True if renew_server: ca_cert, ca_key, _, _ = generate_ca_if_needed(cert_dir=cert_dir, force_renew=False) generate_server_cert_if_needed(ca_cert, ca_key, hostnames=hostnames, cert_dir=cert_dir, force_renew=True) return False, True return False, False 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")) except Exception as e: return {"valid": False, "error": f"Invalid certificate format: {e}"} seats = None license_id = None issuer_url = None try: ext_seats = cert.extensions.get_extension_for_oid(OID_LOGAR_SEATS) if ext_seats and ext_seats.value: raw_val = ext_seats.value.value if hasattr(ext_seats.value, "value") else str(ext_seats.value) if isinstance(raw_val, bytes): seats = int(raw_val.decode("utf-8").strip()) else: seats = int(str(raw_val).strip()) except (x509.ExtensionNotFound, ValueError, Exception): seats = None try: ext_lic = cert.extensions.get_extension_for_oid(OID_LOGAR_LICENSE_ID) if ext_lic and ext_lic.value: raw_val = ext_lic.value.value if hasattr(ext_lic.value, "value") else str(ext_lic.value) license_id = raw_val.decode("utf-8").strip() if isinstance(raw_val, bytes) else str(raw_val).strip() except (x509.ExtensionNotFound, Exception): license_id = None try: ext_url = cert.extensions.get_extension_for_oid(OID_LOGAR_ISSUER_URL) if ext_url and ext_url.value: raw_val = ext_url.value.value if hasattr(ext_url.value, "value") else str(ext_url.value) issuer_url = raw_val.decode("utf-8").strip() if isinstance(raw_val, bytes) else str(raw_val).strip() except (x509.ExtensionNotFound, Exception): issuer_url = None features = [] try: ext_feat = cert.extensions.get_extension_for_oid(OID_LOGAR_FEATURES) if ext_feat and ext_feat.value: raw_val = ext_feat.value.value if hasattr(ext_feat.value, "value") else str(ext_feat.value) feat_str = raw_val.decode("utf-8").strip() if isinstance(raw_val, bytes) else str(raw_val).strip() if feat_str.startswith("["): features = json.loads(feat_str) else: features = [f.strip() for f in feat_str.split(",") if f.strip()] except Exception: features = [] not_before = getattr(cert, "not_valid_before_utc", None) or cert.not_valid_before.replace(tzinfo=datetime.timezone.utc) not_after = getattr(cert, "not_valid_after_utc", None) or cert.not_valid_after.replace(tzinfo=datetime.timezone.utc) now = datetime.datetime.now(datetime.timezone.utc) is_expired = now > not_after days_remaining = max(0, (not_after - now).days) if not is_expired else 0 is_ca = False try: bc_ext = cert.extensions.get_extension_for_oid(x509.OID_BASIC_CONSTRAINTS) if bc_ext and bc_ext.value: is_ca = bc_ext.value.ca except Exception: pass subject_cn = "" for attr in cert.subject: if attr.oid == NameOID.COMMON_NAME: subject_cn = attr.value break issuer_cn = "" for attr in cert.issuer: if attr.oid == NameOID.COMMON_NAME: issuer_cn = attr.value 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, "seats": seats, "license_id": license_id, "issuer_url": issuer_url, "features": features, "subject_cn": subject_cn, "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(), "is_expired": is_expired, "days_remaining": days_remaining } def generate_ca_key_if_needed(cert_dir: str = "certs") -> Tuple[rsa.RSAPrivateKey, str]: """Ensures ca.key exists in cert_dir, generating a 4096-bit RSA key if not.""" os.makedirs(cert_dir, exist_ok=True) ca_key_path = os.path.join(cert_dir, "ca.key") if os.path.exists(ca_key_path): with open(ca_key_path, "r", encoding="utf-8") as f: key_pem = f.read() try: ca_key = serialization.load_pem_private_key(key_pem.encode("utf-8"), password=None) return ca_key, key_pem except Exception: pass ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096) key_pem = ca_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption() ).decode("utf-8") with open(ca_key_path, "w", encoding="utf-8") as f: f.write(key_pem) return ca_key, key_pem def generate_ca_csr( cert_dir: str = "certs", common_name: str = "LOGAR-Sub-CA", organization: str = "LOGAR", country: str = "AT" ) -> Tuple[str, str]: """ Generates a Certificate Signing Request (CSR) for a subordinate CA using ca.key. Returns (csr_pem, ca_key_pem). """ ca_key, key_pem = generate_ca_key_if_needed(cert_dir) subject = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, country), x509.NameAttribute(NameOID.ORGANIZATION_NAME, organization), x509.NameAttribute(NameOID.COMMON_NAME, common_name), ]) csr = ( x509.CertificateSigningRequestBuilder() .subject_name(subject) .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 ) .sign(ca_key, hashes.SHA256()) ) csr_pem = csr.public_bytes(serialization.Encoding.PEM).decode("utf-8") csr_path = os.path.join(cert_dir, "ca.csr") with open(csr_path, "w", encoding="utf-8") as f: f.write(csr_pem) return csr_pem, key_pem def create_license_request_lic( cert_dir: str = "certs", license_server: str = "https://logar.eibl.tech", server_fingerprint: str = "", hostname: str = "", requested_seats: int = 10, common_name: str = "LOGAR-Sub-CA" ) -> Tuple[Dict[str, Any], str]: """ Packages a CA CSR and server metadata into an exportable .lic file dictionary and JSON string. """ csr_pem, _ = generate_ca_csr(cert_dir=cert_dir, common_name=common_name) lic_data = { "format": "LOGAR_LICENSE_REQUEST", "version": "1.0", "license_server": license_server, "server_fingerprint": server_fingerprint, "hostname": hostname, "requested_seats": requested_seats, "created_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "csr": csr_pem } lic_json = json.dumps(lic_data, indent=2) lic_path = os.path.join(cert_dir, "logar_server.lic") with open(lic_path, "w", encoding="utf-8") as f: f.write(lic_json) return lic_data, lic_json def sign_ca_csr_with_seats( csr_pem: str, master_ca_cert_pem: str, master_ca_key_pem: str, seats: int, days_valid: int = 365, license_id: Optional[str] = None, license_server_url: str = "https://logar.eibl.tech", features: Optional[List[str]] = None ) -> str: """ Signs a subordinate CA CSR using the master license authority Root CA. Embeds the approved seat count, license ID, and optional enterprise features via custom X.509 extensions. Returns the signed PEM certificate. """ csr = x509.load_pem_x509_csr(csr_pem.encode("utf-8")) if not csr.is_signature_valid: raise ValueError("Invalid CSR signature") master_ca_cert = x509.load_pem_x509_certificate(master_ca_cert_pem.encode("utf-8")) master_ca_key = serialization.load_pem_private_key(master_ca_key_pem.encode("utf-8"), password=None) now = datetime.datetime.now(datetime.timezone.utc) if not license_id: license_id = f"LIC-{csr.public_key().public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo).hex()[:12].upper()}" builder = ( x509.CertificateBuilder() .subject_name(csr.subject) .issuer_name(master_ca_cert.subject) .public_key(csr.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(csr.public_key()), critical=False ) .add_extension( x509.AuthorityKeyIdentifier.from_issuer_public_key(master_ca_key.public_key()), critical=False ) .add_extension( x509.UnrecognizedExtension(OID_LOGAR_SEATS, str(seats).encode("utf-8")), critical=False ) .add_extension( x509.UnrecognizedExtension(OID_LOGAR_LICENSE_ID, license_id.encode("utf-8")), critical=False ) .add_extension( x509.UnrecognizedExtension(OID_LOGAR_ISSUER_URL, license_server_url.encode("utf-8")), critical=False ) ) if features: feat_str = json.dumps(features) if isinstance(features, list) else str(features) builder = builder.add_extension( x509.UnrecognizedExtension(OID_LOGAR_FEATURES, feat_str.encode("utf-8")), critical=False ) signed_cert = builder.sign(master_ca_key, hashes.SHA256()) return signed_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8") def enroll_and_sync_license_ws( cert_dir: str = "certs", license_id: Optional[str] = None, license_authority_ws_url: str = "wss://logar.eibl.tech/ws/license", server_fingerprint: str = "", hostname: str = "", comm_cert_pem: Optional[str] = None, master_root_ca_pem: Optional[str] = None, seats_taken: Optional[int] = None, timeout: int = 15, verify_tls: bool = True ) -> Tuple[bool, str, Optional[Dict[str, Any]]]: """ Connects to the Master License Authority via WebSocket protocol, authenticating and submitting the server Sub-CA CSR to receive a cryptographically signed Sub-CA license certificate with embedded seat quotas. """ os.makedirs(cert_dir, exist_ok=True) ca_key_path = os.path.join(cert_dir, "ca.key") ca_cert_path = os.path.join(cert_dir, "ca.crt") # 1. Load or generate client Sub-CA private key if os.path.exists(ca_key_path): with open(ca_key_path, "r", encoding="utf-8") as f: ca_key_pem = f.read() ca_key = serialization.load_pem_private_key(ca_key_pem.encode("utf-8"), password=None) else: ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096) ca_key_pem = ca_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption() ).decode("utf-8") with open(ca_key_path, "w", encoding="utf-8") as f: f.write(ca_key_pem) # 2. Build Sub-CA CSR csr_common_name = f"LOGAR-Server-CA-{license_id or 'Node'}" subject = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR Subordinate Node"), x509.NameAttribute(NameOID.COMMON_NAME, csr_common_name), ]) csr = ( x509.CertificateSigningRequestBuilder() .subject_name(subject) .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 ) .sign(ca_key, hashes.SHA256()) ) csr_pem = csr.public_bytes(serialization.Encoding.PEM).decode("utf-8") # 3. Connect via WebSocket try: from websockets.sync.client import connect as ws_connect except ImportError: return False, "websockets library not available", None ssl_context = None if license_authority_ws_url.startswith("wss://"): ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH) if not verify_tls: ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE elif master_root_ca_pem: ssl_context.load_verify_locations(cadata=master_root_ca_pem) else: upstream_path = os.path.join(cert_dir, "upstream_ca.crt") if os.path.exists(upstream_path): ssl_context.load_verify_locations(cafile=upstream_path) try: with ws_connect( license_authority_ws_url, ssl=ssl_context, open_timeout=timeout, close_timeout=timeout ) as ws: payload = { "action": "enroll_or_sync", "license_id": license_id, "communication_cert": comm_cert_pem, "server_csr": csr_pem, "server_fingerprint": server_fingerprint, "hostname": hostname } if seats_taken is not None: payload["seats_taken"] = seats_taken ws.send(json.dumps(payload)) raw_reply = ws.recv(timeout=timeout) reply = json.loads(raw_reply) if reply.get("status") != "success": return False, f"WebSocket licensing error: {reply.get('message', raw_reply)}", None new_ca_pem = reply.get("ca_cert") if not new_ca_pem: return False, "License response missing 'ca_cert'", None # Backup old ca.crt if present if os.path.exists(ca_cert_path): ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d_%H%M%S") try: os.replace(ca_cert_path, f"{ca_cert_path}.{ts}.bak") except Exception: pass with open(ca_cert_path, "w", encoding="utf-8") as f: f.write(new_ca_pem) upstream_ca = reply.get("master_root_ca") or master_root_ca_pem if upstream_ca: with open(os.path.join(cert_dir, "upstream_ca.crt"), "w", encoding="utf-8") as f: f.write(upstream_ca) lic_info = extract_license_info_from_cert(new_ca_pem) return True, f"License {license_id} successfully enrolled via WebSocket ({reply.get('seats')} seats)", { "license_id": license_id, "seats": reply.get("seats"), "ca_cert_pem": new_ca_pem, "master_root_ca_pem": upstream_ca, "license_info": lic_info } except Exception as e: return False, f"WebSocket enrollment failed: {str(e)}", None def import_license_package( lic_content_or_path: str, cert_dir: str = "certs", server_fingerprint: str = "", hostname: str = "" ) -> Dict[str, Any]: """ Imports and validates a license package or certificate. Supports: 1. Bootstrap JSON .lic package (auto-enrolls via WebSocket to receive signed CA cert) 2. Activated JSON .lic package containing 'ca_cert' and optional 'master_root_ca' 3. Raw PEM Subordinate CA certificate 4. File path pointing to any of the above Saves the signed CA certificate to cert_dir/ca.crt (backing up any old ca.crt), and if 'master_root_ca' is present, saves it to cert_dir/upstream_ca.crt for mTLS trust. Returns extracted license attributes and PEM contents. """ os.makedirs(cert_dir, exist_ok=True) raw = lic_content_or_path if os.path.exists(lic_content_or_path) and os.path.isfile(lic_content_or_path): with open(lic_content_or_path, "r", encoding="utf-8") as f: raw = f.read() ca_pem = None master_root_ca_pem = None trimmed = raw.strip() if trimmed.startswith("{"): try: data = json.loads(trimmed) except Exception as e: raise ValueError(f"Failed to parse JSON .lic package: {e}") # If this is a bootstrap license package, auto-enroll over WebSocket if data.get("format") == "LOGAR_BOOTSTRAP_LICENSE" or (data.get("communication_cert") and not data.get("ca_cert")): lic_id = data.get("license_id") ws_url = data.get("license_authority", "wss://logar.eibl.tech/ws/license") comm_cert = data.get("communication_cert") master_ca = data.get("master_root_ca") if master_ca: with open(os.path.join(cert_dir, "upstream_ca.crt"), "w", encoding="utf-8") as f: f.write(master_ca) ok, msg, res = enroll_and_sync_license_ws( cert_dir=cert_dir, license_id=lic_id, license_authority_ws_url=ws_url, server_fingerprint=server_fingerprint or data.get("server_fingerprint", ""), hostname=hostname or data.get("hostname", ""), comm_cert_pem=comm_cert, master_root_ca_pem=master_ca ) if not ok or not res: raise ValueError(msg) return res ca_pem = data.get("ca_cert") or data.get("cert_pem") master_root_ca_pem = data.get("master_root_ca") or data.get("upstream_ca") elif "-----BEGIN CERTIFICATE-----" in trimmed: ca_pem = trimmed if not ca_pem: raise ValueError("Provided license package does not contain a valid 'ca_cert' PEM certificate.") info = extract_license_info_from_cert(ca_pem) if not info.get("valid"): raise ValueError(f"Invalid CA certificate in license: {info.get('error')}") # Backup existing ca.crt if present ca_path = os.path.join(cert_dir, "ca.crt") if os.path.exists(ca_path): ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d_%H%M%S") try: os.replace(ca_path, f"{ca_path}.{ts}.bak") except Exception: pass with open(ca_path, "w", encoding="utf-8") as f: f.write(ca_pem) # Save upstream_ca.crt if master_root_ca is present if master_root_ca_pem: upstream_path = os.path.join(cert_dir, "upstream_ca.crt") with open(upstream_path, "w", encoding="utf-8") as f: f.write(master_root_ca_pem) return { "ca_cert_pem": ca_pem, "master_root_ca_pem": master_root_ca_pem, "license_info": info, "seats": info.get("seats"), "features": info.get("features", []), "license_id": info.get("license_id"), "issuer_url": info.get("issuer_url"), "fingerprint": info.get("fingerprint"), "expires_at": info.get("not_valid_after") } def renew_ca_cert_mtls( cert_dir: str = "certs", license_server_url: str = "https://logar.eibl.tech", upstream_ca_pem: Optional[str] = None, requested_seats: Optional[int] = None, seat_increment: int = 0, seats_taken: Optional[int] = None, active_seats: Optional[int] = None, hostname: Optional[str] = None, server_fingerprint: Optional[str] = None, timeout: int = 15, verify_tls: bool = True ) -> Tuple[bool, str, Optional[str]]: """ Connects to the master License Authority using mutual TLS (mTLS), authenticating with the server's current ca.crt and ca.key, and requests an automated renewal. Supports requesting seat quota increases or increments during renewal and reporting active seat usage. If renewed, writes backup of previous ca.crt and saves new ca.crt in cert_dir. Returns (success, message, new_ca_cert_pem). """ ca_cert_path = os.path.join(cert_dir, "ca.crt") ca_key_path = os.path.join(cert_dir, "ca.key") if not os.path.exists(ca_cert_path) or not os.path.exists(ca_key_path): return False, f"Missing ca.crt or ca.key in {cert_dir}", None renew_url = f"{license_server_url.rstrip('/')}/api/license/renew" parsed = urllib.parse.urlparse(renew_url) try: with open(ca_cert_path, "r", encoding="utf-8") as f: current_pem = f.read() info = extract_license_info_from_cert(current_pem) effective_seats_taken = seats_taken if seats_taken is not None else active_seats payload_dict = { "license_id": info.get("license_id"), "fingerprint": info.get("fingerprint"), "current_cert": current_pem, "seats_taken": effective_seats_taken, "active_seats": effective_seats_taken, "hostname": hostname or "", "server_fingerprint": server_fingerprint or "" } if requested_seats is not None: payload_dict["requested_seats"] = requested_seats if seat_increment: payload_dict["seat_increment"] = seat_increment payload = json.dumps(payload_dict).encode("utf-8") path = parsed.path or "/api/license/renew" headers = { "Content-Type": "application/json", "User-Agent": "LOGAR-Server-mTLS/2.0" } if parsed.scheme == "https": ssl_ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH) if not verify_tls: ssl_ctx.check_hostname = False ssl_ctx.verify_mode = ssl.CERT_NONE elif upstream_ca_pem: ssl_ctx.load_verify_locations(cadata=upstream_ca_pem) upstream_ca_path = os.path.join(cert_dir, "upstream_ca.crt") if os.path.exists(upstream_ca_path) and verify_tls and not upstream_ca_pem: ssl_ctx.load_verify_locations(cafile=upstream_ca_path) ssl_ctx.load_cert_chain(certfile=ca_cert_path, keyfile=ca_key_path) conn = http.client.HTTPSConnection(parsed.netloc, context=ssl_ctx, timeout=timeout) else: conn = http.client.HTTPConnection(parsed.netloc, timeout=timeout) conn.request("POST", path, body=payload, headers=headers) resp = conn.getresponse() resp_data = resp.read().decode("utf-8") conn.close() if resp.status != 200: return False, f"License server returned HTTP {resp.status}: {resp_data}", None result = json.loads(resp_data) new_cert_pem = result.get("ca_cert") if not new_cert_pem: return False, f"License server response missing 'ca_cert': {resp_data}", None # Backup old ca.crt ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d_%H%M%S") os.replace(ca_cert_path, f"{ca_cert_path}.{ts}.bak") # Write new ca.crt with open(ca_cert_path, "w", encoding="utf-8") as f: f.write(new_cert_pem) # Update upstream_ca.crt if master root CA certificate was provided master_root_pem = result.get("master_root_ca") if master_root_pem: upstream_path = os.path.join(cert_dir, "upstream_ca.crt") with open(upstream_path, "w", encoding="utf-8") as f: f.write(master_root_pem) return True, "Certificate renewed successfully via mTLS", new_cert_pem except Exception as e: return False, f"mTLS renewal exception: {str(e)}", None