Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Configurar un autorizador AWS Lambda para la autenticación OIDC

Esta guía asume que ya ha configurado el proveedor de identidad (IdP) de su elección para proporcionar tokens de acceso compatibles con los requisitos de la función de autenticación del HealthImaging OIDC.

1. Configure las funciones de IAM para el acceso a la API DICOMWeb

Antes de configurar el autorizador Lambda, cree las funciones de IAM para que las asuma HealthImaging al procesar las solicitudes de API. DICOMWeb La función Lambda del autorizador devuelve el ARN de uno de estos roles tras una verificación correcta del token, HealthImaging lo que permite ejecutar las solicitudes con los permisos adecuados.

A continuación, se muestra un ejemplo de una política que permite a los roles asociados acceder a una API de HealthImaging DICOMWeb solo lectura:

{
    "Version": "2012-10-17",		 	 	 
    "Statement": [
        {
            "Sid": "MedicalImagingDicomWebOperations",
            "Effect": "Allow",
            "Action": [
                "medical-imaging:SearchDICOMInstances",
                "medical-imaging:GetImageSetMetadata",
                "medical-imaging:GetDICOMSeriesMetadata",
                "medical-imaging:SearchDICOMStudies",
                "medical-imaging:GetDICOMBulkdata",
                "medical-imaging:SearchDICOMSeries",
                "medical-imaging:GetDICOMInstanceMetadata",
                "medical-imaging:GetDICOMInstance",
                "medical-imaging:GetDICOMInstanceFrames"
            ],
            "Resource": "arn:aws:medical-imaging:{Region}:{Account}:datastore/{DatastoreId}"
        }
    ]
}

A continuación, se muestra un ejemplo de la política de relaciones de confianza que debe asociarse a los roles:

{
    "Version": "2012-10-17",		 	 	 
    "Statement": [
        {
            "Sid": "OIDCRoleFederation",
            "Effect": "Allow",
            "Principal": {
                "Service": "medical-imaging.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

El autorizador Lambda que creará en el siguiente paso puede evaluar las reclamaciones de token y devolver el ARN de la función correspondiente. A continuación, AWS HealthImaging suplantará esta función para ejecutar la solicitud de DICOMWeb API con los permisos correspondientes.

Por ejemplo:

Este mecanismo le permite asignar el modelo de autorización de su IdP a HealthImaging permisos específicos de AWS mediante funciones de IAM.

2. Creación y configuración de la función de autorización Lambda

Cree una función Lambda que verifique el token JWT y devuelva el ARN del rol de IAM correspondiente en función de la evaluación de las afirmaciones del token. El servicio de imágenes médicas invoca esta función y pasa un evento que contiene el identificador del HealthImaging almacén de datos, la DICOMWeb operación y el token de acceso que se encuentran en la solicitud HTTP:

{
  "datastoreId": "{datastore id}",
  "operation": "{Healthimaging API name e.g. GetDICOMInstance}",
  "bearerToken": "{access token}"
}

La función autorizadora de Lambda debe devolver una respuesta JSON con la siguiente estructura:

{
  "isTokenValid": {true or false},
  "roleArn": "{role arn or empty string meaning to deny the request explicitly}"
}

Puede consultar el ejemplo de implementación para obtener más información.

Para que el HealthImaging servicio esté autorizado a invocar la función de autorización lambda, debe tener una política de recursos que permita HealthImaging al servicio invocarla. Esta política de recursos se puede crear en el menú de permisos de la pestaña de configuración de lambda o mediante: AWS CLI

aws lambda add-permission \
    --function-name YourAuthorizerFunctionName \
    --statement-id HealthImagingInvoke \
    --action lambda:InvokeFunction \
    --principal medical-imaging.amazonaws.com

Esta política de recursos permite que el HealthImaging servicio invoque su autorizador Lambda al DICOMWeb autenticar las solicitudes de API.

A continuación, se muestra un ejemplo de política de recursos de lambda:

{
  "Version": "2012-10-17",		 	 	 
  "Id": "default",
  "Statement": [
    {
      "Sid": "LambaAuthorizer-HealthImagingInvokePermission",
      "Effect": "Allow",
      "Principal": {
        "Service": "medical-imaging.amazonaws.com"
      },
      "Action": "lambda:InvokeFunction",
      "Resource": "arn:aws:lambda:{Region}:{Account}::function:{LambdaAuthorizerFunctionName}",
      "Condition": {
        "ArnLike": {
          "AWS:SourceArn": "arn:aws:medical-imaging:{Region}:{Account}:datastore/{DatastoreId}"
        }
      }
    }
  ]
}

3. Cree un nuevo almacén de datos con autenticación OIDC

Para habilitar la autenticación OIDC, debe crear un nuevo almacén de datos con el parámetro "». AWS CLI lambda-authorizer-arn La autenticación OIDC no se puede habilitar en los almacenes de datos existentes sin ponerse en contacto con Support. AWS

Este es un ejemplo de cómo crear un almacén de datos nuevo con la autenticación OIDC habilitada:

aws medical-imaging create-datastore \
    --datastore-name YourDatastoreName \
    --lambda-authorizer-arn YourAuthorizerFunctionArn

Puede comprobar si un almacén de datos específico tiene habilitada la función de autenticación OIDC mediante el comando AWS CLI get-datastore y verificando si el atributo "" está presente: lambdaAuthorizerArn

aws medical-imaging get-datastore --datastore-id YourDatastoreId

{
    "datastoreProperties": {
        "datastoreId": YourdatastoreId,
        "datastoreName": YourDatastoreName,
        "datastoreStatus": "ACTIVE",
        "lambdaAuthorizerArn": YourAuthorizerFunctionArn,
        "datastoreArn": YourDatastoreArn,
        "createdAt": "2025-09-30T14:16:04.015000-05:00",
        "updatedAt": "2025-09-30T14:16:04.015000-05:00"
    }
}

Códigos de excepción

En caso de error de autenticación, HealthImaging devuelve los siguientes códigos de respuesta al error HTTP y mensajes del cuerpo:

Ejemplo de implementación

En este ejemplo de Python, se muestra una función autorizadora lambda que verifica los tokens de acceso de AWS Cognito a HealthImaging partir de eventos y devuelve un ARN de rol de IAM con los privilegios adecuados. DICOMWeb

El autorizador Lambda implementa dos mecanismos de almacenamiento en caché para reducir las llamadas externas y la latencia de respuesta. El JWKS (conjunto de claves web JSON) se obtiene una vez cada hora y se almacena en la carpeta temporal de la función, lo que permite que las invocaciones posteriores a la función lo lean localmente en lugar de obtenerlo desde la red pública. También observará que se crea una instancia de un objeto de diccionario token_cache en el contexto global de esta función Lambda. Todas las invocaciones que reutilizan el mismo contexto Lambda calentado comparten las variables globales. Gracias a esto, los tokens verificados correctamente se pueden almacenar en este diccionario y consultarlos rápidamente durante la próxima ejecución de esta misma función Lambda. El método de almacenamiento en caché representa un enfoque generalista que podría adaptarse a los tokens de acceso emitidos por la mayoría de los proveedores de identidad. Para ver una opción de almacenamiento en caché específica de AWS Cognito, consulte la sección Gestión del grupo de usuarios y la sección de almacenamiento en caché de la documentación de Cognito.AWS

import json
import os
import time
import logging
from jose import jwk, jwt
from jose.exceptions import ExpiredSignatureError, JWTClaimsError, JWTError
import requests
import tempfile

# Configure logging
logger = logging.getLogger()
log_level = os.environ.get('LOG_LEVEL', 'WARNING').upper()
logger.setLevel(getattr(logging, log_level, logging.WARNING))

# Global token cache with TTL
token_cache = {}

# JWKS cache file path
JWKS_CACHE_FILE = os.path.join(tempfile.gettempdir(), 'jwks.json')
JWKS_CACHE_TTL = 3600  # 1 hour

# Load environment variables once
USER_POOL_ID = os.environ['USER_POOL_ID']
CLIENT_ID = os.environ['CLIENT_ID']
ROLE_ARN = os.environ.get('AHIDICOMWEB_READONLY_ROLE_ARN', '')

def cleanup_expired_tokens():
    """Remove expired tokens from cache"""
    now = int(time.time())
    expired_keys = [token for token, data in token_cache.items() if now > data['cache_expiry']]
    for token in expired_keys:
        del token_cache[token]

def get_cached_jwks():
    """Get JWKS from cache file if valid, otherwise return None """
    try:
        if os.path.exists(JWKS_CACHE_FILE):
            # Check if cache file is still valid
            cache_age = time.time() - os.path.getmtime(JWKS_CACHE_FILE)
            if cache_age < JWKS_CACHE_TTL:
                with open(JWKS_CACHE_FILE, 'r') as f:
                    jwks = json.load(f)
                    logger.debug(f'Using cached JWKS (age: {int(cache_age)}s)')
                    return jwks
            else:
                logger.debug(f'JWKS cache expired (age: {int(cache_age)}s)')
    except Exception as e:
        logger.debug(f'Error reading JWKS cache: {e}')
    
    return None

def cache_jwks(jwks):
    """Cache JWKS to file"""
    try:
        with open(JWKS_CACHE_FILE, 'w') as f:
            json.dump(jwks, f)
        logger.debug('JWKS cached successfully')
    except Exception as e:
        logger.debug(f'Error caching JWKS: {e}')

def fetch_jwks(jwks_url):
    """Fetch JWKS from URL and cache it"""
    logger.debug('Fetching JWKS from URL')
    jwks = requests.get(jwks_url, timeout=10).json()
    # Convert to dict for faster lookups
    jwks['keys_by_kid'] = {key['kid']: key for key in jwks['keys']}
    cache_jwks(jwks)
    return jwks

def is_token_cached(token):
    if token not in token_cache:
        return None
    
    cached = token_cache[token]
    now = int(time.time())
    
    if now > cached['cache_expiry']:
        del token_cache[token]
        return None
    
    return cached

def cache_token(token, payload):
    now = int(time.time())
    token_exp = payload.get('exp')
    cache_expiry = min(now + 60, token_exp)  # 1 minute or token expiry, whichever is sooner
    
    token_cache[token] = {
        'payload': payload,
        'cache_expiry': cache_expiry,
        'role_arn': ROLE_ARN
    }

def handler(event, context):
    cleanup_expired_tokens() # start be removing expired tokens from the cache
    try:
        # Extract token from bearerToken or authorizationToken field
        token = event.get('bearerToken')
        if not token:
            raise Exception('No token provided')
        
        # Check cache first
        cached = is_token_cached(token)
        if cached:
            logger.debug('Token found in cache, skipping verification')
            return {
                'isTokenValid': True,
                'roleArn': cached['role_arn']
            }
        
        # Get Cognito configuration
        region = context.invoked_function_arn.split(':')[3]
        
        # Get JWKS (cached or fresh)
        jwks_url = f'https://cognito-idp.{region}.amazonaws.com/{USER_POOL_ID}/.well-known/jwks.json'
        jwks = get_cached_jwks()
        if not jwks:
            jwks = fetch_jwks(jwks_url)
        
        # Decode token header to get kid
        headers = jwt.get_unverified_headers(token)
        kid = headers['kid']
        
        # Find the correct key
        key = None
        for jwk_key in jwks['keys']:
            if jwk_key['kid'] == kid:
                key = jwk_key
                break
        
        if not key:
            # Key not found - try refreshing JWKS in case of key rotation
            logger.debug('Key not found in cached JWKS, fetching fresh JWKS')
            jwks = fetch_jwks(jwks_url)
            for jwk_key in jwks['keys']:
                if jwk_key['kid'] == kid:
                    key = jwk_key
                    break
        
        if not key:
            raise Exception('Public key not found')
        
        # Construct the public key
        public_key = jwk.construct(key)
        
        # Verify and decode the token (includes expiry validation)
        payload = jwt.decode(
            token,
            public_key,
            algorithms=['RS256'],
            audience=CLIENT_ID,
            issuer=f'https://cognito-idp.{region}.amazonaws.com/{USER_POOL_ID}'
        )
        
        logger.debug('Token validated successfully')
        logger.debug('User: %s', payload.get('username', 'unknown'))
        
        # Cache the validated token
        cache_token(token, payload)
        
        # Return authorization response
        return {
            'isTokenValid': True,
            'roleArn': ROLE_ARN
        }
        
    except ExpiredSignatureError:
        logger.debug('Token expired')
        return {
            'isTokenValid': False,
            'roleArn': ''
        }
    except JWTClaimsError:
        logger.debug('Invalid token claims')
        return {
            'isTokenValid': False,
            'roleArn': ''
        }
    except JWTError as e:
        logger.debug('JWT validation error: %s', e)
        return {
            'isTokenValid': False,
            'roleArn': ''
        }
    except Exception as e:
        logger.debug('Authorization failed: %s', e)
        return {
            'isTokenValid': False,
            'roleArn': ''
        }