Asegurar API Gateway con Lambda Authorizers Personalizados

Descripcion general

Los Lambda authorizers de API Gateway interceptan peticiones antes de que lleguen a tu handler, validan tokens y retornan politicas IAM que permiten o niegan acceso. Esto separa la autenticacion de la logica de negocio y funciona con cualquier formato de token (JWT, OAuth, custom). A continuacion: construir un authorizer JWT, generar politicas IAM, pasar contexto a handlers, configurar caching y manejar tipos de authorizer request/response.

Cuando Usar Esto

• For alternatives, see Complete Guide to Authentication Patterns.

• APIs serverless que necesitan autenticacion personalizada (JWT, OAuth, API keys con lookup en base de datos)

• Control de acceso fino por ruta o metodo

• Integracion con proveedores de identidad de terceros (Auth0, Cognito, Okta)

• Cualquier arquitectura API Gateway + Lambda que necesita validacion de tokens

Prerrequisitos

• Python 3.11+
• API Gateway (REST o HTTP API)
• Paquete PyJWT para validacion JWT

Solucion

1. JWT Lambda Authorizer (Tipo Request)

import json
import jwt
import time
import os

JWT_SECRET = os.environ.get('JWT_SECRET', 'your-secret-key')
JWT_ALGORITHM = 'HS256'

def lambda_handler(event, context):
    try:
        token = extract_token(event)
        claims = validate_jwt(token)

        principal_id = claims['sub']
        policy = generate_policy(
            principal_id=principal_id,
            effect='Allow',
            resource=event['routeArn'],
            context={
                'userId': principal_id,
                'email': claims.get('email', ''),
                'role': claims.get('role', 'user'),
            },
        )
        return policy

    except Exception as e:
        print(f"Authorization failed: {e}")
        return generate_policy(
            principal_id='unauthorized',
            effect='Deny',
            resource=event['routeArn'],
            context={},
        )

def extract_token(event: dict) -> str:
    headers = event.get('headers', {})
    auth_header = headers.get('Authorization') or headers.get('authorization', '')

    if not auth_header.startswith('Bearer '):
        raise ValueError("Missing or invalid Authorization header")

    return auth_header[7:]

def validate_jwt(token: str) -> dict:
    try:
        claims = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
        if claims.get('exp', 0) < time.time():
            raise ValueError("Token expired")
        return claims
    except jwt.InvalidTokenError as e:
        raise ValueError(f"Invalid token: {e}")

def generate_policy(principal_id: str, effect: str, resource: str, context: dict) -> dict:
    return {
        'principalId': principal_id,
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [
                {
                    'Action': 'execute-api:Invoke',
                    'Effect': effect,
                    'Resource': resource,
                }
            ],
        },
        'context': context,
    }

2. Template SAM con Authorizer

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Resources:
  AuthFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: authorizer.lambda_handler
      Runtime: python3.11
      CodeUri: src/
      Environment:
        Variables:
          JWT_SECRET: '{{resolve:secretsmanager:jwt-secret:SecretString}}'

  ApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: handler.lambda_handler
      Runtime: python3.11
      CodeUri: src/

  MyApi:
    Type: AWS::Serverless::Api
    Properties:
      StageName: prod
      Auth:
        DefaultAuthorizer: JwtAuthorizer
        Authorizers:
          JwtAuthorizer:
            FunctionArn: !GetAtt AuthFunction.Arn
            Identity:
              Header: Authorization
              ValidationExpression: ^Bearer [a-zA-Z0-9._-]+$
            AuthorizerResultTtlInSeconds: 300

Resources:
  ProtectedResource:
    Type: AWS::Serverless::Function
    Properties:
      Handler: handler.lambda_handler
      Runtime: python3.11
      CodeUri: src/
      Events:
        GetUsers:
          Type: Api
          Properties:
            RestApiId: !Ref MyApi
            Path: /users
            Method: GET
            Auth:
              Authorizer: JwtAuthorizer

3. Acceder al Contexto del Authorizer en el Handler

