Source Code for Module backend.server.rhnServer.satellite_cert

  1  # 
  2  # Copyright (c) 2008--2013 Red Hat, Inc. 
  3  # 
  4  # This software is licensed to you under the GNU General Public License, 
  5  # version 2 (GPLv2). There is NO WARRANTY for this software, express or 
  6  # implied, including the implied warranties of MERCHANTABILITY or FITNESS 
  7  # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 
  8  # along with this software; if not, see 
  9  # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. 
 10  # 
 11  # Red Hat trademarks are not licensed under GPLv2. No permission is 
 12  # granted to use or replicate Red Hat trademarks that are incorporated 
 13  # in this software or its documentation. 
 14  # 
 15  # 
 16  # This module exposes the SatelliteCert class, used for parsing a satellite 
 17  # certificate 
 18  # 
 19   
 20  import string 
 21  import sys 
 22  from xml.dom.minidom import parseString 
 23  from xml.sax import SAXParseException 
 24   
 25   
 26 -class ParseException(Exception): 
 27      pass 
 28   
 29   
 30  # Generic class to represent items (like channel families) 
 31 -class Item: 
 32      # pylint: disable=R0903 
 33   
 34      # Name to be displayed by repr() 
 35      pretty_name = None 
 36      # Attribute name in the parent class 
 37      attribute_name = None 
 38      # Mapping from XML to local storage 
 39      attributes = {} 
 40   
 41 -    def __init__(self, node=None): 
 42          if not node: 
 43              return 
 44          for attr_name, storage_name in self.attributes.items(): 
 45              attr = node.getAttribute(attr_name) 
 46              # Make sure we stringify the attribute - it may get out as unicode 
 47              setattr(self, storage_name, attr) 
 48   
 49 -    def __repr__(self): 
 50          return "<%s; %s>" % (self.pretty_name, 
 51                               string.join( 
 52                                   ['%s="%s"' % (x, getattr(self, x)) for x in self.attributes.values()], 
 53                                   ', ' 
 54                               )) 
 55   
 56   
 57 -class ChannelFamily(Item): 
 58      # pylint: disable=R0903 
 59   
 60      pretty_name = "channel family" 
 61      attribute_name = 'channel_families' 
 62      attributes = {'family': 'name', 'quantity': 'quantity', 'flex': 'flex'} 
 63   
 64   
 65 -class Slots: 
 66      _db_label = None 
 67      _slot_name = None 
 68   
 69 -    def __init__(self, quantity): 
 70          self.quantity = quantity 
 71   
 72 -    def get_quantity(self): 
 73          return self.quantity 
 74   
 75 -    def get_db_label(self): 
 76          """Returns the label of this type of slot in the database""" 
 77          return self._db_label 
 78   
 79 -    def get_slot_name(self): 
 80          """Returns the name of the slot, used by 
 81          rhn_entitlements.modify_org_service""" 
 82          return self._slot_name 
 83   
 84   
 85 -class ManagementSlots(Slots): 
 86      _db_label = 'enterprise_entitled' 
 87      _slot_name = 'enterprise' 
 88   
 89   
 90 -class ProvisioningSlots(Slots): 
 91      _db_label = 'provisioning_entitled' 
 92      _slot_name = 'provisioning' 
 93   
 94  # Slots for virt entitlements support 
 95   
 96   
 97 -class VirtualizationSlots(Slots): 
 98      _db_label = 'virtualization_host' 
 99      _slot_name = 'virtualization' 
