Setting the file. One moment.
Blob Inventory · Foundry Iq · microsoft/azure-skills · Skills Docs
Back to the top of the page File Cu Canary
— line 303
This file
Number 10.34
Position 34 of 77
Type Python
Size 17 KB
Lines 330 helpers/ blob_inventory.py
Python · 330 lines · 17 KB
MANAGEMENT_AUDIENCE
, HelperFailure, TokenProvider, Transport, azure_cli_token,
12 digest, http_request, reject_secrets, require_allowed_fields,
13 )
14 except ImportError :
15 from _common import ( # type: ignore[no-redef]
16 MANAGEMENT_AUDIENCE , HelperFailure, TokenProvider, Transport, azure_cli_token,
17 digest, http_request, reject_secrets, require_allowed_fields,
18 )
19
20
21 STORAGE_AUDIENCE = "https://storage.azure.com/"
22 ARM_VERSION = "2025-06-01"
23 STORAGE_VERSION = "2025-05-05"
24 ACCOUNT_ID = re.compile(
25 r " ^ /subscriptions/ [ 0-9a-fA-F- ] {36} /resourceGroups/ [ ^/;?# \r\n ] + /"
26 r "providers/Microsoft \. Storage/storageAccounts/ ([ a-z0-9 ] {3,24} )$ "
27 )
28 CONTAINER = re.compile( r " ^[ a-z0-9 ](?:[ a-z0-9- ] {1,61} [ a-z0-9 ])$ " )
29
30
31 def _fail (code: str , message: str , request_id: str | None = None ) -> HelperFailure:
32 return HelperFailure(code, message, blocked_at = "source-preflight" , request_id = request_id)
33
34
35 def _strip_etag_quotes (value: str ) -> str :
36 # Blob List XML returns unquoted ETags; DFS HTTP responses return
37 # RFC 7232-quoted ETags for the identical underlying value.
38 return value[ 1 : - 1 ] if len (value) >= 2 and value[ 0 ] == '"' and value[ - 1 ] == '"' else value
39
40
41 def validate_boundary (value: Any) -> dict[ str , Any]:
42 if not isinstance (value, dict ):
43 raise _fail( "boundary-invalid" , "An explicitly selected Storage boundary is required." )
44 reject_secrets(value)
45 require_allowed_fields(value, { "storage_id" , "container" , "prefix" , "is_adls" }, label = "boundary" )
46 if (
47 not isinstance (value.get( "storage_id" ), str )
48 or ACCOUNT_ID .fullmatch(value[ "storage_id" ]) is None
49 or not isinstance (value.get( "container" ), str )
50 or CONTAINER .fullmatch(value[ "container" ]) is None
51 or "--" in value[ "container" ]
52 or not isinstance (value.get( "prefix" ), str )
53 or type (value.get( "is_adls" )) is not bool
54 ):
55 raise _fail( "boundary-invalid" , "Exact Storage ID, container, explicit prefix (including empty root), and subtype are required." )
56 prefix = value[ "prefix" ]
57 try :
58 value[ "storage_id" ].encode( "utf-8" )
59 prefix.encode( "utf-8" )
60 except UnicodeError as exc:
61 raise _fail( "boundary-invalid" , "Storage ID and prefix must be valid UTF-8." ) from exc
62 if (
63 prefix.startswith( "/" )
64 or " \\ " in prefix
65 or any ( ord (c) < 32 for c in prefix)
66 or ( any (part in { "." , ".." , "" } for part in prefix.rstrip( "/" ).split( "/" )) and prefix != "" )
67 or "//" in prefix
68 or value[ "is_adls" ] and prefix.endswith( "/" )
69 or not value[ "is_adls" ] and prefix and not prefix.endswith( "/" )
70 ):
71 raise _fail(
72 "boundary-ambiguous" ,
73 "Use explicit empty root, a Blob folder prefix ending in '/', or an ADLS directory path without a trailing '/'. Arbitrary partial-name prefixes are not verified." ,
74 )
75 return dict (value)
76
77
78 def validate_limits (value: Any) -> dict[ str , int ]:
79 if not isinstance (value, dict ):
80 raise _fail( "input-schema-invalid" , "inventory_limits must be an object." )
81 require_allowed_fields(value, { "max_pages" , "max_objects" , "max_requests" , "deadline_seconds" },
82 label = "inventory limits" )
83 for field, maximum in (
84 ( "max_pages" , 1000 ), ( "max_objects" , 100000 ), ( "max_requests" , 200000 ),
85 ( "deadline_seconds" , 3600 ),
86 ):
87 if type (value.get(field)) is not int or not 1 <= value[field] <= maximum:
88 raise _fail( "input-schema-invalid" , "Inventory limits must be bounded positive integers." )
89 return value
90
91
92 class _Reader :
93 def __init__ (
94 self, limits: dict[ str , int ], token_provider: TokenProvider,
95 transport: Transport, monotonic: Callable[[], float ],
96 deadline: float | None = None ,
97 ) -> None :
98 self .limits = limits
99 self .transport = transport
100 self .monotonic = monotonic
101 self .deadline = monotonic() + limits[ "deadline_seconds" ]
102 if deadline is not None :
103 self .deadline = min ( self .deadline, deadline)
104 self .token_provider = token_provider
105 self .tokens: dict[ str , str ] = {}
106 self .request_ids: list[ str ] = []
107 self .requests = 0
108
109 def read (self, method: str , url: str , * , arm: bool = False , raw: bool = False ) -> Any:
110 if method not in { "GET" , "HEAD" }:
111 raise _fail( "source-write-forbidden" , "Storage probes may only read." )
112 remaining = self .deadline - self .monotonic()
113 if self .requests >= self .limits[ "max_requests" ] or remaining <= 0 :
114 raise _fail( "inventory-incomplete" , "Inventory exceeded the approved request or time bound." )
115 audience = MANAGEMENT_AUDIENCE if arm else STORAGE_AUDIENCE
116 if audience not in self .tokens:
117 self .tokens[audience] = self .token_provider(audience)
118 remaining = self .deadline - self .monotonic()
119 if remaining <= 0 :
120 raise _fail( "inventory-incomplete" , "Inventory deadline elapsed during authentication." )
121 self .requests += 1
122 try :
123 result = self .transport(
124 method, url, self .tokens[audience], timeout = min ( 30 , remaining),
125 headers = {} if arm else { "x-ms-version" : STORAGE_VERSION , "Accept" : "application/xml" },
126 raw_response = raw, max_response_bytes = 8 * 1024 * 1024 , follow_redirects = False ,
127 response_deadline = time.monotonic() + remaining,
128 )
129 except HelperFailure as failure:
130 if failure.code in { "response-too-large" , "response-deadline-exceeded" }:
131 raise HelperFailure(
132 "inventory-incomplete" , "Source evidence exceeded the body or time bound." ,
133 blocked_at = "source-preflight" , request_id = failure.request_id, status = failure.http_status,
134 ) from failure
135 raise HelperFailure(
136 "source-inaccessible" , "Source evidence could not be read; inaccessible is not absent." ,
137 blocked_at = "source-preflight" , request_id = failure.request_id, status = failure.http_status,
138 ) from failure
139 if result.request_id:
140 self .request_ids.append(result.request_id)
141 if result.status != 200 :
142 raise HelperFailure(
143 "source-inaccessible" , "Source evidence could not be read; inaccessible is not absent." ,
144 blocked_at = "source-preflight" , request_id = result.request_id, status = result.status,
145 )
146 if self .monotonic() >= self .deadline:
147 raise _fail( "inventory-incomplete" , "Inventory deadline elapsed during a read." , result.request_id)
148 return result
149
150
151 def _account (boundary: dict[ str , Any], reader: _Reader) -> dict[ str , Any]:
152 response = reader.read(
153 "GET" , f "https://management.azure.com { quote(boundary[ 'storage_id' ], safe = '/' ) } ?api-version= { ARM_VERSION } " , arm = True ,
154 )
155 body = response.body
156 if not isinstance (body, dict ) or body.get( "id" ) != boundary[ "storage_id" ]:
157 raise _fail( "storage-identity-mismatch" , "Storage readback does not match the selected resource." , response.request_id)
158 properties = body.get( "properties" )
159 if not isinstance (properties, dict ):
160 raise _fail( "hns-unverified" , "Storage readback must explicitly establish HNS." , response.request_id)
161 # isHnsEnabled is set only at account creation and is immutable thereafter; ARM omits it
162 # from readback whenever it was never explicitly enabled, which documented behavior treats
163 # as a permanent, unambiguous false (Blob, not ADLS) rather than an unknown/undetermined state.
164 raw_hns = properties.get( "isHnsEnabled" )
165 if type (raw_hns) is not bool :
166 if raw_hns is not None :
167 raise _fail( "hns-unverified" , "Storage readback must explicitly establish HNS." , response.request_id)
168 hns_value = False
169 else :
170 hns_value = raw_hns
171 if hns_value != boundary[ "is_adls" ]:
172 raise _fail( "source-drift" , "Storage HNS differs from the selected subtype." , response.request_id)
173 if properties.get( "provisioningState" ) != "Succeeded" :
174 raise _fail( "storage-not-ready" , "Storage account provisioning is not complete." , response.request_id)
175 endpoints = properties.get( "primaryEndpoints" )
176 account = boundary[ "storage_id" ].rsplit( "/" , 1 )[ 1 ]
177 selected_endpoints = {}
178 for kind in ( "blob" , "dfs" ) if boundary[ "is_adls" ] else ( "blob" ,):
179 expected = f "https:// { account } . { kind } .core.windows.net/"
180 if not isinstance (endpoints, dict ) or endpoints.get(kind) != expected:
181 raise _fail( "storage-endpoint-unverified" , "Storage service endpoint is not the selected public-cloud account endpoint." , response.request_id)
182 selected_endpoints[kind] = expected
183 return {
184 "id" : body[ "id" ], "hns" : hns_value, "endpoints" : selected_endpoints,
185 "network_digest" : digest({k: properties.get(k) for k in
186 ( "publicNetworkAccess" , "networkAcls" , "privateEndpointConnections" )}),
187 }
188
189
190 def _object_name (element: Any) -> str :
191 if element is None or not isinstance (element.text, str ) or not element.text:
192 raise _fail( "inventory-invalid" , "Listed object has no exact name." )
193 name = element.text
194 if element.get( "Encoded" ) == "true" :
195 if re.search( r "% (?! [ 0-9A-Fa-f ] {2}) " , name):
196 raise _fail( "inventory-invalid" , "Listed encoded name is ambiguous." )
197 try :
198 name = unquote(name, errors = "strict" )
199 except UnicodeError as exc:
200 raise _fail( "inventory-invalid" , "Listed encoded name is invalid." ) from exc
201 elif element.get( "Encoded" ) not in { None , "false" }:
202 raise _fail( "inventory-invalid" , "Unknown listed-name encoding." )
203 if any ( ord (c) < 32 for c in name):
204 raise _fail( "inventory-invalid" , "Control characters in object names are unsupported." )
205 return name
206
207
208 def _objects (
209 boundary: dict[ str , Any], account: dict[ str , Any], reader: _Reader,
210 ) -> list[dict[ str , Any]]:
211 prefix = boundary[ "prefix" ]
212 if boundary[ "is_adls" ] and prefix:
213 prefix += "/"
214 base = account[ "endpoints" ][ "blob" ] + boundary[ "container" ]
215 marker = ""
216 markers: set[ str ] = set ()
217 objects: dict[ str , dict[ str , Any]] = {}
218 for _ in range (reader.limits[ "max_pages" ]):
219 query = { "restype" : "container" , "comp" : "list" , "prefix" : prefix,
220 "maxresults" : str ( min ( 5000 , reader.limits[ "max_objects" ])), "marker" : marker}
221 response = reader.read( "GET" , base + "?" + urlencode(query), raw = True )
222 payload = response.body
223 if not isinstance (payload, bytes ) or b "<!DOCTYPE" in payload.upper() or b "<!ENTITY" in payload.upper():
224 raise _fail( "inventory-invalid" , "Expected bounded Blob listing XML without declarations." , response.request_id)
225 try :
226 root = ElementTree.fromstring(payload)
227 except ElementTree.ParseError as exc:
228 raise _fail( "inventory-invalid" , "Blob listing XML is invalid." , response.request_id) from exc
229 if root.tag != "EnumerationResults" or len (root.findall( "Blobs" )) != 1 or len (root.findall( "NextMarker" )) != 1 :
230 raise _fail( "inventory-incomplete" , "Blob listing must contain objects and an explicit continuation marker." , response.request_id)
231 if any (child.tag != "Blob" for child in root.find( "Blobs" )):
232 raise _fail( "inventory-incomplete" , "Hierarchical listing cannot prove complete flat inventory." , response.request_id)
233 for item in root.findall( "Blobs/Blob" ):
234 name = _object_name(item.find( "Name" ))
235 etag = item.findtext( "Properties/Etag" )
236 size = item.findtext( "Properties/Content-Length" )
237 kind = item.findtext( "Properties/ResourceType" )
238 if (
239 not name.startswith(prefix) or name in objects or not etag
240 or not isinstance (size, str ) or re.fullmatch( r " [ 0-9 ] + " , size) is None
241 or boundary[ "is_adls" ] and kind not in { "file" , "directory" }
242 or boundary[ "is_adls" ] and (
243 " \\ " in name or any (part in { "" , "." , ".." } for part in name.split( "/" ))
244 )
245 or item.find( "Snapshot" ) is not None
246 ):
247 raise _fail( "inventory-invalid" , "Object identity is duplicate, outside scope, or incomplete." , response.request_id)
248 objects[name] = {
249 "path" : name, "url" : base + "/" + quote(name, safe = "/" ), "etag" : etag,
250 "size" : int (size), "version" : item.findtext( "VersionId" ),
251 "kind" : kind if boundary[ "is_adls" ] else "file" ,
252 }
253 if len (objects) > reader.limits[ "max_objects" ]:
254 raise _fail( "inventory-incomplete" , "Inventory exceeds the approved object bound." , response.request_id)
255 marker = root.findtext( "NextMarker" ) or ""
256 if not marker:
257 return [objects[name] for name in sorted (objects)]
258 if marker in markers:
259 raise _fail( "inventory-incomplete" , "Storage repeated a continuation token." , response.request_id)
260 markers.add(marker)
261 raise _fail( "inventory-incomplete" , "Unconsumed pages exceed the approved page bound." )
262
263
264 def _adls_paths (
265 boundary: dict[ str , Any], account: dict[ str , Any], objects: list[dict[ str , Any]], reader: _Reader,
266 ) -> list[dict[ str , Any]]:
267 paths: dict[ str , str ] = { "" : "directory" , boundary[ "prefix" ]: "directory" }
268 identities = {item[ "path" ]: item for item in objects}
269 for name, kind in [(boundary[ "prefix" ], "directory" )] + [(item[ "path" ], item[ "kind" ]) for item in objects]:
270 paths[name] = kind
271 parts = name.split( "/" )
272 for index in range ( 1 , len (parts)):
273 paths[ "/" .join(parts[:index])] = "directory"
274 records = []
275 for path in sorted (paths):
276 base_url = account[ "endpoints" ][ "dfs" ] + boundary[ "container" ] + "/" + quote(path, safe = "/" )
277 acl_response = reader.read( "HEAD" , base_url + "?action=getAccessControl&upn=false" )
278 headers = {key.lower(): value for key, value in acl_response.headers.items()}
279 fields = ( "x-ms-owner" , "x-ms-group" , "x-ms-permissions" , "x-ms-acl" , "etag" )
280 if any ( not headers.get(field) for field in fields):
281 raise _fail( "adls-evidence-unverified" , "Exact ADLS path type and ACL/property readback are required." , acl_response.request_id)
282 # getAccessControl never returns x-ms-resource-type; the filesystem root itself has
283 # no resource type either, so only non-root paths can be, and must be, type-verified
284 # via a separate plain getProperties HEAD.
285 if path:
286 props_response = reader.read( "HEAD" , base_url)
287 properties = {key.lower(): value for key, value in props_response.headers.items()}
288 resource_type = properties.get( "x-ms-resource-type" )
289 if resource_type != paths[path]:
290 raise _fail( "adls-evidence-unverified" , "Exact ADLS path type and ACL/property readback are required." , props_response.request_id)
291 if (
292 not properties.get( "etag" )
293 or _strip_etag_quotes(properties[ "etag" ]) != _strip_etag_quotes(headers[ "etag" ])
294 ):
295 raise _fail( "source-drift" , "ADLS access-control and properties ETags disagree." , props_response.request_id)
296 if path in identities and _strip_etag_quotes(headers[ "etag" ]) != _strip_etag_quotes(identities[path][ "etag" ]):
297 raise _fail( "source-drift" , "Blob and DFS path ETags disagree." , acl_response.request_id)
298 records.append({ "path" : path, "kind" : paths[path],
299 "properties_digest" : digest({field: headers[field] for field in fields})})
300 return records
301
302
303 def _snapshot (boundary: dict[ str , Any], reader: _Reader) -> dict[ str , Any]:
304 account = _account(boundary, reader)
305 objects = _objects(boundary, account, reader)
306 adls = _adls_paths(boundary, account, objects, reader) if boundary[ "is_adls" ] else []
307 return { "boundary" : boundary, "account" : account, "objects" : objects, "adls_paths" : adls}
308
309
310 def discover (
311 boundary: Any, limits: Any, * ,
312 token_provider: TokenProvider = azure_cli_token, transport: Transport = http_request,
313 monotonic: Callable[[], float ] = time.monotonic,
314 deadline: float | None = None ,
315 ) -> dict[ str , Any]:
316 boundary = validate_boundary(boundary)
317 reader = _Reader(validate_limits(limits), token_provider, transport, monotonic, deadline)
318 first = _snapshot(boundary, reader)
319 second = _snapshot(boundary, reader)
320 if digest(first) != digest(second):
321 raise _fail( "source-drift" , "Consecutive complete source observations differ; no stable evidence is available." )
322 return {
323 "status" : "discovered" , ** second, "inventory_digest" : digest(second),
324 "mutation" : "none" , "writes_performed" : [], "request_ids" : reader.request_ids,
325 "operator_reachability" : "verified" , "managed_ingestion_reachability" : "not-proven" ,
326 "warnings" : [
327 "Observations are not an atomic Storage snapshot or a lock; source objects can change afterward." ,
328 "ACL readback establishes observed metadata, not effective Search principal permissions." ,
329 ],
330 }