import json

def lambda_handler(event, context):
    # El contexto del authorizer esta disponible en el event
    auth_context = event.get('requestContext', {}).get('authorizer', {})

    user_id = auth_context.get('userId', 'unknown')
    email = auth_context.get('email', 'unknown')
    role = auth_context.get('role', 'user')

    if role != 'admin':
        return {
            'statusCode': 403,
            'body': json.dumps({'error': 'Insufficient permissions'}),
        }

    return {
        'statusCode': 200,
        'body': json.dumps({
            'message': f'Hello {email}',
            'userId': user_id,
        }),
    }

4. Control de Acceso Basado en Roles

import json
import jwt
import os

JWT_SECRET = os.environ['JWT_SECRET']

def lambda_handler(event, context):
    try:
        token = extract_token(event)
        claims = jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
        role = claims.get('role', 'user')
        method = event['httpMethod']
        resource = event['routeArn']

        # Definir permisos basados en rol
        permissions = {
            'admin': ['GET', 'POST', 'PUT', 'DELETE'],
            'editor': ['GET', 'POST', 'PUT'],
            'user': ['GET'],
        }

        allowed_methods = permissions.get(role, [])

        if method in allowed_methods:
            return generate_policy(claims['sub'], 'Allow', resource, {
                'userId': claims['sub'],
                'role': role,
            })
        else:
            return generate_policy(claims['sub'], 'Deny', resource, {
                'userId': claims['sub'],
                'role': role,
            })

    except Exception as e:
        print(f"Auth error: {e}")
        return generate_policy('error', 'Deny', event['routeArn'], {})

def generate_policy(principal_id, effect, resource, context):
    return {
        'principalId': principal_id,
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [{
                'Action': 'execute-api:Invoke',
                'Effect': effect,
                'Resource': resource,
            }],
        },
        'context': context,
    }

def extract_token(event):
    headers = event.get('headers', {})
    auth_header = headers.get('Authorization') or headers.get('authorization', '')
    if not auth_header.startswith('Bearer '):
        raise ValueError("Missing Authorization header")
    return auth_header[7:]

5. Authorizer con Lookup en Base de Datos (API Key)

import json
import boto3
import hashlib

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['API_KEYS_TABLE'])

def lambda_handler(event, context):
    api_key = event.get('headers', {}).get('x-api-key', '')

    if not api_key:
        return generate_deny(event['routeArn'], 'Missing API key')

    # Hash del API key para lookup
    key_hash = hashlib.sha256(api_key.encode()).hexdigest()

    response = table.get_item(Key={'apiKeyHash': key_hash})

    if 'Item' not in response:
        return generate_deny(event['routeArn'], 'Invalid API key')

    item = response['Item']

    # Verificar si la key esta activa
    if item.get('status') != 'active':
        return generate_deny(event['routeArn'], 'API key inactive')

    # Verificar rate limits o quotas
    tier = item.get('tier', 'free')

    return generate_allow(
        principal_id=item['clientId'],
        resource=event['routeArn'],
        context={
            'clientId': item['clientId'],
            'tier': tier,
        },
    )

def generate_allow(principal_id, resource, context):
    return {
        'principalId': principal_id,
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [{
                'Action': 'execute-api:Invoke',
                'Effect': 'Allow',
                'Resource': resource,
            }],
        },
        'context': context,
    }

def generate_deny(resource, reason):
    return {
        'principalId': 'unauthorized',
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [{
                'Action': 'execute-api:Invoke',
                'Effect': 'Deny',
                'Resource': resource,
            }],
        },
        'context': {'denyReason': reason},
    }

6. Configuracion de Caching

# Caching del authorizer — TTL en segundos
# Cacheado por valor del header de Identity (ej., Authorization header)
Auth:
  DefaultAuthorizer: JwtAuthorizer
  Authorizers:
    JwtAuthorizer:
      FunctionArn: !GetAtt AuthFunction.Arn
      Identity:
        Header: Authorization
        ValidationExpression: ^Bearer .+$
      AuthorizerResultTtlInSeconds: 300  # Cachear por 5 minutos

Como Funciona

1. Flujo de peticion: El cliente envia una peticion con un header Authorization. API Gateway la intercepta, extrae el token e invoca el authorizer Lambda con los detalles de la peticion.
2. Validacion de token: El authorizer valida el token (firma JWT, expiracion, claims). Si es valido, genera una politica IAM con Effect: Allow. Si es invalido, Effect: Deny.
3. Politica IAM: El Resource de la politica es el ARN de la ruta accedida (execute-api:Invoke en arn:aws:execute-api:...). API Gateway evalua la politica — Allow procede al handler, Deny retorna 403.
4. Pase de contexto: El objeto context en la respuesta del authorizer se pasa al handler via event.requestContext.authorizer. Usalo para pasar user ID, rol u otros claims.
5. Caching: API Gateway cachea la respuesta del authorizer identificada por el token. Peticiones subsecuentes con el mismo token saltan el authorizer por AuthorizerResultTtlInSeconds. Esto reduce invocaciones de Lambda y latencia.

Variantes

HTTP API (v2) Authorizer

# HTTP API usa un formato de evento ligeramente diferente
def lambda_handler(event, context):
    # event['headers'] esta en lowercase en HTTP API
    token = event.get('headers', {}).get('authorization', '').replace('Bearer ', '')

    # event['routeArn'] esta disponible
    # event['requestContext']['http']['method'] para el metodo

    claims = validate_jwt(token)
    return {
        'principalId': claims['sub'],
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [{
                'Action': 'execute-api:Invoke',
                'Effect': 'Allow',
                'Resource': event['routeArn'],
            }],
        },
        'context': {'userId': claims['sub']},
    }

Cognito User Pool Authorizer (Sin Lambda Custom)

# Usar Cognito directamente — no se necesita authorizer custom
Auth:
  DefaultAuthorizer: CognitoAuthorizer
  Authorizers:
    CognitoAuthorizer:
      UserPoolArn: !GetAtt UserPool.Arn
      Identity:
        Header: Authorization

Response Authorizer (Pre-token generation)

# Para HTTP API con authorizer OAuth2/JWT (sin Lambda)
# Configurar directamente en API Gateway
# Esto valida JWT de un emisor externo (Auth0, Okta)

Mejores Practicas

• Habilitar caching: Establece AuthorizerResultTtlInSeconds a 300-900 segundos. Esto reduce invocaciones del authorizer en 90%+ para peticiones repetidas con el mismo token.
• Validar expiracion del token: Verifica el claim exp antes de aceptar. El caching de API Gateway puede servir politicas stale para tokens expirados — establece TTL mas corto que la expiracion del token.
• Usar politicas IAM de minimo privilegio: Solo permite execute-api:Invoke en el ARN de ruta especifico, no *.
• Pasar contexto util: Incluye user ID, rol y permisos en el objeto context. Los handlers pueden usar esto sin re-validar el token.
• Manejar errores graceful: Retorna Deny en cualquier fallo de validacion. No lances excepciones — API Gateway trata las excepciones como errores 500, no 403.
• Usar ValidationExpression: Una regex en el header Authorization previene invocar el authorizer para tokens malformados.

Errores Comunes

• No cachear el authorizer: Cada peticion invoca el authorizer Lambda, agregando 50-200ms de latencia e incrementando costos. Siempre establece un TTL.
• Cachear demasiado tiempo: Si el TTL excede la expiracion del token, los tokens revocados permanecen validos. Establece TTL a 300 segundos o menos.
• Usar Deny con recurso wildcard: Deny en * bloquea todo el acceso al API. Deniega solo el ARN de ruta especifico.
• No manejar headers faltantes: Si el header Authorization esta ausente, headers.get('Authorization') retorna None. Maneja este caso explicitamente.
• Lanzar excepciones en lugar de retornar Deny: Las excepciones causan errores 500. Retorna una politica Deny para cualquier fallo de validacion.