j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-11 12:46:49 +00:00
Code Issues Projects Releases Wiki Activity

fix(console): an SSH compute's VNC console goes through ssh -W

Browse source 
- The tunnel is `ssh -W` to the console's address or Unix socket: the
  compute needs no nc and runs no shell. The old tunnel, copied from
  virt-manager, built a shell script with the console's address and socket
  path in it and eval'd it on the compute, and always passed "-p 22",
  overriding a Port in the SSH configuration.
- On an SSH compute the console always goes through the tunnel, also when
  it listens on a network address: it was sent there unencrypted.
- ssh never prompts (BatchMode) and gives up connecting after 10 seconds;
  when it fails, what it said is logged. The tunnel ends with the session.
This commit is contained in:
catborise 2026-10-26 19:04:58 +03:00
parent 504d86faa3
commit 294b008811
4 changed files with 130 additions and 301 deletions
Download patch file Download diff file Expand all files Collapse all files

42
console/tests/test_sshtunnels.py Normal file
View file

@ -0,0 +1,42 @@
"""The SSH console tunnel: ssh -W to the console's address or socket, never a
prompt, no remote shell, and its error output when it fails."""
import sys
from unittest import TestCase
from unittest.mock import patch
from console import sshtunnels
class SshTunnelTests(TestCase):
def test_ssh_forwards_to_the_console_without_a_shell_or_a_prompt(self):
with patch.object(sshtunnels.subprocess, "Popen") as popen:
sock, process = sshtunnels.open_tunnel("kvm1.example.com", "root", "127.0.0.1:5900")
sock.close()
self.assertEqual(popen.call_args.args[0], [
"ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10", "-W", "127.0.0.1:5900",
"-l", "root", "--", "kvm1.example.com"])
self.assertIs(process, popen.return_value)
def test_the_target_on_the_compute(self):
for address, unix_socket, target in (
("192.0.2.40", None, "192.0.2.40:5900"),
("2001:db8::40", None, "[2001:db8::40]:5900"),
(None, None, "127.0.0.1:5900"),
("0.0.0.0", None, "127.0.0.1:5900"),
("::", None, "[::1]:5900"),
("localhost", None, "localhost:5900"),
("127.0.0.1", "/run/vnc.sock", "/run/vnc.sock"),
):
with self.subTest(address=address, unix_socket=unix_socket):
self.assertEqual(sshtunnels.tunnel_target(address, 5900, unix_socket), target)
def test_a_failed_tunnel_reports_what_ssh_said(self):
failing = sshtunnels.subprocess.Popen(
[sys.executable, "-c", "import sys; sys.stderr.write('Host key verification failed.'); sys.exit(255)"],
stderr=sshtunnels.subprocess.PIPE)
failing.wait() # ssh fails at once, before the session
self.assertEqual(sshtunnels.close_tunnel(failing), "Host key verification failed.")
ended = sshtunnels.subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"],
stderr=sshtunnels.subprocess.PIPE)
self.assertEqual(sshtunnels.close_tunnel(ended), "") # ended with the session: no error

33
console/tests/test_views.py
View file

@ -506,7 +506,7 @@ class NovncdDaemonLogicTestCase(TestCase):
@patch.object(novncd_mod, "get_connection_infos")
def test_a_cookie_does_not_choose_the_console(self, mock_get_info):
# another tab's console in a cookie: the URL's own token is used
mock_get_info.return_value = ("host1", 22, "root", 1, "192.168.1.50", 5900, None)
mock_get_info.return_value = ("host1", "root", 1, "192.168.1.50", 5900, None)
handler = self.CompatibilityMixIn()
handler.headers = {"cookie": "token=1-uuid-of-another-tab"}
handler.path = "/novncd/?token=1-uuid-of-this-tab"
@ -520,7 +520,6 @@ class NovncdDaemonLogicTestCase(TestCase):
def test_compatibility_mixin_token_from_query_params(self, mock_get_info):
mock_get_info.return_value = (
"host2",
22,
"root",
1,
"192.168.1.50",
@ -544,7 +543,7 @@ class NovncdDaemonLogicTestCase(TestCase):
@patch.object(novncd_mod, "get_connection_infos")
def test_a_local_socket_host_reaches_a_console_on_its_loopback(self, mock_get_info):
# libvirt on the panel's host, the VM's console listening on 127.0.0.1
mock_get_info.return_value = ("localhost", None, "", novncd_mod.CONN_SOCKET, "127.0.0.1", 5902, None)
mock_get_info.return_value = ("localhost", "", novncd_mod.CONN_SOCKET, "127.0.0.1", 5902, None)
handler = self.CompatibilityMixIn()
handler.headers = {}
handler.path = "/novncd/?token=1-uuid"
@ -555,6 +554,32 @@ class NovncdDaemonLogicTestCase(TestCase):
handler._new_client(daemon=False, socket_factory=socket_factory)
socket_factory.assert_called_once_with("127.0.0.1", 5902, connect=True)
def test_an_ssh_hosts_console_always_goes_through_the_tunnel(self):
# whatever address the console listens on, the VNC stream is not sent unencrypted
for console_host, console_socket, target in (
("192.0.2.40", None, "192.0.2.40:5900"),
("0.0.0.0", None, "127.0.0.1:5900"),
("::", None, "[::1]:5900"),
("127.0.0.1", None, "127.0.0.1:5900"),
(None, "/run/libvirt/qemu/vnc.sock", "/run/libvirt/qemu/vnc.sock"),
):
with self.subTest(console_host=console_host), patch.object(
novncd_mod, "get_connection_infos",
return_value=("kvm1.example.com", "root", novncd_mod.CONN_SSH, console_host, 5900, console_socket),
), patch.object(novncd_mod, "open_tunnel", return_value=(MagicMock(), MagicMock())) as open_tunnel, \
patch.object(novncd_mod, "close_tunnel", return_value="") as close_tunnel:
handler = self.CompatibilityMixIn()
handler.path = "/novncd/?token=1-uuid"
handler.msg = MagicMock()
handler.do_proxy = MagicMock()
handler.verbose = False
socket_factory = MagicMock()
handler._new_client(daemon=False, socket_factory=socket_factory)
open_tunnel.assert_called_once_with("kvm1.example.com", "root", target)
socket_factory.assert_not_called()
# the tunnel ends with the session
close_tunnel.assert_called_once_with(open_tunnel.return_value[1])
@patch.object(novncd_mod, "get_connection_infos")
def test_compatibility_mixin_missing_token_safe_exit(self, mock_get_info):
handler = self.CompatibilityMixIn()
@ -594,7 +619,7 @@ class NovncdDaemonLogicTestCase(TestCase):
infos = novncd_mod.get_connection_infos(token)
self.assertEqual(infos[0], "127.0.0.1")
self.assertEqual(infos[5], 5900)
self.assertEqual(infos[4], 5900)
# the VM access was checked for, not whichever VM has its name now
self.assertEqual(mock_wvm.call_args.kwargs["uuid"], mock_instance.uuid)