IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2025-02-25 18:55:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

Code cleanup

Browse Source 
This patch removes two files which seem to be long obsoleted and not
used any more.
This commit is contained in:
Jan Zeleny 2011-02-14 07:17:10 -05:00 committed by Rob Crittenden
parent f558ffe294
commit 1e9f923c49
2 changed files with 0 additions and 893 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

445
ipaserver/plugins/ldapapi.py
View File

@ -1,445 +0,0 @@
# Authors:
# Rob Crittenden <rcritten@redhat.com>
# Jason Gerard DeRose <jderose@redhat.com>
#
# Copyright (C) 2008 Red Hat
# see file 'COPYING' for use and warranty information
#
# 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/>.
"""
Backend plugin for LDAP.
This wraps the python-ldap bindings.
"""
import ldap as _ldap
from ipalib import api
from ipalib import errors
from ipalib.crud import CrudBackend
from ipaserver import servercore, ipaldap
import krbV
class ldap(CrudBackend):
"""
LDAP backend plugin.
"""
def __init__(self):
self.dn = _ldap.dn
super(ldap, self).__init__()
def create_connection(self, ccache):
if ccache is None:
raise errors.CCacheError()
conn = ipaldap.IPAdmin(self.env.ldap_host, self.env.ldap_port)
principal = krbV.CCache(
name=ccache, context=krbV.default_context()
).principal().name
conn.set_krbccache(ccache, principal)
return conn
def destroy_connection(self):
self.conn.unbind_s()
def make_user_dn(self, uid):
"""
Construct user dn from uid.
"""
return 'uid=%s,%s,%s' % (
self.dn.escape_dn_chars(uid),
self.api.env.container_user,
self.api.env.basedn,
)
def make_group_dn(self, cn):
"""
Construct group dn from cn.
"""
return 'cn=%s,%s,%s' % (
self.dn.escape_dn_chars(cn),
self.api.env.container_group,
self.api.env.basedn,
)
def make_hostgroup_dn(self, cn):
"""
Construct group of hosts dn from cn.
"""
return 'cn=%s,%s,%s' % (
self.dn.escape_dn_chars(cn),
self.api.env.container_hostgroup,
self.api.env.basedn,
)
def make_taskgroup_dn(self, cn):
"""
Construct group of tasks dn from cn.
"""
return 'cn=%s,%s,%s' % (
self.dn.escape_dn_chars(cn),
self.api.env.container_taskgroup,
self.api.env.basedn,
)
def make_service_dn(self, principal):
"""
Construct service principal dn from principal name
"""
return 'krbprincipalname=%s,%s,%s' % (
self.dn.escape_dn_chars(principal),
self.api.env.container_service,
self.api.env.basedn,
)
def make_host_dn(self, hostname):
"""
Construct host dn from hostname
"""
return 'fqdn=%s,%s,%s' % (
self.dn.escape_dn_chars(hostname),
self.api.env.container_host,
self.api.env.basedn,
)
def make_application_dn(self, appname):
"""
Construct application dn from cn.
"""
return 'cn=%s,%s,%s' % (
self.dn.escape_dn_chars(appname),
self.api.env.container_applications,
self.api.env.basedn,
)
def make_policytemplate_dn(self, appname, uuid):
"""
Construct policytemplate dn from appname
"""
return 'ipaUniqueID=%s,%s' % (
self.dn.escape_dn_chars(uuid),
self.make_application_dn(appname),
)
def make_role_application_dn(self, appname):
"""
Construct application role container from appname
"""
return 'cn=%s,%s,%s' % (
self.dn.escape_dn_chars(appname),
self.api.env.container_roles,
self.api.env.basedn,
)
def make_role_policytemplate_dn(self, appname, uuid):
"""
Construct policytemplate dn from uuid and appname
"""
return 'ipaUniqueID=%s,cn=templates,%s' % (
self.dn.escape_dn_chars(uuid),
self.make_role_application_dn(appname),
)
def make_policygroup_dn(self, uuid):
"""
Construct policygroup dn from uuid
"""
return 'ipaUniqueID=%s,%s,%s' % (
self.dn.escape_dn_chars(uuid),
self.api.env.container_policygroups,
self.api.env.basedn,
)
def make_policy_dn(self, uuid, polgroup_uuid):
"""
Construct policy dn from uuid of policy and its policygroup
"""
return 'ipaUniqueID=%s,%s' % (
self.dn.escape_dn_chars(uuid),
self.make_policygroup_dn(polgroup_uuid),
)
def make_policy_blob_dn(self, blob_uuid, policy_uuid, polgroup_uuid):
"""
Construct policy blob dn from uuids of its policy and policygroup
objects
"""
return 'ipaUniqueID=%s,%s' % (
self.dn.escape_dn_chars(blob_uuid),
self.make_policy_dn(policy_uuid, polgroup_uuid),
)
def make_policylink_dn(self, uuid):
"""
Construct policy link dn from uuid
"""
return 'ipaUniqueID=%s,%s,%s' % (
self.dn.escape_dn_chars(uuid),
self.api.env.container_policylinks,
self.api.env.basedn,
)
def get_object_type(self, attribute):
"""
Based on attribute, make an educated guess as to the type of
object we're looking for.
"""
attribute = attribute.lower()
object_type = None
if attribute == "uid": # User
object_type = "posixAccount"
elif attribute == "cn": # Group
object_type = "ipaUserGroup"
elif attribute == "krbprincipalname": # Service
object_type = "krbPrincipal"
return object_type
def find_entry_dn(self, key_attribute, primary_key, object_type=None, base=None):
"""
Find an existing entry's dn from an attribute
"""
key_attribute = key_attribute.lower()
if not object_type:
object_type = self.get_object_type(key_attribute)
if not object_type:
return None
search_filter = "(&(objectclass=%s)(%s=%s))" % (
object_type,
key_attribute,
self.dn.escape_dn_chars(primary_key)
)
if not base:
base = self.api.env.container_accounts
search_base = "%s, %s" % (base, self.api.env.basedn)
entry = servercore.get_sub_entry(search_base, search_filter, ['dn', 'objectclass'])
return entry.get('dn')
def get_base_entry(self, searchbase, searchfilter, attrs):
return servercore.get_base_entry(searchbase, searchfilter, attrs)
def get_sub_entry(self, searchbase, searchfilter, attrs):
return servercore.get_sub_entry(searchbase, searchfilter, attrs)
def get_one_entry(self, searchbase, searchfilter, attrs):
return servercore.get_one_entry(searchbase, searchfilter, attrs)
def get_ipa_config(self):
"""Return a dictionary of the IPA configuration"""
return servercore.get_ipa_config()
def mark_entry_active(self, dn):
return servercore.mark_entry_active(dn)
def mark_entry_inactive(self, dn):
return servercore.mark_entry_inactive(dn)
def _generate_search_filters(self, **kw):
"""Generates a search filter based on a list of words and a list
of fields to search against.
Returns a tuple of two filters: (exact_match, partial_match)
"""
# construct search pattern for a single word
# (|(f1=word)(f2=word)...)
exact_pattern = "(|"
for field in kw.keys():
exact_pattern += "(%s=%s)" % (field, kw[field])
exact_pattern += ")"
sub_pattern = "(|"
for field in kw.keys():
sub_pattern += "(%s=*%s*)" % (field, kw[field])
sub_pattern += ")"
# construct the giant match for all words
exact_match_filter = "(&" + exact_pattern + ")"
partial_match_filter = "(|" + sub_pattern + ")"
return (exact_match_filter, partial_match_filter)
def _get_scope(self, scope_str):
scope_dict = {'one' : _ldap.SCOPE_ONELEVEL,
'subtree' : _ldap.SCOPE_SUBTREE,
'base' : _ldap.SCOPE_BASE }
return scope_dict.get(scope_str, _ldap.SCOPE_SUBTREE)
def modify_password(self, dn, **kw):
return servercore.modify_password(dn, kw.get('oldpass', ''), kw.get('newpass'))
def add_member_to_group(self, memberdn, groupdn, memberattr='member'):
"""
Add a new member to a group.
:param memberdn: the DN of the member to add
:param groupdn: the DN of the group to add a member to
"""
return servercore.add_member_to_group(memberdn, groupdn, memberattr)
def remove_member_from_group(self, memberdn, groupdn, memberattr='member'):
"""
Remove a new member from a group.
:param memberdn: the DN of the member to remove
:param groupdn: the DN of the group to remove a member from
"""
return servercore.remove_member_from_group(memberdn, groupdn, memberattr)
# The CRUD operations
def strip_none(self, kw):
"""
Remove any None values present in the LDAP attribute dict.
"""
for (key, value) in kw.iteritems():
if value is None:
continue
if type(value) in (list, tuple):
value = filter(
lambda v: type(v) in (str, unicode, bool, int, float),
value
)
if len(value) > 0:
yield (key, value)
else:
assert type(value) in (str, unicode, bool, int, float)
yield (key, value)
def create(self, **kw):
if servercore.entry_exists(kw['dn']):
raise errors.DuplicateEntry
kw = dict(self.strip_none(kw))
entry = ipaldap.Entry(kw['dn'])
# dn isn't allowed to be in the entry itself
del kw['dn']
# Fill in our new entry
for k in kw:
entry.setValues(k, kw[k])
servercore.add_entry(entry)
return self.retrieve(entry.dn)
def retrieve(self, dn, attributes=None):
return servercore.get_entry_by_dn(dn, attributes)
def update(self, dn, **kw):
result = self.retrieve(dn, ["*"] + kw.keys())
entry = ipaldap.Entry((dn, servercore.convert_scalar_values(result)))
start_keys = kw.keys()
kw = dict(self.strip_none(kw))
end_keys = kw.keys()
removed_keys = list(set(start_keys) - set(end_keys))
for k in kw:
entry.setValues(k, kw[k])
for k in removed_keys:
entry.delAttr(k)
servercore.update_entry(entry.toDict(), removed_keys)
return self.retrieve(dn)
def delete(self, dn):
return servercore.delete_entry(dn)
def search(self, **kw):
objectclass = kw.get('objectclass')
sfilter = kw.get('filter')
attributes = kw.get('attributes')
base = kw.get('base')
scope = kw.get('scope')
exactonly = kw.get('exactonly', None)
if attributes:
del kw['attributes']
else:
attributes = ['*']
if objectclass:
del kw['objectclass']
if base:
del kw['base']
if sfilter:
del kw['filter']
if scope:
del kw['scope']
if exactonly is not None:
del kw['exactonly']
if kw:
(exact_match_filter, partial_match_filter) = self._generate_search_filters(**kw)
else:
(exact_match_filter, partial_match_filter) = ('', '')
if objectclass:
exact_match_filter = '(&(objectClass=%s)%s)' % (objectclass, exact_match_filter)
partial_match_filter = '(&(objectClass=%s)%s)' % (objectclass, partial_match_filter)
else:
exact_match_filter = '(&(objectClass=*)%s)' % exact_match_filter
partial_match_filter = '(&(objectClass=*)%s)' % partial_match_filter
if sfilter:
exact_match_filter = '(%s%s)' % (sfilter, exact_match_filter)
partial_match_filter = '(%s%s)' % (sfilter, partial_match_filter)
search_scope = self._get_scope(scope)
if not base:
base = self.api.env.container_accounts
search_base = '%s,%s' % (base, self.api.env.basedn)
try:
exact_results = servercore.search(search_base,
exact_match_filter, attributes, scope=search_scope)
except errors.NotFound:
exact_results = [0]
if not exactonly:
try:
partial_results = servercore.search(search_base,
partial_match_filter, attributes, scope=search_scope)
except errors.NotFound:
partial_results = [0]
else:
partial_results = [0]
exact_counter = exact_results[0]
partial_counter = partial_results[0]
exact_results = exact_results[1:]
partial_results = partial_results[1:]
# Remove exact matches from the partial_match list
exact_dns = set(map(lambda e: e.get('dn'), exact_results))
partial_results = filter(lambda e: e.get('dn') not in exact_dns,
partial_results)
if (exact_counter == -1) or (partial_counter == -1):
counter = -1
else:
counter = len(exact_results) + len(partial_results)
results = [counter]
for r in exact_results + partial_results:
results.append(r)
return results
api.register(ldap)

