Azure Blob：可审阅的Python条件传输工程
前置是前四课的对象、权限与条件语义。目标是把真实Python SDK请求映射到人工可检查的结果。示例未执行；作者没有安装依赖、导入SDK、启动Azurite或服务、连接账号、发起SDK/API请求。下面命令只供未来读者在自己的授权隔离环境执行。
Azure Blob 工程前置与输入
Azure Blob 工程前置与输入：读者自有账号与私有容器必须真实存在，数据角色与网络可达性必须预先满足。工程固定Python 3.10以上、azure-storage-blob 12.25.1、azure-identity 1.21.0和azure-core 1.33.0，使用Entra身份链；空配置故意不能运行成功，没有演示token或模拟后端。基础路径要求HNS关闭的标准通用v2账号。
把模板复制成config.local.json，填写真实HTTPS Blob根地址、容器及全新随机run_id；输入文件相对工程根目录，长度限制1 byte至1 MiB。不得指向生产对象或复用旧run_id。身份配置由读者管理员完成，管理角色不替代数据权限；DefaultAzureCredential不负责授予角色。工程不创建账号、容器、RBAC或保护策略。README给出依赖安装与三条读者自跑命令，本轮没有执行。
Azure Blob 工程请求顺序
Azure Blob 工程请求顺序：先创建唯一对象，再条件更新，故意用旧ETag制造冲突；最后暂存固定块并条件提交，下载比对原输入。每一步都调用真实SDK，异常会停止后续流程，不能用布尔值代替服务结果。state.local.json在首个写请求前记录精确目标，响应丢失时仍知道需要人工检查的名称。
overwrite=False保护首次创建，第二次写以首个ETag作为If-Match；旧ETag冲突只接受412。stage_block逐块上传，commit_block_list给出有序列表并带当前ETag，防止并发覆盖。下载逐字节与输入比较后才打印VERIFIED。工程不为恢复演示删除历史版本，也不为“成功输出”吞掉身份或权限错误。
Azure Blob 工程预期与清理
Azure Blob 工程预期与清理：成功预期包含EXPECTED_CONFLICT与VERIFIED，失败预期必须区分403、404、412和结果未知。所有预期均为人工推导，不是本轮运行证据。若返回上传冲突而不是预期步骤，不应改成无条件覆盖；若下载内容不符，保留对象与state调查，不将其当作可重试成功。
cleanup只根据同一state删除当前的一个精确名称对象，核对exercise_run元数据与ETag；从不删除容器、账号、前缀下其他对象、历史版本或快照。快照存在可能使删除失败，须人工审阅；软删除保留与历史版本继续计费。首次上传响应丢失时created=false，工程拒绝自动清理。元数据不防同权限攻击者，隔离容器是本例必要假设。
Azure Blob 完整同源工程
Azure Blob 完整同源工程：源码、依赖、配置、输入和README都从examples/p2-azure-blob直接包含，同一份文件用于阅读和未来读者自跑。版本锁定约束评审对象，静态解析只证明语法与文档目标可读，不证明真实认证、SDK兼容、网络、费用或恢复结果。以下源码不是简化伪代码。
参考：工程说明与读者步骤
markdown
# Azure Blob 条件传输工程
示例未执行。作者只做源码/配置/AST静态核对，没有安装依赖、导入Azure SDK、访问账号或运行模拟器。下列是未来读者在自己隔离环境的操作说明，不是本次运行记录。依赖固定为azure-storage-blob 12.25.1、azure-identity 1.21.0、azure-core 1.33.0；Python要求3.10或以上。版本锁定便于评审，并不保证适合所有生产系统。
## 真实前置与输入
读者自有真实Azure订阅中的标准通用v2存储账号、现存私有容器、可达的HTTPS Blob endpoint及获授权身份必不可少。基础练习使用HNS关闭的账号；版本/软删除由账号管理员预先按恢复目标设置，工程不修改账号保护策略。不要使用生产容器。容器只允许本练习操作者写入，避免其他写入者伪造相同元数据；元数据是误操作防护，不是安全隔离边界。
身份须在目标容器作用域具有Storage Blob Data Contributor等允许读、写、删、列举的**数据平面**权限。订阅Owner/Contributor管理权限不自动等于数据访问权；角色传播可能有延迟。DefaultAzureCredential按链选择身份，读者需事先配置自己的服务主体环境变量AZURE_TENANT_ID、AZURE_CLIENT_ID、AZURE_CLIENT_SECRET，或使用已授权托管身份/开发凭据。不要把秘密写入配置或提交；如使用托管身份应按部署环境缩窄credential chain。身份登录、角色配置、网络防火墙/私有DNS由读者环境管理员完成，本工程不执行这些管理操作。不提供假账号、假token或默认成功模式。
复制config.example.json为config.local.json。填写自己真实的account_url（HTTPS账号Blob根地址，无路径/查询串）、现存container与新的随机run_id（16至40位小写字母数字，绝不复用）。空模板故意不能成功。input相对于本工程目录；默认UTF-8文件在input/message.txt。block_bytes=16是为了让短文本也形成多个真实块，生产应按内存/吞吐量调优。读者可用独立虚拟环境安装requirements.txt；以下命令仅供读者将来执行。
```powershell
python -m venv .venv
.\.venv\Scripts\python -m pip install -r requirements.txt
.\.venv\Scripts\python app.py run --config config.local.json
.\.venv\Scripts\python app.py inspect --config config.local.json
.\.venv\Scripts\python app.py cleanup --config config.local.json
```
## 手工预期与失败分支
成功预期：run创建唯一名称p2-azure-blob/{run_id}/document.txt，以首次ETag更新一次；再用旧ETag写入必须打印EXPECTED_CONFLICT status=412；显式stage_block与commit_block_list后下载内容必须与原输入逐字节相同，打印VERIFIED bytes与sha256。摘要仅是本地比对证据，不是服务端提供的备份证明。run保留对象和state.local.json供inspect观察，列出的版本数取决于账号版本策略。
失败预期：空配置在本地校验失败；身份缺失抛CredentialUnavailableError/ClientAuthenticationError；无数据角色或网络规则可能403；容器不存在404；已存在对象因overwrite=False返回冲突，不得改成overwrite=True掩盖。ETag过期412须重新读取并由业务决定合并。活动lease可能导致409/412，本例不抢占lease。版本列举权限不足等失败也应如实保留异常。SDK会对可重试故障进行有限重试，响应丢失不能由异常断言“从未写入”。若首次上传完成但响应丢失，state.created仍为false，清理拒绝：请核对精确名称/账号请求日志/内容与所有权后人工处理，不自动接管。
输入是1 byte至1 MiB，本工程有意限制读入内存的练习范围；若小块配置使块数超过50000，程序在远程写入前拒绝，应增大block_bytes。练习中上传、下载、列举、删除产生真实请求和可能费用。程序不包含预制成功输出、不模拟SDK、不捕获异常后假装通过。
## 恢复与清理边界
cleanup只删除state记录的**一个当前Blob**，先核对account/container/run_id/精确名称和exercise_run元数据，再以读取的ETag条件删除。它不删除容器、账号、其他前缀对象，也不清理版本或快照；存在快照导致删除失败时保留异常供人工核对，不自动增加delete_snapshots='include'。本例不创建快照，但外部策略或操作者可能创建。软删除/版本化开启时，历史对象仍保留并计费，须按读者账号保留策略处理。state保留作人工核对，重新练习应先确认当前对象和历史残留，再存档该state并使用全新run_id。
读取某version_id或snapshot验证内容，可在独立恢复流程中选择；不要直接用旧ETag覆盖现有当前对象。容器删除需走容器软删除恢复路径；Blob undelete并不恢复已删除容器。保护期过后、未开启保护、被授权的永久删除或账号故障均可能超出本例恢复能力。跨账号隔离副本与定期恢复演练另行设计。参考：依赖
text
azure-storage-blob==12.25.1
azure-identity==1.21.0
azure-core==1.33.0参考：配置模板
json
{
"account_url": "",
"container": "",
"run_id": "",
"input": "input/message.txt",
"block_bytes": 16
}参考：输入
text
Azure Blob 条件写入与分块提交练习。
This file belongs to the reader-owned isolated exercise.参考：完整源码
python
"""Reader-run Azure data-plane exercise. Not executed by the documentation author."""
import argparse
import hashlib
import json
import re
from pathlib import Path
from urllib.parse import urlparse
from azure.core import MatchConditions
from azure.core.exceptions import HttpResponseError, ResourceNotFoundError
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobBlock, BlobServiceClient, ContentSettings
ROOT = Path(__file__).resolve().parent
STATE = ROOT / "state.local.json"
def load_config(filename):
cfg = json.loads(Path(filename).read_text(encoding="utf-8"))
url = urlparse(cfg["account_url"])
if url.scheme != "https" or not url.hostname or url.query or url.path not in ("", "/"):
raise ValueError("account_url must be your HTTPS account Blob endpoint without SAS/path")
if not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{1,61}[a-z0-9])", cfg["container"]):
raise ValueError("supply an existing reader-owned container")
if "--" in cfg["container"]:
raise ValueError("container cannot contain consecutive hyphens")
if not re.fullmatch(r"[a-z0-9]{16,40}", cfg["run_id"]):
raise ValueError("run_id must be a new 16-40 character lowercase random identifier")
if not 1 <= cfg["block_bytes"] <= 4 * 1024 * 1024:
raise ValueError("block_bytes outside exercise limit")
return cfg
def read_state(cfg):
state = json.loads(STATE.read_text(encoding="utf-8"))
for key in ("account_url", "container", "run_id"):
if state[key] != cfg[key]:
raise ValueError("local state belongs to another exercise configuration")
expected = f"p2-azure-blob/{cfg['run_id']}/document.txt"
if state["name"] != expected:
raise ValueError("refuse a name outside this exercise")
return state
def save_state(state):
# Reserve before remote creation; preserve the name even if response is lost.
STATE.write_text(json.dumps(state, indent=2) + "\n", encoding="utf-8")
def owned_blob(container, state):
blob = container.get_blob_client(state["name"])
props = blob.get_blob_properties()
if props.metadata.get("exercise_run") != state["run_id"]:
raise ValueError("ownership metadata differs; inspect manually, do not delete")
return blob, props
def run(container, cfg):
if STATE.exists():
raise ValueError("state already exists; inspect/clean the previous run first")
payload = (ROOT / cfg["input"]).read_bytes()
if not payload or len(payload) > 1024 * 1024:
raise ValueError("exercise input must contain 1 byte to 1 MiB")
if (len(payload) + cfg["block_bytes"] - 1) // cfg["block_bytes"] > 50000:
raise ValueError("increase block_bytes: a committed block blob allows at most 50000 blocks")
state = {key: cfg[key] for key in ("account_url", "container", "run_id")}
state.update(name=f"p2-azure-blob/{cfg['run_id']}/document.txt", created=False)
save_state(state)
blob = container.get_blob_client(state["name"])
metadata = {"exercise_run": cfg["run_id"]}
# No overwrite: an existing resource is never adopted as ours.
first = blob.upload_blob(payload, overwrite=False, metadata=metadata,
content_settings=ContentSettings(content_type="text/plain; charset=utf-8"))
state["created"] = True
save_state(state)
original_etag = first["etag"]
changed = payload + b"\nrevision two\n"
blob.upload_blob(changed, overwrite=True, etag=original_etag,
match_condition=MatchConditions.IfNotModified, metadata=metadata)
try:
blob.upload_blob(b"stale write", overwrite=True, etag=original_etag,
match_condition=MatchConditions.IfNotModified, metadata=metadata)
except HttpResponseError as exc:
if exc.status_code != 412:
raise
print("EXPECTED_CONFLICT status=412")
else:
raise RuntimeError("stale ETag unexpectedly accepted; stop and inspect")
current = blob.get_blob_properties()
blocks = []
# Equal-length raw IDs; SDK encodes the supplied IDs for the REST protocol.
# A fresh blob/run isolates IDs from other writers; retries reuse ID+bytes.
for number, offset in enumerate(range(0, len(payload), cfg["block_bytes"])):
block_id = f"{number:08d}"
part = payload[offset:offset + cfg["block_bytes"]]
blob.stage_block(block_id=block_id, data=part, length=len(part))
blocks.append(BlobBlock(block_id=block_id))
blob.commit_block_list(blocks, etag=current.etag,
match_condition=MatchConditions.IfNotModified,
metadata=metadata,
content_settings=ContentSettings(content_type="text/plain; charset=utf-8"))
data = blob.download_blob().readall()
if data != payload:
raise RuntimeError("download differs from input")
print(f"VERIFIED bytes={len(data)} sha256={hashlib.sha256(data).hexdigest()}")
print("RESOURCE_RETAINED use inspect then cleanup with the same configuration")
def inspect(container, cfg):
state = read_state(cfg)
blob, props = owned_blob(container, state)
print(json.dumps({"name": state["name"], "etag": props.etag,
"size": props.size, "version_id": props.version_id}, default=str))
versions = container.list_blobs(name_starts_with=state["name"],
include=["versions", "snapshots", "deleted", "metadata"])
for item in versions:
if item.name == state["name"]:
print(json.dumps({"version_id": item.version_id,
"snapshot": item.snapshot, "deleted": item.deleted}, default=str))
def cleanup(container, cfg):
state = read_state(cfg)
if not state["created"]:
raise ValueError("creation not confirmed; inspect response-loss case manually")
# Never enumerate/delete a prefix, container, account, old versions or snapshots.
try:
blob, props = owned_blob(container, state)
blob.delete_blob(etag=props.etag, match_condition=MatchConditions.IfNotModified)
except ResourceNotFoundError:
print("CURRENT_BLOB_ABSENT histories may still exist; review account retention")
else:
print("CURRENT_BLOB_DELETED histories/soft-deleted data may retain charges")
print("STATE_RETAINED for manual recovery/history inspection")
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("action", choices=["run", "inspect", "cleanup"])
parser.add_argument("--config", default=str(ROOT / "config.local.json"))
args = parser.parse_args()
cfg = load_config(args.config)
# Credential selection is the reader's responsibility; no embedded key or SAS.
with DefaultAzureCredential(exclude_interactive_browser_credential=True) as credential:
with BlobServiceClient(cfg["account_url"], credential=credential,
retry_total=3, connection_timeout=10, read_timeout=30) as service:
container = service.get_container_client(cfg["container"])
container.get_container_properties()
{"run": run, "inspect": inspect, "cleanup": cleanup}[args.action](container, cfg)
if __name__ == "__main__":
main()参考：本地文件排除
text
.venv/
config.local.json
state.local.json
__pycache__/练习题与答案
题一：第一次upload成功但响应超时，能否删除state后重试？答：不能据异常认定没写，应先核对精确对象与请求证据，避免遗留付费对象或错误接管。题二：412是否证明权限正确且数据已恢复？答：不证明；它只说明此次请求的相关前置条件不满足，要核对服务错误码和具体请求步骤。题三：cleanup后仍列出旧版本是否bug？答：不是，本例明确保留历史数据，读者需按账号保护策略审核清理。
官方依据
接下来用恢复诊断评审失败分支。
继续阅读
- Azure Blob身份与保护契约
- Azure Blob对象与入门
- Azure Blob身份与保护契约