python - 使用 Boto3 使用自定义日志记录信息为 get_object 创建一个预签名的 S3 URL？

Question

我正在使用 Boto3 和 s3_client.generate_presigned_url 创建一个预签名的 get_object url，即

response = s3_client.generate_presigned_url('get_object',
                                            Params={'Bucket': bucket_name,
                                                    'Key': object_name},
                                            ExpiresIn=expiration)

根据the Boto 3 documentation .

我需要将请求 URL 的用户的身份添加到预签名的 URL 本身中，这样就可以立即看出使用了谁的凭据来生成它!现在 AWS Documentation表示可以在 URL 中包含自定义查询字符串参数:

You can include custom information to be stored in the access log record for a request by adding a custom query-string parameter to the URL for the request. Amazon S3 ignores query-string parameters that begin with "x-", but includes those parameters in the access log record for the request, as part of the Request-URI field of the log record. For example, a GET request for s3.amazonaws.com/awsexamplebucket/photos/2019/08/puppy.jpg?x-user=johndoe works the same as the same request for s3.amazonaws.com/awsexamplebucket/photos/2019/08/puppy.jpg, except that the x-user=johndoe string is included in the Request-URI field for the associated log record. This functionality is available in the REST interface only.

Javascript SDK 库在 Github issues 中有以下配方:

var req = s3.getObject({Bucket: 'mybucket', Key: 'mykey'});
req.on('build', function() { req.httpRequest.path += '?x-foo=value'; });
console.log(req.presign());

用于创建预签名的 url，并将 ?x-foo=value 嵌入到 URL 中。从评论来看，它似乎也有效!

如何在 Python (3) 中实现同样的效果，最好(但不一定)使用 Boto 3？

附言请注意，我不是在问如何强制客户端传递 header 或类似内容 - 事实上我无法控制客户端，我只是给它提供 URL。

Answer 1

您可以在 boto3 和 botocore events 中使用类似的方法.感兴趣的事件是 "provide-client-params.s3.GetObject" 和 "before-sign.s3.GetObject"。 provide-client-params处理程序可以修改 API 参数和上下文，并且 before-sign 接收 AWSRequest 进行签名，因此我们可以将参数注入(inject) URL。

import boto3

from botocore.client import Config
from urllib.parse import urlencode

# Ensure signature V4 mode, required for including the parameters in the signature
s3 = boto3.client("s3", config=Config(signature_version="s3v4"))

def is_custom(k):
    return k.lower().startswith("x-")

def client_param_handler(*, params, context, **_kw):
    # Store custom parameters in context for later event handlers
    context["custom_params"] = {k: v for k, v in params.items() if is_custom(k)}
    # Remove custom parameters from client parameters,
    # because validation would fail on them
    return {k: v for k, v in params.items() if not is_custom(k)}

def request_param_injector(*, request, **_kw):
    if request.context["custom_params"]:
        request.url += "&" if "?" in request.url else "?"
        request.url += urlencode(request.context["custom_params"])

# NOTE: The logic can be included with other client methods as well by simply
# dropping ".GetObject" from the event names
s3.meta.events.register("provide-client-params.s3.GetObject", client_param_handler)
s3.meta.events.register("before-sign.s3.GetObject", request_param_injector)

有了事件处理程序，传递自定义参数很简单，只需将它们包含在 Params 字典中即可:

response = s3.generate_presigned_url(
    'get_object',
    Params={'Bucket': bucket_name,
            'Key': object_name,
            'x-foo': 'value'},
    ExpiresIn=expiration)