你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站,请访问 https://docs.azure.cn

适用于 Python 的 Azure 机密账本客户端库 - 版本 1.1.1

Azure 机密账本提供一项服务,用于记录到不可变、防篡改的账本。 作为 Azure 机密计算 组合的一部分,Azure 机密账本在安全、基于硬件的受信任执行环境中运行,也称为 enclave。 它基于 Microsoft Research 的 机密联盟框架

源代码 | 包 (PyPI) | 包 (Conda) | API 参考文档 | 产品文档

入门

安装包

使用 pip 安装 azure-confidentialledgerazure-identity

pip install azure-identity azure-confidentialledger

azure-identity 用于 Azure Active Directory 身份验证,如下所示。

先决条件

  • Azure 订阅
  • Python 3.6 或更高版本
  • Azure 机密账本的运行实例。
  • 机密账本中的注册用户,通常在 创建 ARM 资源期间分配,具有 Administrator 权限。

验证客户端

使用 Azure Active Directory

本文档演示如何使用 DefaultAzureCredential 通过 Azure Active Directory 向机密账本进行身份验证。 但是, ConfidentialLedgerClient 接受任何 azure 标识 凭据。 有关其他凭据的详细信息,请参阅 azure-identity 文档。

使用客户端证书

作为 Azure Active Directory 的替代方法,客户端可以选择使用客户端证书通过相互 TLS 进行身份验证。 azure.confidentialledger.ConfidentialLedgerCertificateCredential 可用于此目的。

创建客户端

DefaultAzureCredential 将自动处理大多数 Azure SDK 客户端方案。 若要开始,请为在机密账本中注册的 AAD 标识设置环境变量。

export AZURE_CLIENT_ID="generated app id"
export AZURE_CLIENT_SECRET="random password"
export AZURE_TENANT_ID="tenant id"

然后, DefaultAzureCredential 将能够对 进行身份验证 ConfidentialLedgerClient

构造客户端还需要机密账本的 URL 和 ID,可以从 Azure CLI 或 Azure 门户获取这些 URL 和 ID。 检索到这些值后,请替换以下示例中的 和 "https://my-ledger-id.confidential-ledger.azure.com" 实例"my-ledger-id"。 可能还需要将 替换为 "https://identity.confidential-ledger.core.azure.com" 账本的 ARM 说明中的 主机名 identityServiceUri

由于机密账本使用安全生成并存储在 enclave 中的自签名证书,因此必须先从机密账本标识服务检索每个机密账本的签名证书。

from azure.confidentialledger import ConfidentialLedgerClient
from azure.confidentialledger.certificate import ConfidentialLedgerCertificateClient
from azure.identity import DefaultAzureCredential

identity_client = ConfidentialLedgerCertificateClient()
network_identity = identity_client.get_ledger_identity(
    ledger_id="my-ledger-id"
)

ledger_tls_cert_file_name = "ledger_certificate.pem"
with open(ledger_tls_cert_file_name, "w") as cert_file:
    cert_file.write(network_identity["ledgerTlsCertificate"])

credential = DefaultAzureCredential()
ledger_client = ConfidentialLedgerClient(
    endpoint="https://my-ledger-id.confidential-ledger.azure.com",
    credential=credential,
    ledger_certificate_path=ledger_tls_cert_file_name
)

方便地, ConfidentialLedgerClient 构造函数将提取账本 TLS 证书 (,并将其写入指定的文件) (如果提供了不存在的文件)。 用户负责根据需要删除创建的文件。

from azure.confidentialledger import ConfidentialLedgerClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
ledger_client = ConfidentialLedgerClient(
    endpoint="https://my-ledger-id.confidential-ledger.azure.com",
    credential=credential,
    ledger_certificate_path="ledger_certificate.pem"
)

# The ledger TLS certificate is written to `ledger_certificate.pem`.

为了明确文件用于账本 TLS 证书,后续示例会将账本 TLS 证书显式写入文件。

关键概念

账本条目和事务

每次写入 Azure 机密账本都会在服务中生成不可变的账本条目。 写入(也称为事务)由事务 ID 唯一标识,事务 ID 随每次写入而递增。 写入后,可以随时检索账本条目。

集合

虽然大多数用例只涉及每个机密账本的一个集合,但我们提供集合 ID 功能,以防语义或逻辑上不同的数据组需要存储在同一机密账本中。

账本条目由其 collectionId检索。 对于未collectionId指定 的写入条目,机密账本将始终采用由服务确定collectionId的常量。

用户

用户直接使用机密账本进行管理,而不是通过 Azure 进行管理。 用户可能基于 AAD,由其 AAD 对象 ID 标识,或者基于证书,由其 PEM 证书指纹标识。

