Engineering the Nexus Release: How I Built Secure E2EE Network Sync into a Linux Clipboard Manager (v1.5.1)

  ┌─────────────────────────────────────────────────────────────┐
  │                     LOCAL NETWORK (LAN)                     │
  │                                                             │
  │   ┌──────────────┐   mDNS Discovery    ┌──────────────┐    │
  │   │  Device A    │ ◄──────────────────► │  Device B    │    │
  │   │  (Arch)      │                      │  (Kali)      │    │
  │   │              │   X25519 Handshake   │              │    │
  │   │  ghostboard  │ ──── PIN + ECDH ───► │  ghostboard  │    │
  │   │              │                      │              │    │
  │   │  HTTPServer  │ ◄── AES-256-GCM ──── │  HTTPServer  │    │
  │   │  :PORT       │   /api/sync E2EE     │  :PORT       │    │
  │   └──────┬───────┘                      └──────┬───────┘    │
  │          │                                     │            │
  │       ghost.db                              ghost.db        │
  │    (trusted_peers)                       (trusted_peers)    │
  └─────────────────────────────────────────────────────────────┘

# core/network_discovery.py
import socket
from zeroconf import ServiceBrowser, Zeroconf, ServiceInfo, IPVersion
from PyQt6.QtCore import pyqtSignal, QThread

_SERVICE_TYPE = "_dotghost._tcp.local."


class DotGhostDiscovery(QThread):
    peer_found = pyqtSignal(str, str, str, int)  # node_id, name, ip, port
    peer_lost  = pyqtSignal(str)                 # node_id

    def __init__(self, node_id: str, device_name: str, port: int):
        super().__init__()
        self.node_id     = node_id
        self.device_name = device_name
        self.port        = port
        self.zeroconf    = None

    def run(self):
        self.zeroconf = Zeroconf(ip_version=IPVersion.V4Only)

        properties = {
            b'node_id':     self.node_id.encode('utf-8'),
            b'device_name': self.device_name.encode('utf-8'),
            b'version':     b'1',
        }

        instance_name = f"{self.node_id}.{_SERVICE_TYPE}"

        self.info = ServiceInfo(
            type_=_SERVICE_TYPE,
            name=instance_name,
            addresses=[socket.inet_aton(get_local_ip())],
            port=self.port,
            properties=properties,
            server=f"{self.node_id}.local."
        )

        self.zeroconf.register_service(self.info)
        self.browser = ServiceBrowser(self.zeroconf, _SERVICE_TYPE, self)
        self.exec()  # Qt event loop keeps the thread alive

    # ── zeroconf callbacks ──────────────────────────────────────

    def add_service(self, zc: Zeroconf, type_: str, name: str):
        info = zc.get_service_info(type_, name)
        if not info:
            return

        props    = info.properties
        node_id  = props.get(b'node_id', b'').decode()
        dev_name = props.get(b'device_name', b'Unknown').decode()

        if node_id == self.node_id:  # skip self
            return

        ip = socket.inet_ntoa(info.addresses[0])
        self.peer_found.emit(node_id, dev_name, ip, info.port)

    def remove_service(self, zc: Zeroconf, type_: str, name: str):
        node_id = name.replace(f".{_SERVICE_TYPE}", "")
        self.peer_lost.emit(node_id)

    def stop(self):
        if self.zeroconf:
            self.zeroconf.unregister_service(self.info)
            self.zeroconf.close()
        self.quit()

  Device A                              Device B
     │                                     │
     │  1. Generate ephemeral X25519 key   │
     │  2. Derive wrap key from PIN+salt   │
     │  3. Encrypt pubkey → send ─────────►│
     │                                     │  4. Decrypt pubkey with PIN+salt
     │                                     │  5. Generate ephemeral X25519 key
     │◄──────────────── send encrypted ────│  6. Derive shared secret (ECDH)
     │                                     │  7. Encrypt own pubkey → send
     │  8. Derive shared secret (ECDH)     │
     │  9. Discard ephemeral keys          │  9. Discard ephemeral keys
     │                                     │
     │      Shared Secret stored in DB     │

# core/pairing.py
import os
import base64
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import x25519
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

_KDF_ITERATIONS = 100_000  # OWASP minimum for PBKDF2-SHA256


def derive_handshake_key(pin: str, salt: bytes) -> bytes:
    """
    Derive a 256-bit wrapping key from a 6-digit PIN + dynamic salt.
    The salt is generated fresh per-pairing session and sent in plaintext —
    its job is to prevent precomputed PIN dictionaries, not to be secret.
    """
    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=32,
        salt=salt,
        iterations=_KDF_ITERATIONS,
    )
    return kdf.derive(pin.encode("utf-8"))


def generate_pairing_keys() -> tuple[x25519.X25519PrivateKey, bytes]:
    """
    Generate a fresh ephemeral X25519 key pair.
    These keys live only for the duration of the handshake.
    """
    private_key = x25519.X25519PrivateKey.generate()
    public_key_bytes = private_key.public_key().public_bytes(
        encoding=serialization.Encoding.Raw,
        format=serialization.PublicFormat.Raw
    )
    return private_key, public_key_bytes


def encrypt_pairing_payload(public_key_bytes: bytes, handshake_key: bytes) -> str:
    """
    Encrypt the public key using the PIN-derived wrapping key.
    Layout: [ 12 bytes nonce | ciphertext + 16 byte GCM tag ]
    """
    aesgcm = AESGCM(handshake_key)
    nonce = os.urandom(12)
    ciphertext = aesgcm.encrypt(nonce, public_key_bytes, None)
    return base64.b64encode(nonce + ciphertext).decode("utf-8")


