# Copyright 2016-2017 OpenStack Foundation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import struct
import urllib.parse as urlparse
import urllib3
from tempest.api.compute import base
from tempest.common import compute
from tempest import config
from tempest.lib import decorators
CONF = config.CONF
[docs]
class NoVNCConsoleTestJSON(base.BaseV2ComputeTest):
"""Test novnc console"""
create_default_network = True
@classmethod
def skip_checks(cls):
super(NoVNCConsoleTestJSON, cls).skip_checks()
if not CONF.compute_feature_enabled.vnc_console:
raise cls.skipException('VNC Console feature is disabled.')
def setUp(self):
super(NoVNCConsoleTestJSON, self).setUp()
self._websocket = None
def tearDown(self):
super(NoVNCConsoleTestJSON, self).tearDown()
if self._websocket is not None:
self._websocket.close()
# NOTE(zhufl): Because server_check_teardown will raise Exception
# which will prevent other cleanup steps from being executed, so
# server_check_teardown should be called after super's tearDown.
self.server_check_teardown()
@classmethod
def setup_clients(cls):
super(NoVNCConsoleTestJSON, cls).setup_clients()
cls.client = cls.servers_client
@classmethod
def resource_setup(cls):
super(NoVNCConsoleTestJSON, cls).resource_setup()
cls.server = cls.create_test_server(wait_until="ACTIVE")
cls.use_get_remote_console = False
if not cls.is_requested_microversion_compatible('2.5'):
cls.use_get_remote_console = True
def _validate_novnc_html(self, vnc_url):
"""Verify we can connect to novnc and get back the javascript."""
resp = urllib3.PoolManager().request('GET', vnc_url)
# Make sure that the GET request was accepted by the novncproxy
self.assertEqual(resp.status, 200, 'Got a Bad HTTP Response on the '
'initial call: ' + str(resp.status))
# Do some basic validation to make sure it is an expected HTML document
resp_data = resp.data.decode()
# This is needed in the case of example: <html lang="en">
self.assertRegex(resp_data, '<html.*>',
'Not a valid html document in the response.')
self.assertIn('</html>', resp_data,
'Not a valid html document in the response.')
# Just try to make sure we got JavaScript back for noVNC, since we
# won't actually use it since not inside of a browser
self.assertIn('noVNC', resp_data,
'Not a valid noVNC javascript html document.')
self.assertIn('<script', resp_data,
'Not a valid noVNC javascript html document.')
def _validate_rfb_negotiation(self):
"""Verify we can connect to novnc and do the websocket connection."""
# Turn the Socket into a WebSocket to do the communication
data = self._websocket.receive_frame()
self.assertFalse(data is None or not data,
'Token must be invalid because the connection '
'closed.')
# Parse the RFB version from the data to make sure it is valid
# and belong to the known supported RFB versions.
version = float("%d.%d" % (int(data[4:7], base=10),
int(data[8:11], base=10)))
# Add the max RFB versions supported
supported_versions = [3.3, 3.8]
self.assertIn(version, supported_versions,
'Bad RFB Version: ' + str(version))
# Send our RFB version to the server
self._websocket.send_frame(data)
# Get the sever authentication type and make sure None is supported
data = self._websocket.receive_frame()
self.assertIsNotNone(data, 'Expected authentication type None.')
data_length = len(data)
if version == 3.3:
# For RFB 3.3: in the security handshake, rather than a two-way
# negotiation, the server decides the security type and sends a
# single word(4 bytes).
self.assertEqual(
data_length, 4, 'Expected authentication type None.')
self.assertIn(1, [int(data[i]) for i in (0, 3)],
'Expected authentication type None.')
else:
self.assertGreaterEqual(
len(data), 2, 'Expected authentication type None.')
self.assertIn(
1,
[int(data[i + 1]) for i in range(int(data[0]))],
'Expected authentication type None.')
# Send to the server that we only support authentication
# type None
self._websocket.send_frame(bytes((1,)))
# The server should send 4 bytes of 0's if security
# handshake succeeded
data = self._websocket.receive_frame()
self.assertEqual(
len(data), 4,
'Server did not think security was successful.')
self.assertEqual(
[int(i) for i in data], [0, 0, 0, 0],
'Server did not think security was successful.')
# Say to leave the desktop as shared as part of client initialization
self._websocket.send_frame(bytes((1,)))
# Get the server initialization packet back and make sure it is the
# right structure where bytes 20-24 is the name length and
# 24-N is the name
data = self._websocket.receive_frame()
data_length = len(data) if data is not None else 0
self.assertFalse(data_length <= 24 or
data_length != (struct.unpack(">L",
data[20:24])[0] + 24),
'Server initialization was not the right format.')
# Since the rest of the data on the screen is arbitrary, we will
# close the socket and end our validation of the data at this point
# Assert that the latest check was false, meaning that the server
# initialization was the right format
self.assertFalse(data_length <= 24 or
data_length != (struct.unpack(">L",
data[20:24])[0] + 24))
def _validate_websocket_upgrade(self):
"""Verify that the websocket upgrade was successful.
Parses response and ensures that required response
fields are present and accurate.
(https://tools.ietf.org/html/rfc7231#section-6.2.2)
"""
self.assertTrue(
self._websocket.response.startswith(b'HTTP/1.1 101 Switching '
b'Protocols'),
'Incorrect HTTP return status code: {}'.format(
str(self._websocket.response)
)
)
_required_header = 'upgrade: websocket'
_response = str(self._websocket.response).lower()
self.assertIn(
_required_header,
_response,
'Did not get the expected WebSocket HTTP Response.'
)
[docs]
@decorators.idempotent_id('c640fdff-8ab4-45a4-a5d8-7e6146cbd0dc')
def test_novnc(self):
"""Test accessing novnc console of server"""
if self.use_get_remote_console:
body = self.client.get_remote_console(
self.server['id'], console_type='novnc',
protocol='vnc')['remote_console']
else:
body = self.client.get_vnc_console(self.server['id'],
type='novnc')['console']
self.assertEqual('novnc', body['type'])
# Do the initial HTTP Request to novncproxy to get the NoVNC JavaScript
self._validate_novnc_html(body['url'])
# Do the WebSockify HTTP Request to novncproxy to do the RFB connection
self._websocket = compute.create_websocket(body['url'])
# Validate that we successfully connected and upgraded to Web Sockets
self._validate_websocket_upgrade()
# Validate the RFB Negotiation to determine if a valid VNC session
self._validate_rfb_negotiation()
[docs]
@decorators.idempotent_id('f9c79937-addc-4aaa-9e0e-841eef02aeb7')
def test_novnc_bad_token(self):
"""Test accessing novnc console with bad token
Do the WebSockify HTTP Request to novnc proxy with a bad token,
the novnc proxy should reject the connection and closed it.
"""
if self.use_get_remote_console:
body = self.client.get_remote_console(
self.server['id'], console_type='novnc',
protocol='vnc')['remote_console']
else:
body = self.client.get_vnc_console(self.server['id'],
type='novnc')['console']
self.assertEqual('novnc', body['type'])
# Do the WebSockify HTTP Request to novncproxy with a bad token
parts = urlparse.urlparse(body['url'])
qparams = urlparse.parse_qs(parts.query)
if 'path' in qparams:
qparams['path'] = urlparse.unquote(qparams['path'][0]).replace(
'token=', 'token=bad')
elif 'token' in qparams:
qparams['token'] = 'bad' + qparams['token'][0]
new_query = urlparse.urlencode(qparams)
new_parts = urlparse.ParseResult(parts.scheme, parts.netloc,
parts.path, parts.params, new_query,
parts.fragment)
url = urlparse.urlunparse(new_parts)
self._websocket = compute.create_websocket(url)
# Make sure the novncproxy rejected the connection and closed it
data = self._websocket.receive_frame()
self.assertTrue(data is None or not data,
"The novnc proxy actually sent us some data, but we "
"expected it to close the connection.")
