Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Configura un autorizzatore AWS Lambda per l'autenticazione OIDC

Questa guida presuppone che tu abbia già configurato il tuo Identity Provider (IdP) preferito per fornire token di accesso compatibili con i requisiti della funzionalità di HealthImaging autenticazione OIDC.

1. Configura i ruoli IAM per l'accesso alle API DICOMWeb

Prima di configurare l'autorizzatore Lambda, crea ruoli IAM da assumere durante HealthImaging DICOMWeb l'elaborazione delle richieste API. La funzione authorizer Lambda restituisce uno di questi ruoli ARN dopo una corretta verifica del token, HealthImaging permettendo di eseguire le richieste con le autorizzazioni appropriate.

Ecco un esempio di policy che consente ai ruoli associati di accedere a un'API di HealthImaging DICOMWeb sola lettura:

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

Ecco un esempio della politica di relazione di fiducia che dovrebbe essere associata al/ai ruolo/i:

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

L'autorizzatore Lambda che creerai nel passaggio successivo può valutare le attestazioni del token e restituire l'ARN del ruolo appropriato. AWS HealthImaging impersonerà quindi questo ruolo per eseguire la richiesta DICOMWeb API con le autorizzazioni corrispondenti.

Ad esempio:

Questo meccanismo ti consente di mappare il modello di autorizzazione del tuo IdP a specifiche HealthImaging autorizzazioni AWS tramite ruoli IAM.

2. Creazione e configurazione della funzione Lambda Authorizer

Crea una funzione Lambda che verifichi il token JWT e restituisca l'ARN del ruolo IAM appropriato in base alla valutazione delle dichiarazioni del token. Questa funzione viene richiamata dal servizio di imaging sanitario e trasmette un evento che contiene l'ID del HealthImaging datastore, l' DICOMWeb operazione e il token di accesso trovati nella richiesta HTTP:

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

La funzione di autorizzazione Lambda deve restituire una risposta JSON con la seguente struttura:

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

Puoi fare riferimento all'esempio di implementazione per ulteriori informazioni.

Affinché il HealthImaging servizio sia autorizzato a richiamare la funzione di autorizzazione lambda, deve disporre di una politica delle risorse che HealthImaging consenta al servizio di richiamarla. Questa politica delle risorse può essere creata nel menu di autorizzazione della scheda di configurazione lambda o utilizzando: AWS CLI

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

Questa politica delle risorse consente al HealthImaging servizio di richiamare l'autorizzatore Lambda durante DICOMWeb l'autenticazione delle richieste API.

Ecco un esempio di politica delle risorse 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. Crea un nuovo datastore con l'autenticazione OIDC

Per abilitare l'autenticazione OIDC, è necessario creare un nuovo datastore utilizzando il parametro "». AWS CLI lambda-authorizer-arn L'autenticazione OIDC non può essere abilitata su datastore esistenti senza contattare l'assistenza. AWS

Ecco un esempio di come creare un nuovo datastore con l'autenticazione OIDC abilitata:

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

Puoi verificare se uno specifico datastore ha la funzionalità di autenticazione OIDC abilitata utilizzando il comando AWS CLI get-datastore e verificando se l'attributo "" è 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"
    }
}

Codici di eccezione

In caso di errore di autenticazione HealthImaging restituisce i seguenti codici di risposta di errore HTTP e messaggi del corpo:

Esempio di implementazione

Questo esempio in Python dimostra una funzione di autorizzazione lambda che verifica i token di accesso a AWS Cognito dagli eventi HealthImaging e restituisce un ARN del ruolo IAM con i privilegi appropriati. DICOMWeb

L'autorizzatore Lambda implementa due meccanismi di memorizzazione nella cache per ridurre le chiamate esterne e la latenza di risposta. Il JWKS (JSON Web Key Set) viene recuperato una volta ogni ora e archiviato nella cartella temporanea della funzione, per consentire alle successive chiamate di funzione di leggerlo localmente anziché recuperarlo dalla rete pubblica. Noterai anche che un oggetto dizionario token_cache viene istanziato nel contesto globale di questa funzione Lambda. Le variabili globali sono condivise da tutte le chiamate che riutilizzano lo stesso contesto Lambda riscaldato. Grazie a ciò, i token verificati con successo possono essere archiviati in questo dizionario e consultati rapidamente durante la successiva esecuzione della stessa funzione Lambda. Il metodo di memorizzazione nella cache rappresenta un approccio generalista che potrebbe adattarsi ai token di accesso emessi dalla maggior parte dei provider di identità. Per un'opzione di caching specifica per AWS Cognito, consulta la sezione Gestione del pool di utenti e la sezione caching della documentazione di 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': ''
        }