Context Navigation

source: fedd/federation/access.py @ ecca6eb

axis_examplecompt_changesinfo-opsversion-2.00version-3.01version-3.02

Last change on this file since ecca6eb was ecca6eb, checked in by Ted Faber <faber@…>, 15 years ago
checkpoint
Property mode set to `100644`
File size: 51.9 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
10	from threading import *
11	import subprocess
12	import time
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 topdl
23	import httplib
24	import tempfile
25	from urlparse import urlparse
26
27
28	# Make log messages disappear if noone configures a fedd logger
29	class nullHandler(logging.Handler):
30	def emit(self, record): pass
31
32	fl = logging.getLogger("fedd.access")
33	fl.addHandler(nullHandler())
34
35	class access:
36	"""
37	The implementation of access control based on mapping users to projects.
38
39	Users can be mapped to existing projects or have projects created
40	dynamically. This implements both direct requests and proxies.
41	"""
42
43	class parse_error(RuntimeError): pass
44
45
46	proxy_RequestAccess= service_caller('RequestAccess')
47	proxy_ReleaseAccess= service_caller('ReleaseAccess')
48
49	def __init__(self, config=None, auth=None):
50	"""
51	Initializer. Pulls parameters out of the ConfigParser's access section.
52	"""
53
54	# Make sure that the configuration is in place
55	if not config:
56	raise RunTimeError("No config to fedd.access")
57
58	self.project_priority = config.getboolean("access", "project_priority")
59	self.allow_proxy = config.getboolean("access", "allow_proxy")
60
61	self.boss = config.get("access", "boss")
62	self.ops = config.get("access", "ops")
63	self.domain = config.get("access", "domain")
64	self.fileserver = config.get("access", "fileserver")
65	self.eventserver = config.get("access", "eventserver")
66	self.certdir = config.get("access","certdir")
67	self.ssh_privkey_file = config.get("access","ssh_privkey_file")
68
69	self.attrs = { }
70	self.access = { }
71	self.restricted = [ ]
72	self.projects = { }
73	self.keys = { }
74	self.types = { }
75	self.allocation = { }
76	self.state = {
77	'projects': self.projects,
78	'allocation' : self.allocation,
79	'keys' : self.keys,
80	'types': self.types
81	}
82	self.log = logging.getLogger("fedd.access")
83	set_log_level(config, "access", self.log)
84	self.state_lock = Lock()
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.soap_services = {\
114	'RequestAccess': soap_handler("RequestAccess", self.RequestAccess),
115	'ReleaseAccess': soap_handler("ReleaseAccess", self.ReleaseAccess),
116	'StartSegment': soap_handler("StartSegment", self.StartSegment),
117	}
118	self.xmlrpc_services = {\
119	'RequestAccess': xmlrpc_handler('RequestAccess',
120	self.RequestAccess),
121	'ReleaseAccess': xmlrpc_handler('ReleaseAccess',
122	self.ReleaseAccess),
123	'StartSegment': xmlrpc_handler('StartSegment',
124	self.StartSegment),
125	}
126
127
128	if not config.has_option("allocate", "uri"):
129	self.allocate_project = \
130	allocate_project_local(config, auth)
131	else:
132	self.allocate_project = \
133	allocate_project_remote(config, auth)
134
135	# If the project allocator exports services, put them in this object's
136	# maps so that classes that instantiate this can call the services.
137	self.soap_services.update(self.allocate_project.soap_services)
138	self.xmlrpc_services.update(self.allocate_project.xmlrpc_services)
139
140
141	def read_access(self, config):
142	"""
143	Read a configuration file and set internal parameters.
144
145	The format is more complex than one might hope. The basic format is
146	attribute value pairs separated by colons(:) on a signle line. The
147	attributes in bool_attrs, emulab_attrs and id_attrs can all be set
148	directly using the name: value syntax. E.g.
149	boss: hostname
150	sets self.boss to hostname. In addition, there are access lines of the
151	form (tb, proj, user) -> (aproj, auser) that map the first tuple of
152	names to the second for access purposes. Names in the key (left side)
153	can include "<NONE> or <ANY>" to act as wildcards or to require the
154	fields to be empty. Similarly aproj or auser can be <SAME> or
155	<DYNAMIC> indicating that either the matching key is to be used or a
156	dynamic user or project will be created. These names can also be
157	federated IDs (fedid's) if prefixed with fedid:. Finally, the aproj
158	can be followed with a colon-separated list of node types to which that
159	project has access (or will have access if dynamic).
160	Testbed attributes outside the forms above can be given using the
161	format attribute: name value: value. The name is a single word and the
162	value continues to the end of the line. Empty lines and lines startin
163	with a # are ignored.
164
165	Parsing errors result in a self.parse_error exception being raised.
166	"""
167	lineno=0
168	name_expr = "["+string.ascii_letters + string.digits + "\.\-_]+"
169	fedid_expr = "fedid:[" + string.hexdigits + "]+"
170	key_name = "(<ANY>\|<NONE>\|"+fedid_expr + "\|"+ name_expr + ")"
171	access_proj = "(<DYNAMIC>(?::" + name_expr +")*\|"+ \
172	"<SAME>" + "(?::" + name_expr + ")*\|" + \
173	fedid_expr + "(?::" + name_expr + ")*\|" + \
174	name_expr + "(?::" + name_expr + ")*)"
175	access_name = "(<DYNAMIC>\|<SAME>\|" + fedid_expr + "\|"+ name_expr + ")"
176
177	restricted_re = re.compile("restricted:\s(.)", re.IGNORECASE)
178	attr_re = re.compile('attribute:\s([\._\-a-z0-9]+)\s+value:\s(.*)',
179	re.IGNORECASE)
180	access_re = re.compile('\('+key_name+'\s,\s'+key_name+'\s,\s'+
181	key_name+'\s\)\s->\s\('+access_proj + '\s,\s*' +
182	access_name + '\s,\s' + access_name + '\s*\)', re.IGNORECASE)
183
184	def parse_name(n):
185	if n.startswith('fedid:'): return fedid(hexstr=n[len('fedid:'):])
186	else: return n
187
188	def auth_name(n):
189	if isinstance(n, basestring):
190	if n =='<any>' or n =='<none>': return None
191	else: return unicode(n)
192	else:
193	return n
194
195	f = open(config, "r");
196	for line in f:
197	lineno += 1
198	line = line.strip();
199	if len(line) == 0 or line.startswith('#'):
200	continue
201
202	# Extended (attribute: x value: y) attribute line
203	m = attr_re.match(line)
204	if m != None:
205	attr, val = m.group(1,2)
206	self.attrs[attr] = val
207	continue
208
209	# Restricted entry
210	m = restricted_re.match(line)
211	if m != None:
212	val = m.group(1)
213	self.restricted.append(val)
214	continue
215
216	# Access line (t, p, u) -> (ap, cu, su) line
217	m = access_re.match(line)
218	if m != None:
219	access_key = tuple([ parse_name(x) for x in m.group(1,2,3)])
220	auth_key = tuple([ auth_name(x) for x in access_key])
221	aps = m.group(4).split(":");
222	if aps[0] == 'fedid:':
223	del aps[0]
224	aps[0] = fedid(hexstr=aps[0])
225
226	cu = parse_name(m.group(5))
227	su = parse_name(m.group(6))
228
229	access_val = (access_project(aps[0], aps[1:]),
230	parse_name(m.group(5)), parse_name(m.group(6)))
231
232	self.access[access_key] = access_val
233	self.auth.set_attribute(auth_key, "access")
234	continue
235
236	# Nothing matched to here: unknown line - raise exception
237	f.close()
238	raise self.parse_error("Unknown statement at line %d of %s" % \
239	(lineno, config))
240	f.close()
241
242	def get_users(self, obj):
243	"""
244	Return a list of the IDs of the users in dict
245	"""
246	if obj.has_key('user'):
247	return [ unpack_id(u['userID']) \
248	for u in obj['user'] if u.has_key('userID') ]
249	else:
250	return None
251
252	def write_state(self):
253	if self.state_filename:
254	try:
255	f = open(self.state_filename, 'w')
256	pickle.dump(self.state, f)
257	except IOError, e:
258	self.log.error("Can't write file %s: %s" % \
259	(self.state_filename, e))
260	except pickle.PicklingError, e:
261	self.log.error("Pickling problem: %s" % e)
262	except TypeError, e:
263	self.log.error("Pickling problem (TypeError): %s" % e)
264
265
266	def read_state(self):
267	"""
268	Read a new copy of access state. Old state is overwritten.
269
270	State format is a simple pickling of the state dictionary.
271	"""
272	if self.state_filename:
273	try:
274	f = open(self.state_filename, "r")
275	self.state = pickle.load(f)
276
277	self.allocation = self.state['allocation']
278	self.projects = self.state['projects']
279	self.keys = self.state['keys']
280	self.types = self.state['types']
281
282	self.log.debug("[read_state]: Read state from %s" % \
283	self.state_filename)
284	except IOError, e:
285	self.log.warning(("[read_state]: No saved state: " +\
286	"Can't open %s: %s") % (self.state_filename, e))
287	except EOFError, e:
288	self.log.warning(("[read_state]: " +\
289	"Empty or damaged state file: %s:") % \
290	self.state_filename)
291	except pickle.UnpicklingError, e:
292	self.log.warning(("[read_state]: No saved state: " + \
293	"Unpickling failed: %s") % e)
294
295	# Add the ownership attributes to the authorizer. Note that the
296	# indices of the allocation dict are strings, but the attributes are
297	# fedids, so there is a conversion.
298	for k in self.allocation.keys():
299	for o in self.allocation[k].get('owners', []):
300	self.auth.set_attribute(o, fedid(hexstr=k))
301
302
303	def permute_wildcards(self, a, p):
304	"""Return a copy of a with various fields wildcarded.
305
306	The bits of p control the wildcards. A set bit is a wildcard
307	replacement with the lowest bit being user then project then testbed.
308	"""
309	if p & 1: user = ["<any>"]
310	else: user = a[2]
311	if p & 2: proj = "<any>"
312	else: proj = a[1]
313	if p & 4: tb = "<any>"
314	else: tb = a[0]
315
316	return (tb, proj, user)
317
318	def find_access(self, search):
319	"""
320	Search the access DB for a match on this tuple. Return the matching
321	access tuple and the user that matched.
322
323	NB, if the initial tuple fails to match we start inserting wildcards in
324	an order determined by self.project_priority. Try the list of users in
325	order (when wildcarded, there's only one user in the list).
326	"""
327	if self.project_priority: perm = (0, 1, 2, 3, 4, 5, 6, 7)
328	else: perm = (0, 2, 1, 3, 4, 6, 5, 7)
329
330	for p in perm:
331	s = self.permute_wildcards(search, p)
332	# s[2] is None on an anonymous, unwildcarded request
333	if s[2] != None:
334	for u in s[2]:
335	if self.access.has_key((s[0], s[1], u)):
336	return (self.access[(s[0], s[1], u)], u)
337	else:
338	if self.access.has_key(s):
339	return (self.access[s], None)
340	return None, None
341
342	def lookup_access(self, req, fid):
343	"""
344	Determine the allowed access for this request. Return the access and
345	which fields are dynamic.
346
347	The fedid is needed to construct the request
348	"""
349	# Search keys
350	tb = None
351	project = None
352	user = None
353	# Return values
354	rp = access_project(None, ())
355	ru = None
356
357	if req.has_key('project'):
358	p = req['project']
359	if p.has_key('name'):
360	project = unpack_id(p['name'])
361	user = self.get_users(p)
362	else:
363	user = self.get_users(req)
364
365	user_fedids = [ u for u in user if isinstance(u, fedid)]
366	# Determine how the caller is representing itself. If its fedid shows
367	# up as a project or a singleton user, let that stand. If neither the
368	# usernames nor the project name is a fedid, the caller is a testbed.
369	if project and isinstance(project, fedid):
370	if project == fid:
371	# The caller is the project (which is already in the tuple
372	# passed in to the authorizer)
373	owners = user_fedids
374	owners.append(project)
375	else:
376	raise service_error(service_error.req,
377	"Project asserting different fedid")
378	else:
379	if fid not in user_fedids:
380	tb = fid
381	owners = user_fedids
382	owners.append(fid)
383	else:
384	if len(fedids) > 1:
385	raise service_error(service_error.req,
386	"User asserting different fedid")
387	else:
388	# Which is a singleton
389	owners = user_fedids
390	# Confirm authorization
391
392	for u in user:
393	self.log.debug("[lookup_access] Checking access for %s" % \
394	((tb, project, u),))
395	if self.auth.check_attribute((tb, project, u), 'access'):
396	self.log.debug("[lookup_access] Access granted")
397	break
398	else:
399	self.log.debug("[lookup_access] Access Denied")
400	else:
401	raise service_error(service_error.access, "Access denied")
402
403	# This maps a valid user to the Emulab projects and users to use
404	found, user_match = self.find_access((tb, project, user))
405
406	if found == None:
407	raise service_error(service_error.access,
408	"Access denied - cannot map access")
409
410	# resolve <dynamic> and <same> in found
411	dyn_proj = False
412	dyn_create_user = False
413	dyn_service_user = False
414
415	if found[0].name == "<same>":
416	if project != None:
417	rp.name = project
418	else :
419	raise service_error(\
420	service_error.server_config,
421	"Project matched <same> when no project given")
422	elif found[0].name == "<dynamic>":
423	rp.name = None
424	dyn_proj = True
425	else:
426	rp.name = found[0].name
427	rp.node_types = found[0].node_types;
428
429	if found[1] == "<same>":
430	if user_match == "<any>":
431	if user != None: rcu = user[0]
432	else: raise service_error(\
433	service_error.server_config,
434	"Matched <same> on anonymous request")
435	else:
436	rcu = user_match
437	elif found[1] == "<dynamic>":
438	rcu = None
439	dyn_create_user = True
440	else:
441	rcu = found[1]
442
443	if found[2] == "<same>":
444	if user_match == "<any>":
445	if user != None: rsu = user[0]
446	else: raise service_error(\
447	service_error.server_config,
448	"Matched <same> on anonymous request")
449	else:
450	rsu = user_match
451	elif found[2] == "<dynamic>":
452	rsu = None
453	dyn_service_user = True
454	else:
455	rsu = found[2]
456
457	return (rp, rcu, rsu), (dyn_create_user, dyn_service_user, dyn_proj),\
458	owners
459
460	def build_response(self, alloc_id, ap):
461	"""
462	Create the SOAP response.
463
464	Build the dictionary description of the response and use
465	fedd_utils.pack_soap to create the soap message. ap is the allocate
466	project message returned from a remote project allocation (even if that
467	allocation was done locally).
468	"""
469	# Because alloc_id is already a fedd_services_types.IDType_Holder,
470	# there's no need to repack it
471	msg = {
472	'allocID': alloc_id,
473	'emulab': {
474	'domain': self.domain,
475	'boss': self.boss,
476	'ops': self.ops,
477	'fileServer': self.fileserver,
478	'eventServer': self.eventserver,
479	'project': ap['project']
480	},
481	}
482	if len(self.attrs) > 0:
483	msg['emulab']['fedAttr'] = \
484	[ { 'attribute': x, 'value' : y } \
485	for x,y in self.attrs.iteritems()]
486	return msg
487
488	def RequestAccess(self, req, fid):
489	"""
490	Handle the access request. Proxy if not for us.
491
492	Parse out the fields and make the allocations or rejections if for us,
493	otherwise, assuming we're willing to proxy, proxy the request out.
494	"""
495
496	def gateway_hardware(h):
497	if h == 'GWTYPE': return self.attrs.get('connectorType', 'GWTYPE')
498	else: return h
499
500	# The dance to get into the request body
501	if req.has_key('RequestAccessRequestBody'):
502	req = req['RequestAccessRequestBody']
503	else:
504	raise service_error(service_error.req, "No request!?")
505
506	if req.has_key('destinationTestbed'):
507	dt = unpack_id(req['destinationTestbed'])
508
509	if dt == None or dt in self.testbed:
510	# Request for this fedd
511	found, dyn, owners = self.lookup_access(req, fid)
512	restricted = None
513	ap = None
514
515	# If this is a request to export a project and the access project
516	# is not the project to export, access denied.
517	if req.has_key('exportProject'):
518	ep = unpack_id(req['exportProject'])
519	if ep != found[0].name:
520	raise service_error(service_error.access,
521	"Cannot export %s" % ep)
522
523	# Check for access to restricted nodes
524	if req.has_key('resources') and req['resources'].has_key('node'):
525	resources = req['resources']
526	restricted = [ gateway_hardware(t) for n in resources['node'] \
527	if n.has_key('hardware') \
528	for t in n['hardware'] \
529	if gateway_hardware(t) \
530	in self.restricted ]
531	inaccessible = [ t for t in restricted \
532	if t not in found[0].node_types]
533	if len(inaccessible) > 0:
534	raise service_error(service_error.access,
535	"Access denied (nodetypes %s)" % \
536	str(', ').join(inaccessible))
537	# These collect the keys for the two roles into single sets, one
538	# for creation and one for service. The sets are a simple way to
539	# eliminate duplicates
540	create_ssh = set([ x['sshPubkey'] \
541	for x in req['createAccess'] \
542	if x.has_key('sshPubkey')])
543
544	service_ssh = set([ x['sshPubkey'] \
545	for x in req['serviceAccess'] \
546	if x.has_key('sshPubkey')])
547
548	if len(create_ssh) > 0 and len(service_ssh) >0:
549	if dyn[1]:
550	# Compose the dynamic project request
551	# (only dynamic, dynamic currently allowed)
552	preq = { 'AllocateProjectRequestBody': \
553	{ 'project' : {\
554	'user': [ \
555	{ \
556	'access': [ { 'sshPubkey': s } \
557	for s in service_ssh ],
558	'role': "serviceAccess",\
559	}, \
560	{ \
561	'access': [ { 'sshPubkey': s } \
562	for s in create_ssh ],
563	'role': "experimentCreation",\
564	}, \
565	], \
566	}\
567	}\
568	}
569	if restricted != None and len(restricted) > 0:
570	preq['AllocateProjectRequestBody']['resources'] = \
571	{'node': [ { 'hardware' : [ h ] } \
572	for h in restricted ] }
573	ap = self.allocate_project.dynamic_project(preq)
574	else:
575	preq = {'StaticProjectRequestBody' : \
576	{ 'project': \
577	{ 'name' : { 'localname' : found[0].name },\
578	'user' : [ \
579	{\
580	'userID': { 'localname' : found[1] }, \
581	'access': [ { 'sshPubkey': s }
582	for s in create_ssh ],
583	'role': 'experimentCreation'\
584	},\
585	{\
586	'userID': { 'localname' : found[2] }, \
587	'access': [ { 'sshPubkey': s }
588	for s in service_ssh ],
589	'role': 'serviceAccess'\
590	},\
591	]}\
592	}\
593	}
594	if restricted != None and len(restricted) > 0:
595	preq['StaticProjectRequestBody']['resources'] = \
596	{'node': [ { 'hardware' : [ h ] } \
597	for h in restricted ] }
598	ap = self.allocate_project.static_project(preq)
599	else:
600	raise service_error(service_error.req,
601	"SSH access parameters required")
602	# keep track of what's been added
603	allocID, alloc_cert = generate_fedid(subj="alloc", log=self.log)
604	aid = unicode(allocID)
605
606	self.state_lock.acquire()
607	self.allocation[aid] = { }
608	try:
609	pname = ap['project']['name']['localname']
610	except KeyError:
611	pname = None
612
613	if dyn[1]:
614	if not pname:
615	self.state_lock.release()
616	raise service_error(service_error.internal,
617	"Misformed allocation response?")
618	if self.projects.has_key(pname): self.projects[pname] += 1
619	else: self.projects[pname] = 1
620	self.allocation[aid]['project'] = pname
621	else:
622	# sproject is a static project associated with this allocation.
623	self.allocation[aid]['sproject'] = pname
624
625	if ap.has_key('resources'):
626	if not pname:
627	self.state_lock.release()
628	raise service_error(service_error.internal,
629	"Misformed allocation response?")
630	self.allocation[aid]['types'] = set()
631	nodes = ap['resources'].get('node', [])
632	for n in nodes:
633	for h in n.get('hardware', []):
634	if self.types.has_key((pname, h)):
635	self.types[(pname, h)] += 1
636	else:
637	self.types[(pname, h)] = 1
638	self.allocation[aid]['types'].add((pname,h))
639
640
641	self.allocation[aid]['keys'] = [ ]
642
643	try:
644	for u in ap['project']['user']:
645	uname = u['userID']['localname']
646	if u['role'] == 'experimentCreation':
647	self.allocation[aid]['user'] = uname
648	for k in [ k['sshPubkey'] for k in u['access'] \
649	if k.has_key('sshPubkey') ]:
650	kv = "%s:%s" % (uname, k)
651	if self.keys.has_key(kv): self.keys[kv] += 1
652	else: self.keys[kv] = 1
653	self.allocation[aid]['keys'].append((uname, k))
654	except KeyError:
655	self.state_lock.release()
656	raise service_error(service_error.internal,
657	"Misformed allocation response?")
658
659
660	self.allocation[aid]['owners'] = owners
661	self.write_state()
662	self.state_lock.release()
663	for o in owners:
664	self.auth.set_attribute(o, allocID)
665	try:
666	f = open("%s/%s.pem" % (self.certdir, aid), "w")
667	print >>f, alloc_cert
668	f.close()
669	except IOError, e:
670	raise service_error(service_error.internal,
671	"Can't open %s/%s : %s" % (self.certdir, aid, e))
672	resp = self.build_response({ 'fedid': allocID } , ap)
673	return resp
674	else:
675	if self.allow_proxy:
676	resp = self.proxy_RequestAccess.call_service(dt, req,
677	self.cert_file, self.cert_pwd,
678	self.trusted_certs)
679	if resp.has_key('RequestAccessResponseBody'):
680	return resp['RequestAccessResponseBody']
681	else:
682	return None
683	else:
684	raise service_error(service_error.access,
685	"Access proxying denied")
686
687	def ReleaseAccess(self, req, fid):
688	# The dance to get into the request body
689	if req.has_key('ReleaseAccessRequestBody'):
690	req = req['ReleaseAccessRequestBody']
691	else:
692	raise service_error(service_error.req, "No request!?")
693
694	if req.has_key('destinationTestbed'):
695	dt = unpack_id(req['destinationTestbed'])
696	else:
697	dt = None
698
699	if dt == None or dt in self.testbed:
700	# Local request
701	try:
702	if req['allocID'].has_key('localname'):
703	auth_attr = aid = req['allocID']['localname']
704	elif req['allocID'].has_key('fedid'):
705	aid = unicode(req['allocID']['fedid'])
706	auth_attr = req['allocID']['fedid']
707	else:
708	raise service_error(service_error.req,
709	"Only localnames and fedids are understood")
710	except KeyError:
711	raise service_error(service_error.req, "Badly formed request")
712
713	self.log.debug("[access] deallocation requested for %s", aid)
714	if not self.auth.check_attribute(fid, auth_attr):
715	self.log.debug("[access] deallocation denied for %s", aid)
716	raise service_error(service_error.access, "Access Denied")
717
718	# If we know this allocation, reduce the reference counts and
719	# remove the local allocations. Otherwise report an error. If
720	# there is an allocation to delete, del_users will be a dictonary
721	# of sets where the key is the user that owns the keys in the set.
722	# We use a set to avoid duplicates. del_project is just the name
723	# of any dynamic project to delete. We're somewhat lazy about
724	# deleting authorization attributes. Having access to something
725	# that doesn't exist isn't harmful.
726	del_users = { }
727	del_project = None
728	del_types = set()
729
730	if self.allocation.has_key(aid):
731	self.log.debug("Found allocation for %s" %aid)
732	self.state_lock.acquire()
733	for k in self.allocation[aid]['keys']:
734	kk = "%s:%s" % k
735	self.keys[kk] -= 1
736	if self.keys[kk] == 0:
737	if not del_users.has_key(k[0]):
738	del_users[k[0]] = set()
739	del_users[k[0]].add(k[1])
740	del self.keys[kk]
741
742	if self.allocation[aid].has_key('project'):
743	pname = self.allocation[aid]['project']
744	self.projects[pname] -= 1
745	if self.projects[pname] == 0:
746	del_project = pname
747	del self.projects[pname]
748
749	if self.allocation[aid].has_key('types'):
750	for t in self.allocation[aid]['types']:
751	self.types[t] -= 1
752	if self.types[t] == 0:
753	if not del_project: del_project = t[0]
754	del_types.add(t[1])
755	del self.types[t]
756
757	del self.allocation[aid]
758	self.write_state()
759	self.state_lock.release()
760	# If we actually have resources to deallocate, prepare the call.
761	if del_project or del_users:
762	msg = { 'project': { }}
763	if del_project:
764	msg['project']['name']= {'localname': del_project}
765	users = [ ]
766	for u in del_users.keys():
767	users.append({ 'userID': { 'localname': u },\
768	'access' : \
769	[ {'sshPubkey' : s } for s in del_users[u]]\
770	})
771	if users:
772	msg['project']['user'] = users
773	if len(del_types) > 0:
774	msg['resources'] = { 'node': \
775	[ {'hardware': [ h ] } for h in del_types ]\
776	}
777	if self.allocate_project.release_project:
778	msg = { 'ReleaseProjectRequestBody' : msg}
779	self.allocate_project.release_project(msg)
780	return { 'allocID': req['allocID'] }
781	else:
782	raise service_error(service_error.req, "No such allocation")
783
784	else:
785	if self.allow_proxy:
786	resp = self.proxy_ReleaseAccess.call_service(dt, req,
787	self.cert_file, self.cert_pwd,
788	self.trusted_certs)
789	if resp.has_key('ReleaseAccessResponseBody'):
790	return resp['ReleaseAccessResponseBody']
791	else:
792	return None
793	else:
794	raise service_error(service_error.access,
795	"Access proxying denied")
796
797
798
799	class proxy_emulab_segment:
800	class ssh_cmd_timeout(RuntimeError): pass
801
802	def __init__(self, log=None, keyfile=None, debug=False):
803	self.log = log or logging.getLogger(\
804	'fedd.access.proxy_emulab_segment')
805	self.ssh_privkey_file = keyfile
806	self.debug = debug
807	self.ssh_exec="/usr/bin/ssh"
808	self.scp_exec = "/usr/bin/scp"
809	self.ssh_cmd_timeout = access.proxy_emulab_segment.ssh_cmd_timeout
810
811	def scp_file(self, file, user, host, dest=""):
812	"""
813	scp a file to the remote host. If debug is set the action is only
814	logged.
815	"""
816
817	scp_cmd = [self.scp_exec, '-o', 'IdentitiesOnly yes',
818	'-o', 'StrictHostKeyChecking yes', '-i',
819	self.ssh_privkey_file, file,
820	"%s@%s:%s" % (user, host, dest)]
821	rv = 0
822
823	try:
824	dnull = open("/dev/null", "w")
825	except IOError:
826	self.log.debug("[ssh_file]: failed to open " + \
827	"/dev/null for redirect")
828	dnull = Null
829
830	self.log.debug("[scp_file]: %s" % " ".join(scp_cmd))
831	if not self.debug:
832	rv = subprocess.call(scp_cmd, stdout=dnull,
833	stderr=dnull, close_fds=True, close_fds=True)
834
835	return rv == 0
836
837	def ssh_cmd(self, user, host, cmd, wname=None, timeout=None):
838	"""
839	Run a remote command on host as user. If debug is set, the action
840	is only logged. Commands are run without stdin, to avoid stray
841	SIGTTINs.
842	"""
843	sh_str = ("%s -n -o 'IdentitiesOnly yes' -o " + \
844	"'StrictHostKeyChecking yes' -i %s %s@%s %s") % \
845	(self.ssh_exec, self.ssh_privkey_file,
846	user, host, cmd)
847
848	try:
849	dnull = open("/dev/null", "w")
850	except IOError:
851	self.log.debug("[ssh_cmd]: failed to open /dev/null " + \
852	"for redirect")
853	dnull = Null
854
855	self.log.debug("[ssh_cmd]: %s" % sh_str)
856	if not self.debug:
857	if dnull:
858	sub = subprocess.Popen(sh_str, shell=True, stdout=dnull,
859	stderr=dnull, close_fds=True)
860	else:
861	sub = subprocess.Popen(sh_str, shell=True, close_fds=True)
862	if timeout:
863	i = 0
864	rv = sub.poll()
865	while i < timeout:
866	if rv is not None: break
867	else:
868	time.sleep(1)
869	rv = sub.poll()
870	i += 1
871	else:
872	self.log.debug("Process exceeded runtime: %s" % sh_str)
873	os.kill(sub.pid, signal.SIGKILL)
874	raise self.ssh_cmd_timeout();
875	return rv == 0
876	else:
877	return sub.wait() == 0
878	else:
879	if timeout == 0:
880	self.log.debug("debug timeout raised on %s " % sh_str)
881	raise self.ssh_cmd_timeout()
882	else:
883	return True
884
885	class start_segment(proxy_emulab_segment):
886	def __init__(self, log=None, keyfile=None, debug=False):
887	access.proxy_emulab_segment.__init__(self, log=log,
888	keyfile=keyfile, debug=debug)
889	self.null = """
890	set ns [new Simulator]
891	source tb_compat.tcl
892
893	set a [$ns node]
894
895	$ns rtproto Session
896	$ns run
897	"""
898
899	def get_state(self, user, host, pid, eid):
900	# command to test experiment state
901	expinfo_exec = "/usr/testbed/bin/expinfo"
902	# Regular expressions to parse the expinfo response
903	state_re = re.compile("State:\s+(\w+)")
904	no_exp_re = re.compile("^No\s+such\s+experiment")
905	swapping_re = re.compile("^No\s+information\s+available.")
906	state = None # Experiment state parsed from expinfo
907	# The expinfo ssh command. Note the identity restriction to use
908	# only the identity provided in the pubkey given.
909	cmd = [self.ssh_exec, '-o', 'IdentitiesOnly yes', '-o',
910	'StrictHostKeyChecking yes', '-i',
911	self.ssh_privkey_file, "%s@%s" % (user, host),
912	expinfo_exec, pid, eid]
913
914	dev_null = None
915	try:
916	dev_null = open("/dev/null", "a")
917	except IOError, e:
918	self.log.error("[get_state]: can't open /dev/null: %s" %e)
919
920	if self.debug:
921	state = 'swapped'
922	rv = 0
923	else:
924	self.log.debug("Checking state")
925	status = subprocess.Popen(cmd, stdout=subprocess.PIPE,
926	stderr=dev_null, close_fds=True)
927	for line in status.stdout:
928	m = state_re.match(line)
929	if m: state = m.group(1)
930	else:
931	for reg, st in ((no_exp_re, "none"),
932	(swapping_re, "swapping")):
933	m = reg.match(line)
934	if m: state = st
935	rv = status.wait()
936
937	# If the experiment is not present the subcommand returns a
938	# non-zero return value. If we successfully parsed a "none"
939	# outcome, ignore the return code.
940	if rv != 0 and state != 'none':
941	raise service_error(service_error.internal,
942	"Cannot get status of segment:%s/%s" % (pid, eid))
943	elif state not in ('active', 'swapped', 'swapping', 'none'):
944	raise service_error(service_error.internal,
945	"Cannot get status of segment:%s/%s" % (pid, eid))
946	else:
947	self.log.debug("State is %s" % state)
948	return state
949
950
951	def __call__(self, parent, eid, pid, user, tclfile, tmpdir, timeout=0):
952	"""
953	Start a sub-experiment on a federant.
954
955	Get the current state, modify or create as appropriate, ship data
956	and configs and start the experiment. There are small ordering
957	differences based on the initial state of the sub-experiment.
958	"""
959	# ops node in the federant
960	host = "%s%s" % (parent.ops, parent.domain)
961	# Configuration directories on the remote machine
962	proj_dir = "/proj/%s/exp/%s/tmp" % (pid, eid)
963	softdir = "/proj/%s/software/%s" % (pid, eid)
964	# Local software dir
965	lsoftdir = "%s/software" % tmpdir
966
967	state = self.get_state(user, host, pid, eid)
968
969	if not self.scp_file(tclfile, user, host):
970	return False
971
972	if state == 'none':
973	# Create a null copy of the experiment so that we capture any
974	# logs there if the modify fails. Emulab software discards the
975	# logs from a failed startexp
976	try:
977	f = open("%s/null.tcl" % tmpdir, "w")
978	print >>f, self.null
979	f.close()
980	except IOError, e:
981	raise service_error(service_error.internal,
982	"Cannot stage tarfile/rpm: %s" % e.strerror)
983
984	if not self.scp_file("%s/null.tcl" % tmpdir, user, host):
985	return False
986	self.log.info("[start_segment]: Creating %s" % eid)
987	timedout = False
988	try:
989	if not self.ssh_cmd(user, host,
990	("/usr/testbed/bin/startexp -i -f -w -p %s " +
991	"-e %s null.tcl") % (pid, eid), "startexp",
992	timeout=60 * 10):
993	return False
994	except self.ssh_cmd_timeout:
995	timedout = True
996
997	if timedout:
998	state = self.get_state(user, host, pid, eid)
999	if state != "swapped":
1000	return False
1001
1002	# Open up a temporary file to contain a script for setting up the
1003	# filespace for the new experiment.
1004	self.log.info("[start_segment]: creating script file")
1005	try:
1006	sf, scriptname = tempfile.mkstemp()
1007	scriptfile = os.fdopen(sf, 'w')
1008	except IOError:
1009	return False
1010
1011	scriptbase = os.path.basename(scriptname)
1012
1013	# Script the filesystem changes
1014	print >>scriptfile, "/bin/rm -rf %s" % proj_dir
1015	# Clear and create the software directory
1016	print >>scriptfile, "/bin/rm -rf %s/*" % softdir
1017	print >>scriptfile, 'mkdir -p %s' % proj_dir
1018	if os.path.isdir(lsoftdir):
1019	print >>scriptfile, 'mkdir -p %s' % softdir
1020	print >>scriptfile, "rm -f %s" % scriptbase
1021	scriptfile.close()
1022
1023	# Move the script to the remote machine
1024	# XXX: could collide tempfile names on the remote host
1025	if self.scp_file(scriptname, user, host, scriptbase):
1026	os.remove(scriptname)
1027	else:
1028	return False
1029
1030	# Execute the script (and the script's last line deletes it)
1031	if not self.ssh_cmd(user, host, "sh -x %s" % scriptbase):
1032	return False
1033
1034	for f in os.listdir(tmpdir):
1035	if not os.path.isdir("%s/%s" % (tmpdir, f)):
1036	if not self.scp_file("%s/%s" % (tmpdir, f), user, host,
1037	"%s/%s" % (proj_dir, f)):
1038	return False
1039	if os.path.isdir(lsoftdir):
1040	for f in os.listdir(lsoftdir):
1041	if not os.path.isdir("%s/%s" % (lsoftdir, f)):
1042	if not self.scp_file("%s/%s" % (lsoftdir, f),
1043	user, host, "%s/%s" % (softdir, f)):
1044	return False
1045	# Stage the new configuration (active experiments will stay swapped
1046	# in now)
1047	self.log.info("[start_segment]: Modifying %s" % eid)
1048	try:
1049	if not self.ssh_cmd(user, host,
1050	"/usr/testbed/bin/modexp -r -s -w %s %s %s" % \
1051	(pid, eid, tclfile.rpartition('/')[2]),
1052	"modexp", timeout= 60 * 10):
1053	return False
1054	except self.ssh_cmd_timeout:
1055	self.log.error("Modify command failed to complete in time")
1056	# There's really no way to see if this succeeded or failed, so
1057	# if it hangs, assume the worst.
1058	return False
1059	# Active experiments are still swapped, this swaps the others in.
1060	if state != 'active':
1061	self.log.info("[start_segment]: Swapping %s" % eid)
1062	timedout = False
1063	try:
1064	if not self.ssh_cmd(user, host,
1065	"/usr/testbed/bin/swapexp -w %s %s in" % (pid, eid),
1066	"swapexp", timeout=10*60):
1067	return False
1068	except self.ssh_cmd_timeout:
1069	timedout = True
1070
1071	# If the command was terminated, but completed successfully,
1072	# report success.
1073	if timedout:
1074	self.log.debug("[start_segment]: swapin timed out " +\
1075	"checking state")
1076	state = self.get_state(user, host, pid, eid)
1077	self.log.debug("[start_segment]: state is %s" % state)
1078	return state == 'active'
1079	# Everything has gone OK.
1080	return True
1081
1082	class stop_segment(proxy_emulab_segment):
1083	def __init__(self, log=None, keyfile=None, debug=False):
1084	experiment_control_local.emulab_segment.__init__(self,
1085	log=log, keyfile=keyfile, debug=debug)
1086
1087	def __call__(self, tb, eid, tbparams):
1088	"""
1089	Stop a sub experiment by calling swapexp on the federant
1090	"""
1091	user = tbparams[tb]['user']
1092	host = "%s%s" % (tbparams[tb]['host'], tbparams[tb]['domain'])
1093	pid = tbparams[tb]['project']
1094
1095	self.log.info("[stop_segment]: Stopping %s on %s" % (eid, tb))
1096	rv = False
1097	try:
1098	# Clean out tar files: we've gone over quota in the past
1099	self.ssh_cmd(user, host, "rm -rf /proj/%s/rpms/%s" % (pid, eid))
1100	self.ssh_cmd(user, host, "rm -rf /proj/%s/tarfiles/%s" % \
1101	(pid, eid))
1102	rv = self.ssh_cmd(user, host,
1103	"/usr/testbed/bin/swapexp -w %s %s out" % (pid, eid))
1104	except self.ssh_cmd_timeout:
1105	rv = False
1106	return rv
1107
1108	def generate_portal_configs(self, topo, pubkey_base, secretkey_base,
1109	tmpdir, master):
1110
1111	seer_out = False
1112	for p in [ e for e in topo.elements \
1113	if isinstance(e, topdl.Computer) and e.get_attribute('portal')]:
1114	myname = e.name[0]
1115	peer = e.get_attribute('peer')
1116	lexp = e.get_attribute('experiment')
1117	lproj, leid = lexp.split('/', 1)
1118	ldomain = e.get_attribute('domain')
1119	mexp = e.get_attribute('masterexperiment')
1120	mproj, meid = mexp.split("/", 1)
1121	mdomain = e.get_attribute('masterdomain')
1122	scriptdir = e.get_attribute('scriptdir')
1123	active = e.get_attribute('active')
1124	type = e.get_attribute('portal_type')
1125	segid = fedid(hexstr=e.get_attribute('peer_segment'))
1126	for e in topo.elements:
1127	if isinstance(e, topdl.Segment) and e.id.fedid == segid:
1128	seg = e
1129	break
1130	else:
1131	raise service_error(service_error.req,
1132	"Can't find segment for portal %s" % myname)
1133
1134	rexp = seg.get_attribute('experiment')
1135	rproj, reid = rexp.split("/", 1)
1136	rdomain = seg.get_attribute('domain')
1137	cfn = "%s/%s.%s.%s%s.gw.conf" % \
1138	(tmpdir, myname, leid, lproj, ldomain)
1139	try:
1140	f = open(cfn, "w")
1141	print >>f, "Active: %s" % active
1142	print >>f, "BossName: boss"
1143	print >>f, "FsName: fs"
1144	print >>f, "EventServerName: event-server%s" % ldomain
1145	print >>f, "RemoteEventServerName: event-server%s" % rdomain
1146	print >>f, "SeerControl: control.%s.%s%s" % \
1147	(meid.lower(), mproj.lower(), mdomain)
1148	print >>f, "Type: %s" % type
1149	print >>f, "RemoteExperiment: %s" % rexp
1150	print >>f, "LocalExperiment: %s" % lexp
1151	print >>f, "RemoteConfigFile: " + \
1152	"/proj/%s/exp/%s/tmp/%s.%s.%s%s.gw.conf" \
1153	% (rproj, reid, peer.lower(), reid.lower(),
1154	rproj.lower(), rdomain)
1155	print >>f, "Peer: %s.%s.%s%s" % \
1156	(peer.lower(), reid.lower(), rproj.lower(), rdomain)
1157	print >>f, "Pubkeys: /proj/%s/exp/%s/tmp/%s" % \
1158	(lproj, leid, pubkey_base)
1159	print >>f, "Privkey: /proj/%s/exp/%s/tmp/%s" % \
1160	(lproj, leid, secretkey_base)
1161	f.close()
1162	except IOError, e:
1163	raise service_error(service_error.internal,
1164	"Can't write protal config %s: %s" % (cfn, e))
1165
1166	# XXX: This little seer config file needs to go away.
1167	if not seer_out:
1168	try:
1169	seerfn = "%s/seer.conf" % tmpdir
1170	f = open(seerfn, "w")
1171	if not master:
1172	print >>f, "ControlNode: control.%s.%s%s" % \
1173	(meid.lower(), mproj.lower(), mdomain)
1174	print >>f, "ExperimentID: %s" % mexp
1175	f.close()
1176	except IOError, e:
1177	raise service_error(service_error.internal,
1178	"Can't write seer.conf: %s" %e)
1179	seer_out = True
1180
1181
1182	def generate_client_conf(self, user, proj, eid, tmpdir):
1183	try:
1184	f = open("%s/client.conf" % tmpdir, "w")
1185	if self.attrs.has_key('SMBshare'):
1186	print >>f, "SMBshare: %s" % self.attrs['SMBshare']
1187	print >>f, "ProjectUser: %s" % user
1188	print >>f, "ProjectName: %s" % proj
1189	print >>f, "ExperimentID: %s/%s" % (proj, eid)
1190	f.close()
1191	except IOError, e:
1192	raise service_error(service_error.internal,
1193	"Cannot write client.conf: %s" %s)
1194
1195	def generate_ns2(self, topo, expfn, softdir, master):
1196	t = topo.clone()
1197
1198	# Weed out the things we aren't going to instantiate: Segments, portal
1199	# substrates, and portal interfaces. (The copi in the for loop allows
1200	# us to delete from e.elements in side the for loop).
1201	for e in [e for e in t.elements]:
1202	if isinstance(e, topdl.Segment):
1203	t.elements.remove(e)
1204	if isinstance(e, topdl.Computer):
1205	e.interface = [i for i in e.interface \
1206	if not i.get_attribute('portal')]
1207	t.substrates = [ s for s in t.substrates \
1208	if not s.get_attribute('portal')]
1209	t.incorporate_elements()
1210
1211	# Localize the software locations
1212	for e in t.elements:
1213	for s in getattr(e, 'software', []):
1214	s.location = re.sub("^.*/", softdir, s.location)
1215
1216
1217	# Customize the ns2 output for local portal commands and images
1218	filters = []
1219
1220	if master: cmdname = 'MasterConnectorCmd'
1221	else:cmdname = 'SlaveConnectorCmd'
1222
1223	if self.attrs.has_key(cmdname):
1224	filters.append(topdl.generate_portal_command_filter(
1225	self.attrs.get(cmdname)))
1226
1227	if self.attrs.has_key('connectorImage'):
1228	filters.append(topdl.generate_portal_image_filter(
1229	self.attrs.get('connectorImage')))
1230
1231	if self.attrs.has_key('connectorType'):
1232	filters.append(topdl.generate_portal_hardware_filter(
1233	self.attrs.get('connectorType')))
1234
1235	# Convert to ns and write it out
1236	expfile = topdl.topology_to_ns2(t, filters)
1237	try:
1238	f = open(expfn, "w")
1239	print >>f, expfile
1240	f.close()
1241	except IOError:
1242	raise service_error(service_error.internal,
1243	"Cannot write experiment file %s: %s" % (expfn,e))
1244
1245	def StartSegment(self, req, fid):
1246	def get_url(url, cf, tmpdir):
1247	po = urlparse(url)
1248	fn = po.path.rpartition('/')[2]
1249	try:
1250	conn = httplib.HTTPSConnection(po.hostname, port=po.port,
1251	cert_file=cf, key_file=cf)
1252	conn.putrequest('GET', po.path)
1253	conn.endheaders()
1254	response = conn.getresponse()
1255
1256	lf = open("%s/%s" % (tmpdir, fn), "w")
1257	buf = response.read(4096)
1258	while buf:
1259	lf.write(buf)
1260	buf = response.read(4096)
1261	lf.close()
1262	except IOError, e:
1263	raise service_error(service_error.internal,
1264	"Erro writing tempfile: %s" %e)
1265	except httplib.HTTPException, e:
1266	raise service_error(service_error.internal,
1267	"Error retrieving data: %s" % e)
1268
1269	configs = set(('hosts', 'ssh_pubkey', 'ssh_secretkey'))
1270
1271	try:
1272	req = req['StartSegmentRequestBody']
1273	except KeyError:
1274	raise service_error(server_error.req, "Badly formed request")
1275
1276	auth_attr = req['allocID']['fedid']
1277	aid = "%s" % auth_attr
1278	attrs = req.get('fedAttr', [])
1279	if not self.auth.check_attribute(fid, auth_attr):
1280	raise service_error(service_error.access, "Access denied")
1281
1282	if req.has_key('segmentdescription') and \
1283	req['segmentdescription'].has_key('topdldescription'):
1284	topo = \
1285	topdl.Topology(**req['segmentdescription']['topdldescription'])
1286	else:
1287	raise service_error(service_error.req,
1288	"Request missing segmentdescription'")
1289
1290	master = req.get('master', False)
1291
1292	certfile = "%s/%s.pem" % (self.certdir, auth_attr)
1293	try:
1294	tmpdir = tempfile.mkdtemp(prefix="access-")
1295	softdir = "%s/software" % tmpdir
1296	except IOError:
1297	raise service_error(service_error.internal, "Cannot create tmp dir")
1298
1299	sw = set()
1300	for e in topo.elements:
1301	for s in getattr(e, 'software', []):
1302	sw.add(s.location)
1303	if len(sw) > 0:
1304	os.mkdir(softdir)
1305	for s in sw:
1306	get_url(s, certfile, softdir)
1307
1308	for a in attrs:
1309	if a['attribute'] in configs:
1310	get_url(a['value'], certfile, tmpdir)
1311	if a['attribute'] == 'ssh_pubkey':
1312	pubkey_base = a['value'].rpartition('/')[2]
1313	if a['attribute'] == 'ssh_secretkey':
1314	secretkey_base = a['value'].rpartition('/')[2]
1315	if a['attribute'] == 'experiment_name':
1316	ename = a['value']
1317
1318	proj = None
1319	user = None
1320	self.state_lock.acquire()
1321	if self.allocation.has_key(aid):
1322	proj = self.allocation[aid].get('project', None)
1323	if not proj:
1324	proj = self.allocation[aid].get('sproject', None)
1325	user = self.allocation[aid].get('user', None)
1326	self.state_lock.release()
1327
1328	if not proj:
1329	raise service_error(service_error.internal,
1330	"Can't find project for %s" %aid)
1331
1332	if not user:
1333	raise service_error(service_error.internal,
1334	"Can't find creation user for %s" %aid)
1335
1336	expfile = "%s/experiment.tcl" % tmpdir
1337
1338	self.generate_portal_configs(topo, pubkey_base,
1339	secretkey_base, tmpdir, master)
1340	self.generate_ns2(topo, expfile,
1341	"/proj/%s/software/%s/" % (proj, ename), master)
1342	self.generate_client_conf(user, proj, ename, tmpdir)
1343	starter = self.start_segment(keyfile=self.ssh_privkey_file, debug=False)
1344	starter(self, ename, proj, user, expfile, tmpdir)
1345
1346	return { 'allocID': req['allocID'] }

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

Download in other formats: