Coverage for ipalib/plugins/trust : 38%

# Authors: 

#     Alexander Bokovoy <abokovoy@redhat.com> 

# 

# Copyright (C) 2011  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/>. 

from ipalib.plugins.baseldap import * 

from ipalib import api, Str, StrEnum, Password, DefaultFrom, _, ngettext, Object 

from ipalib.parameters import Enum 

from ipalib import Command 

from ipalib import errors 

from ipapython import ipautil 

from ipalib import util 

try: 

    import pysss_murmur #pylint: disable=F0401 

    _murmur_installed = True 

except Exception, e: 

    _murmur_installed = False 

if api.env.in_server and api.env.context in ['lite', 'server']: 

    try: 

        import ipaserver.dcerpc #pylint: disable=F0401 

        _bindings_installed = True 

    except Exception, e: 

        _bindings_installed = False 

__doc__ = _(""" 

Manage trust relationship between realms 

""") 

trust_output_params = ( 

    Str('ipantflatname', 

        label=_('Domain NetBIOS name')), 

    Str('ipanttrusteddomainsid', 

        label=_('Domain Security Identifier')), 

    Str('trustdirection', 

        label=_('Trust direction')), 

    Str('trusttype', 

        label=_('Trust type')), 

    Str('truststatus', 

        label=_('Trust status')), 

) 

_trust_type_dict = {1 : _('Non-Active Directory domain'), 

                    2 : _('Active Directory domain'), 

                    3 : _('RFC4120-compliant Kerberos realm')} 

_trust_direction_dict = {1 : _('Trusting forest'), 

                         2 : _('Trusted forest'), 

                         3 : _('Two-way trust')} 

_trust_status = {1 : _('Established and verified'), 

                 2 : _('Waiting for confirmation by remote side')} 

_trust_type_dict_unknown = _('Unknown') 

def trust_type_string(level): 

    """ 

    Returns a string representing a type of the trust. The original field is an enum: 

      LSA_TRUST_TYPE_DOWNLEVEL  = 0x00000001, 

      LSA_TRUST_TYPE_UPLEVEL    = 0x00000002, 

      LSA_TRUST_TYPE_MIT        = 0x00000003 

    """ 

    string = _trust_type_dict.get(int(level), _trust_type_dict_unknown) 

    return unicode(string) 

def trust_direction_string(level): 

    """ 

    Returns a string representing a direction of the trust. The original field is a bitmask taking two bits in use 

      LSA_TRUST_DIRECTION_INBOUND  = 0x00000001, 

      LSA_TRUST_DIRECTION_OUTBOUND = 0x00000002 

    """ 

    string = _trust_direction_dict.get(int(level), _trust_type_dict_unknown) 

    return unicode(string) 

def trust_status_string(level): 

    string = _trust_direction_dict.get(int(level), _trust_type_dict_unknown) 

    return unicode(string) 

class trust(LDAPObject): 

    """ 

    Trust object. 

    """ 

    trust_types = ('ad', 'ipa') 

    container_dn = api.env.container_trusts 

    object_name = _('trust') 

    object_name_plural = _('trusts') 

    object_class = ['ipaNTTrustedDomain'] 

    default_attributes = ['cn', 'ipantflatname', 'ipanttrusteddomainsid', 

        'ipanttrusttype', 'ipanttrustattributes', 'ipanttrustdirection', 'ipanttrustpartner', 

        'ipantauthtrustoutgoing', 'ipanttrustauthincoming', 'ipanttrustforesttrustinfo', 

        'ipanttrustposixoffset', 'ipantsupportedencryptiontypes' ] 

    label = _('Trusts') 

    label_singular = _('Trust') 

    takes_params = ( 

        Str('cn', 

            cli_name='realm', 

            label=_('Realm name'), 

            primary_key=True, 

        ), 

    ) 

def make_trust_dn(env, trust_type, dn): 

    assert isinstance(dn, DN) 

    if trust_type in trust.trust_types: 

        container_dn = DN(('cn', trust_type), env.container_trusts, env.basedn) 

        return DN(dn[0], container_dn) 

    return dn 

