1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 19:26:48 +00:00

feat(console): add comprehensive test suite and expand send keys menu

- Add 23 unit and integration tests covering console views, authorization, URL/path normalization, and novncd daemon token extraction
- Refactor console/novncd arg parsing into __main__ block for clean test imports
- Expand Send Key(s) menu into categorized sections (System, Navigation/Desktop, Terminal Control, Virtual Consoles)
- Implement non-destructive LIFO sendKeyCombo using public rfb.sendKey API, eliminating vendor debt
- Wire Send Key actions across Lite, Full, and PTY (xterm) console templates
- Add modern navigator.keyboard.lock support during fullscreen sessions
This commit is contained in:
catborise 2026-09-26 12:10:35 +03:00
parent a444a2ba40
commit 591429b0c9
7 changed files with 679 additions and 80 deletions

View file

@ -65,16 +65,6 @@ def get_parser(default_host=WS_HOST, default_port=WS_PORT or 6080, default_cert=
return parser
options = get_parser().parse_args()
FORMAT = "%(asctime)s - %(name)s - %(levelname)s : %(message)s"
if options.debug:
logging.basicConfig(level=logging.DEBUG, format=FORMAT)
options.verbose = True
elif options.verbose:
logging.basicConfig(level=logging.INFO, format=FORMAT)
else:
logging.basicConfig(level=logging.WARNING, format=FORMAT)
try:
from websockify import WebSocketProxy
@ -302,6 +292,17 @@ else:
if __name__ == "__main__":
options = get_parser().parse_args()
FORMAT = "%(asctime)s - %(name)s - %(levelname)s : %(message)s"
if options.debug:
logging.basicConfig(level=logging.DEBUG, format=FORMAT)
options.verbose = True
elif options.verbose:
logging.basicConfig(level=logging.INFO, format=FORMAT)
else:
logging.basicConfig(level=logging.WARNING, format=FORMAT)
if USE_HANDLER:
# Create the WebSocketProxy with NovaProxyRequestHandler handler
server = WebSocketProxy(

View file

@ -35,7 +35,7 @@
<body>
<div id='main_container' class="container">
<nav class="navbar sticky-top navbar-expand-md navbar-dark bg-primary" arial-label="console navbar">
<nav class="navbar sticky-top navbar-expand-md navbar-dark bg-primary" aria-label="console navbar">
<div class="container">
<a class="navbar-brand">{{ instance.name }}</a>
<button class="navbar-toggler" type="button" data-bs-toggle="collapse" data-bs-target=".navbar-collapse"
@ -44,31 +44,50 @@
</button>
<div class="collapse navbar-collapse">
<ul class="navbar-nav me-auto my-2 my-lg-0 navbar-nav-scroll" style="--bs-scroll-height: 100px;">
<ul class="navbar-nav me-auto my-2 my-lg-0">
<li class="nav-item dropdown">
<a class="nav-link dropdown-toggle" href="#" data-bs-toggle="dropdown" role="button"
id="dropdownMenuLink" aria-haspopup="true" aria-expanded="false">
{% trans "Send key(s)" %}
</a>
<div class="dropdown-menu" aria-labelledby="dropdownMenuLink">
<a class="dropdown-item" id="ctrlaltdel" href='#'>Ctrl+Alt+Del</a>
<div class="dropdown-menu shadow" aria-labelledby="dropdownMenuLink" style="min-width: 250px;">
<h6 class="dropdown-header">{% trans "System" %}</h6>
<a class="dropdown-item" id="ctrlaltdel" href="#">Ctrl+Alt+Del</a>
<a class="dropdown-item" id="ctrlaltbackspace" href="#">Ctrl+Alt+Backspace</a>
<div class="dropdown-divider"></div>
<a class="dropdown-item" id="ctrlaltf1" href='#'>Ctrl+Alt+F1</a>
<a class="dropdown-item" id="ctrlaltf2" href='#'>Ctrl+Alt+F2</a>
<a class="dropdown-item" id="ctrlaltf3" href='#'>Ctrl+Alt+F3</a>
<a class="dropdown-item" id="ctrlaltf4" href='#'>Ctrl+Alt+F4</a>
<a class="dropdown-item" id="ctrlaltf5" href='#'>Ctrl+Alt+F5</a>
<a class="dropdown-item" id="ctrlaltf6" href='#'>Ctrl+Alt+F6</a>
<a class="dropdown-item" id="ctrlaltf7" href='#'>Ctrl+Alt+F7</a>
<a class="dropdown-item" id="ctrlaltf8" href='#'>Ctrl+Alt+F8</a>
<a class="dropdown-item" id="ctrlaltf9" href='#'>Ctrl+Alt+F9</a>
<a class="dropdown-item" id="ctrlaltf10" href='#'>Ctrl+Alt+F10</a>
<a class="dropdown-item" id="ctrlaltf11" href='#'>Ctrl+Alt+F11</a>
<a class="dropdown-item" id="ctrlaltf12" href='#'>Ctrl+Alt+F12</a>
<h6 class="dropdown-header">{% trans "Navigation & Desktop" %}</h6>
<a class="dropdown-item" id="alttab" href="#">Alt+Tab</a>
<a class="dropdown-item" id="superkey" href="#">Windows / Super Key</a>
<a class="dropdown-item" id="sendesc" href="#">Escape</a>
<a class="dropdown-item" id="sendtab" href="#">Tab</a>
<div class="dropdown-divider"></div>
<h6 class="dropdown-header">{% trans "Terminal Control" %}</h6>
<a class="dropdown-item" id="ctrlc" href="#">Ctrl+C</a>
<a class="dropdown-item" id="ctrlz" href="#">Ctrl+Z</a>
<a class="dropdown-item" id="ctrld" href="#">Ctrl+D</a>
<a class="dropdown-item" id="ctrlbackslash" href="#">Ctrl+\</a>
<div class="dropdown-divider"></div>
<h6 class="dropdown-header">{% trans "Virtual Consoles (Ctrl+Alt+F#)" %}</h6>
<div class="px-3 py-1">
<div class="row g-1 text-center">
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf1" href="#">F1</a></div>
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf2" href="#">F2</a></div>
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf3" href="#">F3</a></div>
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf4" href="#">F4</a></div>
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf5" href="#">F5</a></div>
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf6" href="#">F6</a></div>
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf7" href="#">F7</a></div>
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf8" href="#">F8</a></div>
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf9" href="#">F9</a></div>
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf10" href="#">F10</a></div>
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf11" href="#">F11</a></div>
<div class="col-3"><a class="btn btn-sm btn-outline-secondary w-100 py-1" id="ctrlaltf12" href="#">F12</a></div>
</div>
</div>
</div>
</li>
<li class="nav-item" id="fullscreen_button">
<a class="nav-link" href='#'>{% trans "Fullscreen" %}</a>
<a class="nav-link" href="#">{% trans "Fullscreen" %}</a>
</li>
{% block navbarmenu %}{% endblock %}
</ul>

View file

@ -69,6 +69,116 @@
src='{% static "js/novnc/vendor/browser-es-module-loader/dist/browser-es-module-loader.js" %}'></script>
<!-- actual script modules -->
<script type="module" crossorigin="anonymous" src="{% static "js/novnc/app/ui.js" %}"></script>
<script type="module" crossorigin="anonymous">
import UI from '{% static "js/novnc/app/ui.js" %}';
import KeyTable from '{% static "js/novnc/core/input/keysym.js" %}';
const fnKeys = [
{ keysym: KeyTable.XK_F1, code: 'F1' },
{ keysym: KeyTable.XK_F2, code: 'F2' },
{ keysym: KeyTable.XK_F3, code: 'F3' },
{ keysym: KeyTable.XK_F4, code: 'F4' },
{ keysym: KeyTable.XK_F5, code: 'F5' },
{ keysym: KeyTable.XK_F6, code: 'F6' },
{ keysym: KeyTable.XK_F7, code: 'F7' },
{ keysym: KeyTable.XK_F8, code: 'F8' },
{ keysym: KeyTable.XK_F9, code: 'F9' },
{ keysym: KeyTable.XK_F10, code: 'F10' },
{ keysym: KeyTable.XK_F11, code: 'F11' },
{ keysym: KeyTable.XK_F12, code: 'F12' }
];
function sendSingleKey(keysym, code) {
if (!UI.rfb) { return false; }
UI.rfb.sendKey(keysym, code, true);
UI.rfb.sendKey(keysym, code, false);
return false;
}
function sendKeyCombo(keys) {
if (!UI.rfb) { return false; }
for (let i = 0; i < keys.length; i++) {
UI.rfb.sendKey(keys[i].keysym, keys[i].code, true);
}
for (let i = keys.length - 1; i >= 0; i--) {
UI.rfb.sendKey(keys[i].keysym, keys[i].code, false);
}
return false;
}
function sendCtrlAltDel() {
if (UI.rfb && typeof UI.rfb.sendCtrlAltDel === 'function') {
UI.rfb.sendCtrlAltDel();
} else {
sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_Alt_L, code: 'AltLeft' },
{ keysym: KeyTable.XK_Delete, code: 'Delete' }
]);
}
return false;
}
function sendCtrlAltFN(f) {
if (f < 0 || f >= fnKeys.length) { return false; }
return sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_Alt_L, code: 'AltLeft' },
fnKeys[f]
]);
}
function bindKeyAction(id, handler) {
const el = document.getElementById(id);
if (el) {
el.addEventListener('click', (e) => {
e.preventDefault();
handler();
});
}
}
bindKeyAction('fullscreen_button', () => {
if (typeof UI.fullscreen === 'function') {
UI.fullscreen();
}
});
bindKeyAction('ctrlaltdel', sendCtrlAltDel);
bindKeyAction('ctrlaltbackspace', () => sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_Alt_L, code: 'AltLeft' },
{ keysym: KeyTable.XK_BackSpace, code: 'Backspace' }
]));
bindKeyAction('alttab', () => sendKeyCombo([
{ keysym: KeyTable.XK_Alt_L, code: 'AltLeft' },
{ keysym: KeyTable.XK_Tab, code: 'Tab' }
]));
bindKeyAction('superkey', () => sendSingleKey(KeyTable.XK_Super_L, 'MetaLeft'));
bindKeyAction('sendesc', () => sendSingleKey(KeyTable.XK_Escape, 'Escape'));
bindKeyAction('sendtab', () => sendSingleKey(KeyTable.XK_Tab, 'Tab'));
bindKeyAction('ctrlc', () => sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_c, code: 'KeyC' }
]));
bindKeyAction('ctrlz', () => sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_z, code: 'KeyZ' }
]));
bindKeyAction('ctrld', () => sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_d, code: 'KeyD' }
]));
bindKeyAction('ctrlbackslash', () => sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_backslash, code: 'Backslash' }
]));
for (let i = 0; i < 12; i++) {
bindKeyAction('ctrlaltf' + (i + 1), () => sendCtrlAltFN(i));
}
</script>
<!-- end scripts -->
{% endblock %}

