使用 Python 模擬 AWS API Server

Evan
evan.fang
Published in
16 min readMay 27, 2022

當在使用 aws-cli,或是 boto3 之類的 SDK,與 AWS 互動的時候,實際上是將指令包裝成 RESTful 的 request,送往 AWS 的 Server。

本文將簡單地研究一下,究竟是送出了怎樣的 request,而 AWS 又是回了怎樣的 response。

假設我們在 AWS 的 SQS 上有一個 Queue,叫做 myqueue。

首先,來觀察一下 AWS 是怎麼回應 cli 的 request 的。以 SQS 為例,嘗試用 cli 發出一個 request,並檢視其 response。在指令中加上 --debug 的選項,可以檢視 cli 執行時產生的詳細訊息。

以 get-queue-url 這隻 API 為例:

aws sqs get-queue-url --queue-name myqueue --debug

輸出的 log 非常多,我們去除掉 log 中不重要的部分,觀察以下訊息即可。

aws-cli 準備要送出的 request:

Making request for OperationModel(name=GetQueueUrl) with params: {'url_path': '/', 
'query_string': '',
'method': 'POST',
'headers': {'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8', 'User-Agent': 'aws-cli/2.2.26 Python/3.8.8 Darwin/20.4.0 exe/x86_64 prompt/off command/sqs.get-queue-url'}, 'body': {'Action': 'GetQueueUrl', 'Version': '2012-11-05', 'QueueName': 'myqueue'}, 'url': 'https://sqs.ap-southeast-1.amazonaws.com/',
'context': {'client_region': 'ap-southeast-1',
'client_config': <botocore.config.Config object at 0x7ff27168d4f0>, 'has_streaming_input': False, 'auth_type': None}}

根據 request 內容,aws-cli 計算 API Signature 的過程:

Calculating signature using v4 auth.  CanonicalRequest: POST /  content-type:application/x-www-form-urlencoded; charset=utf-8 host:sqs.ap-southeast-1.amazonaws.com x-amz-date:20220527T022550Z  content-type;host;x-amz-date 1fa485b2bb1e54cfe8824e0716e98e4addbaddb134688d10ae6a1cbf5940f503  StringToSign: AWS4-HMAC-SHA256 20220527T022550Z 20220527/ap-southeast-1/sqs/aws4_request 1088ca9d24efa785c830c44ecf3ff5d2b61f37763a60e0c53dbf8a81161dccbd  Signature: c4f0f66898fa2b27088be7c9b248e60268a2a6075713f82ffcdedf39a417e0ce

以上訊息表明了使用的簽章版本是 v4,依序計算出 CanonicalRequest、StringToSign,然後產生了 Signature。此 Signature 包含了此次 request 的所有必要資訊,AWS Server 收到此 request 後將對此簽章進行驗算,保證發出 request 的人有資格進行此次操作。

關於簽章的算法,可以參考官方的說明

加上簽章等資訊後的 request:

<AWSPreparedRequest 
stream_output=False,
method=POST,
url=https://sqs.ap-southeast-1.amazonaws.com/,
headers={'Content-Type': b'application/x-www-form-urlencoded; charset=utf-8', 'User-Agent': b'aws-cli/2.2.26 Python/3.8.8 Darwin/20.4.0 exe/x86_64 prompt/off command/sqs.get-queue-url', 'X-Amz-Date': b'20220527T022550Z', 'Authorization': b'AWS4-HMAC-SHA256 Credential=AKIAVO7ZJXICLJI7KMEH/20220527/ap-southeast-1/sqs/aws4_request, SignedHeaders=content-type;host;x-amz-date, Signature=c4f0f66898fa2b27088be7c9b248e60268a2a6075713f82ffcdedf39a417e0ce', 'Content-Length': '67'}>

從 AWS 收到的 response:

