414 lines
14 KiB
Python
414 lines
14 KiB
Python
# Copyright (C) 2010-2024 RhodeCode GmbH
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License, version 3
|
|
# (only), as published by the Free Software Foundation.
|
|
#
|
|
# 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 Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
# This program is dual-licensed. If you wish to learn more about the
|
|
# RhodeCode Enterprise Edition, including its added features, Support services,
|
|
# and proprietary license terms, please see https://rhodecode.com/licenses/
|
|
|
|
"""
|
|
RhodeCode authentication plugin for LDAP
|
|
"""
|
|
|
|
import logging
|
|
|
|
import colander
|
|
|
|
from rhodecode.authentication.plugins.services.ldap_dao import LdapDao
|
|
from rhodecode.translation import _
|
|
from rhodecode.authentication.base import RhodeCodeExternalAuthPlugin, hybrid_property
|
|
from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase, TwoFactorAuthnPluginSettingsSchemaMixin
|
|
from rhodecode.authentication.routes import AuthnPluginResourceBase
|
|
from rhodecode.lib.colander_utils import strip_whitespace
|
|
from rhodecode.lib.exceptions import LdapUsernameError, LdapPasswordError, LdapImportError
|
|
from rhodecode.lib.str_utils import safe_str
|
|
from rhodecode.model.db import User
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class LdapError(Exception):
|
|
pass
|
|
|
|
|
|
def plugin_factory(plugin_id, *args, **kwargs):
|
|
"""
|
|
Factory function that is called during plugin discovery.
|
|
It returns the plugin instance.
|
|
"""
|
|
plugin = RhodeCodeAuthPlugin(plugin_id)
|
|
return plugin
|
|
|
|
|
|
class LdapAuthnResource(AuthnPluginResourceBase):
|
|
pass
|
|
|
|
|
|
class LdapSettingsSchema(TwoFactorAuthnPluginSettingsSchemaMixin, AuthnPluginSettingsSchemaBase):
|
|
tls_kind_choices = ["PLAIN", "LDAPS", "START_TLS"]
|
|
tls_reqcert_choices = ["NEVER", "ALLOW", "TRY", "DEMAND", "HARD"]
|
|
search_scope_choices = ["BASE", "ONELEVEL", "SUBTREE"]
|
|
|
|
host = colander.SchemaNode(
|
|
colander.String(),
|
|
default="",
|
|
description=_(
|
|
"Host[s] of the LDAP Server \n"
|
|
"(e.g., 192.168.2.154, or ldap-server.domain.com.\n "
|
|
"Multiple servers can be specified using commas"
|
|
),
|
|
preparer=strip_whitespace,
|
|
title=_("LDAP Host"),
|
|
widget="string",
|
|
)
|
|
port = colander.SchemaNode(
|
|
colander.Int(),
|
|
default=389,
|
|
description=_(
|
|
"Custom port that the LDAP server is listening on. Default value is: 389, use 636 for LDAPS (SSL)"
|
|
),
|
|
preparer=strip_whitespace,
|
|
title=_("Port"),
|
|
validator=colander.Range(min=0, max=65536),
|
|
widget="int",
|
|
)
|
|
|
|
timeout = colander.SchemaNode(
|
|
colander.Int(),
|
|
default=60 * 5,
|
|
description=_("Timeout for LDAP connection"),
|
|
preparer=strip_whitespace,
|
|
title=_("Connection timeout"),
|
|
validator=colander.Range(min=1),
|
|
widget="int",
|
|
)
|
|
|
|
dn_user = colander.SchemaNode(
|
|
colander.String(),
|
|
default="",
|
|
description=_(
|
|
"Optional user DN/account to connect to LDAP if authentication is required. \n"
|
|
"e.g., cn=admin,dc=mydomain,dc=com, or "
|
|
"uid=root,cn=users,dc=mydomain,dc=com, or admin@mydomain.com"
|
|
),
|
|
missing="",
|
|
preparer=strip_whitespace,
|
|
title=_("Bind account"),
|
|
widget="string",
|
|
)
|
|
dn_pass = colander.SchemaNode(
|
|
colander.String(),
|
|
default="",
|
|
description=_("Password to authenticate for given user DN."),
|
|
missing="",
|
|
preparer=strip_whitespace,
|
|
title=_("Bind account password"),
|
|
widget="password",
|
|
)
|
|
tls_kind = colander.SchemaNode(
|
|
colander.String(),
|
|
default=tls_kind_choices[0],
|
|
description=_("TLS Type"),
|
|
title=_("Connection Security"),
|
|
validator=colander.OneOf(tls_kind_choices),
|
|
widget="select",
|
|
)
|
|
tls_reqcert = colander.SchemaNode(
|
|
colander.String(),
|
|
default=tls_reqcert_choices[0],
|
|
description=_(
|
|
"Require Cert over TLS?. Self-signed and custom "
|
|
"certificates can be used when\n `RhodeCode Certificate` "
|
|
"found in admin > settings > system info page is extended."
|
|
),
|
|
title=_("Certificate Checks"),
|
|
validator=colander.OneOf(tls_reqcert_choices),
|
|
widget="select",
|
|
)
|
|
tls_cert_file = colander.SchemaNode(
|
|
colander.String(),
|
|
default="",
|
|
description=_(
|
|
"This specifies the PEM-format file path containing "
|
|
"certificates for use in TLS connection.\n"
|
|
"If not specified `TLS Cert dir` will be used"
|
|
),
|
|
title=_("TLS Cert file"),
|
|
missing="",
|
|
widget="string",
|
|
)
|
|
tls_cert_dir = colander.SchemaNode(
|
|
colander.String(),
|
|
default=LdapDao.default_tls_cert_dir,
|
|
description=_(
|
|
"This specifies the path of a directory that contains individual CA certificates in separate files."
|
|
),
|
|
title=_("TLS Cert dir"),
|
|
widget="string",
|
|
)
|
|
base_dn = colander.SchemaNode(
|
|
colander.String(),
|
|
default="",
|
|
description=_(
|
|
"Base DN to search. Dynamic bind is supported. Add `$login` marker "
|
|
"in it to be replaced with current user username \n"
|
|
"(e.g., dc=mydomain,dc=com, or ou=Users,dc=mydomain,dc=com)"
|
|
),
|
|
missing="",
|
|
preparer=strip_whitespace,
|
|
title=_("Base DN"),
|
|
widget="string",
|
|
)
|
|
filter = colander.SchemaNode(
|
|
colander.String(),
|
|
default="",
|
|
description=_(
|
|
"Filter to narrow results \n"
|
|
"(e.g., (&(objectCategory=Person)(objectClass=user)), or \n"
|
|
"(memberof=cn=rc-login,ou=groups,ou=company,dc=mydomain,dc=com)))"
|
|
),
|
|
missing="",
|
|
preparer=strip_whitespace,
|
|
title=_("LDAP Search Filter"),
|
|
widget="string",
|
|
)
|
|
|
|
search_scope = colander.SchemaNode(
|
|
colander.String(),
|
|
default=search_scope_choices[2],
|
|
description=_("How deep to search LDAP. If unsure set to SUBTREE"),
|
|
title=_("LDAP Search Scope"),
|
|
validator=colander.OneOf(search_scope_choices),
|
|
widget="select",
|
|
)
|
|
attr_login = colander.SchemaNode(
|
|
colander.String(),
|
|
default="uid",
|
|
description=_("LDAP Attribute to map to user name (e.g., uid, or sAMAccountName)"),
|
|
preparer=strip_whitespace,
|
|
title=_("Login Attribute"),
|
|
missing_msg=_("The LDAP Login attribute of the CN must be specified"),
|
|
widget="string",
|
|
)
|
|
attr_email = colander.SchemaNode(
|
|
colander.String(),
|
|
default="",
|
|
description=_(
|
|
"LDAP Attribute to map to email address (e.g., mail).\n"
|
|
"Emails are a crucial part of RhodeCode. \n"
|
|
"If possible add a valid email attribute to ldap users."
|
|
),
|
|
missing="",
|
|
preparer=strip_whitespace,
|
|
title=_("Email Attribute"),
|
|
widget="string",
|
|
)
|
|
attr_firstname = colander.SchemaNode(
|
|
colander.String(),
|
|
default="",
|
|
description=_("LDAP Attribute to map to first name (e.g., givenName)"),
|
|
missing="",
|
|
preparer=strip_whitespace,
|
|
title=_("First Name Attribute"),
|
|
widget="string",
|
|
)
|
|
attr_lastname = colander.SchemaNode(
|
|
colander.String(),
|
|
default="",
|
|
description=_("LDAP Attribute to map to last name (e.g., sn)"),
|
|
missing="",
|
|
preparer=strip_whitespace,
|
|
title=_("Last Name Attribute"),
|
|
widget="string",
|
|
)
|
|
sync_active_directory_users = colander.SchemaNode(
|
|
colander.Bool(),
|
|
default=False,
|
|
description=_(
|
|
"A cron job that periodically retrieves all users from an LDAP-based Active Directory server and syncs "
|
|
"them with the RhodeCode database.\n"
|
|
"Note: This feature is specific to Active Directory. Enabling it for other types of LDAP servers will have no effect."
|
|
),
|
|
missing=False,
|
|
preparer=strip_whitespace,
|
|
title=_("Sync Active Directory Users"),
|
|
widget="bool",
|
|
)
|
|
|
|
|
|
def get_ldap_args(settings: dict):
|
|
return {
|
|
"server": settings.get("host", ""),
|
|
"base_dn": settings.get("base_dn", ""),
|
|
"port": settings.get("port"),
|
|
"bind_dn": settings.get("dn_user"),
|
|
"bind_pass": settings.get("dn_pass"),
|
|
"tls_kind": settings.get("tls_kind"),
|
|
"tls_reqcert": settings.get("tls_reqcert"),
|
|
"tls_cert_file": settings.get("tls_cert_file"),
|
|
"tls_cert_dir": settings.get("tls_cert_dir"),
|
|
"search_scope": settings.get("search_scope"),
|
|
"attr_login": settings.get("attr_login"),
|
|
"ldap_version": 3,
|
|
"ldap_filter": settings.get("filter"),
|
|
"timeout": settings.get("timeout"),
|
|
}
|
|
|
|
|
|
class RhodeCodeAuthPlugin(RhodeCodeExternalAuthPlugin):
|
|
uid = "ldap"
|
|
# used to define dynamic binding in the
|
|
DYNAMIC_BIND_VAR = "$login"
|
|
_settings_unsafe_keys = ["dn_pass"]
|
|
|
|
def includeme(self, config):
|
|
config.add_authn_plugin(self)
|
|
config.add_authn_resource(self.get_id(), LdapAuthnResource(self))
|
|
config.add_view(
|
|
"rhodecode.authentication.views.AuthnPluginViewBase",
|
|
attr="settings_get",
|
|
renderer="rhodecode:templates/admin/auth/plugin_settings.mako",
|
|
request_method="GET",
|
|
route_name="auth_home",
|
|
context=LdapAuthnResource,
|
|
)
|
|
config.add_view(
|
|
"rhodecode.authentication.views.AuthnPluginViewBase",
|
|
attr="settings_post",
|
|
renderer="rhodecode:templates/admin/auth/plugin_settings.mako",
|
|
request_method="POST",
|
|
route_name="auth_home",
|
|
context=LdapAuthnResource,
|
|
)
|
|
|
|
def get_settings_schema(self):
|
|
return LdapSettingsSchema()
|
|
|
|
def get_display_name(self, load_from_settings=False):
|
|
return _("LDAP")
|
|
|
|
@classmethod
|
|
def docs(cls):
|
|
return "https://docs.rhodecode.com/RhodeCode-Enterprise/auth/auth-ldap.html"
|
|
|
|
@hybrid_property
|
|
def name(self):
|
|
return "ldap"
|
|
|
|
def use_fake_password(self):
|
|
return True
|
|
|
|
def user_activation_state(self):
|
|
def_user_perms = User.get_default_user().AuthUser().permissions["global"]
|
|
return "hg.extern_activate.auto" in def_user_perms
|
|
|
|
def try_dynamic_binding(self, username, password, current_args):
|
|
"""
|
|
Detects marker inside our original bind, and uses dynamic auth if
|
|
present
|
|
"""
|
|
|
|
org_bind = current_args["bind_dn"]
|
|
passwd = current_args["bind_pass"]
|
|
|
|
def has_bind_marker(_username):
|
|
if self.DYNAMIC_BIND_VAR in _username:
|
|
return True
|
|
|
|
# we only passed in user with "special" variable
|
|
if org_bind and has_bind_marker(org_bind) and not passwd:
|
|
log.debug(
|
|
"Using dynamic user/password binding for ldap authentication. Replacing `%s` with username",
|
|
self.DYNAMIC_BIND_VAR,
|
|
)
|
|
current_args["bind_dn"] = org_bind.replace(self.DYNAMIC_BIND_VAR, username)
|
|
current_args["bind_pass"] = password
|
|
|
|
return current_args
|
|
|
|
def auth(self, userobj, username, password, settings, **kwargs):
|
|
"""
|
|
Given a user object (which may be null), username, a plaintext password,
|
|
and a settings object (containing all the keys needed as listed in
|
|
settings()), authenticate this user's login attempt.
|
|
|
|
Return None on failure. On success, return a dictionary of the form:
|
|
|
|
see: RhodeCodeAuthPluginBase.auth_func_attrs
|
|
This is later validated for correctness
|
|
"""
|
|
|
|
if not username or not password:
|
|
log.debug("Empty username or password skipping...")
|
|
return None
|
|
|
|
ldap_args = get_ldap_args(settings)
|
|
|
|
ldap_attrs = self.try_dynamic_binding(username, password, ldap_args)
|
|
|
|
log.debug("Checking for ldap authentication.")
|
|
|
|
try:
|
|
ldap_dao = LdapDao(**ldap_args)
|
|
(user_dn, ldap_attrs) = ldap_dao.authenticate_ldap(username, password)
|
|
log.debug("Got ldap DN response %s", user_dn)
|
|
|
|
def get_ldap_attr(k) -> str:
|
|
return safe_str(ldap_attrs.get(settings.get(k), [b""])[0])
|
|
|
|
# old attrs fetched from RhodeCode database
|
|
admin = getattr(userobj, "admin", False)
|
|
active = getattr(userobj, "active", True)
|
|
email = getattr(userobj, "email", "")
|
|
username = getattr(userobj, "username", username)
|
|
firstname = getattr(userobj, "firstname", "")
|
|
lastname = getattr(userobj, "lastname", "")
|
|
extern_type = getattr(userobj, "extern_type", "")
|
|
|
|
groups = []
|
|
|
|
user_attrs = {
|
|
"username": username,
|
|
"firstname": get_ldap_attr("attr_firstname") or firstname,
|
|
"lastname": get_ldap_attr("attr_lastname") or lastname,
|
|
"groups": groups,
|
|
"user_group_sync": False,
|
|
"email": get_ldap_attr("attr_email") or email,
|
|
"admin": admin,
|
|
"active": active,
|
|
"active_from_extern": None,
|
|
"extern_name": user_dn,
|
|
"extern_type": extern_type,
|
|
}
|
|
|
|
log.debug("ldap user: %s", user_attrs)
|
|
log.info(
|
|
"user `%s` authenticated correctly",
|
|
user_attrs["username"],
|
|
extra={"action": "user_auth_ok", "auth_module": "auth_ldap", "username": user_attrs["username"]},
|
|
)
|
|
|
|
return user_attrs
|
|
|
|
except (LdapUsernameError, LdapPasswordError, LdapImportError):
|
|
log.exception("LDAP related exception")
|
|
return None
|
|
except (Exception,):
|
|
log.exception("Other exception")
|
|
return None
|
|
|
|
|
|
def includeme(config):
|
|
plugin_id = f"egg:rhodecode-enterprise-ce#{RhodeCodeAuthPlugin.uid}"
|
|
plugin_factory(plugin_id).includeme(config)
|