Setting the file. One moment.
Cleanup Receipts · Foundry Iq · microsoft/azure-skills · Skills Docs
Back to the top of the page File Cu Canary
— line 131
This file
Number 10.26
Position 26 of 77
Type Python
Size 14 KB
Lines 239 helpers/ _cleanup_receipts.py
Python · 239 lines · 14 KB
._common
import
HelperFailure, digest, load_approved_input, reject_secrets, odata_name, validate_search_endpoint
12 except ImportError :
13 import _bootstrap_io as private_io
14 from _common import HelperFailure, digest, load_approved_input, reject_secrets, odata_name, validate_search_endpoint
15
16
17 def fail (code, message):
18 return HelperFailure(code, message, blocked_at = "cleanup-provenance" )
19
20
21 def version_identity (value):
22 get = value.get if isinstance (value, dict ) else lambda key: getattr (value, key, None )
23 identifier, created = get( "id" ), get( "created_at" )
24 if isinstance (created, datetime):
25 created = int (created.timestamp()) if created.tzinfo is not None else None
26 if not isinstance (identifier, str ) or not identifier.strip() or type (created) is not int or created <= 0 :
27 return None
28 return { "id" : identifier, "created_at" : created}
29
30
31 def add_argument (parser):
32 parser.add_argument( "--cleanup-receipt-dir" , type = Path,
33 help = "Existing absolute private directory for filtered original-create evidence." )
34
35
36 class Capture :
37 def __init__ (self, directory, document):
38 self .directory = private_io.private_directory( str (directory))
39 self .plan_digest = digest(document[ "plan" ])
40 self .owner = document[ "plan" ].get( "owner" )
41 self .records = {}
42 self .summaries = []
43
44 def start (self, target, acknowledgement):
45 acknowledgement.setdefault( "version_identity" , None )
46 record = {
47 "schema_version" : "1.0" , "kind" : "cleanup-creation-receipt" ,
48 "plan_digest" : self .plan_digest, "owner" : self .owner, "target" : target,
49 "state" : "acknowledged" , "acknowledgement" : acknowledgement, "snapshot" : None ,
50 }
51 key = digest(target)
52 self .records[key] = record
53 self ._persist(record)
54
55 def finish (self, target, snapshot):
56 record = copy.deepcopy( self .records[digest(target)])
57 ack = record[ "acknowledgement" ]
58 snapshot.setdefault( "version_identity" , None )
59 if (snapshot[ "version_identity" ] != ack[ "version_identity" ]
60 or (target[ "type" ] == "prompt-agent-version" and ack[ "version_identity" ] is None )):
61 raise fail( "creation-version-unproven" , "Native version identity/creation time must match the original acknowledgement." )
62 if not isinstance (ack[ "request_id" ], str ) or not ack[ "request_id" ].strip():
63 raise fail( "creation-request-id-unavailable" , "Original native acknowledgement lacks a request ID." )
64 if snapshot[ "definition_digest" ] != ack[ "definition_digest" ]:
65 raise fail( "definition-drift" , "Readback differs from the acknowledged original creation." )
66 record.update( state = "verified" , snapshot = snapshot)
67 self ._persist(record)
68
69 def _persist (self, record):
70 reject_secrets(record)
71 value = { ** record, "integrity" : digest(record)}
72 try :
73 path = private_io.private_file( self .directory, uuid.uuid4().hex + ".cleanup.json" , value)
74 except OSError as exc:
75 raise fail( "cleanup-receipt-persistence-failed" , "Acknowledged creation evidence could not be retained privately." ) from exc
76 self .summaries.append({ "target" : record[ "target" ], "state" : record[ "state" ],
77 "receipt_file" : str (path), "evidence_digest" : value[ "integrity" ]})
78
79
80 def search_target (plan):
81 return { "type" : plan[ "resource_type" ], "endpoint" : validate_search_endpoint(plan[ "endpoint" ]),
82 ** {key: plan[key] for key in ( "name" , "api_version" )}}
83
84
85 def search_ack (capture, plan, response):
86 try :
87 from . import search_reconcile as search, _cleanup_dependencies as dependencies
88 except ImportError :
89 import search_reconcile as search, _cleanup_dependencies as dependencies
90 generated = None
91 if plan[ "resource_type" ] == "knowledge-source" :
92 try :
93 generated = dependencies.generated(response.body) if isinstance (response.body, dict ) else None
94 except HelperFailure:
95 pass
96 capture.start(search_target(plan), {
97 "operation" : "search-create" , "status" : response.status, "request_id" : response.request_id,
98 "definition_digest" : digest(search._definition(plan[ "desired" ])),
99 "etag_evidence" : search.response_etags(response), "generated" : generated, "version" : None ,
100 })
101
102
103 def search_finish (capture, plan, current, token, transport):
104 try :
105 from . import search_reconcile as search, _cleanup_dependencies as dependencies
106 except ImportError :
107 import search_reconcile as search, _cleanup_dependencies as dependencies
108 target = search_target(plan)
109 ack = capture.records[digest(target)][ "acknowledgement" ]
110 etag = search.resolve_etag(ack[ "etag_evidence" ], ack[ "request_id" ])
111 if ack[ "status" ] != 201 or not etag or current is None or current.get( "@odata.etag" ) != etag:
112 raise fail( "creation-version-unproven" , "Require original HTTP 201 and the unchanged acknowledged ETag, not a later GET version." )
113 children = []
114 if plan[ "resource_type" ] == "knowledge-source" :
115 if ack[ "generated" ] is None or dependencies.generated(current) != ack[ "generated" ]:
116 raise fail( "generated-creation-evidence-unavailable" , "The original create response did not identify this exact generated cascade." )
117 for kind, name in sorted (ack[ "generated" ].items()):
118 url = f " { plan[ 'endpoint' ].rstrip( '/' ) } / { dependencies. COLLECTIONS [kind] } (' { odata_name(name) } ')?api-version= { plan[ 'api_version' ] } "
119 child, _ = search.read_resource(url, token, transport = transport)
120 if child is None or child.get( "name" ) != name or not child.get( "@odata.etag" ):
121 raise fail( "generated-version-unavailable" , "An original generated child lacks exact version readback." )
122 children.append({ "type" : kind, "name" : name, "etag" : child[ "@odata.etag" ], "definition_digest" : digest(child)})
123 refreshed, _ = search.read_resource(search.resource_url(plan), token, transport = transport)
124 if (refreshed is None or refreshed.get( "@odata.etag" ) != etag
125 or search._definition(refreshed) != search._definition(current)
126 or dependencies.generated(refreshed) != ack[ "generated" ]):
127 raise fail( "definition-drift" , "Source changed during original generated-child capture." )
128 capture.finish(target, { "definition_digest" : digest(search._definition(current)), "etag" : etag, "generated" : children})
129
130
131 def project_target (plan, kind, * , name = None , version = None ):
132 target = { "type" : kind, "project_resource_id" : plan[ "project_resource_id" ].casefold(),
133 "project_endpoint" : plan[ "project_endpoint" ].rstrip( "/" ),
134 "name" : name or plan[ "connection" ][ "name" ]}
135 if version is not None :
136 target[ "version" ] = version
137 return target
138
139
140 def connection_ack (capture, plan, response):
141 body = response.body if isinstance (response.body, dict ) else {}
142 capture.start(project_target(plan, "project-connection" ), {
143 "operation" : "project-connection-create" , "status" : response.status, "request_id" : response.request_id,
144 "definition_digest" : digest(body), "version" : None , "generated" : None ,
145 "etag_evidence" : { "body" : body.get( "etag" ) or body.get( "@odata.etag" ),
146 "headers" : [v for k, v in response.headers.items() if k.casefold() == "etag" ]},
147 })
148
149
150 def connection_finish (capture, plan, body):
151 try :
152 from .search_reconcile import resolve_etag
153 except ImportError :
154 from search_reconcile import resolve_etag
155 target = project_target(plan, "project-connection" )
156 ack = capture.records[digest(target)][ "acknowledgement" ]
157 etag = resolve_etag(ack[ "etag_evidence" ], ack[ "request_id" ])
158 if ack[ "status" ] != 201 or not etag or etag != (body.get( "etag" ) or body.get( "@odata.etag" )):
159 raise fail( "creation-version-unproven" , "Connection readback must retain its original HTTP 201 ETag." )
160 capture.finish(target, { "definition_digest" : digest(body), "etag" : etag, "generated" : []})
161
162
163 def sdk_response_hook (metadata):
164 def capture (response):
165 native = response.http_response
166 metadata[ "request_id" ] = next ((v for k, v in native.headers.items()
167 if k.casefold() in ( "request-id" , "x-request-id" , "x-ms-request-id" , "apim-request-id" )), None )
168 metadata[ "status" ] = native.status_code
169 return capture
170
171
172 def agent_ack (capture, plan, created, metadata):
173 target = project_target(plan, "prompt-agent-version" , name = created.name, version = str (created.version))
174 capture.start(target, {
175 "operation" : "agents.create_version" , "status" : metadata.get( "status" ), "request_id" : metadata.get( "request_id" ),
176 "definition_digest" : digest(created.definition.as_dict()), "version" : str (created.version),
177 "etag_evidence" : None , "generated" : None ,
178 "version_identity" : version_identity(created),
179 })
180 if (metadata.get( "status" ) not in ( 200 , 201 ) or created.name != plan[ "agent" ][ "name" ]
181 or str (created.version) == plan[ "agent" ][ "version" ] or version_identity(created) is None ):
182 raise fail( "ownership-unproven" , "The original SDK return must identify a newly created version in the approved agent." )
183 return target
184
185
186 def load (input_path, receipt_path, target):
187 try :
188 from .blob_recheck import read_private
189 from . import search_reconcile as search
190 except ImportError :
191 from blob_recheck import read_private
192 import search_reconcile as search
193 _, plan, fingerprint = load_approved_input(input_path)
194 record = read_private(receipt_path)
195 reject_secrets(record)
196 fields = { "schema_version" , "kind" , "plan_digest" , "owner" , "target" , "state" , "acknowledgement" , "snapshot" , "integrity" }
197 if ( not isinstance (record, dict ) or set (record) != fields or record[ "schema_version" ] != "1.0"
198 or record[ "kind" ] != "cleanup-creation-receipt" or record[ "state" ] != "verified"
199 or record[ "plan_digest" ] != fingerprint or record[ "owner" ] != plan.get( "owner" )
200 or record[ "target" ] != target
201 or record[ "integrity" ] != digest({k: v for k, v in record.items() if k != "integrity" })):
202 raise fail( "ownership-unproven" , "Require the exact original verified producer receipt and original approval; ACK/recovery/reuse cannot substitute." )
203 ack, snapshot = record[ "acknowledgement" ], record[ "snapshot" ]
204 if ( not isinstance (ack, dict ) or set (ack) != { "operation" , "status" , "request_id" , "definition_digest" , "etag_evidence" , "generated" , "version" , "version_identity" }
205 or not isinstance (snapshot, dict ) or set (snapshot) != { "definition_digest" , "etag" , "generated" , "version_identity" }
206 or not isinstance (ack[ "request_id" ], str ) or not ack[ "request_id" ].strip()
207 or snapshot[ "definition_digest" ] != ack[ "definition_digest" ]
208 or not isinstance (snapshot[ "definition_digest" ], str ) or not search. SHA256 .fullmatch(snapshot[ "definition_digest" ])
209 or not isinstance (snapshot[ "generated" ], list )):
210 raise fail( "ownership-unproven" , "Producer receipt lacks complete acknowledged version evidence." )
211 if (snapshot[ "version_identity" ] != ack[ "version_identity" ]
212 or (target[ "type" ] == "prompt-agent-version" and (
213 not isinstance (snapshot[ "version_identity" ], dict )
214 or set (snapshot[ "version_identity" ]) != { "id" , "created_at" }
215 or version_identity(snapshot[ "version_identity" ]) != snapshot[ "version_identity" ]))):
216 raise fail( "ownership-unproven" , "Native version birth identity is missing or changed." )
217 if target[ "type" ] != "prompt-agent-version" and (
218 not isinstance (ack[ "etag_evidence" ], dict ) or set (ack[ "etag_evidence" ]) != { "body" , "headers" }
219 or not isinstance (ack[ "etag_evidence" ][ "headers" ], list )
220 ):
221 raise fail( "ownership-unproven" , "Native ETag evidence must be complete." )
222 if target[ "type" ] in ( "knowledge-base" , "knowledge-source" ):
223 if (ack[ "operation" ] != "search-create" or ack[ "status" ] != 201
224 or search.resolve_etag(ack[ "etag_evidence" ], ack[ "request_id" ]) != snapshot[ "etag" ]
225 or not isinstance (snapshot[ "etag" ], str ) or not snapshot[ "etag" ].strip()):
226 raise fail( "creation-version-unproven" , "Original Search create acknowledgement and snapshot versions disagree." )
227 elif target[ "type" ] == "project-connection" :
228 if (ack[ "operation" ] != "project-connection-create" or ack[ "status" ] != 201
229 or search.resolve_etag(ack[ "etag_evidence" ], ack[ "request_id" ]) != snapshot[ "etag" ]
230 or not snapshot[ "etag" ] or snapshot[ "generated" ]):
231 raise fail( "creation-version-unproven" , "Original connection acknowledgement and readback versions disagree." )
232 elif target[ "type" ] == "prompt-agent-version" :
233 if (ack[ "operation" ] not in ( "agents.create_version" , "agents.create" ) or ack[ "status" ] not in ( 200 , 201 )
234 or (ack[ "operation" ] == "agents.create" and ack[ "status" ] != 200 )
235 or ack[ "version" ] != target[ "version" ] or snapshot[ "etag" ] is not None or snapshot[ "generated" ]):
236 raise fail( "ownership-unproven" , "Require original SDK version-create return and its exact readback." )
237 else :
238 raise fail( "cleanup-kind-unsupported" , "No producer receipt contract supports this target." )
239 return plan, record