mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-30 03:14:22 +00:00
Added LDAP Auth
This commit is contained in:
parent
b537997f4f
commit
5fb18ad275
@ -1,5 +1,6 @@
|
||||
import binascii
|
||||
import weakref
|
||||
import ldap3
|
||||
from typing import Optional
|
||||
from typing import MutableMapping # noqa
|
||||
from typing import Tuple
|
||||
@ -46,11 +47,12 @@ class ProxyAuth:
|
||||
self.nonanonymous = False
|
||||
self.htpasswd = None
|
||||
self.singleuser = None
|
||||
self.ldapserver = None
|
||||
self.authenticated = weakref.WeakKeyDictionary() # type: MutableMapping[connections.ClientConnection, Tuple[str, str]]
|
||||
"""Contains all connections that are permanently authenticated after an HTTP CONNECT"""
|
||||
|
||||
def enabled(self) -> bool:
|
||||
return any([self.nonanonymous, self.htpasswd, self.singleuser])
|
||||
return any([self.nonanonymous, self.htpasswd, self.singleuser, self.ldapserver])
|
||||
|
||||
def is_proxy_auth(self) -> bool:
|
||||
"""
|
||||
@ -99,6 +101,21 @@ class ProxyAuth:
|
||||
elif self.htpasswd:
|
||||
if self.htpasswd.check_password(username, password):
|
||||
return username, password
|
||||
elif self.ldapserver:
|
||||
if not username or not password:
|
||||
return None
|
||||
dn = ctx.options.proxyauth.split(":")[2]
|
||||
parts = dn.split("?")
|
||||
conn = ldap3.Connection(
|
||||
self.ldapserver,
|
||||
parts[0] + username + parts[1],
|
||||
password,
|
||||
auto_bind=True)
|
||||
if conn:
|
||||
conn.search(parts[1][1:], '('+parts[0]+username+')', attributes=['objectclass'])
|
||||
if ctx.options.proxyauth.split(":")[3] in conn.entries[0]['objectclass']:
|
||||
conn.unbind()
|
||||
return username, password
|
||||
|
||||
return None
|
||||
|
||||
@ -129,6 +146,17 @@ class ProxyAuth:
|
||||
raise exceptions.OptionsError(
|
||||
"Could not open htpasswd file: %s" % p
|
||||
)
|
||||
elif ctx.options.proxyauth.startswith("ldap"):
|
||||
parts = ctx.options.proxyauth.split(":")
|
||||
if len(parts) != 4:
|
||||
raise exceptions.OptionsError(
|
||||
"Invalid ldap specification"
|
||||
)
|
||||
if parts[0] == "ldaps":
|
||||
server = ldap3.Server(parts[1], use_ssl=True)
|
||||
elif parts[0] == "ldap":
|
||||
server = ldap3.Server(parts[1])
|
||||
self.ldapserver = server
|
||||
else:
|
||||
parts = ctx.options.proxyauth.split(':')
|
||||
if len(parts) != 2:
|
||||
|
@ -199,8 +199,12 @@ class Options(optmanager.OptManager):
|
||||
"""
|
||||
Require proxy authentication. Value may be "any" to require
|
||||
authenticaiton but accept any credentials, start with "@" to specify
|
||||
a path to an Apache htpasswd file, or be of the form
|
||||
"username:password".
|
||||
a path to an Apache htpasswd file, be of the form
|
||||
"username:password", or be of the form
|
||||
"ldap[s]:url_server_ldap:dn:group", the dn must include "?", which will be
|
||||
the username prompted, and the group is the group the user must belong to
|
||||
an example would be
|
||||
"ldap:ldap.forumsys.com:uid=?,dc=example,dc=com:person".
|
||||
"""
|
||||
)
|
||||
self.add_option(
|
||||
|
@ -1,4 +1,5 @@
|
||||
import binascii
|
||||
import ldap3
|
||||
|
||||
import pytest
|
||||
|
||||
@ -41,6 +42,13 @@ def test_configure():
|
||||
ctx.configure(up, proxyauth=None)
|
||||
assert not up.nonanonymous
|
||||
|
||||
ctx.configure(up, proxyauth="ldap:ldap.forumsys.com:uid=?,dc=example,dc=com:person")
|
||||
assert up.ldapserver
|
||||
ctx.configure(up, proxyauth="ldaps:ldap.forumsys.com:uid=?,dc=example,dc=com:person")
|
||||
assert up.ldapserver
|
||||
with pytest.raises(exceptions.OptionsError):
|
||||
ctx.configure(up, proxyauth="ldapldap.forumsys.com:uid=?dc=example,dc=com:person")
|
||||
|
||||
with pytest.raises(exceptions.OptionsError):
|
||||
ctx.configure(
|
||||
up,
|
||||
@ -109,6 +117,24 @@ def test_check():
|
||||
)
|
||||
assert not up.check(f)
|
||||
|
||||
ctx.configure(
|
||||
up,
|
||||
proxyauth="ldap:ldap.forumsys.com:uid=?,dc=example,dc=com:person"
|
||||
)
|
||||
f.request.headers["Proxy-Authorization"] = proxyauth.mkauth(
|
||||
"einstein", "password"
|
||||
)
|
||||
assert up.check(f)
|
||||
f.request.headers["Proxy-Authorization"] = proxyauth.mkauth(
|
||||
"", ""
|
||||
)
|
||||
assert not up.check(f)
|
||||
with pytest.raises(ldap3.core.exceptions.LDAPBindError):
|
||||
f.request.headers["Proxy-Authorization"] = proxyauth.mkauth(
|
||||
"einstein", "foo"
|
||||
)
|
||||
assert not up.check(f)
|
||||
|
||||
|
||||
def test_authenticate():
|
||||
up = proxyauth.ProxyAuth()
|
||||
|
Loading…
Reference in New Issue
Block a user