Dex Authentication Failure for Pipelines SDK on Server - "Login credentials were probably invalid"

[Saad17]

Kubeflow Version: 1.7
Environment: AWS EKS
Dex Version: v1

Issue:
I am encountering an issue where authentication using the Pipelines SDK from outside the Kubernetes cluster fails on a specific server, while it works correctly on my local machine.

Error Message:
Login credentials were probably invalid - No redirect after POST to: https://{host}/dex/auth/local/login?back=&state=jorhb7dhzlt3mwebygf4xl57i

This error occurs despite using valid credentials, which have been verified against the Dex configuration file.

Steps to Reproduce
Deploy Kubeflow v1.7 on AWS EKS.
Attempt to authenticate using the Pipelines SDK from a server external to the cluster.
Observe the Dex pod being killed and restarted during the authentication attempt.
Logs and Investigations
Checked the logs from the Dex pod, but no relevant logs appeared.
Fetched events from the "auth" namespace where Dex is running, but no significant events related to the issue were found.
Dex pod gets killed and restarted when a request is made to fetch the session-cookie for authentication.
Istio Ingress Gateway logs
"GET / HTTP/1.1" 302 UAEX ext_authz_denied - "-" 0 350 4 4 "125.0.16.229" "ELB-HealthChecker/2.0" "ee373b34-f6ab-9686-a421-4863aebd05de" "125.0.26.132:8080" "-" outbound|80||centraldashboard.kubeflow.svc.cluster.local - 125.0.26.132:8080 125.0.16.229:2192 - -

Observations
The issue is isolated to a specific server. On my local machine, the authentication works successfully.
The problem seems to be triggered during the authorization phase when trying to fetch the session-cookie, causing the Dex pod to restart.
Expected Behavior
Authentication should succeed using valid credentials without causing the Dex pod to crash or restart.

Relevant Code Snippet
This is the code used to authenticate the Pipelines SDK:

def get_istio_auth_session(url: str, username: str, password: str) -> dict:
"""
Determine if the specified URL is secured by Dex and try to obtain a session cookie.
WARNING: only Dex staticPasswords and LDAP authentication are currently supported
(we default default to using staticPasswords if both are enabled)

:param url: Kubeflow server URL, including protocol
:param username: Dex `staticPasswords` or `LDAP` username
:param password: Dex `staticPasswords` or `LDAP` password
:return: auth session information
"""
# define the default return object
auth_session = {
    "endpoint_url": url,  # KF endpoint URL
    "redirect_url": None,  # KF redirect URL, if applicable
    "dex_login_url": None,  # Dex login URL (for POST of credentials)
    "is_secured": None,  # True if KF endpoint is secured
    "session_cookie": None  # Resulting session cookies in the form "key1=value1; key2=value2"
}

# use a persistent session (for cookies)
with requests.Session() as s:

    ################
    # Determine if Endpoint is Secured
    ################
    logger.info(f"email: {username},,      password: {password},,")
    logger.info(f"url: {url}")
    resp = s.get(url, allow_redirects=True)
    logger.info(f"initial resp: {resp.text} ")
    
    if resp.status_code != 200:
        raise RuntimeError(
            f"HTTP status code '{resp.status_code}' for GET against: {url}"
        )
    
    logger.info(f"initial request headers: {resp.headers}")
    logger.info(f"---------------------------------------------------")
    logger.info(f"initial request response: {resp.text}")
    auth_session["redirect_url"] = resp.url

    # if we were NOT redirected, then the endpoint is UNSECURED
    if len(resp.history) == 0:
        auth_session["is_secured"] = False
        return auth_session
    else:
        auth_session["is_secured"] = True

    ################
    # Get Dex Login URL
    ################
    redirect_url_obj = urlsplit(auth_session["redirect_url"])

    # if we are at `/auth?=xxxx` path, we need to select an auth type
    if re.search(r"/auth$", redirect_url_obj.path):
        #######
        # TIP: choose the default auth type by including ONE of the following
        #######

        # OPTION 1: set "staticPasswords" as default auth type
        redirect_url_obj = redirect_url_obj._replace(
            path=re.sub(r"/auth$", "/auth/local", redirect_url_obj.path)
        )
        # OPTION 2: set "ldap" as default auth type 
        # redirect_url_obj = redirect_url_obj._replace(
        #     path=re.sub(r"/auth$", "/auth/ldap", redirect_url_obj.path)
        # )

    # if we are at `/auth/xxxx/login` path, then no further action is needed (we can use it for login POST)
    if re.search(r"/auth/.*/login$", redirect_url_obj.path):
        auth_session["dex_login_url"] = redirect_url_obj.geturl()

    # else, we need to be redirected to the actual login page
    else:
        # this GET should redirect us to the `/auth/xxxx/login` path
        resp = s.get(redirect_url_obj.geturl(), allow_redirects=True)
        if resp.status_code != 200:
            raise RuntimeError(
                f"HTTP status code '{resp.status_code}' for GET against: {redirect_url_obj.geturl()}"
            )

        # set the login url
        auth_session["dex_login_url"] = resp.url
    logger.info(f"initial redirect url: {auth_session['redirect_url']}")
    logger.info(f"dex login url: {auth_session['dex_login_url']}")
    ################
    # Attempt Dex Login
    ################
    resp = s.post(
        auth_session["dex_login_url"],
        data={"login": username, "password": password},
        allow_redirects=True
    )
    logger.info(f"dex login response headers: {resp.headers}")
    logger.info(f"Dex login response is: {resp.text}")
    if len(resp.history) == 0:
        raise RuntimeError(
            f"Login credentials were probably invalid - "
            f"No redirect after POST to: {auth_session['dex_login_url']}"
        )

    # store the session cookies in a "key1=value1; key2=value2" string
    auth_session["session_cookie"] = "; ".join([f"{c.name}={c.value}" for c in s.cookies])

return auth_session