-
Notifications
You must be signed in to change notification settings - Fork 1
/
subscriptionmanager.py
98 lines (79 loc) · 3.8 KB
/
subscriptionmanager.py
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
"""
Module for communcating with subscription-manager, part of virt-who
Copyright (C) 2011 Radek Novacek <[email protected]>
This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License
as published by the Free Software Foundation; either version 2
of the License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
"""
import os
import sys
import rhsm.connection as rhsm_connection
import rhsm.certificate as rhsm_certificate
import rhsm.config as rhsm_config
class SubscriptionManagerError(Exception):
def __init__(self, message):
self.message = message
def __str__(self):
return self.message
class SubscriptionManager:
""" Class for interacting subscription-manager. """
def __init__(self, logger, username=None, password=None):
self.logger = logger
self.cert_uuid = None
self.username = username
self.password = password
self.config = rhsm_config.initConfig(rhsm_config.DEFAULT_CONFIG_PATH)
self.readConfig()
# Consumer ID obtained from consumer certificate
self.cert_uuid = self.uuid()
def readConfig(self):
""" Parse rhsm.conf in order to obtain consumer
certificate and key paths. """
consumerCertDir = self.config.get("rhsm", "consumerCertDir")
cert = 'cert.pem'
key = 'key.pem'
self.cert_file = os.path.join(consumerCertDir, cert)
self.key_file = os.path.join(consumerCertDir, key)
if not os.access(self.cert_file, os.R_OK) or not os.access(self.key_file, os.R_OK):
raise SubscriptionManagerError("Unable to read certificate, system is not registered or you are not root")
def connect(self, Connection=rhsm_connection.UEPConnection):
""" Connect to the subscription-manager. """
self.logger.debug(self.config)
kwargs = {
'host': self.config.get('server', 'hostname'),
'ssl_port': int(self.config.get('server', 'port')),
'handler': self.config.get('server', 'prefix'),
'proxy_hostname': self.config.get('server', 'proxy_hostname'),
'proxy_port': self.config.get('server', 'proxy_port'),
'proxy_user': self.config.get('server', 'proxy_user'),
'proxy_password': self.config.get('server', 'proxy_password')
}
if self.username and self.password:
kwargs['username'] = self.username
kwargs['password'] = self.password
else:
kwargs['cert_file'] = self.cert_file
kwargs['key_file'] = self.key_file
self.connection = Connection(**kwargs)
def hypervisorCheckIn(self, owner, env, mapping, type=None):
""" Send hosts to guests mapping to subscription manager. """
self.logger.debug("Sending update in hosts-to-guests mapping: %s" % mapping)
# Send the mapping
return self.connection.hypervisorCheckIn(owner, env, mapping)
def uuid(self):
""" Read consumer certificate and get consumer UUID from it. """
if not self.cert_uuid:
try:
certificate = rhsm_certificate.create_from_file(self.cert_file)
self.cert_uuid = certificate.subject["CN"]
except Exception, e:
raise SubscriptionManagerError("Unable to open certificate %s (%s):" % (self.cert_file, str(e)))
return self.cert_uuid