1020 lines
38 KiB
Python
1020 lines
38 KiB
Python
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")
|
|
|
|
|
|
|
|
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 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 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"),
|
|
"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
|
|
|
|
|