集成AWS API Gateway与Azure AD验证方案

import json
import logging
import os

import time
import traceback
import urllib.request

import jwt as JWT
from jose import jwk, jwt
from jose.utils import base64url_decode

logger = logging.getLogger()
logger.setLevel(os.environ.get("LOG_LEVEL", "INFO"))

app_client_id = os.getenv("APP_CLIENT_ID")
keys_url = "https://login.microsoftonline.com/common/discovery/v2.0/keys"
with urllib.request.urlopen(keys_url) as f:
    response = f.read()
keys = json.loads(response.decode("utf-8"))["keys"]


def lambda_handler(event, context):
    if event.get("type") != "TOKEN":
        return json.dumps({})
    print(event)
    try:
        req_token = event.get("authorizationToken")
        method = event["methodArn"]
        empty_response = {
            "principalId": "Unknown",
            "policyDocument": {
                "Version": "2012-10-17",
                "Statement": [
                    {"Action": "execute-api:Invoke", "Effect": "Deny", "Resource": "*"}
                ],
            },
        }

        headers = JWT.get_unverified_header(req_token)
        kid = headers["kid"]
        # search for the kid in the downloaded public keys
        key_index = -1
        for i in range(len(keys)):
            if kid == keys[i]["kid"]:
                key_index = i
                break
        if key_index == -1:
            print("Public key not found in keys")
            return empty_response
        public_key = jwk.construct(keys[key_index], algorithm="RS256")
        # get the last two sections of the token, message and signature (encoded in base64)
        message, encoded_signature = str(req_token).rsplit(".", 1)
        # decode the signature
        decoded_signature = base64url_decode(encoded_signature.encode("utf-8"))
        # verify the signature
        if not public_key.verify(message.encode("utf8"), decoded_signature):
            print("Signature verification failed")
            return empty_response
        print("Signature successfully verified")
        claims = jwt.get_unverified_claims(req_token)
        print("claims = ", claims)
        # Verify the token expiration
        if time.time() > claims["exp"] or claims["aud"] != app_client_id:
            print("Token is expired or was not issued for this client")
            return {
                "principalId": claims["preferred_username"],
                "policyDocument": {
                    "Version": "2012-10-17",
                    "Statement": [
                        {"Action": "execute-api:Invoke", "Effect": "Deny", "Resource": method}
                    ],
                },
            }
        success_resp = {
            "principalId": claims["preferred_username"],
            "policyDocument": {
                "Version": "2012-10-17",
                "Statement": [
                    {"Action": "execute-api:Invoke", "Effect": "Allow", "Resource": method}
                ],
            },
        }
        print(success_resp)
        return success_resp

    except Exception as e:
        print("........Error.... !!!!!!....... ", e)
        logger.critical(traceback.format_exc(limit=25))
        return

pip3 install --target ./python pyjwt python-jose
zip -r auth-lib.zip ./python

msalInstance.acquireTokenSilent(request)
    .then((response) => {
        var idToken = response.idToken;
        })