Coverage for gws-app/gws/plugin/auth_provider/ldap/__init__.py: 93%
185 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-05 13:35 +0200
1"""LDAP authentication provider.
3Authenticates users against an LDAP or Active Directory server. The server is
4given as an LDAP URL, a subset of the RFC 2255 form::
6 ldap://host:port/baseDN?searchAttribute
8``searchAttribute`` is the attribute that holds the login name. If ``ssl`` is
9configured, the connection uses ``ldaps``. A bind DN and a password can be
10given; this DN must be allowed to search the directory. Without a bind DN,
11the provider binds anonymously. The connection is tested at configuration
12time.
14Authentication with ``username`` and ``password`` credentials works as
15follows:
17- connect to the server, binding as the bind DN if configured,
18- search below the base DN for the entry with ``searchAttribute = username``;
19 more than one entry is an error,
20- reject Active Directory accounts with the ``ACCOUNTDISABLE`` flag in
21 ``userAccountControl``,
22- bind as the found DN with the password,
23- connect again and assign roles from the ``users`` rules.
25A ``users`` rule assigns its ``roles`` if the account is one of the entries
26found by the ``matches`` filter, or if the account is listed in the
27``member``, ``members`` or ``uniqueMember`` attribute of a group found by the
28``memberOf`` filter. The attributes of the entry are passed to the user
29record. If the entry has no ``displayName``, ``displayNameFormat`` creates
30one from the attributes.
32Example::
34 auth.providers+ {
35 type "ldap"
36 url "ldap://ldap.example.com:389/dc=example,dc=com?uid"
37 bindDN "cn=admin,dc=example,dc=com"
38 bindPassword "secret"
39 activeDirectory false
40 users [
41 { matches "(uid=admin)" roles ["admin"] }
42 { memberOf "(cn=editors)" roles ["editor"] }
43 ]
44 }
46References:
47 https://datatracker.ietf.org/doc/html/rfc2255
48"""
50from typing import Optional
52import contextlib
54import ldap
55import ldap.filter
57import gws
58import gws.base.auth
59import gws.lib.net
62class UserSpec(gws.Data):
63 """Rule that assigns GWS roles to LDAP accounts."""
65 roles: list[str]
66 """GWS roles assigned to matching accounts."""
67 matches: Optional[str]
68 """LDAP filter the account has to match."""
69 memberOf: Optional[str]
70 """LDAP filter for groups the account has to be a member of."""
73class SSLConfig(gws.Config):
74 """SSL settings for the LDAP connection."""
76 ca: Optional[gws.FilePath]
77 """Path to the CA certificate file."""
78 crt: Optional[gws.FilePath]
79 """Path to the client certificate file."""
80 key: Optional[gws.FilePath]
81 """Path to the client key file."""
84@gws.ext.config.authProvider('ldap')
85class Config(gws.base.auth.provider.Config):
86 """Authentication against an LDAP or Active Directory server."""
88 activeDirectory: bool = True
89 """The LDAP server is an Active Directory."""
90 bindDN: Optional[str]
91 """DN to bind as for user lookups, anonymous bind if empty."""
92 bindPassword: Optional[str]
93 """Password for the bind DN."""
94 displayNameFormat: Optional[gws.FormatStr]
95 """Format string for the user's display name."""
96 users: list[UserSpec]
97 """Rules that assign GWS roles to LDAP accounts."""
98 timeout: gws.Duration = '30'
99 """Network timeout for LDAP connections."""
100 url: str
101 """LDAP server URL."""
102 ssl: Optional[SSLConfig]
103 """SSL settings, enables ldaps."""
106@gws.ext.object.authProvider('ldap')
107class Object(gws.base.auth.provider.Object):
108 """LDAP authentication provider."""
110 serverUrl: str
111 """Server URL without path and query, with the ``ldap`` or ``ldaps`` scheme."""
112 baseDN: str
113 """Base DN for searches."""
114 loginAttribute: str
115 """Attribute that holds the login name."""
116 timeout: int
117 """Network timeout in seconds."""
118 ssl: Optional[SSLConfig]
119 """SSL settings, or ``None``."""
120 activeDirectory: bool
121 """The server is an Active Directory."""
122 bindDN: str
123 """DN to bind as, empty for an anonymous bind."""
124 bindPassword: str
125 """Password for the bind DN."""
126 displayNameFormat: str
127 """Format string for the display name, empty if none."""
128 users: list[UserSpec]
129 """Rules that assign roles to accounts."""
131 def configure(self):
132 self.timeout = self.cfg('timeout', default=30)
134 self.activeDirectory = self.cfg('activeDirectory', default=True)
135 self.bindDN = self.cfg('bindDN', default='')
136 self.bindPassword = self.cfg('bindPassword', default='')
137 self.displayNameFormat = self.cfg('displayNameFormat', default='')
138 self.ssl = self.cfg('ssl')
139 self.users = self.cfg('users', default=[])
141 proto = 'ldaps' if self.ssl else 'ldap'
142 p = gws.lib.net.parse_url(self.cfg('url'))
144 self.serverUrl = proto + '://' + p.netloc
145 self.baseDN = p.path.strip('/')
146 self.loginAttribute = p.query
148 try:
149 with self._connection():
150 gws.log.debug(f'LDAP connection {self.uid!r} ok')
151 except Exception as exc:
152 raise gws.Error(f'LDAP connection error: {exc.__class__.__name__}') from exc
154 def authenticate(self, method, credentials):
155 username = credentials.get('username', '').strip()
156 password = credentials.get('password', '').strip()
157 if not username or not password:
158 return
160 with self._connection() as conn:
161 rec = self._get_user_record(conn, username, password)
162 if not rec:
163 return
165 # NB need to rebind as admin
166 with self._connection() as conn:
167 return self._make_user(conn, rec)
169 def get_user(self, local_uid):
170 with self._connection() as conn:
171 users = self._find(conn, _make_filter({self.loginAttribute: local_uid}))
172 if len(users) == 1:
173 return self._make_user(conn, users[0])
175 ##
177 def _get_user_record(self, conn, username, password):
178 """Find the entry for a login name and bind as it with the password."""
179 users = self._find(conn, _make_filter({self.loginAttribute: username}))
181 if len(users) == 0:
182 return
183 if len(users) > 1:
184 raise gws.AuthenticationError(f'multiple entries for {username!r}')
186 rec = users[0]
188 # check for AD disabled accounts
189 uac = str(rec.get('userAccountControl', ''))
190 if uac and uac.isdigit():
191 if int(uac) & _MS_ACCOUNTDISABLE:
192 raise gws.ForbiddenError('ACCOUNTDISABLE flag set')
194 try:
195 conn.simple_bind_s(rec['dn'], password)
196 return rec
197 except ldap.INVALID_CREDENTIALS:
198 raise gws.AuthenticationError(f'wrong password for {username!r}')
199 except ldap.LDAPError as exc:
200 gws.log.exception()
201 raise gws.ForbiddenError(f'LDAP error {exc.__class__.__name__}') from exc
203 def _make_user(self, conn, rec):
204 """Create a user from an LDAP entry."""
205 user_rec = dict(rec)
206 user_rec['roles'] = self._roles_for_user(conn, rec)
208 if not user_rec.get('displayName') and self.displayNameFormat:
209 user_rec['displayName'] = gws.u.format_map(self.displayNameFormat, rec)
211 login = user_rec.pop(self.loginAttribute, '')
212 user_rec['localUid'] = user_rec['loginName'] = login
214 return gws.base.auth.user.from_record(self, user_rec)
216 def _roles_for_user(self, conn, rec):
217 """Return the roles of all ``users`` rules that match the entry."""
218 user_dn = rec['dn']
219 roles = set()
221 for u in self.users:
222 if u.get('matches'):
223 for dct in self._find(conn, u.matches):
224 if dct['dn'] == user_dn:
225 roles.update(u.roles)
227 if u.get('memberOf'):
228 for dct in self._find(conn, u.memberOf):
229 if _is_member_of(dct, user_dn):
230 roles.update(u.roles)
232 return sorted(roles)
234 def _find(self, conn, flt):
235 """Search below the base DN and return the entries as dicts."""
236 try:
237 res = conn.search_s(self.baseDN, ldap.SCOPE_SUBTREE, flt)
238 except ldap.NO_SUCH_OBJECT:
239 return []
241 dcts = []
243 for dn, data in res:
244 if dn:
245 d = _as_dict(data)
246 d['dn'] = dn
247 dcts.append(d)
249 return dcts
251 @contextlib.contextmanager
252 def _connection(self):
253 """Open a connection, bound as the bind DN if configured."""
254 conn = ldap.initialize(self.serverUrl)
255 conn.set_option(ldap.OPT_NETWORK_TIMEOUT, self.timeout)
257 if self.ssl:
258 if self.root.app.developer_option('ldap.ssl_insecure'):
259 conn.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
260 else:
261 conn.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_DEMAND)
262 if self.ssl.ca:
263 conn.set_option(ldap.OPT_X_TLS_CACERTFILE, self.ssl.ca)
264 if self.ssl.crt and self.ssl.key:
265 conn.set_option(ldap.OPT_X_TLS_CERTFILE, self.ssl.crt)
266 conn.set_option(ldap.OPT_X_TLS_KEYFILE, self.ssl.key)
267 conn.set_option(ldap.OPT_X_TLS_NEWCTX, 0)
269 if self.activeDirectory:
270 # see https://www.python-ldap.org/faq.html#usage
271 conn.set_option(ldap.OPT_REFERRALS, 0)
273 if self.bindDN:
274 conn.simple_bind_s(self.bindDN, self.bindPassword)
276 try:
277 yield conn
278 finally:
279 conn.unbind_s()
282def _as_dict(data):
283 """Convert attribute values to strings, unwrapping single values."""
284 d = {}
286 for k, v in data.items():
287 if not v:
288 continue
289 if not isinstance(v, list):
290 v = [v]
291 v = [gws.u.to_str(s) for s in v]
292 d[k] = v[0] if len(v) == 1 else v
294 return d
297def _make_filter(filter_dict):
298 """Create an AND filter that matches the given attribute values."""
299 conds = ''.join(
300 '({}={})'.format(
301 ldap.filter.escape_filter_chars(k, 1),
302 ldap.filter.escape_filter_chars(v, 1),
303 )
304 for k, v in filter_dict.items()
305 )
306 return '(&' + conds + ')'
309def _is_member_of(group_dict, user_dn):
310 """Check if a DN is listed as a member of a group entry."""
311 for key in 'member', 'members', 'uniqueMember':
312 if key in group_dict and user_dn in group_dict[key]:
313 return True
316# https://support.microsoft.com/en-us/help/305144
318_MS_SCRIPT = 0x0001
319_MS_ACCOUNTDISABLE = 0x0002
320_MS_HOMEDIR_REQUIRED = 0x0008
321_MS_LOCKOUT = 0x0010
322_MS_PASSWD_NOTREQD = 0x0020
323_MS_PASSWD_CANT_CHANGE = 0x0040
324_MS_ENCRYPTED_TEXT_PWD_ALLOWED = 0x0080
325_MS_TEMP_DUPLICATE_ACCOUNT = 0x0100
326_MS_NORMAL_ACCOUNT = 0x0200
327_MS_INTERDOMAIN_TRUST_ACCOUNT = 0x0800
328_MS_WORKSTATION_TRUST_ACCOUNT = 0x1000
329_MS_SERVER_TRUST_ACCOUNT = 0x2000
330_MS_DONT_EXPIRE_PASSWORD = 0x10000
331_MS_MNS_LOGON_ACCOUNT = 0x20000
332_MS_SMARTCARD_REQUIRED = 0x40000
333_MS_TRUSTED_FOR_DELEGATION = 0x80000
334_MS_NOT_DELEGATED = 0x100000
335_MS_USE_DES_KEY_ONLY = 0x200000
336_MS_DONT_REQ_PREAUTH = 0x400000
337_MS_PASSWORD_EXPIRED = 0x800000
338_MS_TRUSTED_TO_AUTH_FOR_DELEGATION = 0x1000000
339_MS_PARTIAL_SECRETS_ACCOUNT = 0x04000000