Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Richten Sie einen AWS Lambda-Authorizer für die OIDC-Authentifizierung ein

In diesem Handbuch wird davon ausgegangen, dass Sie den Identity Provider (IdP) Ihrer Wahl bereits so konfiguriert haben, dass er Zugriffstoken bereitstellt, die mit den Anforderungen der HealthImaging OIDC-Authentifizierungsfunktion kompatibel sind.

1. Konfigurieren Sie IAM-Rollen für den API-Zugriff DICOMWeb

Erstellen Sie vor der Konfiguration des Lambda-Autorisierers IAM-Rollen, die bei der Verarbeitung DICOMWeb von HealthImaging API-Anfragen verwendet werden sollen. Die Autorisierungs-Lambda-Funktion gibt nach erfolgreicher Token-Überprüfung den ARN einer dieser Rollen zurück, sodass HealthImaging die Anfragen mit den entsprechenden Berechtigungen ausgeführt werden können.

Hier ist ein Beispiel für eine Richtlinie, die zugehörigen Rollen den Zugriff auf eine HealthImaging DICOMWeb schreibgeschützte API ermöglicht:

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

Hier ist ein Beispiel für die Vertrauensstellungsrichtlinie, die den Rollen zugeordnet werden sollte:

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

Der Lambda-Autorisierer, den Sie im nächsten Schritt erstellen, kann die Token-Ansprüche auswerten und den ARN der entsprechenden Rolle zurückgeben. AWS HealthImaging gibt sich dann als diese Rolle aus, um die DICOMWeb API-Anfrage mit den entsprechenden Berechtigungen auszuführen.

Beispiel:

Dieser Mechanismus ermöglicht es Ihnen, das Autorisierungsmodell Ihres IdP über IAM-Rollen bestimmten HealthImaging AWS-Berechtigungen zuzuordnen.

2. Lambda Authorizer-Funktion erstellen und konfigurieren

Erstellen Sie eine Lambda-Funktion, die das JWT-Token verifiziert und den entsprechenden IAM-Rollen-ARN auf der Grundlage der Bewertung der Token-Ansprüche zurückgibt. Diese Funktion wird vom Health Imaging Service aufgerufen und hat ein Ereignis übergeben, das die HealthImaging Datenspeicher-ID, den DICOMWeb Vorgang und das Zugriffstoken enthält, die in der HTTP-Anfrage gefunden wurden:

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

Die Lambda-Autorisierungsfunktion muss eine JSON-Antwort mit der folgenden Struktur zurückgeben:

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

Weitere Informationen finden Sie im Implementierungsbeispiel.

Damit der HealthImaging Dienst autorisiert werden kann, die Lambda-Autorisierungsfunktion aufzurufen, muss er über eine Ressourcenrichtlinie verfügen, die es dem Dienst ermöglicht HealthImaging , ihn aufzurufen. Diese Ressourcenrichtlinie kann im Berechtigungsmenü der Registerkarte „Lambda-Konfiguration“ oder mithilfe von: erstellt werden AWS CLI

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

Diese Ressourcenrichtlinie ermöglicht es dem HealthImaging Dienst, Ihren Lambda-Authorizer aufzurufen, wenn API-Anfragen authentifiziert DICOMWeb werden.

Hier ist ein Beispiel für eine Lambda-Ressourcenrichtlinie:

{
  "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. Erstellen Sie einen neuen Datenspeicher mit OIDC-Authentifizierung

Um die OIDC-Authentifizierung zu aktivieren, müssen Sie einen neuen Datenspeicher mit dem Parameter "" erstellen. AWS CLI lambda-authorizer-arn Die OIDC-Authentifizierung kann nicht für bestehende Datenspeicher aktiviert werden, ohne den Support zu kontaktieren. AWS

Hier ist ein Beispiel für die Erstellung eines neuen Datenspeichers mit aktivierter OIDC-Authentifizierung:

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

Sie können überprüfen, ob für einen bestimmten Datenspeicher die OIDC-Authentifizierungsfunktion aktiviert ist, indem Sie den Befehl AWS CLI get-datastore verwenden und überprüfen, ob das Attribut "" vorhanden ist: 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"
    }
}

Ausnahmecodes

Im Falle eines Authentifizierungsfehlers werden die folgenden HTTP-Fehlerantwortcodes und Textnachrichten HealthImaging zurückgegeben:

Beispiel für eine Implementierung

Dieses Python-Beispiel demonstriert eine Lambda-Autorisierungsfunktion, die AWS Cognito-Zugriffstoken anhand von HealthImaging Ereignissen überprüft und einen IAM-Rollen-ARN mit entsprechenden Rechten zurückgibt. DICOMWeb

Der Lambda-Authorizer implementiert zwei Caching-Mechanismen, um externe Aufrufe und Antwortlatenz zu reduzieren. Das JWKS (JSON Web Key Set) wird einmal pro Stunde abgerufen und im temporären Ordner der Funktion gespeichert, sodass es bei nachfolgenden Funktionsaufrufen lokal gelesen werden kann, anstatt es aus dem öffentlichen Netzwerk abzurufen. Sie werden auch feststellen, dass ein token_cache-Wörterbuchobjekt im globalen Kontext dieser Lambda-Funktion instanziiert wird. Globale Variablen werden von allen Aufrufen gemeinsam genutzt, die denselben erwärmten Lambda-Kontext wiederverwenden. Dadurch können erfolgreich verifizierte Token in diesem Wörterbuch gespeichert und bei der nächsten Ausführung derselben Lambda-Funktion schnell nachgeschlagen werden. Die Caching-Methode stellt einen generalistischen Ansatz dar, der für Zugriffstoken geeignet ist, die von den meisten Identitätsanbietern ausgegeben wurden. Eine AWS Cognito-spezifische Caching-Option finden Sie in den Abschnitten Benutzerpool verwalten und Caching in der AWS Cognito-Dokumentation.

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': ''
        }