用 Azure Blob Storage 给客户发限时下载链接

用 Azure Blob Storage 给客户发限时下载链接：脚本上传文件到私有容器，生成只读、最长 72 小时的 SAS 链接；本地用 Azurite 模拟器测试，生产环境改用 Entra ID 身份签发用户委托 SAS，不需要账户密钥。

先看清成品

本机在 Docker 中运行 Azurite，4 个测试全部通过：上传后用链接能下载、去掉 SAS 访问被拒绝、超过 72 小时被拒绝、中文文件名的响应头编码正确。第一版因中文文件名报错（见调试示例），已修复。没有连接真实 Azure 存储账户，用户委托 SAS 路径只验证到 DefaultAzureCredential 因无凭证而失败。

注意：Azure 按用量计费，价格以官方价格页和账单为准，本文不列金额。中国大陆的 Azure 由世纪互联运营（azure.cn），账号、门户和终结点（如 *.chinacloudapi.cn）与全球版不同，本文按全球版编写。不要共享账号或把密钥提交到仓库。

隐私提示：链接在有效期内谁拿到都能下载。不要用它发身份证件、合同原件等敏感文件，除非对方明确需要并已约定保护方式。

做完之后你会得到什么？

需要准备哪些环境？

文件结构

22-azure-C-src/
├── share.py
└── test_share.py

第 1 步：启动 Azurite

docker run -d --name azr -p 10000:10000 mcr.microsoft.com/azure-storage/azurite \
  azurite-blob --blobHost 0.0.0.0 --skipApiVersionCheck

--skipApiVersionCheck 避免新版 SDK 的 API 版本比 Azurite 新时被拒绝。

第 2 步：脚本

share.py：

"""上传文件到 Blob 容器，生成限时只读下载链接。
- 设置 AZURE_STORAGE_CONNECTION_STRING：用账户密钥签名（本地 Azurite 测试用）
- 否则用 --account 和 Entra ID 登录身份，生成用户委托 SAS（生产推荐，不需要账户密钥）
"""
import argparse, mimetypes, os, sys
from urllib.parse import quote
from datetime import datetime, timedelta, timezone
from pathlib import Path
from azure.storage.blob import BlobServiceClient, BlobSasPermissions, ContentSettings, generate_blob_sas

MAX_HOURS = 72

def client(account):
    cs = os.environ.get("AZURE_STORAGE_CONNECTION_STRING")
    if cs:
        return BlobServiceClient.from_connection_string(cs)
    from azure.identity import DefaultAzureCredential
    return BlobServiceClient(f"https://{account}.blob.core.windows.net", credential=DefaultAzureCredential())

def disposition(name):
    # HTTP 头只能是 latin-1；中文文件名按 RFC 6266/5987 写成 filename*=UTF-8''...
    fallback = name.encode("ascii", "replace").decode().replace("?", "_").replace('"', "_")
    return f'attachment; filename="{fallback}"; filename*=UTF-8\'\'{quote(name)}'

def share(svc, container, path, hours):
    if not 1 <= hours <= MAX_HOURS:
        raise ValueError(f"有效期应为 1-{MAX_HOURS} 小时")
    p = Path(path)
    cc = svc.get_container_client(container)
    if not cc.exists(): cc.create_container()  # 默认私有，不开放匿名访问
    ctype = mimetypes.guess_type(p.name)[0] or "application/octet-stream"
    with open(p, "rb") as f:
        bc = cc.upload_blob(p.name, f, overwrite=True, content_settings=ContentSettings(
            content_type=ctype, content_disposition=disposition(p.name)))
    now = datetime.now(timezone.utc)
    start, expiry = now - timedelta(minutes=5), now + timedelta(hours=hours)  # start 提前 5 分钟，容忍时钟误差
    common = dict(account_name=svc.account_name, container_name=container, blob_name=p.name,
                  permission=BlobSasPermissions(read=True), start=start, expiry=expiry)
    cred = svc.credential
    if getattr(cred, "account_key", None):
        sas = generate_blob_sas(account_key=cred.account_key, **common)
    else:
        sas = generate_blob_sas(user_delegation_key=svc.get_user_delegation_key(start, expiry), **common)
    return f"{bc.url}?{sas}", expiry

