mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
Comments, docstrings and live test names cited item IDs of a roadmap that is not part of the repository. They now say what is meant, or nothing where the sentence stands without them. Test assertions and behavior are unchanged.
687 lines
26 KiB
Python
687 lines
26 KiB
Python
# 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 / "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"], "")
|
|
|
|
@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("'", 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("'", main_script_block)
|
|
self.assertNotIn("'", 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("'off'", content)
|
|
self.assertNotIn("'off'", 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)
|