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

feat(server): add mTLS listener, client licensing schema, and enrollment endpoint in src/Server.py

Browse Source
This commit is contained in:
me0nline
2026-09-04 20:53:48 +02:00
parent e83a5b3e0f
commit 491d2b1194
1 changed files with 296 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server.py
+296 -61
View File
@@ -23,9 +23,16 @@ from pgpy.constants import (
SymmetricKeyAlgorithm,
CompressionAlgorithm
)
import ssl
from pydantic import BaseModel
from fastapi import FastAPI, HTTPException
import uvicorn
try:
from src import server_enrollment as enrollment
except ImportError:
import server_enrollment as enrollment
CONFIG_FILE_NAME = "server_config.json"
DEFAULT_DB_FILE = "logar_state.db"
EVALUATION_WINDOW_HOURS = 12
@@ -37,6 +44,13 @@ app = FastAPI(title="LOGAR Cloud Ingestion & Hermes Hub", version="2.0.0")
SERVER_STATE: Dict[str, Any] = {}
class ClientEnrollRequest(BaseModel):
client_id: str
hostname: str
os: str
enrollment_secret: str
def generate_server_keypair(server_name: str):
"""Generates an OpenPGP RSA 2048 key with encryption capability."""
key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 2048)
@@ -60,12 +74,21 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
print(f"[*] Loading server configuration from: {os.path.abspath(config_path)}")
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
if "enrollment_secret" not in config:
config["enrollment_secret"] = secrets.token_hex(24)
if "max_seats" not in config:
config["max_seats"] = 10
if "cert_dir" not in config:
config["cert_dir"] = "certs"
if "tls_enabled" not in config:
config["tls_enabled"] = True
return config
print(f"[!] Config '{config_path}' not found. Initializing first-run configuration...")
server_name = "LOGAR-Cloud-Hub"
private_key, public_key, fingerprint = generate_server_keypair(server_name)
auth_token = secrets.token_hex(24)
enrollment_secret = secrets.token_hex(24)
config = {
"server_name": server_name,
@@ -74,6 +97,10 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
"hermes_host": "0.0.0.0",
"hermes_port": 8443,
"auth_token": auth_token,
"enrollment_secret": enrollment_secret,
"max_seats": 10,
"cert_dir": "certs",
"tls_enabled": True,
"db_path": DEFAULT_DB_FILE,
"evaluation_window_hours": EVALUATION_WINDOW_HOURS,
"min_persistence_runs": RUN_THRESHOLD,
@@ -95,7 +122,9 @@ def create_client_config(
server_host: str,
server_port: int,
output_path: str,
config_path: str = CONFIG_FILE_NAME
config_path: str = CONFIG_FILE_NAME,
hermes_host: Optional[str] = None,
hermes_port: Optional[int] = None
) -> Dict[str, Any]:
"""Creates a client configuration file containing the server address, auth token, and encryption-only key/fingerprint."""
server_conf = load_or_init_config(config_path)
@@ -103,6 +132,10 @@ def create_client_config(
client_conf = {
"server_host": server_host,
"server_port": server_port,
"hermes_host": hermes_host or server_conf.get("hermes_host", "127.0.0.1"),
"hermes_port": hermes_port or server_conf.get("hermes_port", 8443),
"enrollment_secret": server_conf.get("enrollment_secret"),
"cert_dir": "certs",
"server_fingerprint": server_conf["server_fingerprint"],
"server_public_key": server_conf["public_key"],
"auth_token": server_conf["auth_token"]
@@ -121,8 +154,8 @@ def create_client_config(
return client_conf
def init_db(db_path: str):
"""Initializes the SQLite schema for multi-run temporal tracking."""
def init_db(db_path: str, enrollment_secret: Optional[str] = None, max_seats: int = 10):
"""Initializes the SQLite schema for multi-run temporal tracking, client tracking, and license quota."""
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS active_issues (
@@ -149,6 +182,29 @@ def init_db(db_path: str):
log_count INTEGER
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS license_config (
id INTEGER PRIMARY KEY CHECK (id = 1),
max_seats INTEGER NOT NULL DEFAULT 10,
enrollment_secret TEXT NOT NULL
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS clients (
client_id TEXT PRIMARY KEY,
hostname TEXT NOT NULL,
os_type TEXT NOT NULL,
cert_fingerprint TEXT NOT NULL,
status TEXT DEFAULT 'active',
first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
if enrollment_secret:
conn.execute("""
INSERT OR IGNORE INTO license_config (id, max_seats, enrollment_secret)
VALUES (1, ?, ?)
""", (max_seats, enrollment_secret))
conn.commit()
conn.close()
@@ -252,15 +308,60 @@ def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: i
}
def init_mtls_server_context(cert_dir: str = "certs") -> ssl.SSLContext:
"""Initializes TLS 1.3 server SSLContext with client certificate requirement (mTLS)."""
ca_file = os.path.join(cert_dir, "ca.crt")
srv_cert = os.path.join(cert_dir, "server.crt")
srv_key = os.path.join(cert_dir, "server.key")
ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ctx.load_cert_chain(certfile=srv_cert, keyfile=srv_key)
ctx.load_verify_locations(cafile=ca_file)
ctx.verify_mode = ssl.CERT_REQUIRED
ctx.minimum_version = ssl.TLSVersion.TLSv1_3
return ctx
async def handle_socket_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
"""
Authenticated TCP socket handler.
Protocol:
- 4-byte big-endian prefix: payload length
- Payload: JSON with auth_token and encrypted_payload (OpenPGP ASCII armored)
- Response: 4-byte length + JSON confirmation
mTLS TCP socket handler.
Extracts client CN (client_id) from the TLS handshake,
validates active license status in SQLite, updates last_seen,
reads 4-byte big-endian length-prefixed JSON payload,
and ingests candidate logs into the temporal evaluation engine.
"""
addr = writer.get_extra_info("peername")
client_id = None
ssl_obj = writer.get_extra_info("ssl_object")
if ssl_obj:
peercert = ssl_obj.getpeercert()
if peercert and "subject" in peercert:
for rdn in peercert["subject"]:
for key, val in rdn:
if key == "commonName":
client_id = val
break
# If mTLS is enforced, verify client in accounting database
if SERVER_STATE.get("tls_enabled", False):
if not client_id:
writer.close()
await writer.wait_closed()
return
db_path = SERVER_STATE["config"]["db_path"]
conn = sqlite3.connect(db_path)
c = conn.cursor()
c.execute("SELECT status FROM clients WHERE client_id = ?", (client_id,))
row = c.fetchone()
if not row or row[0] != "active":
conn.close()
writer.close()
await writer.wait_closed()
return
c.execute("UPDATE clients SET last_seen = CURRENT_TIMESTAMP WHERE client_id = ?", (client_id,))
conn.commit()
conn.close()
try:
# Read 4-byte length prefix
length_bytes = await reader.readexactly(4)
@@ -269,26 +370,20 @@ async def handle_socket_client(reader: asyncio.StreamReader, writer: asyncio.Str
raise ValueError(f"Invalid frame size: {length}")
payload_bytes = await reader.readexactly(length)
envelope = json.loads(payload_bytes.decode("utf-8"))
raw_payload = json.loads(payload_bytes.decode("utf-8"))
# Authenticate socket client
expected_token = SERVER_STATE["config"]["auth_token"]
provided_token = envelope.get("auth_token")
if not secrets.compare_digest(str(provided_token), str(expected_token)):
err_msg = json.dumps({"status": "error", "message": "Authentication failed"}).encode("utf-8")
writer.write(struct.pack(">I", len(err_msg)) + err_msg)
await writer.drain()
writer.close()
await writer.wait_closed()
return
# Support both direct JSON payload over mTLS and legacy OpenPGP envelope
if "encrypted_payload" in raw_payload and SERVER_STATE.get("private_key_obj"):
pgp_msg = pgpy.PGPMessage.from_blob(raw_payload["encrypted_payload"])
priv_key = SERVER_STATE["private_key_obj"]
decrypted_obj = priv_key.decrypt(pgp_msg)
log_payload = json.loads(decrypted_obj.message)
else:
log_payload = raw_payload
# Decrypt payload using server's OpenPGP private key
encrypted_armored = envelope.get("encrypted_payload", "")
pgp_msg = pgpy.PGPMessage.from_blob(encrypted_armored)
priv_key = SERVER_STATE["private_key_obj"]
decrypted_obj = priv_key.decrypt(pgp_msg)
decrypted_json_str = decrypted_obj.message
log_payload = json.loads(decrypted_json_str)
# Attach authenticated client_id if not present
if client_id and "server" not in log_payload:
log_payload["server"] = client_id
# Ingest and apply 12h window / 4-run rule
res = process_ingested_logs(
@@ -317,47 +412,158 @@ async def handle_socket_client(reader: asyncio.StreamReader, writer: asyncio.Str
pass
@app.get("/api/hermes/report")
def get_hermes_report():
@app.post("/api/client/enroll")
def enroll_client(req: ClientEnrollRequest):
"""
Agentic Integration endpoint: Consumed by Hermes to fetch anomalies that have persisted
across the 12-hour evaluation window and satisfied the 4-run rule.
Enrolls an edge client by validating the enrollment secret,
checking license seat limits, issuing a signed client certificate + key,
and recording the client in the SQLite accounting database.
"""
db_path = SERVER_STATE["config"]["db_path"]
window_hours = SERVER_STATE["config"]["evaluation_window_hours"]
min_runs = SERVER_STATE["config"]["min_persistence_runs"]
now = datetime.now(timezone.utc)
conn = sqlite3.connect(db_path)
c = conn.cursor()
# 1. Validate enrollment secret against license_config
c.execute("SELECT enrollment_secret, max_seats FROM license_config WHERE id = 1")
row = c.fetchone()
if not row:
conn.close()
raise HTTPException(status_code=500, detail="License configuration not initialized")
expected_secret, max_seats = row
if not secrets.compare_digest(str(req.enrollment_secret), str(expected_secret)):
conn.close()
raise HTTPException(status_code=403, detail="Invalid enrollment secret")
ca_cert = SERVER_STATE.get("ca_cert")
ca_key = SERVER_STATE.get("ca_key")
ca_cert_pem = SERVER_STATE.get("ca_cert_pem")
if not ca_cert or not ca_key:
conn.close()
raise HTTPException(status_code=500, detail="Root CA not loaded on server")
# 2. Check if client_id already registered
c.execute("SELECT status FROM clients WHERE client_id = ?", (req.client_id,))
client_row = c.fetchone()
if client_row:
if client_row[0] == "revoked":
conn.close()
raise HTTPException(status_code=403, detail="Client certificate has been revoked")
# Re-issue for existing active client
client_cert_pem, client_key_pem = enrollment.issue_client_cert(req.client_id, ca_cert, ca_key)
fp = enrollment.calculate_cert_fingerprint(client_cert_pem)
c.execute("""
UPDATE clients
SET hostname = ?, os_type = ?, cert_fingerprint = ?, last_seen = CURRENT_TIMESTAMP
WHERE client_id = ?
""", (req.hostname, req.os, fp, req.client_id))
conn.commit()
conn.close()
print(f"[+] Re-enrolled active client: {req.client_id} ({req.hostname})")
return {
"ca_cert": ca_cert_pem,
"client_cert": client_cert_pem,
"client_key": client_key_pem
}
# 3. New client: check seat limits
c.execute("SELECT COUNT(*) FROM clients WHERE status = 'active'")
active_count = c.fetchone()[0]
if active_count >= max_seats:
conn.close()
raise HTTPException(status_code=403, detail="License seat limit reached")
# 4. Issue signed cert + key
client_cert_pem, client_key_pem = enrollment.issue_client_cert(req.client_id, ca_cert, ca_key)
fp = enrollment.calculate_cert_fingerprint(client_cert_pem)
c.execute("""
INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status)
VALUES (?, ?, ?, ?, 'active')
""", (req.client_id, req.hostname, req.os, fp))
conn.commit()
conn.close()
print(f"[+] Successfully enrolled new client: {req.client_id} ({req.hostname}) [Seats: {active_count + 1}/{max_seats}]")
return {
"ca_cert": ca_cert_pem,
"client_cert": client_cert_pem,
"client_key": client_key_pem
}
@app.get("/api/clients")
def list_clients():
"""Returns all registered clients and license seat usage."""
db_path = SERVER_STATE["config"]["db_path"]
conn = sqlite3.connect(db_path)
c = conn.cursor()
c.execute("SELECT max_seats FROM license_config WHERE id = 1")
lic_row = c.fetchone()
max_seats = lic_row[0] if lic_row else 10
c.execute("SELECT client_id, hostname, os_type, cert_fingerprint, status, first_seen, last_seen FROM clients")
rows = c.fetchall()
conn.close()
clients = [
{
"client_id": r[0],
"hostname": r[1],
"os_type": r[2],
"cert_fingerprint": r[3],
"status": r[4],
"first_seen": r[5],
"last_seen": r[6]
}
for r in rows
]
active_count = sum(1 for cl in clients if cl["status"] == "active")
return {
"active_seats": active_count,
"max_seats": max_seats,
"clients": clients
}
@app.get("/api/hermes/report")
def get_verified_anomalies_for_hermes():
"""
Ingestion endpoint for Hermes agentic workflows.
Returns only verified anomalies that have satisfied the 4-run persistence rule
within the active 12-hour evaluation window. Transient blips (< 4 runs) are excluded.
"""
db_path = SERVER_STATE["config"]["db_path"]
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status
FROM active_issues
WHERE status = 'VERIFIED'
ORDER BY last_seen DESC
""")
rows = cursor.fetchall()
conn.close()
report = []
for r in rows:
last_seen_dt = datetime.fromisoformat(r[8])
# Only return anomalies active within the evaluation window
if (now - last_seen_dt) <= timedelta(hours=window_hours):
report.append({
"fingerprint": r[0],
"site": r[1],
"server": r[2],
"signature": r[3],
"severity": r[4],
"message": r[5],
"os_type": r[6],
"first_seen": r[7],
"last_seen": r[8],
"consecutive_runs": r[9],
"evaluation_window": f"{window_hours}h",
"verified": True,
"status": r[10]
})
report.append({
"fingerprint": r[0],
"site": r[1],
"server": r[2],
"signature": r[3],
"severity": r[4],
"message": r[5],
"os_type": r[6],
"first_seen": r[7],
"last_seen": r[8],
"consecutive_runs": r[9],
"evaluation_window": f"{SERVER_STATE['config']['evaluation_window_hours']}h",
"verified": True,
"status": r[10]
})
return report
@@ -400,26 +606,30 @@ def health_check():
"server_name": SERVER_STATE["config"]["server_name"],
"fingerprint": SERVER_STATE["config"]["server_fingerprint"],
"tcp_port": SERVER_STATE["config"]["tcp_port"],
"hermes_port": SERVER_STATE["config"]["hermes_port"]
"hermes_port": SERVER_STATE["config"]["hermes_port"],
"tls_enabled": SERVER_STATE.get("tls_enabled", False)
}
async def run_server():
"""Runs the TCP socket listener and the Hermes REST API concurrently."""
"""Runs the mTLS TCP socket listener and the Hermes REST API concurrently."""
config = SERVER_STATE["config"]
tcp_host = config["tcp_host"]
tcp_port = int(config["tcp_port"])
hermes_host = config["hermes_host"]
hermes_port = int(config["hermes_port"])
ssl_ctx = SERVER_STATE.get("ssl_ctx")
# Start TCP Socket Server
tcp_server = await asyncio.start_server(handle_socket_client, tcp_host, tcp_port)
print(f"[*] LOGAR TCP Socket Server listening on {tcp_host}:{tcp_port}")
# Start mTLS / TCP Socket Server
tcp_server = await asyncio.start_server(handle_socket_client, tcp_host, tcp_port, ssl=ssl_ctx)
mode_str = "mTLS TLSv1.3" if ssl_ctx else "Plain TCP"
print(f"[*] LOGAR {mode_str} Socket Server listening on {tcp_host}:{tcp_port}")
# Start FastAPI / Uvicorn server for Hermes
# Start FastAPI / Uvicorn server for Hermes & Enrollment
uv_config = uvicorn.Config(app, host=hermes_host, port=hermes_port, log_level="warning")
uv_server = uvicorn.Server(uv_config)
print(f"[*] Hermes Reporting API available at http://{hermes_host}:{hermes_port}/api/hermes/report")
print(f"[*] Client Enrollment API available at http://{hermes_host}:{hermes_port}/api/client/enroll")
await asyncio.gather(
tcp_server.serve_forever(),
@@ -437,12 +647,35 @@ def main():
args = parser.parse_args()
config = load_or_init_config(args.config)
init_db(config["db_path"])
init_db(
config["db_path"],
enrollment_secret=config.get("enrollment_secret"),
max_seats=config.get("max_seats", 10)
)
# Load OpenPGP private key into memory
# Initialize dynamic PKI (Root CA and Server TLS Certificate)
cert_dir = config.get("cert_dir", "certs")
ca_cert, ca_key, ca_pem, ca_key_pem = enrollment.generate_ca_if_needed(cert_dir=cert_dir)
srv_cert, srv_key, srv_pem, srv_key_pem = enrollment.generate_server_cert_if_needed(
ca_cert, ca_key,
hostnames=[config.get("tcp_host"), "127.0.0.1", "localhost"],
cert_dir=cert_dir
)
# Initialize mTLS SSLContext if enabled
ssl_ctx = None
if config.get("tls_enabled", True):
ssl_ctx = init_mtls_server_context(cert_dir=cert_dir)
# Load OpenPGP private key into memory (legacy fallback)
priv_key_obj, _ = pgpy.PGPKey.from_blob(config["private_key"])
SERVER_STATE["config"] = config
SERVER_STATE["private_key_obj"] = priv_key_obj
SERVER_STATE["ca_cert"] = ca_cert
SERVER_STATE["ca_key"] = ca_key
SERVER_STATE["ca_cert_pem"] = ca_pem
SERVER_STATE["ssl_ctx"] = ssl_ctx
SERVER_STATE["tls_enabled"] = config.get("tls_enabled", True)
if args.create_client_config:
port = args.server_port or config["tcp_port"]
@@ -456,7 +689,9 @@ def main():
print("=" * 60)
print(f" LOGAR Server Hub: {config['server_name']}")
print(f" Encryption Fingerprint: {config['server_fingerprint']}")
print(f" Transport Security: {'mTLS (TLS 1.3)' if ssl_ctx else 'Plain TCP'}")
print(f" License Quota: {config.get('max_seats', 10)} Active Seats")
print(f" Server Encryption Fingerprint: {config['server_fingerprint']}")
print(f" Evaluation Window: {config['evaluation_window_hours']} hours | 4-Run Rule: Warnings | Immediate Pass: Errors")
print("=" * 60)