feat(licensing): add Master License Authority program with CSR signing, exact seat quotas, and mTLS renewals
This commit is contained in:
@@ -0,0 +1,635 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
LOGAR Master License Authority (logar.eibl.tech)
|
||||
|
||||
Authoritative PKI service and CLI utility responsible for:
|
||||
1. Maintaining the Master Root CA.
|
||||
2. Processing intermediate CA CSRs (or .lic request files) and issuing signed
|
||||
subordinate CA certificates embedding exact seat quotas via custom enterprise X.509 OIDs.
|
||||
3. Providing an automated mTLS renewal endpoint (/api/license/renew) for active servers.
|
||||
4. Serving an administrative WebUI and REST API to manage and revoke server licenses.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import sqlite3
|
||||
import argparse
|
||||
import datetime
|
||||
from typing import Tuple, Optional, Dict, Any
|
||||
|
||||
from cryptography import x509
|
||||
from cryptography.x509.oid import NameOID
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
|
||||
# Ensure src/ is importable
|
||||
SRC_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
if SRC_DIR not in sys.path:
|
||||
sys.path.insert(0, SRC_DIR)
|
||||
|
||||
import server_enrollment as enrollment
|
||||
|
||||
try:
|
||||
from fastapi import FastAPI, HTTPException, Request, Header
|
||||
from fastapi.responses import HTMLResponse, JSONResponse
|
||||
import uvicorn
|
||||
FASTAPI_AVAILABLE = True
|
||||
except ImportError:
|
||||
FASTAPI_AVAILABLE = False
|
||||
|
||||
|
||||
DEFAULT_DB = "license_authority.db"
|
||||
DEFAULT_MASTER_CERT_DIR = "master_ca_certs"
|
||||
|
||||
|
||||
def init_authority_db(db_path: str = DEFAULT_DB):
|
||||
"""Initializes the SQLite schema for tracking issued server licenses and revocations."""
|
||||
conn = sqlite3.connect(db_path)
|
||||
c = conn.cursor()
|
||||
c.execute("""
|
||||
CREATE TABLE IF NOT EXISTS licenses (
|
||||
license_id TEXT PRIMARY KEY,
|
||||
server_fingerprint TEXT,
|
||||
hostname TEXT,
|
||||
seats INTEGER NOT NULL,
|
||||
issued_at TEXT NOT NULL,
|
||||
expires_at TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'active',
|
||||
renewal_count INTEGER NOT NULL DEFAULT 0,
|
||||
cert_pem TEXT NOT NULL,
|
||||
csr_pem TEXT,
|
||||
client_cert_fingerprint TEXT
|
||||
)
|
||||
""")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
def generate_master_root_ca(
|
||||
cert_dir: str = DEFAULT_MASTER_CERT_DIR,
|
||||
common_name: str = "LOGAR-Master-Root-CA",
|
||||
organization: str = "LOGAR Global Licensing",
|
||||
country: str = "AT"
|
||||
) -> Tuple[str, str]:
|
||||
"""Generates the Master Root CA certificate and private key if not already existing."""
|
||||
os.makedirs(cert_dir, exist_ok=True)
|
||||
cert_path = os.path.join(cert_dir, "master_root_ca.crt")
|
||||
key_path = os.path.join(cert_dir, "master_root_ca.key")
|
||||
|
||||
if os.path.exists(cert_path) and os.path.exists(key_path):
|
||||
with open(cert_path, "r", encoding="utf-8") as f:
|
||||
cert_pem = f.read()
|
||||
with open(key_path, "r", encoding="utf-8") as f:
|
||||
key_pem = f.read()
|
||||
return cert_pem, key_pem
|
||||
|
||||
ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096)
|
||||
subject = issuer = x509.Name([
|
||||
x509.NameAttribute(NameOID.COUNTRY_NAME, country),
|
||||
x509.NameAttribute(NameOID.ORGANIZATION_NAME, organization),
|
||||
x509.NameAttribute(NameOID.COMMON_NAME, common_name),
|
||||
])
|
||||
|
||||
now = datetime.datetime.now(datetime.timezone.utc)
|
||||
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())
|
||||
)
|
||||
|
||||
cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
|
||||
key_pem = ca_key.private_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
||||
encryption_algorithm=serialization.NoEncryption()
|
||||
).decode("utf-8")
|
||||
|
||||
with open(cert_path, "w", encoding="utf-8") as f:
|
||||
f.write(cert_pem)
|
||||
with open(key_path, "w", encoding="utf-8") as f:
|
||||
f.write(key_pem)
|
||||
|
||||
return cert_pem, key_pem
|
||||
|
||||
|
||||
def issue_license_from_csr(
|
||||
csr_pem: str,
|
||||
seats: int,
|
||||
days_valid: int = 365,
|
||||
license_id: Optional[str] = None,
|
||||
server_fingerprint: str = "",
|
||||
hostname: str = "",
|
||||
db_path: str = DEFAULT_DB,
|
||||
cert_dir: str = DEFAULT_MASTER_CERT_DIR,
|
||||
license_server_url: str = "https://logar.eibl.tech"
|
||||
) -> Dict[str, Any]:
|
||||
"""Issues a subordinate CA certificate with the requested number of seats and registers it in the DB."""
|
||||
init_authority_db(db_path)
|
||||
master_cert_pem, master_key_pem = generate_master_root_ca(cert_dir)
|
||||
|
||||
if not license_id:
|
||||
csr_obj = x509.load_pem_x509_csr(csr_pem.encode("utf-8"))
|
||||
der_pub = csr_obj.public_key().public_bytes(
|
||||
serialization.Encoding.DER,
|
||||
serialization.PublicFormat.SubjectPublicKeyInfo
|
||||
)
|
||||
license_id = f"LIC-{der_pub.hex()[:12].upper()}"
|
||||
|
||||
ca_cert_pem = enrollment.sign_ca_csr_with_seats(
|
||||
csr_pem=csr_pem,
|
||||
master_ca_cert_pem=master_cert_pem,
|
||||
master_ca_key_pem=master_key_pem,
|
||||
seats=seats,
|
||||
days_valid=days_valid,
|
||||
license_id=license_id,
|
||||
license_server_url=license_server_url
|
||||
)
|
||||
|
||||
info = enrollment.extract_license_info_from_cert(ca_cert_pem)
|
||||
issued_at = info["not_valid_before"]
|
||||
expires_at = info["not_valid_after"]
|
||||
fp = info["fingerprint"]
|
||||
|
||||
conn = sqlite3.connect(db_path)
|
||||
c = conn.cursor()
|
||||
c.execute("""
|
||||
INSERT INTO licenses (license_id, server_fingerprint, hostname, seats, issued_at, expires_at, status, renewal_count, cert_pem, csr_pem, client_cert_fingerprint)
|
||||
VALUES (?, ?, ?, ?, ?, ?, 'active', 0, ?, ?, ?)
|
||||
ON CONFLICT(license_id) DO UPDATE SET
|
||||
seats = excluded.seats,
|
||||
expires_at = excluded.expires_at,
|
||||
status = 'active',
|
||||
cert_pem = excluded.cert_pem,
|
||||
client_cert_fingerprint = excluded.client_cert_fingerprint
|
||||
""", (license_id, server_fingerprint, hostname, seats, issued_at, expires_at, ca_cert_pem, csr_pem, fp))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"license_id": license_id,
|
||||
"seats": seats,
|
||||
"issued_at": issued_at,
|
||||
"expires_at": expires_at,
|
||||
"fingerprint": fp,
|
||||
"ca_cert": ca_cert_pem,
|
||||
"master_root_ca": master_cert_pem,
|
||||
"license_server": license_server_url
|
||||
}
|
||||
|
||||
|
||||
def renew_license_cert(
|
||||
current_cert_pem: str,
|
||||
days_valid: int = 365,
|
||||
db_path: str = DEFAULT_DB,
|
||||
cert_dir: str = DEFAULT_MASTER_CERT_DIR,
|
||||
license_server_url: str = "https://logar.eibl.tech"
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Validates an existing intermediate CA cert presenting for mTLS renewal,
|
||||
verifies that the license is active, and issues a renewed certificate preserving seat quota.
|
||||
"""
|
||||
init_authority_db(db_path)
|
||||
master_cert_pem, master_key_pem = generate_master_root_ca(cert_dir)
|
||||
master_cert = x509.load_pem_x509_certificate(master_cert_pem.encode("utf-8"))
|
||||
|
||||
current_cert = x509.load_pem_x509_certificate(current_cert_pem.encode("utf-8"))
|
||||
|
||||
# Extract license info
|
||||
info = enrollment.extract_license_info_from_cert(current_cert_pem)
|
||||
license_id = info.get("license_id")
|
||||
seats = info.get("seats") or 10
|
||||
|
||||
# Verify license record in database
|
||||
conn = sqlite3.connect(db_path)
|
||||
c = conn.cursor()
|
||||
c.execute("SELECT status, renewal_count, csr_pem, seats FROM licenses WHERE license_id = ?", (license_id,))
|
||||
row = c.fetchone()
|
||||
if row:
|
||||
status, renewals, csr_pem, db_seats = row
|
||||
if status == "revoked":
|
||||
conn.close()
|
||||
raise ValueError(f"License {license_id} is REVOKED and cannot be renewed.")
|
||||
seats = db_seats or seats
|
||||
else:
|
||||
# Auto-enroll license if missing from DB (signed by our master CA)
|
||||
renewals = 0
|
||||
csr_pem = None
|
||||
|
||||
# Build renewed certificate using the same public key and subject
|
||||
master_key = serialization.load_pem_private_key(master_key_pem.encode("utf-8"), password=None)
|
||||
now = datetime.datetime.now(datetime.timezone.utc)
|
||||
|
||||
builder = (
|
||||
x509.CertificateBuilder()
|
||||
.subject_name(current_cert.subject)
|
||||
.issuer_name(master_cert.subject)
|
||||
.public_key(current_cert.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(current_cert.public_key()),
|
||||
critical=False
|
||||
)
|
||||
.add_extension(
|
||||
x509.AuthorityKeyIdentifier.from_issuer_public_key(master_key.public_key()),
|
||||
critical=False
|
||||
)
|
||||
.add_extension(
|
||||
x509.UnrecognizedExtension(enrollment.OID_LOGAR_SEATS, str(seats).encode("utf-8")),
|
||||
critical=False
|
||||
)
|
||||
.add_extension(
|
||||
x509.UnrecognizedExtension(enrollment.OID_LOGAR_LICENSE_ID, license_id.encode("utf-8")),
|
||||
critical=False
|
||||
)
|
||||
.add_extension(
|
||||
x509.UnrecognizedExtension(enrollment.OID_LOGAR_ISSUER_URL, license_server_url.encode("utf-8")),
|
||||
critical=False
|
||||
)
|
||||
)
|
||||
|
||||
new_cert = builder.sign(master_key, hashes.SHA256())
|
||||
new_pem = new_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
|
||||
new_info = enrollment.extract_license_info_from_cert(new_pem)
|
||||
new_fp = new_info["fingerprint"]
|
||||
|
||||
c.execute("""
|
||||
UPDATE licenses
|
||||
SET renewal_count = renewal_count + 1,
|
||||
expires_at = ?,
|
||||
cert_pem = ?,
|
||||
client_cert_fingerprint = ?
|
||||
WHERE license_id = ?
|
||||
""", (new_info["not_valid_after"], new_pem, new_fp, license_id))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return {
|
||||
"status": "renewed",
|
||||
"license_id": license_id,
|
||||
"seats": seats,
|
||||
"ca_cert": new_pem,
|
||||
"expires_at": new_info["not_valid_after"],
|
||||
"fingerprint": new_fp
|
||||
}
|
||||
|
||||
|
||||
def revoke_license(license_id: str, db_path: str = DEFAULT_DB) -> bool:
|
||||
"""Marks a license as revoked in the database."""
|
||||
init_authority_db(db_path)
|
||||
conn = sqlite3.connect(db_path)
|
||||
c = conn.cursor()
|
||||
c.execute("UPDATE licenses SET status = 'revoked' WHERE license_id = ?", (license_id,))
|
||||
changed = c.rowcount > 0
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return changed
|
||||
|
||||
|
||||
def list_licenses(db_path: str = DEFAULT_DB) -> list:
|
||||
"""Lists all registered licenses."""
|
||||
init_authority_db(db_path)
|
||||
conn = sqlite3.connect(db_path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
c = conn.cursor()
|
||||
c.execute("SELECT license_id, server_fingerprint, hostname, seats, issued_at, expires_at, status, renewal_count, client_cert_fingerprint FROM licenses ORDER BY issued_at DESC")
|
||||
rows = [dict(r) for r in c.fetchall()]
|
||||
conn.close()
|
||||
return rows
|
||||
|
||||
|
||||
# --- FastAPI Web Service (for deployment at logar.eibl.tech) ---
|
||||
if FASTAPI_AVAILABLE:
|
||||
app = FastAPI(title="LOGAR License Authority", version="2.0.1")
|
||||
|
||||
@app.post("/api/license/issue")
|
||||
async def api_issue(request: Request):
|
||||
try:
|
||||
body = await request.json()
|
||||
except Exception:
|
||||
raise HTTPException(status_code=400, detail="Invalid JSON body")
|
||||
|
||||
csr_pem = body.get("csr")
|
||||
seats = int(body.get("seats", 10))
|
||||
server_fp = body.get("server_fingerprint", "")
|
||||
hostname = body.get("hostname", "")
|
||||
lic_id = body.get("license_id")
|
||||
|
||||
if not csr_pem:
|
||||
raise HTTPException(status_code=400, detail="Missing 'csr' in request")
|
||||
|
||||
try:
|
||||
result = issue_license_from_csr(
|
||||
csr_pem=csr_pem,
|
||||
seats=seats,
|
||||
license_id=lic_id,
|
||||
server_fingerprint=server_fp,
|
||||
hostname=hostname
|
||||
)
|
||||
return result
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
@app.post("/api/license/renew")
|
||||
async def api_renew(request: Request):
|
||||
try:
|
||||
body = await request.json()
|
||||
except Exception:
|
||||
raise HTTPException(status_code=400, detail="Invalid JSON body")
|
||||
|
||||
current_cert = body.get("current_cert")
|
||||
if not current_cert:
|
||||
raise HTTPException(status_code=400, detail="Missing 'current_cert' in request")
|
||||
|
||||
try:
|
||||
result = renew_license_cert(current_cert)
|
||||
return result
|
||||
except ValueError as ve:
|
||||
raise HTTPException(status_code=403, detail=str(ve))
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
@app.get("/api/licenses")
|
||||
def api_list():
|
||||
return {"licenses": list_licenses()}
|
||||
|
||||
@app.post("/api/license/revoke")
|
||||
async def api_revoke(request: Request):
|
||||
body = await request.json()
|
||||
lic_id = body.get("license_id")
|
||||
if not lic_id:
|
||||
raise HTTPException(status_code=400, detail="Missing 'license_id'")
|
||||
ok = revoke_license(lic_id)
|
||||
if not ok:
|
||||
raise HTTPException(status_code=404, detail="License not found")
|
||||
return {"status": "revoked", "license_id": lic_id}
|
||||
|
||||
@app.get("/", response_class=HTMLResponse)
|
||||
def web_dashboard():
|
||||
licenses = list_licenses()
|
||||
rows = []
|
||||
for l in licenses:
|
||||
badge_cls = "badge-active" if l["status"] == "active" else "badge-revoked"
|
||||
rows.append(f"""
|
||||
<tr>
|
||||
<td><code>{l['license_id']}</code></td>
|
||||
<td><strong>{l['seats']}</strong></td>
|
||||
<td><span class="badge {badge_cls}">{l['status'].upper()}</span></td>
|
||||
<td>{l['hostname'] or 'N/A'}</td>
|
||||
<td><code>{l['client_cert_fingerprint'][:16]}...</code></td>
|
||||
<td>{l['expires_at'][:10]}</td>
|
||||
<td>{l['renewal_count']}</td>
|
||||
<td>
|
||||
<button class="btn btn-danger" onclick="revokeLic('{l['license_id']}')">Revoke</button>
|
||||
</td>
|
||||
</tr>
|
||||
""")
|
||||
rows_html = "".join(rows) if rows else "<tr><td colspan='8' style='text-align: center; color: #64748b;'>No licenses issued yet.</td></tr>"
|
||||
|
||||
html_tpl = """<!DOCTYPE html>
|
||||
<html>
|
||||
<head>
|
||||
<title>LOGAR Master License Authority (logar.eibl.tech)</title>
|
||||
<meta charset="utf-8">
|
||||
<style>
|
||||
body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; background: #0f172a; color: #f8fafc; margin: 0; padding: 2rem; }
|
||||
.container { max-width: 1100px; margin: 0 auto; }
|
||||
h1 { color: #38bdf8; display: flex; align-items: center; gap: 10px; }
|
||||
.card { background: #1e293b; border-radius: 8px; padding: 1.5rem; margin-bottom: 2rem; border: 1px solid #334155; }
|
||||
table { width: 100%; border-collapse: collapse; margin-top: 1rem; }
|
||||
th, td { text-align: left; padding: 0.75rem; border-bottom: 1px solid #334155; font-size: 0.9rem; }
|
||||
th { color: #94a3b8; font-weight: 600; text-transform: uppercase; font-size: 0.75rem; }
|
||||
.badge { padding: 0.25rem 0.5rem; border-radius: 4px; font-weight: bold; font-size: 0.75rem; }
|
||||
.badge-active { background: #065f46; color: #34d399; }
|
||||
.badge-revoked { background: #7f1d1d; color: #f87171; }
|
||||
code { background: #0f172a; padding: 2px 6px; border-radius: 4px; font-family: monospace; color: #a5b4fc; }
|
||||
.btn { padding: 0.4rem 0.8rem; border: none; border-radius: 4px; cursor: pointer; font-weight: 600; }
|
||||
.btn-primary { background: #0284c7; color: white; }
|
||||
.btn-danger { background: #dc2626; color: white; }
|
||||
textarea, input { background: #0f172a; border: 1px solid #475569; color: white; padding: 0.5rem; border-radius: 4px; width: 100%; box-sizing: border-box; }
|
||||
.form-group { margin-bottom: 1rem; }
|
||||
label { display: block; margin-bottom: 0.5rem; color: #cbd5e1; font-size: 0.85rem; }
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<div class="container">
|
||||
<h1><span>🛡️</span> LOGAR Master License Authority</h1>
|
||||
<p style="color: #94a3b8;">Authoritative PKI for issuing seat-bearing Intermediate CA certificates and automatic mTLS renewals.</p>
|
||||
|
||||
<div class="card">
|
||||
<h3>Issue Subordinate CA Certificate from CSR / .lic</h3>
|
||||
<div class="form-group">
|
||||
<label>Seats Quota (Exact Seat Limit)</label>
|
||||
<input type="number" id="seats" value="25" min="1" max="10000" style="max-width: 200px;">
|
||||
</div>
|
||||
<div class="form-group">
|
||||
<label>CSR (PEM) or .lic JSON content</label>
|
||||
<textarea id="csrText" rows="6" placeholder="-----BEGIN CERTIFICATE REQUEST----- ... or JSON .lic"></textarea>
|
||||
</div>
|
||||
<button class="btn btn-primary" onclick="issueLicense()">Sign & Issue CA Certificate</button>
|
||||
<div id="issueOutput" style="margin-top: 1rem; display: none;"></div>
|
||||
</div>
|
||||
|
||||
<div class="card">
|
||||
<h3>Active & Issued Server Licenses</h3>
|
||||
<table>
|
||||
<thead>
|
||||
<tr>
|
||||
<th>License ID</th>
|
||||
<th>Seats</th>
|
||||
<th>Status</th>
|
||||
<th>Host</th>
|
||||
<th>Cert Fingerprint</th>
|
||||
<th>Expires</th>
|
||||
<th>Renewals</th>
|
||||
<th>Actions</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
__ROWS_HTML__
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<script>
|
||||
async function issueLicense() {
|
||||
let raw = document.getElementById('csrText').value.trim();
|
||||
const seats = parseInt(document.getElementById('seats').value, 10);
|
||||
let csr = raw;
|
||||
let fp = "";
|
||||
let host = "";
|
||||
|
||||
if (raw.startsWith('{')) {
|
||||
try {
|
||||
const parsed = JSON.parse(raw);
|
||||
csr = parsed.csr;
|
||||
fp = parsed.server_fingerprint || "";
|
||||
host = parsed.hostname || "";
|
||||
} catch(e) {
|
||||
alert("Invalid JSON .lic file format");
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
const resp = await fetch('/api/license/issue', {
|
||||
method: 'POST',
|
||||
headers: {'Content-Type': 'application/json'},
|
||||
body: JSON.stringify({ csr: csr, seats: seats, server_fingerprint: fp, hostname: host })
|
||||
});
|
||||
const data = await resp.json();
|
||||
const out = document.getElementById('issueOutput');
|
||||
out.style.display = 'block';
|
||||
if (resp.ok) {
|
||||
out.innerHTML = `
|
||||
<div style="background: #064e3b; padding: 1rem; border-radius: 4px; border: 1px solid #059669;">
|
||||
<h4 style="margin: 0 0 0.5rem 0; color: #34d399;">License Issued Successfully! (Seats: ${data.seats})</h4>
|
||||
<p style="margin: 0; font-size: 0.85rem;">License ID: <code>${data.license_id}</code></p>
|
||||
<p style="margin: 0.5rem 0 0 0; font-size: 0.85rem;">Copy the signed CA certificate below or download it as <code>ca.crt</code>:</p>
|
||||
<textarea rows="6" readonly style="margin-top: 0.5rem;">${data.ca_cert}</textarea>
|
||||
</div>
|
||||
`;
|
||||
} else {
|
||||
out.innerHTML = `<div style="background: #7f1d1d; padding: 1rem; border-radius: 4px;">Error: ${data.detail}</div>`;
|
||||
}
|
||||
}
|
||||
|
||||
async function revokeLic(licId) {
|
||||
if (!confirm(`Are you sure you want to revoke license ${licId}?`)) return;
|
||||
const resp = await fetch('/api/license/revoke', {
|
||||
method: 'POST',
|
||||
headers: {'Content-Type': 'application/json'},
|
||||
body: JSON.stringify({ license_id: licId })
|
||||
});
|
||||
if (resp.ok) {
|
||||
location.reload();
|
||||
} else {
|
||||
alert("Failed to revoke license");
|
||||
}
|
||||
}
|
||||
</script>
|
||||
</body>
|
||||
</html>"""
|
||||
return html_tpl.replace("__ROWS_HTML__", rows_html)
|
||||
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="LOGAR Master License Authority (logar.eibl.tech)")
|
||||
parser.add_argument("--init-master-ca", action="store_true", help="Initialize the Master Root CA")
|
||||
parser.add_argument("--sign-lic", type=str, help="Path to .lic file or CSR PEM file to sign")
|
||||
parser.add_argument("--seats", type=int, default=25, help="Number of seats to grant (default: 25)")
|
||||
parser.add_argument("--out", type=str, default="ca.crt", help="Output path for signed CA cert")
|
||||
parser.add_argument("--revoke", type=str, help="License ID to revoke")
|
||||
parser.add_argument("--list", action="store_true", help="List all issued licenses")
|
||||
parser.add_argument("--serve", action="store_true", help="Run the FastAPI / mTLS renewal service")
|
||||
parser.add_argument("--host", type=str, default="0.0.0.0", help="Service host (default: 0.0.0.0)")
|
||||
parser.add_argument("--port", type=int, default=8080, help="Service port (default: 8080)")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.init_master_ca:
|
||||
generate_master_root_ca()
|
||||
print(f"[+] Master Root CA initialized in {DEFAULT_MASTER_CERT_DIR}/")
|
||||
return
|
||||
|
||||
if args.list:
|
||||
lics = list_licenses()
|
||||
print(f"\n{'License ID':<20} {'Seats':<8} {'Status':<10} {'Host':<20} {'Expires':<12} {'Renewals':<8}")
|
||||
print("-" * 80)
|
||||
for l in lics:
|
||||
print(f"{l['license_id']:<20} {l['seats']:<8} {l['status']:<10} {l['hostname'] or 'N/A':<20} {l['expires_at'][:10]:<12} {l['renewal_count']:<8}")
|
||||
return
|
||||
|
||||
if args.revoke:
|
||||
if revoke_license(args.revoke):
|
||||
print(f"[+] License {args.revoke} revoked successfully.")
|
||||
else:
|
||||
print(f"[-] License {args.revoke} not found.")
|
||||
return
|
||||
|
||||
if args.sign_lic:
|
||||
with open(args.sign_lic, "r", encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
|
||||
csr_pem = content
|
||||
server_fp = ""
|
||||
hostname = ""
|
||||
if content.strip().startswith("{"):
|
||||
data = json.loads(content)
|
||||
csr_pem = data["csr"]
|
||||
server_fp = data.get("server_fingerprint", "")
|
||||
hostname = data.get("hostname", "")
|
||||
|
||||
res = issue_license_from_csr(
|
||||
csr_pem=csr_pem,
|
||||
seats=args.seats,
|
||||
server_fingerprint=server_fp,
|
||||
hostname=hostname
|
||||
)
|
||||
with open(args.out, "w", encoding="utf-8") as f:
|
||||
f.write(res["ca_cert"])
|
||||
|
||||
print(f"[+] Signed CA certificate issued successfully!")
|
||||
print(f" - License ID: {res['license_id']}")
|
||||
print(f" - Granted Seats: {res['seats']}")
|
||||
print(f" - Expires: {res['expires_at']}")
|
||||
print(f" - Saved to: {args.out}")
|
||||
return
|
||||
|
||||
if args.serve:
|
||||
if not FASTAPI_AVAILABLE:
|
||||
print("[-] FastAPI/uvicorn not installed.")
|
||||
sys.exit(1)
|
||||
generate_master_root_ca()
|
||||
print(f"[*] Starting LOGAR License Authority at http://{args.host}:{args.port}")
|
||||
uvicorn.run(app, host=args.host, port=args.port)
|
||||
return
|
||||
|
||||
parser.print_help()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user