Setting the file. One moment. File Cu Canary
def _resolve_inventory_path
— line 350
This file
- Number
- 10.48
- Position
- 48 of 77
- Type
- Python
- Size
- 40 KB
- Lines
- 1,040
helpers/file_ingest.py
Python·1,040 lines·40 KB
typing
import
Any
13from urllib.parse import parse_qs, urlencode, urlsplit
14
15try:
16 from ._progress import Progress, add_progress_argument, reporting
17except ImportError:
18 from _progress import Progress, add_progress_argument, reporting
19
20try:
21 from ._common import (
22 SEARCH_AUDIENCE,
23 HelperFailure,
24 ReadRecovery,
25 TokenProvider,
26 Transport,
27 azure_cli_token,
28 blocked_result,
29 digest,
30 emit_result,
31 file_digest,
32 http_request,
33 load_approved_input,
34 odata_name,
35 reject_secrets,
36 require_allowed_fields,
37 validate_search_endpoint,
38 )
39except ImportError:
40 from _common import ( # type: ignore[no-redef]
41 SEARCH_AUDIENCE,
42 HelperFailure,
43 ReadRecovery,
44 TokenProvider,
45 Transport,
46 azure_cli_token,
47 blocked_result,
48 digest,
49 emit_result,
50 file_digest,
51 http_request,
52 load_approved_input,
53 odata_name,
54 reject_secrets,
55 require_allowed_fields,
56 validate_search_endpoint,
57 )
58
59
60API_VERSION = "2026-08-01-preview"
61MAX_INVENTORY_PAGES = 200
62MAX_SERVER_FILES = 200
63INVENTORY_READ_TIMEOUT_SECONDS = 60
64MAX_INVENTORY_RESPONSE_BYTES = 1024 * 1024
65MAX_FILE_BYTES = {
66 "free": 50 * 1024 * 1024,
67 "basic": 50 * 1024 * 1024,
68 "dedicated": 100 * 1024 * 1024,
69 "serverless": 100 * 1024 * 1024,
70}
71MEDIA_TYPE = re.compile(
72 r"^[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]*/[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]*$"
73)
74FILE_MEDIA_HINTS = {
75 ".txt": "text/plain", ".md": "text/markdown",
76 ".pdf": "application/pdf", ".html": "text/html", ".htm": "text/html",
77 ".csv": "text/csv", ".json": "application/json", ".sh": "application/x-sh",
78 ".doc": "application/msword",
79 ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
80 ".ppt": "application/vnd.ms-powerpoint",
81 ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
82 ".xls": "application/vnd.ms-excel",
83 ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
84 ".jpeg": "image/jpeg", ".jpg": "image/jpeg", ".png": "image/png",
85 ".bmp": "image/bmp", ".heif": "image/heif", ".heic": "image/heic",
86 ".tiff": "image/tiff", ".tif": "image/tiff", ".gif": "image/gif",
87 ".webp": "image/webp", ".svg": "image/svg+xml", ".avif": "image/avif",
88 ".ico": "image/vnd.microsoft.icon",
89}
90
91
92def validate_api_version(value: Any) -> None:
93 if value != API_VERSION:
94 raise HelperFailure(
95 "api-version-invalid",
96 f"This File helper requires {API_VERSION} for its metadata upload contract. "
97 "The service also supports 2026-05-01-preview minimal extraction; use a compatible client "
98 "or explicitly approve August, never silently change the requested API.",
99 blocked_at="input-resolution",
100 )
101
102
103def media_type_hint(path: str) -> str:
104 """Return a portable filename hint, never a claim of server content detection."""
105 return FILE_MEDIA_HINTS.get(PurePosixPath(path).suffix.lower(), "application/octet-stream")
106
107
108def _reject_credential_path(path: Path) -> None:
109 lowered = [part.lower() for part in path.parts]
110 if (
111 any(part in {".ssh", ".aws", ".azure", ".git", "credentials"} or part == ".env" or part.startswith(".env.")
112 for part in lowered)
113 or lowered[-1] in {"id_rsa", "id_dsa", "id_ecdsa", "id_ed25519"}
114 or path.suffix.lower() in {".pem", ".key", ".pfx", ".p12", ".kdbx"}
115 ):
116 raise HelperFailure(
117 "credential-file-forbidden", "Credential files are outside document ingestion.",
118 blocked_at="input-resolution",
119 )
120
121
122def _odata_name(name: Any) -> str:
123 return odata_name(name)
124
125
126def _list_url(plan: dict[str, Any]) -> str:
127 endpoint = validate_search_endpoint(plan.get("endpoint"))
128 validate_api_version(plan.get("api_version"))
129 name = _odata_name(plan.get("name"))
130 return (
131 f"{endpoint}/knowledgesources('{name}')/files?"
132 + urlencode({"api-version": API_VERSION, "pageSize": 200})
133 )
134
135
136def _normalize_inventory(files: list[dict[str, Any]]) -> list[dict[str, Any]]:
137 fields = (
138 "fileId",
139 "fileName",
140 "prefix",
141 "metadata",
142 "parsingMode",
143 "extractionMode",
144 "fileSizeBytes",
145 "errorMessage",
146 )
147 normalized = [
148 {field: item.get(field) for field in fields if field in item}
149 for item in files
150 ]
151 return sorted(
152 normalized,
153 key=lambda item: (str(item.get("fileName")), str(item.get("fileId"))),
154 )
155
156
157def _list_files(
158 url: str,
159 token: str,
160 *,
161 transport: Transport,
162 recovery: ReadRecovery | None = None,
163) -> tuple[list[dict[str, Any]], list[str]]:
164 try:
165 origin = urlsplit(url)
166 origin_port = origin.port
167 except ValueError as exc:
168 raise HelperFailure(
169 "continuation-url-invalid", "File inventory URL is malformed.",
170 blocked_at="verification",
171 ) from exc
172 current_url: str | None = url
173 seen: set[str] = set()
174 files: list[dict[str, Any]] = []
175 request_ids: list[str] = []
176 requests = 0
177 deadline = time.monotonic() + INVENTORY_READ_TIMEOUT_SECONDS
178 if recovery is not None:
179 deadline = min(deadline, recovery.deadline)
180
181 def inventory_transport(method, target, credential, **options):
182 nonlocal requests
183 if requests >= MAX_INVENTORY_PAGES:
184 raise HelperFailure(
185 "file-list-limit-exceeded", "Complete file inventory exceeds 200 pages/requests.",
186 blocked_at="verification",
187 )
188 requests += 1
189 return transport(method, target, credential, **options)
190
191 while current_url:
192 remaining = deadline - time.monotonic()
193 if remaining <= 0:
194 raise HelperFailure(
195 "file-list-timeout", "Complete file inventory read exceeded its 60-second deadline.",
196 blocked_at="verification",
197 request_id=request_ids[-1] if request_ids else None,
198 )
199 if len(seen) >= MAX_INVENTORY_PAGES:
200 raise HelperFailure(
201 "file-list-limit-exceeded", "Complete file inventory exceeds 200 pages/requests.",
202 blocked_at="verification",
203 request_id=request_ids[-1] if request_ids else None,
204 )
205 try:
206 current_url.encode("ascii")
207 current = urlsplit(current_url)
208 current_port = current.port
209 except (ValueError, UnicodeEncodeError) as exc:
210 raise HelperFailure(
211 "continuation-url-invalid", "Search returned a malformed continuation URL.",
212 blocked_at="verification",
213 ) from exc
214 query = parse_qs(current.query)
215 if (
216 current_url in seen
217 or current.scheme != "https"
218 or current.hostname != origin.hostname
219 or current_port != origin_port
220 or current.path != origin.path
221 or query.get("api-version") != [API_VERSION]
222 or current.username
223 or current.password
224 ):
225 raise HelperFailure(
226 "continuation-url-invalid",
227 "Search returned a continuation URL outside the approved service.",
228 blocked_at="verification",
229 )
230 seen.add(current_url)
231 try:
232 options = dict(timeout=remaining, response_deadline=deadline,
233 max_response_bytes=MAX_INVENTORY_RESPONSE_BYTES, follow_redirects=False)
234 result = (recovery.get(current_url, token, transport=inventory_transport,
235 max_requests=MAX_INVENTORY_PAGES - requests, **options)
236 if recovery is not None else inventory_transport("GET", current_url, token, **options))
237 except HelperFailure as failure:
238 if failure.code not in {"response-deadline-exceeded", "read-recovery-budget-exhausted"}:
239 raise
240 raise HelperFailure(
241 "file-list-timeout", "Complete file inventory read exceeded its effective deadline (at most 60 seconds).",
242 blocked_at="verification", request_id=failure.request_id,
243 status=failure.http_status, warnings=failure.warnings,
244 ) from failure
245 if time.monotonic() >= deadline:
246 raise HelperFailure(
247 "file-list-timeout", "Complete file inventory read exceeded its 60-second deadline.",
248 blocked_at="verification",
249 request_id=result.request_id,
250 )
251 if result.status != 200 or not isinstance(result.body, dict):
252 raise HelperFailure(
253 "file-list-invalid",
254 "File-list readback did not return a JSON object.",
255 blocked_at="reconciliation",
256 request_id=result.request_id,
257 status=result.status,
258 )
259 values = result.body.get("value")
260 if isinstance(values, list) and len(files) + len(values) > MAX_SERVER_FILES:
261 raise HelperFailure(
262 "file-list-limit-exceeded", "Complete file inventory exceeds 200 records.",
263 blocked_at="verification",
264 request_id=result.request_id,
265 )
266 if not isinstance(values, list) or not all(
267 isinstance(item, dict) for item in values
268 ):
269 raise HelperFailure(
270 "file-list-invalid",
271 "File-list readback did not contain an object array.",
272 blocked_at="reconciliation",
273 request_id=result.request_id,
274 )
275 files.extend(values)
276 if result.request_id:
277 request_ids.append(result.request_id)
278 next_link = result.body.get("@odata.nextLink")
279 if next_link is not None and not isinstance(next_link, str):
280 raise HelperFailure(
281 "continuation-url-invalid",
282 "Search returned an invalid continuation URL.",
283 blocked_at="verification",
284 )
285 current_url = next_link
286 return files, recovery.request_ids if recovery is not None else request_ids
287
288
289def _validate_relative_path(value: Any) -> str:
290 if not isinstance(value, str) or not value or "\\" in value:
291 raise HelperFailure(
292 "inventory-path-invalid",
293 "Every inventory path must be a non-empty normalized POSIX path.",
294 blocked_at="input-resolution",
295 )
296 path = PurePosixPath(value)
297 if (
298 path.is_absolute()
299 or path.as_posix() != value
300 or any(part in {"", ".", ".."} for part in path.parts)
301 ):
302 raise HelperFailure(
303 "inventory-path-invalid",
304 f"Inventory path is not safely relative: {value!r}.",
305 blocked_at="input-resolution",
306 )
307 if any("\r" in part or "\n" in part or ":" in part for part in path.parts):
308 raise HelperFailure(
309 "inventory-path-invalid",
310 f"Inventory path contains a forbidden segment: {value!r}.",
311 blocked_at="input-resolution",
312 )
313 return path.as_posix()
314
315
316def _reject_links(path: Path) -> None:
317 for probe in reversed((path, *path.parents)):
318 attributes = getattr(probe.lstat(), "st_file_attributes", 0)
319 if probe.is_symlink() or attributes & stat.FILE_ATTRIBUTE_REPARSE_POINT:
320 raise HelperFailure(
321 "inventory-path-invalid",
322 "Local inventory cannot traverse links or reparse points.",
323 blocked_at="input-resolution",
324 )
325
326
327def resolve_local_root(value: Any) -> Path:
328 """Resolve the explicit data boundary without following links or reparse points."""
329 if not isinstance(value, str) or not os.path.isabs(value):
330 raise HelperFailure(
331 "local-root-invalid",
332 "local_root must be an explicit absolute path.",
333 blocked_at="input-resolution",
334 )
335 try:
336 path = Path(value)
337 _reject_links(path)
338 root = path.resolve(strict=True)
339 if not root.is_dir():
340 raise OSError("not a directory")
341 except OSError as exc:
342 raise HelperFailure(
343 "local-root-invalid",
344 "local_root must be an existing readable real directory.",
345 blocked_at="input-resolution",
346 ) from exc
347 return root
348
349
350def _resolve_inventory_path(root: Path, value: Any) -> Path:
351 relative = _validate_relative_path(value)
352 path = root.joinpath(*PurePosixPath(relative).parts)
353 try:
354 _reject_links(path)
355 resolved = path.resolve(strict=True)
356 resolved.relative_to(root)
357 except (OSError, ValueError) as exc:
358 raise HelperFailure(
359 "inventory-path-invalid",
360 f"Inventory path escapes or is unreadable: {relative}.",
361 blocked_at="input-resolution",
362 ) from exc
363 if not resolved.is_file() or resolved.is_symlink():
364 raise HelperFailure(
365 "inventory-path-invalid",
366 f"Inventory entry is not a regular file: {relative}.",
367 blocked_at="input-resolution",
368 )
369 _reject_credential_path(resolved)
370 return resolved
371
372
373def _resolve_file(root: Path, record: dict[str, Any]) -> Path:
374 relative = _validate_relative_path(record.get("path"))
375 resolved = _resolve_inventory_path(root, relative)
376 try:
377 observed = resolved.stat()
378 matches = (
379 record.get("size") == observed.st_size
380 and record.get("mtime_ns") == observed.st_mtime_ns
381 and record.get("sha256") == file_digest(resolved)
382 )
383 except OSError as exc:
384 raise HelperFailure(
385 "inventory-unreadable", f"Selected file cannot be read: {relative}.",
386 blocked_at="input-resolution",
387 ) from exc
388 if not matches:
389 raise HelperFailure(
390 "inventory-drift",
391 f"Size, modification time, or SHA-256 changed for {relative}.",
392 blocked_at="confirmation",
393 )
394 return resolved
395
396
397def snapshot_inventory(
398 root: Path, paths: list[str], *, service_tier: str
399) -> list[dict[str, Any]]:
400 """Freeze selected files with MIME hints; Search determines actual support."""
401 if not isinstance(service_tier, str) or service_tier not in MAX_FILE_BYTES:
402 raise HelperFailure(
403 "service-tier-invalid", "Select a supported service_tier.",
404 blocked_at="input-resolution",
405 )
406 if not isinstance(paths, list) or not 1 <= len(paths) <= 200:
407 raise HelperFailure(
408 "inventory-invalid", "Select between 1 and 200 explicit file paths.",
409 blocked_at="input-resolution",
410 )
411 relative_paths = [_validate_relative_path(path) for path in paths]
412 if len(set(relative_paths)) != len(relative_paths):
413 raise HelperFailure(
414 "inventory-invalid", "Selected file paths must be unique.",
415 blocked_at="input-resolution",
416 )
417 records = []
418 for relative in sorted(relative_paths):
419 path = _resolve_inventory_path(root, relative)
420 media_type = media_type_hint(relative)
421 try:
422 observed = path.stat()
423 if not 0 < observed.st_size <= MAX_FILE_BYTES[service_tier]:
424 raise HelperFailure(
425 "inventory-invalid", "File size is empty or exceeds the selected tier limit.",
426 blocked_at="input-resolution",
427 )
428 record = {
429 "path": relative, "size": observed.st_size,
430 "mtime_ns": observed.st_mtime_ns, "sha256": file_digest(path),
431 "media_type": media_type,
432 }
433 _resolve_file(root, record)
434 except OSError as exc:
435 raise HelperFailure(
436 "inventory-unreadable", "Selected file cannot be read.",
437 blocked_at="input-resolution",
438 ) from exc
439 records.append(record)
440 return records
441
442
443def read_inventory(
444 plan: dict[str, Any], token: str, *, transport: Transport
445) -> tuple[list[dict[str, Any]], list[str]]:
446 """Read all pages for the exact File source, using guarded continuations."""
447 return _list_files(_list_url(plan), token, transport=transport)
448
449
450def inventory_digest(files: list[dict[str, Any]]) -> str:
451 return digest(_normalize_inventory(files))
452
453
454def reconcile_inventory(
455 plan: dict[str, Any],
456 before: list[dict[str, Any]],
457 *,
458 allow_new_uploads: bool = False,
459) -> dict[str, dict[str, Any]]:
460 """Validate all existing markers before planning or performing any upload."""
461 expected_names = {record["path"] for record in plan["files"]}
462 if any(item.get("fileName") not in expected_names for item in before):
463 raise HelperFailure(
464 "server-inventory-conflict",
465 "The server inventory contains files outside the approved inventory.",
466 blocked_at="reconciliation",
467 )
468 matched = {}
469 for record in plan["files"]:
470 matches = [item for item in before if item.get("fileName") == record["path"]]
471 if len(matches) > 1:
472 raise HelperFailure(
473 "duplicate-file-record", f"Multiple server records exist for {record['path']}.",
474 blocked_at="reconciliation",
475 )
476 if matches:
477 if not _matches(matches[0], plan, record):
478 raise HelperFailure(
479 "file-record-conflict", f"Server record conflicts with approved file {record['path']}.",
480 blocked_at="reconciliation",
481 )
482 matched[record["path"]] = matches[0]
483 elif not allow_new_uploads:
484 raise HelperFailure(
485 "reused-source-upload-forbidden",
486 "New files cannot be uploaded into a reused source because individual-file cleanup is unsupported.",
487 blocked_at="reconciliation",
488 )
489 return matched
490
491
492def _metadata(plan: dict[str, Any], record: dict[str, Any]) -> dict[str, str]:
493 supplied = record.get("metadata") or {}
494 if not isinstance(supplied, dict) or not all(
495 isinstance(key, str) and isinstance(value, str)
496 for key, value in supplied.items()
497 ):
498 raise HelperFailure(
499 "file-metadata-invalid",
500 "File metadata must contain only string keys and values.",
501 blocked_at="input-resolution",
502 )
503 metadata = dict(supplied)
504 metadata.update(
505 {
506 "foundryIqSha256": record["sha256"],
507 "foundryIqSizeBytes": str(record["size"]),
508 "foundryIqInventory": plan["inventory_digest"],
509 "foundryIqOwner": str(plan["owner"]),
510 "foundryIqSource": str(plan["name"]),
511 }
512 )
513 return metadata
514
515
516def _matches(
517 server: dict[str, Any],
518 plan: dict[str, Any],
519 record: dict[str, Any],
520) -> bool:
521 metadata = server.get("metadata") or {}
522 expected = _metadata(plan, record)
523 return (
524 server.get("fileName") == record["path"]
525 and server.get("fileSizeBytes") == record["size"]
526 and server.get("errorMessage") is None
527 and isinstance(metadata, dict)
528 and all(metadata.get(key) == value for key, value in expected.items())
529 )
530
531
532def _multipart(
533 plan: dict[str, Any],
534 record: dict[str, Any],
535 content: bytes,
536 fingerprint: str,
537) -> tuple[bytes, str]:
538 boundary = "foundry-iq-" + fingerprint.removeprefix("sha256:")[:24]
539 metadata = {
540 "fileName": record["path"],
541 "metadata": _metadata(plan, record),
542 }
543 media_type = record.get("media_type")
544 if not isinstance(media_type, str) or not media_type:
545 media_type = media_type_hint(record["path"])
546 pieces = [
547 f"--{boundary}\r\n".encode("ascii"),
548 b'Content-Disposition: form-data; name="metadata"\r\n',
549 b"Content-Type: application/json\r\n\r\n",
550 json.dumps(metadata, sort_keys=True, separators=(",", ":")).encode("utf-8"),
551 b"\r\n",
552 f"--{boundary}\r\n".encode("ascii"),
553 b'Content-Disposition: form-data; name="content"; filename="upload"\r\n',
554 f"Content-Type: {media_type}\r\n\r\n".encode("ascii"),
555 content,
556 b"\r\n",
557 f"--{boundary}--\r\n".encode("ascii"),
558 ]
559 return b"".join(pieces), boundary
560
561
562def _validate_plan(plan: dict[str, Any]) -> tuple[Path, list[dict[str, Any]]]:
563 reject_secrets(plan)
564 require_allowed_fields(
565 plan,
566 {
567 "operation",
568 "outcome",
569 "endpoint",
570 "name",
571 "api_version",
572 "local_root",
573 "files",
574 "inventory_digest",
575 "expected_server_inventory_digest",
576 "service_tier",
577 "extraction_mode",
578 "rbac",
579 "network",
580 "owner",
581 "cleanup_approved",
582 },
583 label="File ingestion plan",
584 )
585 for field, allowed in (
586 ("rbac", {"assignments"}),
587 ("network", {"posture", "evidence"}),
588 ):
589 section = plan.get(field)
590 if section is not None:
591 if not isinstance(section, dict):
592 raise HelperFailure(
593 "input-schema-invalid",
594 f"{field} must be an object.",
595 blocked_at="input-resolution",
596 )
597 require_allowed_fields(section, allowed, label=field)
598 if plan.get("operation") != "ingest" or plan.get("cleanup_approved") is not False:
599 raise HelperFailure(
600 "operation-invalid",
601 "File ingestion requires operation ingest and cleanup_approved false.",
602 blocked_at="input-resolution",
603 )
604 if not isinstance(plan.get("name"), str) or not plan["name"]:
605 raise HelperFailure(
606 "name-invalid",
607 "Knowledge source name is required.",
608 blocked_at="input-resolution",
609 )
610 if not isinstance(plan.get("owner"), str) or not plan["owner"]:
611 raise HelperFailure(
612 "owner-invalid",
613 "File ingestion owner is required.",
614 blocked_at="input-resolution",
615 )
616 root = resolve_local_root(plan.get("local_root"))
617 records = plan.get("files")
618 if not isinstance(records, list) or not records or len(records) > 200:
619 raise HelperFailure(
620 "inventory-invalid",
621 "files must contain between 1 and 200 entries.",
622 blocked_at="input-resolution",
623 )
624 if not all(isinstance(record, dict) for record in records):
625 raise HelperFailure(
626 "inventory-invalid",
627 "Every file inventory entry must be an object.",
628 blocked_at="input-resolution",
629 )
630 for record in records:
631 require_allowed_fields(
632 record,
633 {"path", "size", "mtime_ns", "sha256", "media_type", "metadata"},
634 label="File inventory record",
635 )
636 service_tier = plan.get("service_tier")
637 if service_tier not in MAX_FILE_BYTES:
638 raise HelperFailure(
639 "service-tier-invalid",
640 "service_tier must be free, basic, dedicated, or serverless.",
641 blocked_at="input-resolution",
642 )
643 extraction_mode = plan.get("extraction_mode")
644 if extraction_mode not in {"minimal", "standard"}:
645 raise HelperFailure(
646 "extraction-mode-invalid",
647 "extraction_mode must be minimal or standard.",
648 blocked_at="input-resolution",
649 )
650 paths = [_validate_relative_path(record.get("path")) for record in records]
651 if paths != sorted(paths) or len(paths) != len(set(paths)):
652 raise HelperFailure(
653 "inventory-invalid",
654 "File inventory paths must be unique and byte-sorted.",
655 blocked_at="input-resolution",
656 )
657 if digest(records) != plan.get("inventory_digest"):
658 raise HelperFailure(
659 "inventory-drift",
660 "inventory_digest does not match the approved file records.",
661 blocked_at="confirmation",
662 )
663 for record in records:
664 size = record.get("size")
665 mtime_ns = record.get("mtime_ns")
666 media_type = record.get("media_type")
667 if (
668 not isinstance(size, int)
669 or size <= 0
670 or size > MAX_FILE_BYTES[service_tier]
671 or not isinstance(mtime_ns, int)
672 or mtime_ns <= 0
673 or not isinstance(media_type, str)
674 or MEDIA_TYPE.fullmatch(media_type) is None
675 ):
676 raise HelperFailure(
677 "inventory-invalid",
678 "Every file requires positive size/mtime, media type, and a tier-valid size.",
679 blocked_at="input-resolution",
680 )
681 _resolve_file(root, record)
682 _metadata(plan, record)
683 return root, records
684
685
686def _confirm_ambiguous_upload(
687 list_url: str,
688 token: str,
689 transport: Transport,
690 record: dict[str, Any],
691 plan: dict[str, Any],
692 request_ids: list[str],
693 failure: HelperFailure,
694 warnings: list[str],
695) -> dict[str, Any] | None:
696 """Resolve an ambiguous upload with bounded readback, never upload replay.
697
698 Only ambiguous outcomes (transport failure, timeout, 409/429/5xx) reach
699 this helper. A definitive 200/201 response never calls it. Returns the
700 matching server record when the readback proves the approved file
701 exists, otherwise ``None`` so the caller reports ``partial``.
702 """
703 recovery = ReadRecovery()
704 try:
705 recovery.delay(failure)
706 observed, _ = _list_files(
707 list_url, token, transport=transport, recovery=recovery,
708 )
709 except HelperFailure as read_failure:
710 warnings.append(f"Upload readback failed ({read_failure.code}); original upload failure retained.")
711 return None
712 finally:
713 request_ids.extend(recovery.request_ids)
714 warnings.extend(recovery.diagnostics())
715 matches = [item for item in observed if item.get("fileName") == record["path"]]
716 if len(matches) != 1 or not _matches(matches[0], plan, record):
717 return None
718 return matches[0]
719
720
721def _remaining_files(resources: list[dict[str, Any]]) -> list[dict[str, Any]]:
722 return [
723 {
724 "type": "knowledge-source-file",
725 "fileName": resource["fileName"],
726 "sha256": resource["sha256"],
727 }
728 for resource in resources
729 ]
730
731
732@reporting("file-upload")
733def execute(
734 document: dict[str, Any],
735 *,
736 token_provider: TokenProvider = azure_cli_token,
737 transport: Transport = http_request,
738 allow_new_uploads: bool = False,
739 progress: Progress | None = None,
740 upload_session=None,
741 source_check=None,
742 allow_upload_retry=True,
743) -> dict[str, Any]:
744 if allow_new_uploads or upload_session is not None or len(document.get("plan", {}).get("files", [])) > 1:
745 try:
746 from .file_upload import run_batch
747 except ImportError:
748 from file_upload import run_batch
749 return run_batch(document, token_provider=token_provider, transport=transport, progress=progress,
750 allow_new_uploads=allow_new_uploads, session=upload_session,
751 source_check=source_check, allow_upload_retry=allow_upload_retry)
752 progress.update("file-inventory")
753 plan = document["plan"]
754 fingerprint = document["_computed_fingerprint"]
755 root, records = _validate_plan(plan)
756 list_url = _list_url(plan)
757 token = token_provider(SEARCH_AUDIENCE)
758 readonly = ReadRecovery() if not allow_new_uploads else None
759 before, request_ids = _list_files(list_url, token, transport=transport, recovery=readonly)
760 warnings: list[str] = list(readonly.warnings) if readonly is not None else []
761 if inventory_digest(before) != plan.get("expected_server_inventory_digest"):
762 raise HelperFailure(
763 "server-inventory-drift",
764 "Server file inventory changed after approval.",
765 blocked_at="reconciliation",
766 warnings=readonly.diagnostics() if readonly is not None else [],
767 )
768 matched = reconcile_inventory(plan, before, allow_new_uploads=allow_new_uploads)
769
770 created: list[dict[str, Any]] = []
771 reused: list[dict[str, Any]] = []
772 acknowledged_ids: list[str] = []
773 progress.update("file-upload", uploads_acknowledged=0, files_reused=0)
774 for record in records:
775 progress.update("file-upload", uploads_acknowledged=len(created), files_reused=len(reused))
776 if record["path"] in matched:
777 reused.append(
778 {
779 "fileId": matched[record["path"]].get("fileId"),
780 "fileName": record["path"],
781 "sha256": record["sha256"],
782 }
783 )
784 continue
785
786 path = _resolve_file(root, record)
787 try:
788 content = path.read_bytes()
789 except OSError as exc:
790 raise HelperFailure(
791 "inventory-unreadable",
792 f"Approved file became unreadable: {record['path']}.",
793 blocked_at="execution",
794 writes=created,
795 resources_remaining=_remaining_files(created),
796 partial=bool(created),
797 ) from exc
798 try:
799 post_read_stat = path.stat()
800 except OSError as exc:
801 raise HelperFailure(
802 "inventory-unreadable",
803 f"Approved file became unreadable: {record['path']}.",
804 blocked_at="execution",
805 writes=created,
806 resources_remaining=_remaining_files(created),
807 partial=bool(created),
808 ) from exc
809 content_digest = "sha256:" + hashlib.sha256(content).hexdigest()
810 if (
811 content_digest != record["sha256"]
812 or len(content) != record["size"]
813 or post_read_stat.st_mtime_ns != record["mtime_ns"]
814 ):
815 raise HelperFailure(
816 "inventory-drift",
817 f"Approved file changed before upload: {record['path']}.",
818 blocked_at="confirmation",
819 writes=created,
820 resources_remaining=_remaining_files(created),
821 partial=bool(created),
822 )
823 body, boundary = _multipart(plan, record, content, fingerprint)
824 try:
825 result = transport(
826 "POST",
827 list_url,
828 token,
829 body=body,
830 headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
831 )
832 except HelperFailure as failure:
833 if failure.http_status is not None and 400 <= failure.http_status < 500 and failure.http_status not in {408, 409, 429}:
834 raise HelperFailure(
835 failure.code, failure.message, blocked_at=failure.blocked_at,
836 writes=created, resources_remaining=_remaining_files(created),
837 request_id=failure.request_id, status=failure.http_status, partial=bool(created),
838 ) from failure
839 # The transport outcome is ambiguous (we do not know whether the
840 # server received the write): attempt one readback before
841 # concluding partial, per the ambiguous-write contract.
842 confirmed = _confirm_ambiguous_upload(
843 list_url, token, transport, record, plan, request_ids, failure, warnings
844 )
845 if confirmed is not None:
846 created.append({"fileName": record["path"], "sha256": record["sha256"]})
847 if failure.request_id:
848 request_ids.append(failure.request_id)
849 continue
850 uncertain = {
851 "action": "upload-unverified",
852 "type": "knowledge-source-file",
853 "fileName": record["path"],
854 "sha256": record["sha256"],
855 }
856 raise HelperFailure(
857 failure.code,
858 failure.message,
859 blocked_at=failure.blocked_at,
860 writes=created + [uncertain],
861 resources_remaining=_remaining_files(created) + [uncertain],
862 request_id=failure.request_id,
863 status=failure.http_status,
864 partial=True,
865 warnings=warnings,
866 ) from failure
867 if result.status not in {200, 201}:
868 ambiguous = result.status in {408, 409, 429} or result.status >= 500
869 if ambiguous:
870 confirmed = _confirm_ambiguous_upload(
871 list_url, token, transport, record, plan, request_ids,
872 HelperFailure("upload-failed", "Upload response is ambiguous.",
873 blocked_at="execution", status=result.status,
874 request_id=result.request_id, retry_after=result.retry_after,
875 recovery_deadline=result.recovery_deadline),
876 warnings,
877 )
878 if confirmed is not None:
879 created.append(
880 {"fileName": record["path"], "sha256": record["sha256"]}
881 )
882 if result.request_id:
883 request_ids.append(result.request_id)
884 continue
885 uncertain = {
886 "action": "upload-unverified",
887 "type": "knowledge-source-file",
888 "fileName": record["path"],
889 "sha256": record["sha256"],
890 }
891 raise HelperFailure(
892 "upload-failed",
893 f"Upload returned unexpected HTTP {result.status}.",
894 blocked_at="execution",
895 writes=created + ([uncertain] if ambiguous else []),
896 resources_remaining=(
897 _remaining_files(created) + ([uncertain] if ambiguous else [])
898 ),
899 request_id=result.request_id,
900 status=result.status,
901 partial=bool(created) or ambiguous,
902 warnings=warnings,
903 )
904 # A definitive 200/201 response is not ambiguous: trust it rather than
905 # re-listing the whole source after every single file. The complete
906 # inventory is verified once, in bulk, after the loop.
907 created.append({"fileName": record["path"], "sha256": record["sha256"]})
908 if result.request_id:
909 request_ids.append(result.request_id)
910 acknowledged_ids.append(ReadRecovery.safe_id(result.request_id))
911
912 progress.update("file-readback", uploads_acknowledged=len(created), files_reused=len(reused))
913 ack_warnings = (["Acknowledged upload request IDs: " + ", ".join(acknowledged_ids)]
914 if acknowledged_ids else [])
915 recovery = ReadRecovery()
916 try:
917 after, after_request_ids = _list_files(
918 list_url, token, transport=transport, recovery=recovery,
919 )
920 except HelperFailure as failure:
921 raise HelperFailure(
922 failure.code,
923 failure.message,
924 blocked_at=failure.blocked_at,
925 writes=created + failure.writes,
926 resources_remaining=(
927 _remaining_files(created) + failure.resources_remaining
928 ),
929 resources_reused=failure.resources_reused,
930 resources_unverified=failure.resources_unverified,
931 warnings=[*warnings, *ack_warnings, *failure.warnings, *recovery.diagnostics()],
932 request_id=failure.request_id,
933 status=failure.http_status,
934 partial=bool(created or failure.partial),
935 ) from failure
936 request_ids.extend(after_request_ids)
937 warnings.extend(recovery.warnings)
938 if len(after) != len(records):
939 raise HelperFailure(
940 "readback-mismatch",
941 "Final server inventory count differs from the approved inventory.",
942 blocked_at="verification",
943 writes=created,
944 resources_remaining=_remaining_files(created),
945 request_id=request_ids[-1] if request_ids else None,
946 partial=bool(created),
947 warnings=[*warnings, *ack_warnings, *recovery.diagnostics()],
948 )
949 verified: list[dict[str, Any]] = []
950 for record in records:
951 matches = [item for item in after if item.get("fileName") == record["path"]]
952 if len(matches) != 1 or not _matches(matches[0], plan, record):
953 raise HelperFailure(
954 "readback-mismatch",
955 f"File readback failed for {record['path']}.",
956 blocked_at="verification",
957 writes=created,
958 resources_remaining=_remaining_files(created),
959 request_id=request_ids[-1] if request_ids else None,
960 partial=bool(created),
961 warnings=[*warnings, *ack_warnings, *recovery.diagnostics()],
962 )
963 verified.append(
964 {
965 "fileId": matches[0].get("fileId"),
966 "fileName": record["path"],
967 "sha256": record["sha256"],
968 "size": record["size"],
969 }
970 )
971
972 progress.update("file-readback", files_verified=len(verified))
973 return {
974 "status": "completed",
975 "outcome": str(plan.get("outcome") or "file-knowledge-source-ingestion"),
976 "approved_plan": {"fingerprint": fingerprint, "confirmed": True},
977 "resources": {
978 "created": created,
979 "reused": reused,
980 "updated": [],
981 "skipped": [],
982 },
983 "api_contracts": [
984 {"operation": "upload-file", "version": API_VERSION, "preview": True}
985 ],
986 "data_movement": {
987 "boundary": {"local_root_digest": digest(str(root))},
988 "result": "exact approved files uploaded directly to Search",
989 },
990 "auth": {"mode": "entra-user", "principals": []},
991 "rbac": plan.get("rbac", {"assignments": []}),
992 "network": plan.get("network", {"posture": "preserved", "evidence": None}),
993 "verification": {
994 "readback": verified,
995 "server_inventory_digest": digest(_normalize_inventory(after)),
996 "request_ids": request_ids,
997 "idempotency": "matching marker metadata is zero-write",
998 },
999 "warnings": warnings,
1000 "ownership": {
1001 "run_owned": created,
1002 "reused_not_owned": reused,
1003 "owner": plan.get("owner"),
1004 },
1005 "cleanup": {
1006 "status": "not-requested",
1007 "separate_confirmation_required": True,
1008 },
1009 }
1010
1011
1012def main(argv: list[str] | None = None) -> int:
1013 parser = argparse.ArgumentParser()
1014 parser.add_argument("--input", type=Path, required=True)
1015 add_progress_argument(parser)
1016 args = parser.parse_args(argv)
1017 fingerprint: str | None = None
1018 owner: Any = None
1019 outcome = "file-knowledge-source-ingestion"
1020 try:
1021 document, plan, fingerprint = load_approved_input(args.input)
1022 document["_computed_fingerprint"] = fingerprint
1023 owner = plan.get("owner")
1024 outcome = str(plan.get("outcome") or outcome)
1025 result = execute(document, progress=Progress("file-upload", enabled=args.progress))
1026 except HelperFailure as failure:
1027 result = blocked_result(
1028 failure,
1029 outcome=outcome,
1030 fingerprint=fingerprint,
1031 owner=owner,
1032 )
1033 emit_result(result)
1034 return 3 if result["status"] == "partial" else 2
1035 emit_result(result)
1036 return 0
1037
1038
1039if __name__ == "__main__":
1040 sys.exit(main())