Authentification à MongoDB Atlas avec la Fédération d'Identité Sortante AWS

J'ai écrit plusieurs articles sur les sujets d'authentification et d'autorisation, en utilisant les modèles PEP et PDP. PEP et PDP pour une autorisation sécurisée avec Cognito, puis j'ai étendu cela avec Amazon Verified Permissions, et plus tard ajouté ABAC par-dessus. Tout cela concernait l'autorisation utilisateur-to-service. Cela signifie qu'un humain se connecte, obtient un jeton, et le backend décide de ce qu'il peut faire.

Mais qu'en est-il de service-to-service ? Machine-to-machine ? Votre Lambda a besoin de parler à une base de données ou d'appeler une API externe, et il n'y a pas d'humain dans la boucle. Vous auriez toujours besoin d'identifiants.

Tout le code source est disponible sur le Serverless Handbook.

La manière habituelle de gérer les jetons M2M

L'une des approches les plus courantes que je vois est d'utiliser Cognito User Pools, Auth0, ou des solutions similaires avec le flux d'identifiants client OAuth 2.0. Vous créez un serveur de ressources, enregistrez un client d'application avec un ID client et un secret, et votre Lambda échange cela contre un jeton d'accès. Cela fonctionne bien, et si vous utilisez déjà Cognito pour l'authentification utilisateur, cela semble naturel de l'étendre pour M2M également. Ce qu'il faut retenir, c'est que les jetons M2M de ces services peuvent rapidement devenir coûteux.

Une autre approche classique consiste simplement à stocker les identifiants dans Secrets Manager. Mots de passe de base de données, clés API, tout ce dont vous avez besoin. Activez la rotation et passez à autre chose. Cela fonctionne, mais c'est une chose de plus à gérer, une chose de plus qui peut tomber en panne, et un autre secret qui peut fuir.

Fédération d'Identité Sortante

Juste avant re:Invent 2025, une sortie très intéressante est apparue : la Fédération d'Identité Sortante AWS IAM.

Nous utilisons la fédération d'identité dans AWS depuis longtemps. Google, Okta, Entra, ou d'autres problèmes émettent un jeton, AWS le fait confiance, et vous obtenez l'accès aux ressources AWS. C'est la fédération entrante. AWS est celui qui fait confiance.

La Fédération d'Identité Sortante est l'inverse. Maintenant, AWS est celui qui émet les jetons. Votre Lambda (ou EC2, ou tâche ECS) appelle STS pour échanger un rôle IAM contre un jeton JWT signé qui dit essentiellement "hé, je suis ce rôle IAM, et voici une preuve cryptographique." Ensuite, vous remettez ce jeton à n'importe quel service externe ou interne auquel vous parlez, et ils peuvent le vérifier à l'aide de flux standard.

Cela signifie, pas de secrets clients ou de mots de passe stockés. Le jeton émis est à durée courte, lié au rôle IAM, et signé par AWS. Le service récepteur a juste besoin de faire confiance à l'émetteur.

Ce qu'il y a dans le jeton ?

Lorsque nous appelons sts:GetWebIdentityToken, nous obtenons un jeton JWT standard, avec quelques revendications qui comptent :

iss - L'émetteur. C'est l'URL de l'émetteur de jetons que nous obtiendrons lorsque nous activerons la fonctionnalité, elle ressemble à ceci "https://a1810f8a-5a75-4e21-b1cd-a6b09f1836cb.tokens.sts.global.api.aws". Elle est unique à notre compte AWS et nous ne pouvons pas la modifier.
sub - Le sujet. C'est l'ARN du rôle IAM. C'est l'identité.
aud - Le public. C'est défini par nous, et cela peut être n'importe quoi, "my-api", "atlas-demo", ce que vous voulez. Les deux côtés ont juste besoin d'être d'accord sur la valeur.
exp - Expiration. La durée courte est tout l'intérêt.

Aperçu de l'architecture

Dans ce billet, nous allons construire deux solutions différentes qui utilisent la Fédération d'Identité Sortante, juste pour montrer comment la configurer pour un service interne, hébergé sur Lambda derrière API Gateway, et une base de données externe, dans ce cas MongoDB Atlas.

Dans notre cas d'usage interne, nous avons une fonction Lambda de travail qui appellera STS pour obtenir un jeton JWT sortant signé. La fonction appelle une API dans API Gateway où nous avons un Lambda Authorizer qui utilise les clés publiques de STS pour valider le JWT.

Dans le cas de MongoDB, nous avons à peu près la même chose. Nous établissons la confiance entre MongoDB et STS en effectuant certaines configurations du côté MongoDB. Notre fonction Lambda de travail appellera STS pour obtenir le JWT et l'utilisera comme authentification vers MongoDB, qui utilise les clés publiques, etc. de STS pour valider l'utilisateur. Pour le cas d'usage MongoDB, nous devons tout cela mapper à un utilisateur, mais nous en parlerons plus tard.