100   
101   
102 -class VirtualizationPlatformSlots(Slots): 
103      _db_label = 'virtualization_host_platform' 
104      _slot_name = 'virtualization_platform' 
105   
106  # NonLinux slots are gone - misa 20050527 
107   
108   
109 -class MonitoringSlots(Slots): 
110      _db_label = 'monitoring_entitled' 
111      _slot_name = 'monitoring' 
112   
113   
114 -class SatelliteCert: 
115   
116      """Satellite certificate class 
117      Usage: 
118      c = SatelliteCert() 
119      c.load('<rhn-cert><rhn-cert-field name="owner">John Doe</rhn-cert-field></rhn-cert>') 
120      print c.owner 
121      """ 
122   
123      fields_scalar = ['product', 'owner', 'issued', 'expires', 'slots', 
124                       'provisioning-slots', 'nonlinux-slots', 
125                       'monitoring-slots', 'virtualization_host', 
126                       'virtualization_host_platform', 'satellite-version', 
127                       'generation', ] 
128      fields_list = {'channel-families': ChannelFamily} 
129   
130      # datesFormat_cert = '%a %b %d %H:%M:%S %Y' ## OLD CERT FORMAT 
131      datesFormat_cert = '%Y-%m-%d %H:%M:%S' 
132      datesFormat_db = '%Y-%m-%d %H:%M:%S' 
133   
134 -    def __init__(self): 
135          for f in self.fields_scalar: 
136              setattr(self, f, None) 
137          for f in self.fields_list.values(): 
138              setattr(self, f.attribute_name, []) 
139          self.signature = None 
140          self._slots = {} 
141          self._root = None 
142   
143 -    def load(self, s): 
144          try: 
145              self._load(s) 
146          except SAXParseException: 
147              raise ParseException(None, sys.exc_info()[2]) 
148          # Now represent the slots in a more meaningful way 
149          self._slots.clear() 
150          for slot_name, (slot_attr, factory) in self._slot_maps.items(): 
151              quantity = getattr(self, slot_attr) 
152              self._slots[slot_name] = factory(quantity) 
153   
154          return self 
155   
156 -    def _load(self, s): 
157          dom_element = parseString(s) 
158          certs = dom_element.getElementsByTagName("rhn-cert") 
159          if not certs: 
160              self._root = None 
161          else: 
162              self._root = certs[0] 
163          for child in self._root.childNodes: 
164              if child.nodeType != child.ELEMENT_NODE: 
165                  # Probably white space 
166                  continue 
167              if child.nodeName == 'rhn-cert-field': 
168                  field_name = child.getAttribute("name") 
169                  if not field_name: 
170                      # XXX Bogus 
171                      continue 
172                  if field_name in self.fields_scalar: 
173                      val = get_text(child) 
174                      if not val: 
175                          continue 
176                      setattr(self, field_name, val) 
177                      continue 
178                  if field_name in self.fields_list: 
179                      val = self.fields_list[field_name](child) 
180                      l = getattr(self, val.attribute_name) 
181                      l.append(val) 
182              elif child.nodeName == 'rhn-cert-signature': 
183                  self.signature = get_text(child) 
184          # Python's docs say: When you are finished with a DOM, you should 
185          # clean it up. This is necessary because some versions of Python do 
186          # not support garbage collection of objects that refer to each other 
187          # in a cycle. Until this restriction is removed from all versions of 
188          # Python, it is safest to write your code as if cycles would not be 
189          # cleaned up. 
190          dom_element.unlink() 
191   
192      _slot_maps = { 
193          'management': ('slots', ManagementSlots), 
194          'provisioning': ('provisioning-slots', ProvisioningSlots), 
195          'monitoring': ('monitoring-slots', MonitoringSlots), 
196          'virtualization': ('virtualization_host', VirtualizationSlots), 
197          'virtualization_platform': ('virtualization_host_platform', VirtualizationPlatformSlots) 
198      } 
199   
200 -    def get_slots(self, slot_type): 
201          if slot_type not in self._slots: 
202              raise AttributeError(slot_type) 
203          return self._slots[slot_type] 
204   
205 -    def get_slot_types(self): 
206          return self._slot_maps.keys() 
207   
208 -    def lookup_slot_by_db_label(self, db_label): 
209          # Given a string like 'sw_mgr_entitled', returns a string 'management' 
210          for label, (_, slot_class) in self._slot_maps.items(): 
211              if slot_class._db_label == db_label: 
212                  return label 
213          return None 
214   
215   
216 -def get_text(node): 
217      return string.join([x.data for x in node.childNodes if x.nodeType == x.TEXT_NODE], "") 
218