test(pipeline): update end-to-end integration test for mTLS and dynamic PKI enrollment in test_pipeline.py

.gitea/workflows/ci.yml

@@ -28,7 +28,7 @@ jobs:
 
       - name: Verify Python Syntax
         run: |
-          python3 -m py_compile src/Server.py src/Win_Client.py src/Linux_Client.py compilation/package_dist.py compilation/upload_release.py tests/test_pipeline.py tests/*.py
+          python3 -m py_compile src/Server.py src/server_enrollment.py src/Win_Client.py src/Linux_Client.py compilation/package_dist.py compilation/upload_release.py tests/test_pipeline.py tests/*.py
 
       - name: Run Component Unit Tests
         run: |

.gitea/workflows/release-windows.yml

@@ -1,12 +1,9 @@
-name: Release Windows Binaries
+name: Release Windows Binaries & Installers
 
 on:
-  release:
-    types: [published, created]
   push:
     tags:
-      - '*'
-      - '**'
+      - 'v*'
   workflow_dispatch:
     inputs:
       tag:
@@ -15,13 +12,13 @@ on:
 
 jobs:
   release-windows:
-    name: Build & Release Windows Binaries
+    name: Build & Release Windows Binaries & Installers
     runs-on: windows-latest
     steps:
-      - name: Checkout Code
+      - name: Checkout Repository
         uses: actions/checkout@v4
 
-      - name: Set up Python
+      - name: Setup Python
         uses: actions/setup-python@v5
         with:
           python-version: '3.12'
@@ -37,7 +34,7 @@ jobs:
             }
           }
           & $py -m pip install --upgrade pip
-          & $py -m pip install pyinstaller -r compilation/requirements.txt
+          & $py -m pip install pyinstaller cryptography -r compilation/requirements.txt
 
       - name: Compile Standalone Windows Binaries
         shell: powershell
@@ -50,6 +47,26 @@ jobs:
           }
           & $py compilation/package_dist.py --target windows
 
+      - name: Compile Inno Setup Installers
+        shell: powershell
+        run: |
+          $iscc = $null
+          if (Test-Path "C:\Program Files (x86)\Inno Setup 6\ISCC.exe") {
+            $iscc = "C:\Program Files (x86)\Inno Setup 6\ISCC.exe"
+          } elseif (Test-Path "C:\Program Files\Inno Setup 6\ISCC.exe") {
+            $iscc = "C:\Program Files\Inno Setup 6\ISCC.exe"
+          } elseif (Get-Command "ISCC.exe" -ErrorAction SilentlyContinue) {
+            $iscc = "ISCC.exe"
+          }
+
+          if ($iscc) {
+            Write-Host "[*] Compiling Windows Inno Setup installers using $iscc..."
+            & $iscc compilation/installer_client.iss
+            & $iscc compilation/installer_server.iss
+          } else {
+            Write-Host "[!] Inno Setup compiler (ISCC.exe) not found on runner host. Skipping installer compilation."
+          }
+
       - name: Publish Windows Release Assets
         shell: powershell
         env:

.gitignore

@@ -11,6 +11,7 @@ build/
 dist/
 *.spec
 *.exe
+!compilation/nssm.exe
 *.bin
 *.dll
 *.so

compilation/install_linux_client.sh

@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+HUB_URL="${1:-http://hub.example.com:8443}"
+ENROLL_SECRET="${2:-}"
+
+INSTALL_DIR="/opt/logar-client"
+CONFIG_DIR="/etc/logar"
+
+echo "[+] Installing LOGAR Client..."
+mkdir -p "${INSTALL_DIR}" "${CONFIG_DIR}/certs"
+
+if [ -f "dist/Linux_Client.bin" ]; then
+    cp dist/Linux_Client.bin "${INSTALL_DIR}/Linux_Client"
+elif [ -f "dist/Linux_Client" ]; then
+    cp dist/Linux_Client "${INSTALL_DIR}/Linux_Client"
+else
+    echo "[!] Warning: dist/Linux_Client binary not found in current directory. Continuing with existing binary if present."
+fi
+
+if [ -f "${INSTALL_DIR}/Linux_Client" ]; then
+    chmod +x "${INSTALL_DIR}/Linux_Client"
+fi
+
+# Bootstrap certificate if missing and enrollment secret is provided
+if [ ! -f "${CONFIG_DIR}/certs/client.crt" ] && [ -n "${ENROLL_SECRET}" ]; then
+    echo "[+] Enrolling client with LOGAR Hub..."
+    MACHINE_ID=$(cat /etc/machine-id 2>/dev/null || hostname)
+    RESPONSE=$(curl -s -X POST "${HUB_URL}/api/client/enroll" \
+        -H "Content-Type: application/json" \
+        -d "{\"client_id\": \"${MACHINE_ID}\", \"hostname\": \"$(hostname)\", \"os\": \"linux\", \"enrollment_secret\": \"${ENROLL_SECRET}\"}")
+    
+    echo "${RESPONSE}" | grep -q "client_cert" || {
+        echo "[!] Enrollment failed: ${RESPONSE}"
+        exit 1
+    }
+
+    if command -v jq >/dev/null 2>&1; then
+        echo "${RESPONSE}" | jq -r .ca_cert > "${CONFIG_DIR}/certs/ca.crt"
+        echo "${RESPONSE}" | jq -r .client_cert > "${CONFIG_DIR}/certs/client.crt"
+        echo "${RESPONSE}" | jq -r .client_key > "${CONFIG_DIR}/certs/client.key"
+    else
+        python3 -c "import sys, json; data=json.loads(sys.stdin.read()); open('${CONFIG_DIR}/certs/ca.crt','w').write(data['ca_cert']); open('${CONFIG_DIR}/certs/client.crt','w').write(data['client_cert']); open('${CONFIG_DIR}/certs/client.key','w').write(data['client_key'])" <<< "${RESPONSE}"
+    fi
+    chmod 600 "${CONFIG_DIR}/certs/client.key"
+    echo "[+] Certificates written to ${CONFIG_DIR}/certs"
+fi
+
+cat <<EOF > /etc/systemd/system/logar-client.service
+[Unit]
+Description=LOGAR Edge Log Aggregator Client
+After=network.target
+
+[Service]
+Type=simple
+ExecStart=${INSTALL_DIR}/Linux_Client --config ${CONFIG_DIR}/config.json
+Restart=always
+RestartSec=5s
+User=root
+
+[Install]
+WantedBy=multi-user.target
+EOF
+
+if command -v systemctl >/dev/null 2>&1; then
+    systemctl daemon-reload
+    systemctl enable --now logar-client.service || true
+    echo "[+] LOGAR Client service configured and activated."
+else
+    echo "[+] Systemd service installed at /etc/systemd/system/logar-client.service"
+fi

compilation/install_linux_server.sh

@@ -0,0 +1,48 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+INSTALL_DIR="/opt/logar-server"
+CONFIG_DIR="/etc/logar"
+
+echo "[+] Installing LOGAR Server..."
+mkdir -p "${INSTALL_DIR}" "${CONFIG_DIR}" "/var/log/logar"
+
+if [ -f "dist/Server.bin" ]; then
+    cp dist/Server.bin "${INSTALL_DIR}/Server"
+elif [ -f "dist/Server" ]; then
+    cp dist/Server "${INSTALL_DIR}/Server"
+elif [ -f "dist/LOGAR_Server" ]; then
+    cp dist/LOGAR_Server "${INSTALL_DIR}/Server"
+else
+    echo "[!] Warning: dist/Server.bin binary not found in current directory. Continuing with existing binary if present."
+fi
+
+if [ -f "${INSTALL_DIR}/Server" ]; then
+    chmod +x "${INSTALL_DIR}/Server"
+fi
+
+cat <<EOF > /etc/systemd/system/logar-server.service
+[Unit]
+Description=LOGAR Hub and Aggregator Engine
+After=network.target
+
+[Service]
+Type=simple
+WorkingDirectory=${INSTALL_DIR}
+ExecStart=${INSTALL_DIR}/Server --config ${CONFIG_DIR}/server_config.json
+Restart=always
+RestartSec=5s
+User=root
+LimitNOFILE=65536
+
+[Install]
+WantedBy=multi-user.target
+EOF
+
+if command -v systemctl >/dev/null 2>&1; then
+    systemctl daemon-reload
+    systemctl enable --now logar-server.service || true
+    echo "[+] LOGAR Server service installed and activated."
+else
+    echo "[+] Systemd service installed at /etc/systemd/system/logar-server.service"
+fi

compilation/installer_client.iss

@@ -0,0 +1,27 @@
+[Setup]
+AppName=LOGAR Client
+AppVersion=1.0.3
+DefaultDirName={autopf}\LOGAR
+OutputDir=..\dist
+OutputBaseFilename=LOGAR-Client-Setup
+PrivilegesRequired=admin
+Compression=lzma
+SolidCompression=yes
+
+[Files]
+Source: "..\dist\Win_Client.exe"; DestDir: "{app}"; Flags: ignoreversion
+Source: "..\compilation\nssm.exe"; DestDir: "{app}"; Flags: ignoreversion
+
+[Dirs]
+Name: "{commonappdata}\LOGAR"; Permissions: users-modify
+
+[Run]
+Filename: "{app}\nssm.exe"; Parameters: "install LOGAR_Client ""{app}\Win_Client.exe"""; Flags: runhidden
+Filename: "{app}\nssm.exe"; Parameters: "set LOGAR_Client AppDirectory ""{app}"""; Flags: runhidden
+Filename: "{app}\nssm.exe"; Parameters: "set LOGAR_Client AppStdout ""{commonappdata}\LOGAR\client.log"""; Flags: runhidden
+Filename: "{app}\nssm.exe"; Parameters: "set LOGAR_Client AppStderr ""{commonappdata}\LOGAR\client_err.log"""; Flags: runhidden
+Filename: "{app}\nssm.exe"; Parameters: "start LOGAR_Client"; Flags: runhidden
+
+[UninstallRun]
+Filename: "{app}\nssm.exe"; Parameters: "stop LOGAR_Client"; Flags: runhidden
+Filename: "{app}\nssm.exe"; Parameters: "remove LOGAR_Client confirm"; Flags: runhidden

compilation/installer_server.iss

@@ -0,0 +1,27 @@
+[Setup]
+AppName=LOGAR Server
+AppVersion=1.0.3
+DefaultDirName={autopf}\LOGAR-Server
+OutputDir=..\dist
+OutputBaseFilename=LOGAR-Server-Setup
+PrivilegesRequired=admin
+Compression=lzma
+SolidCompression=yes
+
+[Files]
+Source: "..\dist\Server.exe"; DestDir: "{app}"; Flags: ignoreversion
+Source: "..\compilation\nssm.exe"; DestDir: "{app}"; Flags: ignoreversion
+
+[Dirs]
+Name: "{commonappdata}\LOGAR-Server"; Permissions: users-modify
+
+[Run]
+Filename: "{app}\nssm.exe"; Parameters: "install LOGAR_Server ""{app}\Server.exe"""; Flags: runhidden
+Filename: "{app}\nssm.exe"; Parameters: "set LOGAR_Server AppDirectory ""{app}"""; Flags: runhidden
+Filename: "{app}\nssm.exe"; Parameters: "set LOGAR_Server AppStdout ""{commonappdata}\LOGAR-Server\server.log"""; Flags: runhidden
+Filename: "{app}\nssm.exe"; Parameters: "set LOGAR_Server AppStderr ""{commonappdata}\LOGAR-Server\server_err.log"""; Flags: runhidden
+Filename: "{app}\nssm.exe"; Parameters: "start LOGAR_Server"; Flags: runhidden
+
+[UninstallRun]
+Filename: "{app}\nssm.exe"; Parameters: "stop LOGAR_Server"; Flags: runhidden
+Filename: "{app}\nssm.exe"; Parameters: "remove LOGAR_Server confirm"; Flags: runhidden

src/Linux_Client.py

@@ -2,9 +2,11 @@ import os
 import sys
 import json
 import socket
+import ssl
 import struct
 import argparse
 import subprocess
+import urllib.request
 import warnings
 from datetime import datetime, timezone, timedelta
 from typing import Optional, Dict, Any, List
@@ -18,6 +20,68 @@ CONFIG_FILE_NAME = "client_config.json"
 STATE_FILE_NAME = "client_state.json"
 
 
+def enroll_client_if_needed(hub_url: str, enrollment_secret: str, cert_dir: str, client_id: str, hostname: str, os_type: str = "linux"):
+    """Bootstraps client enrollment if certificates are missing."""
+    os.makedirs(cert_dir, exist_ok=True)
+    ca_path = os.path.join(cert_dir, "ca.crt")
+    cert_path = os.path.join(cert_dir, "client.crt")
+    key_path = os.path.join(cert_dir, "client.key")
+
+    if os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path):
+        return True
+
+    print(f"[*] Bootstrapping client enrollment with LOGAR Hub at {hub_url}...")
+    enroll_endpoint = f"{hub_url.rstrip('/')}/api/client/enroll"
+    payload = {
+        "client_id": client_id,
+        "hostname": hostname,
+        "os": os_type,
+        "enrollment_secret": enrollment_secret
+    }
+    req = urllib.request.Request(
+        enroll_endpoint,
+        data=json.dumps(payload).encode("utf-8"),
+        headers={"Content-Type": "application/json"}
+    )
+    with urllib.request.urlopen(req, timeout=10) as resp:
+        if resp.status != 200:
+            raise RuntimeError(f"Enrollment failed with status code {resp.status}")
+        data = json.loads(resp.read().decode("utf-8"))
+
+    with open(ca_path, "w", encoding="utf-8") as f:
+        f.write(data["ca_cert"])
+    with open(cert_path, "w", encoding="utf-8") as f:
+        f.write(data["client_cert"])
+    with open(key_path, "w", encoding="utf-8") as f:
+        f.write(data["client_key"])
+
+    try:
+        os.chmod(key_path, 0o600)
+    except Exception:
+        pass
+
+    print(f"[+] Client enrolled successfully! Certificates saved to {os.path.abspath(cert_dir)}")
+    return True
+
+
+def get_tls_socket(hub_host: str, hub_port: int, cert_dir: str):
+    """Establishes an mTLS connection with the LOGAR hub using client certificates."""
+    ca_path = os.path.join(cert_dir, "ca.crt")
+    cert_path = os.path.join(cert_dir, "client.crt")
+    key_path = os.path.join(cert_dir, "client.key")
+
+    if not (os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path)):
+        raise FileNotFoundError(f"mTLS certificates not found in '{cert_dir}'. Enroll client first.")
+
+    ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=ca_path)
+    ctx.load_cert_chain(certfile=cert_path, keyfile=key_path)
+    ctx.minimum_version = ssl.TLSVersion.TLSv1_3
+    ctx.check_hostname = False
+
+    raw_sock = socket.create_connection((hub_host, hub_port), timeout=15)
+    return ctx.wrap_socket(raw_sock, server_hostname=hub_host)
+
+
 def get_state_path(config_path: str, custom_state_path: Optional[str] = None) -> str:
     if custom_state_path:
         return custom_state_path
@@ -227,39 +291,80 @@ def get_recent_linux_logs(hours: int = 24, state: Optional[dict] = None) -> list
 
 def send_encrypted_logs_over_socket(config: dict, logs: list):
     """
-    Encrypts the payload using the server's OpenPGP public key and streams
-    over an authenticated TCP socket.
+    Streams logs to the LOGAR hub.
+    Uses mutual TLS 1.3 (mTLS) with client certificates if available,
+    or falls back to OpenPGP encrypted envelope over TCP.
     """
     server_host = config["server_host"]
     server_port = int(config["server_port"])
-    auth_token = config["auth_token"]
-    pub_key_armored = config["server_public_key"]
-    expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper()
+    cert_dir = config.get("cert_dir", "certs")
+    enrollment_secret = config.get("enrollment_secret")
+    machine_id = get_machine_identifier()
 
-    # Load and verify server public key
+    # Attempt automatic enrollment bootstrap if certs are missing and secret is provided
+    if enrollment_secret:
+        hermes_host = config.get("hermes_host", server_host)
+        hermes_port = config.get("hermes_port", 8443)
+        hub_url = f"http://{hermes_host}:{hermes_port}"
+        try:
+            enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type="linux")
+        except Exception as e:
+            print(f"[!] Warning: Enrollment bootstrap failed: {e}")
+
+    ca_path = os.path.join(cert_dir, "ca.crt")
+    cert_path = os.path.join(cert_dir, "client.crt")
+    key_path = os.path.join(cert_dir, "client.key")
+    has_mtls_certs = os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path)
+
+    if has_mtls_certs:
+        print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over mTLS (TLS 1.3)...")
+        with get_tls_socket(server_host, server_port, cert_dir) as sock:
+            payload = {
+                "server": machine_id,
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "logs": logs
+            }
+            payload_bytes = json.dumps(payload).encode("utf-8")
+            frame = struct.pack(">I", len(payload_bytes)) + payload_bytes
+            sock.sendall(frame)
+
+            resp_len_bytes = sock.recv(4)
+            if not resp_len_bytes:
+                raise ConnectionError("Server closed mTLS connection without response.")
+            resp_len = struct.unpack(">I", resp_len_bytes)[0]
+            resp_bytes = bytearray()
+            while len(resp_bytes) < resp_len:
+                chunk = sock.recv(min(4096, resp_len - len(resp_bytes)))
+                if not chunk:
+                    break
+                resp_bytes.extend(chunk)
+
+            response = json.loads(resp_bytes.decode("utf-8"))
+            print(f"[+] Server response: {response}")
+            return response
+
+    # Fallback to OpenPGP envelope over plain TCP socket
+    auth_token = config.get("auth_token", "")
+    pub_key_armored = config.get("server_public_key")
+    if not pub_key_armored:
+        raise ValueError("No server public key or mTLS certificates available for connection.")
+
+    expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper()
     pub_key, _ = pgpy.PGPKey.from_blob(pub_key_armored)
     actual_fp = str(pub_key.fingerprint).replace(" ", "").upper()
     if expected_fp and actual_fp != expected_fp:
-        raise ValueError(
-            f"Server fingerprint mismatch! Expected {expected_fp}, but key has {actual_fp}."
-        )
+        raise ValueError(f"Server fingerprint mismatch! Expected {expected_fp}, but key has {actual_fp}.")
 
-    machine_id = get_machine_identifier()
-
-    # Prepare batch
     payload = {
         "server": machine_id,
         "timestamp": datetime.now(timezone.utc).isoformat(),
         "logs": logs
     }
     payload_json = json.dumps(payload)
-
-    # Encrypt payload with server's encryption-only key
     pgp_msg = pgpy.PGPMessage.new(payload_json)
     encrypted_msg = pub_key.encrypt(pgp_msg)
     encrypted_armored = str(encrypted_msg)
 
-    # Envelope with socket authentication header
     envelope = {
         "auth_token": auth_token,
         "timestamp": datetime.now(timezone.utc).isoformat(),
@@ -267,17 +372,14 @@ def send_encrypted_logs_over_socket(config: dict, logs: list):
     }
     envelope_bytes = json.dumps(envelope).encode("utf-8")
 
-    # Connect over TCP socket and transmit with 4-byte length prefix framing
     print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over secure TCP socket...")
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
         sock.settimeout(15.0)
         sock.connect((server_host, server_port))
 
-        # Send frame: length (4 bytes big-endian) + envelope
         frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes
         sock.sendall(frame)
 
-        # Receive response length
         resp_len_bytes = sock.recv(4)
         if not resp_len_bytes:
             raise ConnectionError("Server closed connection without response.")

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)
 

src/Win_Client.py

@@ -2,8 +2,10 @@ import os
 import sys
 import json
 import socket
+import ssl
 import struct
 import argparse
+import urllib.request
 import warnings
 from datetime import datetime, timezone, timedelta
 from typing import Optional, Dict, Any, List
@@ -22,6 +24,68 @@ CONFIG_FILE_NAME = "client_config.json"
 STATE_FILE_NAME = "client_state.json"
 
 
+def enroll_client_if_needed(hub_url: str, enrollment_secret: str, cert_dir: str, client_id: str, hostname: str, os_type: str = "windows"):
+    """Bootstraps client enrollment if certificates are missing."""
+    os.makedirs(cert_dir, exist_ok=True)
+    ca_path = os.path.join(cert_dir, "ca.crt")
+    cert_path = os.path.join(cert_dir, "client.crt")
+    key_path = os.path.join(cert_dir, "client.key")
+
+    if os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path):
+        return True
+
+    print(f"[*] Bootstrapping client enrollment with LOGAR Hub at {hub_url}...")
+    enroll_endpoint = f"{hub_url.rstrip('/')}/api/client/enroll"
+    payload = {
+        "client_id": client_id,
+        "hostname": hostname,
+        "os": os_type,
+        "enrollment_secret": enrollment_secret
+    }
+    req = urllib.request.Request(
+        enroll_endpoint,
+        data=json.dumps(payload).encode("utf-8"),
+        headers={"Content-Type": "application/json"}
+    )
+    with urllib.request.urlopen(req, timeout=10) as resp:
+        if resp.status != 200:
+            raise RuntimeError(f"Enrollment failed with status code {resp.status}")
+        data = json.loads(resp.read().decode("utf-8"))
+
+    with open(ca_path, "w", encoding="utf-8") as f:
+        f.write(data["ca_cert"])
+    with open(cert_path, "w", encoding="utf-8") as f:
+        f.write(data["client_cert"])
+    with open(key_path, "w", encoding="utf-8") as f:
+        f.write(data["client_key"])
+
+    try:
+        os.chmod(key_path, 0o600)
+    except Exception:
+        pass
+
+    print(f"[+] Client enrolled successfully! Certificates saved to {os.path.abspath(cert_dir)}")
+    return True
+
+
+def get_tls_socket(hub_host: str, hub_port: int, cert_dir: str):
+    """Establishes an mTLS connection with the LOGAR hub using client certificates."""
+    ca_path = os.path.join(cert_dir, "ca.crt")
+    cert_path = os.path.join(cert_dir, "client.crt")
+    key_path = os.path.join(cert_dir, "client.key")
+
+    if not (os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path)):
+        raise FileNotFoundError(f"mTLS certificates not found in '{cert_dir}'. Enroll client first.")
+
+    ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=ca_path)
+    ctx.load_cert_chain(certfile=cert_path, keyfile=key_path)
+    ctx.minimum_version = ssl.TLSVersion.TLSv1_3
+    ctx.check_hostname = False
+
+    raw_sock = socket.create_connection((hub_host, hub_port), timeout=15)
+    return ctx.wrap_socket(raw_sock, server_hostname=hub_host)
+
+
 def get_state_path(config_path: str, custom_state_path: Optional[str] = None) -> str:
     if custom_state_path:
         return custom_state_path
@@ -194,39 +258,80 @@ def get_recent_windows_logs(hours: int = 24, state: Optional[dict] = None) -> li
 
 def send_encrypted_logs_over_socket(config: dict, logs: list):
     """
-    Encrypts the payload using the server's OpenPGP public key and streams
-    over an authenticated TCP socket. Zero local state is maintained on the client.
+    Streams logs to the LOGAR hub.
+    Uses mutual TLS 1.3 (mTLS) with client certificates if available,
+    or falls back to OpenPGP encrypted envelope over TCP.
     """
     server_host = config["server_host"]
     server_port = int(config["server_port"])
-    auth_token = config["auth_token"]
-    pub_key_armored = config["server_public_key"]
-    expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper()
+    cert_dir = config.get("cert_dir", "certs")
+    enrollment_secret = config.get("enrollment_secret")
+    machine_id = get_machine_identifier()
 
-    # Load and verify server public key
+    # Attempt automatic enrollment bootstrap if certs are missing and secret is provided
+    if enrollment_secret:
+        hermes_host = config.get("hermes_host", server_host)
+        hermes_port = config.get("hermes_port", 8443)
+        hub_url = f"http://{hermes_host}:{hermes_port}"
+        try:
+            enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type="windows")
+        except Exception as e:
+            print(f"[!] Warning: Enrollment bootstrap failed: {e}")
+
+    ca_path = os.path.join(cert_dir, "ca.crt")
+    cert_path = os.path.join(cert_dir, "client.crt")
+    key_path = os.path.join(cert_dir, "client.key")
+    has_mtls_certs = os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path)
+
+    if has_mtls_certs:
+        print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over mTLS (TLS 1.3)...")
+        with get_tls_socket(server_host, server_port, cert_dir) as sock:
+            payload = {
+                "server": machine_id,
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "logs": logs
+            }
+            payload_bytes = json.dumps(payload).encode("utf-8")
+            frame = struct.pack(">I", len(payload_bytes)) + payload_bytes
+            sock.sendall(frame)
+
+            resp_len_bytes = sock.recv(4)
+            if not resp_len_bytes:
+                raise ConnectionError("Server closed mTLS connection without response.")
+            resp_len = struct.unpack(">I", resp_len_bytes)[0]
+            resp_bytes = bytearray()
+            while len(resp_bytes) < resp_len:
+                chunk = sock.recv(min(4096, resp_len - len(resp_bytes)))
+                if not chunk:
+                    break
+                resp_bytes.extend(chunk)
+
+            response = json.loads(resp_bytes.decode("utf-8"))
+            print(f"[+] Server response: {response}")
+            return response
+
+    # Fallback to OpenPGP envelope over plain TCP socket
+    auth_token = config.get("auth_token", "")
+    pub_key_armored = config.get("server_public_key")
+    if not pub_key_armored:
+        raise ValueError("No server public key or mTLS certificates available for connection.")
+
+    expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper()
     pub_key, _ = pgpy.PGPKey.from_blob(pub_key_armored)
     actual_fp = str(pub_key.fingerprint).replace(" ", "").upper()
     if expected_fp and actual_fp != expected_fp:
-        raise ValueError(
-            f"Server fingerprint mismatch! Expected {expected_fp}, but key has {actual_fp}."
-        )
+        raise ValueError(f"Server fingerprint mismatch! Expected {expected_fp}, but key has {actual_fp}.")
 
-    machine_id = get_machine_identifier()
-
-    # Prepare zero-state candidate batch
     payload = {
         "server": machine_id,
         "timestamp": datetime.now(timezone.utc).isoformat(),
         "logs": logs
     }
     payload_json = json.dumps(payload)
-
-    # Encrypt payload with server's encryption-only key
     pgp_msg = pgpy.PGPMessage.new(payload_json)
     encrypted_msg = pub_key.encrypt(pgp_msg)
     encrypted_armored = str(encrypted_msg)
 
-    # Envelope with socket authentication header
     envelope = {
         "auth_token": auth_token,
         "timestamp": datetime.now(timezone.utc).isoformat(),
@@ -234,17 +339,14 @@ def send_encrypted_logs_over_socket(config: dict, logs: list):
     }
     envelope_bytes = json.dumps(envelope).encode("utf-8")
 
-    # Connect over TCP socket and transmit with 4-byte length prefix framing
     print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over secure TCP socket...")
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
         sock.settimeout(15.0)
         sock.connect((server_host, server_port))
 
-        # Send frame: length (4 bytes big-endian) + envelope
         frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes
         sock.sendall(frame)
 
-        # Receive response length
         resp_len_bytes = sock.recv(4)
         if not resp_len_bytes:
             raise ConnectionError("Server closed connection without response.")

src/server_enrollment.py

@@ -0,0 +1,267 @@
+import os
+import datetime
+import ipaddress
+from typing import Tuple, List, Optional
+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
+
+
+def calculate_cert_fingerprint(cert_pem: str) -> str:
+    """Computes SHA-256 fingerprint for a PEM-encoded X.509 certificate."""
+    cert = x509.load_pem_x509_certificate(cert_pem.encode("utf-8"))
+    return cert.fingerprint(hashes.SHA256()).hex().upper()
+
+
+def generate_ca_if_needed(cert_dir: str = "certs", common_name: str = "LOGAR-Root-CA") -> Tuple[x509.Certificate, rsa.RSAPrivateKey, str, str]:
+    """
+    Loads an existing Root CA or generates a self-signed Root CA certificate and private key.
+    Returns (ca_cert_obj, ca_key_obj, ca_cert_pem, ca_key_pem).
+    """
+    os.makedirs(cert_dir, exist_ok=True)
+    ca_cert_path = os.path.join(cert_dir, "ca.crt")
+    ca_key_path = os.path.join(cert_dir, "ca.key")
+
+    if os.path.exists(ca_cert_path) and os.path.exists(ca_key_path):
+        with open(ca_cert_path, "r", encoding="utf-8") as f:
+            ca_cert_pem = f.read()
+        with open(ca_key_path, "r", encoding="utf-8") as f:
+            ca_key_pem = f.read()
+        ca_cert = x509.load_pem_x509_certificate(ca_cert_pem.encode("utf-8"))
+        ca_key = serialization.load_pem_private_key(ca_key_pem.encode("utf-8"), password=None)
+        return ca_cert, ca_key, ca_cert_pem, ca_key_pem
+
+    # Generate RSA 4096 private key for Root CA
+    ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096)
+    subject = issuer = x509.Name([
+        x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"),
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR"),
+        x509.NameAttribute(NameOID.COMMON_NAME, common_name),
+    ])
+
+    now = datetime.datetime.now(datetime.timezone.utc)
+    ca_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())
+    )
+
+    ca_cert_pem = ca_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
+    ca_key_pem = ca_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.TraditionalOpenSSL,
+        encryption_algorithm=serialization.NoEncryption()
+    ).decode("utf-8")
+
+    with open(ca_cert_path, "w", encoding="utf-8") as f:
+        f.write(ca_cert_pem)
+    with open(ca_key_path, "w", encoding="utf-8") as f:
+        f.write(ca_key_pem)
+
+    try:
+        os.chmod(ca_key_path, 0o600)
+    except Exception:
+        pass
+
+    return ca_cert, ca_key, ca_cert_pem, ca_key_pem
+
+
+def generate_server_cert_if_needed(
+    ca_cert: x509.Certificate,
+    ca_key: rsa.RSAPrivateKey,
+    hostnames: Optional[List[str]] = None,
+    cert_dir: str = "certs",
+    days_valid: int = 825
+) -> Tuple[x509.Certificate, rsa.RSAPrivateKey, str, str]:
+    """
+    Loads an existing server certificate or generates a new server TLS certificate signed by the Root CA.
+    Includes SANs for localhost, 127.0.0.1, and specified hostnames.
+    """
+    os.makedirs(cert_dir, exist_ok=True)
+    server_cert_path = os.path.join(cert_dir, "server.crt")
+    server_key_path = os.path.join(cert_dir, "server.key")
+
+    if os.path.exists(server_cert_path) and os.path.exists(server_key_path):
+        with open(server_cert_path, "r", encoding="utf-8") as f:
+            server_cert_pem = f.read()
+        with open(server_key_path, "r", encoding="utf-8") as f:
+            server_key_pem = f.read()
+        srv_cert = x509.load_pem_x509_certificate(server_cert_pem.encode("utf-8"))
+        srv_key = serialization.load_pem_private_key(server_key_pem.encode("utf-8"), password=None)
+        return srv_cert, srv_key, server_cert_pem, server_key_pem
+
+    server_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    subject = x509.Name([
+        x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"),
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR"),
+        x509.NameAttribute(NameOID.COMMON_NAME, "LOGAR-Server-Hub"),
+    ])
+
+    san_list = [
+        x509.DNSName("localhost"),
+        x509.DNSName("LOGAR-Server-Hub"),
+        x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")),
+        x509.IPAddress(ipaddress.IPv6Address("::1")),
+    ]
+
+    if hostnames:
+        for host in hostnames:
+            if not host:
+                continue
+            try:
+                ip_obj = ipaddress.ip_address(host)
+                san_list.append(x509.IPAddress(ip_obj))
+            except ValueError:
+                san_list.append(x509.DNSName(host))
+
+    now = datetime.datetime.now(datetime.timezone.utc)
+    server_cert = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(ca_cert.subject)
+        .public_key(server_key.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=False, path_length=None), critical=True)
+        .add_extension(
+            x509.KeyUsage(
+                digital_signature=True,
+                key_encipherment=True,
+                key_cert_sign=False,
+                crl_sign=False,
+                content_commitment=False,
+                data_encipherment=False,
+                key_agreement=False,
+                encipher_only=False,
+                decipher_only=False
+            ),
+            critical=True
+        )
+        .add_extension(
+            x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]),
+            critical=False
+        )
+        .add_extension(
+            x509.SubjectKeyIdentifier.from_public_key(server_key.public_key()),
+            critical=False
+        )
+        .add_extension(
+            x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()),
+            critical=False
+        )
+        .add_extension(x509.SubjectAlternativeName(san_list), critical=False)
+        .sign(ca_key, hashes.SHA256())
+    )
+
+    server_cert_pem = server_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
+    server_key_pem = server_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.TraditionalOpenSSL,
+        encryption_algorithm=serialization.NoEncryption()
+    ).decode("utf-8")
+
+    with open(server_cert_path, "w", encoding="utf-8") as f:
+        f.write(server_cert_pem)
+    with open(server_key_path, "w", encoding="utf-8") as f:
+        f.write(server_key_pem)
+
+    try:
+        os.chmod(server_key_path, 0o600)
+    except Exception:
+        pass
+
+    return server_cert, server_key, server_cert_pem, server_key_pem
+
+
+def issue_client_cert(
+    client_id: str,
+    ca_cert: x509.Certificate,
+    ca_key: rsa.RSAPrivateKey,
+    days_valid: int = 365
+) -> Tuple[str, str]:
+    """
+    Generates a 2048-bit RSA private key and signs an X.509 client certificate
+    with Common Name set to client_id.
+    Returns (cert_pem, key_pem).
+    """
+    client_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+
+    subject = x509.Name([
+        x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"),
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR"),
+        x509.NameAttribute(NameOID.COMMON_NAME, client_id),
+    ])
+
+    now = datetime.datetime.now(datetime.timezone.utc)
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(ca_cert.subject)
+        .public_key(client_key.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=False, path_length=None), critical=True)
+        .add_extension(
+            x509.KeyUsage(
+                digital_signature=True,
+                key_encipherment=True,
+                key_cert_sign=False,
+                crl_sign=False,
+                content_commitment=False,
+                data_encipherment=False,
+                key_agreement=False,
+                encipher_only=False,
+                decipher_only=False
+            ),
+            critical=True
+        )
+        .add_extension(
+            x509.ExtendedKeyUsage([ExtendedKeyUsageOID.CLIENT_AUTH]),
+            critical=False
+        )
+        .add_extension(
+            x509.SubjectKeyIdentifier.from_public_key(client_key.public_key()),
+            critical=False
+        )
+        .add_extension(
+            x509.AuthorityKeyIdentifier.from_issuer_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 = client_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.TraditionalOpenSSL,
+        encryption_algorithm=serialization.NoEncryption()
+    ).decode("utf-8")
+
+    return cert_pem, key_pem

tests/test_linux_client.py

@@ -198,6 +198,31 @@ class TestLinuxClientComponent(unittest.TestCase):
         self.assertEqual(logs[0]["__CURSOR"], "c3")
         self.assertEqual(newest_cursor, "c4")
 
+    def test_mtls_client_certificate_handling(self):
+        import shutil
+        test_dir = "test_linux_mtls_certs"
+        os.makedirs(test_dir, exist_ok=True)
+        try:
+            from src import server_enrollment as se
+            ca_cert, ca_key, ca_pem, _ = se.generate_ca_if_needed(test_dir)
+            client_cert_pem, client_key_pem = se.issue_client_cert("linux-client-test", ca_cert, ca_key)
+
+            with open(os.path.join(test_dir, "ca.crt"), "w") as f:
+                f.write(ca_pem)
+            with open(os.path.join(test_dir, "client.crt"), "w") as f:
+                f.write(client_cert_pem)
+            with open(os.path.join(test_dir, "client.key"), "w") as f:
+                f.write(client_key_pem)
+
+            # Test missing certs exception
+            empty_dir = "test_empty_linux_certs"
+            os.makedirs(empty_dir, exist_ok=True)
+            with self.assertRaises(FileNotFoundError):
+                Linux_Client.get_tls_socket("127.0.0.1", 9443, empty_dir)
+            shutil.rmtree(empty_dir, ignore_errors=True)
+        finally:
+            shutil.rmtree(test_dir, ignore_errors=True)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_pipeline.py

@@ -3,9 +3,11 @@ import sys
 import json
 import time
 import socket
+import ssl
 import struct
 import sqlite3
 import urllib.request
+import urllib.error
 import warnings
 from datetime import datetime, timezone, timedelta
 
@@ -16,7 +18,9 @@ sys.path.insert(0, ROOT_DIR)
 sys.path.insert(0, SRC_DIR)
 
 warnings.filterwarnings("ignore")
-import pgpy
+
+import Win_Client
+import Linux_Client
 
 # Test server endpoints
 TCP_HOST = "127.0.0.1"
@@ -24,6 +28,7 @@ TCP_PORT = 9443
 HERMES_HOST = "127.0.0.1"
 HERMES_PORT = 8443
 
+
 def run_tests():
     print("=== [1] Verifying server_config.json & client_config.json ===")
     server_cfg_path = "server_config.json" if os.path.exists("server_config.json") else os.path.join(ROOT_DIR, "server_config.json")
@@ -45,56 +50,117 @@ def run_tests():
     print(f"[OK] Verified client_config.json contains no machine/server/site name.")
     print(f"[OK] Fingerprint verified: {client_conf['server_fingerprint']}")
 
-    # Load public key
-    pub_key, _ = pgpy.PGPKey.from_blob(client_conf["server_public_key"])
-    
-    def send_socket_batch(logs, auth_token=client_conf["auth_token"]):
-        payload = {
-            "server": "test-edge-node.corp.internal",
-            "timestamp": datetime.now(timezone.utc).isoformat(),
-            "logs": logs
-        }
-        pgp_msg = pgpy.PGPMessage.new(json.dumps(payload))
-        enc = pub_key.encrypt(pgp_msg)
-        
-        envelope = {
-            "auth_token": auth_token,
-            "timestamp": datetime.now(timezone.utc).isoformat(),
-            "encrypted_payload": str(enc)
-        }
-        envelope_bytes = json.dumps(envelope).encode("utf-8")
-        
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-            s.settimeout(5.0)
-            s.connect((TCP_HOST, TCP_PORT))
-            frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes
+    cert_dir = os.path.join(ROOT_DIR, "test_pipeline_certs")
+    os.makedirs(cert_dir, exist_ok=True)
+    client_id = "test-edge-node.corp.internal"
+    enrollment_secret = server_conf.get("enrollment_secret") or client_conf.get("enrollment_secret")
+
+    print("\n=== [2] Testing Client Dynamic PKI Enrollment API (/api/client/enroll) ===")
+    enroll_url = f"http://{HERMES_HOST}:{HERMES_PORT}/api/client/enroll"
+
+    # 2a. Test rejection on invalid enrollment secret
+    bad_enroll_payload = {
+        "client_id": client_id,
+        "hostname": client_id,
+        "os": "linux",
+        "enrollment_secret": "invalid-secret-xyz"
+    }
+    req_bad = urllib.request.Request(
+        enroll_url,
+        data=json.dumps(bad_enroll_payload).encode("utf-8"),
+        headers={"Content-Type": "application/json"}
+    )
+    try:
+        with urllib.request.urlopen(req_bad, timeout=5):
+            assert False, "Expected HTTP 403 on invalid secret"
+    except urllib.error.HTTPError as e:
+        assert e.code == 403, f"Expected HTTP 403, got {e.code}"
+        print("[OK] Invalid enrollment secret rejected with HTTP 403.")
+
+    # 2b. Test valid client enrollment
+    valid_enroll_payload = {
+        "client_id": client_id,
+        "hostname": client_id,
+        "os": "linux",
+        "enrollment_secret": enrollment_secret
+    }
+    req_valid = urllib.request.Request(
+        enroll_url,
+        data=json.dumps(valid_enroll_payload).encode("utf-8"),
+        headers={"Content-Type": "application/json"}
+    )
+    with urllib.request.urlopen(req_valid, timeout=5) as resp:
+        assert resp.status == 200, f"Expected 200, got {resp.status}"
+        enroll_data = json.loads(resp.read().decode("utf-8"))
+        assert "ca_cert" in enroll_data
+        assert "client_cert" in enroll_data
+        assert "client_key" in enroll_data
+
+    ca_path = os.path.join(cert_dir, "ca.crt")
+    cert_path = os.path.join(cert_dir, "client.crt")
+    key_path = os.path.join(cert_dir, "client.key")
+
+    with open(ca_path, "w", encoding="utf-8") as f:
+        f.write(enroll_data["ca_cert"])
+    with open(cert_path, "w", encoding="utf-8") as f:
+        f.write(enroll_data["client_cert"])
+    with open(key_path, "w", encoding="utf-8") as f:
+        f.write(enroll_data["client_key"])
+    print(f"[OK] Client enrolled successfully. Certificates stored in {cert_dir}")
+
+    # 2c. Verify client shows in /api/clients
+    clients_req = urllib.request.Request(f"http://{HERMES_HOST}:{HERMES_PORT}/api/clients")
+    with urllib.request.urlopen(clients_req, timeout=5) as resp:
+        clients_data = json.loads(resp.read().decode("utf-8"))
+        assert clients_data["active_seats"] >= 1
+        found_c = any(c["client_id"] == client_id for c in clients_data["clients"])
+        assert found_c, f"Client {client_id} should be listed in /api/clients"
+    print(f"[OK] Verified client in /api/clients: Active Seats: {clients_data['active_seats']}/{clients_data['max_seats']}")
+
+    def send_mtls_batch(logs):
+        ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=ca_path)
+        ctx.load_cert_chain(certfile=cert_path, keyfile=key_path)
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_3
+        ctx.check_hostname = False
+
+        raw_sock = socket.create_connection((TCP_HOST, TCP_PORT), timeout=10)
+        with ctx.wrap_socket(raw_sock, server_hostname=TCP_HOST) as s:
+            payload = {
+                "server": client_id,
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "logs": logs
+            }
+            payload_bytes = json.dumps(payload).encode("utf-8")
+            frame = struct.pack(">I", len(payload_bytes)) + payload_bytes
             s.sendall(frame)
-            
+
             resp_len_bytes = s.recv(4)
+            if not resp_len_bytes:
+                raise ConnectionError("Server closed connection without response.")
             resp_len = struct.unpack(">I", resp_len_bytes)[0]
-            resp_bytes = s.recv(resp_len)
+            resp_bytes = bytearray()
+            while len(resp_bytes) < resp_len:
+                chunk = s.recv(min(4096, resp_len - len(resp_bytes)))
+                if not chunk:
+                    break
+                resp_bytes.extend(chunk)
             return json.loads(resp_bytes.decode("utf-8"))
 
-    print("\n=== [2] Testing Socket Authentication Failure ===")
-    bad_resp = send_socket_batch([], auth_token="invalid-token-12345")
-    assert bad_resp.get("status") == "error", f"Expected error, got: {bad_resp}"
-    print(f"[OK] Bad auth rejected correctly: {bad_resp['message']}")
-
+    print("\n=== [3] Testing Temporal Persistence & 4-Run Rule for Warnings over mTLS ===")
     test_signature = "TestServiceDegraded"
     candidate_log = [{
-        "server": "test-edge-node",
+        "server": client_id,
         "os_type": "linux",
         "signature": test_signature,
         "severity": "WARNING",
         "message": "Resource usage high warning"
     }]
 
-    print("\n=== [3] Testing Temporal Persistence & 4-Run Rule for Warnings ===")
     for run_num in range(1, 5):
-        resp = send_socket_batch(candidate_log)
+        resp = send_mtls_batch(candidate_log)
         assert resp.get("status") == "success", f"Run {run_num} failed: {resp}"
         promoted = resp.get("promoted_verified", 0)
-        print(f"[Run {run_num}/4] Ingested successfully. Promoted to verified: {promoted}")
+        print(f"[Run {run_num}/4] Ingested successfully via mTLS. Promoted to verified: {promoted}")
         if run_num < 4:
             assert promoted == 0, f"Expected 0 promoted on run {run_num} for warning, got {promoted}"
         else:
@@ -106,27 +172,27 @@ def run_tests():
     cursor.execute("SELECT run_count, status FROM active_issues WHERE signature = ?", (test_signature,))
     row = cursor.fetchone()
     conn.close()
-    
+
     assert row is not None, "Issue not found in SQLite"
     run_count, status = row
     print(f"[DB Verification] Issue '{test_signature}' -> run_count: {run_count}, status: {status}")
     assert run_count >= 4, f"Expected run_count >= 4, got {run_count}"
     assert status == "VERIFIED", f"Expected status 'VERIFIED', got {status}"
-    print("[OK] 4-Run Rule verified: Warning promoted to VERIFIED anomaly on 4th run!")
+    print("[OK] 4-Run Rule verified: Warning promoted to VERIFIED anomaly on 4th run over mTLS!")
 
-    print("\n=== [3b] Testing Immediate Pass for Errors ===")
+    print("\n=== [4] Testing Immediate Pass for Errors over mTLS ===")
     error_signature = "TestServiceCrashImmediate"
     error_log = [{
-        "server": "test-edge-node",
+        "server": client_id,
         "os_type": "linux",
         "signature": error_signature,
         "severity": "ERROR",
         "message": "Fatal process crash occurred"
     }]
-    err_resp = send_socket_batch(error_log)
+    err_resp = send_mtls_batch(error_log)
     assert err_resp.get("status") == "success", f"Error run failed: {err_resp}"
     print(f"[Run 1/1] Error ingested successfully. Promoted to verified: {err_resp.get('promoted_verified')}")
-    assert err_resp.get("promoted_verified") == 1, f"Expected error to be promoted to verified immediately on run 1, got {err_resp.get('promoted_verified')}"
+    assert err_resp.get("promoted_verified") == 1, f"Expected error to be promoted to verified immediately, got {err_resp.get('promoted_verified')}"
 
     conn = sqlite3.connect(server_conf.get("db_path", "logar_state.db"))
     cursor = conn.cursor()
@@ -140,12 +206,12 @@ def run_tests():
     assert err_status == "VERIFIED", f"Expected status 'VERIFIED', got {err_status}"
     print("[OK] Immediate pass verified: Error promoted to VERIFIED anomaly immediately!")
 
-    print("\n=== [4] Testing Hermes Reporting Endpoint (/api/hermes/report) ===")
+    print("\n=== [5] Testing Hermes Reporting Endpoint (/api/hermes/report) ===")
     req = urllib.request.Request(f"http://{HERMES_HOST}:{HERMES_PORT}/api/hermes/report")
     with urllib.request.urlopen(req, timeout=5) as response:
         assert response.status == 200, f"Expected 200, got {response.status}"
         hermes_data = json.loads(response.read().decode("utf-8"))
-    
+
     print(f"[Hermes API] Returned {len(hermes_data)} verified anomalies:")
     found_warning = False
     found_error = False
@@ -164,19 +230,24 @@ def run_tests():
     assert found_error, f"Error issue {error_signature} should be in Hermes report"
     print("[OK] Hermes reporting validated!")
 
-    print("\n=== [5] Testing Windows Client Script Integration ===")
+    print("\n=== [6] Testing Windows Client Script Integration ===")
     from Win_Client import get_recent_windows_logs
     win_logs = get_recent_windows_logs(hours=24)
     print(f"[Win_Client] Successfully queried Windows logs: {len(win_logs)} candidate entries.")
 
-    print("\n=== [6] Testing Linux Client Script Integration ===")
+    print("\n=== [7] Testing Linux Client Script Integration ===")
     from Linux_Client import get_recent_linux_logs
     linux_logs = get_recent_linux_logs(hours=24)
     print(f"[Linux_Client] Successfully queried Linux logs: {len(linux_logs)} candidate entries.")
 
-    print("\n==========================================")
-    print("  ALL VERIFICATION TESTS PASSED SUCCESSFULLY! ")
-    print("==========================================")
+    import shutil
+    if os.path.exists(cert_dir):
+        shutil.rmtree(cert_dir, ignore_errors=True)
+
+    print("\n=======================================================")
+    print("  ALL VERIFICATION TESTS (mTLS + PKI + PIPELINE) PASSED! ")
+    print("=======================================================")
+
 
 if __name__ == "__main__":
     run_tests()

tests/test_server.py

@@ -176,6 +176,121 @@ class TestServerComponent(unittest.TestCase):
         self.assertEqual(rows["SigWarn"], "TRANSIENT")
         self.assertEqual(rows["SigInfo"], "TRANSIENT")
 
+    def test_license_schema_and_pki_generation(self):
+        secret = "test-secret-12345"
+        Server.init_db(self.test_db, enrollment_secret=secret, max_seats=5)
+
+        conn = sqlite3.connect(self.test_db)
+        c = conn.cursor()
+        c.execute("SELECT max_seats, enrollment_secret FROM license_config WHERE id = 1")
+        row = c.fetchone()
+        conn.close()
+
+        self.assertIsNotNone(row)
+        self.assertEqual(row[0], 5)
+        self.assertEqual(row[1], secret)
+
+        # Test Dynamic PKI
+        test_cert_dir = "test_certs_pki"
+        try:
+            ca_cert, ca_key, ca_pem, ca_key_pem = Server.enrollment.generate_ca_if_needed(cert_dir=test_cert_dir)
+            self.assertIn("BEGIN CERTIFICATE", ca_pem)
+            self.assertIn("BEGIN RSA PRIVATE KEY", ca_key_pem)
+
+            srv_cert, srv_key, srv_pem, srv_key_pem = Server.enrollment.generate_server_cert_if_needed(
+                ca_cert, ca_key, hostnames=["127.0.0.1", "localhost"], cert_dir=test_cert_dir
+            )
+            self.assertIn("BEGIN CERTIFICATE", srv_pem)
+
+            client_cert_pem, client_key_pem = Server.enrollment.issue_client_cert("node-test-1", ca_cert, ca_key)
+            self.assertIn("BEGIN CERTIFICATE", client_cert_pem)
+            self.assertIn("BEGIN RSA PRIVATE KEY", client_key_pem)
+
+            fp = Server.enrollment.calculate_cert_fingerprint(client_cert_pem)
+            self.assertEqual(len(fp), 64)
+        finally:
+            import shutil
+            if os.path.exists(test_cert_dir):
+                shutil.rmtree(test_cert_dir, ignore_errors=True)
+
+    def test_enrollment_endpoint_and_seat_quota(self):
+        from fastapi import HTTPException
+        secret = "super-secret-enrollment"
+        max_seats = 2
+        Server.init_db(self.test_db, enrollment_secret=secret, max_seats=max_seats)
+
+        test_cert_dir = "test_certs_enroll"
+        try:
+            ca_cert, ca_key, ca_pem, _ = Server.enrollment.generate_ca_if_needed(cert_dir=test_cert_dir)
+            Server.SERVER_STATE["config"] = {"db_path": self.test_db}
+            Server.SERVER_STATE["ca_cert"] = ca_cert
+            Server.SERVER_STATE["ca_key"] = ca_key
+            Server.SERVER_STATE["ca_cert_pem"] = ca_pem
+
+            # 1. Invalid secret should raise 403
+            bad_req = Server.ClientEnrollRequest(
+                client_id="client-1",
+                hostname="host-1",
+                os="linux",
+                enrollment_secret="wrong-secret"
+            )
+            with self.assertRaises(HTTPException) as cm:
+                Server.enroll_client(bad_req)
+            self.assertEqual(cm.exception.status_code, 403)
+
+            # 2. Valid enrollment for client 1
+            req1 = Server.ClientEnrollRequest(
+                client_id="client-1",
+                hostname="host-1",
+                os="linux",
+                enrollment_secret=secret
+            )
+            resp1 = Server.enroll_client(req1)
+            self.assertIn("client_cert", resp1)
+            self.assertIn("client_key", resp1)
+            self.assertEqual(resp1["ca_cert"], ca_pem)
+
+            # 3. Valid enrollment for client 2
+            req2 = Server.ClientEnrollRequest(
+                client_id="client-2",
+                hostname="host-2",
+                os="windows",
+                enrollment_secret=secret
+            )
+            resp2 = Server.enroll_client(req2)
+            self.assertIn("client_cert", resp2)
+
+            # 4. Seat quota exhausted: client 3 should raise 403
+            req3 = Server.ClientEnrollRequest(
+                client_id="client-3",
+                hostname="host-3",
+                os="linux",
+                enrollment_secret=secret
+            )
+            with self.assertRaises(HTTPException) as cm:
+                Server.enroll_client(req3)
+            self.assertEqual(cm.exception.status_code, 403)
+            self.assertIn("License seat limit reached", cm.exception.detail)
+
+            # 5. Re-enrollment for existing client 1 should succeed
+            resp1_re = Server.enroll_client(req1)
+            self.assertIn("client_cert", resp1_re)
+
+            # 6. Revoked client should be rejected
+            conn = sqlite3.connect(self.test_db)
+            conn.execute("UPDATE clients SET status = 'revoked' WHERE client_id = 'client-1'")
+            conn.commit()
+            conn.close()
+
+            with self.assertRaises(HTTPException) as cm:
+                Server.enroll_client(req1)
+            self.assertEqual(cm.exception.status_code, 403)
+            self.assertIn("revoked", cm.exception.detail)
+        finally:
+            import shutil
+            if os.path.exists(test_cert_dir):
+                shutil.rmtree(test_cert_dir, ignore_errors=True)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_win_client.py

@@ -184,6 +184,31 @@ class TestWinClientComponent(unittest.TestCase):
         # Only rec 103 and 102 should be processed (101 is already sent, <= 100 breaks early)
         self.assertEqual(logs, [103, 102])
 
+    def test_mtls_client_certificate_handling(self):
+        import shutil
+        test_dir = "test_win_mtls_certs"
+        os.makedirs(test_dir, exist_ok=True)
+        try:
+            from src import server_enrollment as se
+            ca_cert, ca_key, ca_pem, _ = se.generate_ca_if_needed(test_dir)
+            client_cert_pem, client_key_pem = se.issue_client_cert("win-client-test", ca_cert, ca_key)
+
+            with open(os.path.join(test_dir, "ca.crt"), "w") as f:
+                f.write(ca_pem)
+            with open(os.path.join(test_dir, "client.crt"), "w") as f:
+                f.write(client_cert_pem)
+            with open(os.path.join(test_dir, "client.key"), "w") as f:
+                f.write(client_key_pem)
+
+            # Test missing certs exception
+            empty_dir = "test_empty_certs"
+            os.makedirs(empty_dir, exist_ok=True)
+            with self.assertRaises(FileNotFoundError):
+                Win_Client.get_tls_socket("127.0.0.1", 9443, empty_dir)
+            shutil.rmtree(empty_dir, ignore_errors=True)
+        finally:
+            shutil.rmtree(test_dir, ignore_errors=True)
+
 
 if __name__ == "__main__":
     unittest.main()