Setting the file. One moment.
Search Intake · Foundry Iq · microsoft/azure-skills · Skills Docs
Back to the top of the page File Cu Canary
helpers/ search_intake.py
Python · 303 lines · 19 KB
._common
import
(
13 SEARCH_AUDIENCE , HelperFailure, digest, emit_result, normalize_azure_location,
14 reject_secrets, require_allowed_fields, validate_search_endpoint,
15 )
16 from .bootstrap_azure import API
17 from .model_discovery import GUID , Reader, identity_key, resource_group
18 from .search_reconcile import SUPPORTED_API_VERSIONS
19 except ImportError :
20 from _bootstrap_io import failure, read_json, run_cli
21 from _common import (
22 SEARCH_AUDIENCE , HelperFailure, digest, emit_result, normalize_azure_location,
23 reject_secrets, require_allowed_fields, validate_search_endpoint,
24 )
25 from bootstrap_azure import API
26 from model_discovery import GUID , Reader, identity_key, resource_group
27 from search_reconcile import SUPPORTED_API_VERSIONS
28
29 # Azure service naming rules require both initial characters to be alphanumeric.
30 SERVICE_NAME = r " (?= . {2,60} $ ) [ a-z0-9 ][ a-z0-9 ] + (?: - [ a-z0-9 ] + ) * "
31 SERVICE_ID = re.compile(
32 rf "/subscriptions/( { GUID } )/resourceGroups/([^/]+)/providers/Microsoft\.Search/searchServices/([a-z0-9-] {{ 2,60 }} )" ,
33 re.I | re. ASCII ,
34 )
35 INTENTS = ( "use-existing" , "find-candidates" , "create-new" )
36 PROJECTION = "[]. {id:id,name:name,location:location} "
37 TIERS = { "free" : "free" , "basic" : "basic" , "serverless" : "serverless" ,
38 ** {sku: "dedicated" for sku in ( "standard" , "standard2" , "standard3" ,
39 "storage_optimized_l1" , "storage_optimized_l2" )}}
40
41
42 def fail (code, message):
43 return failure( "search-" + code, message)
44
45
46 def service_id (value):
47 match = SERVICE_ID .fullmatch(value) if isinstance (value, str ) else None
48 return match if match and resource_group(match[ 2 ]) and re.fullmatch( SERVICE_NAME , match[ 3 ].lower()) else None
49
50
51 def operation_requirements (value):
52 """Derive outbound MI from chosen helper inputs, never from a hardening profile."""
53 if value is None :
54 return None
55 if not isinstance (value, dict ):
56 raise fail( "operation-invalid" , "Supply the chosen operation, not a compatibility attestation." )
57 kind = value.get( "kind" )
58 if not isinstance (kind, str ):
59 raise fail( "operation-invalid" , "Select a named helper operation." )
60 fields = {
61 "service" : { "kind" , "api_version" },
62 "file-source" : { "kind" , "api_version" , "extraction_mode" , "vectorization" },
63 "blob-source" : { "kind" , "api_version" , "extraction_mode" , "vectorization" },
64 "knowledge-base" : { "kind" , "api_version" , "reasoning_effort" , "output_mode" },
65 }.get(kind)
66 if (fields is None or set (value) != fields or not isinstance (value.get( "api_version" ), str )
67 or value[ "api_version" ] not in SUPPORTED_API_VERSIONS ):
68 raise fail( "operation-unsupported" , "Select exactly the supported helper operation and API fields." )
69 mi = False
70 if kind in ( "file-source" , "blob-source" ):
71 if value[ "extraction_mode" ] not in ( "minimal" , "standard" ) or value[ "vectorization" ] not in ( "none" , "azureOpenAI" ):
72 raise fail( "operation-unsupported" , "Preserve explicit extraction and embedding choices." )
73 if (kind == "file-source" or value[ "extraction_mode" ] == "standard" ) and value[ "api_version" ] != "2026-08-01-preview" :
74 raise fail( "operation-unsupported" , "File and standard extraction require the supported preview API." )
75 mi = kind == "blob-source" or value[ "vectorization" ] == "azureOpenAI"
76 elif kind == "knowledge-base" :
77 if value[ "reasoning_effort" ] not in ( "minimal" , "low" , "medium" ) or value[ "output_mode" ] not in ( "extractiveData" , "answerSynthesis" ):
78 raise fail( "operation-unsupported" , "Keep KB reasoning/output separate from source embeddings and CU." )
79 mi = value[ "reasoning_effort" ] != "minimal" or value[ "output_mode" ] == "answerSynthesis"
80 if mi and value[ "api_version" ] != "2026-08-01-preview" :
81 raise fail( "operation-unsupported" , "The selected KB model mode requires preview." )
82 return mi
83
84
85 def validate (request):
86 if not isinstance (request, dict ) or request.get( "schema_version" ) != "1.0" :
87 raise fail( "input-invalid" , "Supply a schema 1.0 Search intake object." )
88 require_allowed_fields(request, { "schema_version" , "intent" , "service" , "subscription_id" ,
89 "resource_group" , "source_region" , "page" , "operation" }, label = "Search intake" )
90 reject_secrets(request)
91 intent = request.get( "intent" )
92 if intent is not None and intent not in INTENTS :
93 raise fail( "intent-invalid" , "Choose USE EXISTING, FIND CANDIDATES or CREATE NEW." )
94 sub, group = request.get( "subscription_id" ), request.get( "resource_group" )
95 if sub is not None and ( not isinstance (sub, str ) or not re.fullmatch( GUID , sub)):
96 raise fail( "scope-invalid" , "Select a subscription ID, never an account-wide scan." )
97 if group is not None and not resource_group(group):
98 raise fail( "scope-invalid" , "Select a valid resource group." )
99 if type (request.get( "page" , 0 )) is not int or not 0 <= request.get( "page" , 0 ) < 40 :
100 raise fail( "page-invalid" , "Select a zero-based presentation page below 40." )
101 region = request.get( "source_region" )
102 if region is not None and normalize_azure_location(region) is None :
103 raise fail( "region-invalid" , "Use an observed source region, not location inferred from a name." )
104 operation_requirements(request.get( "operation" ))
105 locator, name, target = request.get( "service" ), None , None
106 if locator is not None :
107 match = service_id(locator)
108 if match:
109 target = locator
110 if (sub and sub.lower() != match[ 1 ].lower()) or (group and identity_key(group) != identity_key(match[ 2 ])):
111 raise fail( "scope-conflict" , "Explicit scope disagrees with the supplied Search ID; no context switch." )
112 sub, group, name = match[ 1 ], match[ 2 ], match[ 3 ].lower()
113 elif isinstance (locator, str ) and re.fullmatch( SERVICE_NAME , locator):
114 name = locator
115 elif isinstance (locator, str ) and locator.startswith( "https://" ):
116 endpoint = validate_search_endpoint(locator)
117 name = endpoint.removeprefix( "https://" ).split( "." )[ 0 ]
118 if not re.fullmatch( SERVICE_NAME , name):
119 raise fail( "selector-invalid" , "Supply a clean Search service root, name or ARM ID." )
120 else :
121 raise fail( "selector-invalid" , "Supply a clean Search service root, name or ARM ID." )
122 intent = "create-new" if intent == "create-new" else "use-existing"
123 if request.get( "page" , 0 ) and intent != "find-candidates" :
124 raise fail( "page-invalid" , "Presentation pages apply only to FIND CANDIDATES." )
125 return intent, sub, group, name, target
126
127
128 def row (value, sub, group = None ):
129 match = service_id(value.get( "id" )) if isinstance (value, dict ) else None
130 if not match or match[ 1 ].lower() != sub.lower() or (group and identity_key(group) != identity_key(match[ 2 ])):
131 raise fail( "readback-mismatch" , "Search inventory/readback escaped the selected scope." )
132 name = value.get( "name" , match[ 3 ])
133 if not isinstance (name, str ) or name.lower() != match[ 3 ].lower():
134 raise fail( "readback-mismatch" , "Search name and returned ID disagree." )
135 location = normalize_azure_location(value.get( "location" ))
136 if location is None :
137 raise fail( "readback-invalid" , "Search region metadata is unresolved." )
138 return { "resource_id" : value[ "id" ], "name" : match[ 3 ].lower(),
139 "resource_group" : match[ 2 ], "location" : location}
140
141
142 def get_service (reader, target, sub):
143 try :
144 value = reader.call([ "rest" , "--method" , "get" , "--url" ,
145 "https://management.azure.com" + quote(target, safe = "/" ) + "?api-version=" + API ,
146 "--subscription" , sub])
147 except HelperFailure as error:
148 if error.code == "ResourceNotFound" and error.http_status in ( None , 404 ):
149 return None
150 raise
151 if not isinstance (value, dict ) or identity_key( str (value.get( "id" , "" ))) != identity_key(target):
152 raise fail( "readback-mismatch" , "Exact ARM GET returned another Search identity." )
153 return value
154
155
156 def operational_state (resource, sub, tenant, operation):
157 observed = row(resource, sub)
158 props = resource.get( "properties" )
159 identity = resource.get( "identity" )
160 if identity is None :
161 identity = {}
162 sku = resource.get( "sku" )
163 if not isinstance (props, dict ) or not isinstance (identity, dict ) or not isinstance (sku, dict ):
164 raise fail( "readback-invalid" , "Require complete Search properties, SKU and identity metadata." )
165 tier = TIERS .get( str (sku.get( "name" , "" )).lower())
166 if tier is None :
167 raise fail( "tier-unsupported" , "The chosen helpers do not recognize this Search tier; no profile conversion." )
168 if str (props.get( "provisioningState" , "" )).lower() != "succeeded" or str (props.get( "status" , "" )).lower() != "running" :
169 raise fail( "not-ready" , "The selected Search service is not running/succeeded; no replacement or repair." )
170 endpoint = validate_search_endpoint(props.get( "endpoint" ))
171 if endpoint.lower().rstrip( "/" ) != "https://" + observed[ "name" ] + ".search.windows.net" :
172 raise fail( "endpoint-mismatch" , "The Search endpoint does not match the selected resource." )
173 observed.update( endpoint = endpoint, service_tier = tier, sku = sku[ "name" ])
174 if operation is not None :
175 if str (props.get( "publicNetworkAccess" , "" )).lower() != "enabled" :
176 raise fail( "network-unsupported" , "This path cannot establish private/perimeter reachability; preserve networking and use its owner." )
177 auth = props.get( "authOptions" )
178 both = isinstance (auth, dict ) and isinstance (auth.get( "aadOrApiKey" ), dict )
179 if props.get( "disableLocalAuth" ) is not True and not both:
180 raise fail( "entra-unavailable" , "Entra data-plane authentication is absent or unresolved; never retrieve keys or change authentication." )
181 if operation_requirements(operation):
182 types = {part.strip().lower() for part in str (identity.get( "type" , "" )).split( "," )}
183 if "systemassigned" not in types:
184 raise fail( "outbound-identity-required" , "This Blob/embedding/KB-chat path requires the existing Search system identity; UAMI selection is unsupported." )
185 if not re.fullmatch( GUID , str (identity.get( "principalId" , "" ))) or str (identity.get( "tenantId" , "" )).lower() != tenant.lower():
186 raise fail( "outbound-identity-unverified" , "Search system principal/tenant readback is unresolved for the selected outbound path." )
187 observed[ "system_principal_id" ] = identity[ "principalId" ]
188 material = { "observed" : observed, "identity" : identity,
189 "access" : {key: props.get(key) for key in ( "disableLocalAuth" , "authOptions" ,
190 "publicNetworkAccess" , "networkRuleSet" , "privateEndpointConnections" )}}
191 return observed, digest(material)
192
193
194 def select (request, * , cli = run_cli, clock = time.monotonic):
195 result = { "schema_version" : "1.0" , "status" : "blocked" , "intent" : None , "writes_performed" : [], "candidates" : [],
196 "verification" : { "caller_to_search" : "unverified" , "outbound_access" : "unverified" ,
197 "operation_write_access" : "unverified" , "feature_region_support" : "unverified" },
198 "mutation_approval_required" : False , "execution_required" : False ,
199 "run_owned" : False , "first_failure" : None }
200 try :
201 intent, sub, group, name, target = validate(request)
202 result[ "intent" ] = intent
203 if intent is None :
204 result.update( status = "intent-required" , choices = list ( INTENTS ))
205 return result
206 if intent in ( "use-existing" , "create-new" ) and name is None :
207 result[ "status" ] = "service-required"
208 return result
209 if intent == "create-new" and group is None :
210 result[ "status" ] = "group-required"
211 return result
212 reader = Reader(cli, clock)
213 context = reader.call([ "account" , "show" ] + ([ "--subscription" , sub] if sub else []))
214 if ( not isinstance (context, dict ) or not re.fullmatch( GUID , str (context.get( "id" , "" )))
215 or not re.fullmatch( GUID , str (context.get( "tenantId" , "" )))
216 or (sub and context[ "id" ].lower() != sub.lower())
217 or context.get( "environmentName" ) != "AzureCloud"
218 or str (context.get( "state" , "" )).lower() != "enabled" ):
219 raise fail( "context-unavailable" , "Require the selected enabled public Azure subscription/tenant; no context switch." )
220 sub = context[ "id" ]
221 result[ "scope" ] = { "subscription_id" : sub, "resource_group" : group}
222 if target is None and name and group:
223 target = f "/subscriptions/ { sub } /resourceGroups/ { group } /providers/Microsoft.Search/searchServices/ { name } "
224 if target is None :
225 args = ([ "search" , "service" , "list" , "--resource-group" , group] if group else
226 [ "resource" , "list" , "--resource-type" , "Microsoft.Search/searchServices" ])
227 if name:
228 args += [ "--name" , name]
229 raw = reader.call(args + [ "--subscription" , sub, "--query" , PROJECTION ])
230 if not isinstance (raw, list ) or len (raw) > 200 :
231 raise fail( "inventory-limit" , "Inventory exceeds 200 rows or is malformed; narrow scope, never infer absence." )
232 rows = [row(item, sub, group) for item in raw]
233 if len ({identity_key(item[ "resource_id" ]) for item in rows}) != len (rows):
234 raise fail( "inventory-ambiguous" , "Duplicate Search IDs make inventory ambiguous." )
235 if name and any (item[ "name" ] != name for item in rows):
236 raise fail( "readback-mismatch" , "Exact-name resource lookup returned other names." )
237 if name and len (rows) == 1 :
238 target, group = rows[ 0 ][ "resource_id" ], rows[ 0 ][ "resource_group" ]
239 else :
240 region = normalize_azure_location(request.get( "source_region" ))
241 rows.sort( key =lambda item: (region is not None and item[ "location" ] != region,
242 identity_key(item[ "resource_id" ])))
243 start = request.get( "page" , 0 ) * 5
244 if start and start >= len (rows):
245 raise fail( "page-invalid" , "This page is outside the current scoped inventory; do not infer absence." )
246 for candidate in rows[start:start + 5 ]:
247 candidate[ "selection_input" ] = dict (copy.deepcopy(request), intent = "use-existing" ,
248 subscription_id = sub, resource_group = candidate[ "resource_group" ],
249 service = candidate[ "resource_id" ], page = 0 )
250 result.update( status = "candidate-choice-required" if rows else "no-candidates-in-scope" ,
251 candidates = rows[start:start + 5 ], total = len (rows),
252 remaining = max ( 0 , len (rows) - start - 5 ), unassessed = len (rows),
253 inventory_complete = True ,
254 warnings = [ "Metadata only; each page refreshes the complete bounded scope. No absence/uniqueness claim from the displayed subset. Region proximity is not compatibility." ],
255 next_page = request.get( "page" , 0 ) + 1 if start + 5 < len (rows) else None )
256 return result
257 result[ "scope" ][ "resource_group" ] = group
258 result[ "selection_input" ] = dict (copy.deepcopy(request), intent = intent, subscription_id = sub,
259 resource_group = group, service = target, page = 0 )
260 resource = get_service(reader, target, sub)
261 if intent == "create-new" :
262 if resource is not None :
263 raise fail( "name-conflict" , "CREATE NEW collided with existing Search; do not overwrite, reuse or suffix silently." )
264 result.update( status = "creation-plan-required" , resource_id = target)
265 return result
266 if resource is None :
267 raise fail( "selected-resource-absent" , "The supplied Search is absent in this exact scope; retain USE EXISTING intent." )
268 operation = request.get( "operation" )
269 observed, before = operational_state(resource, sub, context[ "tenantId" ], operation)
270 if operation is not None :
271 stats = reader.call([ "rest" , "--method" , "get" , "--url" , observed[ "endpoint" ]
272 + "/servicestats?api-version=" + operation[ "api_version" ],
273 "--resource" , SEARCH_AUDIENCE , "--subscription" , sub])
274 if not isinstance (stats, dict ) or not isinstance (stats.get( "counters" ), dict ) or not isinstance (stats.get( "limits" ), dict ):
275 raise fail( "readback-invalid" , "Search statistics did not establish an authenticated metadata read." )
276 refreshed = get_service(reader, target, sub)
277 if refreshed is None or operational_state(refreshed, sub, context[ "tenantId" ], operation)[ 1 ] != before:
278 raise fail( "state-drift" , "Search identity/access changed during readback; refresh without changing configuration." )
279 result[ "verification" ][ "caller_to_search" ] = "service-statistics GET verified"
280 result.update( status = "selected" , selected = observed, operation = copy.deepcopy(operation),
281 operation_required = operation is None ,
282 warnings = [ "Selection/read access is not source/KB write permission, feature availability, outbound RBAC/network, ingestion or retrieval proof." ])
283 except HelperFailure as error:
284 result.update( status = "blocked" , first_failure = { "code" : error.code, "message" : error.message,
285 "status" : error.http_status, "request_id" : error.request_id})
286 return result
287
288
289 def main (argv = None ):
290 parser = argparse.ArgumentParser( description = __doc__ )
291 parser.add_argument( "--select" , required = True )
292 args = parser.parse_args(argv)
293 try :
294 result = select(read_json(args.select))
295 except HelperFailure as error:
296 result = { "schema_version" : "1.0" , "status" : "blocked" , "writes_performed" : [],
297 "first_failure" : { "code" : error.code, "message" : error.message}}
298 emit_result(result)
299 return 2 if result[ "status" ] == "blocked" else 0
300
301
302 if __name__ == "__main__" :
303 raise SystemExit (main())