From 99d735fafb2373f89435736897ac297f45250d75 Mon Sep 17 00:00:00 2001 From: beso tsiklauri Date: Sun, 21 Jun 2026 19:02:52 +0400 Subject: [PATCH] feat: self-hosted web terminal + Headlamp-style K8s browser Web terminal (browser PTY, self-hosted, no third party): - backend relay over Redis pub/sub (worker-safe), 2 WS endpoints, TerminalSession audit row, first-message API-key auth (key off URL) - agent: stdlib RFC6455 WS client + pty.fork, hooked into exec poll - frontend: xterm.js TerminalPanel, Terminal tab on nodes - security borrowed from DirectGate: tenant/node scoped, concurrency cap, idle + max-duration timeouts, audited; shell runs as restricted agent user K8s resource browser (Headlamp-style, read-only): - agent collects extended kinds (services/configmaps/statefulsets/ingresses/ pvcs/events/...), secret values stripped on host; tags push with agent_node - frontend Resources tab + YAML drawer (js-yaml), pod Shell + YAML actions - shell into pods via kubectl exec, reusing terminal infra (argv generalized, k8s names regex-validated against injection) Review fixes from earlier pass: - traces background task uses its own DB session (was reusing closed request session) - datetime.utcnow() -> datetime.now(timezone.utc) across routes - Claude model tiering: RCA->opus, chat/topology->haiku, balanced default Chore: add missing vite-env.d.ts, fix Dashboard generic-index type; tsc clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- agent/shipper.py | 270 +++++++++++++++ backend/.env.example | 9 + backend/app/ai/engine.py | 4 +- backend/app/api/routes/assistant.py | 2 +- backend/app/api/routes/deploy_events.py | 4 +- backend/app/api/routes/exec.py | 14 +- backend/app/api/routes/incidents.py | 4 +- backend/app/api/routes/k8s.py | 4 + backend/app/api/routes/tenants.py | 4 +- backend/app/api/routes/terminal.py | 327 ++++++++++++++++++ backend/app/api/routes/traces.py | 48 +-- backend/app/core/config.py | 12 + backend/app/main.py | 3 +- backend/app/models/__init__.py | 2 + backend/app/models/terminal.py | 40 +++ backend/app/tasks/topology_discovery.py | 2 +- frontend/package-lock.json | 44 ++- frontend/package.json | 4 + frontend/src/api/client.ts | 12 + .../src/components/topology/NodeLogsPanel.tsx | 14 +- .../src/components/topology/TerminalPanel.tsx | 158 +++++++++ frontend/src/views/Dashboard.tsx | 4 +- frontend/src/views/K8sView.tsx | 155 ++++++++- frontend/src/vite-env.d.ts | 1 + 24 files changed, 1095 insertions(+), 46 deletions(-) create mode 100644 backend/app/api/routes/terminal.py create mode 100644 backend/app/models/terminal.py create mode 100644 frontend/src/components/topology/TerminalPanel.tsx create mode 100644 frontend/src/vite-env.d.ts diff --git a/agent/shipper.py b/agent/shipper.py index 9ab21aa..f03aac7 100644 --- a/agent/shipper.py +++ b/agent/shipper.py @@ -21,6 +21,7 @@ """ import argparse +import base64 import hashlib import json import logging @@ -28,6 +29,7 @@ import os import re import socket +import ssl import subprocess import sys import time @@ -76,6 +78,12 @@ EXEC_RESULT_URL = f"{API_URL}/api/v1/exec/result" EXEC_POLL_INTERVAL = 3 +# Interactive terminal (PTY relayed to the browser). Opt-out with PYXIS_TERMINAL_ENABLED=0. +TERMINAL_ENABLED = os.environ.get("PYXIS_TERMINAL_ENABLED", "1") not in ("0", "false", "no") +_WS_BASE = API_URL.replace("https://", "wss://", 1).replace("http://", "ws://", 1) +TERMINAL_WS_URL = f"{_WS_BASE}/api/v1/terminal/ws-agent" +TERMINAL_SHELL = os.environ.get("PYXIS_TERMINAL_SHELL", os.environ.get("SHELL", "/bin/bash")) + SHIPPER_PATH = os.environ.get("PYXIS_SHIPPER_PATH", "/opt/pyxis/shipper.py") CONFIG_PATH = os.environ.get("PYXIS_CONFIG_PATH", "/opt/pyxis/config.json") AUTOUPDATE_INTERVAL = int(os.environ.get("PYXIS_AUTOUPDATE_INTERVAL", "300")) # 5 min @@ -720,6 +728,213 @@ def _exec_allowed(cmd: str) -> tuple[bool, str]: return False, "not in allowlist" +# ── Interactive terminal (PTY ↔ WebSocket) ──────────────────────────────────── +# +# Self-hosted, zero third-party. The agent dials the backend over a WebSocket +# (hand-rolled here to keep this file stdlib-only, like the rest of the agent), +# spawns a PTY shell, and pumps bytes both ways. The shell runs as whatever user +# the agent runs as (the restricted 'pyxis' user in the hardened install), so a +# browser terminal is no more privileged than the agent itself. + + +class _WSClient: + """Minimal RFC6455 WebSocket client (text frames). Client frames are masked.""" + + def __init__(self, url: str): + from urllib.parse import urlparse + + u = urlparse(url) + self.secure = u.scheme == "wss" + self.host = u.hostname or "localhost" + self.port = u.port or (443 if self.secure else 80) + self.path = u.path or "/" + self.sock: socket.socket | None = None + self._buf = b"" + + def connect(self, timeout: float = 10.0) -> None: + raw = socket.create_connection((self.host, self.port), timeout=timeout) + if self.secure: + ctx = ssl.create_default_context() + raw = ctx.wrap_socket(raw, server_hostname=self.host) + self.sock = raw + key = base64.b64encode(os.urandom(16)).decode() + req = ( + f"GET {self.path} HTTP/1.1\r\n" + f"Host: {self.host}:{self.port}\r\n" + "Upgrade: websocket\r\nConnection: Upgrade\r\n" + f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n" + ) + self.sock.sendall(req.encode()) + # Read until end of headers + while b"\r\n\r\n" not in self._buf: + chunk = self.sock.recv(4096) + if not chunk: + raise ConnectionError("WS handshake: connection closed") + self._buf += chunk + head, self._buf = self._buf.split(b"\r\n\r\n", 1) + if b"101" not in head.split(b"\r\n", 1)[0]: + raise ConnectionError(f"WS handshake failed: {head[:80]!r}") + + def send_text(self, data: str) -> None: + payload = data.encode() + header = bytearray([0x81]) # FIN + text + n = len(payload) + if n < 126: + header.append(0x80 | n) + elif n < 65536: + header.append(0x80 | 126) + header += n.to_bytes(2, "big") + else: + header.append(0x80 | 127) + header += n.to_bytes(8, "big") + mask = os.urandom(4) + header += mask + masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) + assert self.sock is not None + self.sock.sendall(bytes(header) + masked) + + def _read_exact(self, n: int) -> bytes: + assert self.sock is not None + while len(self._buf) < n: + chunk = self.sock.recv(65536) + if not chunk: + raise ConnectionError("WS closed") + self._buf += chunk + out, self._buf = self._buf[:n], self._buf[n:] + return out + + def recv_text(self) -> str | None: + """Return the next text message, or None on close. Handles ping/pong.""" + while True: + b0, b1 = self._read_exact(2) + opcode = b0 & 0x0F + masked = b1 & 0x80 + length = b1 & 0x7F + if length == 126: + length = int.from_bytes(self._read_exact(2), "big") + elif length == 127: + length = int.from_bytes(self._read_exact(8), "big") + mask = self._read_exact(4) if masked else b"" + payload = self._read_exact(length) + if masked: + payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) + if opcode == 0x8: # close + return None + if opcode == 0x9: # ping → pong + self._send_control(0x8A, payload) + continue + if opcode == 0xA: # pong + continue + return payload.decode("utf-8", "replace") + + def _send_control(self, op: int, payload: bytes) -> None: + mask = os.urandom(4) + masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) + assert self.sock is not None + self.sock.sendall(bytes([op, 0x80 | len(payload)]) + mask + masked) + + def close(self) -> None: + try: + if self.sock: + self._send_control(0x88, b"") + self.sock.close() + except Exception: + pass + + +def run_pty_session(session_id: str, argv: list | None = None) -> None: + """ + Spawn a PTY and bridge it to the backend over a WebSocket. + argv = [] → the node's default shell; otherwise exec that command (e.g. + `kubectl exec -it ...` for a pod shell). argv[0] is resolved via PATH. + """ + if os.name != "posix": + log.warning("terminal: PTY not supported on this platform") + return + import pty + import fcntl + import termios + import struct + import select + + command = list(argv) if argv else [TERMINAL_SHELL] + ws = _WSClient(f"{TERMINAL_WS_URL}/{session_id}") + pid = master = -1 + try: + ws.connect() + ws.send_text(json.dumps({"api_key": API_KEY, "node_name": NODE_NAME})) + log.info("terminal: session %s attaching PTY (%s)", session_id[:8], command[0]) + + pid, master = pty.fork() + if pid == 0: # child → exec the command + os.environ["TERM"] = "xterm-256color" + try: + os.execvp(command[0], command) + except Exception: + pass + os._exit(127) + + # Parent: pump master fd ↔ websocket. A reader thread handles ws→pty + # so the main loop can select() on the pty for pty→ws. + stop = threading.Event() + + def ws_reader() -> None: + while not stop.is_set(): + try: + raw = ws.recv_text() + except Exception: + break + if raw is None: + break + try: + msg = json.loads(raw) + except Exception: + continue + t = msg.get("t") + if t == "d": + os.write(master, base64.b64decode(msg["b"])) + elif t == "resize": + winsize = struct.pack("HHHH", int(msg.get("rows", 24)), + int(msg.get("cols", 80)), 0, 0) + fcntl.ioctl(master, termios.TIOCSWINSZ, winsize) + elif t == "close": + break + stop.set() + + threading.Thread(target=ws_reader, daemon=True).start() + + while not stop.is_set(): + r, _, _ = select.select([master], [], [], 0.5) + if master in r: + try: + data = os.read(master, 65536) + except OSError: + break + if not data: + break + ws.send_text(json.dumps({"t": "d", "b": base64.b64encode(data).decode()})) + try: + ws.send_text(json.dumps({"t": "exit", "code": 0})) + except Exception: + pass + except Exception as e: + log.warning("terminal: session %s error: %s", session_id[:8], e) + finally: + try: + if master >= 0: + os.close(master) + except Exception: + pass + try: + if pid > 0: + os.kill(pid, 9) + os.waitpid(pid, 0) + except Exception: + pass + ws.close() + log.info("terminal: session %s closed", session_id[:8]) + + def exec_loop() -> None: """Poll for commands from the backend and execute them.""" while True: @@ -733,6 +948,21 @@ def exec_loop() -> None: with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read()) + # Interactive terminal sessions requested from the browser. + if TERMINAL_ENABLED: + for item in data.get("pty", []) or []: + try: + spec = json.loads(item) + except Exception: + spec = {"sid": item, "argv": []} + sid = spec.get("sid") + if sid: + threading.Thread( + target=run_pty_session, + args=(sid, spec.get("argv") or []), + daemon=True, + ).start() + cmd_id = data.get("cmd_id") cmd = data.get("cmd") if cmd_id and cmd: @@ -1446,6 +1676,41 @@ def _kubectl_get(resource: str, all_namespaces: bool = False) -> list: return [] +# Resource kinds surfaced in the cluster browser. (kind -> kubectl resource name) +_K8S_RESOURCE_KINDS = { + "services": "services", + "configmaps": "configmaps", + "secrets": "secrets", + "statefulsets": "statefulsets", + "daemonsets": "daemonsets", + "replicasets": "replicasets", + "jobs": "jobs", + "cronjobs": "cronjobs", + "ingresses": "ingresses", + "endpoints": "endpoints", + "networkpolicies": "networkpolicies", + "pvcs": "persistentvolumeclaims", + "pvs": "persistentvolumes", + "serviceaccounts": "serviceaccounts", + "events": "events", +} + + +def _collect_k8s_resources() -> dict: + """Collect the extended resource set for the cluster browser.""" + out: dict[str, list] = {} + for kind, resource in _K8S_RESOURCE_KINDS.items(): + items = _kubectl_get(resource, all_namespaces=True) + if kind == "secrets": + # Never ship secret values — keep metadata + key names only. + for it in items: + data = it.get("data") or {} + it["data"] = {k: None for k in data} + it.pop("stringData", None) + out[kind] = items + return out + + def k8s_monitor_loop() -> None: """Snapshot cluster state and push to backend every K8S_MONITOR_INTERVAL seconds.""" # Check kubectl is available @@ -1457,10 +1722,15 @@ def k8s_monitor_loop() -> None: while True: try: state = { + # Which agent host has kubectl access — backend routes pod-exec here. + "agent_node": NODE_NAME, "nodes": _kubectl_get("nodes"), "pods": _kubectl_get("pods", all_namespaces=True), "deployments": _kubectl_get("deployments", all_namespaces=True), "namespaces": _kubectl_get("namespaces"), + # Headlamp-style resource browser. Full -o json kept so the UI can + # render each item's YAML offline. Secret values are stripped below. + "resources": _collect_k8s_resources(), } payload = json.dumps(state).encode() req = urllib.request.Request( diff --git a/backend/.env.example b/backend/.env.example index 27c3b77..dd1744b 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -1,5 +1,14 @@ DATABASE_URL=postgresql+asyncpg://pyxis:pyxis@localhost:5432/pyxis REDIS_URL=redis://localhost:6379 ANTHROPIC_API_KEY=sk-ant-... +# Model tiers (optional overrides). FAST=chat/extraction, MODEL=balanced, DEEP=incident RCA. +CLAUDE_MODEL=claude-sonnet-4-6 +CLAUDE_MODEL_FAST=claude-haiku-4-5-20251001 +CLAUDE_MODEL_DEEP=claude-opus-4-8 SECRET_KEY=change-me-in-production-use-32-random-bytes DEBUG=false +# Interactive browser terminal (PTY via the agent). Set false to disable entirely. +TERMINAL_ENABLED=true +TERMINAL_IDLE_TIMEOUT=600 +TERMINAL_MAX_DURATION=3600 +TERMINAL_MAX_PER_TENANT=5 diff --git a/backend/app/ai/engine.py b/backend/app/ai/engine.py index 5764c95..e4cb7d1 100644 --- a/backend/app/ai/engine.py +++ b/backend/app/ai/engine.py @@ -271,7 +271,7 @@ async def _run_rca( for p in past_incidents ) or "(no prior incidents)" - log.info("_run_rca: calling Claude for incident %s (model=%s)", incident.id, settings.CLAUDE_MODEL) + log.info("_run_rca: calling Claude for incident %s (model=%s)", incident.id, settings.CLAUDE_MODEL_DEEP) # 7. Call Claude system_prompt = ( @@ -333,7 +333,7 @@ async def _run_rca( try: message = await _anthropic.messages.create( - model=settings.CLAUDE_MODEL, + model=settings.CLAUDE_MODEL_DEEP, max_tokens=2048, system=system_prompt, messages=[{"role": "user", "content": user_prompt}], diff --git a/backend/app/api/routes/assistant.py b/backend/app/api/routes/assistant.py index ebc1d68..7ffae79 100644 --- a/backend/app/api/routes/assistant.py +++ b/backend/app/api/routes/assistant.py @@ -66,7 +66,7 @@ async def chat( try: response = await _anthropic.messages.create( - model=settings.CLAUDE_MODEL, + model=settings.CLAUDE_MODEL_FAST, max_tokens=1024, system=system, messages=messages, diff --git a/backend/app/api/routes/deploy_events.py b/backend/app/api/routes/deploy_events.py index 9cd0cfb..41752bd 100644 --- a/backend/app/api/routes/deploy_events.py +++ b/backend/app/api/routes/deploy_events.py @@ -1,5 +1,5 @@ import uuid -from datetime import datetime +from datetime import datetime, timezone from typing import Any from fastapi import APIRouter, Depends @@ -50,7 +50,7 @@ async def create_deploy_event( version=payload.version, deployed_by=payload.deployed_by, environment=payload.environment, - deployed_at=payload.deployed_at or datetime.utcnow(), + deployed_at=payload.deployed_at or datetime.now(timezone.utc), meta=payload.meta, ) db.add(event) diff --git a/backend/app/api/routes/exec.py b/backend/app/api/routes/exec.py index d729ede..e3097c4 100644 --- a/backend/app/api/routes/exec.py +++ b/backend/app/api/routes/exec.py @@ -184,12 +184,22 @@ async def poll_command( ): r = await get_redis() pending_key = f"exec:pending:{tenant.id}:{node_name}" + + # Pending interactive terminal sessions for this node (drain the whole list). + pty_key = f"pty:pending:{tenant.id}:{node_name}" + pty_sessions: list[str] = [] + while True: + sid = await r.lpop(pty_key) + if sid is None: + break + pty_sessions.append(sid) + raw = await r.get(pending_key) if raw: data = json.loads(raw) await r.delete(pending_key) # consume it — one command at a time - return {"cmd_id": data["cmd_id"], "cmd": data["cmd"]} - return {"cmd_id": None, "cmd": None} + return {"cmd_id": data["cmd_id"], "cmd": data["cmd"], "pty": pty_sessions} + return {"cmd_id": None, "cmd": None, "pty": pty_sessions} # ── Agent: post command result ───────────────────────────────────────────────── diff --git a/backend/app/api/routes/incidents.py b/backend/app/api/routes/incidents.py index 169f04e..053e117 100644 --- a/backend/app/api/routes/incidents.py +++ b/backend/app/api/routes/incidents.py @@ -53,7 +53,7 @@ async def incident_heatmap( tenant: Tenant = Depends(get_current_tenant), db: AsyncSession = Depends(get_db), ): - since = datetime.utcnow() - timedelta(days=days) + since = datetime.now(timezone.utc) - timedelta(days=days) result = await db.execute( select( cast(Incident.started_at, Date).label("date"), @@ -117,7 +117,7 @@ async def update_incident( if payload.status: incident.status = payload.status if payload.status == "resolved": - incident.resolved_at = datetime.utcnow() + incident.resolved_at = datetime.now(timezone.utc) if payload.rca_summary: incident.rca_summary = payload.rca_summary diff --git a/backend/app/api/routes/k8s.py b/backend/app/api/routes/k8s.py index 00c1aee..1b0dd4b 100644 --- a/backend/app/api/routes/k8s.py +++ b/backend/app/api/routes/k8s.py @@ -23,10 +23,14 @@ class K8sState(BaseModel): + agent_node: str = "" # external_id of the agent host with kubectl access nodes: list[dict[str, Any]] = [] pods: list[dict[str, Any]] = [] deployments: list[dict[str, Any]] = [] namespaces: list[dict[str, Any]] = [] + # Extended Headlamp-style browser: {kind: [items]} — services, configmaps, + # statefulsets, ingresses, pvcs, events, etc. (secret values stripped by agent). + resources: dict[str, list[dict[str, Any]]] = {} updated_at: str = "" diff --git a/backend/app/api/routes/tenants.py b/backend/app/api/routes/tenants.py index f2c1417..5fe4c7c 100644 --- a/backend/app/api/routes/tenants.py +++ b/backend/app/api/routes/tenants.py @@ -4,7 +4,7 @@ """ import secrets import uuid -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel @@ -84,7 +84,7 @@ async def tenant_stats(db: AsyncSession = Depends(get_db)): tenants_result = await db.execute(select(Tenant).where(Tenant.is_active == True)) tenants = tenants_result.scalars().all() - since_7d = datetime.utcnow() - timedelta(days=7) + since_7d = datetime.now(timezone.utc) - timedelta(days=7) stats = [] for t in tenants: total = (await db.execute( diff --git a/backend/app/api/routes/terminal.py b/backend/app/api/routes/terminal.py new file mode 100644 index 0000000..d8c4023 --- /dev/null +++ b/backend/app/api/routes/terminal.py @@ -0,0 +1,327 @@ +""" +Interactive browser terminal (PTY) to a node, relayed through the backend. + +Self-hosted, zero third-party. Architecture (security design borrowed from +DirectGate, adapted for a trusted self-hosted relay): + + browser <-- WS --> backend <-- Redis pub/sub --> backend <-- WS --> agent + (xterm) (client ep) (agent ep) (PTY shell) + +Flow: + 1. Browser POSTs /nodes/{node_id} -> backend creates a TerminalSession (pending), + pushes the session id to the node's pty-pending list, returns session_id. + 2. Agent's existing exec poll picks up the pending session id and dials back in + over WS to /ws-agent/{session_id}, spawning a PTY shell. + 3. Browser connects WS to /ws/{session_id}. Backend bridges the two sockets + byte-for-byte via two Redis channels (worker-safe). + +Security: +- API key auth on both ends; tenant + node ownership verified. +- Per-session unguessable id; Redis channels namespaced + TTL'd. +- Concurrency cap, idle timeout, hard max duration. +- Every session audited (who/where/when/bytes). Byte stream itself is never stored. +- Shell runs as the agent's restricted user (see agent docs). + +Wire format on both Redis channels — text JSON, binary base64-wrapped: + {"t":"d","b":""} data chunk + {"t":"resize","cols":N,"rows":M} terminal resize (client -> agent only) + {"t":"ready"} agent attached (agent -> client) + {"t":"exit","code":N} shell exited (agent -> client) + {"t":"close"} teardown (either direction) +""" +import asyncio +import base64 +import json +import logging +import re +from datetime import datetime, timezone + +import redis.asyncio as aioredis +from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect +from pydantic import BaseModel +from sqlalchemy import select, func +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.config import get_settings +from app.core.database import AsyncSessionLocal, get_db +from app.core.deps import get_current_tenant +from app.core.redis import get_redis +from app.models.tenant import Tenant +from app.models.terminal import TerminalSession +from app.models.topology import Node + +router = APIRouter() +log = logging.getLogger(__name__) +settings = get_settings() + +PENDING_TTL = 60 # agent must dial in within 60s of the browser opening + + +def _up(session_id: str) -> str: # client -> agent + return f"pty:{session_id}:up" + + +def _dn(session_id: str) -> str: # agent -> client + return f"pty:{session_id}:dn" + + +def _pending_key(tenant_id: str, node_external_id: str) -> str: + return f"pty:pending:{tenant_id}:{node_external_id}" + + +# ── Browser: open a session ─────────────────────────────────────────────────── + + +class PodExecRequest(BaseModel): + namespace: str + pod: str + container: str | None = None + + +async def _create_session(db: AsyncSession, tenant: Tenant, node_id: str, + target: str, argv: list[str]) -> dict: + """Shared: validate node, enforce caps, create row, queue for the agent.""" + if not settings.TERMINAL_ENABLED: + raise HTTPException(status_code=403, detail="Terminal feature is disabled") + + node_r = await db.execute( + select(Node).where(Node.id == node_id, Node.tenant_id == tenant.id, Node.deleted_at.is_(None)) + ) + node = node_r.scalar_one_or_none() + if node is None: + raise HTTPException(status_code=404, detail="Node not found") + + active_r = await db.execute( + select(func.count()) + .select_from(TerminalSession) + .where(TerminalSession.tenant_id == tenant.id, TerminalSession.status != "closed") + ) + if (active_r.scalar() or 0) >= settings.TERMINAL_MAX_PER_TENANT: + raise HTTPException(status_code=429, detail="Too many open terminal sessions") + + session = TerminalSession( + tenant_id=tenant.id, + node_id=node.id, + # external_id is how the agent identifies itself (matches the exec poll + # contract); the agent WS auth compares against this. + node_name=node.external_id, + target=target, + status="pending", + ) + db.add(session) + await db.commit() + await db.refresh(session) + + # Queue for the agent to pick up on its next exec poll. The payload carries + # the argv the agent should exec under the PTY ([] = the node's default shell). + r = await get_redis() + key = _pending_key(tenant.id, node.external_id) + await r.rpush(key, json.dumps({"sid": session.id, "argv": argv})) + await r.expire(key, PENDING_TTL) + + log.info("terminal: opened session=%s node=%s target=%s", session.id, node.name, target) + return {"session_id": session.id, "ws_path": f"/api/v1/terminal/ws/{session.id}"} + + +@router.post("/nodes/{node_id}") +async def open_terminal( + node_id: str, + tenant: Tenant = Depends(get_current_tenant), + db: AsyncSession = Depends(get_db), +): + """Interactive shell on the node itself.""" + return await _create_session(db, tenant, node_id, "shell", []) + + +_K8S_NAME = re.compile(r"^[a-z0-9][a-z0-9.\-]{0,252}$") + + +@router.post("/k8s") +async def open_pod_terminal( + body: PodExecRequest, + tenant: Tenant = Depends(get_current_tenant), + db: AsyncSession = Depends(get_db), +): + """ + Interactive shell inside a pod via `kubectl exec`, routed to whichever agent + host pushed the cluster state (it has kubectl + kubeconfig). Names are + strictly validated to keep them out of argv injection range. + """ + for val in (body.namespace, body.pod, body.container or "x"): + if not _K8S_NAME.match(val): + raise HTTPException(status_code=400, detail=f"Invalid k8s name: {val!r}") + + # Find the agent host that monitors this cluster. + r = await get_redis() + raw = await r.get(f"k8s:state:{tenant.id}") + agent_node = json.loads(raw).get("agent_node") if raw else None + if not agent_node: + raise HTTPException(status_code=409, detail="No cluster agent is reporting. Is the k8s monitor running?") + node_r = await db.execute( + select(Node).where(Node.external_id == agent_node, Node.tenant_id == tenant.id, Node.deleted_at.is_(None)) + ) + node = node_r.scalar_one_or_none() + if node is None: + raise HTTPException(status_code=404, detail=f"Cluster agent node '{agent_node}' not found in topology") + + argv = ["kubectl", "exec", "-it", "-n", body.namespace, body.pod] + if body.container: + argv += ["-c", body.container] + # Fall back from bash to sh so it works on minimal images. + argv += ["--", "sh", "-c", "exec bash 2>/dev/null || exec sh"] + target = f"pod:{body.namespace}/{body.pod}" + return await _create_session(db, tenant, node.id, target, argv) + + +# ── Shared relay helpers ────────────────────────────────────────────────────── + + +async def _auth_session(api_key: str, session_id: str) -> TerminalSession | None: + """Return the session iff the api_key's tenant owns it and it isn't closed.""" + async with AsyncSessionLocal() as db: + t_r = await db.execute( + select(Tenant).where(Tenant.api_key == api_key, Tenant.is_active == True) # noqa: E712 + ) + tenant = t_r.scalar_one_or_none() + if not tenant: + return None + s_r = await db.execute( + select(TerminalSession).where( + TerminalSession.id == session_id, + TerminalSession.tenant_id == tenant.id, + ) + ) + session = s_r.scalar_one_or_none() + if session is None or session.status == "closed": + return None + return session + + +async def _mark(session_id: str, *, status: str | None = None, + add_in: int = 0, add_out: int = 0, ended: bool = False) -> None: + async with AsyncSessionLocal() as db: + s_r = await db.execute(select(TerminalSession).where(TerminalSession.id == session_id)) + s = s_r.scalar_one_or_none() + if s is None: + return + if status: + s.status = status + s.bytes_in += add_in + s.bytes_out += add_out + if ended: + s.status = "closed" + s.ended_at = datetime.now(timezone.utc) + await db.commit() + + +async def _bridge(websocket: WebSocket, session_id: str, recv_channel: str, + send_channel: str, *, is_client: bool) -> None: + """ + Pump bytes both ways between one WebSocket and a pair of Redis channels. + Counts bytes for audit. Enforces idle + max-duration on the client side. + """ + redis = aioredis.from_url(settings.REDIS_URL, decode_responses=True) + pubsub = redis.pubsub() + await pubsub.subscribe(recv_channel) + loop = asyncio.get_event_loop() + deadline = loop.time() + settings.TERMINAL_MAX_DURATION + + async def ws_to_redis() -> None: + while True: + if is_client: + timeout = settings.TERMINAL_IDLE_TIMEOUT + try: + raw = await asyncio.wait_for(websocket.receive_text(), timeout=timeout) + except asyncio.TimeoutError: + log.info("terminal: session=%s idle timeout", session_id) + return + else: + raw = await websocket.receive_text() + # Audit byte counts (best-effort, batched would be nicer). + try: + msg = json.loads(raw) + if msg.get("t") == "d": + n = len(base64.b64decode(msg["b"])) + await _mark(session_id, add_in=n if is_client else 0, + add_out=0 if is_client else n) + except Exception: + pass + await redis.publish(send_channel, raw) + + async def redis_to_ws() -> None: + async for message in pubsub.listen(): + if message["type"] != "message": + continue + await websocket.send_text(message["data"]) + + async def watchdog() -> None: + await asyncio.sleep(max(1, deadline - loop.time())) + log.info("terminal: session=%s hit max duration", session_id) + + tasks = [asyncio.create_task(ws_to_redis()), + asyncio.create_task(redis_to_ws()), + asyncio.create_task(watchdog())] + try: + await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + finally: + for t in tasks: + t.cancel() + await pubsub.unsubscribe(recv_channel) + await redis.aclose() + + +# ── Browser side ────────────────────────────────────────────────────────────── + + +@router.websocket("/ws/{session_id}") +async def client_ws(websocket: WebSocket, session_id: str): + await websocket.accept() + # First message authenticates (keeps the API key out of the URL / logs). + try: + auth = json.loads(await asyncio.wait_for(websocket.receive_text(), timeout=10)) + except Exception: + await websocket.close(code=4001, reason="auth required") + return + session = await _auth_session(auth.get("api_key", ""), session_id) + if session is None: + await websocket.close(code=4001, reason="unauthorized") + return + + try: + await _bridge(websocket, session_id, _dn(session_id), _up(session_id), is_client=True) + except WebSocketDisconnect: + pass + finally: + r = await get_redis() + await r.publish(_up(session_id), json.dumps({"t": "close"})) + await _mark(session_id, ended=True) + + +# ── Agent side ──────────────────────────────────────────────────────────────── + + +@router.websocket("/ws-agent/{session_id}") +async def agent_ws(websocket: WebSocket, session_id: str): + await websocket.accept() + try: + auth = json.loads(await asyncio.wait_for(websocket.receive_text(), timeout=10)) + except Exception: + await websocket.close(code=4001, reason="auth required") + return + session = await _auth_session(auth.get("api_key", ""), session_id) + if session is None or auth.get("node_name") != session.node_name: + await websocket.close(code=4001, reason="unauthorized") + return + + await _mark(session_id, status="active") + # Tell the client the shell is live. + r = await get_redis() + await r.publish(_dn(session_id), json.dumps({"t": "ready"})) + + try: + await _bridge(websocket, session_id, _up(session_id), _dn(session_id), is_client=False) + except WebSocketDisconnect: + pass + finally: + await r.publish(_dn(session_id), json.dumps({"t": "close"})) + await _mark(session_id, ended=True) diff --git a/backend/app/api/routes/traces.py b/backend/app/api/routes/traces.py index 14a13fa..4c3b8db 100644 --- a/backend/app/api/routes/traces.py +++ b/backend/app/api/routes/traces.py @@ -15,7 +15,7 @@ from sqlalchemy import select, func, Float, Integer from sqlalchemy.ext.asyncio import AsyncSession -from app.core.database import get_db +from app.core.database import get_db, AsyncSessionLocal from app.core.deps import get_current_tenant from app.core.redis import get_redis from app.ingestion.latency_detector import check_span @@ -49,33 +49,35 @@ async def ingest_traces( batch: TraceBatch, background: BackgroundTasks, tenant: Tenant = Depends(get_current_tenant), - db: AsyncSession = Depends(get_db), ): - background.add_task(_process_batch, batch.spans, tenant.id, db) + # The request-scoped session is closed once this handler returns, so the + # background task must open its own session (see _process_batch). + background.add_task(_process_batch, batch.spans, tenant.id) return {"accepted": len(batch.spans)} -async def _process_batch(spans: list[SpanIn], tenant_id: str, db: AsyncSession) -> None: +async def _process_batch(spans: list[SpanIn], tenant_id: str) -> None: redis = await get_redis() - for s in spans: - span = Span( - id=str(uuid.uuid4()), - tenant_id=tenant_id, - trace_id=s.trace_id, - span_id=s.span_id, - parent_span_id=s.parent_span_id, - service=s.service, - operation=s.operation, - duration_ms=s.duration_ms, - status=s.status, - status_code=s.status_code, - attributes=s.attributes, - started_at=s.started_at or datetime.now(timezone.utc), - ) - db.add(span) - await db.flush() - await check_span(span, tenant_id, redis, db) - await db.commit() + async with AsyncSessionLocal() as db: + for s in spans: + span = Span( + id=str(uuid.uuid4()), + tenant_id=tenant_id, + trace_id=s.trace_id, + span_id=s.span_id, + parent_span_id=s.parent_span_id, + service=s.service, + operation=s.operation, + duration_ms=s.duration_ms, + status=s.status, + status_code=s.status_code, + attributes=s.attributes, + started_at=s.started_at or datetime.now(timezone.utc), + ) + db.add(span) + await db.flush() + await check_span(span, tenant_id, redis, db) + await db.commit() # ── Analytics ───────────────────────────────────────────────────────────────── diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 71baa67..e9a20b1 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -10,7 +10,13 @@ class Settings(BaseSettings): REDIS_URL: str = "redis://localhost:6379" ANTHROPIC_API_KEY: str = "" + # Model tiers — match cost to task. Override any via env. + # FAST → cheap/interactive (chat, structured extraction) + # MODEL → balanced default (log playground analysis) + # DEEP → highest quality, runs rarely (incident RCA) CLAUDE_MODEL: str = "claude-sonnet-4-6" + CLAUDE_MODEL_FAST: str = "claude-haiku-4-5-20251001" + CLAUDE_MODEL_DEEP: str = "claude-opus-4-8" SECRET_KEY: str = "change-me-in-production" @@ -20,6 +26,12 @@ class Settings(BaseSettings): # How many knowledge chunks to inject into AI context per query RAG_TOP_K: int = 8 + # Interactive browser terminal (PTY over the agent). Powerful — gate it. + TERMINAL_ENABLED: bool = True + TERMINAL_IDLE_TIMEOUT: int = 600 # close session after N seconds of no client data + TERMINAL_MAX_DURATION: int = 3600 # hard cap on a single session + TERMINAL_MAX_PER_TENANT: int = 5 # concurrent sessions per tenant + class Config: env_file = ".env" extra = "ignore" diff --git a/backend/app/main.py b/backend/app/main.py index 66ede7b..49d8f56 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -21,7 +21,7 @@ logging.getLogger("uvicorn.access").setLevel(logging.WARNING) logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING) log = logging.getLogger(__name__) -from app.api.routes import ingest, topology, incidents, knowledge, ws, tenants, heartbeat, notifications, install, runbooks, deploy_events, analyze, traces, assistant, exec, k8s, connections, metrics +from app.api.routes import ingest, topology, incidents, knowledge, ws, tenants, heartbeat, notifications, install, runbooks, deploy_events, analyze, traces, assistant, exec, k8s, connections, metrics, terminal settings = get_settings() @@ -102,6 +102,7 @@ async def lifespan(app: FastAPI): app.include_router(k8s.router, prefix="/api/v1/k8s", tags=["k8s"]) app.include_router(connections.router, prefix="/api/v1/connections", tags=["connections"]) app.include_router(metrics.router, prefix="/api/v1/metrics", tags=["metrics"]) +app.include_router(terminal.router, prefix="/api/v1/terminal", tags=["terminal"]) @app.get("/health") diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 0c92a33..d441719 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -6,6 +6,7 @@ from app.models.runbook import Runbook from app.models.deploy_event import DeployEvent from app.models.span import Span +from app.models.terminal import TerminalSession __all__ = [ "Tenant", @@ -18,4 +19,5 @@ "Runbook", "DeployEvent", "Span", + "TerminalSession", ] diff --git a/backend/app/models/terminal.py b/backend/app/models/terminal.py new file mode 100644 index 0000000..699eb6a --- /dev/null +++ b/backend/app/models/terminal.py @@ -0,0 +1,40 @@ +import uuid +from datetime import datetime + +from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.database import Base + + +class TerminalSession(Base): + """ + An interactive PTY session opened from the browser to a node's agent. + + The byte stream itself is relayed through Redis pub/sub and never stored; + this row is the audit record — who opened a shell where, when, and how + much data crossed. Security design borrows from DirectGate: every session + is authenticated, tenant-scoped, time-bounded, and audited. + """ + + __tablename__ = "terminal_sessions" + + id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: str(uuid.uuid4())) + tenant_id: Mapped[str] = mapped_column(String, ForeignKey("tenants.id", ondelete="CASCADE"), index=True) + node_id: Mapped[str] = mapped_column(String, ForeignKey("nodes.id", ondelete="CASCADE"), index=True) + + node_name: Mapped[str] = mapped_column(String(256)) + # What the session runs: "shell" (node shell) or "pod:/" (kubectl exec). + target: Mapped[str] = mapped_column(String(256), default="shell") + status: Mapped[str] = mapped_column(String(16), default="pending", index=True) # pending|active|closed + client_ip: Mapped[str | None] = mapped_column(String(64)) + + bytes_in: Mapped[int] = mapped_column(Integer, default=0) # client → shell (keystrokes) + bytes_out: Mapped[int] = mapped_column(Integer, default=0) # shell → client (output) + + started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + __table_args__ = ( + Index("ix_terminal_tenant_started", "tenant_id", "started_at"), + ) diff --git a/backend/app/tasks/topology_discovery.py b/backend/app/tasks/topology_discovery.py index a4c5769..0954e11 100644 --- a/backend/app/tasks/topology_discovery.py +++ b/backend/app/tasks/topology_discovery.py @@ -383,7 +383,7 @@ async def _discover_from_claude( try: message = await _anthropic.messages.create( - model=settings.CLAUDE_MODEL, + model=settings.CLAUDE_MODEL_FAST, max_tokens=512, system="You are analyzing infrastructure logs to discover service dependencies. Be precise and conservative — only report relationships clearly evidenced in the logs.", messages=[{"role": "user", "content": prompt}], diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 687b1dd..43d94bf 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -14,9 +14,12 @@ "@radix-ui/react-tabs": "^1.1.1", "@radix-ui/react-tooltip": "^1.1.4", "@tanstack/react-query": "^5.59.0", + "@xterm/addon-fit": "^0.10.0", + "@xterm/xterm": "^5.5.0", "axios": "^1.7.7", "clsx": "^2.1.1", "date-fns": "^4.1.0", + "js-yaml": "^4.2.0", "lucide-react": "^0.447.0", "react": "^18.3.1", "react-dom": "^18.3.1", @@ -28,6 +31,7 @@ "zustand": "^5.0.0" }, "devDependencies": { + "@types/js-yaml": "^4.0.9", "@types/react": "^18.3.11", "@types/react-dom": "^18.3.1", "@typescript-eslint/eslint-plugin": "^8.8.1", @@ -2850,6 +2854,13 @@ "@types/unist": "*" } }, + "node_modules/@types/js-yaml": { + "version": "4.0.9", + "resolved": "https://registry.npmjs.org/@types/js-yaml/-/js-yaml-4.0.9.tgz", + "integrity": "sha512-k4MGaQl5TGo/iipqb2UDG2UwjXziSWkh0uysQelTlJpX1qGlpUZYm8PnO4DxG1qBomtJUdYJ6qR6xdIah10JLg==", + "dev": true, + "license": "MIT" + }, "node_modules/@types/json-schema": { "version": "7.0.15", "resolved": "https://registry.npmjs.org/@types/json-schema/-/json-schema-7.0.15.tgz", @@ -3195,6 +3206,21 @@ "vite": "^4.2.0 || ^5.0.0 || ^6.0.0 || ^7.0.0" } }, + "node_modules/@xterm/addon-fit": { + "version": "0.10.0", + "resolved": "https://registry.npmjs.org/@xterm/addon-fit/-/addon-fit-0.10.0.tgz", + "integrity": "sha512-UFYkDm4HUahf2lnEyHvio51TNGiLK66mqP2JoATy7hRZeXaGMRDr00JiSF7m63vR5WKATF605yEggJKsw0JpMQ==", + "license": "MIT", + "peerDependencies": { + "@xterm/xterm": "^5.0.0" + } + }, + "node_modules/@xterm/xterm": { + "version": "5.5.0", + "resolved": "https://registry.npmjs.org/@xterm/xterm/-/xterm-5.5.0.tgz", + "integrity": "sha512-hqJHYaQb5OptNunnyAnkHyM8aCjZ1MEIDTQu1iIbbTD/xops91NB5yq1ZK/dC2JDbVWtF23zUtl9JE2NqwT87A==", + "license": "MIT" + }, "node_modules/abort-controller": { "version": "3.0.0", "resolved": "https://registry.npmjs.org/abort-controller/-/abort-controller-3.0.0.tgz", @@ -3307,7 +3333,6 @@ "version": "2.0.1", "resolved": "https://registry.npmjs.org/argparse/-/argparse-2.0.1.tgz", "integrity": "sha512-8+9WqebbFzpX9OR+Wa6O29asIogeRMzcGtAINdpMHHyAg10f05aSFVBbcEqGf/PXw1EjAZ+q2/bEBg3DvurK3Q==", - "dev": true, "license": "Python-2.0" }, "node_modules/aria-hidden": { @@ -5061,10 +5086,19 @@ "license": "MIT" }, "node_modules/js-yaml": { - "version": "4.1.1", - "resolved": "https://registry.npmjs.org/js-yaml/-/js-yaml-4.1.1.tgz", - "integrity": "sha512-qQKT4zQxXl8lLwBtHMWwaTcGfFOZviOJet3Oy/xmGk2gZH677CJM9EvtfdSkgWcATZhj/55JZ0rmy3myCT5lsA==", - "dev": true, + "version": "4.2.0", + "resolved": "https://registry.npmjs.org/js-yaml/-/js-yaml-4.2.0.tgz", + "integrity": "sha512-ePWsvanv0DWuDRsW8dnt+R4jQ31SCRCQ7hhNcPXZPsoBZiemuZNYGf7adZdqX2D86j6rvKp3RpCxVTSb8WQlOw==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/puzrin" + }, + { + "type": "github", + "url": "https://github.com/sponsors/nodeca" + } + ], "license": "MIT", "dependencies": { "argparse": "^2.0.1" diff --git a/frontend/package.json b/frontend/package.json index b518878..c10e845 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -16,9 +16,12 @@ "@radix-ui/react-tabs": "^1.1.1", "@radix-ui/react-tooltip": "^1.1.4", "@tanstack/react-query": "^5.59.0", + "@xterm/addon-fit": "^0.10.0", + "@xterm/xterm": "^5.5.0", "axios": "^1.7.7", "clsx": "^2.1.1", "date-fns": "^4.1.0", + "js-yaml": "^4.2.0", "lucide-react": "^0.447.0", "react": "^18.3.1", "react-dom": "^18.3.1", @@ -30,6 +33,7 @@ "zustand": "^5.0.0" }, "devDependencies": { + "@types/js-yaml": "^4.0.9", "@types/react": "^18.3.11", "@types/react-dom": "^18.3.1", "@typescript-eslint/eslint-plugin": "^8.8.1", diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index b6b8b88..6ecf780 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -157,6 +157,16 @@ export const api = { { timeout: 35000 }, ).then((r) => r.data), }, + terminal: { + open: (nodeId: string) => + apiClient.post<{ session_id: string; ws_path: string }>( + `/api/v1/terminal/nodes/${nodeId}`, + ).then((r) => r.data), + openPod: (body: { namespace: string; pod: string; container?: string }) => + apiClient.post<{ session_id: string; ws_path: string }>( + `/api/v1/terminal/k8s`, body, + ).then((r) => r.data), + }, incidents: { list: (params?: { status_filter?: string }) => apiClient.get("/api/v1/incidents/", { params }).then((r) => r.data), @@ -231,10 +241,12 @@ export const api = { }; export interface K8sState { + agent_node?: string; nodes: Record[]; pods: Record[]; deployments: Record[]; namespaces: Record[]; + resources?: Record[]>; updated_at: string; } diff --git a/frontend/src/components/topology/NodeLogsPanel.tsx b/frontend/src/components/topology/NodeLogsPanel.tsx index 4302482..d149fb9 100644 --- a/frontend/src/components/topology/NodeLogsPanel.tsx +++ b/frontend/src/components/topology/NodeLogsPanel.tsx @@ -9,6 +9,7 @@ import { import { useQuery } from "@tanstack/react-query"; import { api, getErrorMessage } from "../../api/client"; import type { TopologyNode, NodeLogEntry } from "../../api/client"; +import { TerminalPanel } from "./TerminalPanel"; import clsx from "clsx"; // ── Helpers ─────────────────────────────────────────────────────────────────── @@ -59,7 +60,7 @@ interface HistoryEntry { ts: Date; } -type Tab = "logs" | "console" | "config" | "verbosity" | "health"; +type Tab = "logs" | "console" | "terminal" | "config" | "verbosity" | "health"; const AVAILABLE_SOURCES = [ { id: "syslog", label: "Syslog / journald", desc: "All system events via journald" }, @@ -326,6 +327,7 @@ export default function NodeLogsPanel({ node, onClose }: Props) { {([ { id: "logs", icon: , label: "Logs" }, { id: "console", icon: , label: "Console" }, + { id: "terminal", icon: , label: "Terminal" }, { id: "health", icon: , label: "Health" }, { id: "config", icon: , label: "Config" }, { id: "verbosity", icon: , label: "Verbosity" }, @@ -599,6 +601,16 @@ export default function NodeLogsPanel({ node, onClose }: Props) { )} + {/* ── Terminal tab ── */} + {tab === "terminal" && ( +
+ api.terminal.open(node.id)} + onClose={() => setTab("logs")} + /> +
+ )} {/* ── Config tab ── */} {tab === "config" && (
diff --git a/frontend/src/components/topology/TerminalPanel.tsx b/frontend/src/components/topology/TerminalPanel.tsx new file mode 100644 index 0000000..eb6a04b --- /dev/null +++ b/frontend/src/components/topology/TerminalPanel.tsx @@ -0,0 +1,158 @@ +import { useEffect, useRef, useState } from "react"; +import { Terminal } from "@xterm/xterm"; +import { FitAddon } from "@xterm/addon-fit"; +import "@xterm/xterm/css/xterm.css"; +import { X, Loader2 } from "lucide-react"; +import { getErrorMessage } from "../../api/client"; +import { useAppStore } from "../../store"; + +const WS_BASE = + import.meta.env.VITE_WS_URL || + `${window.location.protocol === "https:" ? "wss" : "ws"}://${window.location.host}`; + +function toB64(s: string): string { + const bytes = new TextEncoder().encode(s); + let bin = ""; + bytes.forEach((b) => (bin += String.fromCharCode(b))); + return btoa(bin); +} +function fromB64(b64: string): Uint8Array { + const bin = atob(b64); + const arr = new Uint8Array(bin.length); + for (let i = 0; i < bin.length; i++) arr[i] = bin.charCodeAt(i); + return arr; +} + +interface Props { + label: string; + openSession: () => Promise<{ session_id: string; ws_path: string }>; + onClose: () => void; +} + +export function TerminalPanel({ label, openSession, onClose }: Props) { + const containerRef = useRef(null); + const apiKey = useAppStore((s) => s.apiKey); + const [status, setStatus] = useState<"connecting" | "live" | "closed" | "error">("connecting"); + const [error, setError] = useState(""); + + useEffect(() => { + let term: Terminal | null = null; + let ws: WebSocket | null = null; + let disposed = false; + const fit = new FitAddon(); + + (async () => { + let session; + try { + session = await openSession(); + } catch (e) { + if (!disposed) { + setError(getErrorMessage(e)); + setStatus("error"); + } + return; + } + if (disposed || !containerRef.current) return; + + term = new Terminal({ + fontSize: 13, + fontFamily: "ui-monospace, SFMono-Regular, Menlo, monospace", + theme: { background: "#0a0a14", foreground: "#d4d4e0" }, + cursorBlink: true, + }); + term.loadAddon(fit); + term.open(containerRef.current); + fit.fit(); + + ws = new WebSocket(`${WS_BASE}${session.ws_path}`); + ws.binaryType = "arraybuffer"; + + ws.onopen = () => { + ws!.send(JSON.stringify({ api_key: apiKey })); + term!.writeln("\x1b[2m[pyxis] connecting to agent…\x1b[0m"); + }; + ws.onmessage = (e) => { + let msg; + try { + msg = JSON.parse(e.data); + } catch { + return; + } + if (msg.t === "d") { + term!.write(fromB64(msg.b)); + } else if (msg.t === "ready") { + setStatus("live"); + const dims = fit.proposeDimensions(); + if (dims) ws!.send(JSON.stringify({ t: "resize", cols: dims.cols, rows: dims.rows })); + } else if (msg.t === "exit" || msg.t === "close") { + term!.writeln("\r\n\x1b[2m[pyxis] session ended\x1b[0m"); + setStatus("closed"); + } + }; + ws.onerror = () => { + if (!disposed) setStatus("error"); + }; + ws.onclose = () => { + if (!disposed) setStatus((s) => (s === "live" ? "closed" : s)); + }; + + term.onData((data) => { + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify({ t: "d", b: toB64(data) })); + } + }); + + const onResize = () => { + fit.fit(); + const dims = fit.proposeDimensions(); + if (dims && ws && ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify({ t: "resize", cols: dims.cols, rows: dims.rows })); + } + }; + window.addEventListener("resize", onResize); + (term as Terminal & { _pyxisCleanup?: () => void })._pyxisCleanup = () => + window.removeEventListener("resize", onResize); + })(); + + return () => { + disposed = true; + (term as (Terminal & { _pyxisCleanup?: () => void }) | null)?._pyxisCleanup?.(); + ws?.close(); + term?.dispose(); + }; + // label uniquely identifies the target; openSession identity is intentionally + // excluded so a new thunk each render doesn't tear down the session. + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [label, apiKey]); + + return ( +
+
+ Terminal — {label} + + {status} + + {status === "connecting" && } + +
+ {error ? ( +
{error}
+ ) : ( +
+ )} +
+ ); +} diff --git a/frontend/src/views/Dashboard.tsx b/frontend/src/views/Dashboard.tsx index e12bb60..e12210a 100644 --- a/frontend/src/views/Dashboard.tsx +++ b/frontend/src/views/Dashboard.tsx @@ -62,9 +62,7 @@ function StatCard({ function IncidentRow({ inc, }: { - inc: ReturnType extends Promise - ? T[number] - : never; + inc: Awaited>[number]; }) { const setActive = useAppStore((s) => s.setActiveIncidentId); return ( diff --git a/frontend/src/views/K8sView.tsx b/frontend/src/views/K8sView.tsx index 593b60c..7f5b05c 100644 --- a/frontend/src/views/K8sView.tsx +++ b/frontend/src/views/K8sView.tsx @@ -7,8 +7,11 @@ import { useQuery } from "@tanstack/react-query"; import { Box, RefreshCw, Server, Layers, Package, Circle, CheckCircle2, XCircle, AlertTriangle, Clock, Cpu, HardDrive, + Boxes, FileCode, TerminalSquare, X, } from "lucide-react"; +import yaml from "js-yaml"; import { api } from "../api/client"; +import { TerminalPanel } from "../components/topology/TerminalPanel"; import clsx from "clsx"; // ── K8s data helpers ────────────────────────────────────────────────────────── @@ -108,19 +111,32 @@ function PhaseBadge({ phase }: { phase: string }) { // ── Tabs ────────────────────────────────────────────────────────────────────── -type TabId = "nodes" | "pods" | "workloads"; +type TabId = "nodes" | "pods" | "workloads" | "resources"; const TABS: { id: TabId; label: string; icon: React.ElementType }[] = [ { id: "nodes", label: "Nodes", icon: Server }, { id: "pods", label: "Pods", icon: Package }, { id: "workloads", label: "Workloads", icon: Layers }, + { id: "resources", label: "Resources", icon: Boxes }, ]; +// Friendly labels for the extended resource kinds the agent collects. +const RESOURCE_LABELS: Record = { + services: "Services", configmaps: "ConfigMaps", secrets: "Secrets", + statefulsets: "StatefulSets", daemonsets: "DaemonSets", replicasets: "ReplicaSets", + jobs: "Jobs", cronjobs: "CronJobs", ingresses: "Ingresses", endpoints: "Endpoints", + networkpolicies: "NetworkPolicies", pvcs: "PVCs", pvs: "PVs", + serviceaccounts: "ServiceAccounts", events: "Events", +}; + // ── Component ───────────────────────────────────────────────────────────────── export default function K8sView() { const [tab, setTab] = useState("nodes"); const [nsFilter, setNsFilter] = useState("all"); + const [selected, setSelected] = useState(null); // YAML drawer + const [resKind, setResKind] = useState("services"); // active resource kind + const [podShell, setPodShell] = useState<{ namespace: string; pod: string } | null>(null); const { data, isLoading, isError, refetch, isFetching, dataUpdatedAt } = useQuery({ queryKey: ["k8s-state"], @@ -400,6 +416,7 @@ export default function K8sView() { Restarts Node Age + Actions @@ -428,6 +445,17 @@ export default function K8sView() { {node} {getAge(meta.creationTimestamp)} + +
+ setPodShell({ namespace: meta.namespace, pod: meta.name })}> + + + setSelected(pod)}> + + +
+ ); })} @@ -435,6 +463,16 @@ export default function K8sView() { )} + {/* ── Resources tab (extended kinds) ── */} + {tab === "resources" && ( + + )} + {/* ── Workloads tab (Deployments) ── */} {tab === "workloads" && ( @@ -500,12 +538,127 @@ export default function K8sView() { )} + + {selected && setSelected(null)} />} + {podShell && ( +
setPodShell(null)}> +
e.stopPropagation()}> + api.terminal.openPod(podShell)} + onClose={() => setPodShell(null)} + /> +
+
+ )} ); } // ── Sub-components ───────────────────────────────────────────────────────────── +function RowBtn({ children, title, onClick }: { + children: React.ReactNode; title: string; onClick: () => void; +}) { + return ( + + ); +} + +function ResourceBrowser({ resources, kind, setKind, onView }: { + resources: Record; + kind: string; + setKind: (k: string) => void; + onView: (o: KObj) => void; +}) { + const kinds = Object.keys(RESOURCE_LABELS).filter((k) => (resources[k]?.length ?? 0) >= 0); + const items = resources[kind] ?? []; + return ( +
+
+ {kinds.map((k) => ( + + ))} +
+
+ + + + + + + {items.map((it) => { + const meta = getMeta(it); + return ( + + + + + + + ); + })} + {items.length === 0 && ( + + )} + +
NameNamespaceAgeActions
{meta.name} + {meta.namespace + ? {meta.namespace} + : —} + {getAge(meta.creationTimestamp)} onView(it)}>
+ No {RESOURCE_LABELS[kind] ?? kind} found +
+
+ ); +} + +function YamlDrawer({ obj, onClose }: { obj: KObj; onClose: () => void }) { + let text = ""; + try { + // Drop noisy server-managed fields for readability. + const clean = JSON.parse(JSON.stringify(obj)) as KObj; + if (clean.metadata) { + delete (clean.metadata as KObj).managedFields; + delete (clean.metadata as KObj).annotations?.["kubectl.kubernetes.io/last-applied-configuration"]; + } + text = yaml.dump(clean, { lineWidth: 120, sortKeys: false }); + } catch { + text = JSON.stringify(obj, null, 2); + } + const meta = getMeta(obj); + return ( +
+
e.stopPropagation()}> +
+ + + {obj.kind ?? "Resource"} · {meta.name} + + +
+ 
+          {text}
+
+
+
+ ); +} + function StatCard({ label, value, sub, ok, icon: Icon, }: { diff --git a/frontend/src/vite-env.d.ts b/frontend/src/vite-env.d.ts new file mode 100644 index 0000000..11f02fe --- /dev/null +++ b/frontend/src/vite-env.d.ts @@ -0,0 +1 @@ +///