feat(enrollment): add import_license_package supporting .lic files and seat renewal expansion

src/server_enrollment.py

@@ -654,16 +654,90 @@ def sign_ca_csr_with_seats(
     return signed_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
 
 
+def import_license_package(
+    lic_content_or_path: str,
+    cert_dir: str = "certs"
+) -> Dict[str, Any]:
+    """
+    Imports and validates a license package or certificate.
+    Supports:
+      1. Activated JSON .lic package containing 'ca_cert' and optional 'master_root_ca'
+      2. Raw PEM Subordinate CA certificate
+      3. File path pointing to either 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)
+            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")
+        except Exception as e:
+            raise ValueError(f"Failed to parse JSON .lic package: {e}")
+    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"),
+        "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,
     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.
     If renewed, writes backup of previous ca.crt and saves new ca.crt in cert_dir.
     Returns (success, message, new_ca_cert_pem).
     """
@@ -681,11 +755,17 @@ def renew_ca_cert_mtls(
             current_pem = f.read()
         info = extract_license_info_from_cert(current_pem)
 
-        payload = json.dumps({
+        payload_dict = {
             "license_id": info.get("license_id"),
             "fingerprint": info.get("fingerprint"),
             "current_cert": current_pem
-        }).encode("utf-8")
+        }
+        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 = {
@@ -730,6 +810,13 @@ def renew_ca_cert_mtls(
         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: