Context Navigation

source: fedd/federation/dragon_access.py @ ab847bc

axis_examplecompt_changesinfo-opsversion-3.01version-3.02

Last change on this file since ab847bc was ab847bc, checked in by Ted Faber <faber@…>, 14 years ago
New syntax for testbeds that includes a /instance rider. This allows users to request multiple segments from a single testbed.
Property mode set to `100644`
File size: 30.7 KB

Line
1	#!/usr/local/bin/python
2
3	import os,sys
4	import re
5	import string
6	import copy
7	import pickle
8	import logging
9	import time
10
11	from threading import Thread, Lock
12	from subprocess import Popen, call, PIPE, STDOUT
13
14	from util import *
15	from allocate_project import allocate_project_local, allocate_project_remote
16	from access_project import access_project
17	from fedid import fedid, generate_fedid
18	from authorizer import authorizer
19	from service_error import service_error
20	from remote_service import xmlrpc_handler, soap_handler, service_caller
21
22	import httplib
23	import tempfile
24	from urlparse import urlparse
25
26	import topdl
27	import list_log
28	import proxy_emulab_segment
29	import local_emulab_segment
30
31
32	# Make log messages disappear if noone configures a fedd logger
33	class nullHandler(logging.Handler):
34	def emit(self, record): pass
35
36	fl = logging.getLogger("fedd.access")
37	fl.addHandler(nullHandler())
38
39	class access:
40	"""
41	The implementation of access control based on mapping users to projects.
42
43	Users can be mapped to existing projects or have projects created
44	dynamically. This implements both direct requests and proxies.
45	"""
46
47	class parse_error(RuntimeError): pass
48
49
50	proxy_RequestAccess= service_caller('RequestAccess')
51	proxy_ReleaseAccess= service_caller('ReleaseAccess')
52
53	def __init__(self, config=None, auth=None):
54	"""
55	Initializer. Pulls parameters out of the ConfigParser's access section.
56	"""
57
58	# Make sure that the configuration is in place
59	if not config:
60	raise RunTimeError("No config to dragon_access.access")
61
62	self.project_priority = config.getboolean("access", "project_priority")
63	self.allow_proxy = False
64	self.certdir = config.get("access","certdir")
65	self.create_debug = config.getboolean("access", "create_debug")
66	self.cli_dir = config.get("access", "cli_dir")
67	self.axis2_home = config.get("access", "axis2_home")
68	self.idc_url = config.get("access", "idc")
69	self.domain = config.get("access", "domain")
70	self.duration = config.getint("access", "duration", 120)
71
72	self.attrs = { }
73	self.access = { }
74	# State is a dict of dicts indexed by segment fedid that includes the
75	# owners of the segment as fedids (who can manipulate it, key: owners),
76	# the repo dir/user for the allocation (key: user), Current allocation
77	# log (key: log), and GRI of the reservation once made (key: gri)
78	self.state = { }
79	self.log = logging.getLogger("fedd.access")
80	set_log_level(config, "access", self.log)
81	self.state_lock = Lock()
82	if not (self.cli_dir and self.axis2_home and self.idc_url):
83	self.log.error("Must specify all of cli_dir, axis2_home, " +\
84	"idc in the [access] section of the configuration")
85
86	if auth: self.auth = auth
87	else:
88	self.log.error(\
89	"[access]: No authorizer initialized, creating local one.")
90	auth = authorizer()
91
92	tb = config.get('access', 'testbed')
93	if tb: self.testbed = [ t.strip() for t in tb.split(',') ]
94	else: self.testbed = [ ]
95
96	if config.has_option("access", "accessdb"):
97	self.read_access(config.get("access", "accessdb"))
98
99	self.state_filename = config.get("access", "access_state")
100	self.read_state()
101
102	# Keep cert_file and cert_pwd coming from the same place
103	self.cert_file = config.get("access", "cert_file")
104	if self.cert_file:
105	self.sert_pwd = config.get("access", "cert_pw")
106	else:
107	self.cert_file = config.get("globals", "cert_file")
108	self.sert_pwd = config.get("globals", "cert_pw")
109
110	self.trusted_certs = config.get("access", "trusted_certs") or \
111	config.get("globals", "trusted_certs")
112
113	self.call_GetValue= service_caller('GetValue')
114	self.call_SetValue= service_caller('SetValue')
115
116	self.soap_services = {\
117	'RequestAccess': soap_handler("RequestAccess", self.RequestAccess),
118	'ReleaseAccess': soap_handler("ReleaseAccess", self.ReleaseAccess),
119	'StartSegment': soap_handler("StartSegment", self.StartSegment),
120	'TerminateSegment': soap_handler("TerminateSegment",
121	self.TerminateSegment),
122	}
123	self.xmlrpc_services = {\
124	'RequestAccess': xmlrpc_handler('RequestAccess',
125	self.RequestAccess),
126	'ReleaseAccess': xmlrpc_handler('ReleaseAccess',
127	self.ReleaseAccess),
128	'StartSegment': xmlrpc_handler("StartSegment", self.StartSegment),
129	'TerminateSegment': xmlrpc_handler('TerminateSegment',
130	self.TerminateSegment),
131	}
132
133	def read_access(self, config):
134	"""
135	Read a configuration file and set internal parameters.
136
137	There are access lines of the
138	form (tb, proj, user) -> user that map the first tuple of
139	names to the user for for access purposes. Names in the key (left side)
140	can include "<NONE> or <ANY>" to act as wildcards or to require the
141	fields to be empty. Similarly aproj or auser can be <SAME> or
142	<DYNAMIC> indicating that either the matching key is to be used or a
143	dynamic user or project will be created. These names can also be
144	federated IDs (fedid's) if prefixed with fedid:. The user is the repo
145	directory that contains the DRAGON user credentials.
146	Testbed attributes outside the forms above can be given using the
147	format attribute: name value: value. The name is a single word and the
148	value continues to the end of the line. Empty lines and lines startin
149	with a # are ignored.
150
151	Parsing errors result in a self.parse_error exception being raised.
152	"""
153	lineno=0
154	name_expr = "["+string.ascii_letters + string.digits + "\.\-_]+"
155	fedid_expr = "fedid:[" + string.hexdigits + "]+"
156	key_name = "(<ANY>\|<NONE>\|"+fedid_expr + "\|"+ name_expr + ")"
157
158	attr_re = re.compile('attribute:\s([\._\-a-z0-9]+)\s+value:\s(.*)',
159	re.IGNORECASE)
160	access_re = re.compile('\('+key_name+'\s,\s'+key_name+'\s,\s'+
161	key_name+'\s\)\s->\s\(('+name_expr +')\s\)', re.IGNORECASE)
162
163	def parse_name(n):
164	if n.startswith('fedid:'): return fedid(hexstr=n[len('fedid:'):])
165	else: return n
166
167	def auth_name(n):
168	if isinstance(n, basestring):
169	if n =='<any>' or n =='<none>': return None
170	else: return unicode(n)
171	else:
172	return n
173
174	f = open(config, "r");
175	for line in f:
176	lineno += 1
177	line = line.strip();
178	if len(line) == 0 or line.startswith('#'):
179	continue
180
181	# Extended (attribute: x value: y) attribute line
182	m = attr_re.match(line)
183	if m != None:
184	attr, val = m.group(1,2)
185	self.attrs[attr] = val
186	continue
187
188	# Access line (t, p, u) -> (a) line
189	# XXX: you are here
190	m = access_re.match(line)
191	if m != None:
192	access_key = tuple([ parse_name(x) for x in m.group(1,2,3)])
193	auth_key = tuple([ auth_name(x) for x in access_key])
194	user_name = auth_name(parse_name(m.group(4)))
195
196	self.access[access_key] = user_name
197	self.auth.set_attribute(auth_key, "access")
198	continue
199
200	# Nothing matched to here: unknown line - raise exception
201	f.close()
202	raise self.parse_error("Unknown statement at line %d of %s" % \
203	(lineno, config))
204	f.close()
205
206	def get_users(self, obj):
207	"""
208	Return a list of the IDs of the users in dict
209	"""
210	if obj.has_key('user'):
211	return [ unpack_id(u['userID']) \
212	for u in obj['user'] if u.has_key('userID') ]
213	else:
214	return None
215
216	def write_state(self):
217	if self.state_filename:
218	try:
219	f = open(self.state_filename, 'w')
220	pickle.dump(self.state, f)
221	except IOError, e:
222	self.log.error("Can't write file %s: %s" % \
223	(self.state_filename, e))
224	except pickle.PicklingError, e:
225	self.log.error("Pickling problem: %s" % e)
226	except TypeError, e:
227	self.log.error("Pickling problem (TypeError): %s" % e)
228
229
230	def read_state(self):
231	"""
232	Read a new copy of access state. Old state is overwritten.
233
234	State format is a simple pickling of the state dictionary.
235	"""
236	if self.state_filename:
237	try:
238	f = open(self.state_filename, "r")
239	self.state = pickle.load(f)
240	self.log.debug("[read_state]: Read state from %s" % \
241	self.state_filename)
242	except IOError, e:
243	self.log.warning(("[read_state]: No saved state: " +\
244	"Can't open %s: %s") % (self.state_filename, e))
245	except EOFError, e:
246	self.log.warning(("[read_state]: " +\
247	"Empty or damaged state file: %s:") % \
248	self.state_filename)
249	except pickle.UnpicklingError, e:
250	self.log.warning(("[read_state]: No saved state: " + \
251	"Unpickling failed: %s") % e)
252
253	# Add the ownership attributes to the authorizer. Note that the
254	# indices of the allocation dict are strings, but the attributes are
255	# fedids, so there is a conversion.
256	for k in self.state.keys():
257	for o in self.state[k].get('owners', []):
258	self.auth.set_attribute(o, fedid(hexstr=k))
259	self.auth.set_attribute(fedid(hexstr=k),fedid(hexstr=k))
260
261
262	def permute_wildcards(self, a, p):
263	"""Return a copy of a with various fields wildcarded.
264
265	The bits of p control the wildcards. A set bit is a wildcard
266	replacement with the lowest bit being user then project then testbed.
267	"""
268	if p & 1: user = ["<any>"]
269	else: user = a[2]
270	if p & 2: proj = "<any>"
271	else: proj = a[1]
272	if p & 4: tb = "<any>"
273	else: tb = a[0]
274
275	return (tb, proj, user)
276
277	def find_access(self, search):
278	"""
279	Search the access DB for a match on this tuple. Return the matching
280	user (repo dir).
281
282	NB, if the initial tuple fails to match we start inserting wildcards in
283	an order determined by self.project_priority. Try the list of users in
284	order (when wildcarded, there's only one user in the list).
285	"""
286	if self.project_priority: perm = (0, 1, 2, 3, 4, 5, 6, 7)
287	else: perm = (0, 2, 1, 3, 4, 6, 5, 7)
288
289	for p in perm:
290	s = self.permute_wildcards(search, p)
291	# s[2] is None on an anonymous, unwildcarded request
292	if s[2] != None:
293	for u in s[2]:
294	if self.access.has_key((s[0], s[1], u)):
295	return self.access[(s[0], s[1], u)]
296	else:
297	if self.access.has_key(s):
298	return self.access[s]
299	return None
300
301	def lookup_access(self, req, fid):
302	"""
303	Determine the allowed access for this request. Return the access and
304	which fields are dynamic.
305
306	The fedid is needed to construct the request
307	"""
308	# Search keys
309	tb = None
310	project = None
311	user = None
312	# Return values
313	rp = access_project(None, ())
314	ru = None
315	user_re = re.compile("user:\s(.*)")
316	project_re = re.compile("project:\s(.*)")
317
318	user = [ user_re.findall(x)[0] for x in req.get('credential', []) \
319	if user_re.match(x)]
320	project = [ project_re.findall(x)[0] \
321	for x in req.get('credential', []) \
322	if project_re.match(x)]
323
324	if len(project) == 1: project = project[0]
325	elif len(project) == 0: project = None
326	else:
327	raise service_error(service_error.req,
328	"More than one project credential")
329
330
331	user_fedids = [ u for u in user if isinstance(u, fedid)]
332
333	# Determine how the caller is representing itself. If its fedid shows
334	# up as a project or a singleton user, let that stand. If neither the
335	# usernames nor the project name is a fedid, the caller is a testbed.
336	if project and isinstance(project, fedid):
337	if project == fid:
338	# The caller is the project (which is already in the tuple
339	# passed in to the authorizer)
340	owners = user_fedids
341	owners.append(project)
342	else:
343	raise service_error(service_error.req,
344	"Project asserting different fedid")
345	else:
346	if fid not in user_fedids:
347	tb = fid
348	owners = user_fedids
349	owners.append(fid)
350	else:
351	if len(fedids) > 1:
352	raise service_error(service_error.req,
353	"User asserting different fedid")
354	else:
355	# Which is a singleton
356	owners = user_fedids
357	# Confirm authorization
358
359	for u in user:
360	self.log.debug("[lookup_access] Checking access for %s" % \
361	((tb, project, u),))
362	if self.auth.check_attribute((tb, project, u), 'access'):
363	self.log.debug("[lookup_access] Access granted")
364	break
365	else:
366	self.log.debug("[lookup_access] Access Denied")
367	else:
368	raise service_error(service_error.access, "Access denied")
369
370	# This maps a valid user to the Emulab projects and users to use
371	found = self.find_access((tb, project, user))
372
373	if found == None:
374	raise service_error(service_error.access,
375	"Access denied - cannot map access")
376	return found, owners
377
378	def RequestAccess(self, req, fid):
379	"""
380	Handle the access request. Proxy if not for us.
381
382	Parse out the fields and make the allocations or rejections if for us,
383	otherwise, assuming we're willing to proxy, proxy the request out.
384	"""
385
386	# The dance to get into the request body
387	if req.has_key('RequestAccessRequestBody'):
388	req = req['RequestAccessRequestBody']
389	else:
390	raise service_error(service_error.req, "No request!?")
391
392	if req.has_key('destinationTestbed'):
393	dt = unpack_id(req['destinationTestbed'])
394
395	if dt == None or dt in self.testbed:
396	# Request for this fedd
397	found, owners = self.lookup_access(req, fid)
398	# keep track of what's been added
399	allocID, alloc_cert = generate_fedid(subj="alloc", log=self.log)
400	aid = unicode(allocID)
401
402	self.state_lock.acquire()
403	self.state[aid] = { }
404	self.state[aid]['user'] = found
405	self.state[aid]['owners'] = owners
406	self.write_state()
407	self.state_lock.release()
408	for o in owners:
409	self.auth.set_attribute(o, allocID)
410	self.auth.set_attribute(allocID, allocID)
411
412	try:
413	f = open("%s/%s.pem" % (self.certdir, aid), "w")
414	print >>f, alloc_cert
415	f.close()
416	except IOError, e:
417	raise service_error(service_error.internal,
418	"Can't open %s/%s : %s" % (self.certdir, aid, e))
419	return { 'allocID': { 'fedid': allocID } }
420	else:
421	if self.allow_proxy:
422	resp = self.proxy_RequestAccess.call_service(dt, req,
423	self.cert_file, self.cert_pwd,
424	self.trusted_certs)
425	if resp.has_key('RequestAccessResponseBody'):
426	return resp['RequestAccessResponseBody']
427	else:
428	return None
429	else:
430	raise service_error(service_error.access,
431	"Access proxying denied")
432
433	def ReleaseAccess(self, req, fid):
434	# The dance to get into the request body
435	if req.has_key('ReleaseAccessRequestBody'):
436	req = req['ReleaseAccessRequestBody']
437	else:
438	raise service_error(service_error.req, "No request!?")
439
440	if req.has_key('destinationTestbed'):
441	dt = unpack_id(req['destinationTestbed'])
442	else:
443	dt = None
444
445	if dt == None or dt in self.testbed:
446	# Local request
447	try:
448	if req['allocID'].has_key('localname'):
449	auth_attr = aid = req['allocID']['localname']
450	elif req['allocID'].has_key('fedid'):
451	aid = unicode(req['allocID']['fedid'])
452	auth_attr = req['allocID']['fedid']
453	else:
454	raise service_error(service_error.req,
455	"Only localnames and fedids are understood")
456	except KeyError:
457	raise service_error(service_error.req, "Badly formed request")
458
459	self.log.debug("[access] deallocation requested for %s", aid)
460	if not self.auth.check_attribute(fid, auth_attr):
461	self.log.debug("[access] deallocation denied for %s", aid)
462	raise service_error(service_error.access, "Access Denied")
463
464	self.state_lock.acquire()
465	if self.state.has_key(aid):
466	self.log.debug("Found allocation for %s" %aid)
467	del self.state[aid]
468	self.write_state()
469	self.state_lock.release()
470	# And remove the access cert
471	cf = "%s/%s.pem" % (self.certdir, aid)
472	self.log.debug("Removing %s" % cf)
473	os.remove(cf)
474	return { 'allocID': req['allocID'] }
475	else:
476	self.state_lock.release()
477	raise service_error(service_error.req, "No such allocation")
478
479	else:
480	if self.allow_proxy:
481	resp = self.proxy_ReleaseAccess.call_service(dt, req,
482	self.cert_file, self.cert_pwd,
483	self.trusted_certs)
484	if resp.has_key('ReleaseAccessResponseBody'):
485	return resp['ReleaseAccessResponseBody']
486	else:
487	return None
488	else:
489	raise service_error(service_error.access,
490	"Access proxying denied")
491
492	def extract_parameters(self, top):
493	"""
494	DRAGON currently supports a fixed capacity link between two endpoints.
495	Those endpoints may specify a VPN (or list or range) to use. This
496	extracts the DRAGON endpoints and vpn preferences from the segments (as
497	attributes) and the capacity of the connection as given by the
498	substrate. The two endpoints VLAN choices are intersected to get set
499	of VLANs that are acceptable (no VLAN requiremnets means any is
500	acceptable).
501	"""
502	segments = filter(lambda x: isinstance(x, topdl.Segment), top.elements)
503
504	if len(segments) != 2 or len(top.substrates) != 1:
505	raise service_error(service_error.req,
506	"Requests to DRAGON must have exactlty two segments " +\
507	"and one substrate")
508
509	ends = [ ]
510	for s in segments:
511	ep = s.get_attribute('dragon_endpoint')
512	if not ep:
513	raise service_error(service_error.req,
514	"Missing DRAGON endpoint for %s" % s.id)
515	v = s.get_attribute('vlans')
516	vlans = None
517	vset = set()
518	# the vlans can be a single integer, a comma separated list or a
519	# comma separated lists of dashed ranges. E.g 100 or 100,300 or
520	# 100,300-305,400
521	if v:
522	if v.count(",") > 0 :
523	vl = [ x.strip() for x in v.split(",") ]
524	else:
525	vl = [ v ]
526	for v in vl:
527	try:
528	if v.count("-")> 0:
529	f, t = v.split("-", 1)
530	for i in range(int(f), int(t)+1):
531	vset.add(i)
532	else:
533	vset.add(int(v))
534	except ValueError:
535	raise service_error(service_error.req,
536	"VLAN tags must be integers (%s)" %s.name)
537	if len(vset) > 0:
538	if vlans: vlans &= vest
539	else: vlans = vset
540	ends.append(ep)
541
542
543	sub = top.substrates[0]
544	if sub.capacity:
545	cap = int(sub.capacity.rate / 1000.0)
546	if cap < 1: cap = 1
547	else:
548	cap = 100
549
550
551	# DRAGON's command line tool barfs if the source (ends[0]) is not in
552	# the domain controlled by the IDC. This code ensures this situation.
553	if self.domain and not ends[0].endswith(self.domain):
554	hold = ends[0]
555	for i, e in enumerate(ends):
556	if i == 0: continue
557	if e.endswith(self.domain):
558	ends[0] = e
559	ends[i] = hold
560	break
561	else:
562	raise service_error(service_error.req,
563	"No endpoint in my domain")
564
565
566	return cap, ends[0], ends[1], vlans
567
568
569	def start_segment(self, repo, fr, to, cap, vpns=[], start=None, end=None,
570	log=None):
571	"""
572	Do the actual work of creating the dragon connecton.
573	"""
574	if not log: log = self.log
575	gri_re = re.compile("GRI:\s(.)", re.IGNORECASE)
576	status_re = re.compile("Status:\s(.)", re.IGNORECASE)
577	source_re = re.compile("Source\s+Endpoint:\s(.)", re.IGNORECASE)
578	dest_re = re.compile("Destination\s+Endpoint:\s(.)", re.IGNORECASE)
579	path_re = re.compile("Path:")
580
581	if not vpns:
582	vpns = [ None, None, None, None, None]
583
584	if not start:
585	start = time.time()
586	if not end:
587	end = start + self.duration *60
588
589
590	status = None
591	gri = None
592	rv = None
593	vlan_no = None
594	for v in vpns:
595	cmd = ['env', 'AXIS2_HOME=%s' % self.axis2_home, './run.sh',
596	'createReservation', '-repo', repo , '-url', self.idc_url,
597	'-l2source', fr, '-l2dest', to, '-bwidth', "%s" % cap,
598	'-vlan', "%s" % v, '-desc', 'fedd created connection',
599	'-pathsetup', 'timer-automatic', '-start', "%d" % int(start),
600	'-end', "%d" % int(end)]
601	log.debug("[start_segment]: %s" % " ".join(cmd))
602	if not self.create_debug:
603	p = Popen(cmd, cwd=self.cli_dir,
604	stdout=PIPE, stderr=STDOUT,
605	close_fds=True)
606	for line in p.stdout:
607	m = status_re.match(line)
608	if m:
609	status = m.group(1)
610	continue
611	m = gri_re.match(line)
612	if m:
613	gri = m.group(1)
614	rv = p.wait()
615	else:
616	rv = 0
617	status = 'ACCEPTED'
618	gri = 'debug_gri'
619
620	# Reservation in progress. Poll the IDC until we know the outcome
621	cmd = ['env', 'AXIS2_HOME=%s' % self.axis2_home, './run.sh',
622	'query', '-repo', repo , '-url', self.idc_url, '-gri', gri]
623
624	while status in ('ACCEPTED', 'INSETUP', 'INCREATE', 'PENDING'):
625	log.debug("[start_segment]: %s" % " ".join(cmd))
626	if not self.create_debug:
627	p = Popen(cmd, cwd=self.cli_dir,
628	stdout=PIPE, stderr=STDOUT,
629	close_fds=True)
630	in_path = False
631	vpn1 = None
632	vpn2 = None
633	src = None
634	dest = None
635	for line in p.stdout:
636	if not in_path:
637	m = status_re.match(line)
638	if m:
639	status = m.group(1)
640	continue
641	m = source_re.match(line)
642	if m:
643	src = m.group(1)
644	continue
645	m = dest_re.match(line)
646	if m:
647	dest = m.group(1)
648	continue
649	m = path_re.match(line)
650	if m:
651	in_path = True
652	if src and dest:
653	vpn1_re = re.compile(
654	"\s%s,\s\w+\s,\s(\d+)" % \
655	src.replace("", "\"))
656	vpn2_re = re.compile(
657	"\s%s,\s\w+\s,\s(\d+)" % \
658	dest.replace("", "\"))
659	else:
660	raise service_error(service_error.internal,
661	"Strange output from query")
662	else:
663	m = vpn1_re.match(line)
664	if m:
665	vpn1 = m.group(1)
666	continue
667	m = vpn2_re.match(line)
668	if m:
669	vpn2 = m.group(1)
670	continue
671	rv = p.wait()
672	if vpn1 == vpn2:
673	if v is not None:
674	if int(vpn1) == v:
675	vlan_no = int(v)
676	else:
677	raise service_error(service_error.federant,
678	"Unexpected vlan assignment")
679	else:
680	vlan_no = int(v)
681	else:
682	raise service_error(service_error.internal,
683	"Different VPNs on DRAGON ends")
684	log.debug("Status: %s" % status or "none")
685	else:
686	status = 'ACTIVE'
687	vlan_no = int(v) or 1
688	if status in ('ACCEPTED', 'INSETUP', 'INCREATE', 'PENDING'):
689	time.sleep(45)
690	if status in ('ACTIVE', 'FINISHED', 'CANCELLED'):
691	break
692
693	self.log.debug("made reservation %s %s" % (gri, vlan_no))
694
695	if (rv == 0 and gri and vlan_no and status == 'ACTIVE'):
696	return gri, vlan_no
697	else:
698	raise service_error(service_error.federant,
699	"Cannot make reservation")
700
701	def stop_segment(self, repo, gri, log=None):
702	if not log: log = self.log
703	cmd = ['env', 'AXIS2_HOME=%s' % self.axis2_home, './run.sh',
704	'cancel', '-repo', repo , '-url', self.idc_url, '-gri', gri]
705
706
707	self.log.debug("[stop_segment]: %s" % " ".join(cmd))
708	if not self.create_debug:
709	try:
710	f = open("/dev/null", "w")
711	call(cmd, cwd=self.cli_dir, stdout=f, stderr=f, close_fds=True)
712	f.close()
713	except IOError, e:
714	raise service_error(service_error.internal,
715	"Failed to open /dev/null: %s" % e)
716
717	def export_store_info(self, cf, vlan, connInfo):
718	"""
719	For the export requests in the connection info, install the peer names
720	at the experiment controller via SetValue calls.
721	"""
722
723	for c in connInfo:
724	for p in [ p for p in c.get('parameter', []) \
725	if p.get('type', '') == 'output']:
726
727	if p.get('name', '') != 'vlan_id':
728	self.log.error("Unknown export parameter: %s" % \
729	p.get('name'))
730	continue
731
732	k = p.get('key', None)
733	surl = p.get('store', None)
734	if surl and k:
735	value = "%s" % vlan
736	req = { 'name': k, 'value': value }
737	print "calling SetValue %s %s" % (surl, req)
738	self.call_SetValue(surl, req, cf)
739	else:
740	self.log.error("Bad export request: %s" % p)
741
742	def import_store_info(self, cf, connInfo):
743	"""
744	Pull any import parameters in connInfo in. We translate them either
745	into known member names or fedAddrs.
746	"""
747
748	for c in connInfo:
749	for p in [ p for p in c.get('parameter', []) \
750	if p.get('type', '') == 'input']:
751	name = p.get('name', None)
752	key = p.get('key', None)
753	store = p.get('store', None)
754
755	if name and key and store :
756	req = { 'name': key, 'wait': True }
757	r = self.call_GetValue(store, req, cf)
758	r = r.get('GetValueResponseBody', None)
759	if r :
760	if r.get('name', '') == key:
761	v = r.get('value', None)
762	if v is not None:
763	if name == 'peer':
764	c['peer'] = v
765	else:
766	if c.has_key('fedAttr'):
767	c['fedAttr'].append({
768	'attribute': name, 'value': value})
769	else:
770	c['fedAttr']= [{
771	'attribute': name, 'value': value}]
772	else:
773	raise service_error(service_error.internal,
774	'None value exported for %s' % key)
775	else:
776	raise service_error(service_error.internal,
777	'Different name returned for %s: %s' \
778	% (key, r.get('name','')))
779	else:
780	raise service_error(service_error.internal,
781	'Badly formatted response: no GetValueResponseBody')
782	else:
783	raise service_error(service_error.internal,
784	'Bad Services missing info for import %s' % c)
785
786
787	def StartSegment(self, req, fid):
788	err = None # Any service_error generated after tmpdir is created
789	rv = None # Return value from segment creation
790
791	try:
792	req = req['StartSegmentRequestBody']
793	except KeyError:
794	raise service_error(server_error.req, "Badly formed request")
795
796	auth_attr = req['allocID']['fedid']
797	aid = "%s" % auth_attr
798	attrs = req.get('fedAttr', [])
799	if not self.auth.check_attribute(fid, auth_attr):
800	raise service_error(service_error.access, "Access denied")
801	else:
802	# See if this is a replay of an earlier succeeded StartSegment -
803	# sometimes SSL kills 'em. If so, replay the response rather than
804	# redoing the allocation.
805	self.state_lock.acquire()
806	retval = self.state[aid].get('started', None)
807	self.state_lock.release()
808	if retval:
809	self.log.warning("Duplicate StartSegment for %s: " % aid + \
810	"replaying response")
811	return retval
812
813	certfile = "%s/%s.pem" % (self.certdir, aid)
814
815	if req.has_key('segmentdescription') and \
816	req['segmentdescription'].has_key('topdldescription'):
817	topo = \
818	topdl.Topology(**req['segmentdescription']['topdldescription'])
819	else:
820	raise service_error(service_error.req,
821	"Request missing segmentdescription'")
822
823	connInfo = req.get('connection', [])
824
825	cap, src, dest, vlans = self.extract_parameters(topo)
826	ename = aid
827
828	for a in attrs:
829	if a['attribute'] == 'experiment_name':
830	ename = a['value']
831
832	repo = None
833	self.state_lock.acquire()
834	if self.state.has_key(aid):
835	repo = self.state[aid]['user']
836	self.state[aid]['log'] = [ ]
837	# Create a logger that logs to the experiment's state object as
838	# well as to the main log file.
839	alloc_log = logging.getLogger('fedd.access.%s' % ename)
840	h = logging.StreamHandler(
841	list_log.list_log(self.state[aid]['log']))
842	# XXX: there should be a global one of these rather than
843	# repeating the code.
844	h.setFormatter(logging.Formatter(
845	"%(asctime)s %(name)s %(message)s",
846	'%d %b %y %H:%M:%S'))
847	alloc_log.addHandler(h)
848	self.write_state()
849	self.state_lock.release()
850
851	if not repo:
852	raise service_error(service_error.internal,
853	"Can't find creation user for %s" %aid)
854
855	gri, vlan_no = self.start_segment(repo, src, dest, cap, vlans,
856	log=alloc_log)
857
858	self.export_store_info(certfile, vlan_no, connInfo)
859
860
861	if gri:
862	rtopo = topo.clone()
863	for s in rtopo.substrates:
864	s.set_attribute('vlan', vlan_no)
865	s.set_attribute('gri', gri)
866
867	# Grab the log (this is some anal locking, but better safe than
868	# sorry)
869	self.state_lock.acquire()
870	self.state[aid]['gri'] = gri
871	logv = "".join(self.state[aid]['log'])
872	# It's possible that the StartSegment call gets retried (!).
873	# if the 'started' key is in the allocation, we'll return it rather
874	# than redo the setup.
875	self.state[aid]['started'] = {
876	'allocID': req['allocID'],
877	'allocationLog': logv,
878	'segmentdescription': {
879	'topdldescription': rtopo.to_dict()
880	},
881	}
882	retval = self.state[aid]['started']
883	self.write_state()
884	self.state_lock.release()
885
886	return retval
887	elif err:
888	raise service_error(service_error.federant,
889	"Swapin failed: %s" % err)
890	else:
891	raise service_error(service_error.federant, "Swapin failed")
892
893	def TerminateSegment(self, req, fid):
894	try:
895	req = req['TerminateSegmentRequestBody']
896	except KeyError:
897	raise service_error(server_error.req, "Badly formed request")
898
899	auth_attr = req['allocID']['fedid']
900	aid = "%s" % auth_attr
901
902	self.log.debug("Terminate request for %s" %aid)
903	attrs = req.get('fedAttr', [])
904	if not self.auth.check_attribute(fid, auth_attr):
905	raise service_error(service_error.access, "Access denied")
906
907	self.state_lock.acquire()
908	if self.state.has_key(aid):
909	gri = self.state[aid].get('gri', None)
910	user = self.state[aid].get('user', None)
911	else:
912	gri = None
913	user = None
914	self.state_lock.release()
915	self.log.debug("Stop segment for user: %s gre %s" %(user, gri))
916
917	if not gri:
918	raise service_error(service_error.internal,
919	"Can't find gri for %s" % aid)
920
921	if not user:
922	raise service_error(service_error.internal,
923	"Can't find creation user for %s" % aid)
924
925	self.log.debug("Stop segment for GRI: %s" %gri)
926	self.stop_segment(user, gri)
927	return { 'allocID': req['allocID'] }

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

Download in other formats: