#!/usr/bin/python ################################################################################ # Name: rhnTools.py # # Version: 1.0.0 # # Date: 2012-08-28 # # Author: VA ESE Enterprise Testing Services # # # # Commonly Used Python Objects # # # # Classes # # rhnAPI # # rhnAPI creates an object to hold command line options and arguments, # # RHN Satellite APP connection information and object lists that only # # need to be accessed once during a session. It provides methods to get # # RHN Satellite objects into dictionaries of dictionaries in a form # # that can be exported, imported and compared. It processes command # # line options and arguments upon initialization and establishes a # # connection to the RHN Satellite the first time a RHN Satellite object # # is referenced. All RHN Satellite related exceptions are should be # # in the methods, so processing may continu after an exception occurs # # # ################################################################################ # # Import Python Libraries # import sys, traceback import xmlrpclib # # Import VA Custom Libraries # from pyTools import extractDict class rhnAPI: """ class to manage access to Red Hat Network (RHN) Satellite API upon initilization it processes command line options public methods are provided to get RHN Satellite objects and to create/update objects """ options = None # Initialize command line options args = None # Initialize command line arguments client = None # Initalize xmlrpc client sessionKey = None # Initialize client session Key org = None # Initialize users RHN Satellite Organization sysGroups = None # Initialize Kickstart Profile dictionary userInfo = {} # Initialize User Info dictionary def __init__(self, customOptions = None, descStr = None, epilogStr = None): """ when an instance of this class is created: \tprocess command line options and arguments \taccept an optional list of additional optparse.make_option objects \taccept an optional string with a program description \taccept an optional string with a program epilog \tsee optparse in the Python Library Reference for additional details """ from getpass import getuser # Import getuser function from optparse import OptionParser, OptionValueError, make_option # Import optionparser classes, exceptions and functions def getCommand(option, opt_str, value, parser): """ function to set first command option and error out on multiple commands """ if parser.values.command: # If a command option has already been set # raise an exception raise OptionValueError('cannot use %s with %s' % (opt_str, parser.values.command)) setattr(parser.values, 'command', opt_str.lstrip('-')) # Set the command option usageStr = 'usage: %prog {-l|-e|-i|-c} [options] [arg]' # String for usage message commonOptions = [ # Build standard option strings make_option('-l', '--list', action='callback', callback=getCommand, help='List command for %prog'), make_option('-e', '--export', action='callback', callback=getCommand, help='Export command for %prog'), make_option('-c', '--compare', action='callback', callback=getCommand, help='Compare command for %prog'), make_option('-i', '--import', action='callback', callback=getCommand, help='Import command for %prog'), make_option('-r', '--rhn-satellite', action='store', type='string', dest='satellite', default='localhost', help='RHN Satellite FQDN'), make_option('-u', '--user', action='store', type='string', dest='user', default=getuser(), help='RHN Satellite User ID'), make_option('-v', '--verbose', action='store_true', default=False, help='Enable verbose output messages'), make_option('-d', '--debug', action='count', default=0 , help='Increase debug output messages'), make_option( '--force', action='store_true', default=False, help='Force import to replace/update existing item') ] if customOptions: # If custom options were passed in from the calling application options = commonOptions + customOptions # Add them to the standard options else: # If custom options were not passed in options = commonOptions # Use the standard options # Create an OptionParser object using the messages and options parser = OptionParser(usage = usageStr, description = descStr, epilog = epilogStr, option_list = options) parser.set_defaults(command = None) # Set additional option defaults (self.options, self.args) = parser.parse_args() # Process command line options and arguments and save them locally parser.destroy() # Destroy the option parser now that it has done its job # Build the URL for the RHN Satellite self.client = xmlrpclib.ServerProxy('http://' + self.options.satellite + '/rpc/api') def _execAPI(self, myNamespace, myMethod, *myArgs): """ private method to use the current xmlrpclib client to execute an API call to a namespace and method. Requires a string with the namespace, a string with the method name and optional arguments. The method uses the session key from the class. """ myResult = None if self.sessionKey: # If a Session Key has been set apiArgs = (self.sessionKey, ) + myArgs # Prepend Session Key to the arguments else: # Else, this must be a call that does not use a Session Key apiArgs = myArgs # Just use the arguments passed in myAPI = getattr(self.client, # Assign the XML-RPC Method to myAPI '.'.join((myNamespace, myMethod))) if self.options.debug > 0: # If debug level > 0 print '\tCalling %s.%s%s' % (myNamespace, # Print the API call with arguments myMethod, ('sessionKey',) + myArgs) try: # Try to ... myResult = myAPI(*apiArgs) # Call the API with the arguments except xmlrpclib.Fault, err: # If an xmlrpclib.Fault occurs if err.faultCode == 2950: # If the faultCode is 2950, it is a bad User ID/Password print 'Incorrect User ID/Password' # Print an error message elif err.faultCode == -1: print 'Bad API call: %s.%s%s' % (myNamespace, myMethod, apiArgs) if self.options.debug > 0: # If debug level > 0 print traceback.print_exc(sys.exc_info()) # Print the fault information else: # If it is not faultCode 2950 print 'xmlrpclib.Fault:' # Print exception type print err.faultCode # Print the faultCode print err.faultString # and faultString if self.sessionKey: # If a session is connected self.client.auth.logout(self.sessionKey) # Logout of the session sys.exit(1) # Exit with a status of '1' except: # If any other exception occurs print 'An Exception occured in %s.%s%s' % (myNamespace, # Print the failed call myMethod, apiArgs) print traceback.print_exc(sys.exc_info()) # and the fault information if self.sessionKey: # If a session is connected self.client.auth.logout(self.sessionKey) # Logout of the session sys.exit(2) # Exit with a status of '2' return myResult def _connectSession(self): """ private method to establish connection to RHN Satellite and get users Organization """ from getpass import getpass # Import getpass function try: # Prompt for a password password = getpass('Enter RHN Satellite %s Password for %s: ' % (self.options.satellite, self.options.user)) except KeyboardInterrupt: # If a KeyboardInterrupt occurs print '\n' # Print a sys.exit(3) # and exit with a status of "3" except: # If any other error occurs print traceback.print_exc(sys.exc_info()) # Print the error sys.exit(9) # and exit with a status of "9" self.sessionKey = self._execAPI('auth', # Establish a connection to the RHN Satellite, saving the session key 'login', self.options.user, password) userDetails = self._execAPI('user', # Get the user details for the current users 'getDetails', self.options.user) self.org = userDetails.get('org_id') # Save the current users organization ID # Get a dictionary of Kickstart Profiles using ID as the key and name as the value self.sysGroups = dict((d['id'], d['name']) for d in self._execAPI('systemgroup', 'listAllGroups')) userNames = list(d['login'] for d in self._execAPI('user', # Get a list of User Logins for the current organization 'listUsers')) for userName in userNames: # For each user login lstRoles = self._execAPI('user', # Get their roles 'listRoles', userName) lstDefSysGrps = list(d['name'] for d in self._execAPI('user', # Get their Default Kickstart Profiles 'listDefaultSystemGroups', userName)) lstAssSysGrps = list(d['name'] for d in self._execAPI('user', # Get their Assigned Kickstart Profiles 'listAssignedSystemGroups', userName)) self.userInfo[userName] = {'roles' : lstRoles, # And save them in userInfo dictionary 'default_system_groups' : lstDefSysGrps, 'assigned_system_groups' : lstAssSysGrps} def closeSession(self): """ method to close this object's RHN Satellite Session """ if self.sessionKey: # If a session Key has been saved self._execAPI('auth', 'logout') # Logout of the session self.sessionKey = None # Set the session key to None def getActivationKeys(self, keyList = None): """ method to get a dictionary of Activation Key dictionary objects enhanced by adding configuration channel information, if provisioning is entitled the Activation Key name is used as the dictionary key \taccept an optional list of keys to limit which Activation Keys are returned """ # Define the types of Activation Keys we are not interested in excludedKeyTypes = ('Kickstart re-activation key for', 'Reactivation key for') activationKeys = {} # Initialize an empty dictionary for discovered Activation Keys if not self.sessionKey: # If session key is not set self._connectSession() # Connect to ther RHN Satellite # Call the RHN Satellite API for a list of all Activation Keys allActivationKeys = self.client.activationkey.listActivationKeys(self.sessionKey) for activationKey in allActivationKeys: # For each Activation Key in the list description = activationKey['description'] # Get the description if not description.startswith(excludedKeyTypes): # If the description does not start with one of the excluded types if 'provisioning_entitled' in activationKey['entitlements']: # If the entitlements include the provisioning entitlements # Save the Configuration Deployment setting activationKey['config_deployment'] = self.client.activationkey.checkConfigDeployment(self.sessionKey, activationKey['key']) # and the list of Configuration Channels in the Activation Key dictionary configChannels = self.client.activationkey.listConfigChannels(self.sessionKey, activationKey['key']) lstConfigChannel = [] # Initialize a list for Configuration Channel Labels for channel in configChannels: # For each Configuration Channel lstConfigChannel.append(channel['label']) # Append its label to the list activationKey['config_channels'] = lstConfigChannel # Save the list of labels in the Activation Key dictionary keyID = activationKey.get('key').replace('%d-' % self.org, '') # Get the Activation Key ID, without the organization prefix activationKey['key'] = keyID # Save the organization independent ID in the Activtion Key dictionary if activationKey['server_group_ids']: # If the Activation Key has Server Group IDs # Use a list comprehension to translate them into names activationKey['server_group_names'] = list(self.sysGroups[id] for id in activationKey['server_group_ids']) else: # Otherwise ... activationKey['server_group_names'] = [] # Save an empty list of server group names del activationKey['package_names'] # Remove 'package_names' from the dictionary, since it is not used to import del activationKey['server_group_ids'] # Remove 'server_group_ids' from the dictionary, # since we have the implementation independent 'server_group_names' if keyList: # If a filter list was passed in if keyID in keyList: # If this Activation Key is in the filter list activationKeys[keyID] = activationKey # Add it to the dictionary of Activation Keys else: # If a filter list ws not passed in activationKeys[keyID] = activationKey # Always add this Activation Key to the dictionary of Activation Keys return activationKeys # Return the dictionary of Activation Keys def setActivationKeys(self, newActivationKeys): """ method to create/update Activation Keys \trequires a dictionary of "enhanced" Activation Key dictionary objects see getActivationKeys for details of the enhancements """ # # Helper Functions internal to this method # def _execAPI(myMethod, myArgs): """ function to use the current session to execute an API call to the activationkey namespace. Requires a string with the method name and a tuple with the arguments. The function the existing session key and assumes the first argument is an Activation Key without the Organization prefix. """ myKey = '%s-%s' % (self.org, myArgs[0]) # Prefix key argument with organization # If myMethod is for ConfigChannels, argument is a list if myMethod in ['addConfigChannels', 'removeConfigChannels', 'setConfigChannels']: apiArgs = (self.sessionKey, [myKey]) + myArgs[1:] # Prepend Session Key and list of current Activation Key elif myMethod in ['addServerGroups', 'removeServerGroups']: # If myMethod is for ServerGroups, translate names to local IDs # Use list comprehension to translate group names to local group IDs serverGroupIDs = list(k for k in self.sysGroups.keys() if self.sysGroups[k] in myArgs[1]) apiArgs = (self.sessionKey, myKey) + (serverGroupIDs,) else: # If myMethod is not for ConfigChannels apiArgs = (self.sessionKey, myKey) + myArgs[1:] # Prepend Session Key and current Activation Key myAPI = getattr(self.client.activationkey, myMethod) # Assign the XML-RPC Method to myAPI try: # Try to ... status = myAPI(*apiArgs) # Call the API with the arguments if self.options.verbose: # If verbose command line option set # Print a header print '\tCalling %s for Activation Key %s:' % (myMethod, myKey) for arg in myArgs[1:]: # and for each argument if type(arg) == type(list()): # If the argument is a list for item in arg: # For each item in the list print '\t\t%s' % item # Print the item else: # If it is not a list print '\t\t%s' % arg # Print the argument except xmlrpclib.Fault, err: # If an xmlrpclib.Fault occurs print err.faultCode # Print the faultCode print err.faultString # and the faultString except: # If any other exception occurs print 'An Exception occured in %s' % myMethod # Print where it occured print traceback.print_exc(sys.exc_info()) # and the fault information # # Main Logic for Method # # Get the list of current Activation Key(s) with the same # ID(s) as the Activation Key(s) to import currentActivationKeys = self.getActivationKeys(newActivationKeys.keys()) for newActivationKey in newActivationKeys.values(): # For each new Activation Key if newActivationKey['key'] not in currentActivationKeys.keys(): # If the ID is not one of the current IDs print 'Importing Activation Key %d-%s ...' % (self.org, newActivationKey['key']) if newActivationKey['usage_limit'] == 0: # If the new Activation Key has a usage_limit of "0" newKey = self.client.activationkey.create(self.sessionKey, # Call the function without a usage limit newActivationKey['key'], newActivationKey['description'], newActivationKey['base_channel_label'], newActivationKey['entitlements'], newActivationKey['universal_default']) else: # If the new Actiavation Key has a non zero usage_limit newKey = self.client.activationkey.create(self.sessionKey, # Call the function with a usage limit newActivationKey['key'], newActivationKey['description'], newActivationKey['base_channel_label'], newActivationKey['usage_limit'], newActivationKey['entitlements'], newActivationKey['universal_default']) if 'provisioning_entitled' in newActivationKey['entitlements']: # If the new Activation Key has a provisioning entitlement if newActivationKey['config_deployment']: # If config_deployment is True, enable it in the new Activation Key self.client.activationkey.enableConfigDeployment(self.sessionKey, newKey) else: # If config_deployment is False, disable it in the new Activation Key self.client.activationkey.disableConfigDeployment(self.sessionKey, newKey) # Call Helper function to add Child Channels, set Confiuration Channels, # add Packages and add Kickstart Profiles _execAPI('addChildChannels', (newActivationKey['key'], newActivationKey['child_channel_labels'])) _execAPI('setConfigChannels', (newActivationKey['key'], newActivationKey['config_channels'])) _execAPI('addPackages', (newActivationKey['key'], newActivationKey['packages'])) _execAPI('addServerGroups', (newActivationKey['key'], newActivationKey['server_group_names'])) elif self.options.force: # If this Activation Key exists, and the force command line option was specified # then this is an update print 'Updating Activation Key %d-%s ...' % (self.org, newActivationKey['key']) # Build a new Activation Key details parameter dictionary newDetails = {'description' : newActivationKey['description'], 'base_channel_label' : newActivationKey['base_channel_label'], 'usage_limit' : newActivationKey['usage_limit'], 'universal_default' : newActivationKey['universal_default'], 'disabled' : newActivationKey['disabled']} if newDetails['usage_limit'] == 0: # If usage_limit is 0 newDetails['unlimited_usage_limit'] = True # Set unlimited_usage_limit to True in the new details try: # Try to update the details in this Activation Key status = self.client.activationkey.setDetails(self.sessionKey, '%s-%s' % (self.org, newActivationKey['key']), newDetails) if self.options.verbose: # If verbose command line option specified, print updating details message print '\tUpdated details for %d-%s' % (self.org, newActivationKey['key']) except xmlrpclib.Fault, err: # If a xmlrpclib.Fault occurs print err.faultCode # Print the faultCode print err.faultString # and the faultString except: # If any other exception occurs, print 'Unexpected error:', sys.exc_info()[0] # Print the exception info # Call Helper functions to remove old Child Channels, Packages and Kickstart Profiles _execAPI('removeChildChannels', (newActivationKey['key'], currentActivationKeys[newActivationKey['key']]['child_channel_labels'])) _execAPI('removePackages', (newActivationKey['key'], currentActivationKeys[newActivationKey['key']]['packages'])) _execAPI('removeServerGroups', (newActivationKey['key'], currentActivationKeys[newActivationKey['key']]['server_group_names'])) # Call Helper functions to add Child Channels, set Confiuration Channels, # add Packages and add Kickstart Profiles _execAPI('addChildChannels', (newActivationKey['key'], newActivationKey['child_channel_labels'])) _execAPI('setConfigChannels', (newActivationKey['key'], newActivationKey['config_channels'])) _execAPI('addPackages', (newActivationKey['key'], newActivationKey['packages'])) _execAPI('addServerGroups', (newActivationKey['key'], newActivationKey['server_group_names'])) else: # If this Activation Key exists, and the force command line option was not specified # Print a message print 'Activation Key exists %s and --force was not specified' % newActivationKey['key'] def getSystemGroups(self, groupList = None): """ method to get a dictionary of "enhanced" Kickstart Profile dictionary objects. the Kickstart Profile name is used as the dictionary key. the objects are enhanced by adding admins and default users, and removing id, org_id ans system_count \taccept an optional list of Groups to limit which Kickstart Profiles are returned """ # # Helper Functions internal to this method # def _getDetails(group): """ Remove everything, but name, and description, then add a list of administrator logins and a list of default users. """ name = group['name'] # Get the group name groupKeys = ['name', 'description', 'admins', 'default'] # List of keys to extract # Get the list of Administrators group['admins'] = list(i for i in self.userInfo.keys() if name in self.userInfo[i]['assigned_system_groups'] and 'org_admin' not in self.userInfo[i]['roles']) # Get the list of Default Users group['default'] = list(i for i in self.userInfo.keys() if name in self.userInfo[i]['default_system_groups']) return extractDict(group, groupKeys) # Return the modified group # # Main Logic for Method # systemGroups = {} # Create an empty dictionary for Kickstart Profiles if not self.sessionKey: # If session key is not set self._connectSession() # Connect to ther RHN Satellite # Call the RHN Satellite API for a list of all Kickstart Profiles allSystemGroups = self.client.systemgroup.listAllGroups(self.sessionKey) for systemGroup in allSystemGroups: # For each Kickstart Profile returned name = systemGroup['name'] # Get the name if groupList: # If a filter list was passed if name in groupList: # If the current group is in the filter list systemGroups[name] = _getDetails(systemGroup) # Add the group (with details) to the dictionary of groups else: # If no filter list was passed systemGroups[name] = _getDetails(systemGroup) # Add the group (with details) to the dictionary of groups return systemGroups # Return the dictionary of Kickstart Profiles def setSystemGroups(self, newSystemGroups): """ method to create/update Kickstart Profiles \trequires a didctionary of "enhanced" Kickstart Profile dictionary objects see getSystemGroups for details of the enhancements """ # # Helper Functions internal to this method # def _execAPI(myNamespace, myMethod, myArgs): """ function to use the current session to execute an API call to the activationkey namespace. Requires a string with the namespace (either 'user', or 'systemgroup'), a string with the method name and a tuple with the arguments. The function uses the existing session key. """ apiArgs = (self.sessionKey, ) + myArgs # Prepend Session Key to the arguments if myNamespace == 'systemgroup': # If the Namespace is 'systemgroup' myAPI = getattr(self.client.systemgroup, myMethod) # Assign the XML-RPC Method to myAPI elif myNamespace == 'user': # If the Namespace is 'user' myAPI = getattr(self.client.user, myMethod) # Assign the XML-RPC Method to myAPI else: # If the Namespace is not supported print 'Unsupported Namespace %s' % myNamespace # Print an error message, and return None # Return None try: # Try to ... status = myAPI(*apiArgs) # Call the API with the arguments if self.options.verbose: # If verbose command line option set # Print the API call with arguments print '\tCalling %s.%s(sessionKey, %s)' % (myNamespace, myMethod, str(myArgs)) except xmlrpclib.Fault, err: # If an xmlrpclib.Fault occurs print err.faultCode # Print the faultCode print err.faultString # and the faultString except: # If any other exception occurs print 'An Exception occured in %s' % myMethod # Print where it occured print traceback.print_exc(sys.exc_info()) # and the fault information # # Main Logic for Method currentSystemGroups = self.getSystemGroups(newSystemGroups.keys()) # Get the list of current Kickstart Profile(s) with the same # ID(s) as the Kickstart Profile(s) to import for name, newSystemGroup in newSystemGroups.items(): # For each new Kickstart Profile if name not in currentSystemGroups.keys(): # If the name in not one of the current names print 'Importing Kickstart Profile %s ...' % name # Print a header # Create the new Kickstart Profile newGroup = _execAPI('systemgroup', # Call API for 'systemgroup' namespace 'create', # And the 'create' method, with the following arguments (name, newSystemGroup['description'])) # Get a list of Administrators for the Kickstart Profile with user IDs in this RHN Satellite adminsToAdd = list(l for l in newSystemGroup['admins'] if l in self.userInfo.keys()) if len(adminsToAdd) > 0: # If this Kickstart Profile has Administrators to be added status = _execAPI('systemgroup', # Call API for 'systemgroup' namespace 'addOrRemoveAdmins', # And the 'addOrRemoveAdmins' method, with the following arguments (name, adminsToAdd, 1)) # Get a list of Group Default Users with user IDs in this RHN Satellite defaultUsersToAdd = list(l for l in newSystemGroup['default'] if l in self.userInfo.keys()) if len(defaultUsersToAdd) > 0: # If this Kickstart Profile has Default Users to be added for defaultUser in defaultUsersToAdd: # For each Default User in the list # Add this Kickstart Profile to the users list of Kickstart Profiles status = _execAPI('user', # Call API for 'user' namespace 'addDefaultSystemGroup', # And the 'addDefaultSystemGroup' method, for the following arguments (defaultUser, name)) elif self.options.force: # If this Kickstart Profile exists, and the force command line option was specified print 'Updating Kickstart Profile %s ...' % name # Print a header # Update the Kickstart Profile theGroup = _execAPI('systemgroup', # Call API for 'systemgroup' namespace 'update', # And the 'update' method, with the following arguments (name, newSystemGroup['description'])) # Remove the current Admins from this Kickstart Profile status = _execAPI('systemgroup', # Call API for 'systemgroup' namespace 'addOrRemoveAdmins', # And the 'addOrRemoveAdmins' method, with the following arguments (name, currentSystemGroups[name]['admins'], 0)) for defaultUser in currentSystemGroups[name]['default']: # For each Default User in this Kickstart Profile # Remove the user from this Kickstart Profile status = _execAPI('user', # Call API for 'user' namespace 'removeDefaultSystemGroup', # And the 'addDefaultSystemGroup' method, for the following arguments (defaultUser, name)) # Get a list of Administrators for the Kickstart Profile with user IDs in this RHN Satellite adminsToAdd = list(l for l in newSystemGroup['admins'] if l in self.userInfo.keys()) if len(adminsToAdd) > 0: # If this Kickstart Profile has Administrators to be added status = _execAPI('systemgroup', # Call API for 'systemgroup' namespace 'addOrRemoveAdmins', # And the 'addOrRemoveAdmins' method, with the following arguments (name, adminsToAdd, 1)) # Get a list of Group Default Users with user IDs in this RHN Satellite defaultUsersToAdd = list(l for l in newSystemGroup['default'] if l in self.userInfo.keys()) if len(defaultUsersToAdd) > 0: # If this Kickstart Profile has Default Users to be added for defaultUser in defaultUsersToAdd: # For each Default User in the list # Add this Kickstart Profile to the users list of Kickstart Profiles status = _execAPI('user', # Call API for 'user' namespace 'addDefaultSystemGroup', # And the 'addDefaultSystemGroup' method, for the following arguments (defaultUser, name)) else: # If this Kickstart Profile exists, and the force command line option was not specified # Print a message print 'Kickstart Profile %s exists and --force was not specified' % name def getKickstartSnippets(self, snippetList = None): """ method to get a dictionary of Kickstart Snippets objects the snippet name is used as the dictionary key \taccept an optional list of Snippets to limit which Kickstart Snippits are returned """ # # Helper Functions internal to this method # def _getDetails(snippet): """ Strip out internal 'fragment' and 'file' since they are not needed for import. """ del snippet['fragment'] del snippet['file'] # # Main Logic for Method # kickstartSnippets = {} # Create an empty dictionary for Kickstart Snippets if not self.sessionKey: # If session key is not set self._connectSession() # Connect to ther RHN Satellite # Call the RHN Satellite API for a list of all Kickstart Snippets allKickstartSnippets = self.client.kickstart.snippet.listCustom(self.sessionKey) for kickstartSnippet in allKickstartSnippets: # For each Kickstart Snippet returned if snippetList: # If a filter list was passed if kickstartSnippet['name'] in snippetList: # If the current snippet is in the filter list _getDetails(kickstartSnippet) # Remove unneeded info # Add the snippet to the dictionary of snippets kickstartSnippets[kickstartSnippet['name']] = kickstartSnippet else: # If no filter list was passed _getDetails(kickstartSnippet) # Remove unneeded info kickstartSnippets[kickstartSnippet['name']] = kickstartSnippet # Add the snippet to the dictionary of snippets return kickstartSnippets # Return the dictionary of Kickstart Snippets def setKickstartSnippets(self, newKickstartSnippets): """ method to create/update Kickstart Snippets \trequires a dictionary of Kickstart Snippet dictionary objects """ # Get the list of current Kickstart Snippets with the same # names as the Kickstart Snippet(s) to import currentKickstartSnippets = self.getKickstartSnippets(newKickstartSnippets.keys()) for name, newKickstartSnippet in newKickstartSnippets.items(): # For each new Kickstart Snippet try: if name not in currentKickstartSnippets.keys(): # If the name is not one of the current snippet names print 'Importing Kickstart Snippet %s ...' % name # Print an import header # Create a new snippet newSnippet = self.client.kickstart.snippet.createOrUpdate(self.sessionKey, name, newKickstartSnippet['contents']) elif self.options.force: # If the name is one of the current snippet names, and --force was specified print 'Updating Kickstart Snippet %s ...' % name # Print an update header # Update the existinf snippet newSnippet = self.client.kickstart.snippet.createOrUpdate(self.sessionKey, name, newKickstartSnippet['contents']) else: # If the name is one of the current snippet names, and --force was not specified # Print a message print 'Kickstart Snippet %s exists and --force was not specified' % name except xmlrpclib.Fault, err: # If an xmlrpclib.Fault occurs print err.faultCode # Print the faultCode print err.faultString # and the faultString except: # If any other exception occurs print 'An Exception occured in importing a snippet' # Print where it occured print traceback.print_exc(sys.exc_info()) # and the fault information def getKickstartKeys(self, keyList = None): """ method to get a dictionary of GPG/SSL Key dictionary objects the key description is used as the dictionary key \taccepts an optional list of keys to limit which GPG/SSL Keys are returned """ # # Helper Functions internal to this method # def _getDetails(myKey): """ get details for a Key and add content to key dictionary """ # Get Details for this key myDetails = self.client.kickstart.keys.getDetails(self.sessionKey, myKey['description']) myKey['content'] = myDetails['content'] # Save the contnets from details to a new item # # Main Logic for Method # keys = {} # Create an empty Dictionary for GPG/SSL Keys if not self.sessionKey: # If session key is not set self._connectSession() # Connect to ther RHN Satellite allKeys = self.client.kickstart.keys.listAllKeys(self.sessionKey) # Get all GPG/SSL Keys for key in allKeys: # For each key found if keyList: # If a filter list was passed if key['description'] in keyList: # If this key is in the filter list _getDetails(key) # Add details to this key keys[key['description']] = key # Add to dictionary using 'description' as a key else: # If a filter list was not passed _getDetails(key) # Add details to this key keys[key['description']] = key # Add to dictionary using 'description' as a key return keys # Return dictionary of GPG/SSL Keys def setKickstartKeys(self, newKeys): """ method to create/update GPG/SSL Keys \trequires a dictionary of GPG/SSL Key dictionary objects """ currentKeys = self.getKickstartKeys(newKeys.keys()) # Get existing keys filtered by list of new keys for description, key in newKeys.items(): # For each new GPG/SSL Key try: # Try ... if description not in currentKeys.keys(): # If 'description' not in current GPG/SSL Keys # Print a header message print 'Importing Kickstart %s Key %s ...' % (key['type'], description) status = self.client.kickstart.keys.create(self.sessionKey, # And create the GPG/SSL Key description, key['type'], key['content']) elif self.options.force: # If 'description' in current GPG/SSL Keys, and --force option specified # Print a header message print 'Updating Kickstart %s Key %s ...' % (key['type'], description) status = self.client.kickstart.keys.update(self.sessionKey, # And update the GPG/SSL Key description, key['type'], key['content']) else: # If 'description' in current GPG/SSL Keus, and --force option was not specified # Print a message print 'Kickstart %s Key %s exists and --force was not specified' % (key['type'], description) except xmlrpclib.Fault, err: # If an xmlrpclib.Fault occurs print err.faultCode # Print the faultCode print err.faultString # and the faultString except: # If any other exception occurs # Print where it occured print 'An Exception occured in importing a Kickstart GPG/SSL Key' print traceback.print_exc(sys.exc_info()) # and the fault information def getConfigChannels(self, channelList = None): """ method to get a dictionary of Configuration Channel dictionary objects the config channel label is used as the dictionary key \taccepts an optional list of channel labels to limit which Config Channels are returned Note: Binary file contents cannot be exported until RHN Satellite incorporates Spacewalk 1.7 """ # # Helper Functions internal to this method # def _getDetails(myChannel): """ get details for a files associated with a channel """ # Create a list of keys to include in Configuration Channel dictionary objects channelKeys = ['label', 'name', 'description', 'file', 'directory', 'symlink'] # Get a list of paths in this Config Channel myPaths = list(d['path'] for d in self.client.configchannel.listFiles(self.sessionKey, myChannel['label'])) # Get detailed information for each path myPathInfo = self.client.configchannel.lookupFileInfo(self.sessionKey, myChannel['label'], myPaths) for myDict in myPathInfo: # For each dictionary of path information in this config channel if 'permissions_mode' in myDict: # If the key 'permissions_mode' exists myDict['permissions'] = myDict.pop('permissions_mode') # Rename it to 'permissions', to be compatible with create or update calls # List of lookupFileInfo keys needed for createOrUpdate calls fileKeys = ['contents', 'owner', 'group', 'permissions', 'selinux_ctx', 'macro-start-delimiter', 'macro-end-delimiter', 'binary'] linkKeys = ['target_path', 'selinux_ctx'] # Split out files, directories, and symbolic links myFiles = dict((d['path'], extractDict(d, fileKeys)) for d in myPathInfo if d['type'] == 'file') myDirectories = dict((d['path'], extractDict(d, fileKeys)) for d in myPathInfo if d['type'] =='directory') mySymlinks = dict((d['path'], extractDict(d, linkKeys)) for d in myPathInfo if d['type'] == 'symlink') myChannel['file'] = myFiles # Assign the dictionary of file path objects to the 'file' key myChannel['directory'] = myDirectories # Assign the dictionary of directory path objects to the 'directory' key myChannel['symlink'] = mySymlinks # Assign the dictionary of symlink path objects to the 'symlink' key return extractDict(myChannel, channelKeys) # Return the dictionary of Configuration Channel information with all unneeded keys removed # # Main Logic for Method # channels = {} # Create an empty dictionary for Config Channels if not self.sessionKey: # If session key is not set self._connectSession() # Connect to ther RHN Satellite allChannels = self.client.configchannel.listGlobals(self.sessionKey) # Get a list of all Configuration Channels for channel in allChannels: # For each Config Channel returned if channelList: # If a filter list was passed in if channel['label'] in channelList: # If this Config Channel is in the filter list channels[channel['label']] = _getDetails(channel) # Add (with details) to dictionary using 'label' as a key else: # If a filter list was not passed in channels[channel['label']] = _getDetails(channel) # Add (with details) to dictionary using 'label' as a key return channels # Return dictionary of Configuration Channels def setConfigChannels(self, newConfigChannels): """ method to create/update Configuration Channels \trequires a dictionary of Configuration Channel dictionary objects """ # # Helper Functions internal to this method # def _addOrUpdateDetails(myConfigChannel): """ function to add or update paths (files, directories or symlinks) in a Configuration Channel """ for path, pathInfo in myConfigChannel['file'].items(): # For each path, pathInfo pair of file information if 'binary' in pathInfo: # If 'binary' is a key in pathInfo del pathInfo['binary'] # Remove it from the dictionary # Create or Update this file path on this Configuration Channel pathObj = self.client.configchannel.createOrUpdatePath(self.sessionKey, myConfigChannel['label'], path, False, pathInfo) for path, pathInfo in myConfigChannel['directory'].items(): # For each path, pathInfo pair of directory information # Create or Update this directory path on this Configuration Channel pathObj = self.client.configchannel.createOrUpdatePath(self.sessionKey, myConfigChannel['label'], path, True, pathInfo) for path, pathInfo in myConfigChannel['symlink'].items(): # For each path, pathInfo pair of symlink information # Create or Update this symlink path on this Configuration Channel pathObj = self.client.configchannel.createOrUpdateSymlink(self.sessionKey, myConfigChannel['label'], path, pathInfo) # # Main Logic for Method # # Get current Config Channels, filtered by list of new Config Channels currentConfigChannels = self.getConfigChannels(newConfigChannels.keys()) for label, configChannel in newConfigChannels.items(): # For each new Config Channel try: # Try ... if label not in currentConfigChannels.keys(): # If 'label' not in current Config Channels print 'Importing ConfigChannel %s ...' % label # Print a create message # And create the Config Channel status = self.client.configchannel.create(self.sessionKey, label, configChannel['name'], configChannel['description']) _addOrUpdateDetails(configChannel) # Update the files, directories and symlinks for this Config Channel elif self.options.force: # If 'label' is current Config Channel, and --force option was specified print 'Updating ConfigChannel %s ...' % label # Print an update message # And update the Config Channel status = self.client.configchannel.update(self.sessionKey, label, configChannel['name'], configChannel['description']) _addOrUpdateDetails(configChannel) # Update the files, directories and symlinks for this Config Channel else: # If 'label' is current Config Channel, and --force option was not specified print 'ConfigChannel %s exists and --force was not specified' % label except xmlrpclib.Fault, err: # If an xmlrpclib.Fault occurs print err.faultCode # Print the faultCode print err.faultString # and the faultString except: # If any other exception occurs # Print where it occured print 'An Exception occured in importing a ConfigChannel' print traceback.print_exc(sys.exc_info()) # and the fault information def getKickstartProfiles(self, profileList = None): """ method to get a dictionary of Kckstart Profile dictionary objects the kickstart profile label is used as the dictionary key \taccepts an optional list of profile labels to limit which Kickstart Profiles are returned """ execAPI = self._execAPI # Save a pointer to the class method to call the rhnAPI # # Helper Functions internal to this method # def _getDetails(myProfile): """ get details for a files associated with a channel where a detail is a list, sort so thea compares will always match """ # List of returned script keys to keep scriptKeys = ['contents', 'script_type', 'interpreter', 'chroot', 'template'] myLabel = myProfile['label'] # Get the label for this Kickstart Profile # Get a list of Advanced Option dictionaries for this Kickstart Profile myProfile['advanced_options'] = sorted(execAPI('kickstart.profile', 'getAdvancedOptions', myLabel), key = lambda x:x['name']) # Get a list of Custom Option strings for this Kickstart Profile myProfile['custom_options'] = sorted(i['arguments'] for i in execAPI('kickstart.profile', 'getCustomOptions', myLabel)) # Get a list of Child Channels for this Kickstart Profile myProfile['child_channels'] = sorted(execAPI('kickstart.profile', 'getChildChannels', myLabel)) # Get a dictionary of Variables for this Kickstart Profile myProfile['variables'] = execAPI('kickstart.profile', 'getVariables', myLabel) myProfile['virtualization_type'] = 'none' # We cannot get the virtualization Type, so assume none myProfile['download_url'] = 'default' # Assume the default URL when initially creating the kickstart # Get a list of scripts for this Kickstart Profile myProfile['scripts'] = sorted(list(extractDict(script, scriptKeys) for script in execAPI('kickstart.profile', 'listScripts', myLabel)), key = lambda x:(x['script_type'], x['contents'])) # Get a list of Activation Keys, with their Organization prefix removed for this Kickstart Profile myProfile['activation_keys'] = sorted(list(keys['key'].replace('%d-' % self.org, '') for keys in execAPI('kickstart.profile.keys', 'getActivationKeys', myLabel))) # Get the Configuration Management setting for this Kickstart Profile myProfile['config_management'] = execAPI('kickstart.profile.system', 'checkConfigManagement', myLabel) # Get the Remote Commands setting for this Kickstart Profile myProfile['remote_commands'] = execAPI('kickstart.profile.system', 'checkRemoteCommands', myLabel) # Get a list of GPG/SSL Keys for this Kickstart Profile myProfile['keys'] = sorted(list(keys['description'] for keys in execAPI('kickstart.profile.system', 'listKeys', myLabel))) # Get a list of File Preservations for this Kickstart Profile myProfile['file_preservations'] = sorted(list(keys['name'] for keys in execAPI('kickstart.profile.system', 'listFilePreservations', myLabel))) # Get the Partitioining Scheme for this Kickstart Profile (do not sort) myProfile['partitioning_scheme'] = execAPI('kickstart.profile.system', 'getPartitioningScheme', myLabel) # Get the Registration Type for this Kickstart Profile myProfile['registration_type'] = execAPI('kickstart.profile.system', 'getRegistrationType', myLabel) # Get a list of Software Packages for this Kickstart Profile (do not sort) myProfile['package_list'] = execAPI('kickstart.profile.software', 'getSoftwareList', myLabel) return myProfile # Return the Kickstart Profile with details added # # Main Logic for Method # profiles = {} # Create an empty dictionary for Kickstart Profiles profileKeys = ['label', 'tree_label', 'active'] # List of returned keys to keep for each Kickstart Profile if not self.sessionKey: # If session key is not set self._connectSession() # Connect to ther RHN Satellite # Get a list of all Kickstart Profiles allProfiles = list(extractDict(profile, profileKeys) for profile in execAPI('kickstart', 'listKickstarts')) for profile in allProfiles: # For each Kickstart Profile returned if profileList: # If a filter list was passed in if profile['label'] in profileList: # If this Kickstart Profile is in the filter list profiles[profile['label']] = _getDetails(profile) # Add (with details) to dictionary using 'label' as a key else: # If a filter list was not passed in profiles[profile['label']] = _getDetails(profile) # Add (with details) to dictionary using 'label' as a key return profiles # Return dictionary of Kickstart Profiles def setKickstartProfiles(self, newKickstartProfiles): """ method to create/update Kickstart Profiles \trequires a dictionary of "enhanced" Kickstart Profile dictionary objects see getKickstartProfiles for details of the enhancements """ execAPI = self._execAPI # Save a pointer to the class method to call the rhnAPI # # Helper Functions internal to this method # def updateDetails(ksProf): """ Function to set or replace Kickstart Profile details """ status = execAPI('kickstart', # Set teh Kickstart Profiles Active/Disabled state 'disableProfile', ksProf['label'], not ksProf['active']) status = execAPI('kickstart.profile', # Set the Advanced Options for this Kickstart Profile 'setAdvancedOptions', ksProf['label'], ksProf['advanced_options']) status = execAPI('kickstart.profile', # Set the Child Channels for this Kickstart Profile 'setChildChannels', ksProf['label'], ksProf['child_channels']) status = execAPI('kickstart.profile', # Set the Custom Options for this Kickstart Profile 'setCustomOptions', ksProf['label'], ksProf['custom_options']) status = execAPI('kickstart.profile', # Set the Variables for this Kickstart Profile 'setVariables', ksProf['label'], ksProf['variables']) # Get a list of the current Script IDs oldScriptIDs = list(script['id'] for script in execAPI('kickstart.profile', 'listScripts', ksProf['label'])) for oldScriptID in oldScriptIDs: # For each current Script ID execAPI('kickstart.profile', # Remove the script 'removeScript', ksProf['label'], oldScriptID) for script in ksProf['scripts']: # For each new script execAPI('kickstart.profile', # Add the new script 'addScript', ksProf['label'], script['contents'], script['interpreter'], script['script_type'], script['chroot'], script['template']) # Get a list of current Activation Keys oldActivationKeys = list(keys['key'] for keys in execAPI('kickstart.profile.keys', 'getActivationKeys', ksProf['label'])) for activationKey in oldActivationKeys: # For each Current Activation Key execAPI('kickstart.profile.keys', # Remove the current Activation Key 'removeActivationKey', ksProf['label'], activationKey) for activationKey in ksProf['activation_keys']: # For each new Activation Key execAPI('kickstart.profile.keys', # Add teh Activation Key 'addActivationKey', ksProf['label'], '%s-%s' % (self.org, activationKey)) if ksProf['config_management']: # If 'config_management' is True execAPI('kickstart.profile.system', # Enable Config Management 'enableConfigManagement', ksProf['label']) else: # Otherwise, execAPI('kickstart.profile.system', # Disable Config Management 'disableConfigManagement', ksProf['label']) if ksProf['remote_commands']: # If 'remote_management' is True execAPI('kickstart.profile.system', # Enable Remote Management 'enableRemoteCommands', ksProf['label']) else: # Otherwise, execAPI('kickstart.profile.system', # Disable Remote Management 'disableRemoteCommands', ksProf['label']) # Get a list of current GPG/SSL Keys oldKeys = list(keys['description'] for keys in execAPI('kickstart.profile.system', 'listKeys', ksProf['label'])) execAPI('kickstart.profile.system', # Remove the current GPG/SSL Keys 'removeKeys', ksProf['label'], oldKeys) execAPI('kickstart.profile.system', # Add the new GPG/SSL Keys 'addKeys', ksProf['label'], ksProf['keys']) # Get a list of the current File Preservations oldFilePreservations = list(keys['name'] for keys in execAPI('kickstart.profile.system', 'listFilePreservations', ksProf['label'])) execAPI('kickstart.profile.system', # Remove the current File Preservations 'removeFilePreservations', ksProf['label'], oldFilePreservations) execAPI('kickstart.profile.system', # Add the new File Preservations 'addFilePreservations', ksProf['label'], ksProf['file_preservations']) execAPI('kickstart.profile.system', # Set the new Partitioning Scheme 'setPartitioningScheme', ksProf['label'], ksProf['partitioning_scheme']) execAPI('kickstart.profile.system', # Set the new Registration Type 'setRegistrationType', ksProf['label'], ksProf['registration_type']) execAPI('kickstart.profile.software', # Set the new Software List 'setSoftwareList', ksProf['label'], ksProf['package_list']) # # Main Logic for Method # # Get the list of current Kickstart Profile(s) with the same # ID(s) as the Kickstart Profile(s) to import currentKickstartProfiles = self.getKickstartProfiles(newKickstartProfiles.keys()) for label, kickstartProfile in newKickstartProfiles.items(): # For each new Kickstart Profile if label not in currentKickstartProfiles.keys(): # If 'label' not in current Kickstart Profiles print 'Importing Kickstart Profile %s ...' % label # Print a create message status = execAPI('kickstart', # And create the Kickstart Profile 'createProfileWithCustomUrl', kickstartProfile['label'], kickstartProfile['virtualization_type'], kickstartProfile['tree_label'], kickstartProfile['download_url'], list(d['arguments'] for d in kickstartProfile['advanced_options'] if d['name'] == 'rootpw')[0]) updateDetails(kickstartProfile) # Update supporting details for this Kickstart Profile elif self.options.force: # If 'label' is current Kickstart Profile, and --force was specified print 'Updating Kickstart Profile %s ...' % label # Print an update message status = execAPI('kickstart.profile', # Update the Kickstart Tree (nothing else can be changed anyway) 'setKickstartTree', label, kickstartProfile['tree_label']) updateDetails(kickstartProfile) # Update supporting details for this Kickstart Profile else: # If 'label' is current Kickstart Profile, and --force was not specified print 'Kickstart Profile %s exists and --force was not specified' % label def getChannels(self, channelList = None): """ method to get a dictionary of Kckstart Channel dictionary objects the kickstart channel label is used as the dictionary key \taccepts an optional list of channel labels to limit which Software Channels are returned """ execAPI = self._execAPI # Save a pointer to the class method to call the rhnAPI # # Helper Functions internal to this method # def _getDetails(myLabel): """ get details for a files associated with a channel """ channelMap = ['checksum_label', 'description', 'maintainer_name', 'maintainer_email', 'maintainer_phone', 'gpg_key_url', 'gpg_key_id', 'gpg_key_fp'] channelKeys = ['label', 'name', 'summary', 'arch_name', 'parent_channel_label', 'channel_map', 'channel_repos'] myChannel = execAPI('channel.software', 'getDetails', myLabel) myChannel['channel_map'] = extractDict(myChannel, channelMap) myChannel['channel_repos'] = execAPI('channel.software', 'listChannelRepos', myLabel) print myChannel['channel_repos'] return extractDict(myChannel, channelKeys) # Return the Software Channel with details added # # Main Logic for Method # channels = {} # Create an empty dictionary for Software Channels excludedChannels = ('rhel-', 'rhn-tools-rhel-') # List of returned keys to keep for each Software Channel if not self.sessionKey: # If session key is not set self._connectSession() # Connect to ther RHN Satellite # Get a list of all non-Red Hat Software Channel Labels allChannels = list(channel['label'] for channel in execAPI('channel', 'listAllChannels') if not channel['label'].startswith(excludedChannels)) for channel in allChannels: # For each Software Channel returned if channelList: # If a filter list was passed in if channel in channelList: # If this Software Channel is in the filter list channels[channel] = _getDetails(channel) # Add (with details) to dictionary using 'label' as a key else: # If a filter list was not passed in channels[channel] = _getDetails(channel) # Add (with details) to dictionary using 'label' as a key return channels # Return dictionary of Software Channels