atlas-iac/services/hermes/scripts/patch_ttyd_index.py

#!/usr/bin/env python3
"""Add reliable browser clipboard handling to ttyd's pinned client page."""

from __future__ import annotations

import argparse
import socket
import subprocess
import time
import urllib.request
from pathlib import Path


MARKER = "atlas-ttyd-clipboard"
UPSTREAM_COPY = 'document.execCommand("copy")'
CLIPBOARD_ADAPTER = r"""
<style>
  #atlas-clipboard-status {
    background: #181825;
    border: 1px solid #f9e2af;
    border-radius: 6px;
    color: #f9e2af;
    font: 13px/1.4 monospace;
    padding: 6px 10px;
    position: fixed;
    right: 12px;
    top: 12px;
    z-index: 10000;
  }
</style>
<script id="atlas-ttyd-clipboard">
(() => {
  const showStatus = (message, failed = false) => {
    let status = document.getElementById('atlas-clipboard-status');
    if (!status) {
      status = document.createElement('div');
      status.id = 'atlas-clipboard-status';
      document.body.appendChild(status);
    }
    status.textContent = message;
    status.style.borderColor = failed ? '#f38ba8' : '#a6e3a1';
    status.style.color = failed ? '#f38ba8' : '#a6e3a1';
    clearTimeout(status.removeTimer);
    status.removeTimer = setTimeout(() => status.remove(), 1200);
  };

  const fallbackCopy = (text) => {
    const input = document.createElement('textarea');
    input.value = text;
    input.setAttribute('readonly', '');
    input.style.position = 'fixed';
    input.style.opacity = '0';
    document.body.appendChild(input);
    input.select();
    const copied = document.execCommand('copy');
    input.remove();
    if (!copied) throw new Error('browser rejected clipboard write');
  };

  const install = () => {
    const term = window.term;
    if (!term) {
      setTimeout(install, 50);
      return;
    }
    if (term.__atlasClipboardInstalled) return;
    term.__atlasClipboardInstalled = true;

    const copySelection = async () => {
      const text = term.getSelection();
      if (!text) return false;
      try {
        if (navigator.clipboard && window.isSecureContext) {
          await navigator.clipboard.writeText(text);
        } else {
          fallbackCopy(text);
        }
        showStatus('Copied selection');
        return true;
      } catch (error) {
        try {
          fallbackCopy(text);
          showStatus('Copied selection');
          return true;
        } catch (fallbackError) {
          console.error('[atlas] clipboard copy failed', error, fallbackError);
          showStatus('Clipboard copy failed', true);
          return false;
        }
      }
    };

    window.addEventListener('mouseup', () => {
      if (term.hasSelection()) void copySelection();
    });
    window.addEventListener('keydown', (event) => {
      const copyKey = event.key.toLowerCase() === 'c'
        && (event.ctrlKey || event.metaKey)
        && term.hasSelection();
      if (!copyKey) return;
      event.preventDefault();
      event.stopImmediatePropagation();
      void copySelection();
    }, true);
  };

  install();
})();
</script>
"""


def patch_html(content: str) -> str:
    """Insert the adapter once and disable ttyd's false-success copy call."""
    if content.count(UPSTREAM_COPY) != 1 or content.count("</body>") != 1:
        raise RuntimeError("ttyd index patch context changed")
    if MARKER in content:
        raise RuntimeError("ttyd index is already patched")
    content = content.replace(UPSTREAM_COPY, "void 0", 1)
    return content.replace("</body>", f"{CLIPBOARD_ADAPTER}</body>", 1)


def fetch_embedded_index(ttyd: Path) -> str:
    """Serve and retrieve ttyd's embedded page without vendoring its bundle."""
    with socket.socket() as listener:
        listener.bind(("127.0.0.1", 0))
        port = listener.getsockname()[1]
    process = subprocess.Popen(
        [
            str(ttyd),
            "--interface",
            "127.0.0.1",
            "--port",
            str(port),
            "/bin/sh",
            "-c",
            "sleep 30",
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    try:
        for _ in range(100):
            try:
                with urllib.request.urlopen(
                    f"http://127.0.0.1:{port}/", timeout=1
                ) as response:
                    return response.read().decode("utf-8")
            except OSError:
                if process.poll() is not None:
                    raise RuntimeError("temporary ttyd exited before serving its index")
                time.sleep(0.05)
        raise RuntimeError("timed out retrieving ttyd's embedded index")
    finally:
        process.terminate()
        try:
            process.wait(timeout=3)
        except subprocess.TimeoutExpired:
            process.kill()
            process.wait(timeout=3)


def main() -> int:
    """Generate one patched index for the ttyd sidecar."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("ttyd", type=Path)
    parser.add_argument("output", type=Path)
    args = parser.parse_args()
    content = patch_html(fetch_embedded_index(args.ttyd))
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(content, encoding="utf-8")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())