class trust_add(LDAPCreate): 

    __doc__ = _('Add new trust to use') 

    takes_options = LDAPCreate.takes_options + ( 

        StrEnum('trust_type', 

            cli_name='type', 

            label=_('Trust type (ad for Active Directory, default)'), 

            values=(u'ad',), 

            default=u'ad', 

            autofill=True, 

        ), 

        Str('realm_admin?', 

            cli_name='admin', 

            label=_("Active Directory domain administrator"), 

        ), 

        Password('realm_passwd?', 

            cli_name='password', 

            label=_("Active directory domain adminstrator's password"), 

            confirm=False, 

        ), 

        Str('realm_server?', 

            cli_name='server', 

            label=_('Domain controller for the Active Directory domain (optional)'), 

        ), 

        Password('trust_secret?', 

            cli_name='trust_secret', 

            label=_('Shared secret for the trust'), 

            confirm=False, 

        ), 

        Int('base_id?', 

            cli_name='base_id', 

            label=_('First Posix ID of the range reserved for the trusted domain'), 

        ), 

        Int('range_size?', 

            cli_name='range_size', 

            label=_('Size of the ID range reserved for the trusted domain'), 

            default=200000, 

            autofill=True 

        ), 

    ) 

    msg_summary = _('Added Active Directory trust for realm "%(value)s"') 

    def execute(self, *keys, **options): 

        if 'trust_type' in options: 

            if options['trust_type'] == u'ad': 

                result = self.execute_ad(*keys, **options) 

            else: 

                raise errors.ValidationError(name=_('trust type'), error=_('only "ad" is supported')) 

        else: 

            raise errors.RequirementError(name=_('trust type')) 

        self.add_range(*keys, **options) 

        return result 

    def add_range(self, *keys, **options): 

        new_obj = api.Command['trust_show'](keys[-1]) 

        dom_sid = new_obj['result']['ipanttrusteddomainsid'][0]; 

        range_name = keys[-1].upper()+'_id_range' 

        try: 

            old_range = api.Command['range_show'](range_name) 

        except errors.NotFound, e: 

            old_range = None 

        if old_range: 

            old_dom_sid = old_range['result']['ipanttrusteddomainsid'][0]; 

            if old_dom_sid == dom_sid: 

                return 

            raise errors.ValidationError(name=_('range exists'), 

                    error=_('ID range with the same name but different ' \ 

                            'domain SID already exists. The ID range for ' \ 

                            'the new trusted domain must be created manually.')) 

        if 'base_id' in options: 

            base_id = options['base_id'] 

        else: 

            if not _murmur_installed: 

                raise errors.ValidationError(name=_('missing base_id'), 

                    error=_('pysss_murmur is not available on the server ' \ 

                            'and no base_id is given, ' \ 

                            'ID range must be create manually')) 

            base_id = 200000 + (pysss_murmur.murmurhash3(dom_sid, len(dom_sid), 0xdeadbeef) % 10000) * 200000 

        try: 

            new_range = api.Command['range_add'](range_name, 

                                                 ipabaseid=base_id, 

                                                 ipaidrangesize=options['range_size'], 

                                                 ipabaserid=0, 

                                                 ipanttrusteddomainsid=dom_sid) 

        except Exception, e: 

            raise errors.ValidationError(name=_('ID range exists'), 

                   error = _('ID range already exists, must be added manually')) 

    def execute_ad(self, *keys, **options): 

        # Join domain using full credentials and with random trustdom 

        # secret (will be generated by the join method) 

        trustinstance = None 

        if not _bindings_installed: 

            raise errors.NotFound(name=_('AD Trust setup'), 

                  reason=_('''Cannot perform join operation without Samba 4 support installed. 

                              Make sure you have installed server-trust-ad sub-package of IPA''')) 

        if 'realm_server' not in options: 

            realm_server = None 

        else: 

            realm_server = options['realm_server'] 

        trustinstance = ipaserver.dcerpc.TrustDomainJoins(self.api) 

        if not trustinstance.configured: 

            raise errors.NotFound(name=_('AD Trust setup'), 

                  reason=_('''Cannot perform join operation without own domain configured. 

                              Make sure you have run ipa-adtrust-install on the IPA server first''')) 

        # 1. Full access to the remote domain. Use admin credentials and 

        # generate random trustdom password to do work on both sides 

        if 'realm_admin' in options: 

            realm_admin = options['realm_admin'] 

            names = realm_admin.split('@') 

            if len(names) > 1: 

                # realm admin name is in UPN format, user@realm, check that 

                # realm is the same as the one that we are attempting to trust 

                if keys[-1].lower() != names[-1].lower(): 

                    raise errors.ValidationError(name=_('AD Trust setup'), 

                                 error=_('Trusted domain and administrator account use different realms')) 

                realm_admin = names[0] 

            if 'realm_passwd' not in options: 

                raise errors.ValidationError(name=_('AD Trust setup'), error=_('Realm administrator password should be specified')) 

            realm_passwd = options['realm_passwd'] 

            result = trustinstance.join_ad_full_credentials(keys[-1], realm_server, realm_admin, realm_passwd) 

            if result is None: 

                raise errors.ValidationError(name=_('AD Trust setup'), error=_('Unable to verify write permissions to the AD')) 

            return dict(result=dict(), value=trustinstance.remote_domain.info['dns_domain']) 

        # 2. We don't have access to the remote domain and trustdom password 

        # is provided. Do the work on our side and inform what to do on remote 

        # side. 

        if 'trust_secret' in options: 

            result = trustinstance.join_ad_ipa_half(keys[-1], realm_server, options['trust_secret']) 

            return dict(result=dict(), value=trustinstance.remote_domain.info['dns_domain']) 

        raise errors.ValidationError(name=_('AD Trust setup'), error=_('Not enough arguments specified to perform trust setup')) 