Maintenant, construisons !

Activer la Fédération d'Identité Sortante

Avant de pouvoir utiliser la Fédération d'Identité Sortante, nous devons l'activer, par défaut elle est désactivée. Alors allez dans la Console AWS et IAM > Paramètres du compte.

Défilez jusqu'à la section Fédération d'Identité Sortante et cliquez simplement sur Activer.

Une fois activée, vous obtiendrez immédiatement votre URL d'Émetteur de Jetons spécifique au compte. Copiez-la, vous en aurez besoin plus tard.

Vous pouvez également l'activer via CLI si vous préférez :

aws iam enable-outbound-web-identity-federation

Maintenant, essayons cela et appelons STS pour obtenir un jeton, nous exécutons la commande CLI ci-dessous pour obtenir un jeton pour notre principal connecté.

aws sts get-web-identity-token --audience "my-api" --signing-algorithm RS256

Nous obtiendrons une réponse comme :

{
    "WebIdentityToken": "Jeton encodé en Base64",
    "Expiration": "2026-03-19T08:20:06.177000+00:00"
}

Si nous décodons le WebIdentityToken, en utilisant une page web comme jwt.io, nous pouvons voir les différentes revendications.

{
  "aud": "my-api",
  "sub": "arn:aws:iam::<account_id><iam_role>",
  "https://sts.amazonaws.com/": {
    "org_id": "<AWS_Org_Id>",
    "aws_account": "<account_id>",
    "ou_path": [
      "...."
    ],
    "original_session_exp": "2026-03-19T10:15:04Z",
    "source_region": "eu-west-1",
    "principal_id": "<iam_principal>",
    "identity_store_user_id": "...."
  },
  "iss": "https://....tokens.sts.global.api.aws",
  "exp": 1773908406,
  "iat": 1773908106,
  "jti": "..."
}

Vérifier le jeton avec API Gateway

Avec cela fonctionnant, construisons une API simple qu'une fonction Lambda appelle, où nous avons un Lambda Authorizer qui valide le jeton.

Nous commençons par créer la fonction appelante, c'est assez simple, nous appelons STS pour obtenir un jeton et le présentons lors de l'appel à l'API.

import json
import logging
import os

import boto3
import requests

logger = logging.getLogger()
logger.setLevel(logging.INFO)

sts_client = boto3.client("sts")

API_ENDPOINT = os.environ["API_ENDPOINT"]
AUDIENCE = os.environ["AUDIENCE"]

def get_token():
    response = sts_client.get_web_identity_token(
        Audience=[AUDIENCE],
        DurationSeconds=300,
        SigningAlgorithm="RS256",
    )
    return response["WebIdentityToken"]


def handler(event, context):
    try:
        token = get_token()

        response = requests.get(
            API_ENDPOINT,
            headers={"Authorization": f"Bearer {token}"},
            timeout=10,
        )

        return {
            "statusCode": response.status_code,
            "body": json.dumps(
                {
                    "api_status_code": response.status_code,
                    "api_response": response.json() if response.ok else response.text,
                    "request_id": context.aws_request_id,
                }
            ),
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps(
                {
                    "message": "Erreur lors de l'appel à l'API protégée",
                    "error": str(e),
                }
            ),
        }

Si nous regardons la fonction get_token(), elle est très simple, juste trois paramètres et c'est tout. Nous n'avons pas besoin de récupérer des identifiants de n'importe où. Le Lambda dit simplement "donne-moi un jeton pour ce public" et STS en signe un avec sa propre identité.

Le Lambda Authorizer

Le Lambda Authorizer attaché à API Gateway, c'est là que cela commence à devenir vraiment intéressant. Ce que nous allons faire, c'est récupérer les clés publiques de STS et ensuite valider le jeton à l'aide d'une bibliothèque JWT standard.

Nous créons la découverte à l'aide de l'URL d'émetteur et ajoutons /.well-known/openid-configuration Cela nous indique où nous pouvons trouver les clés publiques et quelle devrait être la valeur de l'émetteur. Ensuite, nous créons un PyJWKClient qui gère la récupération et le cache des clés.

import json
import logging
import os
import urllib.request

import jwt

logger = logging.getLogger()
logger.setLevel(logging.INFO)

AUDIENCE = os.environ["AUDIENCE"]
ISSUER_URL = os.environ["ISSUER_URL"]

DISCOVERY_URL = f"{ISSUER_URL}/.well-known/openid-configuration"

discovery_doc = json.loads(urllib.request.urlopen(DISCOVERY_URL).read())
JWKS_URI = discovery_doc["jwks_uri"]
EXPECTED_ISSUER = discovery_doc["issuer"]

jwks_client = jwt.PyJWKClient(JWKS_URI)

