diff --git a/src/server_enrollment.py b/src/server_enrollment.py index b709698..cdfc87f 100644 --- a/src/server_enrollment.py +++ b/src/server_enrollment.py @@ -1,12 +1,24 @@ import os +import json +import ssl import datetime import ipaddress -from typing import Tuple, List, Optional +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") + + def calculate_cert_fingerprint(cert_pem: str) -> str: """Computes SHA-256 fingerprint for a PEM-encoded X.509 certificate.""" @@ -378,3 +390,349 @@ def check_and_renew_hub_pki( return False, False + +def extract_license_info_from_cert(cert_pem: str) -> 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. + """ + 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 + + 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) + + return { + "valid": True, + "seats": seats, + "license_id": license_id, + "issuer_url": issuer_url, + "subject_cn": subject_cn, + "issuer_cn": issuer_cn, + "is_ca": is_ca, + "is_self_signed": is_self_signed, + "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" +) -> str: + """ + Signs a subordinate CA CSR using the master license authority Root CA. + Embeds the approved seat count and license ID 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 + ) + ) + + signed_cert = builder.sign(master_ca_key, hashes.SHA256()) + return signed_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8") + + +def renew_ca_cert_mtls( + cert_dir: str = "certs", + license_server_url: str = "https://logar.eibl.tech", + upstream_ca_pem: 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. + 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) + + payload = json.dumps({ + "license_id": info.get("license_id"), + "fingerprint": info.get("fingerprint"), + "current_cert": current_pem + }).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(cdata=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) + + return True, "Certificate renewed successfully via mTLS", new_cert_pem + + except Exception as e: + return False, f"mTLS renewal exception: {str(e)}", None + +