class trust_del(LDAPDelete): 

    __doc__ = _('Delete a trust.') 

    msg_summary = _('Deleted trust "%(value)s"') 

    def pre_callback(self, ldap, dn, *keys, **options): 

        assert isinstance(dn, DN) 

        try: 

            result = self.api.Command.trust_show(keys[-1]) 

        except errors.NotFound, e: 

            self.obj.handle_not_found(*keys) 

        return result['result']['dn'] 

class trust_mod(LDAPUpdate): 

    __doc__ = _('Modify a trust.') 

    msg_summary = _('Modified trust "%(value)s"') 

    def pre_callback(self, ldap, dn, *keys, **options): 

        assert isinstance(dn, DN) 

        result = None 

        try: 

            result = self.api.Command.trust_show(keys[-1]) 

        except errors.NotFound, e: 

            self.obj.handle_not_found(*keys) 

        # TODO: we found the trust object, now modify it 

        return result['result']['dn'] 

class trust_find(LDAPSearch): 

    __doc__ = _('Search for trusts.') 

    msg_summary = ngettext( 

        '%(count)d trust matched', '%(count)d trusts matched', 0 

    ) 

    # Since all trusts types are stored within separate containers under 'cn=trusts', 

    # search needs to be done on a sub-tree scope 

    def pre_callback(self, ldap, filters, attrs_list, base_dn, scope, *args, **options): 

        assert isinstance(base_dn, DN) 

        return (filters, base_dn, ldap.SCOPE_SUBTREE) 

class trust_show(LDAPRetrieve): 

    __doc__ = _('Display information about a trust.') 

    has_output_params = LDAPRetrieve.has_output_params + trust_output_params 

    def execute(self, *keys, **options): 

        error = None 

        result = None 

        for trust_type in trust.trust_types: 

            options['trust_show_type'] = trust_type 

            try: 

                result = super(trust_show, self).execute(*keys, **options) 

            except errors.NotFound, e: 

                result = None 

                error = e 

            if result: 

                 result['result']['trusttype'] = [trust_type_string(result['result']['ipanttrusttype'][0])] 

                 result['result']['trustdirection'] = [trust_direction_string(result['result']['ipanttrustdirection'][0])] 

                 break 

        if error or not result: 

            self.obj.handle_not_found(*keys) 

        return result 

    def pre_callback(self, ldap, dn, entry_attrs, *keys, **options): 

        assert isinstance(dn, DN) 

        if 'trust_show_type' in options: 

            return make_trust_dn(self.env, options['trust_show_type'], dn) 

        return dn 

api.register(trust) 

api.register(trust_add) 

api.register(trust_mod) 

api.register(trust_del) 

api.register(trust_find) 

api.register(trust_show)