def handler(event, context):
    try:
        token = extract_token(event)
        if not token:
            return generate_policy("Deny", event["methodArn"])

        signing_key = jwks_client.get_signing_key_from_jwt(token)

        decoded = jwt.decode(
            token,
            signing_key.key,
            algorithms=["RS256"],
            audience=OIDC_AUDIENCE,
            issuer=EXPECTED_ISSUER,
        )

        return generate_policy(
            "Allow",
            event["methodArn"],
            principal_id=decoded.get("sub", "unknown"),
            context={"sub": decoded.get("sub", ""), "iss": decoded.get("iss", "")},
        )

    except jwt.ExpiredSignatureError:
        logger.warning("Le jeton a expiré")
    except jwt.InvalidAudienceError:
        logger.warning("Revendication d'audience invalide")
    except jwt.InvalidIssuerError:
        logger.warning("Revendication d'émetteur invalide")
    except Exception as e:
        logger.error("Échec de la vérification du jeton : %s", e)

    return generate_policy("Deny", event["methodArn"])

La fonction jwt.decode() est le cœur de tout et validera les jetons réels par rapport aux valeurs attendues, et lancera une exception en cas de problème.

Si tout passe, nous autorisons la demande et transmettons la revendication sub afin que le backend sache qui appelle.

Il est important de noter qu'il n'y a rien de spécifique à AWS dans cette logique de vérification. C'est juste une validation standard de jeton JWT.

Authentification à MongoDB Atlas

Maintenant, mettons cela dans un cas d'usage réel, en nous connectant à un cluster MongoDB Atlas en utilisant des jetons de la Fédération d'Identité Sortante IAM, éliminant ainsi le besoin de stocker des mots de passe dans Secrets Manager.

Il y a une chose très importante à noter. Pour que cela fonctionne, vous devrez exécuter un cluster dédié M10+ MongoDB Atlas avec MongoDB 7.0.11 ou supérieur. La Fédération d'Identité de Charge de Travail n'est pas prise en charge sur les clusters gratuits ou partagés (M0, M2, M5).

L'idée est plutôt simple et directe. Notre fonction Lambda appelle STS pour obtenir un jeton, puis présente ce jeton à Atlas au lieu d'un nom d'utilisateur et d'un mot de passe. Atlas validera le jeton, en utilisant des flux standard, et associera le rôle IAM à un utilisateur de base de données.

Configuration d'Atlas

La première étape maintenant est de configurer les choses du côté MongoDB, connectez-vous à votre compte et naviguez vers Fédération sous Identité et accès.

Sous Fédération, sélectionnez Fournisseurs d'Identité, et nous allons commencer à configurer cela.

Sur l'écran suivant, sélectionnez Identité de Charge de Travail afin que nous puissions commencer à ajouter des détails.

Sur l'écran de configuration, remplissez le nom et la description. L'URL d'émetteur est celle de la Console IAM AWS, que nous avons vue plus tôt dans cet article. Pour audience, définissez une valeur de votre choix, il est important que ce soit la même sur MongoDB et AWS config (lors de l'appel à STS). Le public doit correspondre. Pour la revendication utilisateur, gardez la valeur par défaut sub

Enregistrez et terminez votre configuration et vous devriez voir l'écran d'aperçu.

Ensuite, nous devons connecter le fournisseur d'identité que nous venons de créer à notre organisation. Sélectionnez Organisations dans le menu et commencez à connecter.

Dans la boîte de dialogue contextuelle, sélectionnez le fournisseur que nous venons de créer et cliquez sur Connecter.

Dans la vue suivante, nous devrions voir l'aperçu de la connexion.

Déploiement de l'application de test

C'était beaucoup d'étapes, mais maintenant nous pouvons déployer notre application de test et ensuite effectuer la configuration finale dans Atlas avant de tester. Nous allons garder l'infrastructure de test minimale. Juste une fonction Lambda qui a les autorisations sts:GetWebIdentityToken :

Resources:
  DatabaseOpFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub ${Application}-database-op
      CodeUri: src/DatabaseOp/
      Handler: database_op.handler
      Environment:
        Variables:
          MONGODB_URI: !Ref MongoDbUri
          OIDC_AUDIENCE: !Ref OidcAudience
      Policies:
        - Statement:
            - Effect: Allow
              Action:
                - sts:GetWebIdentityToken
              Resource: "*"

