Context Navigation

source: fedd/federation/access.py @ 1da6a23

axis_examplecompt_changesinfo-opsversion-2.00version-3.01version-3.02

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

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

Download in other formats: