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
|
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import ldap
import logging
from ldap.filter import filter_format
from odoo import api, fields, models, tools
from odoo.tools.pycompat import to_native
_logger = logging.getLogger(__name__)
class CompanyLDAP(models.Model):
_name = 'res.company.ldap'
_order = 'sequence'
_rec_name = 'ldap_server'
sequence = fields.Integer(default=10)
company = fields.Many2one('res.company', string='Company', required=True, ondelete='cascade')
ldap_server = fields.Char(string='LDAP Server address', required=True, default='127.0.0.1')
ldap_server_port = fields.Integer(string='LDAP Server port', required=True, default=389)
ldap_binddn = fields.Char('LDAP binddn',
help="The user account on the LDAP server that is used to query the directory. "
"Leave empty to connect anonymously.")
ldap_password = fields.Char(string='LDAP password',
help="The password of the user account on the LDAP server that is used to query the directory.")
ldap_filter = fields.Char(string='LDAP filter', required=True)
ldap_base = fields.Char(string='LDAP base', required=True)
user = fields.Many2one('res.users', string='Template User',
help="User to copy when creating new users")
create_user = fields.Boolean(default=True,
help="Automatically create local user accounts for new users authenticating via LDAP")
ldap_tls = fields.Boolean(string='Use TLS',
help="Request secure TLS/SSL encryption when connecting to the LDAP server. "
"This option requires a server with STARTTLS enabled, "
"otherwise all authentication attempts will fail.")
@api.multi
def get_ldap_dicts(self):
"""
Retrieve res_company_ldap resources from the database in dictionary
format.
:return: ldap configurations
:rtype: list of dictionaries
"""
ldaps = self.sudo().search([('ldap_server', '!=', False)], order='sequence')
res = ldaps.read([
'id',
'company',
'ldap_server',
'ldap_server_port',
'ldap_binddn',
'ldap_password',
'ldap_filter',
'ldap_base',
'user',
'create_user',
'ldap_tls'
])
return res
def connect(self, conf):
"""
Connect to an LDAP server specified by an ldap
configuration dictionary.
:param dict conf: LDAP configuration
:return: an LDAP object
"""
uri = 'ldap://%s:%d' % (conf['ldap_server'], conf['ldap_server_port'])
connection = ldap.initialize(uri)
if conf['ldap_tls']:
connection.start_tls_s()
return connection
def authenticate(self, conf, login, password):
"""
Authenticate a user against the specified LDAP server.
In order to prevent an unintended 'unauthenticated authentication',
which is an anonymous bind with a valid dn and a blank password,
check for empty passwords explicitely (:rfc:`4513#section-6.3.1`)
:param dict conf: LDAP configuration
:param login: username
:param password: Password for the LDAP user
:return: LDAP entry of authenticated user or False
:rtype: dictionary of attributes
"""
if not password:
return False
entry = False
try:
filter = filter_format(conf['ldap_filter'], (login,))
except TypeError:
_logger.warning('Could not format LDAP filter. Your filter should contain one \'%s\'.')
return False
try:
results = self.query(conf, tools.ustr(filter))
# Get rid of (None, attrs) for searchResultReference replies
results = [i for i in results if i[0]]
if len(results) == 1:
dn = results[0][0]
conn = self.connect(conf)
conn.simple_bind_s(dn, to_native(password))
conn.unbind()
entry = results[0]
except ldap.INVALID_CREDENTIALS:
return False
except ldap.LDAPError as e:
_logger.error('An LDAP exception occurred: %s', e)
return entry
def query(self, conf, filter, retrieve_attributes=None):
"""
Query an LDAP server with the filter argument and scope subtree.
Allow for all authentication methods of the simple authentication
method:
- authenticated bind (non-empty binddn + valid password)
- anonymous bind (empty binddn + empty password)
- unauthenticated authentication (non-empty binddn + empty password)
.. seealso::
:rfc:`4513#section-5.1` - LDAP: Simple Authentication Method.
:param dict conf: LDAP configuration
:param filter: valid LDAP filter
:param list retrieve_attributes: LDAP attributes to be retrieved. \
If not specified, return all attributes.
:return: ldap entries
:rtype: list of tuples (dn, attrs)
"""
results = []
try:
conn = self.connect(conf)
ldap_password = conf['ldap_password'] or ''
ldap_binddn = conf['ldap_binddn'] or ''
conn.simple_bind_s(to_native(ldap_binddn), to_native(ldap_password))
results = conn.search_st(to_native(conf['ldap_base']), ldap.SCOPE_SUBTREE, filter, retrieve_attributes, timeout=60)
conn.unbind()
except ldap.INVALID_CREDENTIALS:
_logger.error('LDAP bind failed.')
except ldap.LDAPError as e:
_logger.error('An LDAP exception occurred: %s', e)
return results
def map_ldap_attributes(self, conf, login, ldap_entry):
"""
Compose values for a new resource of model res_users,
based upon the retrieved ldap entry and the LDAP settings.
:param dict conf: LDAP configuration
:param login: the new user's login
:param tuple ldap_entry: single LDAP result (dn, attrs)
:return: parameters for a new resource of model res_users
:rtype: dict
"""
return {
'name': ldap_entry[1]['cn'][0],
'login': login,
'company_id': conf['company'][0]
}
@api.model
def get_or_create_user(self, conf, login, ldap_entry):
"""
Retrieve an active resource of model res_users with the specified
login. Create the user if it is not initially found.
:param dict conf: LDAP configuration
:param login: the user's login
:param tuple ldap_entry: single LDAP result (dn, attrs)
:return: res_users id
:rtype: int
"""
user_id = False
login = tools.ustr(login.lower().strip())
self.env.cr.execute("SELECT id, active FROM res_users WHERE lower(login)=%s", (login,))
res = self.env.cr.fetchone()
if res:
if res[1]:
user_id = res[0]
elif conf['create_user']:
_logger.debug("Creating new Odoo user \"%s\" from LDAP" % login)
values = self.map_ldap_attributes(conf, login, ldap_entry)
SudoUser = self.env['res.users'].sudo()
if conf['user']:
values['active'] = True
user_id = SudoUser.browse(conf['user'][0]).copy(default=values).id
else:
user_id = SudoUser.create(values).id
return user_id
|