Outputs:
  DatabaseOpFunctionRoleArn:
    Description: ARN du rôle IAM pour la fonction Lambda (utilisez ceci dans la cartographie des rôles OIDC d'Atlas)
    Value: !GetAtt DatabaseOpFunctionRole.Arn

Nous ajoutons l'ARN du rôle aux sorties de la pile, car nous en aurons besoin pour la dernière partie d'Atlas. Le code est divisé dans le gestionnaire Lambda et un client qui gérera la connexion à la base de données MongoDB.

Le pilote pymongo a un excellent mécanisme de rappel pour OIDC. Nous lui donnons une classe, et le pilote l'appelle chaque fois qu'il a besoin d'un jeton. Le pilote appelle fetch() lors de la première connexion et à nouveau lorsque le jeton est sur le point d'expirer. Nous définissons la valeur à 280 secondes, ce qui est un peu en dessous des 300 secondes réelles, donc il y a une marge.

Définir authMechanism="MONGODB-OIDC" indique au pilote de sauter le nom d'utilisateur/mot de passe et d'utiliser OIDC à la place.

import os
import boto3
from pymongo import MongoClient
from pymongo.auth_oidc import OIDCCallback, OIDCCallbackResult

sts_client = boto3.client("sts")
cached_client = None


def get_oidc_token():
    audience = os.environ["OIDC_AUDIENCE"]

    response = sts_client.get_web_identity_token(
        Audience=[audience],
        DurationSeconds=300,
        SigningAlgorithm="RS256",
    )

    return response["WebIdentityToken"]


class AwsOidcCallback(OIDCCallback):
    def fetch(self, context):
        token = get_oidc_token()
        return OIDCCallbackResult(access_token=token, expires_in_seconds=280)


def get_mongo_client():
    global cached_client

    if cached_client is not None:
        return cached_client

    uri = os.environ["MONGODB_URI"]

    properties = {"OIDC_CALLBACK": AwsOidcCallback()}

    cached_client = MongoClient(
        uri,
        authMechanism="MONGODB-OIDC",
        authMechanismProperties=properties,
    )

    return cached_client

Puis dans notre gestionnaire Lambda, nous créons le client MongoDB et interagissons avec la base de données, dans ce simple exemple, en insérant simplement un document dans une collection items.

def handler(event, context):
    try:
        client = get_mongo_client()
        db = client["sample_db"]
        collection = db["items"]

        document = {
            "message": "Bonjour depuis Lambda avec authentification OIDC",
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "request_id": context.aws_request_id,
        }

        result = collection.insert_one(document)
        doc = collection.find_one({"_id": result.inserted_id})

        return {
            "statusCode": 200,
            "body": json.dumps({
                "message": "Connecté avec succès à MongoDB Atlas avec OIDC",
                "insertedId": str(result.inserted_id),
            }),
        }
    except Exception as e:
        logger.error("Erreur : %s", e)
        return {"statusCode": 500, "body": json.dumps({"error": str(e)})}

Créer un utilisateur de base de données dans Atlas

Maintenant, nous devons faire la dernière étape dans Atlas, créer un utilisateur de base de données et le mapper à l'ARN du rôle IAM, afin que nous puissions obtenir les autorisations pour interagir avec la base de données. Allez dans Accès à la base de données et ajoutez un nouvel utilisateur de base de données.

Sélectionnez Authentification fédérée comme méthode d'authentification et définissez le nom d'utilisateur sur l'ARN du rôle IAM de la fonction Lambda, récupérez-le dans les sorties de la pile. Assignez à l'utilisateur les autorisations Lire et écrire sur n'importe quelle base de données.

Cet ARN de rôle correspondra à la revendication sub dans le JWT. C'est ainsi qu'Atlas sait quel utilisateur de base de données utiliser.

Autoriser l'accès réseau

Avant de tester tout cela, il y a une autre chose que nous devons faire, nous devons autoriser l'accès réseau pour notre fonction Lambda. Comme nous ne tournons pas dans un VPC, nous devons autoriser 0.0.0.0/0 dans une configuration de production, ne faites pas cela, mais pour ce test, c'est très bien.

Naviguez vers Liste d'accès IP dans le menu

Puis ajoutez 0.0.0.0/0, définissez-le comme temporaire afin qu'il soit supprimé automatiquement.

C'est tout ! Nous pouvons maintenant utiliser la Fédération d'Identité Sortante dans STS pour nous authentifier auprès de MongoDB Atlas et lire et écrire des données, pas de mots de passe, pas de secrets à gérer !!

Tests

Il est temps de faire un petit test, pour cela, invoquez simplement la fonction Lambda déployée. Ensuite, allez dans Explorateur de données dans Atlas et vérifiez que les données sont là.

Derniers mots

J'aime vraiment ce modèle. Pas de secrets stockés. Des jetons qui expirent en 5 minutes. Vous savez exactement quel rôle IAM a fait chaque connexion. Et cela fonctionne avec tout ce qui parle OIDC, pas seulement MongoDB. Tout service externe qui peut vérifier les jetons OIDC peut utiliser la même fonction get_oidc_token(). Le même code, une destination différente.

Découvrez mes autres articles sur jimmydqv.com et suivez-moi sur LinkedIn et X pour plus de contenu serverless.

Maintenant, allez construire !