View file

@ -34,6 +34,7 @@
<script type="module" crossorigin="anonymous">
// RFB holds the API to connect and communicate with a VNC server
import RFB from '{% static "js/novnc/core/rfb.js" %}';
import KeyTable from '{% static "js/novnc/core/input/keysym.js" %}';
let rfb;
let desktopName;
@ -66,14 +67,64 @@
desktopName = e.detail.name;
}
function sendSingleKey(keysym, code) {
if (!rfb) { return false; }
rfb.sendKey(keysym, code, true);
rfb.sendKey(keysym, code, false);
return false;
}
function sendKeyCombo(keys) {
if (!rfb) { return false; }
for (let i = 0; i < keys.length; i++) {
rfb.sendKey(keys[i].keysym, keys[i].code, true);
}
for (let i = keys.length - 1; i >= 0; i--) {
rfb.sendKey(keys[i].keysym, keys[i].code, false);
}
return false;
}
// Since most operating systems will catch Ctrl+Alt+Del
// before they get a chance to be intercepted by the browser,
// we provide a way to emulate this key sequence.
function sendCtrlAltDel() {
rfb.sendCtrlAltDel();
if (rfb && typeof rfb.sendCtrlAltDel === 'function') {
rfb.sendCtrlAltDel();
} else {
sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_Alt_L, code: 'AltLeft' },
{ keysym: KeyTable.XK_Delete, code: 'Delete' }
]);
}
return false;
}
const fnKeys = [
{ keysym: KeyTable.XK_F1, code: 'F1' },
{ keysym: KeyTable.XK_F2, code: 'F2' },
{ keysym: KeyTable.XK_F3, code: 'F3' },
{ keysym: KeyTable.XK_F4, code: 'F4' },
{ keysym: KeyTable.XK_F5, code: 'F5' },
{ keysym: KeyTable.XK_F6, code: 'F6' },
{ keysym: KeyTable.XK_F7, code: 'F7' },
{ keysym: KeyTable.XK_F8, code: 'F8' },
{ keysym: KeyTable.XK_F9, code: 'F9' },
{ keysym: KeyTable.XK_F10, code: 'F10' },
{ keysym: KeyTable.XK_F11, code: 'F11' },
{ keysym: KeyTable.XK_F12, code: 'F12' }
];
function sendCtrlAltFN(f) {
if (f < 0 || f >= fnKeys.length) { return false; }
return sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_Alt_L, code: 'AltLeft' },
fnKeys[f]
]);
}
function fullscreen() {
if (document.fullscreenElement || // alternative standard method
document.mozFullScreenElement || // currently working methods
@ -89,32 +140,40 @@
document.msExitFullscreen();
}
} else {
if (document.documentElement.requestFullscreen) {
document.documentElement.requestFullscreen();
} else if (document.documentElement.mozRequestFullScreen) {
document.documentElement.mozRequestFullScreen();
} else if (document.documentElement.webkitRequestFullscreen) {
document.documentElement.webkitRequestFullscreen(Element.ALLOW_KEYBOARD_INPUT);
} else if (document.body.msRequestFullscreen) {
document.body.msRequestFullscreen();
const el = document.documentElement;
const request = el.requestFullscreen ||
el.mozRequestFullScreen ||
el.webkitRequestFullscreen ||
el.msRequestFullscreen;
if (request) {
request.call(el).catch(() => {});
}
}
return false;
}
function sendCtrlAltFN(f) {
rfb.sendCtrlAltFN(f);
return false;
}
document.addEventListener('fullscreenchange', () => {
if (document.fullscreenElement) {
if ('keyboard' in navigator && typeof navigator.keyboard.lock === 'function') {
navigator.keyboard.lock(['AltLeft', 'AltRight', 'Tab', 'Escape']).catch(() => {});
}
} else {
if ('keyboard' in navigator && typeof navigator.keyboard.unlock === 'function') {
navigator.keyboard.unlock();
}
}
});
function machineShutdown() {
rfb.machineShutdown();
if (rfb) { rfb.machineShutdown(); }
return false;
}
function machineReboot() {
rfb.machineReboot();
if (rfb) { rfb.machineReboot(); }
return false;
}
function machineReset() {
rfb.machineReset();
if (rfb) { rfb.machineReset(); }
return false;
}
@ -123,6 +182,16 @@
document.getElementById('noVNC_status').textContent = text;
}
function bindKeyAction(id, handler) {
const el = document.getElementById(id);
if (el) {
el.addEventListener('click', (e) => {
e.preventDefault();
handler();
});
}
}
// This function extracts the value of one variable from the
// query string. If the variable isn't defined in the URL
// it returns the default value instead.
@ -143,25 +212,46 @@
return defaultValue;
}
document.getElementById('sendCtrlAltDelButton').onclick = sendCtrlAltDel;
document.getElementById('machineShutdownButton').onclick = machineShutdown;
document.getElementById('machineRebootButton').onclick = machineReboot;
document.getElementById('machineResetButton').onclick = machineReset;
document.getElementById('fullscreen_button').onclick = fullscreen;
bindKeyAction('sendCtrlAltDelButton', sendCtrlAltDel);
bindKeyAction('machineShutdownButton', machineShutdown);
bindKeyAction('machineRebootButton', machineReboot);
bindKeyAction('machineResetButton', machineReset);
bindKeyAction('fullscreen_button', fullscreen);
document.getElementById('ctrlaltdel').addEventListener('click', sendCtrlAltDel);
document.getElementById('ctrlaltf1').addEventListener('click', function () { sendCtrlAltFN(0); });
document.getElementById('ctrlaltf2').addEventListener('click', function () { sendCtrlAltFN(1); });
document.getElementById('ctrlaltf3').addEventListener('click', function () { sendCtrlAltFN(2); });
document.getElementById('ctrlaltf4').addEventListener('click', function () { sendCtrlAltFN(3); });
document.getElementById('ctrlaltf5').addEventListener('click', function () { sendCtrlAltFN(4); });
document.getElementById('ctrlaltf6').addEventListener('click', function () { sendCtrlAltFN(5); });
document.getElementById('ctrlaltf7').addEventListener('click', function () { sendCtrlAltFN(6); });
document.getElementById('ctrlaltf8').addEventListener('click', function () { sendCtrlAltFN(7); });
document.getElementById('ctrlaltf9').addEventListener('click', function () { sendCtrlAltFN(8); });
document.getElementById('ctrlaltf10').addEventListener('click', function () { sendCtrlAltFN(9); });
document.getElementById('ctrlaltf11').addEventListener('click', function () { sendCtrlAltFN(10); });
document.getElementById('ctrlaltf12').addEventListener('click', function () { sendCtrlAltFN(11); });
bindKeyAction('ctrlaltdel', sendCtrlAltDel);
bindKeyAction('ctrlaltbackspace', () => sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_Alt_L, code: 'AltLeft' },
{ keysym: KeyTable.XK_BackSpace, code: 'Backspace' }
]));
bindKeyAction('alttab', () => sendKeyCombo([
{ keysym: KeyTable.XK_Alt_L, code: 'AltLeft' },
{ keysym: KeyTable.XK_Tab, code: 'Tab' }
]));
bindKeyAction('superkey', () => sendSingleKey(KeyTable.XK_Super_L, 'MetaLeft'));
bindKeyAction('sendesc', () => sendSingleKey(KeyTable.XK_Escape, 'Escape'));
bindKeyAction('sendtab', () => sendSingleKey(KeyTable.XK_Tab, 'Tab'));
bindKeyAction('ctrlc', () => sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_c, code: 'KeyC' }
]));
bindKeyAction('ctrlz', () => sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_z, code: 'KeyZ' }
]));
bindKeyAction('ctrld', () => sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_d, code: 'KeyD' }
]));
bindKeyAction('ctrlbackslash', () => sendKeyCombo([
{ keysym: KeyTable.XK_Control_L, code: 'ControlLeft' },
{ keysym: KeyTable.XK_backslash, code: 'Backslash' }
]));
for (let i = 0; i < 12; i++) {
bindKeyAction('ctrlaltf' + (i + 1), () => sendCtrlAltFN(i));
}
// Read parameters specified in the URL query string
// By default, use the host and port of server that served this file

