diff options
| -rwxr-xr-x | session_app.py.publish | 58 | ||||
| -rw-r--r-- | session_ldap.py | 44 |
2 files changed, 88 insertions, 14 deletions
diff --git a/session_app.py.publish b/session_app.py.publish index b09cb59..ac37e17 100755 --- a/session_app.py.publish +++ b/session_app.py.publish @@ -14,7 +14,6 @@ # 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 base. # Run: # FLASK_APP=session_app.py FLASK_DEBUG=1 flask run --host 0.0.0.0 # Dependencies: @@ -41,10 +40,12 @@ app.config['LDAP_URI'] = "ldaps://ipa.internal.com:636" 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_DISPLAY_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['LDAP_USER_KERBEROS_PRINCIPAL_ATTRIB'] = "krbPrincipalName" app.config['minutes'] = 2 app.permanent_session_lifetime=datetime.timedelta(minutes=app.config['minutes']) @@ -93,7 +94,19 @@ def requires_authn_kerberos(function): token = ''.join(header.split()[1:]) rc = _gssapi_authenticate(token) if rc == kerberos.AUTH_GSS_COMPLETE: - response = function(ctx.kerberos_user, *args, **kwargs) + # if ldap config options are set, then do kerberos -> short username resolution + user = ctx.kerberos_user + if 'LDAP_USER_KERBEROS_PRINCIPAL_ATTRIB' in app.config: + user = session_ldap.get_ldap_attrib_from_krbPrincipalName( + server_uri=get_next_ldap_server(app), + bind_dn=app.config['LDAP_BIND_DN'], + bind_pw=app.config['LDAP_BIND_PASSWORD'], + search_base=app.config['LDAP_USER_BASE'], + user_attrib=app.config['LDAP_USER_DISPLAY_ATTRIB'], + user_krbPrincipalName=user, + krbPrincipalName_attrib=app.config['LDAP_USER_KERBEROS_PRINCIPAL_ATTRIB'] + ) + response = function(user, *args, **kwargs) response = make_response(response) if ctx.kerberos_token is not None: response.headers['WWW-Authenticate'] = ' '.join(['negotiate', ctx.kerberos_token]) @@ -131,9 +144,29 @@ def requires_authn_ldap(function): username = request.form['username'] pw = request.form['password'] #print(f"DEBUG: requires_authn_ldap with username={username} and pw={pw}") - ll = ldap_login(username,pw) + # learn dn of user + this_uri = get_next_ldap_server(app) + this_user = session_ldap.list_matching_users( + server_uri=this_uri, + bind_dn=app.config['LDAP_BIND_DN'], + bind_pw=app.config['LDAP_BIND_PASSWORD'], + user_base=app.config['LDAP_USER_BASE'], + username=username, + user_match_attrib=app.config['LDAP_USER_MATCH_ATTRIB'] + ) + # list_matching_users always returns list, so if it contains <> 1 we are in trouble + if len(this_user) != 1: + print(f"WARNING: cannot determine unique user for {app.config['LDAP_USER_MATCH_ATTRIB']}={username} which returned {tihs_user}") + return _unauthorized_ldap() + this_user = this_user[0] + print(f"DEBUG: requires_authn_ldap: found in ldap the username {this_user}") + ll = ldap_login(this_user,pw) if ll: - return function(ll.user,*args, **kwargs) + shortuser = session_ldap.get_ldap_username_attrib_from_dn( + authenticated_user=ll, + user_match_attrib=app.config['LDAP_USER_DISPLAY_ATTRIB'] + ) + return function(shortuser,*args, **kwargs) else: return _unauthorized_ldap() return decorated @@ -185,9 +218,8 @@ def login(user="None"): if request.authorization: return redirect(url_for("login_basic"),code=307) return handle_login_ldap_from_non_ldap(request) - -def ldap_login(username,password): - #print(f"DEBUG: Trying user {username} with pw '{password}'") + +def get_next_ldap_server(app): # on first ldap_login attempt, cache this lookup result: if 'LDAP_HOSTS' not in app.config: this_domain = urlparse(app.config['LDAP_URI']).hostname @@ -205,12 +237,16 @@ def ldap_login(username,password): if up.port: this_netloc += f":{up.port}" this_uri = up._replace(netloc=this_netloc).geturl() + return this_uri + +def ldap_login(username,password): + #print(f"DEBUG: Trying user {username} with pw '{password}'") + this_uri = get_next_ldap_server(app) # Perform the ldap interactions user = session_ldap.authenticated_user( - this_uri, - app.config['LDAP_USER_FORMAT'], - username, - password + server_uri=this_uri, + user_dn=username, + password=password ) if user: return user diff --git a/session_ldap.py b/session_ldap.py index 423f322..1b2dc12 100644 --- a/session_ldap.py +++ b/session_ldap.py @@ -93,6 +93,44 @@ def get_ldap_user_groups(server_uri, bind_dn, bind_pw,user_dn,user_attrib_member 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") +def get_ldap_attrib_from_krbPrincipalName(server_uri = None, bind_dn = "", bind_pw = "", connection = None, search_base = "", user_attrib = "uid", user_krbPrincipalName = "", krbPrincipalName_attrib = "krbPrincipalName"): + if connection and isinstance(connection, ldap3.core.connection.Connection): + conn = connection + else: + server = ldap3.Server(server_uri) + conn = ldap3.Connection(server, auto_bind=True,user=bind_dn, password=bind_pw) + conn.search( + search_base=search_base, + search_scope="SUBTREE", + search_filter=f"({krbPrincipalName_attrib}={user_krbPrincipalName})", + attributes=[user_attrib] + ) + entry = conn.entries[0] + if user_attrib == "dn": + return entry.entry_dn + else: + return entry.entry_attributes_as_dict[entry.entry_attributes[0]][0] + +def get_ldap_username_attrib_from_dn(server_uri = None, bind_dn = "", bind_pw = "", authenticated_user = None, user_match_attrib = "dn", user_dn = None): + # Needs (server_uri, bind_dn, bind_pw, user_dn) or (authenticated_user) + if authenticated_user and isinstance(authenticated_user, ldap3.core.connection.Connection): + conn = authenticated_user + search_base=authenticated_user.extend.standard.who_am_i().replace("dn: ","") + else: + # then we have to use a new connection + server = ldap3.Server(server_uri) + conn = ldap3.Connection(server, auto_bind=True,user=bind_dn, password=bind_pw) + search_base=user_dn, + # so now conn is the connection regardless of how we got there, and search_base + #print(f"DEBUG: search_base {search_base} attributes {user_match_attrib}") + conn.search( + search_base=search_base, + search_scope="BASE", + search_filter="(cn=*)", + attributes=[user_match_attrib] + ) + entry = conn.entries[0] + if user_match_attrib == "dn": + return entry.entry_dn + else: + return entry.entry_attributes_as_dict[entry.entry_attributes[0]][0] |