codekingpro/portable-devtools
114k
1##########################################################################2#3# pgAdmin 4 - PostgreSQL Tools4#5# Copyright (C) 2013 - 2024, The pgAdmin Development Team6# This software is released under the PostgreSQL Licence7#8##########################################################################9 10"""A blueprint module implementing the ldap authentication."""11 12import ssl13import config14from ldap3 import Connection, Server, Tls, ALL, ALL_ATTRIBUTES, ANONYMOUS,\15 SIMPLE, AUTO_BIND_TLS_BEFORE_BIND, AUTO_BIND_NO_TLS, set_config_parameter16from ldap3.core.exceptions import LDAPSocketOpenError, LDAPBindError,\17 LDAPInvalidScopeError, LDAPAttributeError, LDAPInvalidFilterError,\18 LDAPStartTLSError, LDAPSSLConfigurationError19from flask_babel import gettext20from urllib.parse import urlparse21 22from .internal import BaseAuthentication23from pgadmin.model import User, ServerGroup, db, Role24from flask import current_app25from pgadmin.tools.user_management import create_user26from pgadmin.utils.constants import LDAP27from sqlalchemy import func28from flask_security import login_user29 30ERROR_SEARCHING_LDAP_DIRECTORY = gettext(31 "Error searching the LDAP directory: {}")32 33ERROR_CONNECTING_LDAP_SERVER = gettext(34 "Error connecting to the LDAP server: {}\n")35 36if config.LDAP_IGNORE_MALFORMED_SCHEMA:37 set_config_parameter('IGNORE_MALFORMED_SCHEMA',38 config.LDAP_IGNORE_MALFORMED_SCHEMA)39 40 41class LDAPAuthentication(BaseAuthentication):42 """Ldap Authentication Class"""43 44 def get_source_name(self):45 return LDAP46 47 def get_friendly_name(self):48 return gettext("ldap")49 50 def authenticate(self, form):51 self.username = form.data['email']52 self.password = form.data['password']53 self.dedicated_user = True54 self.start_tls = False55 user_email = None56 57 # Check the dedicated ldap user58 self.bind_user = getattr(config, 'LDAP_BIND_USER', None)59 self.bind_pass = getattr(config, 'LDAP_BIND_PASSWORD', None)60 61 # Check for the anonymous binding62 self.anonymous_bind = getattr(config, 'LDAP_ANONYMOUS_BIND', False)63 64 if self.bind_user and not self.bind_pass:65 return False, gettext(66 "LDAP configuration error: Set the bind password.")67 68 # if no dedicated ldap user is configured then use the login69 # username and password70 if not self.bind_user and not self.bind_pass and\71 self.anonymous_bind is False:72 73 user_dn = config.LDAP_BIND_FORMAT\74 .format(75 LDAP_USERNAME=self.username,76 LDAP_BASE_DN=config.LDAP_BASE_DN,77 LDAP_USERNAME_ATTRIBUTE=config.LDAP_USERNAME_ATTRIBUTE78 )79 80 self.bind_user = user_dn81 self.bind_pass = self.password82 self.dedicated_user = False83 84 # Connect ldap server85 status, msg = self.connect()86 87 if not status:88 return status, msg89 90 status, ldap_user = self.search_ldap_user()91 92 if not status:93 return status, ldap_user94 95 # If dedicated user is configured96 if self.dedicated_user:97 # Get the user DN from the user ldap entry98 self.bind_user = ldap_user.entry_dn99 self.bind_pass = self.password100 self.anonymous_bind = False101 status, msg = self.connect()102 103 if not status:104 return status, msg105 106 if 'mail' in ldap_user:107 mail = ldap_user['mail'].value108 if isinstance(mail, list) and len(mail) > 0:109 user_email = mail[0]110 else:111 user_email = ldap_user['mail'].value112 113 return self.__auto_create_user(user_email)114 115 def connect(self):116 """Setup the connection to the LDAP server and authenticate the user.117 """118 status, server = self._configure_server()119 120 if not status:121 return status, server122 123 auto_bind = AUTO_BIND_TLS_BEFORE_BIND if self.start_tls \124 else AUTO_BIND_NO_TLS125 126 # Create the connection127 try:128 if self.anonymous_bind:129 self.conn = Connection(server,130 auto_bind=auto_bind,131 authentication=ANONYMOUS132 )133 else:134 self.conn = Connection(server,135 user=self.bind_user,136 password=self.bind_pass,137 auto_bind=auto_bind,138 authentication=SIMPLE139 )140 141 except LDAPSocketOpenError as e:142 current_app.logger.exception(143 ERROR_CONNECTING_LDAP_SERVER.format(e))144 return False, ERROR_CONNECTING_LDAP_SERVER.format(e.args[0])145 except LDAPBindError as e:146 current_app.logger.exception(147 "Error binding to the LDAP server.")148 return False, gettext("Error binding to the LDAP server: {}\n".149 format(e.args[0]))150 except LDAPStartTLSError as e:151 current_app.logger.exception(152 "Error starting TLS: {}\n".format(e))153 return False, gettext("Error starting TLS: {}\n"154 ).format(e.args[0])155 except Exception as e:156 current_app.logger.exception(157 ERROR_CONNECTING_LDAP_SERVER.format(e))158 return False, ERROR_CONNECTING_LDAP_SERVER.format(e.args[0])159 160 return True, None161 162 def login(self, form):163 user = getattr(form, 'user', None)164 if user is None:165 if config.LDAP_DN_CASE_SENSITIVE:166 user = User.query.filter_by(username=self.username).first()167 else:168 user = User.query.filter(169 func.lower(User.username) == func.lower(170 self.username)).first()171 172 if user is None:173 current_app.logger.exception(174 self.messages('USER_DOES_NOT_EXIST'))175 return False, self.messages('USER_DOES_NOT_EXIST')176 177 # Login user through flask_security178 status = login_user(user)179 if not status:180 current_app.logger.exception(self.messages('LOGIN_FAILED'))181 return False, self.messages('LOGIN_FAILED')182 current_app.logger.info(183 "LDAP user {0} logged in.".format(user))184 return True, None185 186 def __auto_create_user(self, user_email):187 """Add the ldap user to the internal SQLite database."""188 if config.LDAP_AUTO_CREATE_USER:189 if config.LDAP_DN_CASE_SENSITIVE:190 user = User.query.filter_by(username=self.username).first()191 else:192 user = User.query.filter(193 func.lower(User.username) == func.lower(194 self.username)).first()195 196 if user is None:197 create_msg = ("Creating user {0} with email {1} "198 "from auth source LDAP.")199 current_app.logger.info(create_msg.format(self.username,200 user_email))201 return create_user({202 'username': self.username,203 'email': user_email,204 'role': 2,205 'active': True,206 'auth_source': LDAP207 })208 209 return True, None210 211 def __configure_tls(self):212 ca_cert_file = getattr(config, 'LDAP_CA_CERT_FILE', None)213 cert_file = getattr(config, 'LDAP_CERT_FILE', None)214 key_file = getattr(config, 'LDAP_KEY_FILE', None)215 cert_validate = ssl.CERT_NONE216 217 if ca_cert_file and cert_file and key_file:218 cert_validate = ssl.CERT_REQUIRED219 220 try:221 tls = Tls(222 local_private_key_file=key_file,223 local_certificate_file=cert_file,224 validate=cert_validate,225 version=ssl.PROTOCOL_TLSv1_2,226 ca_certs_file=ca_cert_file)227 except LDAPSSLConfigurationError as e:228 current_app.logger.exception(229 "LDAP configuration error: {}\n".format(e))230 return False, gettext("LDAP configuration error: {}\n").format(231 e.args[0])232 return True, tls233 234 def _configure_server(self):235 # Parse the server URI236 uri = getattr(config, 'LDAP_SERVER_URI', None)237 238 if uri:239 uri = urlparse(uri)240 241 # Create the TLS configuration object if required242 tls = None243 244 if isinstance(uri, str):245 return False, gettext(246 "LDAP configuration error: Set the proper LDAP URI.")247 248 if uri.scheme == 'ldaps' or config.LDAP_USE_STARTTLS:249 status, tls = self.__configure_tls()250 if not status:251 return status, tls252 253 if uri.scheme != 'ldaps' and config.LDAP_USE_STARTTLS:254 self.start_tls = True255 256 try:257 # Create the server object258 server = Server(uri.hostname,259 port=uri.port,260 use_ssl=(uri.scheme == 'ldaps'),261 get_info=ALL,262 tls=tls,263 connect_timeout=config.LDAP_CONNECTION_TIMEOUT)264 except ValueError as e:265 return False, "LDAP configuration error: {}.".format(e)266 267 return True, server268 269 def search_ldap_user(self):270 """Get a list of users from the LDAP server based on config271 search criteria."""272 try:273 search_base_dn = config.LDAP_SEARCH_BASE_DN274 if (not search_base_dn or search_base_dn == '<Search-Base-DN>')\275 and (self.anonymous_bind or self.dedicated_user):276 return False, gettext("LDAP configuration error: "277 "Set the Search Domain.")278 elif not search_base_dn or search_base_dn == '<Search-Base-DN>':279 search_base_dn = config.LDAP_BASE_DN280 281 search_filter = "({0}={1})".format(config.LDAP_USERNAME_ATTRIBUTE,282 self.username)283 if config.LDAP_SEARCH_FILTER:284 search_filter = "(&{0}{1})".format(search_filter,285 config.LDAP_SEARCH_FILTER)286 287 self.conn.search(search_base=search_base_dn,288 search_filter=search_filter,289 search_scope=config.LDAP_SEARCH_SCOPE,290 attributes=ALL_ATTRIBUTES291 )292 293 except LDAPInvalidScopeError as e:294 current_app.logger.exception(295 ERROR_SEARCHING_LDAP_DIRECTORY.format(e.args[0])296 )297 return False, ERROR_SEARCHING_LDAP_DIRECTORY.format(e.args[0])298 except LDAPAttributeError as e:299 current_app.logger.exception(300 ERROR_SEARCHING_LDAP_DIRECTORY.format(e)301 )302 return False, ERROR_SEARCHING_LDAP_DIRECTORY.format(e.args[0])303 except LDAPInvalidFilterError as e:304 current_app.logger.exception(305 ERROR_SEARCHING_LDAP_DIRECTORY.format(e)306 )307 return False, ERROR_SEARCHING_LDAP_DIRECTORY.format(e.args[0])308 309 results = len(self.conn.entries)310 if results > 1:311 return False, ERROR_SEARCHING_LDAP_DIRECTORY.format(312 gettext("More than one result found."))313 elif results < 1:314 return False, ERROR_SEARCHING_LDAP_DIRECTORY.format(315 gettext("Could not find the specified user."))316 return True, self.conn.entries[0]317