柚子快報邀請碼778899分享：APIGateway的認證

http://yzkb.51969.com/

APIGateway的支持的認證如下：

我們從表格中可以看到，HTTP API 不支持資源策略的功能，另外是通過JWT的方式集成Cognito的。

對于REST API則是沒有顯示說明支持JWT認證，這個我們可以通過Lambda 自定義的方式來實現(xiàn)。

所以按照這個說法，除了資源策略，各種認證方式HTTP API和REST API 都能夠實現(xiàn)。

資源策略

note：HTTP API沒有資源策略，所以這個部分都都是關于REST API的。

先談資源策略，因為這個是兩個API唯一不同的地方。

資源策略默認為空，對于公有API來是完全放開的，但是如果寫了任意一條策略，那么其他的策略都會變成Deny，但是對于私有API來說，沒有資源策略則意味著完全私有。

下面是三種資源策略：

允許特定的賬戶訪問APIGateway，因為訪問人是賬戶，所以這個就需要開啟IAM驗證。

{

"Version": "2012-10-17",

"Statement": [

{

"Effect": "Allow",

"Principal": {

"AWS": [

"arn:aws:iam::{{otherAWSAccountID}}:root",

"arn:aws:iam::{{otherAWSAccountID}}:user/{{otherAWSUserName}}",

"arn:aws:iam::{{otherAWSAccountID}}:role/{{otherAWSRoleName}}"

]

},

"Action": "execute-api:Invoke",

"Resource": [

"execute-api:/{{stageNameOrWildcard*}}/{{httpVerbOrWildcard*}}/{{resourcePathOrWildcard*}}"

]

}

]

}

基于IP的訪問策略如下：。

{

"Version": "2012-10-17",

"Statement": [

{

"Effect": "Deny",

"Principal": "*",

"Action": "execute-api:Invoke",

"Resource": "execute-api:/{{stageNameOrWildcard}}/{{httpVerbOrWildcard}}/{{resourcePathOrWildcard}}",

"Condition" : {

"IpAddress": {

"aws:SourceIp": [ "{{sourceIpOrCIDRBlock}}", "{{sourceIpOrCIDRBlock}}" ]

}

}

},

{

"Effect": "Allow",

"Principal": "*",

"Action": "execute-api:Invoke",

"Resource": "execute-api:/{{stageNameOrWildcard}}/{{httpVerbOrWildcard}}/{{resourcePathOrWildcard}}"

}

]

}

允許來自來自特定VPC的流量：

{

"Version": "2012-10-17",

"Statement": [

{

"Effect": "Deny",

"Principal": "*",

"Action": "execute-api:Invoke",

"Resource": "execute-api:/{{stageNameOrWildcard}}/{{httpVerbOrWildcard}}/{{resourcePathOrWildcard}}",

"Condition": {

"StringNotEquals": {

"aws:sourceVpc": "{{vpcID}}"

}

}

},

{

"Effect": "Allow",

"Principal": "*",

"Action": "execute-api:Invoke",

"Resource": "execute-api:/{{stageNameOrWildcard}}/{{httpVerbOrWildcard}}/{{resourcePathOrWildcard}}"

}

]

}

Lambda授權方

Lambda的認證方式可以自定義認證的方式，以下是一個官方提供的RSET API認證的例子，當然也在這個代碼中實現(xiàn)JWT的頒發(fā)認證，以及SSO中我們常用的Oauth，SAML和OIDC協(xié)議 。

REST API

# 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

這個是官方文檔的一張圖： 也就是說當訪問APIGatewa有的時候會帶上一個憑證，然后這個憑證會被傳遞到負責驗證的Lambda中，這個lambda會根據傳遞的請求頭會返回allow或者deny的資源策略，或者unauthorized的異常。

在Header中添加{Authorization: allow}，是可以請求成功的。

在Header中添加{Authorization: deny}，可以按照預期攔截。 在Header中添加未認證的token，即不在黑白名單內的，報錯500 符合預期

HTTP API

然后我們再來看HTTP API，由于沒有資源策略，所以授權方函數的代碼和之前不一樣。授權方的配置如下：

再來看一看代碼，

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

在Header中添加{Authorization: secretToken}，是可以請求成功的。

如果在請求的時候沒有添加Authorization的Header，這個時候是返回401 “message”: “Unauthorized”，由于我們沒有在請求的時候帶入身份，所以會返回401。

如果傳遞的token不對，那么會報錯403 “message”: “Forbidden”。即我們傳遞了一個token到后端，但是沒有通過認證，也就說沒有對應的權限。

Cognito 授權

note：

由于中國區(qū)沒有Cognito用戶池，所以此功能在中國區(qū)不可用。 REST API 提供了直接集成Cognito的方式，對于HTTP API而言可以使用JWT的方式來支持Cognito。