View file

@ -105,7 +105,35 @@
return defaultValue;
}
window.onresize = resize
window.onload = resize
function bindTerminalKey(id, str) {
const el = document.getElementById(id);
if (el) {
el.addEventListener('click', (e) => {
e.preventDefault();
socket.emit("pty_input", {"input": str});
});
}
}
bindTerminalKey('ctrlc', '\x03');
bindTerminalKey('ctrld', '\x04');
bindTerminalKey('ctrlz', '\x1a');
bindTerminalKey('ctrlbackslash', '\x1c');
bindTerminalKey('sendtab', '\t');
bindTerminalKey('sendesc', '\x1b');
const fsBtn = document.getElementById('fullscreen_button');
if (fsBtn) {
fsBtn.addEventListener('click', (e) => {
e.preventDefault();
if (document.fullscreenElement) {
document.exitFullscreen().catch(() => {});
} else {
document.documentElement.requestFullscreen().catch(() => {});
}
});
}
window.onresize = resize;
window.onload = resize;
</script>
{% endblock %}

View file

@ -1,3 +1,353 @@
from django.test import TestCase
import importlib.machinery
from pathlib import Path
from unittest.mock import MagicMock, patch
# Create your tests here.
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.test import TestCase, override_settings
from django.urls import reverse
from instances.models import Instance
from libvirt import libvirtError
# Dynamically load console/novncd script for testing
NOVNCD_PATH = Path(__file__).resolve().parent / "novncd"
novncd_mod = importlib.machinery.SourceFileLoader("novncd_mod", str(NOVNCD_PATH)).load_module()
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")
self.assertEqual(response.cookies["token"].value, self.token)
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_superuser_spice_lite(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "spice"
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=lite"
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertTemplateUsed(response, "console-spice-lite.html")
@patch("console.views.wvmInstance")
def test_console_superuser_spice_full(self, mock_wvm):
mock_conn = MagicMock()
mock_conn.get_console_type.return_value = "spice"
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-spice-full.html")
@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")
@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, 500)
self.assertIn("permission", response.content.decode("utf-8").lower())
@patch("console.views.wvmInstance")
def test_console_user_with_view_instances_perm(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
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, 200)
@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_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")

