Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Configurer un autorisateur AWS Lambda pour l'authentification OIDC

Ce guide part du principe que vous avez déjà configuré le fournisseur d'identité (IdP) de votre choix pour fournir des jetons d'accès compatibles avec les exigences de la fonctionnalité d'authentification HealthImaging OIDC.

1. Configuration des rôles IAM pour l'accès aux DICOMWeb API

Avant de configurer l'autorisateur Lambda, créez des rôles IAM HealthImaging à assumer lors du traitement des demandes d'API. DICOMWeb La fonction Lambda d'autorisation renvoie l'ARN de l'un de ces rôles après une vérification réussie du jeton, HealthImaging ce qui permet d'exécuter les demandes avec les autorisations appropriées.

Voici un exemple de politique permettant aux rôles associés d'accéder à une API en HealthImaging DICOMWeb lecture seule :

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

Voici un exemple de politique de relation de confiance qui doit être associée au (x) rôle (s) :

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

L'autorisateur Lambda que vous allez créer à l'étape suivante peut évaluer les demandes de jetons et renvoyer l'ARN du rôle approprié. AWS HealthImaging se fera ensuite passer pour ce rôle pour exécuter la demande d' DICOMWeb API avec les autorisations correspondantes.

Exemples :

Ce mécanisme vous permet de mapper le modèle d'autorisation de votre IdP à des HealthImaging autorisations AWS spécifiques via des rôles IAM.

2. Création et configuration de la fonction d'autorisation Lambda

Créez une fonction Lambda qui vérifiera le jeton JWT et renverra l'ARN du rôle IAM approprié en fonction de l'évaluation des réclamations du jeton. Cette fonction est invoquée par le service d'imagerie médicale et transmet un événement contenant l'identifiant de la HealthImaging banque de données, l' DICOMWeb opération et le jeton d'accès trouvés dans la requête HTTP :

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

La fonction d'autorisation Lambda doit renvoyer une réponse JSON avec la structure suivante :

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

Vous pouvez vous référer à l'exemple de mise en œuvre pour plus d'informations.

Pour que le HealthImaging service soit autorisé à invoquer la fonction d'autorisation Lambda, il doit disposer d'une politique de ressources autorisant le HealthImaging service à l'invoquer. Cette politique de ressources peut être créée dans le menu des autorisations de l'onglet de configuration Lambda ou en utilisant AWS CLI :

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

Cette politique de ressources permet au HealthImaging service d'invoquer votre autorisateur Lambda lors de l'authentification DICOMWeb des demandes d'API.

Voici un exemple de politique de ressources 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. Création d'une nouvelle banque de données avec authentification OIDC

Pour activer l'authentification OIDC, vous devez créer une nouvelle banque de données à l' AWS CLI aide du paramètre « »lambda-authorizer-arn. L'authentification OIDC ne peut pas être activée sur les banques de données existantes sans contacter le Support. AWS

Voici un exemple de création d'une nouvelle banque de données avec l'authentification OIDC activée :

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

Vous pouvez vérifier si la fonctionnalité d'authentification OIDC est activée dans une banque de données spécifique en utilisant la commande AWS CLI get-datastore et en vérifiant si l'attribut « » est présent : 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"
    }
}

Codes d'exception

En cas d'échec de l'authentification, HealthImaging renvoie les codes de réponse d'erreur HTTP et le corps des messages suivants :

Exemple de mise en œuvre

Cet exemple Python illustre une fonction d'autorisation Lambda qui vérifie les jetons d'accès AWS Cognito à HealthImaging partir d'événements et renvoie un ARN de rôle IAM avec les privilèges appropriés. DICOMWeb

L'autorisateur Lambda implémente deux mécanismes de mise en cache pour réduire les appels externes et la latence des réponses. Le JWKS (JSON Web Key Set) est extrait une fois par heure et stocké dans le dossier temporaire de la fonction, ce qui permet aux invocations ultérieures de la fonction de le lire localement au lieu de le récupérer sur le réseau public. Vous remarquerez également qu'un objet du dictionnaire token_cache est instancié dans le contexte global de cette fonction Lambda. Les variables globales sont partagées par toutes les invocations qui réutilisent le même contexte Lambda réchauffé. Grâce à cela, les jetons vérifiés avec succès peuvent être stockés dans ce dictionnaire et recherchés rapidement lors de la prochaine exécution de cette même fonction Lambda. La méthode de mise en cache représente une approche généraliste qui pourrait s'adapter aux jetons d'accès émis par la plupart des fournisseurs d'identité. Pour une option de mise en cache spécifique à AWS Cognito, reportez-vous à la section Gestion du groupe d'utilisateurs et à la section relative à la mise en cache de la documentation 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': ''
        }