sfa/dummy/dummydriver.py

   1 import time
   2 import datetime
   3 #
   4 from sfa.util.faults import MissingSfaInfo, UnknownSfaType, \
   5     RecordNotFound, SfaNotImplemented, SliverDoesNotExist, SearchFailed, \
   6     UnsupportedOperation, Forbidden
   7 from sfa.util.sfalogging import logger
   8 from sfa.util.defaultdict import defaultdict
   9 from sfa.util.sfatime import utcparse, datetime_to_string, datetime_to_epoch
  10 from sfa.util.xrn import Xrn, hrn_to_urn, get_leaf
  11 from sfa.util.cache import Cache
  12
  13 # one would think the driver should not need to mess with the SFA db, but..
  14 from sfa.storage.model import RegRecord, SliverAllocation
  15 from sfa.trust.credential import Credential
  16
  17 # used to be used in get_ticket
  18 #from sfa.trust.sfaticket import SfaTicket
  19 from sfa.rspecs.version_manager import VersionManager
  20 from sfa.rspecs.rspec import RSpec
  21
  22 # the driver interface, mostly provides default behaviours
  23 from sfa.managers.driver import Driver
  24
  25 from sfa.dummy.dummyshell import DummyShell
  26 from sfa.dummy.dummyaggregate import DummyAggregate
  27 from sfa.dummy.dummyslices import DummySlices
  28 from sfa.dummy.dummyxrn import DummyXrn, slicename_to_hrn, hostname_to_hrn, hrn_to_dummy_slicename, xrn_to_hostname
  29
  30
  31 def list_to_dict(recs, key):
  32     """
  33     convert a list of dictionaries into a dictionary keyed on the
  34     specified dictionary key
  35     """
  36     return dict ( [ (rec[key],rec) for rec in recs ] )
  37
  38 #
  39 # DummyShell is just an xmlrpc serverproxy where methods can be sent as-is;
  40 #
  41 class DummyDriver (Driver):
  42
  43     # the cache instance is a class member so it survives across incoming requests
  44     cache = None
  45
  46     def __init__ (self, api):
  47         Driver.__init__ (self, api)
  48         config = api.config
  49         self.hrn = config.SFA_INTERFACE_HRN
  50         self.root_auth = config.SFA_REGISTRY_ROOT_AUTH
  51         self.shell = DummyShell (config)
  52         self.testbedInfo = self.shell.GetTestbedInfo()
  53
  54     def check_sliver_credentials(self, creds, urns):
  55         # build list of cred object hrns
  56         slice_cred_names = []
  57         for cred in creds:
  58             slice_cred_hrn = Credential(cred=cred).get_gid_object().get_hrn()
  59             slice_cred_names.append(DummyXrn(xrn=slice_cred_hrn).dummy_slicename())
  60
  61         # look up slice name of slivers listed in urns arg
  62         slice_ids = []
  63         for urn in urns:
  64             sliver_id_parts = Xrn(xrn=urn).get_sliver_id_parts()
  65             try:
  66                 slice_ids.append(int(sliver_id_parts[0]))
  67             except ValueError:
  68                 pass
  69
  70         if not slice_ids:
  71              raise Forbidden("sliver urn not provided")
  72
  73         slices = self.shell.GetSlices({'slice_ids': slice_ids})
  74         sliver_names = [slice['slice_name'] for slice in slices]
  75
  76         # make sure we have a credential for every specified sliver ierd
  77         for sliver_name in sliver_names:
  78             if sliver_name not in slice_cred_names:
  79                 msg = "Valid credential not found for target: %s" % sliver_name
  80                 raise Forbidden(msg)
  81
  82     ########################################
  83     ########## registry oriented
  84     ########################################
  85
  86     def augment_records_with_testbed_info (self, sfa_records):
  87         return self.fill_record_info (sfa_records)
  88
  89     ##########
  90     def register (self, sfa_record, hrn, pub_key):
  91         type = sfa_record['type']
  92         dummy_record = self.sfa_fields_to_dummy_fields(type, hrn, sfa_record)
  93
  94         if type == 'authority':
  95             pointer = -1
  96
  97         elif type == 'slice':
  98             slices = self.shell.GetSlices({'slice_name': dummy_record['slice_name']})
  99             if not slices:
 100                  pointer = self.shell.AddSlice(dummy_record)
 101             else:
 102                  pointer = slices[0]['slice_id']
 103
 104         elif type == 'user':
 105             users = self.shell.GetUsers({'email':sfa_record['email']})
 106             if not users:
 107                 pointer = self.shell.AddUser(dummy_record)
 108             else:
 109                 pointer = users[0]['user_id']
 110
 111             # Add the user's key
 112             if pub_key:
 113                 self.shell.AddUserKey({'user_id' : pointer, 'key' : pub_key})
 114
 115         elif type == 'node':
 116             nodes = self.shell.GetNodes(dummy_record['hostname'])
 117             if not nodes:
 118                 pointer = self.shell.AddNode(dummy_record)
 119             else:
 120                 pointer = users[0]['node_id']
 121
 122         return pointer
 123
 124     ##########
 125     def update (self, old_sfa_record, new_sfa_record, hrn, new_key):
 126         pointer = old_sfa_record['pointer']
 127         type = old_sfa_record['type']
 128         dummy_record=self.sfa_fields_to_dummy_fields(type, hrn, new_sfa_record)
 129
 130         # new_key implemented for users only
 131         if new_key and type not in [ 'user' ]:
 132             raise UnknownSfaType(type)
 133
 134
 135         if type == "slice":
 136             self.shell.UpdateSlice({'slice_id': pointer, 'fields': dummy_record})
 137
 138         elif type == "user":
 139             self.shell.UpdateUser({'user_id': pointer, 'fields': dummy_record})
 140
 141             if new_key:
 142                 self.shell.AddUserKey({'user_id' : pointer, 'key' : new_key})
 143
 144         elif type == "node":
 145             self.shell.UpdateNode({'node_id': pointer, 'fields': dummy_record})
 146
 147
 148         return True
 149
 150
 151     ##########
 152     def remove (self, sfa_record):
 153         type=sfa_record['type']
 154         pointer=sfa_record['pointer']
 155         if type == 'user':
 156             self.shell.DeleteUser({'user_id': pointer})
 157         elif type == 'slice':
 158             self.shell.DeleteSlice({'slice_id': pointer})
 159         elif type == 'node':
 160             self.shell.DeleteNode({'node_id': pointer})
 161
 162         return True
 163
 164
 165
 166
 167
 168     ##
 169     # Convert SFA fields to Dummy testbed fields for use when registering or updating
 170     # registry record in the dummy testbed
 171     #
 172
 173     def sfa_fields_to_dummy_fields(self, type, hrn, sfa_record):
 174
 175         dummy_record = {}
 176
 177         if type == "slice":
 178             dummy_record["slice_name"] = hrn_to_dummy_slicename(hrn)
 179
 180         elif type == "node":
 181             if "hostname" not in sfa_record:
 182                 raise MissingSfaInfo("hostname")
 183             dummy_record["hostname"] = sfa_record["hostname"]
 184             if "type" in sfa_record:
 185                dummy_record["type"] = sfa_record["type"]
 186             else:
 187                dummy_record["type"] = "dummy_type"
 188
 189         elif type == "authority":
 190             dummy_record["name"] = hrn
 191
 192         elif type == "user":
 193             dummy_record["user_name"] = sfa_record["email"].split('@')[0]
 194             dummy_record["email"] = sfa_record["email"]
 195
 196         return dummy_record
 197
 198     ####################
 199     def fill_record_info(self, records):
 200         """
 201         Given a (list of) SFA record, fill in the DUMMY TESTBED specific
 202         and SFA specific fields in the record.
 203         """
 204         if not isinstance(records, list):
 205             records = [records]
 206
 207         self.fill_record_dummy_info(records)
 208         self.fill_record_hrns(records)
 209         self.fill_record_sfa_info(records)
 210         return records
 211
 212     def fill_record_dummy_info(self, records):
 213         """
 214         Fill in the DUMMY specific fields of a SFA record. This
 215         involves calling the appropriate DUMMY method to retrieve the
 216         database record for the object.
 217
 218         @param record: record to fill in field (in/out param)
 219         """
 220         # get ids by type
 221         node_ids, slice_ids, user_ids = [], [], []
 222         type_map = {'node': node_ids, 'slice': slice_ids, 'user': user_ids}
 223
 224         for record in records:
 225             for type in type_map:
 226                 if type == record['type']:
 227                     type_map[type].append(record['pointer'])
 228
 229         # get dummy records
 230         nodes, slices, users = {}, {}, {}
 231         if node_ids:
 232             node_list = self.shell.GetNodes({'node_ids':node_ids})
 233             nodes = list_to_dict(node_list, 'node_id')
 234         if slice_ids:
 235             slice_list = self.shell.GetSlices({'slice_ids':slice_ids})
 236             slices = list_to_dict(slice_list, 'slice_id')
 237         if user_ids:
 238             user_list = self.shell.GetUsers({'user_ids': user_ids})
 239             users = list_to_dict(user_list, 'user_id')
 240
 241         dummy_records = {'node': nodes, 'slice': slices, 'user': users}
 242
 243
 244         # fill record info
 245         for record in records:
 246             # records with pointer==-1 do not have dummy info.
 247             if record['pointer'] == -1:
 248                 continue
 249
 250             for type in dummy_records:
 251                 if record['type'] == type:
 252                     if record['pointer'] in dummy_records[type]:
 253                         record.update(dummy_records[type][record['pointer']])
 254                         break
 255             # fill in key info
 256             if record['type'] == 'user':
 257                 record['key_ids'] = []
 258                 record['keys'] = []
 259                 for key in dummy_records['user'][record['pointer']]['keys']:
 260                      record['key_ids'].append(-1)
 261                      record['keys'].append(key)
 262
 263         return records
 264
 265     def fill_record_hrns(self, records):
 266         """
 267         convert dummy ids to hrns
 268         """
 269
 270         # get ids
 271         slice_ids, user_ids, node_ids = [], [], []
 272         for record in records:
 273             if 'user_ids' in record:
 274                 user_ids.extend(record['user_ids'])
 275             if 'slice_ids' in record:
 276                 slice_ids.extend(record['slice_ids'])
 277             if 'node_ids' in record:
 278                 node_ids.extend(record['node_ids'])
 279
 280         # get dummy records
 281         slices, users, nodes = {}, {}, {}
 282         if user_ids:
 283             user_list = self.shell.GetUsers({'user_ids': user_ids})
 284             users = list_to_dict(user_list, 'user_id')
 285         if slice_ids:
 286             slice_list = self.shell.GetSlices({'slice_ids': slice_ids})
 287             slices = list_to_dict(slice_list, 'slice_id')
 288         if node_ids:
 289             node_list = self.shell.GetNodes({'node_ids': node_ids})
 290             nodes = list_to_dict(node_list, 'node_id')
 291
 292         # convert ids to hrns
 293         for record in records:
 294             # get all relevant data
 295             type = record['type']
 296             pointer = record['pointer']
 297             testbed_name = self.testbed_name()
 298             auth_hrn = self.hrn
 299             if pointer == -1:
 300                 continue
 301
 302             if 'user_ids' in record:
 303                 emails = [users[user_id]['email'] for user_id in record['user_ids'] \
 304                           if user_id in  users]
 305                 usernames = [email.split('@')[0] for email in emails]
 306                 user_hrns = [".".join([auth_hrn, testbed_name, username]) for username in usernames]
 307                 record['users'] = user_hrns
 308             if 'slice_ids' in record:
 309                 slicenames = [slices[slice_id]['slice_name'] for slice_id in record['slice_ids'] \
 310                               if slice_id in slices]
 311                 slice_hrns = [slicename_to_hrn(auth_hrn, slicename) for slicename in slicenames]
 312                 record['slices'] = slice_hrns
 313             if 'node_ids' in record:
 314                 hostnames = [nodes[node_id]['hostname'] for node_id in record['node_ids'] \
 315                              if node_id in nodes]
 316                 node_hrns = [hostname_to_hrn(auth_hrn, login_base, hostname) for hostname in hostnames]
 317                 record['nodes'] = node_hrns
 318
 319
 320         return records
 321
 322     def fill_record_sfa_info(self, records):
 323
 324         def startswith(prefix, values):
 325             return [value for value in values if value.startswith(prefix)]
 326
 327         # get user ids
 328         user_ids = []
 329         for record in records:
 330             user_ids.extend(record.get("user_ids", []))
 331
 332         # get sfa records for all records associated with these records.
 333         # we'll replace pl ids (person_ids) with hrns from the sfa records
 334         # we obtain
 335
 336         # get the registry records
 337         user_list, users = [], {}
 338         user_list = self.api.dbsession().query (RegRecord).filter(RegRecord.pointer.in_(user_ids))
 339         # create a hrns keyed on the sfa record's pointer.
 340         # Its possible for multiple records to have the same pointer so
 341         # the dict's value will be a list of hrns.
 342         users = defaultdict(list)
 343         for user in user_list:
 344             users[user.pointer].append(user)
 345
 346         # get the dummy records
 347         dummy_user_list, dummy_users = [], {}
 348         dummy_user_list = self.shell.GetUsers({'user_ids': user_ids})
 349         dummy_users = list_to_dict(dummy_user_list, 'user_id')
 350
 351         # fill sfa info
 352         for record in records:
 353             # skip records with no pl info (top level authorities)
 354             #if record['pointer'] == -1:
 355             #    continue
 356             sfa_info = {}
 357             type = record['type']
 358             logger.info("fill_record_sfa_info - incoming record typed %s"%type)
 359             if (type == "slice"):
 360                 # all slice users are researchers
 361                 record['geni_urn'] = hrn_to_urn(record['hrn'], 'slice')
 362                 record['PI'] = []
 363                 record['researcher'] = []
 364                 for user_id in record.get('user_ids', []):
 365                     hrns = [user.hrn for user in users[user_id]]
 366                     record['researcher'].extend(hrns)
 367
 368             elif (type.startswith("authority")):
 369                 record['url'] = None
 370                 logger.info("fill_record_sfa_info - authority xherex")
 371
 372             elif (type == "node"):
 373                 sfa_info['dns'] = record.get("hostname", "")
 374                 # xxx TODO: URI, LatLong, IP, DNS
 375
 376             elif (type == "user"):
 377                 logger.info('setting user.email')
 378                 sfa_info['email'] = record.get("email", "")
 379                 sfa_info['geni_urn'] = hrn_to_urn(record['hrn'], 'user')
 380                 sfa_info['geni_certificate'] = record['gid']
 381                 # xxx TODO: PostalAddress, Phone
 382             record.update(sfa_info)
 383
 384
 385     ####################
 386     def update_relation (self, subject_type, target_type, relation_name, subject_id, target_ids):
 387         # hard-wire the code for slice/user for now, could be smarter if needed
 388         if subject_type =='slice' and target_type == 'user' and relation_name == 'researcher':
 389             subject=self.shell.GetSlices ({'slice_id': subject_id})[0]
 390             if 'user_ids' not in subject.keys():
 391                  subject['user_ids'] = []
 392             current_target_ids = subject['user_ids']
 393             add_target_ids = list ( set (target_ids).difference(current_target_ids))
 394             del_target_ids = list ( set (current_target_ids).difference(target_ids))
 395             logger.debug ("subject_id = %s (type=%s)"%(subject_id,type(subject_id)))
 396             for target_id in add_target_ids:
 397                 self.shell.AddUserToSlice ({'user_id': target_id, 'slice_id': subject_id})
 398                 logger.debug ("add_target_id = %s (type=%s)"%(target_id,type(target_id)))
 399             for target_id in del_target_ids:
 400                 logger.debug ("del_target_id = %s (type=%s)"%(target_id,type(target_id)))
 401                 self.shell.DeleteUserFromSlice ({'user_id': target_id, 'slice_id': subject_id})
 402         else:
 403             logger.info('unexpected relation %s to maintain, %s -> %s'%(relation_name,subject_type,target_type))
 404
 405
 406     ########################################
 407     ########## aggregate oriented
 408     ########################################
 409
 410     def testbed_name (self): return "dummy"
 411
 412     def aggregate_version (self):
 413         return {}
 414
 415     def list_resources (self, version=None, options=None):
 416         if options is None: options={}
 417         aggregate = DummyAggregate(self)
 418         rspec =  aggregate.list_resources(version=version, options=options)
 419         return rspec
 420
 421     def describe(self, urns, version, options=None):
 422         if options is None: options={}
 423         aggregate = DummyAggregate(self)
 424         return aggregate.describe(urns, version=version, options=options)
 425
 426     def status (self, urns, options=None):
 427         if options is None: options={}
 428         aggregate = DummyAggregate(self)
 429         desc =  aggregate.describe(urns, version='GENI 3')
 430         status = {'geni_urn': desc['geni_urn'],
 431                   'geni_slivers': desc['geni_slivers']}
 432         return status
 433
 434
 435     def allocate (self, urn, rspec_string, expiration, options=None):
 436         if options is None: options={}
 437         xrn = Xrn(urn)
 438         aggregate = DummyAggregate(self)
 439         slices = DummySlices(self)
 440         slice_record=None
 441         users = options.get('geni_users', [])
 442         if users:
 443             slice_record = users[0].get('slice_record', {})
 444
 445         # parse rspec
 446         rspec = RSpec(rspec_string)
 447         requested_attributes = rspec.version.get_slice_attributes()
 448
 449         # ensure slice record exists
 450         slice = slices.verify_slice(xrn.hrn, slice_record, expiration=expiration, options=options)
 451         # ensure person records exists
 452         #persons = slices.verify_persons(xrn.hrn, slice, users, peer, sfa_peer, options=options)
 453
 454         # add/remove slice from nodes
 455         request_nodes = rspec.version.get_nodes_with_slivers()
 456         nodes = slices.verify_slice_nodes(urn, slice, request_nodes)
 457
 458         return aggregate.describe([xrn.get_urn()], version=rspec.version)
 459
 460     def provision(self, urns, options=None):
 461         if options is None: options={}
 462         # update users
 463         slices = DummySlices(self)
 464         aggregate = DummyAggregate(self)
 465         slivers = aggregate.get_slivers(urns)
 466         slice = slivers[0]
 467         geni_users = options.get('geni_users', [])
 468         #users = slices.verify_users(None, slice, geni_users, options=options)
 469         # update sliver allocation states and set them to geni_provisioned
 470         sliver_ids = [sliver['sliver_id'] for sliver in slivers]
 471         dbsession=self.api.dbsession()
 472         SliverAllocation.set_allocations(sliver_ids, 'geni_provisioned',dbsession)
 473         version_manager = VersionManager()
 474         rspec_version = version_manager.get_version(options['geni_rspec_version'])
 475         return self.describe(urns, rspec_version, options=options)
 476
 477     def delete(self, urns, options=None):
 478         if options is None: options={}
 479         # collect sliver ids so we can update sliver allocation states after
 480         # we remove the slivers.
 481         aggregate = DummyAggregate(self)
 482         slivers = aggregate.get_slivers(urns)
 483         if slivers:
 484             slice_id = slivers[0]['slice_id']
 485             node_ids = []
 486             sliver_ids = []
 487             for sliver in slivers:
 488                 node_ids.append(sliver['node_id'])
 489                 sliver_ids.append(sliver['sliver_id'])
 490
 491             # determine if this is a peer slice
 492             # xxx I wonder if this would not need to use PlSlices.get_peer instead
 493             # in which case plc.peers could be deprecated as this here
 494             # is the only/last call to this last method in plc.peers
 495             slice_hrn = DummyXrn(auth=self.hrn, slicename=slivers[0]['slice_name']).get_hrn()
 496             try:
 497                 self.shell.DeleteSliceFromNodes({'slice_id': slice_id, 'node_ids': node_ids})
 498                 # delete sliver allocation states
 499                 dbsession=self.api.dbsession()
 500                 SliverAllocation.delete_allocations(sliver_ids,dbsession)
 501             finally:
 502                 pass
 503
 504         # prepare return struct
 505         geni_slivers = []
 506         for sliver in slivers:
 507             geni_slivers.append(
 508                 {'geni_sliver_urn': sliver['sliver_id'],
 509                  'geni_allocation_status': 'geni_unallocated',
 510                  'geni_expires': datetime_to_string(utcparse(sliver['expires']))})
 511         return geni_slivers
 512
 513     def renew (self, urns, expiration_time, options=None):
 514         if options is None: options={}
 515         aggregate = DummyAggregate(self)
 516         slivers = aggregate.get_slivers(urns)
 517         if not slivers:
 518             raise SearchFailed(urns)
 519         slice = slivers[0]
 520         requested_time = utcparse(expiration_time)
 521         record = {'expires': int(datetime_to_epoch(requested_time))}
 522         self.shell.UpdateSlice({'slice_id': slice['slice_id'], 'fileds': record})
 523         description = self.describe(urns, 'GENI 3', options)
 524         return description['geni_slivers']
 525
 526     def perform_operational_action (self, urns, action, options=None):
 527         if options is None: options={}
 528         # Dummy doesn't support operational actions. Lets pretend like it
 529         # supports start, but reject everything else.
 530         action = action.lower()
 531         if action not in ['geni_start']:
 532             raise UnsupportedOperation(action)
 533
 534         # fault if sliver is not full allocated (operational status is geni_pending_allocation)
 535         description = self.describe(urns, 'GENI 3', options)
 536         for sliver in description['geni_slivers']:
 537             if sliver['geni_operational_status'] == 'geni_pending_allocation':
 538                 raise UnsupportedOperation(action, "Sliver must be fully allocated (operational status is not geni_pending_allocation)")
 539         #
 540         # Perform Operational Action Here
 541         #
 542
 543         geni_slivers = self.describe(urns, 'GENI 3', options)['geni_slivers']
 544         return geni_slivers
 545
 546     def shutdown (self, xrn, options=None):
 547         if options is None: options={}
 548         xrn = DummyXrn(xrn=xrn, type='slice')
 549         slicename = xrn.pl_slicename()
 550         slices = self.shell.GetSlices({'name': slicename}, ['slice_id'])
 551         if not slices:
 552             raise RecordNotFound(slice_hrn)
 553         slice_id = slices[0]['slice_id']
 554         slice_tags = self.shell.GetSliceTags({'slice_id': slice_id, 'tagname': 'enabled'})
 555         if not slice_tags:
 556             self.shell.AddSliceTag(slice_id, 'enabled', '0')
 557         elif slice_tags[0]['value'] != "0":
 558             tag_id = slice_tags[0]['slice_tag_id']
 559             self.shell.UpdateSliceTag(tag_id, '0')
 560         return 1