Context Navigation

source: fedd/federation/emulab_access.py @ 968b84b

Last change on this file since 968b84b was 0b217d1, checked in by Ted Faber <faber@…>, 10 years ago
Strip some dead wood
Property mode set to `100644`
File size: 40.5 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
[9c3e77f]	13	import socket
[19cc408]	14
[f8582c9]	15	from threading import *
[8e6fe4d]	16	from M2Crypto.SSL import SSLError
[f8582c9]	17
[f771e2f]	18	from access import access_base
	19
[ec4fb42]	20	from util import *
[6bedbdba]	21	from deter import fedid, generate_fedid
[6e63513]	22	from authorizer import authorizer, abac_authorizer
[6a0c9f4]	23	from service_error import service_error
[9460b1e]	24	from remote_service import xmlrpc_handler, soap_handler, service_caller
[e83f2f2]	25	from proof import proof as access_proof
[11a08b0]	26
[6c57fe9]	27	import httplib
	28	import tempfile
	29	from urlparse import urlparse
	30
[6bedbdba]	31	from deter import topdl
[11860f52]	32	import list_log
[06c1dba]	33	import emulab_segment
[8fbef04]	34	import emulab_shared_nat_segment
[11860f52]	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
[ee950c2]	44	class access(access_base):
[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
[53b5c18]	52	max_name_len = 19
	53
[3f6bc5f]	54	def __init__(self, config=None, auth=None):
[866c983]	55	"""
	56	Initializer. Pulls parameters out of the ConfigParser's access section.
	57	"""
	58
[f771e2f]	59	access_base.__init__(self, config, auth)
[866c983]	60
[53b5c18]	61	self.max_name_len = access.max_name_len
	62
[866c983]	63	self.allow_proxy = config.getboolean("access", "allow_proxy")
	64
	65	self.domain = config.get("access", "domain")
[eeb0088]	66	self.userconfdir = config.get("access","userconfdir")
	67	self.userconfcmd = config.get("access","userconfcmd")
[fe28bb2]	68	self.userconfurl = config.get("access","userconfurl")
[9b3627e]	69	self.federation_software = config.get("access", "federation_software")
	70	self.portal_software = config.get("access", "portal_software")
[e76f38a]	71	self.local_seer_software = config.get("access", "local_seer_software")
	72	self.local_seer_image = config.get("access", "local_seer_image")
	73	self.local_seer_start = config.get("access", "local_seer_start")
[49051fb]	74	self.seer_master_start = config.get("access", "seer_master_start")
[ecca6eb]	75	self.ssh_privkey_file = config.get("access","ssh_privkey_file")
[3bddd24]	76	self.ssh_pubkey_file = config.get("access","ssh_pubkey_file")
[6280b1f]	77	self.ssh_port = config.get("access","ssh_port") or "22"
[181aeb4]	78	self.boss = config.get("access", "boss")
[814b5e5]	79	self.ops = config.get("access", "ops")
[0b217d1]	80	# XXX: Delete these
	81	# self.xmlrpc_cert = config.get("access", "xmlrpc_cert")
	82	# self.xmlrpc_certpw = config.get("access", "xmlrpc_certpw")
[5e1fb7b]	83
	84	self.dragon_endpoint = config.get("access", "dragon")
	85	self.dragon_vlans = config.get("access", "dragon_vlans")
	86	self.deter_internal = config.get("access", "deter_internal")
	87
	88	self.tunnel_config = config.getboolean("access", "tunnel_config")
	89	self.portal_command = config.get("access", "portal_command")
	90	self.portal_image = config.get("access", "portal_image")
	91	self.portal_type = config.get("access", "portal_type") or "pc"
	92	self.portal_startcommand = config.get("access", "portal_startcommand")
	93	self.node_startcommand = config.get("access", "node_startcommand")
[4ffa6f8]	94	self.nat_portal = config.get("access", "nat_portal")
[8fbef04]	95	self.shared_nat = config.getboolean("access", "shared_nat")
[5e1fb7b]	96
[9c3e77f]	97	self.uri = 'https://%s:%d' % (socket.getfqdn(),
	98	self.get_port(config.get("globals", "services", "23235")))
	99
[f771e2f]	100	self.federation_software = self.software_list(self.federation_software)
	101	self.portal_software = self.software_list(self.portal_software)
	102	self.local_seer_software = self.software_list(self.local_seer_software)
[11860f52]	103
	104	self.access_type = self.access_type.lower()
[8fbef04]	105	if self.shared_nat:
	106	self.start_segment = emulab_shared_nat_segment.start_segment
	107	self.stop_segment = emulab_shared_nat_segment.stop_segment
	108	self.info_segment = emulab_shared_nat_segment.info_segment
	109	self.operation_segment = emulab_shared_nat_segment.operation_segment
	110	self.exports_segment = emulab_shared_nat_segment.exports_segment
	111	else:
	112	self.start_segment = emulab_segment.start_segment
	113	self.stop_segment = emulab_segment.stop_segment
	114	self.info_segment = emulab_segment.info_segment
	115	self.operation_segment = emulab_segment.operation_segment
	116	self.exports_segment = emulab_segment.exports_segment
[866c983]	117
	118	self.restricted = [ ]
[0b217d1]	119	# XXX
	120	#tb = config.get('access', 'testbed')
	121	#if tb: self.testbed = [ t.strip() for t in tb.split(',') ]
	122	#else: self.testbed = [ ]
[866c983]	123
[6e63513]	124	# authorization information
	125	self.auth_type = config.get('access', 'auth_type') \
[ee950c2]	126	or 'abac'
[6e63513]	127	self.auth_dir = config.get('access', 'auth_dir')
	128	accessdb = config.get("access", "accessdb")
	129	# initialize the authorization system
[ee950c2]	130	if self.auth_type == 'abac':
[6e63513]	131	self.auth = abac_authorizer(load=self.auth_dir)
[c002cb2]	132	self.access = [ ]
[6e63513]	133	if accessdb:
[78f2668]	134	self.read_access(accessdb, self.access_tuple)
[6e63513]	135	else:
	136	raise service_error(service_error.internal,
	137	"Unknown auth_type: %s" % self.auth_type)
[f771e2f]	138
[06cc65b]	139	# read_state in the base_class
[f771e2f]	140	self.state_lock.acquire()
[ee950c2]	141	if 'allocation' not in self.state: self.state['allocation']= { }
[f771e2f]	142	self.allocation = self.state['allocation']
	143	self.state_lock.release()
[a20a20f]	144	self.exports = {
	145	'SMB': self.export_SMB,
	146	'seer': self.export_seer,
	147	'tmcd': self.export_tmcd,
	148	'userconfig': self.export_userconfig,
	149	'project_export': self.export_project_export,
	150	'local_seer_control': self.export_local_seer,
	151	'seer_master': self.export_seer_master,
	152	'hide_hosts': self.export_hide_hosts,
	153	}
	154
	155	if not self.local_seer_image or not self.local_seer_software or \
	156	not self.local_seer_start:
	157	if 'local_seer_control' in self.exports:
	158	del self.exports['local_seer_control']
	159
	160	if not self.local_seer_image or not self.local_seer_software or \
	161	not self.seer_master_start:
	162	if 'seer_master' in self.exports:
	163	del self.exports['seer_master']
[866c983]	164
	165
	166	self.soap_services = {\
	167	'RequestAccess': soap_handler("RequestAccess", self.RequestAccess),
	168	'ReleaseAccess': soap_handler("ReleaseAccess", self.ReleaseAccess),
[cc8d8e9]	169	'StartSegment': soap_handler("StartSegment", self.StartSegment),
[e76f38a]	170	'TerminateSegment': soap_handler("TerminateSegment",
	171	self.TerminateSegment),
[6e33086]	172	'InfoSegment': soap_handler("InfoSegment", self.InfoSegment),
[b709861]	173	'OperationSegment': soap_handler("OperationSegment",
	174	self.OperationSegment),
[866c983]	175	}
	176	self.xmlrpc_services = {\
	177	'RequestAccess': xmlrpc_handler('RequestAccess',
	178	self.RequestAccess),
	179	'ReleaseAccess': xmlrpc_handler('ReleaseAccess',
	180	self.ReleaseAccess),
[5ae3857]	181	'StartSegment': xmlrpc_handler("StartSegment", self.StartSegment),
	182	'TerminateSegment': xmlrpc_handler('TerminateSegment',
	183	self.TerminateSegment),
[6e33086]	184	'InfoSegment': xmlrpc_handler("InfoSegment", self.InfoSegment),
[b709861]	185	'OperationSegment': xmlrpc_handler("OperationSegment",
	186	self.OperationSegment),
[866c983]	187	}
	188
[68f03a6]	189	self.call_SetValue = service_caller('SetValue', log=self.log)
[2ee4226]	190	self.call_GetValue = service_caller('GetValue', log=self.log)
[866c983]	191
[9c3e77f]	192	@staticmethod
	193	def get_port(ps):
	194	'''
	195	Take a fedd service string and return the first port. Used in
	196	creating the testbed uri identifier.
	197	'''
	198	p = ps.split(',')
	199	smallport = p[0].split(':')
	200	try:
	201	rv = int(smallport[0])
	202	except ValueError:
	203	rv = 23235
	204	return rv
	205
[6e63513]	206	@staticmethod
[78f2668]	207	def access_tuple(str):
[6e63513]	208	"""
[f77a256]	209	Convert a string of the form (id, id, id) into an access_project. This
	210	is called by read_access to convert to local attributes. It returns a
	211	tuple of the form (project, user, certificate_file).
[6e63513]	212	"""
	213
	214	str = str.strip()
[f77a256]	215	if str.startswith('(') and str.endswith(')') and str.count(',') == 2:
[725c55d]	216	# The slice takes the parens off the string.
[f77a256]	217	proj, user, cert = str[1:-1].split(',')
	218	return (proj.strip(), user.strip(), cert.strip())
[6e63513]	219	else:
	220	raise self.parse_error(
[f77a256]	221	'Bad mapping (unbalanced parens or more than 2 commas)')
[6e63513]	222
[06cc65b]	223	# RequestAccess support routines
	224
[f77a256]	225	def save_project_state(self, aid, pname, uname, certf, owners):
[f771e2f]	226	"""
[ee950c2]	227	Save the project, user, and owners associated with this allocation.
	228	This creates the allocation entry.
[f771e2f]	229	"""
	230	self.state_lock.acquire()
	231	self.allocation[aid] = { }
[ee950c2]	232	self.allocation[aid]['project'] = pname
	233	self.allocation[aid]['user'] = uname
[f77a256]	234	self.allocation[aid]['cert'] = certf
[f771e2f]	235	self.allocation[aid]['owners'] = owners
	236	self.write_state()
	237	self.state_lock.release()
	238	return (pname, uname)
[19cc408]	239
[06cc65b]	240	# End of RequestAccess support routines
	241
[19cc408]	242	def RequestAccess(self, req, fid):
[866c983]	243	"""
	244	Handle the access request. Proxy if not for us.
	245
	246	Parse out the fields and make the allocations or rejections if for us,
	247	otherwise, assuming we're willing to proxy, proxy the request out.
	248	"""
	249
	250	def gateway_hardware(h):
[5e1fb7b]	251	if h == 'GWTYPE': return self.portal_type or 'GWTYPE'
[866c983]	252	else: return h
	253
[43197eb]	254	def get_export_project(svcs):
	255	"""
	256	if the service requests includes one to export a project, return
	257	that project.
	258	"""
	259	rv = None
	260	for s in svcs:
	261	if s.get('name', '') == 'project_export' and \
	262	s.get('visibility', '') == 'export':
	263	if not rv:
[1f6a573]	264	for a in s.get('fedAttr', []):
[43197eb]	265	if a.get('attribute', '') == 'project' \
	266	and 'value' in a:
	267	rv = a['value']
	268	else:
	269	raise service_error(service_error, access,
	270	'Requesting multiple project exports is ' + \
	271	'not supported');
	272	return rv
	273
[8cab4c2]	274	self.log.info("RequestAccess called by %s" % fid)
[866c983]	275	# The dance to get into the request body
	276	if req.has_key('RequestAccessRequestBody'):
	277	req = req['RequestAccessRequestBody']
	278	else:
	279	raise service_error(service_error.req, "No request!?")
	280
[1f6a573]	281	# if this includes a project export request, construct a filter such
	282	# that only the ABAC attributes mapped to that project are checked for
	283	# access.
	284	if 'service' in req:
	285	ep = get_export_project(req['service'])
[de86b35]	286	if ep: pf = lambda(a): a.value[0] == ep
	287	else: pf = None
[1f6a573]	288	else:
	289	ep = None
	290	pf = None
	291
[c573278]	292	if self.auth.import_credentials(
	293	data_list=req.get('abac_credential', [])):
	294	self.auth.save()
[cde9b98]	295	else:
	296	self.log.debug('failed to import incoming credentials')
[866c983]	297
[ee950c2]	298	if self.auth_type == 'abac':
	299	found, owners, proof = self.lookup_access(req, fid, filter=pf)
[6e63513]	300	else:
	301	raise service_error(service_error.internal,
	302	'Unknown auth_type: %s' % self.auth_type)
[f771e2f]	303	ap = None
[866c983]	304
[f771e2f]	305	# keep track of what's been added
	306	allocID, alloc_cert = generate_fedid(subj="alloc", log=self.log)
	307	aid = unicode(allocID)
	308
[f77a256]	309	pname, uname = self.save_project_state(aid, found[0], found[1],
	310	found[2], owners)
[f771e2f]	311
	312	services, svc_state = self.export_services(req.get('service',[]),
	313	pname, uname)
	314	self.state_lock.acquire()
	315	# Store services state in global state
	316	for k, v in svc_state.items():
	317	self.allocation[aid][k] = v
[c65b7e4]	318	self.append_allocation_authorization(aid,
	319	set([(o, allocID) for o in owners]), state_attr='allocation')
[f771e2f]	320	self.write_state()
	321	self.state_lock.release()
	322	try:
	323	f = open("%s/%s.pem" % (self.certdir, aid), "w")
	324	print >>f, alloc_cert
	325	f.close()
	326	except EnvironmentError, e:
[8cab4c2]	327	self.log.info("RequestAccess failed for by %s: internal error" \
	328	% fid)
[f771e2f]	329	raise service_error(service_error.internal,
	330	"Can't open %s/%s : %s" % (self.certdir, aid, e))
[8cab4c2]	331	self.log.debug('RequestAccess Returning allocation ID: %s' % allocID)
[f771e2f]	332	resp = self.build_access_response({ 'fedid': allocID } ,
[ee950c2]	333	pname, services, proof)
[f771e2f]	334	return resp
[d81971a]	335
	336	def ReleaseAccess(self, req, fid):
[8cab4c2]	337	self.log.info("ReleaseAccess called by %s" % fid)
[866c983]	338	# The dance to get into the request body
	339	if req.has_key('ReleaseAccessRequestBody'):
	340	req = req['ReleaseAccessRequestBody']
	341	else:
	342	raise service_error(service_error.req, "No request!?")
	343
[8cf2b90e]	344	try:
	345	if req['allocID'].has_key('localname'):
	346	auth_attr = aid = req['allocID']['localname']
	347	elif req['allocID'].has_key('fedid'):
	348	aid = unicode(req['allocID']['fedid'])
	349	auth_attr = req['allocID']['fedid']
	350	else:
	351	raise service_error(service_error.req,
	352	"Only localnames and fedids are understood")
	353	except KeyError:
	354	raise service_error(service_error.req, "Badly formed request")
	355
[725c55d]	356	self.log.debug("[access] deallocation requested for %s by %s" % \
	357	(aid, fid))
[e83f2f2]	358	access_ok, proof = self.auth.check_attribute(fid, auth_attr,
	359	with_proof=True)
	360	if not access_ok:
[8cf2b90e]	361	self.log.debug("[access] deallocation denied for %s", aid)
	362	raise service_error(service_error.access, "Access Denied")
	363
	364	self.state_lock.acquire()
	365	if aid in self.allocation:
	366	self.log.debug("Found allocation for %s" %aid)
[c65b7e4]	367	self.clear_allocation_authorization(aid, state_attr='allocation')
[8cf2b90e]	368	del self.allocation[aid]
	369	self.write_state()
	370	self.state_lock.release()
[ee950c2]	371	# Remove the access cert
[8cf2b90e]	372	cf = "%s/%s.pem" % (self.certdir, aid)
	373	self.log.debug("Removing %s" % cf)
	374	os.remove(cf)
[8cab4c2]	375	self.log.info("ReleaseAccess succeeded for %s" % fid)
[e83f2f2]	376	return { 'allocID': req['allocID'], 'proof': proof.to_dict() }
[8cf2b90e]	377	else:
	378	self.state_lock.release()
	379	raise service_error(service_error.req, "No such allocation")
[866c983]	380
[06cc65b]	381	# These are subroutines for StartSegment
[8cf2b90e]	382	def generate_ns2(self, topo, expfn, softdir, connInfo):
[06cc65b]	383	"""
	384	Convert topo into an ns2 file, decorated with appropriate commands for
	385	the particular testbed setup. Convert all requests for software, etc
	386	to point at the staged copies on this testbed and add the federation
	387	startcommands.
	388	"""
[617592b]	389	class dragon_commands:
	390	"""
	391	Functor to spit out approrpiate dragon commands for nodes listed in
	392	the connectivity description. The constructor makes a dict mapping
	393	dragon nodes to their parameters and the __call__ checks each
	394	element in turn for membership.
	395	"""
	396	def __init__(self, map):
	397	self.node_info = map
	398
	399	def __call__(self, e):
	400	s = ""
	401	if isinstance(e, topdl.Computer):
[49051fb]	402	if self.node_info.has_key(e.name):
[fefa026]	403	info = self.node_info[e.name]
	404	for ifname, vlan, type in info:
[617592b]	405	for i in e.interface:
	406	if i.name == ifname:
	407	addr = i.get_attribute('ip4_address')
	408	subs = i.substrate[0]
	409	break
	410	else:
	411	raise service_error(service_error.internal,
	412	"No interface %s on element %s" % \
[49051fb]	413	(ifname, e.name))
[1cf8e2c]	414	# XXX: do netmask right
[617592b]	415	if type =='link':
[06cc65b]	416	s = ("tb-allow-external ${%s} " + \
	417	"dragonportal ip %s vlan %s " + \
	418	"netmask 255.255.255.0\n") % \
[e07c8f3]	419	(topdl.to_tcl_name(e.name), addr, vlan)
[617592b]	420	elif type =='lan':
[06cc65b]	421	s = ("tb-allow-external ${%s} " + \
	422	"dragonportal " + \
[617592b]	423	"ip %s vlan %s usurp %s\n") % \
[e07c8f3]	424	(topdl.to_tcl_name(e.name), addr,
	425	vlan, subs)
[617592b]	426	else:
	427	raise service_error(service_error_internal,
	428	"Unknown DRAGON type %s" % type)
	429	return s
	430
	431	class not_dragon:
[06cc65b]	432	"""
	433	Return true if a node is in the given map of dragon nodes.
	434	"""
[617592b]	435	def __init__(self, map):
	436	self.nodes = set(map.keys())
	437
	438	def __call__(self, e):
[49051fb]	439	return e.name not in self.nodes
[69692a9]	440
[4d68ba6]	441	def have_portals(top):
	442	'''
	443	Return true if the topology has a portal node
	444	'''
	445	# The else is on the for
	446	for e in top.elements:
	447	if isinstance(e, topdl.Computer) and e.get_attribute('portal'):
	448	return True
	449	else:
	450	return False
	451
	452
[06cc65b]	453	# Main line of generate_ns2
[ecca6eb]	454	t = topo.clone()
	455
[06cc65b]	456	# Create the map of nodes that need direct connections (dragon
	457	# connections) from the connInfo
[617592b]	458	dragon_map = { }
	459	for i in [ i for i in connInfo if i['type'] == 'transit']:
	460	for a in i.get('fedAttr', []):
	461	if a['attribute'] == 'vlan_id':
	462	vlan = a['value']
	463	break
	464	else:
[8cf2b90e]	465	raise service_error(service_error.internal, "No vlan tag")
[617592b]	466	members = i.get('member', [])
	467	if len(members) > 1: type = 'lan'
	468	else: type = 'link'
	469
	470	try:
	471	for m in members:
[8cf2b90e]	472	if m['element'] in dragon_map:
[617592b]	473	dragon_map[m['element']].append(( m['interface'],
	474	vlan, type))
	475	else:
	476	dragon_map[m['element']] = [( m['interface'],
	477	vlan, type),]
	478	except KeyError:
	479	raise service_error(service_error.req,
	480	"Missing connectivity info")
	481
[319fb0a]	482	def output_fixed_filter(e):
	483	if not isinstance(e, topdl.Computer): return ""
	484	fn = e.get_attribute('fixed')
	485	if fn is None:
	486	return ""
	487	else:
[f725eda]	488	return 'tb-fix-node ${%s} %s\n' % (topdl.to_tcl_name(e.name), fn)
	489
	490	def output_external_access(e):
	491	'''
	492	Allow nodes that have explicit external access requests attached to
	493	have external access. This will be replaced by a real risky
	494	experiment filter.
	495	'''
	496	if not isinstance(e, topdl.Computer): return ""
	497	ex = e.get_attribute('containers:external')
	498	if ex is None:
	499	return ""
	500	else:
	501	return 'tb-allow-external ${%s}\n' % topdl.to_tcl_name(e.name)
[319fb0a]	502
[35aa3ae]	503	# Weed out the things we aren't going to instantiate: Segments, portal
	504	# substrates, and portal interfaces. (The copy in the for loop allows
	505	# us to delete from e.elements in side the for loop). While we're
	506	# touching all the elements, we also adjust paths from the original
	507	# testbed to local testbed paths and put the federation commands into
	508	# the start commands
[4d68ba6]	509	local = len(dragon_map) == 0 and not have_portals(t)
	510	if local: routing = 'Static'
	511	else: routing = 'Manual'
	512
[aada25c]	513	if local:
	514	self.log.debug("Local experiment.")
[35aa3ae]	515	for e in [e for e in t.elements]:
	516	if isinstance(e, topdl.Segment):
	517	t.elements.remove(e)
[43649f1]	518	if isinstance(e, topdl.Computer):
[e76f38a]	519	self.add_kit(e, self.federation_software)
[5e1fb7b]	520	if e.get_attribute('portal') and self.portal_startcommand:
[9b3627e]	521	# Add local portal support software
[e76f38a]	522	self.add_kit(e, self.portal_software)
[43649f1]	523	# Portals never have a user-specified start command
[5e1fb7b]	524	e.set_attribute('startup', self.portal_startcommand)
[4d68ba6]	525	elif not local and self.node_startcommand:
[43649f1]	526	if e.get_attribute('startup'):
[d87778f]	527	e.set_attribute('startup', "%s \\$USER '%s'" % \
[5e1fb7b]	528	(self.node_startcommand,
	529	e.get_attribute('startup')))
[43649f1]	530	else:
[5e1fb7b]	531	e.set_attribute('startup', self.node_startcommand)
[0297248]	532
[49051fb]	533	dinf = [i[0] for i in dragon_map.get(e.name, []) ]
[35aa3ae]	534	# Remove portal interfaces that do not connect to DRAGON
	535	e.interface = [i for i in e.interface \
[617592b]	536	if not i.get_attribute('portal') or i.name in dinf ]
[9b3627e]	537	# Fix software paths
	538	for s in getattr(e, 'software', []):
	539	s.location = re.sub("^.*/", softdir, s.location)
[35aa3ae]	540
	541	t.substrates = [ s.clone() for s in t.substrates ]
	542	t.incorporate_elements()
[ecca6eb]	543
	544	# Customize the ns2 output for local portal commands and images
[f725eda]	545	filters = [output_fixed_filter, output_external_access]
[ecca6eb]	546
[5e1fb7b]	547	if self.dragon_endpoint:
[617592b]	548	add_filter = not_dragon(dragon_map)
	549	filters.append(dragon_commands(dragon_map))
[69692a9]	550	else:
	551	add_filter = None
	552
[5e1fb7b]	553	if self.portal_command:
[8fbef04]	554	if self.shared_nat: suffix = 'shared nat 0/0/tcp rdr 22/tcp'
	555	else: suffix = ''
	556
[5e1fb7b]	557	filters.append(topdl.generate_portal_command_filter(
[8fbef04]	558	self.portal_command, add_filter=add_filter, suffix=suffix))
[ecca6eb]	559
[5e1fb7b]	560	if self.portal_image:
[ecca6eb]	561	filters.append(topdl.generate_portal_image_filter(
[5e1fb7b]	562	self.portal_image))
[ecca6eb]	563
[5e1fb7b]	564	if self.portal_type:
[ecca6eb]	565	filters.append(topdl.generate_portal_hardware_filter(
[5e1fb7b]	566	self.portal_type))
[ecca6eb]	567
	568	# Convert to ns and write it out
[4d68ba6]	569	expfile = topdl.topology_to_ns2(t, filters, routing=routing)
[ecca6eb]	570	try:
	571	f = open(expfn, "w")
	572	print >>f, expfile
	573	f.close()
[d3c8759]	574	except EnvironmentError:
[ecca6eb]	575	raise service_error(service_error.internal,
	576	"Cannot write experiment file %s: %s" % (expfn,e))
[f9ef40b]	577
[2c1fd21]	578	def export_store_info(self, cf, proj, ename, connInfo):
	579	"""
[26821ac]	580	For the export requests in the connection info, send out teh SetValue
	581	calls for parameters initialized by the segment's export_segment
	582	object. Values are all in the connInfo parameter's value fields.
[2c1fd21]	583	"""
	584
	585	for c in connInfo:
[26821ac]	586	for p in c.get('parameter', []):
	587	# Ignore non-output parameters
	588	if p.get('type', '') != 'output':
	589	continue
	590	# Sanity check store, key and value
	591	surl = p.get('store', None)
	592	key = p.get('key', None)
	593	value = p.get('value', None)
	594	if surl is None:
	595	self.log.error("Export parameter without store: %s" % \
	596	p.get('name'))
	597	continue
	598	if key is None:
	599	self.log.error("Export parameter without key: %s" % \
	600	p.get('name'))
	601	continue
	602	if value is None:
	603	self.log.error("Unknown (unset) export parameter: %s" % \
[2c1fd21]	604	p.get('name'))
	605	continue
[26821ac]	606	# Set the store value
	607	req = { 'name': key, 'value': value }
	608	self.call_SetValue(surl, req, cf)
[2c1fd21]	609
[49051fb]	610	def add_seer_node(self, topo, name, startup):
	611	"""
	612	Add a seer node to the given topology, with the startup command passed
[06cc65b]	613	in. Used by configure seer_services.
[49051fb]	614	"""
	615	c_node = topdl.Computer(
	616	name=name,
	617	os= topdl.OperatingSystem(
	618	attribute=[
	619	{ 'attribute': 'osid',
	620	'value': self.local_seer_image },
	621	]),
	622	attribute=[
	623	{ 'attribute': 'startup', 'value': startup },
	624	]
	625	)
	626	self.add_kit(c_node, self.local_seer_software)
	627	topo.elements.append(c_node)
	628
	629	def configure_seer_services(self, services, topo, softdir):
[8cf2b90e]	630	"""
	631	Make changes to the topology required for the seer requests being made.
	632	Specifically, add any control or master nodes required and set up the
	633	start commands on the nodes to interconnect them.
	634	"""
	635	local_seer = False # True if we need to add a control node
	636	collect_seer = False # True if there is a seer-master node
	637	seer_master= False # True if we need to add the seer-master
[49051fb]	638	for s in services:
	639	s_name = s.get('name', '')
	640	s_vis = s.get('visibility','')
	641
[8cf2b90e]	642	if s_name == 'local_seer_control' and s_vis == 'export':
[49051fb]	643	local_seer = True
	644	elif s_name == 'seer_master':
	645	if s_vis == 'import':
	646	collect_seer = True
	647	elif s_vis == 'export':
	648	seer_master = True
	649
	650	# We've got the whole picture now, so add nodes if needed and configure
	651	# them to interconnect properly.
	652	if local_seer or seer_master:
	653	# Copy local seer control node software to the tempdir
	654	for l, f in self.local_seer_software:
	655	base = os.path.basename(f)
	656	copy_file(f, "%s/%s" % (softdir, base))
	657	# If we're collecting seers somewhere the controllers need to talk to
	658	# the master. In testbeds that export the service, that will be a
	659	# local node that we'll add below. Elsewhere it will be the control
	660	# portal that will port forward to the exporting master.
	661	if local_seer:
	662	if collect_seer:
[acaa9b9]	663	startup = "%s -C %s" % (self.local_seer_start, "seer-master")
[49051fb]	664	else:
	665	startup = self.local_seer_start
	666	self.add_seer_node(topo, 'control', startup)
	667	# If this is the seer master, add that node, too.
	668	if seer_master:
[e07c8f3]	669	self.add_seer_node(topo, 'seer-master',
	670	"%s -R -n -R seer-master -R -A -R sink" % \
	671	self.seer_master_start)
[49051fb]	672
[06cc65b]	673	def retrieve_software(self, topo, certfile, softdir):
	674	"""
	675	Collect the software that nodes in the topology need loaded and stage
	676	it locally. This implies retrieving it from the experiment_controller
	677	and placing it into softdir. Certfile is used to prove that this node
	678	has access to that data (it's the allocation/segment fedid). Finally
	679	local portal and federation software is also copied to the same staging
	680	directory for simplicity - all software needed for experiment creation
	681	is in softdir.
	682	"""
	683	sw = set()
	684	for e in topo.elements:
	685	for s in getattr(e, 'software', []):
	686	sw.add(s.location)
	687	for s in sw:
	688	self.log.debug("Retrieving %s" % s)
	689	try:
[815cd26]	690	get_url(s, certfile, softdir, log=self.log)
[06cc65b]	691	except:
	692	t, v, st = sys.exc_info()
	693	raise service_error(service_error.internal,
	694	"Error retrieving %s: %s" % (s, v))
[49051fb]	695
[06cc65b]	696	# Copy local federation and portal node software to the tempdir
	697	for s in (self.federation_software, self.portal_software):
	698	for l, f in s:
	699	base = os.path.basename(f)
	700	copy_file(f, "%s/%s" % (softdir, base))
[49051fb]	701
[6c57fe9]	702
[06cc65b]	703	def initialize_experiment_info(self, attrs, aid, certfile, tmpdir):
	704	"""
	705	Gather common configuration files, retrieve or create an experiment
	706	name and project name, and return the ssh_key filenames. Create an
	707	allocation log bound to the state log variable as well.
	708	"""
[3df9b33]	709	configs = ('hosts', 'ssh_pubkey', 'ssh_secretkey',
[9a52a80]	710	'seer_ca_pem', 'seer_node_pem', 'route.tgz')
[06cc65b]	711	ename = None
	712	pubkey_base = None
	713	secretkey_base = None
	714	proj = None
	715	user = None
	716	alloc_log = None
[05c41f5]	717	nonce_experiment = False
[262328f]	718	vchars_re = '[^' + string.ascii_letters + string.digits + '-]'
[06cc65b]	719
[f3898f7]	720	self.state_lock.acquire()
	721	if aid in self.allocation:
	722	proj = self.allocation[aid].get('project', None)
	723	self.state_lock.release()
	724
	725	if not proj:
	726	raise service_error(service_error.internal,
	727	"Can't find project for %s" %aid)
	728
[06cc65b]	729	for a in attrs:
	730	if a['attribute'] in configs:
	731	try:
	732	self.log.debug("Retrieving %s from %s" % \
	733	(a['attribute'], a['value']))
[815cd26]	734	get_url(a['value'], certfile, tmpdir, log=self.log)
[06cc65b]	735	except:
	736	t, v, st = sys.exc_info()
	737	raise service_error(service_error.internal,
	738	"Error retrieving %s: %s" % (a.get('value', ""), v))
	739	if a['attribute'] == 'ssh_pubkey':
	740	pubkey_base = a['value'].rpartition('/')[2]
	741	if a['attribute'] == 'ssh_secretkey':
	742	secretkey_base = a['value'].rpartition('/')[2]
	743	if a['attribute'] == 'experiment_name':
	744	ename = a['value']
	745
[f3898f7]	746	# Names longer than the emulab max are discarded
[c167378]	747	if ename and len(ename) <= self.max_name_len:
[262328f]	748	# Clean up the experiment name so that emulab will accept it.
	749	ename = re.sub(vchars_re, '-', ename)
	750
	751	else:
[06cc65b]	752	ename = ""
	753	for i in range(0,5):
	754	ename += random.choice(string.ascii_letters)
[05c41f5]	755	nonce_experiment = True
[53b5c18]	756	self.log.warn("No experiment name or suggestion too long: " + \
	757	"picked one randomly: %s" % ename)
[06cc65b]	758
	759	if not pubkey_base:
	760	raise service_error(service_error.req,
	761	"No public key attribute")
	762
	763	if not secretkey_base:
	764	raise service_error(service_error.req,
	765	"No secret key attribute")
[6c57fe9]	766
[06cc65b]	767	self.state_lock.acquire()
	768	if aid in self.allocation:
	769	user = self.allocation[aid].get('user', None)
[f77a256]	770	cert = self.allocation[aid].get('cert', None)
[06cc65b]	771	self.allocation[aid]['experiment'] = ename
[05c41f5]	772	self.allocation[aid]['nonce'] = nonce_experiment
[06cc65b]	773	self.allocation[aid]['log'] = [ ]
	774	# Create a logger that logs to the experiment's state object as
	775	# well as to the main log file.
	776	alloc_log = logging.getLogger('fedd.access.%s' % ename)
	777	h = logging.StreamHandler(
	778	list_log.list_log(self.allocation[aid]['log']))
	779	# XXX: there should be a global one of these rather than
	780	# repeating the code.
	781	h.setFormatter(logging.Formatter(
	782	"%(asctime)s %(name)s %(message)s",
	783	'%d %b %y %H:%M:%S'))
	784	alloc_log.addHandler(h)
	785	self.write_state()
	786	self.state_lock.release()
	787
	788	if not user:
	789	raise service_error(service_error.internal,
	790	"Can't find creation user for %s" %aid)
	791
[f77a256]	792	return (ename, proj, user, cert, pubkey_base, secretkey_base, alloc_log)
[06cc65b]	793
[6e33086]	794	def decorate_topology(self, info, t):
[06cc65b]	795	"""
[6e33086]	796	Copy the physical mapping and status onto the topology. Used by
	797	StartSegment and InfoSegment
[06cc65b]	798	"""
[6e33086]	799	def add_new(ann, attr):
	800	for a in ann:
	801	if a not in attr: attr.append(a)
	802
[cebcdce]	803	def merge_os(os, e):
	804	if len(e.os) == 0:
	805	# No OS at all:
	806	if os.get_attribute('emulab_access:image'):
	807	os.set_attribute('emulab_access:initial_image',
	808	os.get_attribute('emulab_access:image'))
	809	e.os = [ os ]
	810	elif len(e.os) == 1:
	811	# There's one OS, copy the initial image and replace
	812	eos = e.os[0]
	813	initial = eos.get_attribute('emulab_access:initial_image')
	814	if initial:
	815	os.set_attribute('emulab_access:initial_image', initial)
	816	e.os = [ os]
	817	else:
	818	# Multiple OSes, replace or append
	819	for eos in e.os:
	820	if os.name == eos.name:
	821	eos.version = os.version
	822	eos.version = os.distribution
	823	eos.version = os.distributionversion
	824	for a in os.attribute:
	825	if eos.get_attribute(a.attribute):
	826	eos.remove_attribute(a.attribute)
	827	eos.set_attribute(a.attribute, a.value)
	828	break
	829	else:
	830	e.os.append(os)
	831
	832
	833	if t is None: return
[7653f01]	834	i = 0 # For fake debugging instantiation
[06cc65b]	835	# Copy the assigned names into the return topology
[c6f867c]	836	for e in t.elements:
[29d5f7c]	837	if isinstance(e, topdl.Computer):
	838	if not self.create_debug:
[6e33086]	839	if e.name in info.node:
[1ae1aa2]	840	add_new(("%s%s" %
	841	(info.node[e.name].pname, self.domain),),
	842	e.localname)
[6527d60]	843	add_new(("%s%s" %
	844	(info.node[e.name].lname, self.domain),),
	845	e.localname)
[1ae1aa2]	846	e.status = info.node[e.name].status
	847	os = info.node[e.name].getOS()
[cebcdce]	848	if os: merge_os(os, e)
[29d5f7c]	849	else:
	850	# Simple debugging assignment
[6e33086]	851	add_new(("node%d%s" % (i, self.domain),), e.localname)
[29d5f7c]	852	e.status = 'active'
[6e33086]	853	add_new(('testop1', 'testop2'), e.operation)
[29d5f7c]	854	i += 1
	855
[6527d60]	856	for s in t.substrates:
	857	if s.name in info.subs:
	858	sub = info.subs[s.name]
	859	if sub.cap is not None:
	860	s.capacity = topdl.Capacity(sub.cap, 'max')
	861	if sub.delay is not None:
	862	s.delay = topdl.Latency(sub.delay, 'max')
	863	# XXX interfaces
	864
[6e33086]	865
	866	def finalize_experiment(self, starter, topo, aid, alloc_id, proof):
	867	"""
	868	Store key bits of experiment state in the global repository, including
	869	the response that may need to be replayed, and return the response.
	870	"""
[9c3e77f]	871	def get_localnames(t):
	872	names = [ ]
	873	for e in t.elements:
	874	if isinstance(e, topdl.Computer):
	875	n = e.get_attribute('testbed')
	876	if n is not None and n not in names:
	877	names.append(n)
	878	return names
[6e33086]	879	i = 0
	880	t = topo.clone()
	881	self.decorate_topology(starter, t)
[06cc65b]	882	# Grab the log (this is some anal locking, but better safe than
	883	# sorry)
	884	self.state_lock.acquire()
[9c3e77f]	885	# Put information about this testbed into the topdl
	886	tb = topdl.Testbed(self.uri, "deter",
	887	localname=get_localnames(t),
	888	attribute=[
	889	{
	890	'attribute': 'project',
	891	'value': self.allocation[aid]['project']
	892	},
	893	{
	894	'attribute': 'experiment',
	895	'value': self.allocation[aid]['experiment']
	896	}])
	897	t.elements.append(tb)
[06cc65b]	898	logv = "".join(self.allocation[aid]['log'])
	899	# It's possible that the StartSegment call gets retried (!).
	900	# if the 'started' key is in the allocation, we'll return it rather
	901	# than redo the setup.
	902	self.allocation[aid]['started'] = {
	903	'allocID': alloc_id,
	904	'allocationLog': logv,
	905	'segmentdescription': {
[b709861]	906	'topdldescription': t.to_dict()
[06cc65b]	907	},
[e83f2f2]	908	'proof': proof.to_dict(),
[06cc65b]	909	}
[b709861]	910	self.allocation[aid]['topo'] = t
[06cc65b]	911	retval = copy.copy(self.allocation[aid]['started'])
	912	self.write_state()
	913	self.state_lock.release()
	914	return retval
	915
	916	# End of StartSegment support routines
	917
	918	def StartSegment(self, req, fid):
[b770aa0]	919	err = None # Any service_error generated after tmpdir is created
	920	rv = None # Return value from segment creation
	921
[8cab4c2]	922	self.log.info("StartSegment called by %s" % fid)
[cc8d8e9]	923	try:
	924	req = req['StartSegmentRequestBody']
[06cc65b]	925	auth_attr = req['allocID']['fedid']
	926	topref = req['segmentdescription']['topdldescription']
[cc8d8e9]	927	except KeyError:
	928	raise service_error(server_error.req, "Badly formed request")
[ecca6eb]	929
[e02cd14]	930	connInfo = req.get('connection', [])
	931	services = req.get('service', [])
[ecca6eb]	932	aid = "%s" % auth_attr
[6c57fe9]	933	attrs = req.get('fedAttr', [])
[e83f2f2]	934
	935	access_ok, proof = self.auth.check_attribute(fid, auth_attr,
	936	with_proof=True)
	937	if not access_ok:
[8cab4c2]	938	self.log.info("StartSegment for %s failed: access denied" % fid)
[ecca6eb]	939	raise service_error(service_error.access, "Access denied")
[cd06678]	940	else:
	941	# See if this is a replay of an earlier succeeded StartSegment -
	942	# sometimes SSL kills 'em. If so, replay the response rather than
	943	# redoing the allocation.
	944	self.state_lock.acquire()
	945	retval = self.allocation[aid].get('started', None)
	946	self.state_lock.release()
	947	if retval:
	948	self.log.warning("Duplicate StartSegment for %s: " % aid + \
	949	"replaying response")
	950	return retval
	951
	952	# A new request. Do it.
[6c57fe9]	953
[06cc65b]	954	if topref: topo = topdl.Topology(**topref)
[6c57fe9]	955	else:
	956	raise service_error(service_error.req,
	957	"Request missing segmentdescription'")
[2761484]	958
[6c57fe9]	959	certfile = "%s/%s.pem" % (self.certdir, auth_attr)
	960	try:
	961	tmpdir = tempfile.mkdtemp(prefix="access-")
[ecca6eb]	962	softdir = "%s/software" % tmpdir
[c23684d]	963	os.mkdir(softdir)
[d3c8759]	964	except EnvironmentError:
[8cab4c2]	965	self.log.info("StartSegment for %s failed: internal error" % fid)
[6c57fe9]	966	raise service_error(service_error.internal, "Cannot create tmp dir")
	967
[b770aa0]	968	# Try block alllows us to clean up temporary files.
	969	try:
[06cc65b]	970	self.retrieve_software(topo, certfile, softdir)
[f77a256]	971	ename, proj, user, xmlrpc_cert, pubkey_base, secretkey_base, \
	972	alloc_log = self.initialize_experiment_info(attrs, aid,
	973	certfile, tmpdir)
[c0f409a]	974	# A misconfigured cert in the ABAC map can be confusing...
	975	if not os.access(xmlrpc_cert, os.R_OK):
	976	self.log.error("Cannot open user's emulab SSL cert: %s" % \
	977	xmlrpc_cert)
	978	raise service_error(service_error.internal,
	979	"Cannot open user's emulab SSL cert: %s" % xmlrpc_cert)
	980
[9b3627e]	981
[f3898f7]	982	if '/' in proj: proj, gid = proj.split('/')
	983	else: gid = None
	984
	985
[06cc65b]	986	# Set up userconf and seer if needed
[c200d36]	987	self.configure_userconf(services, tmpdir)
[49051fb]	988	self.configure_seer_services(services, topo, softdir)
[06cc65b]	989	# Get and send synch store variables
[26821ac]	990	exporter = self.exports_segment(keyfile=self.ssh_privkey_file,
	991	debug=self.create_debug, log=alloc_log, boss=self.boss,
	992	ops=self.ops, cert=xmlrpc_cert)
[8fbef04]	993	if not exporter(self, ename, proj, user, connInfo, tmpdir, gid):
	994	raise service_error(service_error.internal,
	995	'Failed to gather export variables')
[2761484]	996	self.export_store_info(certfile, proj, ename, connInfo)
	997	self.import_store_info(certfile, connInfo)
	998
[b770aa0]	999	expfile = "%s/experiment.tcl" % tmpdir
	1000
	1001	self.generate_portal_configs(topo, pubkey_base,
[06cc65b]	1002	secretkey_base, tmpdir, proj, ename, connInfo, services)
[b770aa0]	1003	self.generate_ns2(topo, expfile,
[8cf2b90e]	1004	"/proj/%s/software/%s/" % (proj, ename), connInfo)
[f07fa49]	1005
[b770aa0]	1006	starter = self.start_segment(keyfile=self.ssh_privkey_file,
[181aeb4]	1007	debug=self.create_debug, log=alloc_log, boss=self.boss,
[f77a256]	1008	ops=self.ops, cert=xmlrpc_cert)
[f3898f7]	1009	rv = starter(self, ename, proj, user, expfile, tmpdir, gid=gid)
[b770aa0]	1010	except service_error, e:
[8cab4c2]	1011	self.log.info("StartSegment for %s failed: %s" % (fid, e))
[b770aa0]	1012	err = e
[06cc65b]	1013	except:
	1014	t, v, st = sys.exc_info()
[5f6ebd0]	1015	self.log.info("StartSegment for %s failed:unexpected error: %s" \
	1016	% (fid, traceback.extract_tb(st)))
[06cc65b]	1017	err = service_error(service_error.internal, "%s: %s" % \
	1018	(v, traceback.extract_tb(st)))
[b770aa0]	1019
[574055e]	1020	# Walk up tmpdir, deleting as we go
[06cc65b]	1021	if self.cleanup: self.remove_dirs(tmpdir)
	1022	else: self.log.debug("[StartSegment]: not removing %s" % tmpdir)
[574055e]	1023
[fd556d1]	1024	if rv:
[8cab4c2]	1025	self.log.info("StartSegment for %s succeeded" % fid)
[e83f2f2]	1026	return self.finalize_experiment(starter, topo, aid, req['allocID'],
	1027	proof)
[b770aa0]	1028	elif err:
	1029	raise service_error(service_error.federant,
	1030	"Swapin failed: %s" % err)
[fd556d1]	1031	else:
	1032	raise service_error(service_error.federant, "Swapin failed")
[5ae3857]	1033
	1034	def TerminateSegment(self, req, fid):
[8cab4c2]	1035	self.log.info("TerminateSegment called by %s" % fid)
[5ae3857]	1036	try:
	1037	req = req['TerminateSegmentRequestBody']
	1038	except KeyError:
	1039	raise service_error(server_error.req, "Badly formed request")
	1040
	1041	auth_attr = req['allocID']['fedid']
	1042	aid = "%s" % auth_attr
	1043	attrs = req.get('fedAttr', [])
[e83f2f2]	1044
	1045	access_ok, proof = self.auth.check_attribute(fid, auth_attr,
	1046	with_proof=True)
	1047	if not access_ok:
[5ae3857]	1048	raise service_error(service_error.access, "Access denied")
	1049
	1050	self.state_lock.acquire()
[06cc65b]	1051	if aid in self.allocation:
[5ae3857]	1052	proj = self.allocation[aid].get('project', None)
	1053	user = self.allocation[aid].get('user', None)
[f77a256]	1054	xmlrpc_cert = self.allocation[aid].get('cert', None)
[5ae3857]	1055	ename = self.allocation[aid].get('experiment', None)
[05c41f5]	1056	nonce = self.allocation[aid].get('nonce', False)
[1d913e13]	1057	else:
	1058	proj = None
	1059	user = None
	1060	ename = None
[05c41f5]	1061	nonce = False
[f77a256]	1062	xmlrpc_cert = None
[5ae3857]	1063	self.state_lock.release()
	1064
	1065	if not proj:
[8cab4c2]	1066	self.log.info("TerminateSegment failed for %s: cannot find project"\
	1067	% fid)
[5ae3857]	1068	raise service_error(service_error.internal,
	1069	"Can't find project for %s" % aid)
[f3898f7]	1070	else:
	1071	if '/' in proj: proj, gid = proj.split('/')
	1072	else: gid = None
[5ae3857]	1073
	1074	if not user:
[8cab4c2]	1075	self.log.info("TerminateSegment failed for %s: cannot find user"\
	1076	% fid)
[5ae3857]	1077	raise service_error(service_error.internal,
	1078	"Can't find creation user for %s" % aid)
	1079	if not ename:
[8cab4c2]	1080	self.log.info(
	1081	"TerminateSegment failed for %s: cannot find experiment"\
	1082	% fid)
[5ae3857]	1083	raise service_error(service_error.internal,
	1084	"Can't find experiment name for %s" % aid)
[b90c44d]	1085
[fd556d1]	1086	stopper = self.stop_segment(keyfile=self.ssh_privkey_file,
[c7141dc]	1087	debug=self.create_debug, boss=self.boss, ops=self.ops,
[f77a256]	1088	cert=xmlrpc_cert)
[05c41f5]	1089	stopper(self, user, proj, ename, gid, nonce)
[8cab4c2]	1090	self.log.info("TerminateSegment succeeded for %s %s %s" % \
	1091	(fid, proj, ename))
[b90c44d]	1092	self.state_lock.acquire()
	1093	# Remove the started flag/info - the segment is no longer started
	1094	if aid in self.allocation:
[e62fb86]	1095	if 'started' in self.allocation[aid]:
	1096	del self.allocation[aid]['started']
[b90c44d]	1097	self.write_state()
	1098	self.state_lock.release()
[e83f2f2]	1099	return { 'allocID': req['allocID'], 'proof': proof.to_dict() }
[45e880d]	1100
	1101	def InfoSegment(self, req, fid):
[8cab4c2]	1102	self.log.info("InfoSegment called by %s" % fid)
[45e880d]	1103	try:
	1104	req = req['InfoSegmentRequestBody']
	1105	except KeyError:
	1106	raise service_error(server_error.req, "Badly formed request")
	1107
	1108	auth_attr = req['allocID']['fedid']
	1109	aid = "%s" % auth_attr
	1110
	1111	access_ok, proof = self.auth.check_attribute(fid, auth_attr,
	1112	with_proof=True)
	1113	if not access_ok:
	1114	raise service_error(service_error.access, "Access denied")
	1115
	1116	self.state_lock.acquire()
	1117	if aid in self.allocation:
	1118	topo = self.allocation[aid].get('topo', None)
	1119	if topo: topo = topo.clone()
	1120	proj = self.allocation[aid].get('project', None)
	1121	user = self.allocation[aid].get('user', None)
[f77a256]	1122	xmlrpc_cert = self.allocation[aid].get('cert', None)
[45e880d]	1123	ename = self.allocation[aid].get('experiment', None)
	1124	else:
	1125	proj = None
	1126	user = None
	1127	ename = None
[b709861]	1128	topo = None
[f77a256]	1129	xmlrpc_cert = None
[45e880d]	1130	self.state_lock.release()
	1131
	1132	if not proj:
[8cab4c2]	1133	self.log.info("InfoSegment failed for %s: cannot find project"% fid)
[45e880d]	1134	raise service_error(service_error.internal,
	1135	"Can't find project for %s" % aid)
	1136	else:
	1137	if '/' in proj: proj, gid = proj.split('/')
	1138	else: gid = None
	1139
	1140	if not user:
[8cab4c2]	1141	self.log.info("InfoSegment failed for %s: cannot find user"% fid)
[45e880d]	1142	raise service_error(service_error.internal,
	1143	"Can't find creation user for %s" % aid)
	1144	if not ename:
[8cab4c2]	1145	self.log.info("InfoSegment failed for %s: cannot find exp"% fid)
[45e880d]	1146	raise service_error(service_error.internal,
	1147	"Can't find experiment name for %s" % aid)
	1148	info = self.info_segment(keyfile=self.ssh_privkey_file,
[c7141dc]	1149	debug=self.create_debug, boss=self.boss, ops=self.ops,
[f77a256]	1150	cert=xmlrpc_cert)
[6e33086]	1151	info(self, user, proj, ename)
[8cab4c2]	1152	self.log.info("InfoSegment gathered info for %s %s %s %s" % \
	1153	(fid, user, proj, ename))
[6e33086]	1154	self.decorate_topology(info, topo)
[1ae1aa2]	1155	self.state_lock.acquire()
	1156	if aid in self.allocation:
	1157	self.allocation[aid]['topo'] = topo
	1158	self.write_state()
	1159	self.state_lock.release()
[8cab4c2]	1160	self.log.info("InfoSegment updated info for %s %s %s %s" % \
	1161	(fid, user, proj, ename))
[1ae1aa2]	1162
[7f57435]	1163	rv = {
[45e880d]	1164	'allocID': req['allocID'],
	1165	'proof': proof.to_dict(),
	1166	}
[8cab4c2]	1167	self.log.info("InfoSegment succeeded info for %s %s %s %s" % \
	1168	(fid, user, proj, ename))
[7f57435]	1169	if topo:
	1170	rv['segmentdescription'] = { 'topdldescription' : topo.to_dict() }
	1171	return rv
[b709861]	1172
	1173	def OperationSegment(self, req, fid):
	1174	def get_pname(e):
	1175	"""
	1176	Get the physical name of a node
	1177	"""
	1178	if e.localname:
	1179	return re.sub('\..*','', e.localname[0])
	1180	else:
	1181	return None
	1182
[8cab4c2]	1183	self.log.info("OperationSegment called by %s" % fid)
[b709861]	1184	try:
	1185	req = req['OperationSegmentRequestBody']
	1186	except KeyError:
	1187	raise service_error(server_error.req, "Badly formed request")
	1188
	1189	auth_attr = req['allocID']['fedid']
	1190	aid = "%s" % auth_attr
	1191
	1192	access_ok, proof = self.auth.check_attribute(fid, auth_attr,
	1193	with_proof=True)
	1194	if not access_ok:
[8cab4c2]	1195	self.log.info("OperationSegment failed for %s: access denied" % fid)
[b709861]	1196	raise service_error(service_error.access, "Access denied")
	1197
	1198	op = req.get('operation', None)
	1199	targets = req.get('target', None)
	1200	params = req.get('parameter', None)
	1201
	1202	if op is None :
[8cab4c2]	1203	self.log.info("OperationSegment failed for %s: no operation" % fid)
[b709861]	1204	raise service_error(service_error.req, "missing operation")
	1205	elif targets is None:
[8cab4c2]	1206	self.log.info("OperationSegment failed for %s: no targets" % fid)
[b709861]	1207	raise service_error(service_error.req, "no targets")
	1208
[f77a256]	1209	self.state_lock.acquire()
[b709861]	1210	if aid in self.allocation:
	1211	topo = self.allocation[aid].get('topo', None)
	1212	if topo: topo = topo.clone()
[f77a256]	1213	xmlrpc_cert = self.allocation[aid].get('cert', None)
[b709861]	1214	else:
	1215	topo = None
[f77a256]	1216	xmlrpc_cert = None
	1217	self.state_lock.release()
[b709861]	1218
	1219	targets = copy.copy(targets)
	1220	ptargets = { }
	1221	for e in topo.elements:
	1222	if isinstance(e, topdl.Computer):
	1223	if e.name in targets:
	1224	targets.remove(e.name)
	1225	pn = get_pname(e)
	1226	if pn: ptargets[e.name] = pn
	1227
	1228	status = [ operation_status(t, operation_status.no_target) \
	1229	for t in targets]
	1230
	1231	ops = self.operation_segment(keyfile=self.ssh_privkey_file,
[c7141dc]	1232	debug=self.create_debug, boss=self.boss, ops=self.ops,
[f77a256]	1233	cert=xmlrpc_cert)
[1ae1aa2]	1234	ops(self, op, ptargets, params, topo)
[8cab4c2]	1235	self.log.info("OperationSegment operated for %s" % fid)
[b709861]	1236
	1237	status.extend(ops.status)
[8cab4c2]	1238	self.log.info("OperationSegment succeed for %s" % fid)
[b709861]	1239
	1240	return {
	1241	'allocID': req['allocID'],
	1242	'status': [s.to_dict() for s in status],
	1243	'proof': proof.to_dict(),
	1244	}

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

Download in other formats: