diff --git a/nest/dashboard/static/css/sessions.css b/nest/dashboard/static/css/sessions.css new file mode 100644 index 0000000..ee49b3a --- /dev/null +++ b/nest/dashboard/static/css/sessions.css @@ -0,0 +1,54 @@ +.machine-tabs { display: flex; gap: 8px; margin: 0 0 24px; padding-bottom: 12px; border-bottom: 1px solid var(--line); } +.machine-tab { display: inline-flex; align-items: center; gap: 8px; padding: 8px 16px; border-radius: 22px; color: var(--muted); font-size: 11px; } +.machine-tab.active { background: var(--white); color: var(--ink); box-shadow: var(--shadow-card); } +.machine-tab .icon { width: 14px; height: 14px; } +.session-section-heading { margin-bottom: 14px; } +.session-workspace { display: grid; grid-template-columns: 205px minmax(0, 1fr); gap: 16px; align-items: stretch; } +.session-history { padding: 14px 10px; min-width: 0; display: flex; flex-direction: column; } +.session-history > header { display: flex; align-items: center; justify-content: space-between; padding: 0 6px 10px; } +.session-history h3 { font-size: 11px; } +.session-history > .panel-footnote { margin-top: auto; padding: 15px 8px 0; } +.session-list { display: grid; gap: 6px; max-height: 510px; overflow-y: auto; } +.session-list-empty { padding: 24px 9px; color: var(--muted); font-size: 10px; line-height: 1.8; } +.session-item { display: grid; grid-template-columns: 1fr auto; gap: 7px; padding: 12px 10px; border: 1px solid transparent; border-radius: 12px; color: var(--ink-soft); } +.session-item:hover { background: #fff8; } +.session-item.active { background: var(--white); border-color: #d9d0e4; } +.session-item strong { overflow-wrap: anywhere; font-size: 11px; } +.session-item small { grid-column: 1 / -1; color: var(--muted); font-size: 9px; } +.session-state { align-self: center; width: 6px; height: 6px; border-radius: 50%; background: var(--lilac); } +.session-state.active { background: var(--sage); box-shadow: 0 0 0 3px var(--sage-soft); } +.session-state.failed, .session-state.interrupted { background: var(--peach); } +.terminal-panel { min-width: 0; overflow: hidden; border-radius: 17px; background: #25222d; color: #e8e3ef; box-shadow: var(--shadow-card); } +.terminal-toolbar { min-height: 56px; display: flex; flex-wrap: wrap; align-items: center; justify-content: space-between; gap: 12px; padding: 12px 16px; border-bottom: 1px solid #ffffff12; background: #302c39; } +.terminal-toolbar > div:first-child { display: flex; flex-wrap: wrap; align-items: center; gap: 10px; } +.terminal-toolbar strong { font-size: 11px; } +.terminal-status { color: #b8afc6; font-size: 9px; } +.terminal-toolbar .button { color: #e8e3ef; background: #ffffff0c; border-color: #ffffff17; } +.terminal-toolbar .button:hover { background: #ffffff19; } +.terminal-lights { display: inline-flex; gap: 4px; margin-right: 3px; } +.terminal-lights i { height: 6px; width: 6px; border-radius: 50%; background: var(--peach); } +.terminal-lights i:nth-child(2) { background: var(--yellow); } +.terminal-lights i:nth-child(3) { background: var(--sage); } +.session-terminal { height: 480px; padding: 14px 12px 10px; overflow: hidden; } +.session-terminal .xterm { height: 100%; } +.session-terminal .xterm-viewport { scrollbar-width: thin; scrollbar-color: #766a88 #25222d; } +.terminal-blank { min-height: 480px; display: flex; flex-direction: column; align-items: center; justify-content: center; gap: 17px; text-align: center; padding: 24px; } +.terminal-blank > .icon { width: 37px; height: 37px; color: #b3a5ca; } +.terminal-blank h3 { font-size: 20px; font-weight: 450; } +.terminal-blank p { color: #b2a9bf; font-size: 11px; line-height: 1.8; } +.terminal-footer { min-height: 43px; padding: 11px 16px; color: #b5abc4; font-size: 9px; border-top: 1px solid #ffffff0d; overflow-wrap: anywhere; } +.terminal-support { margin-top: 18px; padding: 15px 19px; font-size: 10px; color: var(--muted); } +.terminal-support summary { cursor: pointer; color: var(--ink-soft); } +.terminal-support p { margin: 12px 0; max-width: 680px; line-height: 1.8; } +.terminal-support .muted { margin-left: 12px; } +@media (max-width: 1100px) { + .session-workspace { grid-template-columns: minmax(0, 1fr); } + .session-list { grid-template-columns: repeat(auto-fill, minmax(150px, 1fr)); max-height: 190px; } + .session-history > .panel-footnote { display: none; } +} +@media (max-width: 700px) { + .session-terminal { height: 400px; } + .terminal-blank { min-height: 330px; } + .session-section-heading { flex-wrap: wrap; } + .terminal-support .muted { display: block; margin: 10px 0 0; } +} diff --git a/nest/dashboard/static/index.html b/nest/dashboard/static/index.html index a635bf0..63c6c32 100644 --- a/nest/dashboard/static/index.html +++ b/nest/dashboard/static/index.html @@ -9,8 +9,13 @@ + - + + + + + diff --git a/pyproject.toml b/pyproject.toml index aa7b709..839c531 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,4 +26,4 @@ packages = ["nest", "nest.dashboard", "uvm", "uvm.firecracker", "uvm.routers"] include-package-data = false [tool.setuptools.package-data] -"nest.dashboard" = ["static/*.html", "static/css/*.css", "static/js/*.js", "static/fonts/*.woff2", "static/fonts/*.txt"] +"nest.dashboard" = ["static/*.html", "static/css/*.css", "static/js/*.js", "static/fonts/*.woff2", "static/fonts/*.txt", "static/vendor/*.md", "static/vendor/*/LICENSE", "static/vendor/*/lib/*.js", "static/vendor/*/css/*.css"] diff --git a/tests/test_guest_terminal.py b/tests/test_guest_terminal.py new file mode 100644 index 0000000..64b82df --- /dev/null +++ b/tests/test_guest_terminal.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import asyncio +import os +import pwd +import socket +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from unittest.mock import Mock, patch + +from tests.test_guest_exec import FakeScope +from tests.test_lifecycle import LifecycleTestCase +from uvm import guest_agent as agent +from uvm.errors import ConflictError, GuestAgentUnavailableError, NotFoundError +from uvm.guest_terminal import GuestTerminal + + +class GuestTerminalTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.scope = FakeScope() + self.addCleanup(self.scope.temporary.cleanup) + self.home = self.enterContext(tempfile.TemporaryDirectory()) + username = pwd.getpwuid(os.geteuid()).pw_name + original_popen = subprocess.Popen + # Exercise the actual guest child entry point with a temporary account + # home, so interactive shells never read/write the developer's dotfiles. + script = f""" +import os, runpy +from pathlib import Path +from types import SimpleNamespace +agent = runpy.run_path({agent.__file__!r}) +agent['pwd'].getpwnam = lambda _: SimpleNamespace(pw_name={username!r}, pw_uid=os.geteuid(), + pw_gid=os.getegid(), pw_dir={self.home!r}, pw_shell='/bin/sh') +agent['child_main'](Path({str(self.scope.path)!r}), {username!r}, None) +""" + + def launch(_args, **kwargs): + return original_popen([sys.executable, "-I", "-c", script], **kwargs) + + self.enterContext(patch.object(agent.subprocess, "Popen", side_effect=launch)) + left, right = socket.socketpair() + self.terminal = GuestTerminal(left) + self.addCleanup(self.terminal.close) + + def runner(connection, account, rows, cols): + return agent.run_terminal(connection, account, rows, cols, scope_factory=lambda: self.scope) + + self.thread = threading.Thread(target=agent.serve_connection, + args=(right, {"vm_id": "vm-test", "username": username}), + kwargs={"terminal_runner": runner}, daemon=True) + self.thread.start() + self.addCleanup(self.thread.join, 3) + agent.receive_frame(left, 4096, time.monotonic() + 2) + await asyncio.to_thread(self.terminal.start, 24, 80) + self.prompt = b"# " if os.geteuid() == 0 else b"$ " + await self.until(self.prompt) + + async def until(self, wanted): + data = bytearray() + try: + async with asyncio.timeout(5): + while wanted not in data: + message = await self.terminal.read() + self.assertEqual(message["type"], "output", message) + data.extend(message["data"]) + except TimeoutError: + self.fail(f"Missing {wanted!r} in terminal output: {bytes(data)!r}") + return bytes(data) + + async def asyncTearDown(self): + self.terminal.close() + await asyncio.to_thread(self.thread.join, 3) + self.assertFalse(self.thread.is_alive()) + + async def test_terminal_has_tty_persistent_shell_state_and_exit_code(self): + await self.terminal.write(b"test -t 0 && printf 'REAL-TTY\\n'\r") + await self.until(b"\r\nREAL-TTY\r\n") + await self.terminal.write(b"export NEST_TEST=hello; cd /\r") + await self.terminal.write(b"printf 'STATE:%s:%s\\n' \"$NEST_TEST\" \"$PWD\"\r") + await self.until(b"STATE:hello:/\r\n") + await self.terminal.write(b"exit 7\r") + async with asyncio.timeout(5): + while True: + message = await self.terminal.read() + if message["type"] == "exit": + self.assertEqual(message["exit_code"], 7) + break + + async def test_resize_and_control_c_reach_the_guest_foreground_job(self): + await self.terminal.resize(40, 120) + await self.terminal.write(b"stty size\r") + await self.until(b"40 120\r\n") + await self.terminal.write(b"sleep 20\r") + await asyncio.sleep(0.1) + await self.terminal.write(b"\x03") + await self.until(self.prompt) + await self.terminal.write(b"printf 'INTERRUPTED\\n'\r") + await self.until(b"\r\nINTERRUPTED\r\n") + + async def test_explicit_end_terminates_the_shell(self): + await self.terminal.write(b"printf 'READY\\n'\r") + await self.until(b"\r\nREADY\r\n") + await self.terminal.end() + async with asyncio.timeout(5): + while (message := await self.terminal.read())["type"] != "exit": + pass + self.assertIsInstance(message["exit_code"], int) + self.assertGreater(self.scope.kills, 0) + + +class TerminalCompatibilityTests(unittest.TestCase): + def test_old_agent_rejection_explains_how_to_upgrade(self): + left, right = socket.socketpair() + terminal = GuestTerminal(left) + self.addCleanup(terminal.close) + + def old_agent(): + with right: + agent.receive_frame(right, 4096, time.monotonic() + 2) + agent.send_frame(right, {"error": "invalid execution request fields", "outcome_unknown": False}, + 4096, time.monotonic() + 2) + + thread = threading.Thread(target=old_agent) + thread.start() + try: + with self.assertRaisesRegex(GuestAgentUnavailableError, "Install / update terminal support"): + terminal.start(24, 80) + finally: + thread.join(3) + + +class TerminalLifecycleTests(LifecycleTestCase): + def test_terminal_connects_under_lock_and_opens_without_holding_the_lock(self): + vm = self.create_vm(name="worker", ports=()) + terminal = Mock() + + def connect(path, vm_id, username, pid): + self.assertTrue(self.store.operation_active) + self.assertEqual((path, vm_id, username, pid), (self.settings.guest_socket_path(vm.id), vm.id, vm.username, vm.pid)) + return terminal + + def start(rows, cols): + self.assertFalse(self.store.operation_active) + self.assertEqual((rows, cols), (40, 120)) + + self.service._terminal_connector = Mock(side_effect=connect) + terminal.start.side_effect = start + self.assertEqual(self.service.open_terminal(vm.id, rows=40, cols=120), (terminal, vm.username)) + terminal.close.assert_not_called() + + def test_aliases_and_stopped_vms_cannot_create_a_terminal(self): + vm = self.create_vm(name="worker") + self.service._terminal_connector = Mock() + for alias in (vm.name, vm.guest_ip): + with self.assertRaises(NotFoundError): + self.service.open_terminal(alias) + self.service.stop(vm.id) + with self.assertRaises(ConflictError): + self.service.open_terminal(vm.id) + self.service._terminal_connector.assert_not_called() diff --git a/tests/test_sessions.py b/tests/test_sessions.py new file mode 100644 index 0000000..afe24da --- /dev/null +++ b/tests/test_sessions.py @@ -0,0 +1,267 @@ +from __future__ import annotations + +import asyncio +import base64 +import os +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace + +from nest.server import create_app +from nest.session_store import SessionStore +from nest.sessions import SessionManager +from tests.test_server import FASTAPI_AVAILABLE, fake_application +from uvm.errors import ConflictError, GuestAgentUnavailableError, NotFoundError + + +if FASTAPI_AVAILABLE: + from fastapi.testclient import TestClient + from starlette.websockets import WebSocketDisconnect + + +class FakeTerminal: + def __init__(self) -> None: + self.events = asyncio.Queue() + self.events.put_nowait({"type": "output", "data": b"guest$ "}) + self.input = [] + self.sizes = [] + self.closed = False + + async def read(self): + return await self.events.get() + + async def write(self, data): + self.input.append(data) + await self.events.put({"type": "output", "data": data + b"\r\nguest$ "}) + + async def resize(self, rows, cols): + self.sizes.append((rows, cols)) + + async def end(self): + await self.events.put({"type": "exit", "exit_code": 0}) + + def close(self): + self.closed = True + + +def session_application(base): + application = fake_application(api_token="test-token") + application.settings.base = Path(base) + application.terminals = [] + application.vm_ids = {"vm-test", "vm-other"} + + def get(vm_id): + if vm_id not in application.vm_ids: + raise NotFoundError("VM not found") + return SimpleNamespace(id=vm_id) + + def open_terminal(vm_id, **_size): + get(vm_id) + terminal = FakeTerminal() + application.terminals.append(terminal) + return terminal, "worker" + + application.state_store = SimpleNamespace(get=get) + application.lifecycle = SimpleNamespace(open_terminal=open_terminal) + return application + + +class SessionStoreTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.path = Path(temporary.name) / "nest" / "sessions.sqlite3" + self.store = SessionStore(self.path, max_output=8) + self.addCleanup(self.store.close) + + def test_history_is_scoped_by_exact_vm_id_and_survives_reopening(self): + record = self.store.create("vm-a", 24, 80) + self.store.activate(record["id"], "worker") + self.store.append(record["id"], b"hello") + self.store.finish(record["id"], "closed", 7) + self.assertEqual(self.store.list("vm-b"), []) + with self.assertRaises(NotFoundError): + self.store.get("vm-b", record["id"]) + self.store.close() + reopened = SessionStore(self.path) + self.addCleanup(reopened.close) + saved, start, output = reopened.read("vm-a", record["id"], 0) + self.assertEqual((saved["status"], saved["exit_code"], saved["username"], start, output), ("closed", 7, "worker", 0, b"hello")) + self.assertEqual(self.path.stat().st_mode & 0o777, 0o600) + self.assertEqual(self.path.parent.stat().st_mode & 0o777, 0o700) + + def test_restarted_server_marks_only_live_sessions_interrupted(self): + live = self.store.create("vm-a", 24, 80) + ended = self.store.create("vm-a", 24, 80) + self.store.activate(live["id"], "worker") + self.store.finish(ended["id"], "closed", 0) + self.store.close() + reopened = SessionStore(self.path) + self.addCleanup(reopened.close) + self.assertEqual(reopened.get("vm-a", live["id"])["status"], "interrupted") + self.assertIsNotNone(reopened.get("vm-a", live["id"])["ended_at"]) + self.assertEqual(reopened.get("vm-a", ended["id"])["status"], "closed") + + def test_transcript_is_bounded_and_offsets_support_exact_replay(self): + record = self.store.create("vm-a", 24, 80) + self.store.append(record["id"], b"abcdefghij") + self.store.append(record["id"], b"kl") + saved, start, output = self.store.read("vm-a", record["id"], 0) + self.assertEqual((start, output, saved["output_end"], saved["truncated"]), (4, b"efghijkl", 12, True)) + self.assertEqual(self.store.read("vm-a", record["id"], 9)[1:], (9, b"jkl")) + self.assertEqual(self.store.read("vm-a", record["id"], 12)[2], b"") + + def test_second_server_cannot_reconcile_another_servers_live_sessions(self): + record = self.store.create("vm-a", 24, 80) + self.store.activate(record["id"], "worker") + with self.assertRaises(ConflictError): + SessionStore(self.path) + self.assertEqual(self.store.get("vm-a", record["id"])["status"], "active") + + def test_symlinked_history_file_is_not_opened_or_rewritten(self): + self.store.close() + self.path.unlink() + target = self.path.parent / "unrelated" + target.write_bytes(b"keep this") + self.path.symlink_to(target) + with self.assertRaises(OSError): + SessionStore(self.path) + self.assertEqual(target.read_bytes(), b"keep this") + + +class SessionManagerTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.application = session_application(self.temporary.name) + self.manager = SessionManager(self.application) + self.addAsyncCleanup(self.manager.shutdown) + + async def test_failed_launch_is_saved_in_history(self): + def unavailable(*_args, **_kwargs): + raise GuestAgentUnavailableError("update the guest agent") + + self.application.lifecycle.open_terminal = unavailable + with self.assertRaises(GuestAgentUnavailableError): + await self.manager.create("vm-test", 24, 80) + history = await self.manager.list("vm-test") + self.assertEqual(history[0]["status"], "failed") + self.assertEqual(history[0]["error"], "update the guest agent") + + async def test_shutdown_closes_live_channels_and_saves_interrupted_history(self): + record = await self.manager.create("vm-test", 24, 80) + await self.manager.shutdown() + self.assertTrue(self.application.terminals[0].closed) + store = SessionStore(Path(self.temporary.name) / "nest" / "sessions.sqlite3") + self.addCleanup(store.close) + self.assertEqual(store.get("vm-test", record["id"])["status"], "interrupted") + + async def test_cancelling_the_create_request_does_not_orphan_the_shell(self): + import threading + entered, release = threading.Event(), threading.Event() + connect = self.application.lifecycle.open_terminal + + def slow_connect(*args, **kwargs): + entered.set() + release.wait(2) + return connect(*args, **kwargs) + + self.application.lifecycle.open_terminal = slow_connect + request = asyncio.create_task(self.manager.create("vm-test", 24, 80)) + await asyncio.to_thread(entered.wait, 2) + request.cancel() + release.set() + await asyncio.gather(request, return_exceptions=True) + await self.manager.shutdown() + self.assertTrue(self.application.terminals[0].closed) + + +@unittest.skipUnless(FASTAPI_AVAILABLE, "FastAPI and httpx are required") +class SessionApiTests(unittest.TestCase): + def setUp(self): + self.temporary = self.enterContext(tempfile.TemporaryDirectory()) + self.application = session_application(self.temporary) + self.api = create_app(self.application, host="127.0.0.1") + self.client = self.enterContext(TestClient(self.api)) + self.headers = {"X-UVM-Token": "test-token"} + self.base = "/vms/vm-test/sessions" + + def create(self): + response = self.client.post(self.base, headers=self.headers, json={}) + self.assertEqual(response.status_code, 201, response.text) + return response.json() + + def connect(self, record, *, vm_id="vm-test", offset=0): + ws = self.enterContext(self.client.websocket_connect(f"/vms/{vm_id}/sessions/{record['id']}/terminal")) + ws.send_json({"type": "auth", "token": "test-token", "offset": offset}) + return ws + + def output(self, ws): + while True: + message = ws.receive_json() + if message["type"] == "output": + data = base64.b64decode(message["data"]) + end = message["offset"] + len(data) + ws.send_json({"type": "ack", "offset": end}) + return data, end + self.assertEqual(message["type"], "session", message) + + def test_all_http_operations_and_websocket_output_require_authentication(self): + for method, path in (("GET", self.base), ("POST", self.base), + ("GET", self.base + "/session-unknown"), ("POST", self.base + "/session-unknown/close")): + self.assertEqual(self.client.request(method, path, json={}).status_code, 401) + for token in ("wrong", "", "\u00e9"): + with self.client.websocket_connect(self.base + "/session-unknown/terminal") as ws: + ws.send_json({"type": "auth", "token": token}) + with self.assertRaises(WebSocketDisconnect) as raised: + ws.receive_json() + self.assertEqual(raised.exception.code, 4401) + self.assertEqual(self.application.terminals, []) + self.assertFalse((Path(self.temporary) / "nest").exists()) + + def test_foreign_browser_origin_is_rejected(self): + with self.assertRaises(WebSocketDisconnect): + with self.client.websocket_connect(self.base + "/unknown/terminal", headers={"Origin": "https://foreign.example"}): + self.fail("foreign origin accepted") + + def test_create_rejects_unknown_vm_aliases_and_invalid_sizes(self): + for body in ({"rows": 0}, {"cols": True}, {"cols": 401}, {"command": "id"}): + self.assertEqual(self.client.post(self.base, headers=self.headers, json=body).status_code, 422) + self.assertEqual(self.client.post("/vms/name-alias/sessions", headers=self.headers, json={}).status_code, 404) + self.assertEqual(self.application.terminals, []) + + def test_sessions_are_nested_under_the_selected_vm(self): + record = self.create() + self.assertEqual(record["username"], "worker") + other = "/vms/vm-other/sessions" + self.assertEqual(self.client.get(other, headers=self.headers).json(), []) + self.assertEqual(self.client.get(other + "/" + record["id"], headers=self.headers).status_code, 404) + self.assertEqual(self.client.post(other + "/" + record["id"] + "/close", headers=self.headers).status_code, 404) + ws = self.connect(record, vm_id="vm-other") + self.assertEqual(ws.receive_json()["type"], "error") + self.assertFalse(self.application.terminals[0].closed) + + def test_disconnect_reconnect_input_resize_and_closed_history(self): + record = self.create() + ws = self.connect(record) + output, offset = self.output(ws) + self.assertEqual(output, b"guest$ ") + ws.send_json({"type": "resize", "rows": 35, "cols": 120}) + ws.send_json({"type": "input", "data": base64.b64encode(b"pwd\r").decode()}) + output, offset = self.output(ws) + self.assertEqual(output, b"pwd\r\r\nguest$ ") + self.assertEqual(self.application.terminals[0].sizes, [(35, 120)]) + ws.close() + self.assertFalse(self.application.terminals[0].closed) + reopened = self.connect(record, offset=offset) + self.assertEqual(reopened.receive_json()["session"]["status"], "active") + self.assertEqual(len(self.application.terminals), 1) + closed = self.client.post(self.base + "/" + record["id"] + "/close", headers=self.headers) + self.assertEqual(closed.json()["status"], "closed") + self.assertEqual(closed.json()["exit_code"], 0) + replay = self.connect(record) + transcript, _end = self.output(replay) + self.assertEqual(transcript, b"guest$ pwd\r\r\nguest$ ") + self.assertEqual(self.client.get(self.base, headers=self.headers).json()[0]["status"], "closed") + self.assertTrue(self.application.terminals[0].closed)