Receipts

为了强制实施事务完整性保证,Azure 机密账本使用 [Merkle 树][merkle_tree_wiki] 数据结构来记录追加到不可变账本的所有事务块的哈希。 提交写入事务后,Azure 机密账本用户可以通过机密账本中生成的条目获取加密 Merkle 证明或收据,以验证写入操作是否已正确保存。 写入事务收据是系统已提交相应事务的证明,可用于验证该条目是否已有效地追加到账本。

有关 Azure 机密账本写入事务收据的详细信息,请参阅以下 文章

收据验证

获取写入事务的收据后,Azure 机密账本用户可以按照验证算法验证提取的收据的内容。 验证成功证明与收据关联的写入操作已正确追加到不可变账本中。

有关 Azure 机密账本写入事务收据的验证过程的详细信息,请参阅以下 文章

应用程序声明

Azure 机密账本应用程序可以附加任意数据(称为应用程序声明)以写入事务。 这些声明表示写入操作期间执行的操作。 附加到事务时,声明对象的 SHA-256 摘要包含在账本中,并作为写入事务的一部分提交。 这可以保证摘要已就地登录,并且不会被篡改。

稍后,应用程序声明可以在与添加应用程序声明的同一事务对应的接收有效负载中以未摘要的形式显示。 这允许用户利用收据中的信息重新计算在事务期间由 Azure 机密账本实例就地附加和登录的相同声明摘要。 声明摘要可用作写入事务回执验证过程的一部分,为用户提供一种脱机方式来完全验证所记录声明的真实性。

有关应用程序声明格式和摘要计算算法的更多详细信息,请参阅以下链接:

有关 CCF 应用程序声明的详细信息,请参阅以下 CCF 文档页:

机密计算

Azure 机密计算 允许在云中处理数据时隔离和保护数据。 Azure 机密账本在 Azure 机密计算虚拟机上运行,从而通过加密正在使用的数据提供更强大的数据保护。

机密联盟框架

Azure 机密账本基于 Microsoft Research 的开源 机密联盟框架 (CCF) 。 根据 CCF,应用程序由成员联盟管理,这些成员能够提交修改和治理应用程序操作的建议。 在 Azure 机密账本中,Microsoft Azure 拥有一个操作员成员标识,该标识允许它执行治理和维护操作,例如替换机密账本中的不正常节点和升级 enclave 代码。

示例

本部分包含涵盖常见任务的代码片段,包括:

追加条目

需要以防篡改方式永久存储的数据可以通过将条目追加到账本中,将其保存到 Azure 机密账本。

由于机密账本是分布式系统,因此罕见的暂时性故障可能会导致写入丢失。 对于必须保留的条目,建议验证写入是否持久。 对于首选较高客户端吞吐量的不太重要的写入,可以跳过等待步骤。

from azure.confidentialledger import ConfidentialLedgerClient
from azure.confidentialledger.certificate import ConfidentialLedgerCertificateClient
from azure.identity import DefaultAzureCredential

identity_client = ConfidentialLedgerCertificateClient()
network_identity = identity_client.get_ledger_identity(
    ledger_id="my-ledger-id"
)

ledger_tls_cert_file_name = "ledger_certificate.pem"
with open(ledger_tls_cert_file_name, "w") as cert_file:
    cert_file.write(network_identity["ledgerTlsCertificate"])

credential = DefaultAzureCredential()
ledger_client = ConfidentialLedgerClient(
    endpoint="https://my-ledger-id.confidential-ledger.azure.com",
    credential=credential,
    ledger_certificate_path=ledger_tls_cert_file_name
)

post_entry_result = ledger_client.create_ledger_entry(
        {"contents": "Hello world!"}
    )
transaction_id = post_entry_result["transactionId"]

wait_poller = ledger_client.begin_wait_for_commit(transaction_id)
wait_poller.wait()
print(f'Ledger entry at transaction id {transaction_id} has been committed successfully')

或者,客户端可能会在写入账本条目时等待提交。

from azure.confidentialledger import ConfidentialLedgerClient
from azure.confidentialledger.certificate import ConfidentialLedgerCertificateClient
from azure.identity import DefaultAzureCredential

identity_client = ConfidentialLedgerCertificateClient()
network_identity = identity_client.get_ledger_identity(
    ledger_id="my-ledger-id"
)

ledger_tls_cert_file_name = "ledger_certificate.pem"
with open(ledger_tls_cert_file_name, "w") as cert_file:
    cert_file.write(network_identity["ledgerTlsCertificate"])

