refactor(licensing): delegate src/License_Authority.py to LOGAR-LICENSING submodule
This commit is contained in:
+21
-623
@@ -1,635 +1,33 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
"""
|
"""
|
||||||
LOGAR Master License Authority (logar.eibl.tech)
|
LOGAR Master License Authority Proxy
|
||||||
|
Delegates directly to the standalone submodule: LOGAR-LICENSING/License_Authority.py
|
||||||
Authoritative PKI service and CLI utility responsible for:
|
Repo: https://gitea.eibl.tech/me0nline/LOGAR-LICENSING.git
|
||||||
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 os
|
||||||
import sys
|
import sys
|
||||||
import json
|
|
||||||
import sqlite3
|
|
||||||
import argparse
|
|
||||||
import datetime
|
|
||||||
from typing import Tuple, Optional, Dict, Any
|
|
||||||
|
|
||||||
from cryptography import x509
|
# Prefer the submodule implementation
|
||||||
from cryptography.x509.oid import NameOID
|
ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||||||
from cryptography.hazmat.primitives import hashes, serialization
|
SUBMODULE_DIR = os.path.join(ROOT_DIR, "LOGAR-LICENSING")
|
||||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
if SUBMODULE_DIR not in sys.path:
|
||||||
|
sys.path.insert(0, SUBMODULE_DIR)
|
||||||
|
|
||||||
# Ensure src/ is importable
|
from License_Authority import (
|
||||||
SRC_DIR = os.path.dirname(os.path.abspath(__file__))
|
init_authority_db,
|
||||||
if SRC_DIR not in sys.path:
|
extract_license_info_from_cert,
|
||||||
sys.path.insert(0, SRC_DIR)
|
generate_master_root_ca,
|
||||||
|
sign_ca_csr_with_seats,
|
||||||
import server_enrollment as enrollment
|
issue_license_from_csr,
|
||||||
|
renew_license_cert,
|
||||||
try:
|
revoke_license,
|
||||||
from fastapi import FastAPI, HTTPException, Request, Header
|
list_licenses,
|
||||||
from fastapi.responses import HTMLResponse, JSONResponse
|
main,
|
||||||
import uvicorn
|
OID_LOGAR_SEATS,
|
||||||
FASTAPI_AVAILABLE = True
|
OID_LOGAR_LICENSE_ID,
|
||||||
except ImportError:
|
OID_LOGAR_ISSUER_URL
|
||||||
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__":
|
if __name__ == "__main__":
|
||||||
main()
|
main()
|
||||||
|
|||||||
Reference in New Issue
Block a user