def decrypt_pairing_payload(payload: str, handshake_key: bytes) -> bytes:
    """Reverse of encrypt_pairing_payload. Raises InvalidTag on wrong PIN."""
    raw = base64.b64decode(payload)
    nonce, ciphertext = raw[:12], raw[12:]
    aesgcm = AESGCM(handshake_key)
    return aesgcm.decrypt(nonce, ciphertext, None)


def derive_shared_secret(
    private_key: x25519.X25519PrivateKey,
    peer_public_key_bytes: bytes
) -> bytes:
    """
    Complete the ECDH exchange. The result is a raw 32-byte shared secret.
    Both sides arrive at the same value without it ever being transmitted.
    """
    peer_public_key = x25519.X25519PublicKey.from_public_bytes(peer_public_key_bytes)
    return private_key.exchange(peer_public_key)

# Storage schema for trusted peers
"""
CREATE TABLE trusted_peers (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    node_id       TEXT UNIQUE NOT NULL,
    device_name   TEXT NOT NULL,
    shared_secret BLOB NOT NULL,      -- raw 32 bytes from ECDH
    paired_at     TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""

# core/api_server.py
import json
import time
import urllib.parse
from collections import defaultdict
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Lock


class _RateLimiter:
    """Sliding window rate limiter — 3 pairing attempts per 60s per IP."""

    def __init__(self, max_attempts: int = 3, window: int = 60):
        self._attempts = defaultdict(list)
        self._lock     = Lock()
        self.max       = max_attempts
        self.window    = window

    def is_allowed(self, ip: str) -> bool:
        now = time.time()
        with self._lock:
            # drop timestamps outside the window
            self._attempts[ip] = [
                t for t in self._attempts[ip] if now - t < self.window
            ]
            if len(self._attempts[ip]) >= self.max:
                return False
            self._attempts[ip].append(now)
            return True


_rate_limiter = _RateLimiter()


class GhostAPIHandler(BaseHTTPRequestHandler):

    def log_message(self, format, *args):
        pass  # silence default HTTP logs

    def _send_response(self, code: int, body: dict):
        payload = json.dumps(body).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def do_POST(self):
        parsed    = urllib.parse.urlparse(self.path)
        client_ip = self.client_address[0]

        # ── /api/pair — pairing PIN exchange ──────────────────────
        if parsed.path == '/api/pair':
            if not _rate_limiter.is_allowed(client_ip):
                self._send_response(429, {
                    "status":  "error",
                    "message": "Too many pairing attempts. Try again later."
                })
                return
            # ... PIN verification and key exchange logic
            self._send_response(200, {"status": "paired"})
            return

        # ── /api/sync — incoming E2EE clipboard item ───────────────
        if parsed.path == '/api/sync':
            body = self.rfile.read(int(self.headers.get('Content-Length', 0)))
            data = json.loads(body)

            peer_node_id = data.get("node_id")
            peer = storage.get_trusted_peer(peer_node_id)

            if not peer:
                self._send_response(403, {
                    "status":  "error",
                    "message": "Untrusted peer"
                })
                return

            try:
                plaintext = decrypt_from_peer(
                    data.get("payload"),
                    peer["shared_secret"]
                )
            except Exception:
                # Wrong key or tampered payload — silent drop
                self._send_response(403, {
                    "status":  "error",
                    "message": "Decryption failed"
                })
                return

            item_id = storage.add_item("text", plaintext)

            # Cross-thread UI update via Qt signal — safe from any thread
            self.server.qthread_parent.sync_received.emit(item_id, plaintext)
            self._send_response(201, {"status": "synced"})

  Device A (sender)                         Device B (receiver)
  ─────────────────                         ────────────────────
  1. User copies text
  2. ClipboardMonitor detects change
  3. Encrypt with shared_secret
     [ AES-256-GCM | random 12-byte nonce ]
  4. POST /api/sync ──────────────────────► 5. GhostAPIHandler.do_POST()
     {                                         6. Lookup peer by node_id
       "node_id": "abc123",                    7. Decrypt with shared_secret
       "payload": "<base64 ciphertext>"        8. storage.add_item()
     }                                         9. sync_received.emit()
                                              10. UI updates in main thread
  ◄──────────────── 201 { "status": "synced" }

# .github/workflows/build-all.yml (signing steps)

- name: Sign AppImage (GPG)
  run: |
    echo "${{ secrets.GPG_PRIVATE_KEY }}" | gpg --import --batch --yes
    gpg --batch --yes --pinentry-mode loopback \
      --passphrase "${{ secrets.GPG_PASSPHRASE }}" \
      --detach-sign --armor \
      DotGhostBoard-*.AppImage

- name: Verify AppImage signature
  run: |
    gpg --verify DotGhostBoard-*.AppImage.asc DotGhostBoard-*.AppImage
    echo "✅ AppImage signature verified"

- name: Sign DEB Package (GPG)
  run: |
    echo "${{ secrets.GPG_PRIVATE_KEY }}" | gpg --import --batch --yes
    dpkg-sig --sign builder \
      -k "${{ secrets.GPG_KEY_ID }}" \
      --gpg-options "--passphrase ${{ secrets.GPG_PASSPHRASE }} --pinentry-mode loopback --batch --yes" \
      dotghostboard_*.deb

- name: Generate SHA256 checksums
  run: |
    cd out && sha256sum * > SHA256SUMS.txt

# Verify AppImage
gpg --verify DotGhostBoard-1.5.1-x86_64.AppImage.asc \
             DotGhostBoard-1.5.1-x86_64.AppImage

# Verify DEB
dpkg-sig --verify dotghostboard_1.5.1_amd64.deb

# Verify checksum
sha256sum -c SHA256SUMS.txt