/usr/sbin/spacewalk-channel is in rhn-client-tools 1.8.26-4.
This file is owned by root:root, with mode 0o755.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | #! /usr/bin/python
#
# Copyright (c) 2009--2012 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 socket
import sys
import urlparse
import xmlrpclib
from rhn import rpclib
from optparse import Option, OptionParser
import gettext
t = gettext.translation('rhn-client-tools', fallback=True)
_ = t.ugettext
_LIBPATH = "/usr/share/rhn"
# add to the path if need be
if _LIBPATH not in sys.path:
sys.path.append(_LIBPATH)
from up2date_client.rhnChannel import subscribeChannels, unsubscribeChannels, getChannels
from up2date_client import up2dateAuth, config, up2dateErrors, rhncli, rhnserver
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(rhncli.utf8_encode(msg) + "\n")
sys.exit(code)
# quick check to see if you are a super-user.
if os.getuid() != 0:
systemExit(8, _('ERROR: must be root to execute\n'))
def processCommandline():
"process the commandline, setting the OPTIONS object"
optionsTable = [
Option('-c', '--channel', action='append',
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='store',
help=_('your user name')),
Option('-p', '--password', action='store',
help=_('your password')),
]
optionParser = OptionParser(option_list=optionsTable)
global OPTIONS
OPTIONS, 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)"))
if not OPTIONS.user and not OPTIONS.list and not OPTIONS.base:
print _("Username: "),
OPTIONS.user = sys.stdin.readline().rstrip('\n')
if not OPTIONS.password and not OPTIONS.list and not OPTIONS.base:
OPTIONS.password = getpass.getpass()
def get_available_channels(user, password):
""" return list of available child channels """
cfg = config.initUp2dateConfig()
modified_servers = []
servers = config.getServerlURL()
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:
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:
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 gracefuly if channel is empty """
if not channel:
systemExit(4, _("ERROR: you have to specify at least one channel"))
def main():
if OPTIONS.add:
need_channel(OPTIONS.channel)
result = subscribeChannels(OPTIONS.channel, OPTIONS.user, OPTIONS.password)
if OPTIONS.verbose:
if result == 0:
print _("Channel(s): %s successfully added") % ', '.join(OPTIONS.channel)
else:
sys.stderr.write(rhncli.utf8_encode(_("Error during adding channel(s) %s") % ', '.join(OPTIONS.channel)))
if result != 0:
sys.exit(result)
elif OPTIONS.remove:
need_channel(OPTIONS.channel)
result = unsubscribeChannels(OPTIONS.channel, OPTIONS.user, OPTIONS.password)
if OPTIONS.verbose:
if result == 0:
print _("Channel(s): %s successfully removed") % ', '.join(OPTIONS.channel)
else:
sys.stderr.write(rhncli.utf8_encode(_("Error during removal of channel(s) %s") % ', '.join(OPTIONS.channel)))
if result != 0:
sys.exit(result)
elif OPTIONS.list:
try:
channels = map(lambda x: x['label'], 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?'))
channels.sort()
print '\n'.join(channels)
elif OPTIONS.base:
try:
for channel in getChannels().channels():
# Base channel has no parent
if not channel['parent_channel']:
print channel['label']
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?')
elif OPTIONS.available_channels:
channels = get_available_channels(OPTIONS.user, OPTIONS.password)
channels.sort()
print '\n'.join(channels)
else:
systemExit(3, _("ERROR: you may want to specify --add, --remove or --list"))
try:
sys.excepthook = rhncli.exceptionHandler
processCommandline()
main()
except KeyboardInterrupt:
systemExit(0, "\n" + _("User interrupted process."))
except up2dateErrors.RhnServerException, e:
# do not print traceback, it will scare people
systemExit(1, e)
|