Context Navigation

source: fedd/federation/emulab_access.py @ 5d7f1e8

axis_examplecompt_changesinfo-ops

Last change on this file since 5d7f1e8 was de86b35, checked in by Ted Faber <faber@…>, 14 years ago
Flat out bug in the lambda definition in access. The emulab_access deals with cases when services are requested but not export project. (That was a bug, too)
Property mode set to `100644`
File size: 37.9 KB

Rev	Line
[19cc408]	1	#!/usr/local/bin/python
	2
	3	import os,sys
[eeb0088]	4	import stat # for chmod constants
[19cc408]	5	import re
[ab847bc]	6	import random
[19cc408]	7	import string
	8	import copy
[d81971a]	9	import pickle
[c971895]	10	import logging
[eeb0088]	11	import subprocess
[06cc65b]	12	import traceback
[19cc408]	13
[f8582c9]	14	from threading import *
[8e6fe4d]	15	from M2Crypto.SSL import SSLError
[f8582c9]	16
[f771e2f]	17	from access import access_base
[78f2668]	18	from legacy_access import legacy_access
[f771e2f]	19
[ec4fb42]	20	from util import *
	21	from allocate_project import allocate_project_local, allocate_project_remote
[51cc9df]	22	from fedid import fedid, generate_fedid
[6e63513]	23	from authorizer import authorizer, abac_authorizer
[6a0c9f4]	24	from service_error import service_error
[9460b1e]	25	from remote_service import xmlrpc_handler, soap_handler, service_caller
[11a08b0]	26
[6c57fe9]	27	import httplib
	28	import tempfile
	29	from urlparse import urlparse
	30
[11860f52]	31	import topdl
	32	import list_log
	33	import proxy_emulab_segment
	34	import local_emulab_segment
	35
[0ea11af]	36
	37	# Make log messages disappear if noone configures a fedd logger
[11a08b0]	38	class nullHandler(logging.Handler):
	39	def emit(self, record): pass
	40
	41	fl = logging.getLogger("fedd.access")
	42	fl.addHandler(nullHandler())
[19cc408]	43
[78f2668]	44	class access(access_base, legacy_access):
[19cc408]	45	"""
	46	The implementation of access control based on mapping users to projects.
	47
	48	Users can be mapped to existing projects or have projects created
	49	dynamically. This implements both direct requests and proxies.
	50	"""
	51
[3f6bc5f]	52	def __init__(self, config=None, auth=None):
[866c983]	53	"""
	54	Initializer. Pulls parameters out of the ConfigParser's access section.
	55	"""
	56
[f771e2f]	57	access_base.__init__(self, config, auth)
[866c983]	58
	59	self.allow_proxy = config.getboolean("access", "allow_proxy")
	60
	61	self.domain = config.get("access", "domain")
[eeb0088]	62	self.userconfdir = config.get("access","userconfdir")
	63	self.userconfcmd = config.get("access","userconfcmd")
[fe28bb2]	64	self.userconfurl = config.get("access","userconfurl")
[9b3627e]	65	self.federation_software = config.get("access", "federation_software")
	66	self.portal_software = config.get("access", "portal_software")
[e76f38a]	67	self.local_seer_software = config.get("access", "local_seer_software")
	68	self.local_seer_image = config.get("access", "local_seer_image")
	69	self.local_seer_start = config.get("access", "local_seer_start")
[49051fb]	70	self.seer_master_start = config.get("access", "seer_master_start")
[ecca6eb]	71	self.ssh_privkey_file = config.get("access","ssh_privkey_file")
[3bddd24]	72	self.ssh_pubkey_file = config.get("access","ssh_pubkey_file")
[6280b1f]	73	self.ssh_port = config.get("access","ssh_port") or "22"
[181aeb4]	74	self.boss = config.get("access", "boss")
[814b5e5]	75	self.ops = config.get("access", "ops")
[181aeb4]	76	self.xmlrpc_cert = config.get("access", "xmlrpc_cert")
	77	self.xmlrpc_certpw = config.get("access", "xmlrpc_certpw")
[5e1fb7b]	78
	79	self.dragon_endpoint = config.get("access", "dragon")
	80	self.dragon_vlans = config.get("access", "dragon_vlans")
	81	self.deter_internal = config.get("access", "deter_internal")
	82
	83	self.tunnel_config = config.getboolean("access", "tunnel_config")
	84	self.portal_command = config.get("access", "portal_command")
	85	self.portal_image = config.get("access", "portal_image")
	86	self.portal_type = config.get("access", "portal_type") or "pc"
	87	self.portal_startcommand = config.get("access", "portal_startcommand")
	88	self.node_startcommand = config.get("access", "node_startcommand")
	89
[f771e2f]	90	self.federation_software = self.software_list(self.federation_software)
	91	self.portal_software = self.software_list(self.portal_software)
	92	self.local_seer_software = self.software_list(self.local_seer_software)
[11860f52]	93
	94	self.access_type = self.access_type.lower()
	95	if self.access_type == 'remote_emulab':
	96	self.start_segment = proxy_emulab_segment.start_segment
	97	self.stop_segment = proxy_emulab_segment.stop_segment
	98	elif self.access_type == 'local_emulab':
	99	self.start_segment = local_emulab_segment.start_segment
	100	self.stop_segment = local_emulab_segment.stop_segment
	101	else:
	102	self.start_segment = None
	103	self.stop_segment = None
[866c983]	104
	105	self.restricted = [ ]
[6e63513]	106	# XXX: this should go?
	107	#if config.has_option("access", "accessdb"):
	108	# self.read_access(config.get("access", "accessdb"))
[866c983]	109	tb = config.get('access', 'testbed')
	110	if tb: self.testbed = [ t.strip() for t in tb.split(',') ]
	111	else: self.testbed = [ ]
	112
[6e63513]	113	# authorization information
	114	self.auth_type = config.get('access', 'auth_type') \
	115	or 'legacy'
	116	self.auth_dir = config.get('access', 'auth_dir')
	117	accessdb = config.get("access", "accessdb")
	118	# initialize the authorization system
	119	if self.auth_type == 'legacy':
[c002cb2]	120	self.access = { }
[6e63513]	121	if accessdb:
[78f2668]	122	self.legacy_read_access(accessdb, self.legacy_access_tuple)
[6e63513]	123	elif self.auth_type == 'abac':
	124	self.auth = abac_authorizer(load=self.auth_dir)
[c002cb2]	125	self.access = [ ]
[6e63513]	126	if accessdb:
[78f2668]	127	self.read_access(accessdb, self.access_tuple)
[6e63513]	128	else:
	129	raise service_error(service_error.internal,
	130	"Unknown auth_type: %s" % self.auth_type)
[f771e2f]	131
[06cc65b]	132	# read_state in the base_class
[f771e2f]	133	self.state_lock.acquire()
[06cc65b]	134	for a in ('allocation', 'projects', 'keys', 'types'):
	135	if a not in self.state:
	136	self.state[a] = { }
[f771e2f]	137	self.allocation = self.state['allocation']
	138	self.projects = self.state['projects']
	139	self.keys = self.state['keys']
	140	self.types = self.state['types']
[6e63513]	141	if self.auth_type == "legacy":
	142	# Add the ownership attributes to the authorizer. Note that the
	143	# indices of the allocation dict are strings, but the attributes are
	144	# fedids, so there is a conversion.
	145	for k in self.allocation.keys():
	146	for o in self.allocation[k].get('owners', []):
	147	self.auth.set_attribute(o, fedid(hexstr=k))
	148	if self.allocation[k].has_key('userconfig'):
	149	sfid = self.allocation[k]['userconfig']
	150	fid = fedid(hexstr=sfid)
	151	self.auth.set_attribute(fid, "/%s" % sfid)
[f771e2f]	152	self.state_lock.release()
[a20a20f]	153	self.exports = {
	154	'SMB': self.export_SMB,
	155	'seer': self.export_seer,
	156	'tmcd': self.export_tmcd,
	157	'userconfig': self.export_userconfig,
	158	'project_export': self.export_project_export,
	159	'local_seer_control': self.export_local_seer,
	160	'seer_master': self.export_seer_master,
	161	'hide_hosts': self.export_hide_hosts,
	162	}
	163
	164	if not self.local_seer_image or not self.local_seer_software or \
	165	not self.local_seer_start:
	166	if 'local_seer_control' in self.exports:
	167	del self.exports['local_seer_control']
	168
	169	if not self.local_seer_image or not self.local_seer_software or \
	170	not self.seer_master_start:
	171	if 'seer_master' in self.exports:
	172	del self.exports['seer_master']
[866c983]	173
	174
	175	self.soap_services = {\
	176	'RequestAccess': soap_handler("RequestAccess", self.RequestAccess),
	177	'ReleaseAccess': soap_handler("ReleaseAccess", self.ReleaseAccess),
[cc8d8e9]	178	'StartSegment': soap_handler("StartSegment", self.StartSegment),
[e76f38a]	179	'TerminateSegment': soap_handler("TerminateSegment",
	180	self.TerminateSegment),
[866c983]	181	}
	182	self.xmlrpc_services = {\
	183	'RequestAccess': xmlrpc_handler('RequestAccess',
	184	self.RequestAccess),
	185	'ReleaseAccess': xmlrpc_handler('ReleaseAccess',
	186	self.ReleaseAccess),
[5ae3857]	187	'StartSegment': xmlrpc_handler("StartSegment", self.StartSegment),
	188	'TerminateSegment': xmlrpc_handler('TerminateSegment',
	189	self.TerminateSegment),
[866c983]	190	}
	191
[2761484]	192	self.call_SetValue = service_caller('SetValue')
[2ee4226]	193	self.call_GetValue = service_caller('GetValue', log=self.log)
[866c983]	194
	195	if not config.has_option("allocate", "uri"):
	196	self.allocate_project = \
	197	allocate_project_local(config, auth)
	198	else:
	199	self.allocate_project = \
	200	allocate_project_remote(config, auth)
	201
[e76f38a]	202
[866c983]	203	# If the project allocator exports services, put them in this object's
	204	# maps so that classes that instantiate this can call the services.
	205	self.soap_services.update(self.allocate_project.soap_services)
	206	self.xmlrpc_services.update(self.allocate_project.xmlrpc_services)
[f8582c9]	207
[e76f38a]	208	@staticmethod
[78f2668]	209	def legacy_access_tuple(str):
[06cc65b]	210	"""
[027b87b]	211	Convert a string of the form (id[:resources:resouces], id, id) into a
	212	tuple of the form (project, user, user) where users may be names or
	213	fedids. The resources strings are obsolete and ignored.
[06cc65b]	214	"""
[866c983]	215	def parse_name(n):
	216	if n.startswith('fedid:'): return fedid(hexstr=n[len('fedid:'):])
	217	else: return n
[19cc408]	218
[f771e2f]	219	str = str.strip()
	220	if str.startswith('(') and str.endswith(')'):
	221	str = str[1:-1]
	222	names = [ s.strip() for s in str.split(",")]
	223	if len(names) > 3:
	224	raise self.parse_error("More than three fields in name")
	225	first = names[0].split(":")
	226	if first == 'fedid:':
	227	del first[0]
	228	first[0] = fedid(hexstr=first[0])
[027b87b]	229	names[0] = first[0]
[f771e2f]	230
	231	for i in range(1,2):
	232	names[i] = parse_name(names[i])
	233
	234	return tuple(names)
	235	else:
	236	raise self.parse_error('Bad mapping (unbalanced parens)')
[19cc408]	237
[6e63513]	238	@staticmethod
[78f2668]	239	def access_tuple(str):
[6e63513]	240	"""
	241	Convert a string of the form (id, id) into an access_project. This is
[78f2668]	242	called by read_access to convert to local attributes. It returns
[6e63513]	243	a tuple of the form (project, user, user) where the two users are
	244	always the same.
	245	"""
	246
	247	str = str.strip()
	248	if str.startswith('(') and str.endswith(')') and str.count(',') == 1:
[725c55d]	249	# The slice takes the parens off the string.
	250	proj, user = str[1:-1].split(',')
[027b87b]	251	return (proj.strip(), user.strip(), user.strip())
[6e63513]	252	else:
	253	raise self.parse_error(
	254	'Bad mapping (unbalanced parens or more than 1 comma)')
	255
[19cc408]	256
[06cc65b]	257	# RequestAccess support routines
	258