Response headers:
{ 'x-amzn-RequestId': '9914a4c9-1ca0-5c63-9f98-ffc3244f94ec', 'Date': 'Fri, 27 May 2022 02:25:51 GMT', 'Content-Type': 'text/xml', 'Content-Length': '344' }
Response body:
b'<?xml version="1.0"?> <GetQueueUrlResponse xmlns="http://queue.amazonaws.com/doc/2012-11-05/"> <GetQueueUrlResult> <QueueUrl>https://sqs.ap-southeast-1.amazonaws.com/375795595780/myqueue</QueueUrl> </GetQueueUrlResult> <ResponseMetadata> <RequestId>9914a4c9-1ca0-5c63-9f98-ffc3244f94ec</RequestId> </ResponseMetadata> </GetQueueUrlResponse>'

可以發現,我們拿到的是一個 xml 格式的 response。

有了以上資訊,可以試著實作一個 “假” 的 AWS API Serve。

以使用 Flask 為例:

其中的 xml 的 response 是直接使用了 “正常” 的 AWS Server 的 response。要注意如果 xml 的欄位有漏失,cli 或 SDK 是無法正確解析的。

啟動 Flask app:

export FLASK_APP=server
export FLASK_ENV=development
flask run --host=0.0.0.0 --port=8866

然後試著用 aws-cli 打我們自己的 app:

aws sqs get-queue-url --queue-name myqueue --endpoint-url http://localhost:8866
就像是從真正的 AWS 拿到的 SQS QUEUE URL

也可以使用 boto3 來試一下,也是沒問題,記得要指定

如果想看到詳細的 log 輸出,可以將上面程式碼中的 set_stream_logger 的註解取消,就會看到如 aws-cli debug mode 下的 log 。

接著,考慮一下錯誤處理的部分。

當 Queue 不存在時,AWS 會給出什麼回應?

aws sqs get-queue-url --queue-name oops --profile stranger --debug

得到一個 400 的 response,error message 為 「The specified queue does not exist for this wsdl version.」

https://sqs.ap-southeast-1.amazonaws.com:443 "POST / HTTP/1.1" 400 333 botocore.parsersResponse headers: {'x-amzn-RequestId': '4f8e03bb-b7f4-5763-a042-58b1f749f1b0', 'Date': 'Fri, 27 May 2022 02:59:09 GMT', 'Content-Type': 'text/xml', 'Content-Length': '333'}Response body: b'<?xml version="1.0"?><ErrorResponse xmlns="http://queue.amazonaws.com/doc/2012-11-05/"><Error><Type>Sender</Type><Code>AWS.SimpleQueueService.NonExistentQueue</Code><Message>The specified queue does not exist for this wsdl version.</Message><Detail/></Error><RequestId>4f8e03bb-b7f4-5763-a042-58b1f749f1b0</RequestId></ErrorResponse>'

接著,測試一下若使用了錯誤的 ACCESS KEY,AWS 會做出什麼回應。建立一個名為 stranger 的假 profile 在 aws credential 中:

[stranger]
aws_access_key_id = IRJOKZGUBHQQCUUXHXBK
aws_secret_access_key = hbop0MQb4xIrA1OIucObKq8ARUyCE8CAiRHMKp3g region = ap-northeast-1

試著用這個不存在的身份,來呼叫 AWS API:

aws sqs get-queue-url --queue-name myqueue --profile stranger --debug

得到一個 403 response,error message 是「 The security token included in the request is invalid.」

https://sqs.ap-northeast-1.amazonaws.com:443 "POST / HTTP/1.1" 403 311Response headers: {'x-amzn-RequestId': '07a308a1-7936-5ac6-b3a2-53ab3dd6fc1b', 'Connection': 'close', 'Date': 'Fri, 27 May 2022 03:16:48 GMT', 'Content-Type': 'text/xml', 'Content-Length': '311'} Response body: b'<?xml version="1.0"?><ErrorResponse xmlns="http://queue.amazonaws.com/doc/2012-11-05/"><Error><Type>Sender</Type><Code>InvalidClientTokenId</Code><Message>The security token included in the request is invalid.</Message><Detail/></Error><RequestId>07a308a1-7936-5ac6-b3a2-53ab3dd6fc1b</RequestId></ErrorResponse>'

