As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Configurar um autorizador AWS Lambda para autenticação OIDC

Este guia pressupõe que você já tenha configurado o provedor de identidade (IdP) de sua escolha para fornecer tokens de acesso compatíveis com os requisitos do recurso de autenticação HealthImaging do OIDC.

1. Configurar funções do IAM para acesso à DICOMWeb API

Antes de configurar o autorizador Lambda, crie funções do IAM HealthImaging para assumir ao DICOMWeb processar solicitações de API. A função autorizadora Lambda retorna uma dessas funções ARN após a verificação bem-sucedida do token, HealthImaging permitindo executar as solicitações com as permissões apropriadas.

Aqui está um exemplo de uma política que permite que funções associadas acessem a API HealthImaging DICOMWeb somente para leitura:

{
    "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}"
        }
    ]
}

Aqui está um exemplo da política de relacionamento de confiança que deve ser associada à (s) função (s):

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

O autorizador Lambda que você criará na próxima etapa pode avaliar as reivindicações de token e retornar o ARN da função apropriada. A AWS então HealthImaging representará essa função para executar a solicitação de DICOMWeb API com as permissões correspondentes.

Por exemplo:

Esse mecanismo permite que você mapeie o modelo de autorização do seu IdP para HealthImaging permissões específicas da AWS por meio de funções do IAM.

2. Criar e configurar a função Lambda Authorizer

Crie uma função Lambda que verificará o token JWT e retornará o ARN apropriado da função do IAM com base na avaliação das declarações do token. Essa função é invocada pelo serviço de imagens de saúde e transmitida por um evento que contém o ID do HealthImaging armazenamento de dados, a DICOMWeb operação e o token de acesso encontrados na solicitação HTTP:

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

A função autorizadora do Lambda deve retornar uma resposta JSON com a seguinte estrutura:

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

Você pode consultar o exemplo de implementação para obter mais informações.

Para que o HealthImaging serviço seja autorizado a invocar a função autorizadora lambda, ele deve ter uma política de recursos que permita que o HealthImaging serviço a invoque. Essa política de recursos pode ser criada no menu de permissões da guia de configuração lambda ou usando AWS CLI:

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

Essa política de recursos permite que o HealthImaging serviço invoque seu autorizador Lambda ao DICOMWeb autenticar solicitações de API.

Aqui está um exemplo da política de recursos 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. Crie um novo armazenamento de dados com a autenticação OIDC

Para habilitar a autenticação OIDC, você deve criar um novo armazenamento de dados usando o AWS CLI com o parâmetro "”. lambda-authorizer-arn A autenticação OIDC não pode ser habilitada nos armazenamentos de dados existentes sem entrar em contato com o Support. AWS

Veja um exemplo de como criar um novo armazenamento de dados com a autenticação OIDC ativada:

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

Você pode verificar se um armazenamento de dados específico tem o recurso de autenticação OIDC ativado usando o comando AWS CLI get-datastore e verificando se o 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 exceção

Em caso de falha na autenticação, HealthImaging retorna os seguintes códigos de resposta de erro HTTP e mensagens corporais:

Exemplo de implementação

Este exemplo em Python demonstra uma função autorizadora lambda que verifica os tokens de acesso do Cognito a partir de eventos HealthImaging e retorna um AWS ARN de função do IAM com os privilégios apropriados. DICOMWeb

O autorizador Lambda implementa dois mecanismos de armazenamento em cache para reduzir as chamadas externas e a latência de resposta. O JWKS (JSON Web Key Set) é obtido uma vez a cada hora e armazenado na pasta temporária da função, permitindo que as invocações subsequentes da função o leiam localmente em vez de buscá-lo na rede pública. Você também notará que um objeto de dicionário token_cache é instanciado no contexto global dessa função Lambda. As variáveis globais são compartilhadas por todas as invocações que reutilizam o mesmo contexto aquecido do Lambda. Graças a isso, os tokens verificados com sucesso podem ser armazenados neste dicionário e pesquisados rapidamente durante a próxima execução dessa mesma função Lambda. O método de armazenamento em cache representa uma abordagem generalista que pode caber em tokens de acesso emitidos pela maioria dos provedores de identidade. Para uma opção de armazenamento em cache específica do AWS Cognito, consulte a seção Gerenciando grupos de usuários e a seção de armazenamento em cache da documentação do 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': ''
        }