me0nline/LOGAR
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

feat(pki): add CA CSR generation, seat quota OID extensions, and mTLS renewal support

Browse Source
This commit is contained in:
me0nline
2026-09-24 16:40:54 +02:00
parent f5ff8ab6cc
commit f605375639
1 changed files with 359 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server_enrollment.py
+359 -1
View File
@@ -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