credential = DefaultAzureCredential()
ledger_client = ConfidentialLedgerClient(
    endpoint="https://my-ledger-id.confidential-ledger.azure.com",
    credential=credential,
    ledger_certificate_path=ledger_tls_cert_file_name
)

post_poller = ledger_client.begin_create_ledger_entry(
    {"contents": "Hello world again!"}
)
new_post_result = post_poller.result()
print(
    'The new ledger entry has been committed successfully at transaction id '
    f'{new_post_result["transactionId"]}'
)

检索账本条目

获取早于最新条目的账本条目可能需要一些时间,因为服务正在加载历史条目,因此会提供一个轮询器。

账本条目由集合检索。 返回的值是由事务 ID 标识的指定集合中包含的值。

from azure.confidentialledger import ConfidentialLedgerClient
from azure.confidentialledger.certificate import ConfidentialLedgerCertificateClient
from azure.identity import DefaultAzureCredential

identity_client = ConfidentialLedgerCertificateClient()
network_identity = identity_client.get_ledger_identity(
    ledger_id="my-ledger-id"
)

ledger_tls_cert_file_name = "ledger_certificate.pem"
with open(ledger_tls_cert_file_name, "w") as cert_file:
    cert_file.write(network_identity["ledgerTlsCertificate"])

credential = DefaultAzureCredential()
ledger_client = ConfidentialLedgerClient(
    endpoint="https://my-ledger-id.confidential-ledger.azure.com",
    credential=credential,
    ledger_certificate_path=ledger_tls_cert_file_name
)

post_poller = ledger_client.begin_create_ledger_entry(
    {"contents": "Original hello"}
)
post_result = post_poller.result()

post_transaction_id = post_result["transactionId"]

latest_entry = ledger_client.get_current_ledger_entry()
print(
    f'Current entry (transaction id = {latest_entry["transactionId"]}) '
    f'in collection {latest_entry["collectionId"]}: {latest_entry["contents"]}'
)

post_poller = ledger_client.begin_create_ledger_entry(
    {"contents": "Hello!"}
)
post_result = post_poller.result()

get_entry_poller = ledger_client.begin_get_ledger_entry(post_transaction_id)
older_entry = get_entry_poller.result()
print(
    f'Contents of {older_entry["entry"]["collectionId"]} at {post_transaction_id}: {older_entry["entry"]["contents"]}'
)

创建范围查询

可以通过一系列事务 ID 检索账本条目。 条目将仅从默认或指定的集合中返回。

from azure.confidentialledger import ConfidentialLedgerClient
from azure.confidentialledger.certificate import ConfidentialLedgerCertificateClient
from azure.identity import DefaultAzureCredential

identity_client = ConfidentialLedgerCertificateClient()
network_identity = identity_client.get_ledger_identity(
    ledger_id="my-ledger-id"
)

ledger_tls_cert_file_name = "ledger_certificate.pem"
with open(ledger_tls_cert_file_name, "w") as cert_file:
    cert_file.write(network_identity["ledgerTlsCertificate"])

credential = DefaultAzureCredential()
ledger_client = ConfidentialLedgerClient(
    endpoint="https://my-ledger-id.confidential-ledger.azure.com",
    credential=credential,
    ledger_certificate_path=ledger_tls_cert_file_name
)

post_poller = ledger_client.begin_create_ledger_entry(
    {"contents": "First message"}
)
first_transaction_id = post_poller.result()["transactionId"]

for i in range(10):
    ledger_client.create_ledger_entry(
        {"contents": f"Message {i}"}
    )

post_poller = ledger_client.begin_create_ledger_entry(
    {"contents": "Last message"}
)
last_transaction_id = post_poller.result()["transactionId"]

ranged_result = ledger_client.list_ledger_entries(
    from_transaction_id=first_transaction_id,
    to_transaction_id=last_transaction_id,
)
for entry in ranged_result:
    print(f'Contents at {entry["transactionId"]}: {entry["contents"]}')

管理用户

具有 Administrator 特权的用户可以直接使用机密账本本身管理机密账本的用户。 可用角色 (Reader 只读) 、 Contributor (读取和写入) ,以及 Administrator (读取、写入和添加或删除用户) 。

from azure.confidentialledger import ConfidentialLedgerClient
from azure.confidentialledger.certificate import ConfidentialLedgerCertificateClient
from azure.identity import DefaultAzureCredential

identity_client = ConfidentialLedgerCertificateClient()
network_identity = identity_client.get_ledger_identity(
    ledger_id="my-ledger-id"
)

