-
Notifications
You must be signed in to change notification settings - Fork 2.7k
/
get_and_verify_receipt.py
143 lines (123 loc) · 5.68 KB
/
get_and_verify_receipt.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
"""
FILE: get_and_verify_receipt.py
DESCRIPTION:
This sample demonstrates how to retrieve Confidential Ledger receipts
and verify their content. In this sample, we write a ledger entry, retrieve
a receipt certifying that it was written correctly, and then verify its
content by applying the receipt verification algorithm.
USAGE:
python get_and_verify_receipt.py
Set the environment variables with your own values before running the sample:
1) CONFIDENTIALLEDGER_ENDPOINT - the endpoint of the Confidential Ledger.
"""
import logging
import os
import sys
import tempfile
from azure.confidentialledger import ConfidentialLedgerClient
from azure.confidentialledger.certificate import (
ConfidentialLedgerCertificateClient,
)
from azure.confidentialledger.receipt import (
verify_receipt,
)
from azure.core.exceptions import HttpResponseError
from azure.identity import DefaultAzureCredential
logging.basicConfig(level=logging.ERROR)
LOG = logging.getLogger()
def main():
"""Main method."""
# Set the values of the client ID, tenant ID, and client secret of the AAD application as
# environment variables:
# AZURE_CLIENT_ID, AZURE_TENANT_ID, AZURE_CLIENT_SECRET, CONFIDENTIALLEDGER_ENDPOINT
try:
ledger_endpoint = os.environ["CONFIDENTIALLEDGER_ENDPOINT"]
except KeyError:
LOG.error(
"Missing environment variable 'CONFIDENTIALLEDGER_ENDPOINT' - "
"please set it before running the example"
)
sys.exit(1)
# Under the current URI format, the ledger id is the first part of the ledger endpoint.
# i.e. https://<ledger id>.confidential-ledger.azure.com
ledger_id = ledger_endpoint.replace("https://", "").split(".")[0]
identity_service_client = ConfidentialLedgerCertificateClient() # type: ignore[call-arg]
ledger_certificate = identity_service_client.get_ledger_identity(ledger_id)
# The Confidential Ledger's TLS certificate must be written to a file to be used by the
# ConfidentialLedgerClient. Here, we write it to a temporary file so that is is cleaned up
# automatically when the program exits.
with tempfile.TemporaryDirectory() as tempdir:
ledger_cert_file = os.path.join(tempdir, f"{ledger_id}.pem")
with open(ledger_cert_file, "w") as outfile:
outfile.write(ledger_certificate["ledgerTlsCertificate"])
print(
f"Ledger certificate has been written to {ledger_cert_file}. "
"It will be deleted when the script completes."
)
# Build a client through AAD
ledger_client = ConfidentialLedgerClient(
ledger_endpoint,
credential=DefaultAzureCredential(),
ledger_certificate_path=ledger_cert_file,
)
# Write a ledger entry and wait for the transaction to be committed.
try:
entry_contents = "Hello world!"
post_poller = ledger_client.begin_create_ledger_entry( # type: ignore[attr-defined]
{"contents": entry_contents}
)
post_entry_result = post_poller.result()
transaction_id = post_entry_result["transactionId"]
print(
f"Wrote '{entry_contents}' to the ledger at transaction {transaction_id}."
)
except HttpResponseError as e:
if e.response != None:
print("Request failed: {}".format(e.response.json())) # type: ignore[union-attr]
else:
print("No response found")
raise
# Get a receipt for a ledger entry.
# A receipt can be retrieved for any transaction id to provide cryptographic proof of the
# contents of the transaction.
try:
print(
f"Retrieving a receipt for {transaction_id}. The receipt may be used to "
"cryptographically verify the contents of the transaction."
)
print(
"For more information about receipts, please see "
"https://microsoft.github.io/CCF/main/audit/receipts.html#receipts"
)
get_receipt_poller = ledger_client.begin_get_receipt(transaction_id) # type: ignore[attr-defined]
get_receipt_result = get_receipt_poller.result()
print(f"Receipt for transaction id {transaction_id}: {get_receipt_result}")
except HttpResponseError as e:
if e.response != None:
print("Request failed: {}".format(e.response.json())) # type: ignore[union-attr]
else:
print("No response found")
raise
# Read content of service certificate file saved in previous step.
with open(ledger_cert_file, "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")
raise
if __name__ == "__main__":
main()