[78f2668]	259	def legacy_lookup_access(self, req, fid):
[866c983]	260	"""
[06cc65b]	261	Look up the local access control information mapped to this fedid and
	262	credentials. In this case it is a (project, create_user, access_user)
	263	triple, and a triple of three booleans indicating which, if any will
	264	need to be dynamically created. Finally a list of owners for that
	265	allocation is returned.
	266
	267	lookup_access_base pulls the first triple out, and it is parsed by this
	268	routine into the boolean map. Owners is always the controlling fedid.
[866c983]	269	"""
	270	# Return values
[027b87b]	271	rp = None
[866c983]	272	ru = None
	273	# This maps a valid user to the Emulab projects and users to use
[78f2668]	274	found, match = self.legacy_lookup_access_base(req, fid)
[f771e2f]	275	tb, project, user = match
[866c983]	276
	277	if found == None:
	278	raise service_error(service_error.access,
	279	"Access denied - cannot map access")
	280
	281	# resolve <dynamic> and <same> in found
	282	dyn_proj = False
	283	dyn_create_user = False
	284	dyn_service_user = False
	285
[027b87b]	286	if found[0] == "<same>":
[866c983]	287	if project != None:
[027b87b]	288	rp = project
[866c983]	289	else :
	290	raise service_error(\
	291	service_error.server_config,
	292	"Project matched <same> when no project given")
[027b87b]	293	elif found[0] == "<dynamic>":
	294	rp = None
[866c983]	295	dyn_proj = True
	296	else:
[027b87b]	297	rp = found[0]
[866c983]	298
	299	if found[1] == "<same>":
	300	if user_match == "<any>":
	301	if user != None: rcu = user[0]
	302	else: raise service_error(\
	303	service_error.server_config,
	304	"Matched <same> on anonymous request")
	305	else:
	306	rcu = user_match
	307	elif found[1] == "<dynamic>":
	308	rcu = None
	309	dyn_create_user = True
	310	else:
	311	rcu = found[1]
	312
	313	if found[2] == "<same>":
	314	if user_match == "<any>":
	315	if user != None: rsu = user[0]
	316	else: raise service_error(\
	317	service_error.server_config,
	318	"Matched <same> on anonymous request")
	319	else:
	320	rsu = user_match
	321	elif found[2] == "<dynamic>":
	322	rsu = None
	323	dyn_service_user = True
	324	else:
	325	rsu = found[2]
	326
	327	return (rp, rcu, rsu), (dyn_create_user, dyn_service_user, dyn_proj),\
[f771e2f]	328	[ fid ]
[19cc408]	329
[f771e2f]	330	def do_project_allocation(self, dyn, project, user):
	331	"""
	332	Call the project allocation routines and return the info.
	333	"""
	334	if dyn:
	335	# Compose the dynamic project request
	336	# (only dynamic, dynamic currently allowed)
	337	preq = { 'AllocateProjectRequestBody': \
	338	{ 'project' : {\
	339	'user': [ \
	340	{ \
	341	'access': [ {
	342	'sshPubkey': self.ssh_pubkey_file } ],
	343	'role': "serviceAccess",\
	344	}, \
	345	{ \
	346	'access': [ {
	347	'sshPubkey': self.ssh_pubkey_file } ],
	348	'role': "experimentCreation",\
	349	}, \
	350	], \
	351	}\
	352	}\
	353	}
	354	return self.allocate_project.dynamic_project(preq)
[eeb0088]	355	else:
[f771e2f]	356	preq = {'StaticProjectRequestBody' : \
	357	{ 'project': \
	358	{ 'name' : { 'localname' : project },\
	359	'user' : [ \
	360	{\
	361	'userID': { 'localname' : user }, \
	362	'access': [ {
	363	'sshPubkey': self.ssh_pubkey_file } ],
	364	'role': 'experimentCreation'\
	365	},\
	366	{\
	367	'userID': { 'localname' : user}, \
	368	'access': [ {
	369	'sshPubkey': self.ssh_pubkey_file } ],
	370	'role': 'serviceAccess'\
	371	},\
	372	]}\
	373	}\
	374	}
	375	return self.allocate_project.static_project(preq)
[eeb0088]	376
[f771e2f]	377	def save_project_state(self, aid, ap, dyn, owners):
	378	"""
	379	Parse out and save the information relevant to the project created for
	380	this experiment. That info is largely in ap and owners. dyn indicates
	381	that the project was created dynamically. Return the user and project
	382	names.
	383	"""
	384	self.state_lock.acquire()
	385	self.allocation[aid] = { }
[eeb0088]	386	try:
[f771e2f]	387	pname = ap['project']['name']['localname']
	388	except KeyError:
	389	pname = None
	390
	391	if dyn:
	392	if not pname:
	393	self.state_lock.release()
	394	raise service_error(service_error.internal,
	395	"Misformed allocation response?")
	396	if pname in self.projects: self.projects[pname] += 1
	397	else: self.projects[pname] = 1
	398	self.allocation[aid]['project'] = pname
[43197eb]	399	else:
[f771e2f]	400	# sproject is a static project associated with this allocation.
	401	self.allocation[aid]['sproject'] = pname
[866c983]	402
[f771e2f]	403	self.allocation[aid]['keys'] = [ ]
[5e1fb7b]	404
[f771e2f]	405	try:
	406	for u in ap['project']['user']:
	407	uname = u['userID']['localname']
	408	if u['role'] == 'experimentCreation':
	409	self.allocation[aid]['user'] = uname
	410	for k in [ k['sshPubkey'] for k in u['access'] \
	411	if k.has_key('sshPubkey') ]:
	412	kv = "%s:%s" % (uname, k)
	413	if self.keys.has_key(kv): self.keys[kv] += 1
	414	else: self.keys[kv] = 1
	415	self.allocation[aid]['keys'].append((uname, k))
	416	except KeyError:
	417	self.state_lock.release()
	418	raise service_error(service_error.internal,
	419	"Misformed allocation response?")
	420
	421	self.allocation[aid]['owners'] = owners
	422	self.write_state()
	423	self.state_lock.release()
	424	return (pname, uname)
[19cc408]	425
[06cc65b]	426	# End of RequestAccess support routines
	427
[19cc408]	428	def RequestAccess(self, req, fid):
[866c983]	429	"""
	430	Handle the access request. Proxy if not for us.
	431
	432	Parse out the fields and make the allocations or rejections if for us,
	433	otherwise, assuming we're willing to proxy, proxy the request out.
	434	"""
	435
	436	def gateway_hardware(h):
