Azure Blob存储SAS URL分块上传视频的安全方案及代码咨询

需在Azure Blob容器中为用户及其视频创建存储路径，生成对应视频Blob的SAS URL，通过Python PUT请求分块上传大视频。目前已实现SAS Token及URL生成（代码如下），但对Azure Python SDK中的BinaryClient、BinaryService等组件存在困惑，希望获取最安全的标准实现方法；请检查现有SAS生成代码的正确性，并指导如何通过PUT请求分块上传、处理状态码及最终合并分块。

现有SAS Token生成代码 #

def sas_token_genetation(container_name, blob_name):
    connection_string = f"DefaultEndpointsProtocol=https;AccountName={container_name};AccountKey={account_key};EndpointSuffix=core.windows.net"
    blob_service_client = BlobServiceClient.from_connection_string(connection_string)
    sas_token = generate_blob_sas(
        account_name=blob_service_client.account_name,
        container_name=container_name,
        blob_name=blob_name,
        account_key=account_key,
        permission=BlobSasPermissions(read=True, write=True, update = True, delete= True, list = True, add=True),
        expiry=datetime.utcnow() + timedelta(hours=1)
    )
    return sas_token

现有SAS URL生成代码 #

def sas_url(account_name : str, container_name:str , blob_name : str):
    sas_token = sas_token_genetation(container_name, blob_name)
    connection_string = f"DefaultEndpointsProtocol=https;AccountName={container_name};AccountKey={account_key};EndpointSuffix=core.windows.net"
    blob_service_client = BlobServiceClient.from_connection_string(connection_string)
    sas_url = f"https://{account_name}.blob.core.windows.net/{container_name}/{blob_name}"
    return sas_url

一、现有SAS代码问题检查与修正 #

1. 连接字符串参数错误：现有代码中AccountName误用了container_name，应替换为实际存储账户名而非容器名。
2. SAS URL未拼接Token：生成的URL未附加SAS Token，导致无法通过URL完成授权操作。
3. 权限过度授予：当前请求的权限包含delete、list等非必要权限，分块上传仅需write、add、read（可选，用于验证结果），遵循最小权限原则更安全。

修正后的代码：

from azure.storage.blob import BlobServiceClient, generate_blob_sas, BlobSasPermissions
from datetime import datetime, timedelta

# 建议通过环境变量加载，避免硬编码敏感信息
account_name = "你的存储账户名"
account_key = "你的存储账户密钥"

def sas_token_generation(container_name, blob_name):
    # 修正连接字符串的AccountName参数
    connection_string = f"DefaultEndpointsProtocol=https;AccountName={account_name};AccountKey={account_key};EndpointSuffix=core.windows.net"
    blob_service_client = BlobServiceClient.from_connection_string(connection_string)
    
    # 配置分块上传所需最小权限
    sas_token = generate_blob_sas(
        account_name=blob_service_client.account_name,
        container_name=container_name,
        blob_name=blob_name,
        account_key=account_key,
        permission=BlobSasPermissions(write=True, add=True, read=True),
        expiry=datetime.utcnow() + timedelta(hours=1)
    )
    return sas_token

def sas_url(container_name: str, blob_name: str):
    sas_token = sas_token_generation(container_name, blob_name)
    # 拼接SAS Token到URL末尾，生成可直接使用的授权URL
    sas_url = f"https://{account_name}.blob.core.windows.net/{container_name}/{blob_name}?{sas_token}"
    return sas_url

二、Azure Python SDK组件说明 #

Azure官方azure-storage-blob SDK中并无BinaryClient或BinaryService这类公开组件，你可能混淆了其他SDK的命名。标准分块上传推荐使用SDK内置的BlobClient.stage_block和BlobClient.commit_block_list方法，或直接调用REST API的PUT分块接口，前者封装了错误处理、重试逻辑，更简洁可靠。

三、标准分块上传实现（两种方案） #

方案1：使用Azure Python SDK（推荐） #

SDK已封装分块上传的全流程，无需手动处理HTTP状态码和分块合并：

def upload_large_video_via_sdk(local_file_path, container_name, blob_name):
    connection_string = f"DefaultEndpointsProtocol=https;AccountName={account_name};AccountKey={account_key};EndpointSuffix=core.windows.net"
    blob_client = BlobServiceClient.from_connection_string(connection_string).get_blob_client(container=container_name, blob=blob_name)
    
    # 分块上传：SDK自动处理分块大小、重试、合并
    with open(local_file_path, "rb") as data:
        blob_client.upload_blob(
            data,
            blob_type="BlockBlob",
            chunk_size=4*1024*1024,  # 4MB分块，可根据网络调整，最大支持100MB
            overwrite=True
        )
    print(f"视频 {blob_name} 上传完成")

方案2：手动调用PUT REST API分块上传 #

若需手动实现PUT请求分块上传，步骤如下：

1. 生成带权限的SAS URL；
2. 将本地文件分割为16KB~100MB之间的分块；
3. 逐个上传分块到指定URL；
4. 提交块列表完成合并。

代码示例：

import requests
import base64
import os

def upload_large_video_via_put(local_file_path, container_name, blob_name):
    sas_url = sas_url(container_name, blob_name)
    chunk_size = 4 * 1024 * 1024  # 4MB分块
    block_ids = []
    
    with open(local_file_path, "rb") as f:
        block_index = 0
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            
            # 生成Base64编码的唯一块ID
            block_id = base64.b64encode(f"block_{block_index}".encode()).decode()
            block_ids.append(block_id)
            
            # 上传单个分块
            block_url = f"{sas_url}&comp=block&blockid={block_id}"
            response = requests.put(block_url, data=chunk)
            
            # 处理分块上传状态码
            if response.status_code != 201:
                raise Exception(f"分块 {block_index} 上传失败，状态码：{response.status_code}")
            
            block_index += 1
    
    # 构造块列表XML，提交合并请求
    block_list_xml = f"""<?xml version="1.0" encoding="utf-8"?>
<BlockList>
{"".join([f"<Latest>{block_id}</Latest>" for block_id in block_ids])}
</BlockList>"""
    
    merge_url = f"{sas_url}&comp=blocklist"
    response = requests.put(merge_url, data=block_list_xml, headers={"Content-Type": "application/xml"})
    
    # 处理合并状态码
    if response.status_code == 201:
        print(f"视频 {blob_name} 上传并合并完成")
    else:
        raise Exception(f"分块合并失败，状态码：{response.status_code}")

四、关键状态码说明 #

• 分块上传：成功返回201 Created；403 Forbidden需检查SAS权限或过期时间；400 Bad Request需检查块大小是否合法或块ID格式错误。
• 分块合并：成功返回201 Created；409 Conflict表示块列表中存在无效块ID；411 Length Required表示未指定请求体长度。

内容的提问来源于stack exchange，提问作者CloudExplorer