diff --git a/src/Server.py b/src/Server.py index f6ea7fa..ef1bc00 100644 --- a/src/Server.py +++ b/src/Server.py @@ -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)