API Gateway 认证实践：IAM、Lambda、Cognito 与 JWT

# A simple token-based authorizer example to demonstrate how to use an authorization token
# to allow or deny a request. In this example, the caller named 'user' is allowed to invoke
# a request if the client-supplied token value is 'allow'. The caller is not allowed to invoke
# the request if the token value is 'deny'. If the token value is 'unauthorized' or an empty
# string, the authorizer function returns an HTTP 401 status code. For any other token value,
# the authorizer returns an HTTP 500 status code.
# Note that token values are case-sensitive.

import json


def lambda_handler(event, context):
    token = event['authorizationToken']
    if token == 'allow':
        print('authorized')
        response = generatePolicy('user', 'Allow', event['methodArn'])
    elif token == 'deny':
        print('unauthorized')
        response = generatePolicy('user', 'Deny', event['methodArn'])
    elif token == 'unauthorized':
        print('unauthorized')
        raise Exception('Unauthorized')  # Return a 401 Unauthorized response
        return 'unauthorized'
    try:
        return json.loads(response)
    except BaseException:
        print('unauthorized')
        return 'unauthorized'  # Return a 500 error


def generatePolicy(principalId, effect, resource):
    authResponse = {}
    authResponse['principalId'] = principalId
    if (effect and resource):
        policyDocument = {}
        policyDocument['Version'] = '2012-10-17'
        policyDocument['Statement'] = []
        statementOne = {}
        statementOne['Action'] = 'execute-api:Invoke'
        statementOne['Effect'] = effect
        statementOne['Resource'] = resource
        policyDocument['Statement'] = [statementOne]
        authResponse['policyDocument'] = policyDocument
    authResponse['context'] = {
        "stringKey": "stringval",
        "numberKey": 123,
        "booleanKey": True
    }
    authResponse_JSON = json.dumps(authResponse)
    return authResponse_JSON

import json


def lambda_handler(event, context):
    response = {
        "isAuthorized": False,
        "context": {
            "stringKey": "value",
            "numberKey": 1,
            "booleanKey": True,
            "arrayKey": ["value1", "value2"],
            "mapKey": {"value1": "value2"}
        }
    }

    try:
        if (event["headers"]["authorization"] == "secretToken"):
            response = {
                "isAuthorized": True,
                "context": {
                    "stringKey": "value",
                    "numberKey": 1,
                    "booleanKey": True,
                    "arrayKey": ["value1", "value2"],
                    "mapKey": {"value1": "value2"}
                }
            }
            print('allowed')
            return response
        else:
            print('denied')
            return response
    except BaseException:
        print('denied')
        return response

import boto3
import requests
from requests.auth import AuthBase
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

class BotoSigV4Auth(AuthBase):
    """为 HTTP 请求创建 AWS Signature V4"""
    def __init__(self, service, region):
        self.service = service
        self.region = region
        self.session = boto3.Session()
        self.credentials = self.session.get_credentials()

    def __call__(self, r):
        aws_request = AWSRequest(method=r.method, url=r.url, data=r.body)
        SigV4Auth(self.credentials, self.service, self.region).add_auth(aws_request)
        r.headers.update(dict(aws_request.headers.items()))
        return r

def main():
    # 配置
    service = 'execute-api'
    region = 'us-east-1'
    api_url = 'https://你的api网关.execute-api.us-east-1.amazonaws.com/你的阶段/你的资源'
    
    # 创建 requests Session
    session = requests.Session()
    session.auth = BotoSigV4Auth(service, region)

    # 发送 GET 请求
    response = session.get(api_url)

    # 打印响应
    print("Response Status Code:", response.status_code)
    print("Response Text:", response.text)

if __name__ == "__main__":
    main()

import requests

url = "https://xuhan.au.auth0.com/oauth/token"

payload = {
    "client_id": "iiptrnicFRTaDduDsWQ6W9WlHm0cdvMp",
    "client_secret": "POQsksHOg3330gITitO4-7B_wYBID8xgMN9-Tz8Asp8R6PbXxSg1vq6De8HoIn7p",
    "audience": "https://auth0-jwt-authorizer",
    "grant_type": "client_credentials"
}

headers = {'content-type': "application/json"}

response = requests.post(url, json=payload, headers=headers)

print(response.text)

import requests
from multiprocessing import Pool

def make_request(url):
    headers = {'x-api-key': 'your key api'}
    response = requests.get(url, headers=headers)
    if response.status_code != 200:
        return f"Request failed. Status: {response.status_code} Response: {response.text}"
    return "Request successful."

def main():
    url = "your url"
    process_count = 100  # 你可以根据需要调整进程数量

    with Pool(process_count) as p:
        results = p.map(make_request, [url] * 10)  # 发送10次请求

    # 打印出所有结果，包括成功和失败的
    for result in results:
        print(result)

if __name__ == '__main__':
    main()