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

我正在使用 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。

请您参考如下方法:

您可以在 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)