代码之家  ›  专栏  ›  技术社区  ›  Andre

AWS Lambda自定义JWT验证

  •  0
  • Andre  · 技术社区  · 2 年前

    我构建了它,首先验证JWT令牌,然后提取用户唯一ID(“sub”)。

    在非Lambda环境中,脚本运行良好,但在AWS Lambda中,我收到了一条错误消息。

    可能是什么问题?

    Unexpected error during JWT validation: Unable to find an algorithm for key: {'alg': 'RS256', 'e': 'AQAB', 'kid': 'dmAQX7bVDINFkTGxZc5YCxF5ZA/pcaRsQMUoBbRt4bw=', 'kty': 'RSA', 'n': 'u9hHbyMaI-PWsTG9MtaHjxwBmMez6VeV-ScqIgllBUSQkx8Ao...vGUIG39rb3nPmNVCunBw', 'use': 'sig'}

    这是我的AWS Lambda代码:

    import json
    import os
    import requests
    from jose import jwt, jwk
    
    def get_efs_keys(file_name="/mnt/efs/jwks.json"):
        
        # The jkws.json is obtained from here:
        # https://cognito-idp.<Region>.amazonaws.com/<userPoolId>/.well-known/jwks.json
        
        try:
            with open(file_name, 'r') as file:
                jwks_data = json.load(file)
                return jwks_data.get('keys', [])
        except Exception as e:
            print(f"An error occurred while fetching keys: {e}")
            return []
    
    def validate_jwt(jwt_token, keys):
        if not jwt_token:
            return False, False
    
        try:
            headers = jwt.get_unverified_headers(jwt_token)
            kid = headers.get('kid')
            if not kid:
                return False, False
    
            key = next((key for key in keys if key['kid'] == kid), None)
            if key is None:
                return False, False
    
            public_key = jwk.construct(key)
            decoded_token = jwt.decode(jwt_token, public_key, algorithms=['RS256'], audience=os.environ.get('APP_CLIENT_ID'))
            return True, decoded_token.get('sub', False)
        except jwt.JWTError as e:
            print(f"JWT token validation error: {e}")
            return False, False
        except Exception as e:
            print(f"Unexpected error during JWT validation: {e}")
            return False, False
    
    def lambda_handler(event, context):
        # Get all headers from the event
        headers = event.get('headers', {})
    
        # Get the Authorization header
        authorization_header = headers.get('Authorization', '')
    
        # Parse the Bearer token to get only the access token (case-insensitive)
        if authorization_header.lower().startswith('bearer '):
            access_token = authorization_header[7:]
        else:
            access_token = None
    
        # Get keys from EFS
        keys = get_efs_keys()
    
        # Validate the JWT token
        jwt_valid, sub = validate_jwt(access_token, keys)
    
        # Create a response
        response_body = {
            'access_token': access_token,
            'jwt_valid': jwt_valid,
            'sub': sub
        }
    
        response = {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json'
            },
            'body': json.dumps(response_body)
        }
    
        return response
    

    如果验证成功,“jwt_valid”必须为“True”,“sub”必须分别为唯一值。

    1 回复  |  直到 2 年前
        1
  •  1
  •   TLeitzbach    2 年前

    This 线程提示您缺少依赖项 cryptography . 例如,您需要将其安装在Lambda层中,因为它提供了必要的算法。

    pip install cryptography