[5e1fb7b]	437	if h == 'GWTYPE': return self.portal_type or 'GWTYPE'
[866c983]	438	else: return h
	439
[43197eb]	440	def get_export_project(svcs):
	441	"""
	442	if the service requests includes one to export a project, return
	443	that project.
	444	"""
	445	rv = None
	446	for s in svcs:
	447	if s.get('name', '') == 'project_export' and \
	448	s.get('visibility', '') == 'export':
	449	if not rv:
[1f6a573]	450	for a in s.get('fedAttr', []):
[43197eb]	451	if a.get('attribute', '') == 'project' \
	452	and 'value' in a:
	453	rv = a['value']
	454	else:
	455	raise service_error(service_error, access,
	456	'Requesting multiple project exports is ' + \
	457	'not supported');
	458	return rv
	459
[866c983]	460	# The dance to get into the request body
	461	if req.has_key('RequestAccessRequestBody'):
	462	req = req['RequestAccessRequestBody']
	463	else:
	464	raise service_error(service_error.req, "No request!?")
	465
[1f6a573]	466	# if this includes a project export request, construct a filter such
	467	# that only the ABAC attributes mapped to that project are checked for
	468	# access.
	469	if 'service' in req:
	470	ep = get_export_project(req['service'])
[de86b35]	471	if ep: pf = lambda(a): a.value[0] == ep
	472	else: pf = None
[1f6a573]	473	else:
	474	ep = None
	475	pf = None
	476
[c573278]	477	if self.auth.import_credentials(
	478	data_list=req.get('abac_credential', [])):
	479	self.auth.save()
[866c983]	480
[6e63513]	481	if self.auth_type == "legacy":
[78f2668]	482	found, dyn, owners = self.legacy_lookup_access(req, fid)
[6e63513]	483	elif self.auth_type == 'abac':
[1f6a573]	484	found, dyn, owners = self.lookup_access(req, fid, filter=pf)
[6e63513]	485	else:
	486	raise service_error(service_error.internal,
	487	'Unknown auth_type: %s' % self.auth_type)
[f771e2f]	488	ap = None
[866c983]	489
[1f6a573]	490	# This only happens in legacy lookups, but if this user has access to
	491	# the testbed but not the project to be exported, raise the error.
	492	if ep and ep != found[0]:
	493	raise service_error(service_error.access,
	494	"Cannot export %s" % ep)
[f771e2f]	495
	496	if self.ssh_pubkey_file:
[027b87b]	497	ap = self.do_project_allocation(dyn[1], found[0], found[1])
[f771e2f]	498	else:
	499	raise service_error(service_error.internal,
	500	"SSH access parameters required")
	501	# keep track of what's been added
	502	allocID, alloc_cert = generate_fedid(subj="alloc", log=self.log)
	503	aid = unicode(allocID)
	504
	505	pname, uname = self.save_project_state(aid, ap, dyn[1], owners)
	506
	507	services, svc_state = self.export_services(req.get('service',[]),
	508	pname, uname)
	509	self.state_lock.acquire()
	510	# Store services state in global state
	511	for k, v in svc_state.items():
	512	self.allocation[aid][k] = v
	513	self.write_state()
	514	self.state_lock.release()
	515	# Give the owners the right to change this allocation
	516	for o in owners:
	517	self.auth.set_attribute(o, allocID)
[6e63513]	518	self.auth.save()
[f771e2f]	519	try:
	520	f = open("%s/%s.pem" % (self.certdir, aid), "w")
	521	print >>f, alloc_cert
	522	f.close()
	523	except EnvironmentError, e:
	524	raise service_error(service_error.internal,
	525	"Can't open %s/%s : %s" % (self.certdir, aid, e))
	526	resp = self.build_access_response({ 'fedid': allocID } ,
	527	ap, services)
	528	return resp
[d81971a]	529
[8cf2b90e]	530	def do_release_project(self, del_project, del_users, del_types):
	531	"""
	532	If a project and users has to be deleted, make the call.
	533	"""
	534	msg = { 'project': { }}
	535	if del_project:
	536	msg['project']['name']= {'localname': del_project}
	537	users = [ ]
	538	for u in del_users.keys():
	539	users.append({ 'userID': { 'localname': u },\
	540	'access' : \
	541	[ {'sshPubkey' : s } for s in del_users[u]]\
	542	})
	543	if users:
	544	msg['project']['user'] = users
	545	if len(del_types) > 0:
	546	msg['resources'] = { 'node': \
	547	[ {'hardware': [ h ] } for h in del_types ]\
	548	}
	549	if self.allocate_project.release_project:
	550	msg = { 'ReleaseProjectRequestBody' : msg}
	551	self.allocate_project.release_project(msg)
	552
[d81971a]	553	def ReleaseAccess(self, req, fid):
[866c983]	554	# The dance to get into the request body
	555	if req.has_key('ReleaseAccessRequestBody'):
	556	req = req['ReleaseAccessRequestBody']
	557	else:
	558	raise service_error(service_error.req, "No request!?")
	559
[8cf2b90e]	560	try:
	561	if req['allocID'].has_key('localname'):
	562	auth_attr = aid = req['allocID']['localname']
	563	elif req['allocID'].has_key('fedid'):
	564	aid = unicode(req['allocID']['fedid'])
	565	auth_attr = req['allocID']['fedid']
	566	else:
	567	raise service_error(service_error.req,
	568	"Only localnames and fedids are understood")
	569	except KeyError:
	570	raise service_error(service_error.req, "Badly formed request")
	571
[725c55d]	572	self.log.debug("[access] deallocation requested for %s by %s" % \
	573	(aid, fid))