ledger_tls_cert_file_name = "ledger_certificate.pem"
with open(ledger_tls_cert_file_name, "w") as cert_file:
    cert_file.write(network_identity["ledgerTlsCertificate"])

credential = DefaultAzureCredential()
ledger_client = ConfidentialLedgerClient(
    endpoint="https://my-ledger-id.confidential-ledger.azure.com",
    credential=credential,
    ledger_certificate_path=ledger_tls_cert_file_name
)

user_id = "some AAD object id"
user = ledger_client.create_or_update_user(
    user_id, {"assignedRole": "Contributor"}
)
# A client may now be created and used with AAD credentials (i.e. AAD-issued JWT tokens) for the user identified by `user_id`.

user = ledger_client.get_user(user_id)
assert user["userId"] == user_id
assert user["assignedRole"] == "Contributor"

ledger_client.delete_user(user_id)

# For a certificate-based user, their user ID is the fingerprint for their PEM certificate.
user_id = "PEM certificate fingerprint"
user = ledger_client.create_or_update_user(
    user_id, {"assignedRole": "Reader"}
)

user = ledger_client.get_user(user_id)
assert user["userId"] == user_id
assert user["assignedRole"] == "Reader"

ledger_client.delete_user(user_id)

使用证书身份验证

客户端可以使用相互 TLS 中的客户端证书进行身份验证,而不是通过 Azure Active Directory 令牌进行身份验证。 ConfidentialLedgerCertificateCredential 为此类客户端提供 。

from azure.confidentialledger import (
    ConfidentialLedgerCertificateCredential,
    ConfidentialLedgerClient,
)
from azure.confidentialledger.certificate import ConfidentialLedgerCertificateClient

identity_client = ConfidentialLedgerCertificateClient()
network_identity = identity_client.get_ledger_identity(
    ledger_id="my-ledger-id"
)

ledger_tls_cert_file_name = "ledger_certificate.pem"
with open(ledger_tls_cert_file_name, "w") as cert_file:
    cert_file.write(network_identity["ledgerTlsCertificate"])

credential = ConfidentialLedgerCertificateCredential(
    certificate_path="Path to user certificate PEM file"
)
ledger_client = ConfidentialLedgerClient(
    endpoint="https://my-ledger-id.confidential-ledger.azure.com",
    credential=credential,
    ledger_certificate_path=ledger_tls_cert_file_name
)

验证事务回执

客户端可以利用 SDK 中的收据验证库来验证 Azure 机密 Legder 实例颁发的写入事务回执。 实用工具可用于脱机完全验证收据,因为验证算法不需要连接到机密账本或任何其他 Azure 服务。

将新条目追加到账本 (请参阅 此示例) ,即可获取已提交的写入事务的收据。

from azure.confidentialledger import ConfidentialLedgerClient
from azure.confidentialledger.certificate import ConfidentialLedgerCertificateClient
from azure.identity import DefaultAzureCredential

# Replace this with the Confidential Ledger ID 
ledger_id = "my-ledger-id"

# Setup authentication
credential = DefaultAzureCredential()

# Create a Ledger Certificate client and use it to
# retrieve the service identity for our ledger
identity_client = ConfidentialLedgerCertificateClient()
network_identity = identity_client.get_ledger_identity(
    ledger_id=ledger_id
)

# Save ledger service certificate into a file for later use
ledger_tls_cert_file_name = "ledger_certificate.pem"
with open(ledger_tls_cert_file_name, "w") as cert_file:
    cert_file.write(network_identity["ledgerTlsCertificate"])

# Create Confidential Ledger client
ledger_client = ConfidentialLedgerClient(
    endpoint=f"https://{ledger_id}.confidential-ledger.azure.com",
    credential=credential,
    ledger_certificate_path=ledger_tls_cert_file_name
)

# The method begin_get_receipt returns a poller that
# we can use to wait for the receipt to be available for retrieval 
get_receipt_poller = ledger_client.begin_get_receipt(transaction_id)
get_receipt_result = get_receipt_poller.result()

print(f"Write receipt for transaction id {transaction_id} was successfully retrieved: {get_receipt_result}")

获取写入事务的回执后,可以调用 verify_receipt 函数来验证回执是否有效。 函数可以接受应用程序声明的可选列表,以针对收据声明摘要进行验证。

from azure.confidentialledger.receipt import (
    verify_receipt,
)

# Read contents of service certificate file saved in previous step.
with open(ledger_tls_cert_file_name, "r") as service_cert_file:
    service_cert_content = service_cert_file.read()

# Optionally read application claims, if any
application_claims = get_receipt_result.get("applicationClaims", None) 

