diff options
| -rwxr-xr-x | session_app.py.publish | 12 | ||||
| -rw-r--r-- | session_ldap.py | 60 |
2 files changed, 62 insertions, 10 deletions
diff --git a/session_app.py.publish b/session_app.py.publish index fd403c2..b09cb59 100755 --- a/session_app.py.publish +++ b/session_app.py.publish @@ -14,7 +14,7 @@ # Improve: # move all configs to config file # move all references to references section -# accept a bind credential so we can perform lookups of users who match "uid=%s" under a basedn. +# accept a bind credential so we can perform lookups of users who match "uid=%s" under a base. # Run: # FLASK_APP=session_app.py FLASK_DEBUG=1 flask run --host 0.0.0.0 # Dependencies: @@ -38,9 +38,13 @@ secret_key_value = os.urandom(24) secret_key_value_hex_encoded = binascii.hexlify(secret_key_value) app.config['SECRET_KEY'] = secret_key_value_hex_encoded app.config['LDAP_URI'] = "ldaps://ipa.internal.com:636" -app.config['LDAP_USER_BASEDN'] = "cn=users,cn=accounts,dc=ipa,dc=internal,dc=com" -app.config['LDAP_GROUP_BASEDN'] = "cn=groups,cn=accounts,dc=ipa,dc=internal,dc=com" -app.config['LDAP_USER_FORMAT'] = "uid=%s,cn=users,cn=accounts,dc=ipa,dc=internal,dc=com" +app.config['LDAP_USER_BASE'] = "cn=users,cn=accounts,dc=ipa,dc=internal,dc=com" +app.config['LDAP_GROUP_BASE'] = "cn=groups,cn=accounts,dc=ipa,dc=internal,dc=com" +app.config['LDAP_USER_MATCH_ATTRIB'] = "uid" +app.config['LDAP_USER_ATTRIB_MEMBEROF'] = "memberof" +app.config['LDAP_GROUP_NAME_ATTRIB'] = "cn" +app.config['LDAP_BIND_DN'] = "uid=domainjoin,cn=users,cn=accounts,dc=ipa,dc=internal,dc=com" +app.config['LDAP_BIND_PASSWORD'] = "bulkpassword" app.config['minutes'] = 2 app.permanent_session_lifetime=datetime.timedelta(minutes=app.config['minutes']) diff --git a/session_ldap.py b/session_ldap.py index d12f008..423f322 100644 --- a/session_ldap.py +++ b/session_ldap.py @@ -8,15 +8,27 @@ import ldap3 from ldap3.core.exceptions import LDAPBindError, LDAPPasswordIsMandatoryError -def authenticated_user(server_uri, user_format, username, password): - user = user_format.replace("%s",username) +def list_matching_users(server_uri, bind_dn, bind_pw,user_base, username, user_match_attrib): + search_filter=f"({user_match_attrib}={username})" + server = ldap3.Server(server_uri) + conn = ldap3.Connection(server, auto_bind=True,user=bind_dn, password=bind_pw) + conn.search( + search_base=user_base, + search_filter=search_filter) + print(f"DEBUG: search_base {user_base}") + print(f"DEBUG: search_filter {search_filter}") + result = [] + for i in conn.entries: + result.append(i.entry_dn) + print(f"DEBUG: result {result}") + return result + +def authenticated_user(server_uri, user_dn, password): print(f"server_uri: {server_uri}") - print(f"username: {username}") - print(f"user_format: {user_format}") - print(f"user: {user}") + print(f"user_dn: {user_dn}") try: server = ldap3.Server(server_uri) - conn = ldap3.Connection(server, auto_bind=True, user=user, password=password) + conn = ldap3.Connection(server, auto_bind=True, user=user_dn, password=password) return conn except LDAPBindError as e: if 'invalidCredentials' in str(e): @@ -48,3 +60,39 @@ def list_ldap_servers_for_domain(domain): if not len(namelist): namelist.append(domain) return namelist + +def get_ldap_user_groups(server_uri, bind_dn, bind_pw,user_dn,user_attrib_memberof,group_name_attrib,group_base): + server = ldap3.Server(server_uri) + conn = ldap3.Connection(server, auto_bind=True,user=bind_dn, password=bind_pw) + conn.search( + search_base=user_dn, + search_filter="(cn=*)", # this has the potential to not work in a directory where CN is not a part of any dn? + search_scope="BASE", + attributes=[user_attrib_memberof] + ) + these_groups = conn.entries[0].entry_attributes_as_dict[user_attrib_memberof] + #print(f"DEBUG: these_groups={these_groups}") + result = [] + for group in these_groups: + #print(f"DEBUG: will check for value {group_base} in {group}") + if group_base in group: + if group_name_attrib == "dn": + #print(f"DEBUG: just add group via dn {group}") + result.append(group) + else: + # we need to lookup this group and pick the attribute of it the admin wants. + #print(f"DEBUG: need to lookup group {group} and extract attrib {group_name_attrib}") + conn.search( + search_base=group, + search_filter="(objectClass=*)", + search_scope="BASE", + attributes=[group_name_attrib] + ) + this_group=conn.entries[0].entry_attributes_as_dict[group_name_attrib][0] + #print(f"DEBUG: Group {group} identified as attrib {group_name_attrib}={this_group}") + result.append(this_group) + return result + +def get_ldap_dn_from_krbPrincipalName(server_uri, bind_dn, bind_pw,user_krbPrincipalName): + # goal: return as string the dn + print("stub") |