[8cf2b90e]	574	if not self.auth.check_attribute(fid, auth_attr):
	575	self.log.debug("[access] deallocation denied for %s", aid)
	576	raise service_error(service_error.access, "Access Denied")
	577
	578	# If we know this allocation, reduce the reference counts and
	579	# remove the local allocations. Otherwise report an error. If
	580	# there is an allocation to delete, del_users will be a dictonary
	581	# of sets where the key is the user that owns the keys in the set.
	582	# We use a set to avoid duplicates. del_project is just the name
	583	# of any dynamic project to delete. We're somewhat lazy about
	584	# deleting authorization attributes. Having access to something
	585	# that doesn't exist isn't harmful.
	586	del_users = { }
	587	del_project = None
	588	del_types = set()
	589
	590	self.state_lock.acquire()
	591	if aid in self.allocation:
	592	self.log.debug("Found allocation for %s" %aid)
	593	for k in self.allocation[aid]['keys']:
	594	kk = "%s:%s" % k
	595	self.keys[kk] -= 1
	596	if self.keys[kk] == 0:
	597	if not del_users.has_key(k[0]):
	598	del_users[k[0]] = set()
	599	del_users[k[0]].add(k[1])
	600	del self.keys[kk]
	601
	602	if 'project' in self.allocation[aid]:
	603	pname = self.allocation[aid]['project']
	604	self.projects[pname] -= 1
	605	if self.projects[pname] == 0:
	606	del_project = pname
	607	del self.projects[pname]
	608
	609	if 'types' in self.allocation[aid]:
	610	for t in self.allocation[aid]['types']:
	611	self.types[t] -= 1
	612	if self.types[t] == 0:
	613	if not del_project: del_project = t[0]
	614	del_types.add(t[1])
	615	del self.types[t]
	616
	617	del self.allocation[aid]
	618	self.write_state()
	619	self.state_lock.release()
	620	# If we actually have resources to deallocate, prepare the call.
	621	if del_project or del_users:
	622	self.do_release_project(del_project, del_users, del_types)
	623	# And remove the access cert
	624	cf = "%s/%s.pem" % (self.certdir, aid)
	625	self.log.debug("Removing %s" % cf)
	626	os.remove(cf)
	627	return { 'allocID': req['allocID'] }
	628	else:
	629	self.state_lock.release()
	630	raise service_error(service_error.req, "No such allocation")
[866c983]	631
[06cc65b]	632	# These are subroutines for StartSegment
[8cf2b90e]	633	def generate_ns2(self, topo, expfn, softdir, connInfo):
[06cc65b]	634	"""
	635	Convert topo into an ns2 file, decorated with appropriate commands for
	636	the particular testbed setup. Convert all requests for software, etc
	637	to point at the staged copies on this testbed and add the federation
	638	startcommands.
	639	"""
[617592b]	640	class dragon_commands:
	641	"""
	642	Functor to spit out approrpiate dragon commands for nodes listed in
	643	the connectivity description. The constructor makes a dict mapping
	644	dragon nodes to their parameters and the __call__ checks each
	645	element in turn for membership.
	646	"""
	647	def __init__(self, map):
	648	self.node_info = map
	649
	650	def __call__(self, e):
	651	s = ""
	652	if isinstance(e, topdl.Computer):
[49051fb]	653	if self.node_info.has_key(e.name):
[fefa026]	654	info = self.node_info[e.name]
	655	for ifname, vlan, type in info:
[617592b]	656	for i in e.interface:
	657	if i.name == ifname:
	658	addr = i.get_attribute('ip4_address')
	659	subs = i.substrate[0]
	660	break
	661	else:
	662	raise service_error(service_error.internal,
	663	"No interface %s on element %s" % \
[49051fb]	664	(ifname, e.name))
[1cf8e2c]	665	# XXX: do netmask right
[617592b]	666	if type =='link':
[06cc65b]	667	s = ("tb-allow-external ${%s} " + \
	668	"dragonportal ip %s vlan %s " + \
	669	"netmask 255.255.255.0\n") % \
[e07c8f3]	670	(topdl.to_tcl_name(e.name), addr, vlan)
[617592b]	671	elif type =='lan':
[06cc65b]	672	s = ("tb-allow-external ${%s} " + \
	673	"dragonportal " + \
[617592b]	674	"ip %s vlan %s usurp %s\n") % \
[e07c8f3]	675	(topdl.to_tcl_name(e.name), addr,
	676	vlan, subs)
[617592b]	677	else:
	678	raise service_error(service_error_internal,
	679	"Unknown DRAGON type %s" % type)
	680	return s
	681
	682	class not_dragon:
[06cc65b]	683	"""
	684	Return true if a node is in the given map of dragon nodes.
	685	"""
[617592b]	686	def __init__(self, map):
	687	self.nodes = set(map.keys())
	688
	689	def __call__(self, e):
[49051fb]	690	return e.name not in self.nodes
[69692a9]	691
[06cc65b]	692	# Main line of generate_ns2
[ecca6eb]	693	t = topo.clone()
	694
[06cc65b]	695	# Create the map of nodes that need direct connections (dragon
	696	# connections) from the connInfo
[617592b]	697	dragon_map = { }
	698	for i in [ i for i in connInfo if i['type'] == 'transit']:
	699	for a in i.get('fedAttr', []):
	700	if a['attribute'] == 'vlan_id':
	701	vlan = a['value']
	702	break
	703	else:
[8cf2b90e]	704	raise service_error(service_error.internal, "No vlan tag")
[617592b]	705	members = i.get('member', [])
	706	if len(members) > 1: type = 'lan'
	707	else: type = 'link'
	708
	709	try:
	710	for m in members:
[8cf2b90e]	711	if m['element'] in dragon_map:
[617592b]	712	dragon_map[m['element']].append(( m['interface'],
	713	vlan, type))
	714	else:
	715	dragon_map[m['element']] = [( m['interface'],
	716	vlan, type),]
	717	except KeyError:
	718	raise service_error(service_error.req,
	719	"Missing connectivity info")
	720
[35aa3ae]	721	# Weed out the things we aren't going to instantiate: Segments, portal
	722	# substrates, and portal interfaces. (The copy in the for loop allows
	723	# us to delete from e.elements in side the for loop). While we're
	724	# touching all the elements, we also adjust paths from the original
	725	# testbed to local testbed paths and put the federation commands into
	726	# the start commands
	727	for e in [e for e in t.elements]:
	728	if isinstance(e, topdl.Segment):
	729	t.elements.remove(e)
[43649f1]	730	if isinstance(e, topdl.Computer):
[e76f38a]	731	self.add_kit(e, self.federation_software)
[5e1fb7b]	732	if e.get_attribute('portal') and self.portal_startcommand:
[9b3627e]	733	# Add local portal support software
[e76f38a]	734	self.add_kit(e, self.portal_software)
[43649f1]	735	# Portals never have a user-specified start command
[5e1fb7b]	736	e.set_attribute('startup', self.portal_startcommand)
	737	elif self.node_startcommand:
