使用 Boto3 为 get_object 创建带有自定义日志记录信息的预签名 S3 URL？

Question

我正在使用 Boto3 并s3_client.generate_presigned_url创建一个预签名get_objecturl，即

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

根据Boto 3 文档。

我需要将请求 URL 的用户身份添加到预签名 URL 本身中，以便立即显而易见谁的凭据用于生成它！现在AWS 文档表示可以在 URL 中包含自定义查询字符串参数：

您可以通过向请求的 URL 添加自定义查询字符串参数来包含要存储在请求的访问日志记录中的自定义信息。Amazon S3 会忽略以“x-”开头的查询字符串参数，但会将这些参数包含在请求的访问日志记录中，作为日志记录的 Request-URI 字段的一部分。例如， GET 请求的s3.amazonaws.com/awsexamplebucket/photos/2019/08/puppy.jpg?x-user=johndoe工作方式与 的相同请求相同s3.amazonaws.com/awsexamplebucket/photos/2019/08/puppy.jpg，只是x-user=johndoe字符串包含在Request-URI关联日志记录的字段中。此功能仅在 REST 接口中可用。

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());

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

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

PS 请注意，我并不是问如何强制客户端传递标头或任何类似内容 - 事实上我无法控制客户端，我只是向它提供 URL。

Answer 1

boto3您可以在with botocoreevents中使用类似的方法。感兴趣的事件是"provide-client-params.s3.GetObject"和"before-sign.s3.GetObject"。Provide -client-params处理程序可以修改 API 参数和上下文，并且 before-sign 接收 to Sign 等信息AWSRequest，因此我们可以将参数注入 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)