View file

@ -7,16 +7,9 @@ from django.http.response import HttpResponseServerError
from django.shortcuts import render
from django.utils.translation import gettext_lazy as _
from instances.models import Instance
from django.conf import settings
from libvirt import libvirtError
from vrtManager.instance import wvmInstance
from webvirtcloud.settings import (
SOCKETIO_PUBLIC_HOST,
SOCKETIO_PUBLIC_PATH,
SOCKETIO_PUBLIC_PORT,
WS_PUBLIC_HOST,
WS_PUBLIC_PATH,
WS_PUBLIC_PORT,
)
def console(request):
@ -72,14 +65,18 @@ def console(request):
console_type = conn.get_console_type()
console_websocket_port = conn.get_console_websocket_port()
console_passwd = conn.get_console_passwd()
except libvirtError:
except (libvirtError, ValueError, IndexError, Instance.DoesNotExist):
console_type = None
console_websocket_port = None
console_passwd = None
ws_port = console_websocket_port if console_websocket_port else WS_PUBLIC_PORT
ws_host = WS_PUBLIC_HOST if WS_PUBLIC_HOST else request.get_host()
ws_path = WS_PUBLIC_PATH if WS_PUBLIC_PATH else "/"
ws_public_port = getattr(settings, "WS_PUBLIC_PORT", 6080)
ws_public_host = getattr(settings, "WS_PUBLIC_HOST", None)
ws_public_path = getattr(settings, "WS_PUBLIC_PATH", "/")
ws_port = console_websocket_port if console_websocket_port else ws_public_port
ws_host = ws_public_host if ws_public_host else request.get_host()
ws_path = ws_public_path if ws_public_path else "/"
if ":" in ws_host:
ws_host = re.sub(":[0-9]+", "", ws_host)
@ -91,11 +88,15 @@ def console(request):
console_page = "console-" + console_type + "-" + view_type + ".html"
response = render(request, console_page, locals())
elif console_type == "pty":
socketio_public_host = getattr(settings, "SOCKETIO_PUBLIC_HOST", None)
socketio_public_port = getattr(settings, "SOCKETIO_PUBLIC_PORT", 6081)
socketio_public_path = getattr(settings, "SOCKETIO_PUBLIC_PATH", "socket.io/")
socketio_host = (
SOCKETIO_PUBLIC_HOST if SOCKETIO_PUBLIC_HOST else request.get_host()
socketio_public_host if socketio_public_host else request.get_host()
)
socketio_port = SOCKETIO_PUBLIC_PORT if SOCKETIO_PUBLIC_PORT else 6081
socketio_path = SOCKETIO_PUBLIC_PATH if SOCKETIO_PUBLIC_PATH else "/"
socketio_port = socketio_public_port if socketio_public_port else 6081
socketio_path = socketio_public_path if socketio_public_path else "/"
if ":" in socketio_host:
socketio_host = re.sub(":[0-9]+", "", socketio_host)