[43649f1]	738	if e.get_attribute('startup'):
[d87778f]	739	e.set_attribute('startup', "%s \\$USER '%s'" % \
[5e1fb7b]	740	(self.node_startcommand,
	741	e.get_attribute('startup')))
[43649f1]	742	else:
[5e1fb7b]	743	e.set_attribute('startup', self.node_startcommand)
[0297248]	744
[49051fb]	745	dinf = [i[0] for i in dragon_map.get(e.name, []) ]
[35aa3ae]	746	# Remove portal interfaces that do not connect to DRAGON
	747	e.interface = [i for i in e.interface \
[617592b]	748	if not i.get_attribute('portal') or i.name in dinf ]
[9b3627e]	749	# Fix software paths
	750	for s in getattr(e, 'software', []):
	751	s.location = re.sub("^.*/", softdir, s.location)
[35aa3ae]	752
	753	t.substrates = [ s.clone() for s in t.substrates ]
	754	t.incorporate_elements()
[ecca6eb]	755
	756	# Customize the ns2 output for local portal commands and images
	757	filters = []
	758
[5e1fb7b]	759	if self.dragon_endpoint:
[617592b]	760	add_filter = not_dragon(dragon_map)
	761	filters.append(dragon_commands(dragon_map))
[69692a9]	762	else:
	763	add_filter = None
	764
[5e1fb7b]	765	if self.portal_command:
	766	filters.append(topdl.generate_portal_command_filter(
	767	self.portal_command, add_filter=add_filter))
[ecca6eb]	768
[5e1fb7b]	769	if self.portal_image:
[ecca6eb]	770	filters.append(topdl.generate_portal_image_filter(
[5e1fb7b]	771	self.portal_image))
[ecca6eb]	772
[5e1fb7b]	773	if self.portal_type:
[ecca6eb]	774	filters.append(topdl.generate_portal_hardware_filter(
[5e1fb7b]	775	self.portal_type))
[ecca6eb]	776
	777	# Convert to ns and write it out
	778	expfile = topdl.topology_to_ns2(t, filters)
	779	try:
	780	f = open(expfn, "w")
	781	print >>f, expfile
	782	f.close()
[d3c8759]	783	except EnvironmentError:
[ecca6eb]	784	raise service_error(service_error.internal,
	785	"Cannot write experiment file %s: %s" % (expfn,e))
[f9ef40b]	786
[2c1fd21]	787	def export_store_info(self, cf, proj, ename, connInfo):
	788	"""
	789	For the export requests in the connection info, install the peer names
	790	at the experiment controller via SetValue calls.
	791	"""
	792
	793	for c in connInfo:
	794	for p in [ p for p in c.get('parameter', []) \
	795	if p.get('type', '') == 'output']:
	796
	797	if p.get('name', '') == 'peer':
	798	k = p.get('key', None)
	799	surl = p.get('store', None)
	800	if surl and k and k.index('/') != -1:
	801	value = "%s.%s.%s%s" % \
	802	(k[k.index('/')+1:], ename, proj, self.domain)
	803	req = { 'name': k, 'value': value }
	804	self.log.debug("Setting %s to %s on %s" % \
	805	(k, value, surl))
	806	self.call_SetValue(surl, req, cf)
	807	else:
	808	self.log.error("Bad export request: %s" % p)
	809	elif p.get('name', '') == 'ssh_port':
	810	k = p.get('key', None)
	811	surl = p.get('store', None)
	812	if surl and k:
	813	req = { 'name': k, 'value': self.ssh_port }
	814	self.log.debug("Setting %s to %s on %s" % \
	815	(k, self.ssh_port, surl))
	816	self.call_SetValue(surl, req, cf)
	817	else:
	818	self.log.error("Bad export request: %s" % p)
	819	else:
	820	self.log.error("Unknown export parameter: %s" % \
	821	p.get('name'))
	822	continue
	823
[49051fb]	824	def add_seer_node(self, topo, name, startup):
	825	"""
	826	Add a seer node to the given topology, with the startup command passed
[06cc65b]	827	in. Used by configure seer_services.
[49051fb]	828	"""
	829	c_node = topdl.Computer(
	830	name=name,
	831	os= topdl.OperatingSystem(
	832	attribute=[
	833	{ 'attribute': 'osid',
	834	'value': self.local_seer_image },
	835	]),
	836	attribute=[
	837	{ 'attribute': 'startup', 'value': startup },
	838	]
	839	)
	840	self.add_kit(c_node, self.local_seer_software)
	841	topo.elements.append(c_node)
	842
	843	def configure_seer_services(self, services, topo, softdir):
[8cf2b90e]	844	"""
	845	Make changes to the topology required for the seer requests being made.
	846	Specifically, add any control or master nodes required and set up the
	847	start commands on the nodes to interconnect them.
	848	"""
	849	local_seer = False # True if we need to add a control node
	850	collect_seer = False # True if there is a seer-master node
	851	seer_master= False # True if we need to add the seer-master
[49051fb]	852	for s in services:
	853	s_name = s.get('name', '')
	854	s_vis = s.get('visibility','')
	855
[8cf2b90e]	856	if s_name == 'local_seer_control' and s_vis == 'export':
[49051fb]	857	local_seer = True
	858	elif s_name == 'seer_master':
	859	if s_vis == 'import':
	860	collect_seer = True
	861	elif s_vis == 'export':
	862	seer_master = True
	863
	864	# We've got the whole picture now, so add nodes if needed and configure
	865	# them to interconnect properly.
	866	if local_seer or seer_master:
	867	# Copy local seer control node software to the tempdir
	868	for l, f in self.local_seer_software:
	869	base = os.path.basename(f)
	870	copy_file(f, "%s/%s" % (softdir, base))
	871	# If we're collecting seers somewhere the controllers need to talk to
	872	# the master. In testbeds that export the service, that will be a
	873	# local node that we'll add below. Elsewhere it will be the control
	874	# portal that will port forward to the exporting master.
	875	if local_seer:
	876	if collect_seer:
