Root Zanli
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
sbin
/
Filename :
spacewalk-channel
back
Copy
#!/usr/bin/python3 # # Copyright (c) 2009--2018 Red Hat, Inc. # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. import getpass import os import re import sys from rhn import rpclib from rhn.i18n import sstr from optparse import Option, OptionParser try: # python2 import urlparse import xmlrpclib except ImportError: # python3 import urllib.parse as urlparse import xmlrpc.client as xmlrpclib import gettext t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext from up2date_client.rhnChannel import subscribeChannels, unsubscribeChannels, getChannels from up2date_client import up2dateAuth, config, up2dateErrors, rhncli, rhnserver class Credentials(object): def __init__(self, username=None, password=None): if username is not None: self.user = username if password is not None: self.password = password def __getattr__(self, attr): if attr == 'user': tty = open("/dev/tty", "w") tty.write('Username: ') tty.close() setattr(self, 'user', sys.stdin.readline().rstrip('\n')) return self.user elif attr == 'password': # force user population _user = self.user setattr(self, 'password', getpass.getpass()) return self.password else: raise AttributeError(attr) def user_callback(self, _option, _opt_str, value, _parser): self.user = value def password_callback(self, _option, _opt_str, value, _parser): self.password = value VERBOSE = False def info(text): if VERBOSE: print(text) def systemExit(code, msgs=None): "Exit with a code and optional message(s). Saved a few lines of code." if msgs is not None: if type(msgs) not in [type([]), type(())]: msgs = (msgs, ) for msg in msgs: if hasattr(msg, 'value'): msg = msg.value sys.stderr.write(sstr(msg) + "\n") sys.exit(code) def processCommandline(): "process the command-line" credentials = Credentials() optionsTable = [ Option('-c', '--channel', action='append', dest='channels', help=_('name of channel you want to (un)subscribe')), Option('-a', '--add', action='store_true', help=_('subscribe to channel')), Option('-r', '--remove', action='store_true', help=_('unsubscribe from channel')), Option('-l', '--list', action='store_true', help=_('list channels')), Option('-b', '--base', action='store_true', help=_('show base channel of a system')), Option('-L', '--available-channels', action='store_true', help=_('list all available child channels')), Option('-v', '--verbose', action='store_true', help=_('verbose output')), Option('-u', '--user', action='callback', callback=credentials.user_callback, nargs=1, type='string', help=_('your user name')), Option('-p', '--password', action='callback', callback=credentials.password_callback, nargs=1, type='string', help=_('your password')), ] optionParser = OptionParser(option_list=optionsTable) opts, args = optionParser.parse_args() # we take no extra commandline arguments that are not linked to an option if args: systemExit(1, _("ERROR: these arguments make no sense in this context (try --help)")) # remove confusing stuff delattr(opts, 'user') delattr(opts, 'password') return opts, credentials def get_available_channels(user, password): """ return list of available child channels """ modified_servers = [] servers = config.getServerlURLFromMirror() for server in servers: scheme, netloc, path, query, fragment = urlparse.urlsplit(server) modified_servers.append(urlparse.urlunsplit((scheme, netloc, '/rpc/api', query, fragment))) client = rhnserver.RhnServer(serverOverride=modified_servers) try: key = client.auth.login(user, password) except xmlrpclib.Fault: exc = sys.exc_info()[1] systemExit(1, "Error during client authentication: %s" % exc.faultString) system_id = re.sub('^ID-', '', rpclib.xmlrpclib.loads(up2dateAuth.getSystemId())[0][0]['system_id']) result = [] try: channels = client.system.listChildChannels(key, int(system_id)) except xmlrpclib.Fault: exc = sys.exc_info()[1] systemExit(1, "Error when listing child channels: %s" % exc.faultString) for channel in channels: if 'LABEL' in channel: result.extend([channel['LABEL']]) else: result.extend([channel['label']]) return result def need_channel(channel): """ die gracefully if channel is empty """ if not channel: systemExit(4, _("ERROR: you have to specify at least one channel")) def no_channels(channels): """ for actions that does not require channel as a parameter """ if channels: systemExit(4, _("ERROR: this action does not require channel")) def add_channel(channels, credentials): need_channel(channels) result = subscribeChannels(channels, credentials.user, credentials.password) if result == 0: info(_("Channel(s): %s successfully added") % ', '.join(channels)) else: systemExit(result, _("Error during adding channel(s) %s") % ', '.join(channels)) def remove_channel(channels, credentials): need_channel(channels) result = unsubscribeChannels(channels, credentials.user, credentials.password) if result == 0: info(_("Channel(s): %s successfully removed") % ', '.join(channels)) else: systemExit(result, _("Error during removal of channel(s) %s") % ', '.join(channels)) def list_channels(only_base_channels=False): try: channels = getChannels().channels() except up2dateErrors.NoChannelsError: systemExit(1, _('This system is not associated with any channel.')) except up2dateErrors.NoSystemIdError: systemExit(1, _('Unable to locate SystemId file. Is this system registered?')) for channel in sorted(channels): if not (only_base_channels and channel['parent_channel']): print(channel['label']) def list_available_channels(credentials): channels = get_available_channels(credentials.user, credentials.password) channels.sort() print('\n'.join(channels)) def main(): options, credentials = processCommandline() if options.verbose: VERBOSE = True if options.add: add_channel(options.channels, credentials) elif options.remove: remove_channel(options.channels, credentials) elif options.list: no_channels(options.channels) list_channels() elif options.base: no_channels(options.channels) list_channels(only_base_channels=True) elif options.available_channels: no_channels(options.channels) list_available_channels(credentials) else: systemExit(3, _("ERROR: you may want to specify --add, --remove or --list")) if __name__ == '__main__': # quick check to see if you are a super-user. if os.getuid() != 0: systemExit(8, _('ERROR: must be root to execute\n')) try: sys.excepthook = rhncli.exceptionHandler main() except KeyboardInterrupt: systemExit(0, "\n" + _("User interrupted process.")) except up2dateErrors.RhnServerException: # do not print traceback, it will scare people systemExit(1, sys.exc_info()[1]) else: # If you need some code from here, separate it to some proper place... raise ImportError('This was never supposed to be used as a library')