下面啟動一個Cognito用戶池，由于是簡單測試，所以沒有集成第三方身份提供商。

接下來設置密碼策略并且關閉MFA，然后下一步直接到 Step 5 Integrate your app。

使用托管UI并且設置Cognito domain的URL，以及回調URL。隨后我們通過內置的Cognito UI登錄，會調轉到我們設置的回調函數，同時帶著我們需要的憑證。

接下來是創(chuàng)建用戶，我們接下來要我們使用這個用戶登錄Cognito UI。

然后編輯托管UI的配置選擇Implicit grant

登錄之后會跳轉到我們的設置的回調函數，同時會返回id_token,access_token。 設置Cognito授權方，選擇前面創(chuàng)建好的Cognito用戶池，然后設置加上請求頭Authorization。

填寫之后可以測試，使用前面回調返回的id_token,這里測試之后，重新部署API 然后再使用Postman再次測試。

Postman的設置如下，這里添加了請求頭{ Authorization： }

IAM 授權

IAM 認證比較特殊，對于中國區(qū)而言，如果你沒有備案，那么只能使用IAM認證的方式進行認證。

這里其實就是SignV4的算法，我們可以使用Postman來做簽名，如下：

如果你的應用需要使用SignV4訪問API使用代碼：

import boto3

import requests

from requests.auth import AuthBase

from botocore.auth import SigV4Auth

from botocore.awsrequest import AWSRequest

class BotoSigV4Auth(AuthBase):

"""為 HTTP 請求創(chuàng)建 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/你的階段/你的資源'

# 創(chuàng)建 requests Session

session = requests.Session()

session.auth = BotoSigV4Auth(service, region)

# 發(fā)送 GET 請求

response = session.get(api_url)

# 打印響應

print("Response Status Code:", response.status_code)

print("Response Text:", response.text)

if __name__ == "__main__":

main()

JWT 授權

note： 這個部分屬于HTTP API的認證，REST API

由于OIDC協(xié)議使用JWT作為中間憑證，所以在這里可以使用Auth0來代替JWT的頒發(fā)商。配置如下：

在Applications - APIS中新建API： 然后這個時候，auth0 會自動生成一個Application，后續(xù)我們會使用這個Application的Client ID和Secret ID以及Domain的信息來登錄。

也就是說這三個信息確定了一個身份池，然后符合規(guī)則的用戶可以通過這個身份池來換取JWT?？梢栽贏pplications-Applications 中看到。

配置好之后，可以通過Auth0的API來拿到登錄后的JWT，以下是一個官方給的教程可以用來測試功能，當然也可以集成到APP中。

APIGateway 的Authorization 配置如下：

auth0 也提供了實例代碼供我們測試：

官方提供的代碼很爛，這個功能完全可以使用requests來實現(xiàn)，代碼如下：

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)

然后可以使用Postman來進行驗證，其實就是在請求頭中加上了Authorization: Bearer < your JWT>，這樣是可以通過客戶端加上憑證范訪問APIGateway.

APIKEY

APIKEY本來是用來做限流的功能，比如說某個服務會提供API給開發(fā)者使用，但同時又不希望開發(fā)者濫用這樣的憑證，所以才有了這個功能。很多人會把這個當成限制匿名用戶的一部分，雖然這樣的解釋沒有問題，但是APIKEY的作用仍然是做限流而不是認證。

對于REST API來說APIKEY通常與使用計劃關聯(lián)，然后再再特定的路由中啟動APIKEY。在使用計劃中寫明Burst limit和Rate limit，以及每天或者每月的額度。然后在請求頭中帶上x-api-key: your apikey

速率限制（Burst limit） ：設計用來控制較長時間尺度（如每秒）內的平均請求量，確保服務的穩(wěn)定性和可靠性，防止 API 被過度使用。

突發(fā)限制（Rate limit）： 設計用來處理短時間內的高流量突發(fā)，允許在極短的時間窗口內接受較多請求，但不應持續(xù)太久，以避免服務器資源被迅速耗盡。

使用Python多進程測試代碼如下，實測達到任意限制都會報錯 429 {“message”:“Too Many Requests”}

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) # 發(fā)送10次請求

# 打印出所有結果，包括成功和失敗的

for result in results:

print(result)

if __name__ == '__main__':

main()

對于HTTP API 來說則是直接在階段設置就可以，同樣可以達到限流的效果。

最后關于中國區(qū)

中國區(qū)有一個特殊的流程叫做ICP備案, 如果沒有進行備案的話，那么無論是公網訪問還是內網訪問，都會遇到如下的401報錯

{

"message": null

}

當然如果使用自定義域名的話，那么域名也需要備案。

柚子快報邀請碼778899分享：APIGateway的認證

http://yzkb.51969.com/

參考閱讀