openstack/deb-python-proliantutils
Code Issues Proposed changes

Initial Support to RIS

Browse Source 
Added Support to SecureBoot, Firmware Settings API through RIS.
The API's are fully tested to work only on Gen9 Systems.

Note: This is a breaking change from the prev version of
proliantutils. Driver code must be modified for this
change to work.

Change-Id: Iea8a68330a5042407ab719667cd725946ffcf821
Implements: blueprint ris-support
This commit is contained in:
Anusha Ramineni
2015-01-23 20:03:40 +05:30
parent c7a32b7b99
commit 47608b6eef
6 changed files with 768 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
proliantutils/ilo/client.py Normal file
View File

@@ -0,0 +1,35 @@
# Copyright 2014 Hewlett-Packard Development Company, L.P.
#
# 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.
from proliantutils.ilo import ribcl
# Commenting to satisfy pep8
# from proliantutils.ilo import ris
class IloClient(object):
def __new__(self, host, login, password, timeout=60, port=443,
bios_password=None):
# Object is created based on the server model
client = ribcl.RIBCLOperations(host, login, password, timeout, port)
# Till the full RIS Integration is done, disabling the automatic switch
# between RIS and RIBCL CLient. Defaulting it to RIBCL for now.
# TODO(Anusha): Uncomment when full RIS library is available.
# model = client.get_product_name()
#
# if 'Gen9' in model:
# client = ris.RISOperations(host, login, password, bios_password)
return client

80
proliantutils/ilo/exception.py Normal file
View File

@@ -0,0 +1,80 @@
# Copyright 2014 Hewlett-Packard Development Company, L.P.
#
# 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.
"""Exception Class for iLO"""
class IloError(Exception):
"""Base Exception.
This exception is used when a problem is encountered in
executing an operation on the iLO.
"""
def __init__(self, message, errorcode=None):
super(IloError, self).__init__(message)
class IloClientInternalError(IloError):
"""Internal Error from IloClient.
This exception is raised when iLO client library fails to
communicate properly with the iLO.
"""
def __init__(self, message, errorcode=None):
super(IloError, self).__init__(message)
class IloCommandNotSupportedError(IloError):
"""Command not supported on the platform.
This exception is raised when iLO client library fails to
communicate properly with the iLO
"""
def __init__(self, message, errorcode=None):
super(IloError, self).__init__(message)
class IloLoginFailError(IloError):
"""iLO Login Failed.
This exception is used to communicate a login failure to
the caller.
"""
messages = ['User login name was not found',
'Login failed', 'Login credentials rejected']
statuses = [0x005f, 0x000a]
message = 'Authorization Failed'
def __init__(self, message, errorcode=None):
super(IloError, self).__init__(message)
class IloConnectionError(IloError):
"""Cannot connect to iLO.
This exception is used to communicate an HTTP connection
error from the iLO to the caller.
"""
def __init__(self, message):
super(IloConnectionError, self).__init__(message)
class IloInvalidInputError(IloError):
"""Invalid Input passed.
This exception is used when invalid inputs are passed to
the APIs exposed by this module.
"""
def __init__(self, message):
super(IloInvalidInputError, self).__init__(message)

135
proliantutils/ilo/operations.py Normal file
View File

@@ -0,0 +1,135 @@
# Copyright 2014 Hewlett-Packard Development Company, L.P.
#
# 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.
from proliantutils.ilo import exception
ERRMSG = "The specified operation is not supported on current platform."
class IloOperations:
"""iLO class for performing iLO Operations.
This class provides an OO interface for retrieving information
and managing iLO. It implements the same interface in
python as described in HP iLO 4 Scripting and Command Line Guide.
"""
def get_all_licenses(self):
"""Retrieve license type, key, installation date, etc."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def get_product_name(self):
"""Get the model name of the queried server."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def get_host_power_status(self):
"""Request the power state of the server."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def get_one_time_boot(self):
"""Retrieves the current setting for the one time boot."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def get_vm_status(self, device='FLOPPY'):
"""Returns the virtual media drive status like url, is connected, etc.
"""
raise exception.IloCommandNotSupportedError(ERRMSG)
def reset_server(self):
"""Resets the server."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def press_pwr_btn(self):
"""Simulates a physical press of the server power button."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def hold_pwr_btn(self):
"""Simulate a physical press and hold of the server power button."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def set_host_power(self, power):
"""Toggle the power button of server.
:param power: 'ON' or 'OFF'
"""
raise exception.IloCommandNotSupportedError(ERRMSG)
def set_one_time_boot(self, value):
"""Configures a single boot from a specific device."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def insert_virtual_media(self, url, device='FLOPPY'):
"""Notifies iLO of the location of a virtual media diskette image."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def eject_virtual_media(self, device='FLOPPY'):
"""Ejects the Virtual Media image if one is inserted."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def set_vm_status(self, device='FLOPPY',
boot_option='BOOT_ONCE', write_protect='YES'):
"""Sets the Virtual Media drive status and allows the
boot options for booting from the virtual media.
"""
raise exception.IloCommandNotSupportedError(ERRMSG)
def get_current_boot_mode(self):
"""Retrieves the current boot mode settings."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def get_pending_boot_mode(self):
"""Retrieves the pending boot mode settings."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def get_supported_boot_mode(self):
"""Retrieves the supported boot mode."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def set_pending_boot_mode(self, value):
"""Sets the boot mode of the system for next boot."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def get_persistent_boot(self):
"""Retrieves the boot order of the host."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def get_persistent_boot_device(self):
"""Get the current persistent boot device set for the host."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def set_persistent_boot(self, values=[]):
"""Configures to boot from a specific device."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def update_persistent_boot(self, device_type=[]):
"""Updates persistent boot based on the boot mode."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def get_secure_boot_state(self):
"""Get the status if secure boot is enabled or not."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def set_secure_boot_state(self, secure_boot_enable):
"""Enable/Disable secure boot on the server."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def reset_secure_boot_keys(self):
"""Reset secure boot keys to manufacturing defaults."""
raise exception.IloCommandNotSupportedError(ERRMSG)
def clear_secure_boot_keys(self):
"""Reset all keys."""
raise exception.IloCommandNotSupportedError(ERRMSG)

103
proliantutils/ilo/ribcl.py
View File

@@ -23,6 +23,10 @@ import xml.etree.ElementTree as etree
import six
from proliantutils.ilo import exception
from proliantutils.ilo import operations
POWER_STATE = {
'ON': 'Yes',
'OFF': 'No',
@@ -36,74 +40,11 @@ BOOT_MODE_CMDS = [
]
class IloError(Exception):
"""Base Exception
This exception is used when a problem is encountered in
executing an operation on the iLO.
"""
def __init__(self, message, errorcode=None):
super(IloError, self).__init__(message)
class IloClientInternalError(IloError):
"""Internal Error from IloClient
This exception is raised when iLO client library fails to
communicate properly with the iLO.
"""
def __init__(self, message, errorcode=None):
super(IloError, self).__init__(message)
class IloCommandNotSupportedError(IloError):
"""Command not supported on the platform.
This exception is raised when iLO client library fails to
communicate properly with the iLO
"""
def __init__(self, message, errorcode=None):
super(IloError, self).__init__(message)
class IloLoginFailError(IloError):
"""iLO Login Failed.
This exception is used to communicate a login failure to
the caller.
"""
messages = ['User login name was not found',
'Login failed', 'Login credentials rejected']
statuses = [0x005f, 0x000a]
class IloConnectionError(IloError):
"""Cannot connect to iLO.
This exception is used to communicate an HTTP connection
error from the iLO to the caller.
"""
def __init__(self, message):
super(IloConnectionError, self).__init__(message)
class IloInvalidInputError(IloError):
"""Invalid Input passed
This exception is used when invalid inputs are passed to
the APIs exposed by this module.
"""
def __init__(self, message):
super(IloInvalidInputError, self).__init__(message)
class IloClient:
class RIBCLOperations(operations.IloOperations):
"""iLO class for RIBCL interface for iLO.
This class provides an OO interface for retrieving information
and managing iLO. This class currently uses RIBCL scripting
language to talk to the iLO. It implements the same interface in
python as described in HP iLO 4 Scripting and Command Line Guide.
Implements the base class using RIBCL scripting language to talk
to the iLO.
"""
def __init__(self, host, login, password, timeout=60, port=443):
self.host = host
@@ -130,7 +71,7 @@ class IloClient:
req.add_header("Content-length", len(xml))
data = urllib2.urlopen(req).read()
except (ValueError, urllib2.URLError, urllib2.HTTPError) as e:
raise IloConnectionError(e)
raise exception.IloConnectionError(e)
return data
def _create_dynamic_xml(self, cmdname, tag_name, mode, subelements=None):
@@ -261,7 +202,7 @@ class IloClient:
# XML response is returned by Ilo. Set status to some
# arbitary non-zero value.
status = -1
raise IloClientInternalError(message, status)
raise exception.IloClientInternalError(message, status)
for child in message:
if child.tag != 'RESPONSE':
@@ -275,13 +216,14 @@ class IloClient:
for cmd in BOOT_MODE_CMDS:
if cmd in msg:
msg = "%s not supported on this platform." % cmd
raise IloCommandNotSupportedError(msg, status)
raise (exception.IloCommandNotSupportedError
(msg, status))
else:
raise IloClientInternalError(msg, status)
if (status in IloLoginFailError.statuses or
msg in IloLoginFailError.messages):
raise IloLoginFailError(msg, status)
raise IloError(msg, status)
raise exception.IloClientInternalError(msg, status)
if (status in exception.IloLoginFailError.statuses or
msg in exception.IloLoginFailError.messages):
raise exception.IloLoginFailError(msg, status)
raise exception.IloError(msg, status)
def _execute_command(self, create_command, tag_info, mode, dic={}):
"""Execute a command on the iLO.
@@ -304,6 +246,13 @@ class IloClient:
d[key] = data['GET_ALL_LICENSES']['LICENSE'][key]['VALUE']
return d
def get_product_name(self):
"""Get the model name of the queried server."""
data = self._execute_command(
'GET_PRODUCT_NAME', 'SERVER_INFO', 'read')
return data['GET_PRODUCT_NAME']['PRODUCT_NAME']['VALUE']
def get_host_power_status(self):
"""Request the power state of the server."""
data = self._execute_command(
@@ -351,7 +300,7 @@ class IloClient:
'SET_HOST_POWER', 'SERVER_INFO', 'write', dic)
return data
else:
raise IloInvalidInputError(
raise exception.IloInvalidInputError(
"Invalid input. The expected input is ON or OFF.")
def set_one_time_boot(self, value):
@@ -498,7 +447,7 @@ class IloClient:
nic_list.append(item["value"])
except KeyError as e:
msg = "_get_nic_boot_devices failed with the KeyError:%s"
raise IloError((msg) % e)
raise exception.IloError((msg) % e)
all_nics = pxe_nic_list + nic_list
return all_nics
@@ -515,6 +464,6 @@ class IloClient:
disk_list.append(item["value"])
except KeyError as e:
msg = "_get_disk_boot_devices failed with the KeyError:%s"
raise IloError((msg) % e)
raise exception.IloError((msg) % e)
return disk_list

470
proliantutils/ilo/ris.py Normal file
View File

@@ -0,0 +1,470 @@
# Copyright 2014 Hewlett-Packard Development Company, L.P.
#
# 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.
__author__ = 'HP'
import base64
import gzip
import hashlib
import httplib
import json
import StringIO
import urlparse
from proliantutils.ilo import exception
from proliantutils.ilo import operations
""" Currently this class supports only secure boot and firmware settings
related API's .
TODO : Add rest of the API's that exists in RIBCL. """
class RISOperations(operations.IloOperations):
def __init__(self, host, login, password, bios_password=None):
self.host = host
self.login = login
self.password = password
self.bios_password = bios_password
# Message registry support
self.message_registries = {}
def _rest_op(self, operation, suburi, request_headers, request_body):
"""Generic REST Operation handler."""
url = urlparse.urlparse('https://' + self.host + suburi)
if request_headers is None:
request_headers = dict()
# Use self.login/self.password and Basic Auth
if self.login is not None and self.password is not None:
hr = "BASIC " + base64.b64encode(self.login + ":" + self.password)
request_headers['Authorization'] = hr
redir_count = 4
while redir_count:
conn = None
if url.scheme == 'https':
conn = httplib.HTTPSConnection(host=url.netloc, strict=True)
elif url.scheme == 'http':
conn = httplib.HTTPConnection(host=url.netloc, strict=True)
try:
conn.request(operation, url.path, headers=request_headers,
body=json.dumps(request_body))
resp = conn.getresponse()
body = resp.read()
except Exception as e:
raise exception.IloConnectionError(e)
# NOTE:Do not assume every HTTP operation will return a JSON body.
# For example, ExtendedError structures are only required for
# HTTP 400 errors and are optional elsewhere as they are mostly
# redundant for many of the other HTTP status code. In particular,
# 200 OK responses should not have to return any body.
# NOTE: this makes sure the headers names are all lower cases
# because HTTP says they are case insensitive
headers = dict((x.lower(), y) for x, y in resp.getheaders())
# Follow HTTP redirect
if resp.status == 301 and 'location' in headers:
url = urlparse.urlparse(headers['location'])
redir_count -= 1
else:
break
response = dict()
try:
if body:
response = json.loads(body.decode('utf-8'))
except ValueError:
# if it doesn't decode as json
# NOTE: resources may return gzipped content
# try to decode as gzip (we should check the headers for
# Content-Encoding=gzip)
try:
gzipper = gzip.GzipFile(fileobj=StringIO.StringIO(body))
uncompressed_string = gzipper.read().decode('UTF-8')
response = json.loads(uncompressed_string)
except Exception as e:
raise exception.IloError(e)
return resp.status, headers, response
def _rest_get(self, suburi, request_headers=None):
"""REST GET operation.
HTTP response codes could be 500, 404 etc.
"""
return self._rest_op('GET', suburi, request_headers, None)
def _rest_patch(self, suburi, request_headers, request_body):
"""REST PATCH operation.
HTTP response codes could be 500, 404, 202 etc.
"""
if not isinstance(request_headers, dict):
request_headers = dict()
request_headers['Content-Type'] = 'application/json'
return self._rest_op('PATCH', suburi, request_headers, request_body)
def _rest_put(self, suburi, request_headers, request_body):
"""REST PUT operation.
HTTP response codes could be 500, 404, 202 etc.
"""
if not isinstance(request_headers, dict):
request_headers = dict()
request_headers['Content-Type'] = 'application/json'
return self._rest_op('PUT', suburi, request_headers, request_body)
def _rest_post(self, suburi, request_headers, request_body):
"""REST POST operation.
The response body after the operation could be the new resource, or
ExtendedError, or it could be empty.
"""
if not isinstance(request_headers, dict):
request_headers = dict()
request_headers['Content-Type'] = 'application/json'
return self._rest_op('POST', suburi, request_headers, request_body)
def _rest_delete(self, suburi, request_headers):
"""REST DELETE operation.
HTTP response codes could be 500, 404 etc.
"""
return self._rest_op('DELETE', suburi, request_headers, None)
def _get_type(self, obj):
"""Return the type of an object."""
typever = obj['Type']
typesplit = typever.split('.')
return typesplit[0] + '.' + typesplit[1]
def _operation_allowed(self, headers_dict, operation):
"""Checks if specified operation is allowed on the resource."""
if 'allow' in headers_dict:
if operation in headers_dict['allow']:
return True
return False
def _render_extended_error_message_list(self, extended_error):
"""Parse the ExtendedError object and retruns the message.
Build a list of decoded messages from the extended_error using the
message registries. An ExtendedError JSON object is a response from
the with its own schema. This function knows how to parse the
ExtendedError object and, using any loaded message registries,
render an array of plain language strings that represent
the response.
"""
messages = []
if isinstance(extended_error, dict):
if ('Type' in extended_error and
extended_error['Type'].startswith('ExtendedError.')):
for msg in extended_error['Messages']:
message_id = msg['MessageID']
x = message_id.split('.')
registry = x[0]
msgkey = x[len(x) - 1]
# if the correct message registry is loaded,
# do string resolution
if (registry in self.message_registries and msgkey in
self.message_registries[registry]['Messages']):
rmsgs = self.message_registries[registry]['Messages']
msg_dict = rmsgs[msgkey]
msg_str = message_id + ': ' + msg_dict['Message']
for argn in range(0, msg_dict['NumberOfArgs']):
subst = '%' + str(argn+1)
m = str(msg['MessageArgs'][argn])
msg_str = msg_str.replace(subst, m)
if ('Resolution' in msg_dict and
msg_dict['Resolution'] != 'None'):
msg_str += ' ' + msg_dict['Resolution']
messages.append(msg_str)
else:
# no message registry, simply return the msg object
# in string form
messages.append(str(message_id))
return messages
def _get_extended_error(self, extended_error):
"""Gets the list of decoded messages from the extended_error."""
return self._render_extended_error_message_list(extended_error)
def _get_host_details(self):
"""Get the system details."""
# Assuming only one system present as part of collection,
# as we are dealing with iLO's here.
status, headers, system = self._rest_get('/rest/v1/Systems/1')
if status < 300:
stype = self._get_type(system)
if not (stype == 'ComputerSystem.0' or
stype(system) == 'ComputerSystem.1'):
msg = "%s is not a valid system type " % stype
raise exception.IloError(msg)
else:
msg = self._get_extended_error(system)
raise exception.IloError(msg)
return system
def _check_bios_resource(self, properties=[]):
"""Check if the bios resource exists."""
system = self._get_host_details()
if ('links' in system['Oem']['Hp'] and
'BIOS' in system['Oem']['Hp']['links']):
# Get the BIOS URI and Settings
bios_uri = system['Oem']['Hp']['links']['BIOS']['href']
status, headers, bios_settings = self._rest_get(bios_uri)
if status >= 300:
msg = self._get_extended_error(bios_settings)
raise exception.IloError(msg)
# If property is not None, check if the bios_property is supported
for property in properties:
if property not in bios_settings:
# not supported on this platform
msg = ('\tBIOS Property "' + property + '" is not'
' supported on this system.')
raise exception.IloCommandNotSupportedError(msg)
return headers, bios_settings
else:
msg = ('"links/BIOS" section in ComputerSystem/Oem/Hp'
' does not exist')
raise exception.IloCommandNotSupportedError(msg)
def _get_bios_setting(self, bios_property):
"""Retrieves bios settings of the server."""
headers, bios_settings = self._check_bios_resource([bios_property])
return bios_settings[bios_property]
def _change_bios_setting(self, properties):
"""Change the bios settings to specified values."""
# Get the keys to check if keys are supported.
keys = properties.keys()
# Check if the BIOS resource/property if exists.
headers, bios_settings = self._check_bios_resource(keys)
# if this BIOS resource doesn't support PATCH, go get the Settings.
if not self._operation_allowed(headers, 'PATCH'): # this is GET-only
bios_uri = bios_settings['links']['Settings']['href']
status, headers, bios_settings = self._rest_get(bios_uri)
# this should allow PATCH, else raise error
if not self._operation_allowed(headers, 'PATCH'):
msg = ('PATCH Operation not supported on the resource'
'%s ' % bios_uri)
raise exception.IloError(msg)
request_headers = dict()
if self.bios_password:
bios_password_hash = hashlib.sha256((self.bios_password.encode()).
hexdigest().upper())
request_headers['X-HPRESTFULAPI-AuthToken'] = bios_password_hash
# perform the patch
status, headers, response = self._rest_patch(bios_uri, request_headers,
properties)
if status >= 300:
msg = self._get_extended_error(response)
raise exception.IloError(msg)
def _reset_to_default(self):
"""Change to default bios setting to default values."""
# Check if the BIOS resource if exists.
headers_bios, bios_settings = self._check_bios_resource()
# Get the default configs
base_config_uri = bios_settings['links']['BaseConfigs']['href']
status, headers, config = self._rest_get(base_config_uri)
if status >= 300:
msg = self._get_extended_error(config)
raise exception.IloError(msg)
# if this BIOS resource doesn't support PATCH, go get the Settings
if not self._operation_allowed(headers_bios, 'PATCH'):
# this is GET-only
bios_uri = bios_settings['links']['Settings']['href']
status, headers, bios_settings = self._rest_get(bios_uri)
# this should allow PATCH, else raise error
if not self._operation_allowed(headers, 'PATCH'):
msg = ('PATCH Operation not supported on the resource'
'%s ' % bios_uri)
raise exception.IloError(msg)
new_bios_settings = config['BaseConfigs'][0]['default']
request_headers = dict()
if self.bios_password:
bios_password_hash = hashlib.sha256((self.bios_password.encode()).
hexdigest().upper())
request_headers['X-HPRESTFULAPI-AuthToken'] = bios_password_hash
# perform the patch
status, headers, response = self._rest_patch(bios_uri, request_headers,
new_bios_settings)
if status >= 300:
msg = self._get_extended_error(response)
raise exception.IloError(msg)
def _change_secure_boot_settings(self, property, value):
"""Change secure boot settings on the server."""
system = self._get_host_details()
# find the BIOS URI
if ('links' not in system['Oem']['Hp'] or
'SecureBoot' not in system['Oem']['Hp']['links']):
msg = (' "SecureBoot" resource or feature is not '
'supported on this system')
raise exception.IloCommandNotSupportedError(msg)
secure_boot_uri = system['Oem']['Hp']['links']['SecureBoot']['href']
# Change the property required
new_secure_boot_settings = dict()
new_secure_boot_settings[property] = value
# perform the patch
status, headers, response = self._rest_patch(
secure_boot_uri, None, new_secure_boot_settings)
if status >= 300:
msg = self._get_extended_error(response)
raise exception.IloError(msg)
# Change the bios setting as a workaround to enable secure boot
# Can be removed when fixed for Gen9 snap2
val = self._get_bios_setting('CustomPostMessage')
val = val.rstrip() if val.endswith(" ") else val+" "
self._change_bios_setting({'CustomPostMessage': val})
def get_secure_boot_mode(self):
"""Get the status of secure boot.
:returns: True, if enabled, else False
:raises: IloError, on an error from iLO.
:raises: IloCommandNotSupportedError, if the command is not supported
on the server.
"""
system = self._get_host_details()
if ('links' not in system['Oem']['Hp'] or
'SecureBoot' not in system['Oem']['Hp']['links']):
msg = ('"SecureBoot" resource or feature is not supported'
' on this system')
raise exception.IloCommandNotSupportedError(msg)
secure_boot_uri = system['Oem']['Hp']['links']['SecureBoot']['href']
# get the Secure Boot object
status, headers, secure_boot_settings = self._rest_get(secure_boot_uri)
if status >= 300:
msg = self._get_extended_error(system)
raise exception.IloError(msg)
return secure_boot_settings['SecureBootCurrentState']
def set_secure_boot_mode(self, secure_boot_enable):
"""Enable/Disable secure boot on the server.
:param secure_boot_enable: True, if secure boot needs to be
enabled for next boot, else False.
:raises: IloError, on an error from iLO.
:raises: IloCommandNotSupportedError, if the command is not supported
on the server.
"""
self._change_secure_boot_settings('SecureBootEnable',
secure_boot_enable)
def reset_secure_boot_keys(self):
"""Reset secure boot keys to manufacturing defaults.
:raises: IloError, on an error from iLO.
:raises: IloCommandNotSupportedError, if the command is not supported
on the server.
"""
self._change_secure_boot_settings('ResetToDefaultKeys', True)
def clear_secure_boot_keys(self):
"""Reset all keys.
:raises: IloError, on an error from iLO.
:raises: IloCommandNotSupportedError, if the command is not supported
on the server.
"""
self._change_secure_boot_settings('ResetAllKeys', True)
def get_host_power_status(self):
"""Request the power state of the server.
:returns: Power State of the server, 'ON' or 'OFF'
:raises: IloError, on an error from iLO.
"""
data = self._get_host_details()
return data['Power'].upper()
def get_current_boot_mode(self):
"""Retrieves the current boot mode of the server.
:returns: Current boot mode, LEGACY or UEFI.
:raises: IloError, on an error from iLO.
"""
boot_mode = self._get_bios_setting('BootMode')
if boot_mode == 'LegacyBios':
boot_mode = 'legacy'
return boot_mode.upper()
def set_pending_boot_mode(self, boot_mode):
"""Sets the boot mode of the system for next boot.
:param boot_mode: either 'uefi' or 'bios'.
:raises: IloInvalidInputError, on an invalid input.
:raises: IloError, on an error from iLO.
:raises: IloCommandNotSupportedError, if the command is not supported
on the server.
"""
if boot_mode not in ['uefi', 'bios']:
msg = 'Invalid Boot mode specified'
raise exception.IloInvalidInputError(msg)
boot_properties = {'BootMode': boot_mode}
if boot_mode == 'bios':
boot_properties['BootMode'] = 'LegacyBios'
else:
# If Boot Mode is 'Uefi' set the UEFIOptimizedBoot first.
boot_properties['UefiOptimizedBoot'] = "Enabled"
# Change the Boot Mode
self._change_bios_setting(boot_properties)

43
proliantutils/tests/ilo/test_ribcl.py
View File

@@ -20,6 +20,7 @@ import unittest
import constants
import mock
from proliantutils.ilo import exception
from proliantutils.ilo import ribcl
@@ -27,25 +28,25 @@ class IloRibclTestCase(unittest.TestCase):
def setUp(self):
super(IloRibclTestCase, self).setUp()
self.ilo = ribcl.IloClient("x.x.x.x", "admin", "Admin", 60, 443)
self.ilo = ribcl.RIBCLOperations("x.x.x.x", "admin", "Admin", 60, 443)
def test__request_ilo_connection_failed(self):
self.assertRaises(ribcl.IloConnectionError,
self.assertRaises(exception.IloConnectionError,
self.ilo.get_all_licenses)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_login_fail(self, request_ilo_mock):
request_ilo_mock.return_value = constants.LOGIN_FAIL_XML
self.assertRaises(ribcl.IloError,
self.assertRaises(exception.IloError,
self.ilo.get_all_licenses)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_hold_pwr_btn(self, request_ilo_mock):
request_ilo_mock.return_value = constants.HOLD_PWR_BTN_XML
result = self.ilo.hold_pwr_btn()
self.assertIn('Host power is already OFF.', result)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_get_vm_status_none(self, request_ilo_mock):
request_ilo_mock.return_value = constants.GET_VM_STATUS_XML
result = self.ilo.get_vm_status()
@@ -57,7 +58,7 @@ class IloRibclTestCase(unittest.TestCase):
self.assertIn('IMAGE_INSERTED', result)
self.assertIn('BOOT_OPTION', result)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_get_vm_status_cdrom(self, request_ilo_mock):
request_ilo_mock.return_value = constants.GET_VM_STATUS_CDROM_XML
result = self.ilo.get_vm_status('cdrom')
@@ -69,13 +70,13 @@ class IloRibclTestCase(unittest.TestCase):
self.assertIn('IMAGE_INSERTED', result)
self.assertIn('BOOT_OPTION', result)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_get_vm_status_error(self, request_ilo_mock):
request_ilo_mock.return_value = constants.GET_VM_STATUS_ERROR_XML
self.assertRaises(
ribcl.IloError, self.ilo.get_vm_status)
exception.IloError, self.ilo.get_vm_status)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_get_all_licenses(self, request_ilo_mock):
request_ilo_mock.return_value = constants.GET_ALL_LICENSES_XML
result = self.ilo.get_all_licenses()
@@ -85,58 +86,58 @@ class IloRibclTestCase(unittest.TestCase):
self.assertIn('LICENSE_KEY', result)
self.assertIn('LICENSE_CLASS', result)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_get_one_time_boot(self, request_ilo_mock):
request_ilo_mock.return_value = constants.GET_ONE_TIME_BOOT_XML
result = self.ilo.get_one_time_boot()
self.assertIn('NORMAL', result.upper())
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_get_host_power_status(self, request_ilo_mock):
request_ilo_mock.return_value = constants.GET_HOST_POWER_STATUS_XML
result = self.ilo.get_host_power_status()
self.assertIn('ON', result)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_reset_server(self, request_ilo_mock):
request_ilo_mock.return_value = constants.RESET_SERVER_XML
result = self.ilo.reset_server()
self.assertIn('server being reset', result.lower())
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_press_pwr_btn(self, request_ilo_mock):
request_ilo_mock.return_value = constants.PRESS_POWER_BTN_XML
result = self.ilo.press_pwr_btn()
self.assertIsNone(result)
self.assertTrue(request_ilo_mock.called)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_set_host_power(self, request_ilo_mock):
request_ilo_mock.return_value = constants.SET_HOST_POWER_XML
result = self.ilo.set_host_power('ON')
self.assertIn('Host power is already ON.', result)
self.assertRaises(ribcl.IloInvalidInputError,
self.assertRaises(exception.IloInvalidInputError,
self.ilo.set_host_power, 'ErrorCase')
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_set_one_time_boot(self, request_ilo_mock):
request_ilo_mock.return_value = constants.SET_ONE_TIME_BOOT_XML
self.ilo.set_one_time_boot('NORMAL')
self.assertTrue(request_ilo_mock.called)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_insert_virtual_media(self, request_ilo_mock):
request_ilo_mock.return_value = constants.INSERT_VIRTUAL_MEDIA_XML
result = self.ilo.insert_virtual_media('any_url', 'floppy')
self.assertIsNone(result)
self.assertTrue(request_ilo_mock.called)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_eject_virtual_media(self, request_ilo_mock):
request_ilo_mock.return_value = constants.EJECT_VIRTUAL_MEDIA_XML
self.assertRaises(ribcl.IloError, self.ilo.eject_virtual_media)
self.assertRaises(exception.IloError, self.ilo.eject_virtual_media)
@mock.patch.object(ribcl.IloClient, '_request_ilo')
@mock.patch.object(ribcl.RIBCLOperations, '_request_ilo')
def test_set_vm_status(self, request_ilo_mock):
request_ilo_mock.return_value = constants.SET_VM_STATUS_XML
self.ilo.set_vm_status('cdrom', 'boot_once', 'yes')