Cognito - Admin инициирует проверку подлинности с ошибкой NotAuthorizedException

В настоящее время мы разрабатываем приложение и соответствующий серверный модуль с тремя вариантами входа в систему / регистрации, Google, Apple и с помощью электронной почты / пароля.

Когда пользователь выбирает Apple или Google и его еще нет в Cognito, мы создаем пользователя с admin_create_user из boto3.client.

Пока все хорошо, пользователь создан.

Когда дело доходит до обработки логина, мы сталкиваемся с проблемами с CUSTOM_AUTH для admin_initiate_auth:

    try:
        initiate_resp = cognito_client.admin_initiate_auth(
            UserPoolId=settings.AWS_COGNITO_USER_POOL_ID,
            ClientId=settings.AWS_COGNITO_CLIENT_ID,
            AuthFlow="CUSTOM_AUTH",
            AuthParameters={
                "USERNAME": email,
                "SECRET_HASH": get_secret_hash(email)
            }
        )
        print(initiate_resp)
    except ClientError as e:
        return {"error": f"admin_initiate_auth failed: {str(e)}"}, 400

Мы добавили, согласно документации, лямбда-триггеры для

• определить_авторизованный вызов:

def lambda_handler(event, context):
    """
    Cognito calls this to decide which challenge to present next (or issue tokens).
    We'll always present a CUSTOM_CHALLENGE if there's no successful challenge yet.
    """

    # event["request"]["session"] contains prior challenge attempts.
    # If empty, it's the user's first attempt this auth session.
    session_list = event["request"].get("session", [])

    if len(session_list) == 0:
        # First time => present the custom challenge
        event["response"]["challengeName"] = "CUSTOM_CHALLENGE"
        event["response"]["issueTokens"] = False
        event["response"]["failAuthentication"] = False
        print("DefineAuthChallenge -> Presenting CUSTOM_CHALLENGE (first attempt)")

    else:
        # Check the last challenge
        last_challenge = session_list[-1]
        if last_challenge.get("challengeResult") is True:
            # If they answered correctly, issue tokens
            event["response"]["challengeName"] = "CUSTOM_CHALLENGE"
            event["response"]["issueTokens"] = True
            event["response"]["failAuthentication"] = False
            print("DefineAuthChallenge -> Last challenge success, issuing tokens")

        else:
            # If they failed or haven't answered, prompt again
            event["response"]["challengeName"] = "CUSTOM_CHALLENGE"
            event["response"]["issueTokens"] = False
            event["response"]["failAuthentication"] = False
            print("DefineAuthChallenge -> Re-presenting CUSTOM_CHALLENGE")

    return event

• создать_авторизованный вызов:

def lambda_handler(event, context):
    challenge_name = event["request"]["challengeName"]
    print(challenge_name)
    print(event)
    """
    Invoked after Cognito decides to present a CUSTOM_CHALLENGE,
    giving you a chance to store or define challenge parameters.
    For a Google token check, we usually don't need to store anything.
    """
    if challenge_name != "CUSTOM_CHALLENGE":
        return event
        
    event["response"]["publicChallengeParameters"] = {}
    event["response"]["privateChallengeParameters"] = {}
    event["response"]["challengeMetadata"] = "CUSTOM_CHALLENGE_METADATA"
    print("CreateAuthChallenge -> Setting empty challenge parameters for Google token flow")


    return event

и

• проверить подлинность вызова

import json
import urllib.request

def lambda_handler(event, context):
    challengeName = event["request"]["challengeName"]
    print("Verifying started...")
    """
    Verifies the challenge answer (the Google ID token).
    If it's valid, we set 'answerCorrect' to True. Cognito then issues tokens.
    """

    if challengeName == "CUSTOM_CHALLENGE":
        provided_token = event["request"].get("challengeAnswer")
        if verify_google_id_token(provided_token):
            event["response"]["answerCorrect"] = True
            print("VerifyAuthChallenge -> Google token valid, answerCorrect=True")
        else:
            event["response"]["answerCorrect"] = False
            print("VerifyAuthChallenge -> Google token invalid, answerCorrect=False")


    return event


def verify_google_id_token(id_token: str) -> bool:
    """Verify the Google ID token by calling the tokeninfo endpoint."""
    if not id_token:
        return False

    url = f"https://oauth2.googleapis.com/tokeninfo?id_token={id_token}"

    try:
        with urllib.request.urlopen(url) as resp:
            if resp.status != 200:
                return False
            data = json.loads(resp.read())
    except Exception as e:
        print("Error verifying Google token:", e)
        return False

    if "error" in data:
        return False

    # Optionally check 'aud' == your Google client ID, 
    # 'exp' is in the future, 'email_verified' == "true", etc.
    # Example check:
    # if data.get("aud") != "YOUR_GOOGLE_CLIENT_ID":
    #     return False

    return True

Когда я пытаюсь войти в систему с помощью моего id_token, я получаю следующую ошибку:

[
    {
        "error": "admin_initiate_auth failed: An error occurred (NotAuthorizedException) when calling the AdminInitiateAuth operation: Incorrect username or password."
    },
    400
]

У кого-нибудь есть совет? Был бы очень признателен!

Вернуться на верх