[acaa9b9]	877	startup = "%s -C %s" % (self.local_seer_start, "seer-master")
[49051fb]	878	else:
	879	startup = self.local_seer_start
	880	self.add_seer_node(topo, 'control', startup)
	881	# If this is the seer master, add that node, too.
	882	if seer_master:
[e07c8f3]	883	self.add_seer_node(topo, 'seer-master',
	884	"%s -R -n -R seer-master -R -A -R sink" % \
	885	self.seer_master_start)
[49051fb]	886
[06cc65b]	887	def retrieve_software(self, topo, certfile, softdir):
	888	"""
	889	Collect the software that nodes in the topology need loaded and stage
	890	it locally. This implies retrieving it from the experiment_controller
	891	and placing it into softdir. Certfile is used to prove that this node
	892	has access to that data (it's the allocation/segment fedid). Finally
	893	local portal and federation software is also copied to the same staging
	894	directory for simplicity - all software needed for experiment creation
	895	is in softdir.
	896	"""
	897	sw = set()
	898	for e in topo.elements:
	899	for s in getattr(e, 'software', []):
	900	sw.add(s.location)
	901	for s in sw:
	902	self.log.debug("Retrieving %s" % s)
	903	try:
	904	get_url(s, certfile, softdir)
	905	except:
	906	t, v, st = sys.exc_info()
	907	raise service_error(service_error.internal,
	908	"Error retrieving %s: %s" % (s, v))
[49051fb]	909
[06cc65b]	910	# Copy local federation and portal node software to the tempdir
	911	for s in (self.federation_software, self.portal_software):
	912	for l, f in s:
	913	base = os.path.basename(f)
	914	copy_file(f, "%s/%s" % (softdir, base))
[49051fb]	915
[6c57fe9]	916
[06cc65b]	917	def initialize_experiment_info(self, attrs, aid, certfile, tmpdir):
	918	"""
	919	Gather common configuration files, retrieve or create an experiment
	920	name and project name, and return the ssh_key filenames. Create an
	921	allocation log bound to the state log variable as well.
	922	"""
[6c57fe9]	923	configs = set(('hosts', 'ssh_pubkey', 'ssh_secretkey'))
[06cc65b]	924	ename = None
	925	pubkey_base = None
	926	secretkey_base = None
	927	proj = None
	928	user = None
	929	alloc_log = None
	930
	931	for a in attrs:
	932	if a['attribute'] in configs:
	933	try:
	934	self.log.debug("Retrieving %s from %s" % \
	935	(a['attribute'], a['value']))
	936	get_url(a['value'], certfile, tmpdir)
	937	except:
	938	t, v, st = sys.exc_info()
	939	raise service_error(service_error.internal,
	940	"Error retrieving %s: %s" % (a.get('value', ""), v))
	941	if a['attribute'] == 'ssh_pubkey':
	942	pubkey_base = a['value'].rpartition('/')[2]
	943	if a['attribute'] == 'ssh_secretkey':
	944	secretkey_base = a['value'].rpartition('/')[2]
	945	if a['attribute'] == 'experiment_name':
	946	ename = a['value']
	947
	948	if not ename:
	949	ename = ""
	950	for i in range(0,5):
	951	ename += random.choice(string.ascii_letters)
	952	self.log.warn("No experiment name: picked one randomly: %s" \
	953	% ename)
	954
	955	if not pubkey_base:
	956	raise service_error(service_error.req,
	957	"No public key attribute")
	958
	959	if not secretkey_base:
	960	raise service_error(service_error.req,
	961	"No secret key attribute")
[6c57fe9]	962
[06cc65b]	963	self.state_lock.acquire()
	964	if aid in self.allocation:
	965	proj = self.allocation[aid].get('project', None)
	966	if not proj:
	967	proj = self.allocation[aid].get('sproject', None)
	968	user = self.allocation[aid].get('user', None)
	969	self.allocation[aid]['experiment'] = ename
	970	self.allocation[aid]['log'] = [ ]
	971	# Create a logger that logs to the experiment's state object as
	972	# well as to the main log file.
	973	alloc_log = logging.getLogger('fedd.access.%s' % ename)
	974	h = logging.StreamHandler(
	975	list_log.list_log(self.allocation[aid]['log']))
	976	# XXX: there should be a global one of these rather than
	977	# repeating the code.
	978	h.setFormatter(logging.Formatter(
	979	"%(asctime)s %(name)s %(message)s",
	980	'%d %b %y %H:%M:%S'))
	981	alloc_log.addHandler(h)
	982	self.write_state()
	983	self.state_lock.release()
	984
	985	if not proj:
	986	raise service_error(service_error.internal,
	987	"Can't find project for %s" %aid)
	988
	989	if not user:
	990	raise service_error(service_error.internal,
	991	"Can't find creation user for %s" %aid)
	992
	993	return (ename, proj, user, pubkey_base, secretkey_base, alloc_log)
	994
	995	def finalize_experiment(self, starter, topo, aid, alloc_id):
	996	"""
	997	Store key bits of experiment state in the global repository, including
	998	the response that may need to be replayed, and return the response.
	999	"""
	1000	# Copy the assigned names into the return topology
	1001	embedding = [ ]
	1002	for n in starter.node:
	1003	embedding.append({
	1004	'toponame': n,
	1005	'physname': ["%s%s" % (starter.node[n], self.domain)],
	1006	})
	1007	# Grab the log (this is some anal locking, but better safe than
	1008	# sorry)
	1009	self.state_lock.acquire()
	1010	logv = "".join(self.allocation[aid]['log'])
	1011	# It's possible that the StartSegment call gets retried (!).
	1012	# if the 'started' key is in the allocation, we'll return it rather
	1013	# than redo the setup.
	1014	self.allocation[aid]['started'] = {
	1015	'allocID': alloc_id,
	1016	'allocationLog': logv,
	1017	'segmentdescription': {
	1018	'topdldescription': topo.clone().to_dict()
	1019	},
	1020	'embedding': embedding
	1021	}
	1022	retval = copy.copy(self.allocation[aid]['started'])
	1023	self.write_state()
	1024	self.state_lock.release()
	1025	return retval
	1026
	1027	# End of StartSegment support routines
	1028
	1029	def StartSegment(self, req, fid):
[b770aa0]	1030	err = None # Any service_error generated after tmpdir is created
	1031	rv = None # Return value from segment creation
	1032
[cc8d8e9]	1033	try:
	1034	req = req['StartSegmentRequestBody']