448
ipaserver/servercore.py
View File

@ -1,448 +0,0 @@
# Authors: Rob Crittenden <rcritten@redhat.com>
#
# Copyright (C) 2007 Red Hat
# see file 'COPYING' for use and warranty information
#
# 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
from ipalib.request import context
from ipaserver import ipaldap
from ipalib import errors
from ipalib import api
def convert_entry(ent):
entry = dict(ent.data)
entry['dn'] = ent.dn
# For now convert single entry lists to a string for the ui.
# TODO: we need to deal with multi-values better
for key,value in entry.iteritems():
if isinstance(value,list) or isinstance(value,tuple):
if len(value) == 0:
entry[key] = ''
elif len(value) == 1:
entry[key] = value[0]
return entry
def convert_scalar_values(orig_dict):
"""LDAP update dicts expect all values to be a list (except for dn).
This method converts single entries to a list."""
new_dict={}
for (k,v) in orig_dict.iteritems():
if not isinstance(v, list) and k != 'dn':
v = [v]
new_dict[k] = v
return new_dict
def generate_match_filters(search_fields, criteria_words):
"""Generates a search filter based on a list of words and a list
of fields to search against.
Returns a tuple of two filters: (exact_match, partial_match)"""
# construct search pattern for a single word
# (|(f1=word)(f2=word)...)
search_pattern = "(|"
for field in search_fields:
search_pattern += "(" + field + "=%(match)s)"
search_pattern += ")"
gen_search_pattern = lambda word: search_pattern % {'match':word}
# construct the giant match for all words
exact_match_filter = "(&"
partial_match_filter = "(|"
for word in criteria_words:
exact_match_filter += gen_search_pattern(word)
partial_match_filter += gen_search_pattern("*%s*" % word)
exact_match_filter += ")"
partial_match_filter += ")"
return (exact_match_filter, partial_match_filter)
# TODO: rethink the get_entry vs get_list API calls.
# they currently restrict the data coming back without
# restricting scope. For now adding a get_base/sub_entry()
# calls, but the API isn't great.
def get_entry (base, scope, searchfilter, sattrs=None):
"""Get a specific entry (with a parametized scope).
Return as a dict of values.
Multi-valued fields are represented as lists.
"""
ent=""
ent = context.ldap.conn.getEntry(base, scope, searchfilter, sattrs)
return convert_entry(ent)
def get_base_entry (base, searchfilter, sattrs=None):
"""Get a specific entry (with a scope of BASE).
Return as a dict of values.
Multi-valued fields are represented as lists.
"""
return get_entry(base, ldap.SCOPE_BASE, searchfilter, sattrs)
def get_sub_entry (base, searchfilter, sattrs=None):
"""Get a specific entry (with a scope of SUB).
Return as a dict of values.
Multi-valued fields are represented as lists.
"""
return get_entry(base, ldap.SCOPE_SUBTREE, searchfilter, sattrs)
def get_one_entry (base, searchfilter, sattrs=None):
"""Get the children of an entry (with a scope of ONE).
Return as a list of dict of values.
Multi-valued fields are represented as lists.
"""
return get_list(base, searchfilter, sattrs, ldap.SCOPE_ONELEVEL)
def get_list (base, searchfilter, sattrs=None, scope=ldap.SCOPE_SUBTREE):
"""Gets a list of entries. Each is converted to a dict of values.
Multi-valued fields are represented as lists.
"""
entries = []
entries = context.ldap.conn.getList(base, scope, searchfilter, sattrs)
return map(convert_entry, entries)
def has_nsaccountlock(dn):
"""Check to see if an entry has the nsaccountlock attribute.
This attribute is provided by the Class of Service plugin so
doing a search isn't enough. It is provided by the two
entries cn=inactivated and cn=activated. So if the entry has
the attribute and isn't in either cn=activated or cn=inactivated
then the attribute must be in the entry itself.
Returns True or False
"""
# First get the entry. If it doesn't have nsaccountlock at all we
# can exit early.
entry = get_entry_by_dn(dn, ['dn', 'nsaccountlock', 'memberof'])
if not entry.get('nsaccountlock'):
return False
# Now look to see if they are in activated or inactivated
# entry is a member
memberof = entry.get('memberof')
if isinstance(memberof, basestring):
memberof = [memberof]
for m in memberof:
inactivated = m.find("cn=inactivated")
activated = m.find("cn=activated")
# if they are in either group that means that the nsaccountlock
# value comes from there, otherwise it must be in this entry.
if inactivated >= 0 or activated >= 0:
return False
return True
# General searches
def get_entry_by_dn (dn, sattrs=None):
"""Get a specific entry. Return as a dict of values.
Multi-valued fields are represented as lists.
"""
searchfilter = "(objectClass=*)"
api.log.info("IPA: get_entry_by_dn '%s'" % dn)
return get_base_entry(dn, searchfilter, sattrs)
def get_entry_by_cn (cn, sattrs):
"""Get a specific entry by cn. Return as a dict of values.
Multi-valued fields are represented as lists.
"""
api.log.info("IPA: get_entry_by_cn '%s'" % cn)
# cn = self.__safe_filter(cn)
searchfilter = "(cn=%s)" % cn
return get_sub_entry("cn=accounts," + api.env.basedn, searchfilter, sattrs)
# User support
def entry_exists(dn):
"""Return True if the entry exists, False otherwise."""
try:
get_base_entry(dn, "objectclass=*", ['dn','objectclass'])
return True
except errors.NotFound:
return False
def get_user_by_uid (uid, sattrs):
"""Get a specific user's entry. Return as a dict of values.
Multi-valued fields are represented as lists.
"""
if not isinstance(uid,basestring) or len(uid) == 0:
raise SyntaxError("uid is not a string")
# raise ipaerror.gen_exception(ipaerror.INPUT_INVALID_PARAMETER)
if sattrs is not None and not isinstance(sattrs,list):
raise SyntaxError("sattrs is not a list")
# raise ipaerror.gen_exception(ipaerror.INPUT_INVALID_PARAMETER)
api.log.info("IPA: get_user_by_uid '%s'" % uid)
# uid = self.__safe_filter(uid)
searchfilter = "(uid=" + uid + ")"
return get_sub_entry("cn=accounts," + api.env.basedn, searchfilter, sattrs)
def uid_too_long(uid):
"""Verify that the new uid is within the limits we set. This is a
very narrow test.
Returns True if it is longer than allowed
False otherwise
"""
if not isinstance(uid,basestring) or len(uid) == 0:
# It is bad, but not too long
return False
api.log.debug("IPA: __uid_too_long(%s)" % uid)
try:
config = get_ipa_config()
maxlen = int(config.get('ipamaxusernamelength', 0))
if maxlen > 0 and len(uid) > maxlen:
return True
except Exception, e:
api.log.debug("There was a problem " + str(e))
pass
return False
def update_entry (entry, remove_keys=[]):
"""Update an LDAP entry
entry is a dict
remove_keys is a list of attributes to remove from this entry
This refreshes the record from LDAP in order to obtain the list of
attributes that has changed. It only retrieves the attributes that
are in the update so attributes aren't inadvertantly lost.
"""
assert type(remove_keys) is list
attrs = entry.keys()
o = get_base_entry(entry['dn'], "objectclass=*", attrs + remove_keys)
oldentry = convert_scalar_values(o)
newentry = convert_scalar_values(entry)
# Should be able to get this from either the old or new entry
# but just in case someone has decided to try changing it, use the
# original
try:
moddn = oldentry['dn']
except KeyError, e:
# FIXME: return a missing DN error message
raise e
return context.ldap.conn.updateEntry(moddn, oldentry, newentry)
def add_entry(entry):
"""Add a new entry"""
return context.ldap.conn.addEntry(entry)
def delete_entry(dn):
"""Remove an entry"""
return context.ldap.conn.deleteEntry(dn)
# FIXME, get time and search limit from cn=ipaconfig
def search(base, filter, attributes, timelimit=1, sizelimit=3000, scope=ldap.SCOPE_SUBTREE):
"""Perform an LDAP query"""
try:
timelimit = float(timelimit)
results = context.ldap.conn.getListAsync(base, scope,
filter, attributes, 0, None, None, timelimit, sizelimit)
except ldap.NO_SUCH_OBJECT:
raise errors.NotFound(reason=filter)
counter = results[0]
entries = [counter]
for r in results[1:]:
entries.append(convert_entry(r))
return entries
def uniq_list(x):
"""Return a unique list, preserving order and ignoring case"""
myset = {}
return [myset.setdefault(e.lower(),e) for e in x if e.lower() not in myset]
def get_schema():
"""Retrieves the current LDAP schema from the LDAP server."""
schema_entry = get_base_entry("", "objectclass=*", ['dn','subschemasubentry'])
schema_cn = schema_entry.get('subschemasubentry')
schema = get_base_entry(schema_cn, "objectclass=*", ['*'])
return schema
def get_objectclasses():
"""Returns a list of available objectclasses that the LDAP
server supports. This parses out the syntax, attributes, etc
and JUST returns a lower-case list of the names."""
schema = get_schema()
objectclasses = schema.get('objectclasses')
# Convert this list into something more readable
result = []
for i in range(len(objectclasses)):
oc = objectclasses[i].lower().split(" ")
result.append(oc[3].replace("'",""))
return result
def get_ipa_config():
"""Retrieve the IPA configuration"""
searchfilter = "cn=ipaconfig"
try:
config = get_sub_entry("cn=etc," + api.env.basedn, searchfilter)
except ldap.NO_SUCH_OBJECT:
# FIXME
raise errors.NotFound(reason="IPA configuration cannot be loaded")
return config
def modify_password(dn, oldpass, newpass):
return context.ldap.conn.modifyPassword(dn, oldpass, newpass)
def mark_entry_active (dn):
"""Mark an entry as active in LDAP."""
# This can be tricky. The entry itself can be marked inactive
# by being in the inactivated group. It can also be inactivated by
# being the member of an inactive group.
#
# First we try to remove the entry from the inactivated group. Then
# if it is still inactive we have to add it to the activated group
# which will override the group membership.
res = ""
# First, check the entry status
entry = get_entry_by_dn(dn, ['dn', 'nsAccountlock'])
if entry.get('nsaccountlock', 'false').lower() == "false":
api.log.debug("IPA: already active")
raise errors.AlreadyActive()
if has_nsaccountlock(dn):
api.log.debug("IPA: appears to have the nsaccountlock attribute")
raise errors.HasNSAccountLock()
group = get_entry_by_cn("inactivated", None)
try:
remove_member_from_group(entry.get('dn'), group.get('dn'))
except errors.NotGroupMember:
# Perhaps the user is there as a result of group membership
pass
# Now they aren't a member of inactivated directly, what is the status
# now?
entry = get_entry_by_dn(dn, ['dn', 'nsAccountlock'])
if entry.get('nsaccountlock', 'false').lower() == "false":
# great, we're done
api.log.debug("IPA: removing from inactivated did it.")
return True
# So still inactive, add them to activated
group = get_entry_by_cn("activated", None)
res = add_member_to_group(dn, group.get('dn'))
api.log.debug("IPA: added to activated.")
return res
def mark_entry_inactive (dn):
"""Mark an entry as inactive in LDAP."""
entry = get_entry_by_dn(dn, ['dn', 'nsAccountlock', 'memberOf'])
if entry.get('nsaccountlock', 'false').lower() == "true":
api.log.debug("IPA: already marked as inactive")
raise errors.AlreadyInactive()
if has_nsaccountlock(dn):
api.log.debug("IPA: appears to have the nsaccountlock attribute")
raise errors.HasNSAccountLock()
# First see if they are in the activated group as this will override
# the our inactivation.
group = get_entry_by_cn("activated", None)
try:
remove_member_from_group(dn, group.get('dn'))
except errors.NotGroupMember:
# this is fine, they may not be explicitly in this group
pass
# Now add them to inactivated
group = get_entry_by_cn("inactivated", None)
res = add_member_to_group(dn, group.get('dn'))
return res
def add_member_to_group(member_dn, group_dn, memberattr='member'):
"""
Add a member to an existing group.
"""
api.log.info("IPA: add_member_to_group '%s' to '%s'" % (member_dn, group_dn))
if member_dn.lower() == group_dn.lower():
# You can't add a group to itself
raise errors.RecursiveGroup()
group = get_entry_by_dn(group_dn, None)
if group is None:
raise errors.NotFound(reason="cannot find group %s" % group_dn)
# check to make sure member_dn exists
member_entry = get_base_entry(member_dn, "(objectClass=*)", ['dn','objectclass'])
if not member_entry:
raise errors.NotFound(reason="cannot find member %s" % member_dn)
# Add the new member to the group member attribute
members = group.get(memberattr, [])
if isinstance(members, basestring):
members = [members]
members.append(member_dn)
group[memberattr] = members
return update_entry(group)
def remove_member_from_group(member_dn, group_dn, memberattr='member'):
"""Remove a member_dn from an existing group."""
group = get_entry_by_dn(group_dn, None)
if group is None:
raise errors.NotFound(reason="cannot find group %s" % group_dn)
"""
if group.get('cn') == "admins":
member = get_entry_by_dn(member_dn, ['dn','uid'])
if member.get('uid') == "admin":
raise ipaerror.gen_exception(ipaerror.INPUT_ADMIN_REQUIRED_IN_ADMINS)
"""
api.log.info("IPA: remove_member_from_group '%s' from '%s'" % (member_dn, group_dn))
members = group.get(memberattr, False)
if not members:
raise errors.NotGroupMember()
if isinstance(members,basestring):
members = [members]
for i in range(len(members)):
members[i] = ipaldap.IPAdmin.normalizeDN(members[i])
try:
members.remove(member_dn)
except ValueError:
raise errors.NotGroupMember()
except Exception, e:
raise e
group[memberattr] = members
return update_entry(group)