接著,把 stranger 的 aws_access_key_id 更換成一把真實存在於 AWS 上的 KEY,來測試一下當 KEY 存在,但 SECRET 不正確的時候,AWS 會吐出什麼錯誤訊息。

https://sqs.ap-northeast-1.amazonaws.com:443 "POST / HTTP/1.1" 403 905Response headers: {'x-amzn-RequestId': 'a6d4cd2d-f24c-579b-9b0c-9627b719c86c', 'Connection': 'close', 'Date': 'Fri, 27 May 2022 03:18:45 GMT', 'Content-Type': 'text/xml', 'Content-Length': '905'} Response body: b'<?xml version="1.0"?><ErrorResponse xmlns="http://queue.amazonaws.com/doc/2012-11-05/"><Error><Type>Sender</Type><Code>SignatureDoesNotMatch</Code><Message>The request signature we calculated does not match the signature you provided. Check your AWS Secret Access Key and signing method. Consult the service documentation for details.\n\nThe Canonical String for this request should have been\n\'POST\n/\n\ncontent-type:application/x-www-form-urlencoded; charset=utf-8\nhost:sqs.ap-northeast-1.amazonaws.com\nx-amz-date:20220527T031845Z\n\ncontent-type;host;x-amz-date\n1fa485b2bb1e54cfe8824e0716e98e4addbaddb134688d10ae6a1cbf5940f503\'\n\nThe String-to-Sign should have been\n\'AWS4-HMAC-SHA256\n20220527T031845Z\n20220527/ap-northeast-1/sqs/aws4_request\ndee3ce93987847caacd92bf69dd7ab66dd32ecb65c6066e58b0ee773b499d88d\'\n</Message><Detail/></Error><RequestId>a6d4cd2d-f24c-579b-9b0c-9627b719c86c</RequestId></ErrorResponse>'

一樣得到 403 的錯誤,但錯誤訊息是「The request signature we calculated does not match the signature you provided. Check your AWS Secret Access Key and signing method. Consult the service documentation for details.」;另外還附上的 AWS 算出的簽章,供請求者核對。

有了以上資訊,可以進一步完善我們的 “假” AWS Server 了。

首先,將會用到的 “假” response xml 都集中到另外的 python file 中:

從上到下,分別是 ACCESS KEY 不正確、SECRET 不正確、Queue 找不到、Queue 找到了,四種 get_queue_url 可能會看到的訊息內容。

稍加改良後的版本如下:

我們定義了一個 “假” Queue,以及一組寫死的 credential:

queues = [{"name": "q1", "url": "http://sqs.aws.com/001/q1"}]ACCESS_KEY = "test_key"
ACCESS_SECRET = "test_secret"

在收到 request 的時候,會從 header 中取出 access key id,並檢查系統中是否存在該 key,若不存在,就按照 AWS 的格式,丟出 403 error response:

credential=request.headers["Authorization"].split("Credential=")[1]
aws_access_key_id=credential.split("/")[0]
if aws_access_key_id != ACCESS_KEY:
return Response(INVALID_KEY_ERROR,mimetype="text/xml",status=403)

然後從 header 中取出 signature,並檢查是否正確:

signature = request.headers["Authorization"].split("Signature=")[1]     if signature != _calculate_signature(request):
# return error: signature not match
pass

但由於計算簽章的步驟較為冗長及複雜,並非本文重點,故暫且略過。如需要了解計算的過程,可以參考完整的程式碼:Github

在 Queue 找不到的時候,會回覆 400 error response;反之,則回覆 Queue 的 URL:

if len(queue) == 0:
return Response(NOT_FOUND_ERROR,mimetype="text/xml",status=400)
return Response(GET_QUEUE_URL_RESULT,mimetype="text/xml")

由於 server 端加上了 credential 的檢查,client 端也需要加上對應的 key 以及 secret 才行:

至此,我們已經模擬出一個相當陽春的 AWS API Server。

--

--