if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("file"); ap.add_argument("--container", default="shares")
    ap.add_argument("--hours", type=int, default=24); ap.add_argument("--account")
    a = ap.parse_args()
    if not os.environ.get("AZURE_STORAGE_CONNECTION_STRING") and not a.account:
        sys.exit("请设置 AZURE_STORAGE_CONNECTION_STRING 或传 --account")
    url, exp = share(client(a.account), a.container, a.file, a.hours)
    print("下载链接：", url)
    print("过期时间（北京时间）：", exp.astimezone(timezone(timedelta(hours=8))).strftime("%Y-%m-%d %H:%M"))

要点：

• 权限只有 read，有效期 1-72 小时。
• 容器用默认私有权限创建。
• start 提前 5 分钟，避免对方电脑时钟稍快导致链接“尚未生效”。

第 3 步：测试

test_share.py（连接串里是 Azurite 官方公开的开发账号密钥，不是真实机密）：

"""需要本地 Azurite：docker run -p 10000:10000 mcr.microsoft.com/azure-storage/azurite azurite-blob --blobHost 0.0.0.0"""
import os, urllib.request, urllib.error, pytest
from azure.storage.blob import BlobServiceClient
import share as m

CS = ("DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;"
      "AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;"
      "BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;")

@pytest.fixture
def svc(): return BlobServiceClient.from_connection_string(CS)

def test_share_roundtrip(svc, tmp_path):
    f = tmp_path / "报价单.txt"; f.write_text("总价以合同为准", encoding="utf-8")
    url, _ = m.share(svc, "shares", f, 1)
    assert "sp=r" in url
    assert urllib.request.urlopen(url).read().decode() == "总价以合同为准"

def test_private_without_sas(svc, tmp_path):
    f = tmp_path / "a.txt"; f.write_text("x")
    url, _ = m.share(svc, "shares", f, 1)
    with pytest.raises(urllib.error.HTTPError):
        urllib.request.urlopen(url.split("?")[0])

def test_hours_limit(svc, tmp_path):
    with pytest.raises(ValueError): m.share(svc, "shares", tmp_path / "x", 100)

def test_disposition_chinese():
    d = m.disposition("报价单.txt")
    d.encode("latin-1")  # 不应抛异常
    assert "filename*=UTF-8''%E6%8A%A5%E4%BB%B7%E5%8D%95.txt" in d

python -m pytest -q

检查结果：4 passed。

第 4 步：本地生成链接

export AZURE_STORAGE_CONNECTION_STRING="（与测试中相同的 Azurite 连接串）"
python share.py 报价单.pdf --hours 2

本机实际输出（签名已省略）：

下载链接： http://127.0.0.1:10000/devstoreaccount1/shares/demo.txt?st=2026-10-21T06%3A27%3A49Z&se=2026-10-21T08%3A32%3A49Z&sp=r&sv=2026-10-06&sr=b&sig=...
过期时间（北京时间）： 2026-10-21 16:32

sp=r 表示只读，se 是 UTC 过期时间。

第 5 步：生产环境（本文未执行）

unset AZURE_STORAGE_CONNECTION_STRING
az login
python share.py 报价单.pdf --account 你的存储账户名 --hours 24

脚本通过 DefaultAzureCredential 使用 az login 的身份，调用 get_user_delegation_key 签发链接，适用于本工具 B 篇那种禁用了共享密钥的存储账户。

调试示例

第一版直接写 content_disposition=f'attachment; filename="{p.name}"'，测试中文文件名时实际报错：

FAILED test_share.py::test_share_roundtrip - UnicodeEncodeError: 'latin-1' co...

HTTP 头只能用 latin-1 字符。改成 RFC 6266 的写法：filename 放 ASCII 备用名，filename*=UTF-8'' 放百分号编码的中文名（见 disposition()），并加了测试 test_disposition_chinese。

出错了怎么办？

模板规则

• 链接只读、有效期尽量短，最长 72 小时。
• 生产用用户委托 SAS，不发放账户密钥。
• 敏感文件不用公开链接分享。

项目资源

限时链接是“既要分享又不想公开”的折中办法。把权限、有效期写死在脚本里，再用 Azurite 跑测试，每次发出去的链接都是可控的。