[06cc65b]	1035	auth_attr = req['allocID']['fedid']
	1036	topref = req['segmentdescription']['topdldescription']
[cc8d8e9]	1037	except KeyError:
	1038	raise service_error(server_error.req, "Badly formed request")
[ecca6eb]	1039
[e02cd14]	1040	connInfo = req.get('connection', [])
	1041	services = req.get('service', [])
[ecca6eb]	1042	aid = "%s" % auth_attr
[6c57fe9]	1043	attrs = req.get('fedAttr', [])
[ecca6eb]	1044	if not self.auth.check_attribute(fid, auth_attr):
	1045	raise service_error(service_error.access, "Access denied")
[cd06678]	1046	else:
	1047	# See if this is a replay of an earlier succeeded StartSegment -
	1048	# sometimes SSL kills 'em. If so, replay the response rather than
	1049	# redoing the allocation.
	1050	self.state_lock.acquire()
	1051	retval = self.allocation[aid].get('started', None)
	1052	self.state_lock.release()
	1053	if retval:
	1054	self.log.warning("Duplicate StartSegment for %s: " % aid + \
	1055	"replaying response")
	1056	return retval
	1057
	1058	# A new request. Do it.
[6c57fe9]	1059
[06cc65b]	1060	if topref: topo = topdl.Topology(**topref)
[6c57fe9]	1061	else:
	1062	raise service_error(service_error.req,
	1063	"Request missing segmentdescription'")
[2761484]	1064
[6c57fe9]	1065	certfile = "%s/%s.pem" % (self.certdir, auth_attr)
	1066	try:
	1067	tmpdir = tempfile.mkdtemp(prefix="access-")
[ecca6eb]	1068	softdir = "%s/software" % tmpdir
[c23684d]	1069	os.mkdir(softdir)
[d3c8759]	1070	except EnvironmentError:
[6c57fe9]	1071	raise service_error(service_error.internal, "Cannot create tmp dir")
	1072
[b770aa0]	1073	# Try block alllows us to clean up temporary files.
	1074	try:
[06cc65b]	1075	self.retrieve_software(topo, certfile, softdir)
	1076	ename, proj, user, pubkey_base, secretkey_base, alloc_log = \
	1077	self.initialize_experiment_info(attrs, aid,
	1078	certfile, tmpdir)
[9b3627e]	1079
[06cc65b]	1080	# Set up userconf and seer if needed
[c200d36]	1081	self.configure_userconf(services, tmpdir)
[49051fb]	1082	self.configure_seer_services(services, topo, softdir)
[06cc65b]	1083	# Get and send synch store variables
[2761484]	1084	self.export_store_info(certfile, proj, ename, connInfo)
	1085	self.import_store_info(certfile, connInfo)
	1086
[b770aa0]	1087	expfile = "%s/experiment.tcl" % tmpdir
	1088
	1089	self.generate_portal_configs(topo, pubkey_base,
[06cc65b]	1090	secretkey_base, tmpdir, proj, ename, connInfo, services)
[b770aa0]	1091	self.generate_ns2(topo, expfile,
[8cf2b90e]	1092	"/proj/%s/software/%s/" % (proj, ename), connInfo)
[f07fa49]	1093
[b770aa0]	1094	starter = self.start_segment(keyfile=self.ssh_privkey_file,
[181aeb4]	1095	debug=self.create_debug, log=alloc_log, boss=self.boss,
	1096	cert=self.xmlrpc_cert)
[b770aa0]	1097	rv = starter(self, ename, proj, user, expfile, tmpdir)
	1098	except service_error, e:
	1099	err = e
[06cc65b]	1100	except:
	1101	t, v, st = sys.exc_info()
	1102	err = service_error(service_error.internal, "%s: %s" % \
	1103	(v, traceback.extract_tb(st)))
[b770aa0]	1104
[574055e]	1105	# Walk up tmpdir, deleting as we go
[06cc65b]	1106	if self.cleanup: self.remove_dirs(tmpdir)
	1107	else: self.log.debug("[StartSegment]: not removing %s" % tmpdir)
[574055e]	1108
[fd556d1]	1109	if rv:
[06cc65b]	1110	return self.finalize_experiment(starter, topo, aid, req['allocID'])
[b770aa0]	1111	elif err:
	1112	raise service_error(service_error.federant,
	1113	"Swapin failed: %s" % err)
[fd556d1]	1114	else:
	1115	raise service_error(service_error.federant, "Swapin failed")
[5ae3857]	1116
	1117	def TerminateSegment(self, req, fid):
	1118	try:
	1119	req = req['TerminateSegmentRequestBody']
	1120	except KeyError:
	1121	raise service_error(server_error.req, "Badly formed request")
	1122
	1123	auth_attr = req['allocID']['fedid']
	1124	aid = "%s" % auth_attr
	1125	attrs = req.get('fedAttr', [])
	1126	if not self.auth.check_attribute(fid, auth_attr):
	1127	raise service_error(service_error.access, "Access denied")
	1128
	1129	self.state_lock.acquire()
[06cc65b]	1130	if aid in self.allocation:
[5ae3857]	1131	proj = self.allocation[aid].get('project', None)
	1132	if not proj:
	1133	proj = self.allocation[aid].get('sproject', None)
	1134	user = self.allocation[aid].get('user', None)
	1135	ename = self.allocation[aid].get('experiment', None)
[1d913e13]	1136	else:
	1137	proj = None
	1138	user = None
	1139	ename = None
[5ae3857]	1140	self.state_lock.release()
	1141
	1142	if not proj:
	1143	raise service_error(service_error.internal,
	1144	"Can't find project for %s" % aid)
	1145
	1146	if not user:
	1147	raise service_error(service_error.internal,
	1148	"Can't find creation user for %s" % aid)
	1149	if not ename:
	1150	raise service_error(service_error.internal,
	1151	"Can't find experiment name for %s" % aid)
[fd556d1]	1152	stopper = self.stop_segment(keyfile=self.ssh_privkey_file,
[181aeb4]	1153	debug=self.create_debug, boss=self.boss, cert=self.xmlrpc_cert)
[5ae3857]	1154	stopper(self, user, proj, ename)
	1155	return { 'allocID': req['allocID'] }

Note: See TracBrowser for help on using the repository browser.

Download in other formats: