<s> """<STR_LIT>""" <EOL> import abc <EOL> import os <EOL> import random <EOL> import re <EOL> import shutil <EOL> import socket <EOL> import ssl <EOL> import subprocess <EOL> import sys <EOL> import tempfile <EOL> import types <EOL> import uuid <EOL> from oslo_concurrency import processutils as utils <EOL> import six <EOL> from proliantutils import exception <EOL> from proliantutils . ilo import common <EOL> from proliantutils import log <EOL> if six . PY3 : <EOL> def b ( x ) : <EOL> return bytes ( x , '<STR_LIT:ascii>' ) <EOL> else : <EOL> def b ( x ) : <EOL> return x <EOL> LOG = log . get_logger ( __name__ ) <EOL> SUPPORTED_FIRMWARE_UPDATE_COMPONENTS = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> RAW_FIRMWARE_EXTNS = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> def check_firmware_update_component ( func ) : <EOL> """<STR_LIT>""" <EOL> @ six . wraps ( func ) <EOL> def wrapper ( self , filename , component_type ) : <EOL> """<STR_LIT>""" <EOL> component_type = component_type and component_type . lower ( ) <EOL> if ( component_type not in SUPPORTED_FIRMWARE_UPDATE_COMPONENTS ) : <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> { '<STR_LIT>' : component_type } ) <EOL> LOG . error ( self . _ ( msg ) ) <EOL> raise exception . InvalidInputError ( msg ) <EOL> return func ( self , filename , component_type ) <EOL> return wrapper <EOL> @ six . add_metaclass ( abc . ABCMeta ) <EOL> class FirmwareImageControllerBase ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fw_file ) : <EOL> self . fw_file = fw_file <EOL> file_name , file_ext_with_dot = common . get_filename_and_extension_of ( <EOL> fw_file ) <EOL> self . fw_filename = file_name <EOL> self . fw_file_ext = file_ext_with_dot <EOL> class FirmwareImageUploader ( FirmwareImageControllerBase ) : <EOL> """<STR_LIT>""" <EOL> HTTP_UPLOAD_HEADER = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def upload_file_to ( self , addressinfo , timeout ) : <EOL> """<STR_LIT>""" <EOL> self . hostname , self . port = addressinfo <EOL> self . timeout = timeout <EOL> filename = self . fw_file <EOL> firmware = open ( filename , '<STR_LIT:rb>' ) . read ( ) <EOL> boundary = b ( '<STR_LIT>' + <EOL> str ( random . randint ( <NUM_LIT> , <NUM_LIT> ) ) + '<STR_LIT:z>' ) <EOL> while boundary in firmware : <EOL> boundary = b ( '<STR_LIT>' + <EOL> str ( random . randint ( <NUM_LIT> , <NUM_LIT> ) ) + '<STR_LIT:z>' ) <EOL> parts = [ <EOL> b ( "<STR_LIT>" ) + boundary + <EOL> b ( """<STR_LIT>""" <EOL> """<STR_LIT>""" ) , <EOL> b ( "<STR_LIT>" ) + boundary + <EOL> b ( '''<STR_LIT>''' <EOL> '''<STR_LIT>''' ) + <EOL> b ( filename ) + <EOL> b ( '''<STR_LIT>''' ) , <EOL> firmware , <EOL> b ( "<STR_LIT>" ) + boundary + b ( "<STR_LIT>" ) , <EOL> ] <EOL> total_bytes = sum ( [ len ( x ) for x in parts ] ) <EOL> sock = self . _get_socket ( ) <EOL> sock . write ( b ( self . HTTP_UPLOAD_HEADER % <EOL> ( total_bytes , boundary . decode ( '<STR_LIT:ascii>' ) ) ) ) <EOL> for part in parts : <EOL> sock . write ( part ) <EOL> data = '<STR_LIT>' <EOL> try : <EOL> while True : <EOL> d = sock . read ( ) <EOL> data += d . decode ( '<STR_LIT>' ) <EOL> if not d : <EOL> break <EOL> except socket . sslerror : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> if not data : <EOL> raise exception . IloConnectionError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % { '<STR_LIT>' : self . hostname , <EOL> '<STR_LIT:port>' : self . port , '<STR_LIT:error>' : str ( e ) } ) <EOL> cookie_match = re . search ( '<STR_LIT>' , data ) <EOL> if not cookie_match : <EOL> raise exception . IloError ( "<STR_LIT>" <EOL> "<STR_LIT>" % filename ) <EOL> return cookie_match . group ( <NUM_LIT:1> ) <EOL> def _get_socket ( self , sslversion = ssl . PROTOCOL_TLSv1 ) : <EOL> """<STR_LIT>""" <EOL> err = None <EOL> sock = None <EOL> try : <EOL> for res in socket . getaddrinfo ( <EOL> self . hostname , self . port , <NUM_LIT:0> , socket . SOCK_STREAM ) : <EOL> af , socktype , proto , canonname , sa = res <EOL> try : <EOL> sock = socket . socket ( af , socktype , proto ) <EOL> sock . settimeout ( self . timeout ) <EOL> sock . connect ( sa ) <EOL> except socket . timeout : <EOL> if sock is not None : <EOL> sock . close ( ) <EOL> err = exception . IloConnectionError ( <EOL> "<STR_LIT>" <EOL> % { '<STR_LIT>' : self . hostname , '<STR_LIT:port>' : self . port } ) <EOL> except socket . error : <EOL> if sock is not None : <EOL> sock . close ( ) <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> err = exception . IloConnectionError ( <EOL> "<STR_LIT>" <EOL> % { '<STR_LIT>' : self . hostname , '<STR_LIT:port>' : self . port , <EOL> '<STR_LIT:error>' : str ( e ) } ) <EOL> except Exception : <EOL> raise exception . IloConnectionError ( <EOL> "<STR_LIT>" % self . hostname ) <EOL> if err is not None : <EOL> raise err <EOL> try : <EOL> return ssl . wrap_socket ( sock , ssl_version = sslversion ) <EOL> except socket . sslerror : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> msg = ( getattr ( e , '<STR_LIT>' , None ) or <EOL> getattr ( e , '<STR_LIT:message>' , None ) ) <EOL> if ( '<STR_LIT>' in msg ) and ( <EOL> sslversion == ssl . PROTOCOL_TLSv1 ) : <EOL> return self . _get_socket ( ssl . PROTOCOL_SSLv3 ) <EOL> raise exception . IloConnectionError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % { '<STR_LIT>' : self . hostname , '<STR_LIT:port>' : self . port , <EOL> '<STR_LIT:error>' : str ( e ) } ) <EOL> class FirmwareImageExtractor ( FirmwareImageControllerBase ) : <EOL> """<STR_LIT>""" <EOL> def extract ( self ) : <EOL> """<STR_LIT>""" <EOL> target_file = self . fw_file <EOL> common . add_exec_permission_to ( target_file ) <EOL> temp_dir = tempfile . mkdtemp ( ) <EOL> extract_path = os . path . join ( temp_dir , self . fw_filename ) <EOL> try : <EOL> self . _do_extract ( target_file , extract_path ) <EOL> except exception . ImageExtractionFailed : <EOL> shutil . rmtree ( temp_dir , ignore_errors = True ) <EOL> raise <EOL> firmware_file_path = _get_firmware_file_in_new_path ( extract_path ) <EOL> shutil . rmtree ( temp_dir , ignore_errors = True ) <EOL> if not firmware_file_path : <EOL> raise exception . InvalidInputError ( <EOL> "<STR_LIT>" % target_file ) <EOL> return firmware_file_path , True <EOL> def get_fw_extractor ( fw_file ) : <EOL> """<STR_LIT>""" <EOL> fw_img_extractor = FirmwareImageExtractor ( fw_file ) <EOL> extension = fw_img_extractor . fw_file_ext . lower ( ) <EOL> if extension == '<STR_LIT>' : <EOL> fw_img_extractor . _do_extract = types . MethodType ( <EOL> _extract_scexe_file , fw_img_extractor ) <EOL> elif extension == '<STR_LIT>' : <EOL> fw_img_extractor . _do_extract = types . MethodType ( <EOL> _extract_rpm_file , fw_img_extractor ) <EOL> elif extension in RAW_FIRMWARE_EXTNS : <EOL> def dummy_extract ( self ) : <EOL> """<STR_LIT>""" <EOL> return fw_img_extractor . fw_file , False <EOL> fw_img_extractor . extract = types . MethodType ( <EOL> dummy_extract , fw_img_extractor ) <EOL> else : <EOL> raise exception . InvalidInputError ( <EOL> '<STR_LIT>' % fw_file ) <EOL> return fw_img_extractor <EOL> def _extract_scexe_file ( self , target_file , extract_path ) : <EOL> """<STR_LIT>""" <EOL> unpack_cmd = '<STR_LIT>' + extract_path <EOL> cmd = [ target_file , unpack_cmd ] <EOL> out , err = utils . trycmd ( * cmd ) <EOL> def _extract_rpm_file ( self , target_file , extract_path ) : <EOL> """<STR_LIT>""" <EOL> if not os . path . exists ( extract_path ) : <EOL> os . makedirs ( extract_path ) <EOL> os . chdir ( extract_path ) <EOL> try : <EOL> rpm2cpio = subprocess . Popen ( '<STR_LIT>' + target_file , <EOL> shell = True , <EOL> stdout = subprocess . PIPE ) <EOL> cpio = subprocess . Popen ( '<STR_LIT>' , shell = True , <EOL> stdin = rpm2cpio . stdout ) <EOL> out , err = cpio . communicate ( ) <EOL> except Exception : <EOL> raise exception . ImageExtractionFailed ( <EOL> image_ref = target_file , <EOL> reason = '<STR_LIT>' ) <EOL> def _get_firmware_file ( path ) : <EOL> """<STR_LIT>""" <EOL> for dirpath , dirnames , filenames in os . walk ( path ) : <EOL> for filename in filenames : <EOL> file_name , file_ext = os . path . splitext ( os . path . basename ( filename ) ) <EOL> if file_ext in RAW_FIRMWARE_EXTNS : <EOL> return os . path . join ( dirpath , filename ) <EOL> def _get_firmware_file_in_new_path ( searching_path ) : <EOL> """<STR_LIT>""" <EOL> firmware_file_path = _get_firmware_file ( searching_path ) <EOL> if not firmware_file_path : <EOL> return None <EOL> file_name , file_ext_with_dot = common . get_filename_and_extension_of ( <EOL> firmware_file_path ) <EOL> new_firmware_file_path = os . path . join ( <EOL> tempfile . gettempdir ( ) , str ( uuid . uuid4 ( ) ) + '<STR_LIT:_>' + <EOL> file_name + file_ext_with_dot ) <EOL> os . link ( firmware_file_path , new_firmware_file_path ) <EOL> return new_firmware_file_path </s>
