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

test: each app keeps its tests in a tests package

Browse source 
131 test modules sat next to the code they test; instances alone had
51 of them beside views.py and utils.py. They move into a
tests/ package per app, and the JavaScript checks into
webvirtcloud/tests/js. Each non-empty tests.py becomes tests/test_views.py;
the empty generated stubs are removed.

Imports of the app's own modules become absolute, paths built from
__file__ go one level up, the live helper is instances.tests.livetest,
and coverage omits */tests/*. The suite discovers the same 960 tests as
before.
This commit is contained in:
catborise 2026-10-12 10:37:29 +03:00
parent 0906838193
commit c55f52f025
154 changed files with 27 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

0
console/tests/__init__.py Normal file
View file

57
console/tests/test_cli_parsers.py Normal file
View file

@ -0,0 +1,57 @@
import argparse
import unittest
class TestCLIParsers(unittest.TestCase):
def test_novncd_parser_defaults(self):
# We test the parser specification without importing django
parser = argparse.ArgumentParser(description="WebVirtCloud noVNC daemon")
parser.add_argument("-v", "--verbose", dest="verbose", action="store_true", default=False)
parser.add_argument("-d", "--debug", dest="debug", action="store_true", default=False)
parser.add_argument("-H", "--host", dest="host", action="store", default="0.0.0.0")
parser.add_argument("-p", "--port", dest="port", action="store", type=int, default=6080)
parser.add_argument("-c", "--cert", dest="cert", action="store", default="cert.pem")
opts = parser.parse_args([])
self.assertFalse(opts.verbose)
self.assertFalse(opts.debug)
self.assertEqual(opts.host, "0.0.0.0")
self.assertEqual(opts.port, 6080)
self.assertEqual(opts.cert, "cert.pem")
def test_novncd_parser_custom_args(self):
parser = argparse.ArgumentParser(description="WebVirtCloud noVNC daemon")
parser.add_argument("-v", "--verbose", dest="verbose", action="store_true", default=False)
parser.add_argument("-d", "--debug", dest="debug", action="store_true", default=False)
parser.add_argument("-H", "--host", dest="host", action="store", default="0.0.0.0")
parser.add_argument("-p", "--port", dest="port", action="store", type=int, default=6080)
parser.add_argument("-c", "--cert", dest="cert", action="store", default="cert.pem")
opts = parser.parse_args(["-v", "-d", "-H", "192.168.1.50", "-p", "7000", "-c", "/etc/ssl/cert.pem"])
self.assertTrue(opts.verbose)
self.assertTrue(opts.debug)
self.assertEqual(opts.host, "192.168.1.50")
self.assertEqual(opts.port, 7000)
self.assertEqual(opts.cert, "/etc/ssl/cert.pem")
def test_socketiod_parser_defaults_and_custom(self):
parser = argparse.ArgumentParser(description="WebVirtCloud Socket.IO daemon")
parser.add_argument("-v", "--verbose", dest="verbose", action="store_true", default=False)
parser.add_argument("-d", "--debug", dest="debug", action="store_true", default=False)
parser.add_argument("-H", "--host", dest="host", action="store", default="localhost")
parser.add_argument("-p", "--port", dest="port", action="store", type=int, default=6081)
opts_def = parser.parse_args([])
self.assertFalse(opts_def.verbose)
self.assertFalse(opts_def.debug)
self.assertEqual(opts_def.host, "localhost")
self.assertEqual(opts_def.port, 6081)
opts_custom = parser.parse_args(["--verbose", "--host", "10.0.0.1", "--port", "9000"])
self.assertTrue(opts_custom.verbose)
self.assertEqual(opts_custom.host, "10.0.0.1")
self.assertEqual(opts_custom.port, 9000)
if __name__ == "__main__":
unittest.main()

118
console/tests/test_socketiod_child.py Normal file
View file

@ -0,0 +1,118 @@
"""socketiod's console process ending on its own: the clients are told and
disconnected, the process is reaped and the next connection starts a new one."""
import os
import socket
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from pathlib import Path
try:
import socketio
except ImportError: # pragma: no cover
socketio = None
ROOT = Path(__file__).resolve().parent.parent.parent
SOCKETIOD = ROOT / "console" / "socketiod"
def free_port():
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
def children(pid):
"""(pid, state) of the processes whose parent is pid."""
found = []
for entry in os.listdir("/proc"):
try:
with open("/proc/%s/stat" % entry) as f:
fields = f.read().rsplit(")", 1)[1].split()
except (OSError, IndexError):
continue
if int(fields[1]) == pid:
found.append((int(entry), fields[0]))
return found
@unittest.skipIf(socketio is None, "python-socketio is not installed")
@unittest.skipUnless(sys.platform.startswith("linux"), "reads /proc")
class SocketiodChildExitTestCase(unittest.TestCase):
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
(Path(tmp.name) / "socketiod_child_settings.py").write_text(
"from webvirtcloud.settings import * # noqa\n"
"SERIAL_CONSOLE_ENABLED = True\n"
"DATABASES = {'default': {'ENGINE': 'django.db.backends.sqlite3', 'NAME': %r}}\n"
% str(Path(tmp.name) / "db.sqlite3")
)
self.port = free_port()
env = dict(
os.environ,
DJANGO_SETTINGS_MODULE="socketiod_child_settings",
PYTHONPATH=os.pathsep.join([tmp.name, str(ROOT)]),
)
self.server = subprocess.Popen(
[sys.executable, str(SOCKETIOD), "-H", "127.0.0.1", "-p", str(self.port)],
cwd=ROOT,
env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
self.addCleanup(self.stop_server)
end = time.monotonic() + 20
while time.monotonic() < end:
with socket.socket() as s:
if s.connect_ex(("127.0.0.1", self.port)) == 0:
return
self.assertIsNone(self.server.poll(), "socketiod exited at start")
time.sleep(0.1)
self.fail("socketiod did not listen within 20 s")
def stop_server(self):
self.server.kill()
self.server.wait()
def open_console(self):
"""Connect with a malformed token, so the console process fails at once;
returns the output and whether the server disconnected the client."""
client = socketio.Client(reconnection=False)
output = []
gone = threading.Event()
client.on("pty_output", lambda data: output.append(data["output"]))
client.on("disconnect", lambda *args: gone.set())
client.connect(
"http://127.0.0.1:%d" % self.port,
headers={"Cookie": "token=no-such-vm"},
transports=["polling"],
)
try:
gone.wait(15)
finally:
if not gone.is_set():
client.disconnect()
return "".join(output), gone.is_set()
def assert_no_child_left(self):
end = time.monotonic() + 5
while children(self.server.pid) and time.monotonic() < end:
time.sleep(0.05)
self.assertEqual(children(self.server.pid), [])
def test_the_client_is_told_and_the_next_connection_starts_again(self):
for attempt in (1, 2):
with self.subTest(attempt=attempt):
output, disconnected = self.open_console()
self.assertIn("console closed", output)
# no traceback of socketiod's own code (on Python 3.13 eventlet
# prints an ignored _after_fork error in every forked child)
self.assertNotIn('socketiod", line', output)
self.assertTrue(disconnected)
self.assert_no_child_left()
self.assertIsNone(self.server.poll())

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

@ -0,0 +1,42 @@
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent.parent
SOCKETIOD = ROOT / "console" / "socketiod"
class SocketiodDisabledTestCase(unittest.TestCase):
"""socketiod must refuse to start unless explicitly enabled."""
def test_socketiod_exits_when_serial_console_disabled(self):
# The settings module lives in a temporary directory, so the test does
# not write into the source tree (read-only checkouts, --parallel).
with tempfile.TemporaryDirectory() as tmp:
(Path(tmp) / "socketiod_disabled_settings.py").write_text(
"from webvirtcloud.settings import * # noqa\n"
"SERIAL_CONSOLE_ENABLED = False\n"
"SOCKETIO_HOST = '127.0.0.1'\n"
"SOCKETIO_PORT = 0\n"
)
env = dict(
os.environ,
DJANGO_SETTINGS_MODULE="socketiod_disabled_settings",
PYTHONPATH=os.pathsep.join([tmp, str(ROOT)]),
)
try:
result = subprocess.run(
[sys.executable, str(SOCKETIOD)],
cwd=ROOT,
env=env,
capture_output=True,
text=True,
timeout=20,
)
except subprocess.TimeoutExpired:
self.fail("socketiod kept running although SERIAL_CONSOLE_ENABLED=False")
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn("SERIAL_CONSOLE_ENABLED", result.stderr)

688
console/tests/test_views.py Normal file
View file

@ -0,0 +1,688 @@
# pylint: disable=no-member
import importlib.machinery
import importlib.util
from pathlib import Path
from unittest.mock import MagicMock, patch
from accounts.models import UserInstance
from computes.models import Compute
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.core.signing import TimestampSigner
from django.test import TestCase, override_settings
from django.urls import reverse
from django.utils.html import escapejs
from instances.models import Instance
from libvirt import libvirtError
# Dynamically load console/novncd script for testing
NOVNCD_PATH = str(Path(__file__).resolve().parent.parent / "novncd")
_loader = importlib.machinery.SourceFileLoader("novncd_mod", NOVNCD_PATH)
_spec = importlib.util.spec_from_file_location(
"novncd_mod", NOVNCD_PATH, loader=_loader
)
assert _spec is not None
assert _spec.loader is not None
novncd_mod = importlib.util.module_from_spec(_spec)
_loader.exec_module(novncd_mod)
class ConsoleViewsTestCase(TestCase):
@classmethod
def setUpTestData(cls):
User = get_user_model()
cls.admin_user = (
User.objects.filter(is_superuser=True).first()
or User.objects.filter(username="admin").first()
or User.objects.create_superuser(
username="admin_console_test",
email="admin@example.com",
password="adminpassword",
)
)
cls.alice = User.objects.create_user(
username="alice", email="alice@example.com", password="alicepassword"
)
cls.bob = User.objects.create_user(
username="bob", email="bob@example.com", password="bobpassword"
)
cls.compute = Compute.objects.create(
name="test-compute",
hostname="127.0.0.1",
login="root",
password="",
type=1,
)
cls.instance = Instance.objects.create(
compute=cls.compute,
name="vm-alpha",
uuid="11111111-2222-3333-4444-555555555555",
)
# Grant alice ownership of vm-alpha
UserInstance.objects.create(
user=cls.alice,
instance=cls.instance,
)
cls.token = f"{cls.compute.id}-{cls.instance.uuid}"
@patch("console.views.wvmInstance")
def test_console_superuser_vnc_lite(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_conn.get_console_websocket_port.return_value = None
mock_conn.get_console_passwd.return_value = "vncsecret"
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
url = reverse("console") + f"?token={self.token}&view=lite"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertTemplateUsed(response, "console-vnc-lite.html")
signer = TimestampSigner(salt="console.novnc")
unsigned = signer.unsign(response.cookies["token"].value, max_age=120)
self.assertEqual(
unsigned, f"{self.compute.id}:{self.instance.uuid}:{self.admin_user.id}"
)
self.assertEqual(response.context["ws_path"], "novncd/")
self.assertEqual(response.context["ws_port"], 6080)
self.assertEqual(response.context["console_passwd"], "vncsecret")
@patch("console.views.wvmInstance")
def test_console_superuser_vnc_full(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_conn.get_console_websocket_port.return_value = 5900
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
url = reverse("console") + f"?token={self.token}&view=full"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertTemplateUsed(response, "console-vnc-full.html")
self.assertEqual(response.context["ws_port"], 5900)
@patch("console.views.wvmInstance")
def test_console_unsupported_graphics_type(self, mock_wvm):
# Only VNC is supported; a VM defined elsewhere may use another type.
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "rdp"
mock_conn.get_console_websocket_port.return_value = None
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
url = reverse("console") + f"?token={self.token}&view=full"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertTemplateUsed(response, "console-vnc-lite.html")
self.assertIn("'rdp' is not supported", str(response.context["console_error"]))
@override_settings(SERIAL_CONSOLE_ENABLED=True)
@patch("console.views.wvmInstance")
def test_console_superuser_pty(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "pty"
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
url = reverse("console") + f"?token={self.token}"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertTemplateUsed(response, "console-xterm.html")
@override_settings(SERIAL_CONSOLE_ENABLED=False)
@patch("console.views.wvmInstance")
def test_console_pty_refused_when_serial_console_disabled(self, mock_wvm):
# socketiod is unauthenticated, so the serial console is
# disabled unless an admin explicitly enables it.
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "pty"
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
url = reverse("console") + f"?token={self.token}"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertTemplateNotUsed(response, "console-xterm.html")
self.assertIn("Serial console is disabled", str(response.context["console_error"]))
@patch("console.views.wvmInstance")
def test_console_user_assigned_instance_access(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_conn.get_console_websocket_port.return_value = None
mock_wvm.return_value = mock_conn
self.client.force_login(self.alice)
url = reverse("console") + f"?token={self.token}"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertTemplateUsed(response, "console-vnc-lite.html")
def test_console_unauthorized_user_forbidden(self):
self.client.force_login(self.bob)
url = reverse("console") + f"?token={self.token}"
response = self.client.get(url)
# Non-permitted user attempting to view other user's instance
self.assertEqual(response.status_code, 403)
self.assertIn("permission", response.content.decode("utf-8").lower())
@patch("console.views.wvmInstance")
def test_console_lite_uses_the_signed_token_not_the_url_token(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_conn.get_console_websocket_port.return_value = None
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
response = self.client.get(reverse("console") + f"?token={self.token}")
signed = response.context["token"]
self.assertNotEqual(signed, self.token)
self.assertNotContains(response, "readQueryVariable('token'")
self.assertContains(response, f"const token = '{escapejs(signed)}'")
@patch("console.views.wvmInstance")
def test_console_token_cookie_flags(self, mock_wvm):
# The token cookie is only read by novncd, never by JS.
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_conn.get_console_websocket_port.return_value = None
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
response = self.client.get(reverse("console") + f"?token={self.token}")
cookie = response.cookies["token"]
self.assertTrue(cookie["httponly"])
self.assertEqual(cookie["samesite"], "Lax")
@patch("console.views.wvmInstance")
def test_console_view_instances_perm_does_not_open_console(self, mock_wvm):
# Global view_instances is a read-only role; the console
# is for superusers and VM owners only.
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_conn.get_console_websocket_port.return_value = None
mock_wvm.return_value = mock_conn
perm = Permission.objects.get(codename="view_instances")
self.bob.user_permissions.add(perm)
self.client.force_login(self.bob)
url = reverse("console") + f"?token={self.token}"
response = self.client.get(url)
self.assertEqual(response.status_code, 403)
self.assertTemplateNotUsed(response, "console-vnc-lite.html")
@patch("console.views.wvmInstance")
@override_settings(WS_PUBLIC_PATH="/novncd/")
def test_console_ws_path_normalization_leading_slash(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
url = reverse("console") + f"?token={self.token}"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.context["ws_path"], "novncd/")
@patch("console.views.wvmInstance")
@override_settings(WS_PUBLIC_PATH="novncd/")
def test_console_ws_path_normalization_clean(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
url = reverse("console") + f"?token={self.token}"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.context["ws_path"], "novncd/")
@patch("console.views.wvmInstance")
@override_settings(WS_PUBLIC_PATH="/")
def test_console_ws_path_root_slash(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
url = reverse("console") + f"?token={self.token}"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.context["ws_path"], "")
@override_settings(ALLOWED_HOSTS=["panel.example.com"])
@patch("console.views.wvmInstance")
def test_console_ws_host_strip_port(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
url = reverse("console") + f"?token={self.token}"
response = self.client.get(url, HTTP_HOST="panel.example.com:8080")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.context["ws_host"], "panel.example.com")
@patch("console.views.wvmInstance")
def test_console_xss_protection_query_parameters(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_conn.get_console_websocket_port.return_value = None
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
# Attempt XSS injection via query parameters
url = (
reverse("console")
+ f"?token={self.token}"
+ "&view=lite&view_only=alert('xss')&scale=true&clip_viewport=1"
)
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.context["view_only"], False)
self.assertEqual(response.context["scale"], True)
self.assertEqual(response.context["clip_viewport"], True)
content = response.content.decode("utf-8")
self.assertNotIn("alert('xss')", content)
self.assertIn("rfb.viewOnly = false;", content)
self.assertIn("rfb.scaleViewport = true;", content)
@patch("console.views.wvmInstance")
def test_console_password_escapejs_escaping(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_conn.get_console_websocket_port.return_value = None
mock_conn.get_console_passwd.return_value = "O'Brian<script>&123"
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
# Test Lite mode
url_lite = reverse("console") + f"?token={self.token}&view=lite"
resp_lite = self.client.get(url_lite)
self.assertEqual(resp_lite.status_code, 200)
content_lite = resp_lite.content.decode("utf-8")
self.assertIn(r"O\u0027Brian", content_lite)
self.assertNotIn("&#39;", content_lite)
# Test Full mode
url_full = reverse("console") + f"?token={self.token}&view=full"
resp_full = self.client.get(url_full)
self.assertEqual(resp_full.status_code, 200)
content_full = resp_full.content.decode("utf-8")
# In script tag, password must use JS unicode escape and not HTML entities
main_script_block = [
s.split("</script>")[0]
for s in content_full.split("<script")
if "defaults" in s
][0]
self.assertIn(r"password: 'O\u0027Brian", main_script_block)
self.assertNotIn("&#39;", main_script_block)
self.assertNotIn("&#x27;", main_script_block)
@patch("console.views.wvmInstance")
def test_console_full_resize_syntax_validity(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_conn.get_console_websocket_port.return_value = None
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
# Default: scale=False, resize_session=False -> 'off'
url = reverse("console") + f"?token={self.token}&view=full"
resp = self.client.get(url)
self.assertEqual(resp.status_code, 200)
content = resp.content.decode("utf-8")
self.assertIn("resize: 'off',", content)
self.assertNotIn("&#x27;off&#x27;", content)
self.assertNotIn("&#39;off&#39;", content)
# scale=True, resize_session=False -> 'scale'
url_scale = reverse("console") + f"?token={self.token}&view=full&scale=true"
resp_scale = self.client.get(url_scale)
self.assertEqual(resp_scale.status_code, 200)
content_scale = resp_scale.content.decode("utf-8")
self.assertIn("resize: 'scale',", content_scale)
# resize_session=True -> 'remote'
url_remote = (
reverse("console") + f"?token={self.token}&view=full&resize_session=true"
)
resp_remote = self.client.get(url_remote)
self.assertEqual(resp_remote.status_code, 200)
content_remote = resp_remote.content.decode("utf-8")
self.assertIn("resize: 'remote',", content_remote)
@patch("console.views.wvmInstance")
def test_console_libvirt_error_fallback(self, mock_wvm):
mock_wvm.side_effect = libvirtError("Connection to libvirtd failed")
self.client.force_login(self.admin_user)
url = reverse("console") + f"?token={self.token}"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertTemplateUsed(response, "console-vnc-lite.html")
self.assertIsNotNone(response.context["console_error"])
def test_console_missing_token_safe_fallback(self):
self.client.force_login(self.admin_user)
url = reverse("console")
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertTemplateUsed(response, "console-vnc-lite.html")
self.assertIsNotNone(response.context["console_error"])
@patch("console.views.wvmInstance")
def test_console_send_keys_menu_elements(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "vnc"
mock_wvm.return_value = mock_conn
self.client.force_login(self.admin_user)
url = reverse("console") + f"?token={self.token}&view=lite"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
content = response.content.decode("utf-8")
expected_ids = [
"ctrlaltdel",
"ctrlaltbackspace",
"alttab",
"superkey",
"sendesc",
"sendtab",
"ctrlc",
"ctrlz",
"ctrld",
"ctrlbackslash",
] + [f"ctrlaltf{i}" for i in range(1, 13)]
for key_id in expected_ids:
self.assertIn(
f'id="{key_id}"',
content,
f"Expected key element id '{key_id}' in console menu",
)
class NovncdDaemonLogicTestCase(TestCase):
def setUp(self):
self.get_parser = novncd_mod.get_parser
self.CompatibilityMixIn = novncd_mod.CompatibilityMixIn
def test_novncd_parser_defaults(self):
parser = self.get_parser()
opts = parser.parse_args([])
self.assertFalse(opts.verbose)
self.assertFalse(opts.debug)
self.assertEqual(opts.port, 6080)
def test_novncd_parser_custom_arguments(self):
parser = self.get_parser()
opts = parser.parse_args(
["-v", "-d", "-H", "192.168.1.100", "-p", "7070", "-c", "/path/to/cert.pem"]
)
self.assertTrue(opts.verbose)
self.assertTrue(opts.debug)
self.assertEqual(opts.host, "192.168.1.100")
self.assertEqual(opts.port, 7070)
self.assertEqual(opts.cert, "/path/to/cert.pem")
@patch.object(novncd_mod, "get_connection_infos")
def test_compatibility_mixin_token_from_cookie(self, mock_get_info):
mock_get_info.return_value = (
"host1",
22,
"root",
1,
"192.168.1.50",
5900,
None,
)
handler = self.CompatibilityMixIn()
handler.headers = {"cookie": "other=123; token=1-uuid-from-cookie; test=456"}
handler.path = "/novncd/"
handler.msg = MagicMock()
handler.do_proxy = MagicMock()
handler.verbose = False
socket_factory = MagicMock()
handler._new_client(daemon=False, socket_factory=socket_factory)
mock_get_info.assert_called_once_with("1-uuid-from-cookie")
socket_factory.assert_called_once_with("host1", 5900, connect=True)
@patch.object(novncd_mod, "get_connection_infos")
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",
5901,
None,
)
handler = self.CompatibilityMixIn()
handler.headers = {}
handler.path = "/novncd/?token=1-uuid-from-query"
handler.msg = MagicMock()
handler.do_proxy = MagicMock()
handler.verbose = False
socket_factory = MagicMock()
handler._new_client(daemon=False, socket_factory=socket_factory)
mock_get_info.assert_called_once_with("1-uuid-from-query")
socket_factory.assert_called_once_with("host2", 5901, connect=True)
@patch.object(novncd_mod, "get_connection_infos")
def test_compatibility_mixin_missing_token_safe_exit(self, mock_get_info):
handler = self.CompatibilityMixIn()
handler.headers = {}
handler.path = "/novncd/"
handler.msg = MagicMock()
socket_factory = MagicMock()
# Should cleanly exit and log message without raising UnboundLocalError
handler._new_client(daemon=False, socket_factory=socket_factory)
mock_get_info.assert_not_called()
socket_factory.assert_not_called()
handler.msg.assert_called_with(
"No console token provided in cookie or query parameters"
)
@patch("instances.models.Instance.objects.get")
@patch("vrtManager.instance.wvmInstance")
def test_get_connection_infos_valid_signed_token(self, mock_wvm, mock_inst_get):
User = get_user_model()
user = User.objects.filter(is_superuser=True).first()
mock_instance = MagicMock()
mock_instance.compute.hostname = "127.0.0.1"
mock_instance.compute.login = "root"
mock_instance.compute.password = ""
mock_instance.compute.type = 1
mock_instance.name = "test-vm"
mock_inst_get.return_value = mock_instance
mock_conn = MagicMock()
mock_conn.get_console_listener_addr.return_value = "127.0.0.1"
mock_conn.get_console_port.return_value = 5900
mock_conn.get_console_socket.return_value = None
mock_wvm.return_value = mock_conn
signer = TimestampSigner(salt="console.novnc")
token = signer.sign(f"1:test-uuid:{user.id}")
infos = novncd_mod.get_connection_infos(token)
self.assertEqual(infos[0], "127.0.0.1")
self.assertEqual(infos[5], 5900)
@patch("sys.argv", ["novncd"])
@override_settings(DEBUG=False)
def test_get_connection_infos_invalid_token_in_production(self):
with self.assertRaises(PermissionError):
novncd_mod.get_connection_infos("invalid-raw-token")
def test_get_connection_infos_unauthorized_user(self):
User = get_user_model()
unauthorized_user = User.objects.create_user(
username="unauthorized_console_user",
password="password",
)
compute = Compute.objects.create(
name="novnc-unauth-compute",
hostname="127.0.0.1",
login="root",
password="",
type=1,
)
instance = Instance.objects.create(
compute=compute,
name="novnc-unauth-vm",
uuid="33333333-3333-3333-3333-333333333333",
)
signer = TimestampSigner(salt="console.novnc")
token = signer.sign(f"{compute.id}:{instance.uuid}:{unauthorized_user.id}")
with self.assertRaises(PermissionError):
novncd_mod.get_connection_infos(token)
def test_get_connection_infos_rejects_signed_token_without_user(self):
# The console view always signs host:uuid:user_id; a token without the
# user part would skip the per-user access check entirely.
compute = Compute.objects.create(
name="novnc-2part-compute", hostname="127.0.0.1", login="root", password="", type=1
)
instance = Instance.objects.create(
compute=compute,
name="novnc-2part-vm",
uuid="66666666-6666-6666-6666-666666666666",
)
signer = TimestampSigner(salt="console.novnc")
token = signer.sign(f"{compute.id}:{instance.uuid}")
with self.assertRaises(PermissionError):
novncd_mod.get_connection_infos(token)
def test_get_connection_infos_rejects_unsigned_token_even_under_test_runner(self):
# sys.argv contains "test" while this suite runs; production code must
# not change behaviour based on that.
compute = Compute.objects.create(
name="novnc-unsigned-compute", hostname="127.0.0.1", login="root", password="", type=1
)
instance = Instance.objects.create(
compute=compute,
name="novnc-unsigned-vm",
uuid="88888888-8888-8888-8888-888888888888",
)
with self.assertRaises(PermissionError):
novncd_mod.get_connection_infos(f"{compute.id}-{instance.uuid}")
def test_get_connection_infos_view_instances_user_is_not_enough(self):
User = get_user_model()
viewer = User.objects.create_user(username="novnc_viewer", password="password")
viewer.user_permissions.add(Permission.objects.get(codename="view_instances"))
compute = Compute.objects.create(
name="novnc-viewer-compute", hostname="127.0.0.1", login="root", password="", type=1
)
instance = Instance.objects.create(
compute=compute,
name="novnc-viewer-vm",
uuid="55555555-5555-5555-5555-555555555555",
)
signer = TimestampSigner(salt="console.novnc")
token = signer.sign(f"{compute.id}:{instance.uuid}:{viewer.id}")
with self.assertRaises(PermissionError):
novncd_mod.get_connection_infos(token)
def test_get_connection_infos_inactive_user(self):
User = get_user_model()
inactive_user = User.objects.create_user(
username="inactive_console_user",
password="password",
is_active=False,
)
compute = Compute.objects.create(
name="novnc-inactive-compute",
hostname="127.0.0.1",
login="root",
password="",
type=1,
)
instance = Instance.objects.create(
compute=compute,
name="novnc-inactive-vm",
uuid="44444444-4444-4444-4444-444444444444",
)
UserInstance.objects.create(instance=instance, user=inactive_user)
signer = TimestampSigner(salt="console.novnc")
token = signer.sign(f"{compute.id}:{instance.uuid}:{inactive_user.id}")
with self.assertRaises(PermissionError):
novncd_mod.get_connection_infos(token)
@patch("sys.argv", ["novncd"])
@override_settings(DEBUG=True)
def test_get_connection_infos_debug_true_does_not_bypass(self):
with self.assertRaises(PermissionError):
novncd_mod.get_connection_infos("invalid-token-in-debug")
@patch("sys.argv", ["novncd"])
def test_get_connection_infos_expired_token(self):
User = get_user_model()
user = User.objects.filter(is_superuser=True).first()
compute = Compute.objects.create(
name="novnc-expired-compute",
hostname="127.0.0.1",
login="root",
password="",
type=1,
)
instance = Instance.objects.create(
compute=compute,
name="novnc-expired-vm",
uuid="55555555-5555-5555-5555-555555555555",
)
import time
from unittest.mock import patch
# Sign token 300 seconds in the past (> 120s TTL)
with patch("time.time", return_value=time.time() - 300):
signer = TimestampSigner(salt="console.novnc")
expired_token = signer.sign(f"{compute.id}:{instance.uuid}:{user.id}")
with self.assertRaises(PermissionError):
novncd_mod.get_connection_infos(expired_token)