Skip to content

GitLab

S
snf-occi
Commit 10f732f6 authored by Stavros Sachtouris's avatar Stavros Sachtouris
Browse files
Options

Remove astavoms from repo

parent d0ec3b10
Hide whitespace changes
Inline Side-by-side
Showing with 0 additions and 531 deletions
astavoms/__init__.py deleted 100644 → 0
View file @ d0ec3b10
# Copyright (C) 2012-2015 GRNET S.A.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from kamaki.clients import astakos, ClientError
from astavoms import vomsdir, identity
from inspect import getmembers, ismethod
class SnfOcciUsers(object):
"""Translate OCCI users to Synnefo users and manage their properties"""
def __init__(
self, snf_auth_url, snf_admin_token,
ldap_url, ldap_user, ldap_password, base_dn,
ca_cert_file=None):
self.snf_auth_url, self.snf_admin_token = snf_auth_url, snf_admin_token
self.snf_users = identity.IdentityClient(snf_auth_url, snf_admin_token)
self.snf_admin = astakos.AstakosClient(snf_auth_url, snf_admin_token)
self.snf_admin_id = self.snf_admin.user_info['id']
self.ldap_conf = dict(
ldap_url=ldap_url,
user=ldap_user, password=ldap_password,
base_dn=base_dn,
ca_cert_file=ca_cert_file)
@staticmethod
def dn_to_dict(user_dn):
"""
:param user_dn: a string, typically of the form
/C=<country>/O=<org name>/OU=<domain>/CN=<full user name>
:returns: a dict from a user_dn, all keys are uppercase
"""
return dict(map(
lambda x: (x[0].upper(), x[1]),
[i.split('=') for i in user_dn.split('/') if i]))
@staticmethod
def create_user_email(dn, vo):
user = SnfOcciUsers.dn_to_dict(dn)
full_name = ''.join(
[(c if (c.isalpha() or c.isdigit()) else '_') for c in user['CN']])
vo += '.' if vo else ''
return '%s%s@%s' % (vo, full_name, user['OU'])
def get_cached_user(self, dn, vo):
"""
:returns: (dict) user info from LDAP
:raises KeyError: if not in LDAP
"""
cn = self.dn_to_dict(dn)['CN']
with vomsdir.LDAPUser(**self.ldap_conf) as ldap_user:
k, v = ldap_user.search_by_vo(cn, vo).items()[0]
v['dn'] = k
return v
def cache_user(self, uuid, email, token, dn, vo, cert=None):
"""
:returns: (dict) updated user info from LDAP
"""
cn = self.dn_to_dict(dn)['CN']
with vomsdir.LDAPUser(**self.ldap_conf) as ldap_user:
ldap_user.create(uuid, cn, email, token, vo, dn, cert)
return self.get_cached_user(dn, vo)
def vo_to_project(self, vo):
"""
:returns: project with name==vo and owned by admin
:raises ClientError: 404 if not found
:raises ClientError: 409 if more than one projects match
"""
vo_projects = self.snf_admin.get_projects(
name=vo, owner=self.snf_admin_id)
if vo_projects:
if len(vo_projects) > 1:
raise ClientError(
'More than one projects matching name %s' % vo, 409)
return vo_projects[0]
raise ClientError('No projects matching name %s' % vo, 404)
@staticmethod
def split_full_name(full_name):
names = [name for name in full_name.split(' ') if name]
return names[0], ' '.join(names[1:])
def get_user_info(self, dn, vo):
""" If user not cached, attempt to create it
:returns: (dict) cached user info from LDAP directory
:raises ClientError: 404 if the project is not found
:raises ClientError: 409 if more than one projects match, new user
exists or new user is already enrolled to project
:raises ClientError: (tmp) 400 instead of 409, because of a server bug
:raises KeyError: if dn is not formated as expected
"""
try:
return self.get_cached_user(dn, vo)
except KeyError:
project = self.vo_to_project(vo)
email = self.create_user_email(dn, vo)
dn_dict = self.dn_to_dict(dn)
first, last = self.split_full_name(dn_dict['CN'])
user = self.snf_users.create_user(email, first, last, dn_dict['O'])
self.snf_admin.enroll_member(project['id'], email)
return self.cache_user(
user['id'], email, user['auth_token'], dn, vo)
def wrap_renew_token(self, cls, method, method_name, user_id):
"""Return a wrapped method with renew token feature"""
def wrap(*args, **kwargs):
try:
return method(*args, **kwargs)
except ClientError as ce:
if ce.status not in (401, ):
raise
print 'User', user_id, ':', ce
print 'Renew token and retry %s.%s(...)' % (
cls.__name__, method_name)
user = self.snf_users.renew_user_token(user_id)
with vomsdir.LDAPUser(**self.ldap_conf) as ldap_user:
ldap_user.update_token(user['id'], user['auth_token'])
cls.token = user['auth_token']
return method(*args, **kwargs)
wrap.__name__ = method_name
return wrap
def add_renew_token(self, cls, user_id):
""" In case of authentication failure, cls gets a new token and the
failed method is run again.
:returns: cls with all its methods wrapped
"""
cls_methods = [m for m in getmembers(cls) if (
not m[0].startswith('_')) and ismethod(m[1])]
for name, method in cls_methods:
wrap = self.wrap_renew_token(cls, method, name, user_id)
cls.__setattr__(name, wrap)
return cls
def uuid_from_token(self, token):
"""Lookup token in LDAP dir
:returns: uuid
:raises KeyError: if token not found
"""
with vomsdir.LDAPUser(**self.ldap_conf) as ldap_user:
r = ldap_user.search_by_token(token, ['uid', ])
if not r.values():
raise KeyError("Token not found in LDAP directory")
return r.values()[0]['uid'][0]
astavoms/cli.py deleted 100644 → 0
View file @ d0ec3b10
from kamaki.cli import command
from kamaki.cli.cmds import CommandInit, errors
from kamaki.cli.cmdtree import CommandTree
from kamaki.cli.argument import (
CommaSeparatedListArgument, ValueArgument, KeyValueArgument)
from astavoms.identity import IdentityClient
from kamaki.clients import ClientError
xuser_cmds = CommandTree('xuser', 'SNF User administrator commands')
# ldap_cmds = CommandInit('ldap', 'LDAP User commands')
namespaces = [xuser_cmds, ]
class _XuserInit(CommandInit):
"""Base class for all cuser commands"""
@property
def xuser(self):
self._xuser = getattr(self, '_xuser', IdentityClient(
self.astakos.endpoint_url, self.astakos.token))
return self._xuser
def user_by_email(self, email):
for user in self.xuser.list_users():
if user.get('email', None) == email:
return user
raise ClientError('Not Found', status=404)
@command(xuser_cmds)
class xuser_list(_XuserInit):
"""List snf users"""
arguments = dict(
select=CommaSeparatedListArgument('Keys to display', '--select'))
@errors.Generic.all
def _run(self):
users = self.xuser.list_users()
if self['select']:
users_ = []
for user in users:
user_ = [u for u in user.items() if u[0] in self['select']]
users_.append(dict(user_))
users = users_
self.print_list(users)
def main(self):
self._run()
@command(xuser_cmds)
class xuser_info(_XuserInit):
"""Info on an snf user"""
arguments = dict(
uuid=ValueArgument('Search by uuid', '--uuid'),
email=ValueArgument('Search by email', '--email'),
)
required = ['uuid', 'email']
@errors.Generic.all
def _run(self):
if self['uuid']:
self.print_dict(self.xuser.get_user_details(self['uuid']))
else:
self.print_dict(self.user_by_email(self['email']))
def main(self):
self._run()
@command(xuser_cmds)
class xuser_create(_XuserInit):
"""Create a new SNF user"""
arguments = dict(
email=ValueArgument('Full name', '--email'),
first_name=ValueArgument('First name', '--first-name'),
last_name=ValueArgument('Last name', '--last-name'),
affiliation=ValueArgument('Affiliation', '--affiliation'),
metadata=KeyValueArgument('Key=Value', '--metadata'),
)
required = ('email', 'first_name', 'last_name', 'affiliation')
@errors.Generic.all
def _run(self):
self.print_dict(self.xuser.create_user(
self['email'],
self['first_name'], self['last_name'],
self['affiliation'],
self['metadata'] or None))
def main(self):
self._run()
@command(xuser_cmds)
class xuser_modify(_XuserInit):
"""Modify user information"""
arguments = dict(
email=ValueArgument('Full name', '--email'),
first_name=ValueArgument('First name', '--first-name'),
last_name=ValueArgument('Last name', '--last-name'),
affiliation=ValueArgument('Affiliation', '--affiliation'),
metadata=KeyValueArgument('Key=Value', '--metadata'),
)
required = ['email', 'first_name', 'last_name', 'affiliation', 'metadata']
@errors.Generic.all
def _run(self, uuid):
self.print_dict(self.xuser.modify_user(
uuid,
self['email'] or None,
self['first_name'] or None, self['last_name'] or None,
self['affiliation'] or None,
self['metadata'] or None))
def main(self, uuid):
self._run(uuid)
@command(xuser_cmds)
class xuser_activate(_XuserInit):
"""Activate a user"""
arguments = dict(
uuid=ValueArgument('Search by uuid', '--uuid'),
email=ValueArgument('Search by email', '--email'),
)
required = ['uuid', 'email']
@errors.Generic.all
def _run(self):
uuid = self['uuid'] or self.user_by_email(self['email'])['id']
self.print_dict(self.xuser.activate_user(uuid))
def main(self):
self._run()
@command(xuser_cmds)
class xuser_deactivate(_XuserInit):
"""Deactivate a user"""
arguments = dict(
uuid=ValueArgument('Search by uuid', '--uuid'),
email=ValueArgument('Search by email', '--email'),
)
required = ['uuid', 'email']
@errors.Generic.all
def _run(self):
uuid = self['uuid'] or self.user_by_email(self['email'])['id']
self.print_dict(self.xuser.deactivate_user(uuid))
def main(self):
self._run()
@command(xuser_cmds)
class xuser_newtoken(_XuserInit):
"""Renew user token"""
arguments = dict(
uuid=ValueArgument('Search by uuid', '--uuid'),
email=ValueArgument('Search by email', '--email'),
)
required = ['uuid', 'email']
@errors.Generic.all
def _run(self):
uuid = self['uuid'] or self.user_by_email(self['email'])['id']
self.print_dict(self.xuser.renew_user_token(uuid))
def main(self):
self._run()
astavoms/identity.py deleted 100644 → 0
View file @ d0ec3b10
# Copyright (C) 2012-2015 GRNET S.A.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from kamaki.clients import Client
class IdentityClient(Client):
"""An Extended Identity Client"""
def list_users(self):
"""List all users"""
return self.get('users', success=200).json['users']
def create_user(
self, username, first_name, last_name, affiliation,
metadata=None):
"""Create a new user"""
kwargs = dict(
username=username,
first_name=first_name,
last_name=last_name,
affilication=affiliation)
if metadata:
kwargs['metadata'] = metadata
# Success should be 201, but server is currently returning 200, so...
r = self.post('users', json=dict(user=kwargs), success=(200, 201))
return r.json['user']
def get_user_details(self, user_id):
"""Get user details"""
return self.get('users/%s' % user_id, success=200).json['user']
def modify_user(
self, user_id,
username=None,
first_name=None,
last_name=None,
affilication=None,
password=None,
email=None,
metadata=None):
"""Modify User"""
kwargs = dict()
if username is not None:
kwargs['username'] = username
if first_name is not None:
kwargs['first_name'] = first_name
if last_name is not None:
kwargs['last_name'] = last_name
if affilication is not None:
kwargs['affilication'] = affilication
if password is not None:
kwargs['password'] = password
if email is not None:
kwargs['email'] = email
if metadata is not None:
kwargs['metadata'] = metadata
r = self.put('users/%s' % user_id, json=dict(user=kwargs), success=200)
return r.json['user']
def activate_user(self, user_id):
"""Activate a user"""
r = self.post(
'users/%s/action' % user_id, json=dict(activate={}), success=200)
return r.json['user']
def deactivate_user(self, user_id):
"""Deactivate a user"""
r = self.post(
'users/%s/action' % user_id, json=dict(deactivate={}), success=200)
return r.json['user']
def renew_user_token(self, user_id):
"""Renew user authentication token"""
r = self.post(
'users/%s/action' % user_id, json=dict(renewToken={}), success=200)
return r.json['user']
astavoms/vomsdir.py deleted 100644 → 0
View file @ d0ec3b10
# Copyright (C) 2012-2015 GRNET S.A.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import ldap
import ssl
class LDAPUser:
"""An LDAP manager for VOMS users"""
def __init__(self, ldap_url, user, password, base_dn, ca_cert_file=None):
"""
:raises ldap.LDAPError: if connection fails
"""
self.con = ldap.initialize(ldap_url)
self.base_dn = base_dn
if ca_cert_file:
ldap.set_option(ldap.OPT_X_TLS_CACERTFILE, ca_cert_file)
self.con.start_tls_s()
self.user, self.password = user, password
def __enter__(self):
self.con.simple_bind_s(self.user, self.password)
return self
def __exit__(self, type, value, traceback):
self.con.unbind()
def _search(self, query, attrlist):
return self.con.search_s(
self.base_dn, ldap.SCOPE_SUBTREE, query, attrlist)
def search_by_uid(self, userUID, attrlist=[]):
"""
:return: (dict) of the form dict(dn={...})
"""
query = '(&(objectclass=person)(uid=%s))' % userUID
return dict(self._search(query, attrlist))
def search_by_vo(self, user_cn, user_vo, attrlist=[]):
"""
:return: (dict) of the form dict(dn={...})
"""
query = '(&(objectclass=person)(cn=%s)(sn=%s))' % (user_cn, user_vo)
return dict(self._search(query, attrlist))
def search_by_token(self, token, attrlist=[]):
"""
:return: (dict) of the form dict(dn={...})
"""
query = '(&(objectclass=person)(userpassword=%s))' % token
return dict(self._search(query, attrlist))
def delete_user(self, userUID):
"""Remove a user from the LDAP directory
:raises ldap.NO_SUCH_OBJECT: if this user is not in the LDAP directory
"""
dn = 'uid=%s,%s' % (userUID, self.base_dn)
self.con.delete_s(dn)
def list_users(self, attrlist=[]):
"""
:return: (dict) of the form dict(dn={...}, ...)
"""
return dict(self._search('(&(objectclass=person))', attrlist))
def create(
self, userUID, certCN, email, token, user_vo, userClientDN,
userCert=None):
add_record = [
('objectclass', [
'person', 'organizationalperson', 'inetorgperson', 'pkiuser']),
('uid', [userUID]),
('cn', [certCN]),
('sn', [user_vo]),
('userpassword', [token]),
('mail', [email]),
('givenname', userClientDN),
('ou', ['users'])
]
dn = 'uid=%s,%s' % (userUID, self.base_dn)
self.con.add_s(dn, add_record)
if userCert:
cert_der = ssl.PEM_cert_to_DER_cert(userCert)
mod_attrs = [(ldap.MOD_ADD, 'userCertificate;binary', cert_der)]
self.con.modify_s(dn, mod_attrs)
def update_token(self, userUID, newToken):
dn = 'uid=%s,%s' % (userUID, self.base_dn)
mod_attrs = [(ldap.MOD_REPLACE, 'userpassword', newToken)]
self.con.modify_s(dn, mod_attrs)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment