Coverage for gws-app/gws/plugin/auth_provider/ldap/__init__.py: 93%

1"""LDAP authentication provider. 

2 

3Authenticates users against an LDAP or Active Directory server. The server is 

4given as an LDAP URL, a subset of the RFC 2255 form:: 

5 

6 ldap://host:port/baseDN?searchAttribute 

7 

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. 

13 

14Authentication with ``username`` and ``password`` credentials works as 

15follows: 

16 

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. 

24 

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. 

31 

32Example:: 

33 

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 } 

45 

46References: 

47 https://datatracker.ietf.org/doc/html/rfc2255 

48""" 

49 

50from typing import Optional 

51 

52import contextlib 

53 

54import ldap 

55import ldap.filter 

56 

57import gws 

58import gws.base.auth 

59import gws.lib.net 

60 

61 

62class UserRule(gws.Data): 

63 """Rule that assigns GWS roles to LDAP accounts.""" 

64 

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.""" 

71 

72 

73class SSLConfig(gws.Config): 

74 """SSL settings for the LDAP connection.""" 

75 

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.""" 

82 

83 

84@gws.ext.config.authProvider('ldap') 

85class Config(gws.base.auth.provider.Config): 

86 """Authentication against an LDAP or Active Directory server.""" 

87 

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[UserRule] 

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.""" 

104 

105 

106@gws.ext.object.authProvider('ldap') 

107class Object(gws.base.auth.provider.Object): 

108 """LDAP authentication provider.""" 

109 

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 userRules: list[UserRule] 

129 """Rules that assign roles to accounts.""" 

130 

131 def configure(self): 

132 self.timeout = self.cfg('timeout', default=30) 

133 

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.userRules = self.cfg('users', default=[]) 

140 

141 proto = 'ldaps' if self.ssl else 'ldap' 

142 p = gws.lib.net.parse_url(self.cfg('url')) 

143 

144 self.serverUrl = proto + '://' + p.netloc 

145 self.baseDN = p.path.strip('/') 

146 self.loginAttribute = p.query 

147 

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 

153 

154 def authenticate(self, method, credentials): 

155 username = credentials.get('username', '').strip() 

156 password = credentials.get('password', '') 

157 if not username or not password: 

158 return 

159 

160 with self._connection() as conn: 

161 rec = self._get_user_record(conn, username, password) 

162 if not rec: 

163 return 

164 

165 # NB need to rebind as admin 

166 with self._connection() as conn: 

167 return self._make_user(conn, rec) 

168 

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]) 

174 

175 ## 

176 

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})) 

180 

181 if len(users) == 0: 

182 return 

183 if len(users) > 1: 

184 raise gws.AuthenticationError(f'multiple entries for {username!r}') 

185 

186 rec = users[0] 

187 

188 # check for AD disabled accounts 

189 uac = str(rec.get('userAccountControl', '')) 

190 if uac and uac.isdigit() and (int(uac) & _MS_ACCOUNTDISABLE): 

191 raise gws.ForbiddenError('ACCOUNTDISABLE flag set') 

192 

193 try: 

194 conn.simple_bind_s(rec['dn'], password) 

195 return rec 

196 except ldap.INVALID_CREDENTIALS: 

197 raise gws.AuthenticationError(f'wrong password for {username!r}') 

198 except ldap.LDAPError as exc: 

199 gws.log.exception() 

200 raise gws.ForbiddenError(f'LDAP error {exc!r}') from exc 

201 

202 def _make_user(self, conn, rec): 

203 """Create a user from an LDAP entry.""" 

204 user_rec = dict(rec) 

205 user_rec['roles'] = self._roles_for_user(conn, rec) 

206 

207 if not user_rec.get('displayName') and self.displayNameFormat: 

208 user_rec['displayName'] = gws.u.format_map(self.displayNameFormat, rec) 

209 

210 login = '' 

211 for k in list(user_rec): 

212 if k.lower() == self.loginAttribute.lower(): 

213 login = user_rec.pop(k) 

214 break 

215 user_rec['localUid'] = user_rec['loginName'] = login 

216 

217 return gws.base.auth.user.from_record(self, user_rec) 

218 

219 def _roles_for_user(self, conn, rec): 

220 """Return the roles of all ``users`` rules that match the entry.""" 

221 user_dn = rec['dn'] 

222 roles = set() 

223 

224 for u in self.userRules: 

225 if u.get('matches') and self._matches(conn, user_dn, u.matches): 

226 roles.update(u.roles) 

227 

228 if u.get('memberOf') and self._find(conn, _member_filter(u.memberOf, user_dn)): 

229 roles.update(u.roles) 

230 

231 return sorted(roles) 

232 

233 def _matches(self, conn, dn, flt): 

234 try: 

235 res = conn.search_s(dn, ldap.SCOPE_BASE, flt) 

236 return bool(res) 

237 except ldap.NO_SUCH_OBJECT: 

238 return False 

239 

240 def _find(self, conn, flt): 

241 """Search below the base DN and return the entries as dicts.""" 

242 try: 

243 res = conn.search_s(self.baseDN, ldap.SCOPE_SUBTREE, flt) 

244 except ldap.NO_SUCH_OBJECT: 

245 return [] 

246 

247 dcts = [] 

248 

249 for dn, data in res: 

250 if dn: 

251 d = _as_dict(data) 

252 d['dn'] = dn 

253 dcts.append(d) 

254 

255 return dcts 

256 

257 @contextlib.contextmanager 

258 def _connection(self): 

259 """Open a connection, bound as the bind DN if configured.""" 

260 conn = ldap.initialize(self.serverUrl) 

261 conn.set_option(ldap.OPT_NETWORK_TIMEOUT, self.timeout) 

262 

263 if self.ssl: 

264 if self.root.app.developer_option('ldap.ssl_insecure'): 

265 conn.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER) 

266 else: 

267 conn.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_DEMAND) 

268 if self.ssl.ca: 

269 conn.set_option(ldap.OPT_X_TLS_CACERTFILE, self.ssl.ca) 

270 if self.ssl.crt and self.ssl.key: 

271 conn.set_option(ldap.OPT_X_TLS_CERTFILE, self.ssl.crt) 

272 conn.set_option(ldap.OPT_X_TLS_KEYFILE, self.ssl.key) 

273 conn.set_option(ldap.OPT_X_TLS_NEWCTX, 0) 

274 

275 if self.activeDirectory: 

276 # see https://www.python-ldap.org/faq.html#usage 

277 conn.set_option(ldap.OPT_REFERRALS, 0) 

278 

279 try: 

280 if self.bindDN: 

281 conn.simple_bind_s(self.bindDN, self.bindPassword) 

282 yield conn 

283 finally: 

284 conn.unbind_s() 

285 

286 

287def _as_dict(data): 

288 """Convert attribute values to strings, unwrapping single values.""" 

289 d = {} 

290 

291 for k, v in data.items(): 

292 if not v: 

293 continue 

294 if not isinstance(v, list): 

295 v = [v] 

296 v = [gws.u.to_str(s) for s in v] 

297 d[k] = v[0] if len(v) == 1 else v 

298 

299 return d 

300 

301 

302def _make_filter(filter_dict): 

303 """Create an AND filter that matches the given attribute values.""" 

304 conds = [] 

305 for k, v in filter_dict.items(): 

306 conds.append(f'({k}={_escape(v)})') 

307 return '(&' + ''.join(conds) + ')' 

308 

309 

310def _member_filter(group_filter, user_dn): 

311 """Build a filter for groups that match ``group_filter`` and list ``user_dn`` as a member. 

312 

313 The server evaluates the membership, so this also works for Active Directory groups 

314 whose member list exceeds ``MaxValRange`` and is returned in ranges. 

315 """ 

316 if not group_filter.startswith('('): 

317 group_filter = f'({group_filter})' 

318 dn = _escape(user_dn) 

319 return f'(&{group_filter}(|(member={dn})(members={dn})(uniqueMember={dn})))' 

320 

321 

322def _escape(s): 

323 return ldap.filter.escape_filter_chars(s, 1) 

324 

325 

326# https://support.microsoft.com/en-us/help/305144 

327 

328_MS_SCRIPT = 0x0001 

329_MS_ACCOUNTDISABLE = 0x0002 

330_MS_HOMEDIR_REQUIRED = 0x0008 

331_MS_LOCKOUT = 0x0010 

332_MS_PASSWD_NOTREQD = 0x0020 

333_MS_PASSWD_CANT_CHANGE = 0x0040 

334_MS_ENCRYPTED_TEXT_PWD_ALLOWED = 0x0080 

335_MS_TEMP_DUPLICATE_ACCOUNT = 0x0100 

336_MS_NORMAL_ACCOUNT = 0x0200 

337_MS_INTERDOMAIN_TRUST_ACCOUNT = 0x0800 

338_MS_WORKSTATION_TRUST_ACCOUNT = 0x1000 

339_MS_SERVER_TRUST_ACCOUNT = 0x2000 

340_MS_DONT_EXPIRE_PASSWORD = 0x10000 

341_MS_MNS_LOGON_ACCOUNT = 0x20000 

342_MS_SMARTCARD_REQUIRED = 0x40000 

343_MS_TRUSTED_FOR_DELEGATION = 0x80000 

344_MS_NOT_DELEGATED = 0x100000 

345_MS_USE_DES_KEY_ONLY = 0x200000 

346_MS_DONT_REQ_PREAUTH = 0x400000 

347_MS_PASSWORD_EXPIRED = 0x800000 

348_MS_TRUSTED_TO_AUTH_FOR_DELEGATION = 0x1000000 

349_MS_PARTIAL_SECRETS_ACCOUNT = 0x04000000