[MERGE] auth_ldap: Migrate to new API

This commit is contained in:
Yannick Tivisse
2016-07-29 15:17:03 +02:00
10 changed files with 280 additions and 294 deletions
+8 -8
View File
@@ -1,7 +1,7 @@
Adds support for authentication by LDAP server.
===============================================
This module allows users to login with their LDAP username and password, and
will automatically create OpenERP users for them on the fly.
will automatically create Odoo users for them on the fly.
**Note:** This module only work on servers who have Python's ``ldap`` module installed.
@@ -10,7 +10,7 @@ Configuration:
After installing this module, you need to configure the LDAP parameters in the
Configuration tab of the Company details. Different companies may have different
LDAP servers, as long as they have unique usernames (usernames need to be unique
in OpenERP, even across multiple companies).
in Odoo, even across multiple companies).
Anonymous LDAP binding is also supported (for LDAP servers that allow it), by
simply keeping the LDAP user and password empty in the LDAP configuration.
@@ -26,24 +26,24 @@ manpage: manpage:`ldap.conf(5)`.
Security Considerations:
------------------------
Users' LDAP passwords are never stored in the OpenERP database, the LDAP server
Users' LDAP passwords are never stored in the Odoo database, the LDAP server
is queried whenever a user needs to be authenticated. No duplication of the
password occurs, and passwords are managed in one place only.
OpenERP does not manage password changes in the LDAP, so any change of password
Odoo does not manage password changes in the LDAP, so any change of password
should be conducted by other means in the LDAP directory directly (for LDAP users).
It is also possible to have local OpenERP users in the database along with
It is also possible to have local Odoo users in the database along with
LDAP-authenticated users (the Administrator account is one obvious example).
Here is how it works:
---------------------
* The system first attempts to authenticate users against the local OpenERP
* The system first attempts to authenticate users against the local Odoo
database;
* if this authentication fails (for example because the user has no local
password), the system then attempts to authenticate against LDAP;
As LDAP users have blank passwords by default in the local OpenERP database
As LDAP users have blank passwords by default in the local Odoo database
(which means no access), the first step always fails and the LDAP server is
queried to do the authentication.
@@ -61,6 +61,6 @@ allows pre-setting the default groups and menus of the first-time users.
assigned as local password for each new LDAP user, effectively setting
a *master password* for these users (until manually changed). You
usually do not want this. One easy way to setup a template user is to
login once with a valid LDAP user, let OpenERP create a blank local
login once with a valid LDAP user, let Odoo create a blank local
user with the same login (and a blank password), then rename this new
user to a username that does not exist in LDAP, and setup its groups
+1 -1
View File
@@ -1,3 +1,3 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import users_ldap
import models
+2 -5
View File
@@ -2,17 +2,14 @@
{
'name' : 'Authentication via LDAP',
'version' : '1.0',
'depends' : ['base'],
#'description': < auto-loaded from README file
'category' : 'Extra Tools',
'data' : [
'users_ldap_view.xml',
'user_ldap_installer.xml',
'views/res_company_views.xml',
'views/ldap_installer_views.xml',
'security/ir.model.access.csv',
],
'auto_install': False,
'installable': True,
'external_dependencies' : {
'python' : ['ldap'],
}
+5
View File
@@ -0,0 +1,5 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import res_company
import res_company_ldap
import res_users
+11
View File
@@ -0,0 +1,11 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo import fields, models
class ResCompany(models.Model):
_inherit = "res.company"
ldaps = fields.One2many('res.company.ldap', 'company', string='LDAP Parameters',
copy=True, groups="base.group_system")
+202
View File
@@ -0,0 +1,202 @@
# -*- 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
_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, 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, password.encode('utf-8'))
conn.unbind()
entry = results[0]
except ldap.INVALID_CREDENTIALS:
return False
except ldap.LDAPError, 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(ldap_binddn.encode('utf-8'), ldap_password.encode('utf-8'))
results = conn.search_st(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, 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
+41
View File
@@ -0,0 +1,41 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.exceptions import AccessDenied
from odoo import api, models, registry, SUPERUSER_ID
class Users(models.Model):
_inherit = "res.users"
def _login(self, db, login, password):
user_id = super(Users, self)._login(db, login, password)
if user_id:
return user_id
with registry(db).cursor() as cr:
env = api.Environment(cr, SUPERUSER_ID, {})
cr.execute("SELECT id FROM res_users WHERE lower(login)=%s", (login,))
res = cr.fetchone()
if res:
return False
Ldap = env['res.company.ldap']
for conf in Ldap.get_ldap_dicts():
entry = Ldap.authenticate(conf, login, password)
if entry:
user_id = Ldap.sudo().get_or_create_user(conf, login, entry)
if user_id:
break
return user_id
@api.model
def check_credentials(self, password):
try:
super(Users, self).check_credentials(password)
except AccessDenied:
if self.env.user.active:
Ldap = self.env['res.company.ldap']
for conf in Ldap.get_ldap_dicts():
if Ldap.authenticate(conf, self.env.user.login, password):
return
raise
-267
View File
@@ -1,267 +0,0 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import ldap
import logging
from ldap.filter import filter_format
import openerp.exceptions
from openerp import tools
from openerp.osv import fields, osv
from openerp import SUPERUSER_ID
from openerp.modules.registry import RegistryManager
_logger = logging.getLogger(__name__)
class CompanyLDAP(osv.osv):
_name = 'res.company.ldap'
_order = 'sequence'
_rec_name = 'ldap_server'
def get_ldap_dicts(self, cr, ids=None):
"""
Retrieve res_company_ldap resources from the database in dictionary
format.
:param list ids: Valid ids of model res_company_ldap. If not \
specified, process all resources (unlike other ORM methods).
:return: ldap configurations
:rtype: list of dictionaries
"""
if ids:
id_clause = 'AND id IN (%s)'
args = [tuple(ids)]
else:
id_clause = ''
args = []
cr.execute("""
SELECT id, company, ldap_server, ldap_server_port, ldap_binddn,
ldap_password, ldap_filter, ldap_base, "user", create_user,
ldap_tls
FROM res_company_ldap
WHERE ldap_server != '' """ + id_clause + """ ORDER BY sequence
""", args)
return cr.dictfetchall()
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, filter)
# Get rid of (None, attrs) for searchResultReference replies
results = [i for i in results if i[0]]
if results and len(results) == 1:
dn = results[0][0]
conn = self.connect(conf)
conn.simple_bind_s(dn, password.encode('utf-8'))
conn.unbind()
entry = results[0]
except ldap.INVALID_CREDENTIALS:
return False
except ldap.LDAPError, 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(ldap_binddn.encode('utf-8'), ldap_password.encode('utf-8'))
results = conn.search_st(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, e:
_logger.error('An LDAP exception occurred: %s', e)
return results
def map_ldap_attributes(self, cr, uid, 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
"""
values = { 'name': ldap_entry[1]['cn'][0],
'login': login,
'company_id': conf['company']
}
return values
def get_or_create_user(self, cr, uid, conf, login, ldap_entry,
context=None):
"""
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())
cr.execute("SELECT id, active FROM res_users WHERE lower(login)=%s", (login,))
res = 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)
user_obj = self.pool['res.users']
values = self.map_ldap_attributes(cr, uid, conf, login, ldap_entry)
if conf['user']:
values['active'] = True
user_id = user_obj.copy(cr, SUPERUSER_ID, conf['user'],
default=values)
else:
user_id = user_obj.create(cr, SUPERUSER_ID, values)
return user_id
_columns = {
'sequence': fields.integer('Sequence'),
'company': fields.many2one('res.company', 'Company', required=True,
ondelete='cascade'),
'ldap_server': fields.char('LDAP Server address', required=True),
'ldap_server_port': fields.integer('LDAP Server port', required=True),
'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('LDAP password',
help=("The password of the user account on the LDAP server that is "
"used to query the directory.")),
'ldap_filter': fields.char('LDAP filter', required=True),
'ldap_base': fields.char('LDAP base', required=True),
'user': fields.many2one('res.users', 'Template User',
help="User to copy when creating new users"),
'create_user': fields.boolean('Create user',
help="Automatically create local user accounts for new users authenticating via LDAP"),
'ldap_tls': fields.boolean('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."),
}
_defaults = {
'ldap_server': '127.0.0.1',
'ldap_server_port': 389,
'sequence': 10,
'create_user': True,
}
class res_company(osv.osv):
_inherit = "res.company"
_columns = {
'ldaps': fields.one2many(
'res.company.ldap', 'company', 'LDAP Parameters', copy=True, groups="base.group_system"),
}
class users(osv.osv):
_inherit = "res.users"
def _login(self, db, login, password):
user_id = super(users, self)._login(db, login, password)
if user_id:
return user_id
registry = RegistryManager.get(db)
with registry.cursor() as cr:
cr.execute("SELECT id FROM res_users WHERE lower(login)=%s", (login,))
res = cr.fetchone()
if res:
return False
ldap_obj = registry.get('res.company.ldap')
for conf in ldap_obj.get_ldap_dicts(cr):
entry = ldap_obj.authenticate(conf, login, password)
if entry:
user_id = ldap_obj.get_or_create_user(
cr, SUPERUSER_ID, conf, login, entry)
if user_id:
break
return user_id
def check_credentials(self, cr, uid, password):
try:
super(users, self).check_credentials(cr, uid, password)
except openerp.exceptions.AccessDenied:
cr.execute('SELECT login FROM res_users WHERE id=%s AND active=TRUE',
(int(uid),))
res = cr.fetchone()
if res:
ldap_obj = self.pool['res.company.ldap']
for conf in ldap_obj.get_ldap_dicts(cr):
if ldap_obj.authenticate(conf, res[0], password):
return
raise
@@ -1,9 +1,9 @@
<openerp>
<data>
<?xml version="1.0" encoding="utf-8"?>
<odoo>
<!-- ldap installer Form View -->
<record model="ir.ui.view" id="view_ldap_installer_form">
<field name="name">ldap.installer.form</field>
<record id="view_ldap_installer_form" model="ir.ui.view">
<field name="name">res.company.ldap.form</field>
<field name="model">res.company.ldap</field>
<field name="arch" type="xml">
<form string="LDAP Configuration">
@@ -43,5 +43,4 @@
<field name="view_id" ref="view_ldap_installer_form"/>
</record>
</data>
</openerp>
</odoo>
@@ -1,8 +1,7 @@
<?xml version="1.0"?>
<openerp>
<data>
<record model="ir.ui.view" id="company_form_view">
<field name="name">res.company.form.inherit.users_ldap</field>
<?xml version="1.0" encoding="utf-8"?>
<odoo>
<record id="company_form_view" model="ir.ui.view">
<field name="name">res.company.form.inherit</field>
<field name="model">res.company</field>
<field name="inherit_id" ref="base.view_company_form"/>
<field name="arch" type="xml">
@@ -35,5 +34,4 @@
</xpath>
</field>
</record>
</data>
</openerp>
</odoo>