try:
    # Verify the contents of the receipt.
    verify_receipt(get_receipt_result["receipt"], service_cert_content, application_claims=application_claims)
    print(f"Receipt for transaction id {transaction_id} successfully verified")
except ValueError:
    print(f"Receipt verification for transaction id {transaction_id} failed")

一个完整的示例 Python 程序,演示如何将新条目追加到正在运行的机密账本实例、获取已提交事务的回执,并验证是否可以在 samples 文件夹下找到收据内容: get_and_verify_receipt.py

异步 API

此库包含 Python 3.5+ 上支持的完整异步 API。 若要使用它,必须先安装异步传输,例如 aiohttp。 有关详细信息,请参阅 azure-core 文档

azure.confidentialledger.aio获取异步客户端。 方法与同步客户端具有相同的名称和签名。 可 在此处找到示例。

疑难解答

常规

机密账本客户端会引发 azure-core 中定义的异常。 例如,如果尝试获取不存在的事务, ConfidentialLedgerClient 则会引发 ResourceNotFoundError

from azure.core.exceptions import ResourceNotFoundError
from azure.confidentialledger import ConfidentialLedgerClient
from azure.confidentialledger.certificate import ConfidentialLedgerCertificateClient
from azure.identity import DefaultAzureCredential

identity_client = ConfidentialLedgerCertificateClient()
network_identity = identity_client.get_ledger_identity(
    ledger_id="my-ledger-id"
)

ledger_tls_cert_file_name = "ledger_certificate.pem"
with open(ledger_tls_cert_file_name, "w") as cert_file:
    cert_file.write(network_identity["ledgerTlsCertificate"])

credential = DefaultAzureCredential()
ledger_client = ConfidentialLedgerClient(
    endpoint="https://my-ledger-id.confidential-ledger.azure.com",
    credential=credential,
    ledger_certificate_path=ledger_tls_cert_file_name
)

try:
    ledger_client.begin_get_ledger_entry(
        transaction_id="10000.100000"  # Using a very high id that probably doesn't exist in the ledger if it's relatively new.
    )
except ResourceNotFoundError as e:
    print(e.message)

日志记录

此库使用标准 日志记录 库进行日志记录。 有关 HTTP 会话 (URL、标头等的基本信息,) 在 INFO 级别记录。

可以使用 参数在客户端 logging_enable 上启用详细的调试级别日志记录,包括请求/响应正文和未处理标头:

import logging
import sys

from azure.confidentialledger import ConfidentialLedgerClient
from azure.confidentialledger.certificate import ConfidentialLedgerCertificateClient
from azure.identity import DefaultAzureCredential

# Create a logger for the 'azure' SDK
logger = logging.getLogger('azure')
logger.setLevel(logging.DEBUG)

# Configure a console output
handler = logging.StreamHandler(stream=sys.stdout)
logger.addHandler(handler)

identity_client = ConfidentialLedgerCertificateClient()
network_identity = identity_client.get_ledger_identity(
    ledger_id="my-ledger-id"
)

ledger_tls_cert_file_name = "ledger_certificate.pem"
with open(ledger_tls_cert_file_name, "w") as cert_file:
    cert_file.write(network_identity["ledgerTlsCertificate"])

credential = DefaultAzureCredential()

# This client will log detailed information about its HTTP sessions, at DEBUG level.
ledger_client = ConfidentialLedgerClient(
    endpoint="https://my-ledger-id.confidential-ledger.azure.com",
    credential=credential,
    ledger_certificate_path=ledger_tls_cert_file_name,
    logging_enable=True,
)

同样,即使没有为客户端启用详细日志记录,logging_enable 也可以为单个操作启用:

ledger_client.get_current_ledger_entry(logging_enable=True)

后续步骤

更多示例代码

这些代码示例演示 Azure 机密账本客户端库的常见方案操作。

常见方案

高级方案

其他文档

有关 Azure 机密账本的更广泛文档,请参阅 API 参考文档。 你还可以阅读有关 Microsoft Research 的开源 机密联盟框架的详细信息。

贡献

本项目欢迎贡献和建议。 大多数贡献要求你同意贡献者许可协议 (CLA),并声明你有权(并且确实有权)授予我们使用你的贡献的权利。 有关详细信息,请访问 https://cla.microsoft.com

提交拉取请求时,CLA 机器人将自动确定你是否需要提供 CLA,并相应地修饰 PR(例如标签、注释)。 直接按机器人提供的说明操作。 只需使用 CLA 对所有存储库执行一次这样的操作。

此项目采用了 Microsoft 开放源代码行为准则。 有关详细信息,请参阅行为准则常见问题解答;若有其他任何问题或意见,请联系 opencode@microsoft.com。