summaryrefslogtreecommitdiffstats
path: root/ipalib/plugins/trust.py
blob: f19a0a874057860d0ae32f1cc2336bdc3accf6e5 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
# 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)