Setting the file. One moment. File Cu Canary
def _resolve_inventory_path
— line 329
This file
- Number
- 18.32
- Position
- 32 of 84
- Type
- Python
- Size
- 24 KB
- Lines
- 673
helpers/_file_inventory.py
Python·673 lines·24 KB
parse_qs, urlencode, urlsplit
12
13try:
14 from ._common import (
15 HelperFailure, ReadRecovery, Transport, digest, file_digest, odata_name, reject_secrets,
16 require_allowed_fields, validate_search_endpoint,
17 )
18except ImportError:
19 from _common import (
20 HelperFailure, ReadRecovery, Transport, digest, file_digest, odata_name, reject_secrets,
21 require_allowed_fields, validate_search_endpoint,
22 )
23
24
25API_VERSION = "2026-08-01-preview"
26
27
28MAX_INVENTORY_PAGES = 200
29
30
31MAX_SERVER_FILES = 200
32
33
34INVENTORY_READ_TIMEOUT_SECONDS = 60
35
36
37MAX_INVENTORY_RESPONSE_BYTES = 1024 * 1024
38
39
40MAX_FILE_BYTES = {
41 "free": 50 * 1024 * 1024,
42 "basic": 50 * 1024 * 1024,
43 "dedicated": 100 * 1024 * 1024,
44 "serverless": 100 * 1024 * 1024,
45}
46
47
48MEDIA_TYPE = re.compile(
49 r"^[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]*/[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]*$"
50)
51
52
53FILE_MEDIA_HINTS = {
54 ".txt": "text/plain", ".md": "text/markdown",
55 ".pdf": "application/pdf", ".html": "text/html", ".htm": "text/html",
56 ".csv": "text/csv", ".json": "application/json", ".sh": "application/x-sh",
57 ".doc": "application/msword",
58 ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
59 ".ppt": "application/vnd.ms-powerpoint",
60 ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
61 ".xls": "application/vnd.ms-excel",
62 ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
63 ".jpeg": "image/jpeg", ".jpg": "image/jpeg", ".png": "image/png",
64 ".bmp": "image/bmp", ".heif": "image/heif", ".heic": "image/heic",
65 ".tiff": "image/tiff", ".tif": "image/tiff", ".gif": "image/gif",
66 ".webp": "image/webp", ".svg": "image/svg+xml", ".avif": "image/avif",
67 ".ico": "image/vnd.microsoft.icon",
68}
69
70
71def validate_api_version(value: Any) -> None:
72 if value != API_VERSION:
73 raise HelperFailure(
74 "api-version-invalid",
75 f"This File helper requires {API_VERSION} for its metadata upload contract. "
76 "The service also supports 2026-05-01-preview minimal extraction; use a compatible client "
77 "or explicitly approve August, never silently change the requested API.",
78 blocked_at="input-resolution",
79 )
80
81
82def media_type_hint(path: str) -> str:
83 """Return a portable filename hint, never a claim of server content detection."""
84 return FILE_MEDIA_HINTS.get(PurePosixPath(path).suffix.lower(), "application/octet-stream")
85
86
87def _reject_credential_path(path: Path) -> None:
88 lowered = [part.lower() for part in path.parts]
89 if (
90 any(part in {".ssh", ".aws", ".azure", ".git", "credentials"} or part == ".env" or part.startswith(".env.")
91 for part in lowered)
92 or lowered[-1] in {"id_rsa", "id_dsa", "id_ecdsa", "id_ed25519"}
93 or path.suffix.lower() in {".pem", ".key", ".pfx", ".p12", ".kdbx"}
94 ):
95 raise HelperFailure(
96 "credential-file-forbidden", "Credential files are outside document ingestion.",
97 blocked_at="input-resolution",
98 )
99
100
101def _odata_name(name: Any) -> str:
102 return odata_name(name)
103
104
105def _list_url(plan: dict[str, Any]) -> str:
106 endpoint = validate_search_endpoint(plan.get("endpoint"))
107 validate_api_version(plan.get("api_version"))
108 name = _odata_name(plan.get("name"))
109 return (
110 f"{endpoint}/knowledgesources('{name}')/files?"
111 + urlencode({"api-version": API_VERSION, "pageSize": 200})
112 )
113
114
115def _normalize_inventory(files: list[dict[str, Any]]) -> list[dict[str, Any]]:
116 fields = (
117 "fileId",
118 "fileName",
119 "prefix",
120 "metadata",
121 "parsingMode",
122 "extractionMode",
123 "fileSizeBytes",
124 "errorMessage",
125 )
126 normalized = [
127 {field: item.get(field) for field in fields if field in item}
128 for item in files
129 ]
130 return sorted(
131 normalized,
132 key=lambda item: (str(item.get("fileName")), str(item.get("fileId"))),
133 )
134
135
136def _list_files(
137 url: str,
138 token: str,
139 *,
140 transport: Transport,
141 recovery: ReadRecovery | None = None,
142) -> tuple[list[dict[str, Any]], list[str]]:
143 try:
144 origin = urlsplit(url)
145 origin_port = origin.port
146 except ValueError as exc:
147 raise HelperFailure(
148 "continuation-url-invalid", "File inventory URL is malformed.",
149 blocked_at="verification",
150 ) from exc
151 current_url: str | None = url
152 seen: set[str] = set()
153 files: list[dict[str, Any]] = []
154 request_ids: list[str] = []
155 requests = 0
156 deadline = time.monotonic() + INVENTORY_READ_TIMEOUT_SECONDS
157 if recovery is not None:
158 deadline = min(deadline, recovery.deadline)
159
160 def inventory_transport(method, target, credential, **options):
161 nonlocal requests
162 if requests >= MAX_INVENTORY_PAGES:
163 raise HelperFailure(
164 "file-list-limit-exceeded", "Complete file inventory exceeds 200 pages/requests.",
165 blocked_at="verification",
166 )
167 requests += 1
168 return transport(method, target, credential, **options)
169
170 while current_url:
171 remaining = deadline - time.monotonic()
172 if remaining <= 0:
173 raise HelperFailure(
174 "file-list-timeout", "Complete file inventory read exceeded its 60-second deadline.",
175 blocked_at="verification",
176 request_id=request_ids[-1] if request_ids else None,
177 )
178 if len(seen) >= MAX_INVENTORY_PAGES:
179 raise HelperFailure(
180 "file-list-limit-exceeded", "Complete file inventory exceeds 200 pages/requests.",
181 blocked_at="verification",
182 request_id=request_ids[-1] if request_ids else None,
183 )
184 try:
185 current_url.encode("ascii")
186 current = urlsplit(current_url)
187 current_port = current.port
188 except (ValueError, UnicodeEncodeError) as exc:
189 raise HelperFailure(
190 "continuation-url-invalid", "Search returned a malformed continuation URL.",
191 blocked_at="verification",
192 ) from exc
193 query = parse_qs(current.query)
194 if (
195 current_url in seen
196 or current.scheme != "https"
197 or current.hostname != origin.hostname
198 or current_port != origin_port
199 or current.path != origin.path
200 or query.get("api-version") != [API_VERSION]
201 or current.username
202 or current.password
203 ):
204 raise HelperFailure(
205 "continuation-url-invalid",
206 "Search returned a continuation URL outside the approved service.",
207 blocked_at="verification",
208 )
209 seen.add(current_url)
210 try:
211 options = dict(timeout=remaining, response_deadline=deadline,
212 max_response_bytes=MAX_INVENTORY_RESPONSE_BYTES, follow_redirects=False)
213 result = (recovery.get(current_url, token, transport=inventory_transport,
214 max_requests=MAX_INVENTORY_PAGES - requests, **options)
215 if recovery is not None else inventory_transport("GET", current_url, token, **options))
216 except HelperFailure as failure:
217 if failure.code not in {"response-deadline-exceeded", "read-recovery-budget-exhausted"}:
218 raise
219 raise HelperFailure(
220 "file-list-timeout", "Complete file inventory read exceeded its effective deadline (at most 60 seconds).",
221 blocked_at="verification", request_id=failure.request_id,
222 status=failure.http_status, warnings=failure.warnings,
223 ) from failure
224 if time.monotonic() >= deadline:
225 raise HelperFailure(
226 "file-list-timeout", "Complete file inventory read exceeded its 60-second deadline.",
227 blocked_at="verification",
228 request_id=result.request_id,
229 )
230 if result.status != 200 or not isinstance(result.body, dict):
231 raise HelperFailure(
232 "file-list-invalid",
233 "File-list readback did not return a JSON object.",
234 blocked_at="reconciliation",
235 request_id=result.request_id,
236 status=result.status,
237 )
238 values = result.body.get("value")
239 if isinstance(values, list) and len(files) + len(values) > MAX_SERVER_FILES:
240 raise HelperFailure(
241 "file-list-limit-exceeded", "Complete file inventory exceeds 200 records.",
242 blocked_at="verification",
243 request_id=result.request_id,
244 )
245 if not isinstance(values, list) or not all(
246 isinstance(item, dict) for item in values
247 ):
248 raise HelperFailure(
249 "file-list-invalid",
250 "File-list readback did not contain an object array.",
251 blocked_at="reconciliation",
252 request_id=result.request_id,
253 )
254 files.extend(values)
255 if result.request_id:
256 request_ids.append(result.request_id)
257 next_link = result.body.get("@odata.nextLink")
258 if next_link is not None and not isinstance(next_link, str):
259 raise HelperFailure(
260 "continuation-url-invalid",
261 "Search returned an invalid continuation URL.",
262 blocked_at="verification",
263 )
264 current_url = next_link
265 return files, recovery.request_ids if recovery is not None else request_ids
266
267
268def _validate_relative_path(value: Any) -> str:
269 if not isinstance(value, str) or not value or "\\" in value:
270 raise HelperFailure(
271 "inventory-path-invalid",
272 "Every inventory path must be a non-empty normalized POSIX path.",
273 blocked_at="input-resolution",
274 )
275 path = PurePosixPath(value)
276 if (
277 path.is_absolute()
278 or path.as_posix() != value
279 or any(part in {"", ".", ".."} for part in path.parts)
280 ):
281 raise HelperFailure(
282 "inventory-path-invalid",
283 f"Inventory path is not safely relative: {value!r}.",
284 blocked_at="input-resolution",
285 )
286 if any("\r" in part or "\n" in part or ":" in part for part in path.parts):
287 raise HelperFailure(
288 "inventory-path-invalid",
289 f"Inventory path contains a forbidden segment: {value!r}.",
290 blocked_at="input-resolution",
291 )
292 return path.as_posix()
293
294
295def _reject_links(path: Path) -> None:
296 for probe in reversed((path, *path.parents)):
297 attributes = getattr(probe.lstat(), "st_file_attributes", 0)
298 if probe.is_symlink() or attributes & stat.FILE_ATTRIBUTE_REPARSE_POINT:
299 raise HelperFailure(
300 "inventory-path-invalid",
301 "Local inventory cannot traverse links or reparse points.",
302 blocked_at="input-resolution",
303 )
304
305
306def resolve_local_root(value: Any) -> Path:
307 """Resolve the explicit data boundary without following links or reparse points."""
308 if not isinstance(value, str) or not os.path.isabs(value):
309 raise HelperFailure(
310 "local-root-invalid",
311 "local_root must be an explicit absolute path.",
312 blocked_at="input-resolution",
313 )
314 try:
315 path = Path(value)
316 _reject_links(path)
317 root = path.resolve(strict=True)
318 if not root.is_dir():
319 raise OSError("not a directory")
320 except OSError as exc:
321 raise HelperFailure(
322 "local-root-invalid",
323 "local_root must be an existing readable real directory.",
324 blocked_at="input-resolution",
325 ) from exc
326 return root
327
328
329def _resolve_inventory_path(root: Path, value: Any) -> Path:
330 relative = _validate_relative_path(value)
331 path = root.joinpath(*PurePosixPath(relative).parts)
332 try:
333 _reject_links(path)
334 resolved = path.resolve(strict=True)
335 resolved.relative_to(root)
336 except (OSError, ValueError) as exc:
337 raise HelperFailure(
338 "inventory-path-invalid",
339 f"Inventory path escapes or is unreadable: {relative}.",
340 blocked_at="input-resolution",
341 ) from exc
342 if not resolved.is_file() or resolved.is_symlink():
343 raise HelperFailure(
344 "inventory-path-invalid",
345 f"Inventory entry is not a regular file: {relative}.",
346 blocked_at="input-resolution",
347 )
348 _reject_credential_path(resolved)
349 return resolved
350
351
352def _resolve_file(root: Path, record: dict[str, Any]) -> Path:
353 relative = _validate_relative_path(record.get("path"))
354 resolved = _resolve_inventory_path(root, relative)
355 try:
356 observed = resolved.stat()
357 matches = (
358 record.get("size") == observed.st_size
359 and record.get("mtime_ns") == observed.st_mtime_ns
360 and record.get("sha256") == file_digest(resolved)
361 )
362 except OSError as exc:
363 raise HelperFailure(
364 "inventory-unreadable", f"Selected file cannot be read: {relative}.",
365 blocked_at="input-resolution",
366 ) from exc
367 if not matches:
368 raise HelperFailure(
369 "inventory-drift",
370 f"Size, modification time, or SHA-256 changed for {relative}.",
371 blocked_at="confirmation",
372 )
373 return resolved
374
375
376def snapshot_inventory(
377 root: Path, paths: list[str], *, service_tier: str
378) -> list[dict[str, Any]]:
379 """Freeze selected files with MIME hints; Search determines actual support."""
380 if not isinstance(service_tier, str) or service_tier not in MAX_FILE_BYTES:
381 raise HelperFailure(
382 "service-tier-invalid", "Select a supported service_tier.",
383 blocked_at="input-resolution",
384 )
385 if not isinstance(paths, list) or not 1 <= len(paths) <= 200:
386 raise HelperFailure(
387 "inventory-invalid", "Select between 1 and 200 explicit file paths.",
388 blocked_at="input-resolution",
389 )
390 relative_paths = [_validate_relative_path(path) for path in paths]
391 if len(set(relative_paths)) != len(relative_paths):
392 raise HelperFailure(
393 "inventory-invalid", "Selected file paths must be unique.",
394 blocked_at="input-resolution",
395 )
396 records = []
397 for relative in sorted(relative_paths):
398 path = _resolve_inventory_path(root, relative)
399 media_type = media_type_hint(relative)
400 try:
401 observed = path.stat()
402 if not 0 < observed.st_size <= MAX_FILE_BYTES[service_tier]:
403 raise HelperFailure(
404 "inventory-invalid", "File size is empty or exceeds the selected tier limit.",
405 blocked_at="input-resolution",
406 )
407 record = {
408 "path": relative, "size": observed.st_size,
409 "mtime_ns": observed.st_mtime_ns, "sha256": file_digest(path),
410 "media_type": media_type,
411 }
412 _resolve_file(root, record)
413 except OSError as exc:
414 raise HelperFailure(
415 "inventory-unreadable", "Selected file cannot be read.",
416 blocked_at="input-resolution",
417 ) from exc
418 records.append(record)
419 return records
420
421
422def read_inventory(
423 plan: dict[str, Any], token: str, *, transport: Transport
424) -> tuple[list[dict[str, Any]], list[str]]:
425 """Read all pages for the exact File source, using guarded continuations."""
426 return _list_files(_list_url(plan), token, transport=transport)
427
428
429def inventory_digest(files: list[dict[str, Any]]) -> str:
430 return digest(_normalize_inventory(files))
431
432
433def reconcile_inventory(
434 plan: dict[str, Any],
435 before: list[dict[str, Any]],
436 *,
437 allow_new_uploads: bool = False,
438) -> dict[str, dict[str, Any]]:
439 """Validate all existing markers before planning or performing any upload."""
440 expected_names = {record["path"] for record in plan["files"]}
441 if any(item.get("fileName") not in expected_names for item in before):
442 raise HelperFailure(
443 "server-inventory-conflict",
444 "The server inventory contains files outside the approved inventory.",
445 blocked_at="reconciliation",
446 )
447 matched = {}
448 for record in plan["files"]:
449 matches = [item for item in before if item.get("fileName") == record["path"]]
450 if len(matches) > 1:
451 raise HelperFailure(
452 "duplicate-file-record", f"Multiple server records exist for {record['path']}.",
453 blocked_at="reconciliation",
454 )
455 if matches:
456 if not _matches(matches[0], plan, record):
457 raise HelperFailure(
458 "file-record-conflict", f"Server record conflicts with approved file {record['path']}.",
459 blocked_at="reconciliation",
460 )
461 matched[record["path"]] = matches[0]
462 elif not allow_new_uploads:
463 raise HelperFailure(
464 "reused-source-upload-forbidden",
465 "New files cannot be uploaded into a reused source because individual-file cleanup is unsupported.",
466 blocked_at="reconciliation",
467 )
468 return matched
469
470
471def _metadata(plan: dict[str, Any], record: dict[str, Any]) -> dict[str, str]:
472 supplied = record.get("metadata") or {}
473 if not isinstance(supplied, dict) or not all(
474 isinstance(key, str) and isinstance(value, str)
475 for key, value in supplied.items()
476 ):
477 raise HelperFailure(
478 "file-metadata-invalid",
479 "File metadata must contain only string keys and values.",
480 blocked_at="input-resolution",
481 )
482 metadata = dict(supplied)
483 metadata.update(
484 {
485 "foundryIqSha256": record["sha256"],
486 "foundryIqSizeBytes": str(record["size"]),
487 "foundryIqInventory": plan["inventory_digest"],
488 "foundryIqOwner": str(plan["owner"]),
489 "foundryIqSource": str(plan["name"]),
490 }
491 )
492 return metadata
493
494
495def _matches(
496 server: dict[str, Any],
497 plan: dict[str, Any],
498 record: dict[str, Any],
499) -> bool:
500 metadata = server.get("metadata") or {}
501 expected = _metadata(plan, record)
502 return (
503 server.get("fileName") == record["path"]
504 and server.get("fileSizeBytes") == record["size"]
505 and server.get("errorMessage") is None
506 and isinstance(metadata, dict)
507 and all(metadata.get(key) == value for key, value in expected.items())
508 )
509
510
511def _multipart(
512 plan: dict[str, Any],
513 record: dict[str, Any],
514 content: bytes,
515 fingerprint: str,
516) -> tuple[bytes, str]:
517 boundary = "foundry-iq-" + fingerprint.removeprefix("sha256:")[:24]
518 metadata = {
519 "fileName": record["path"],
520 "metadata": _metadata(plan, record),
521 }
522 media_type = record.get("media_type")
523 if not isinstance(media_type, str) or not media_type:
524 media_type = media_type_hint(record["path"])
525 pieces = [
526 f"--{boundary}\r\n".encode("ascii"),
527 b'Content-Disposition: form-data; name="metadata"\r\n',
528 b"Content-Type: application/json\r\n\r\n",
529 json.dumps(metadata, sort_keys=True, separators=(",", ":")).encode("utf-8"),
530 b"\r\n",
531 f"--{boundary}\r\n".encode("ascii"),
532 b'Content-Disposition: form-data; name="content"; filename="upload"\r\n',
533 f"Content-Type: {media_type}\r\n\r\n".encode("ascii"),
534 content,
535 b"\r\n",
536 f"--{boundary}--\r\n".encode("ascii"),
537 ]
538 return b"".join(pieces), boundary
539
540
541def _validate_plan(plan: dict[str, Any]) -> tuple[Path, list[dict[str, Any]]]:
542 reject_secrets(plan)
543 require_allowed_fields(
544 plan,
545 {
546 "operation",
547 "outcome",
548 "endpoint",
549 "name",
550 "api_version",
551 "local_root",
552 "files",
553 "inventory_digest",
554 "expected_server_inventory_digest",
555 "service_tier",
556 "extraction_mode",
557 "rbac",
558 "network",
559 "owner",
560 "cleanup_approved",
561 },
562 label="File ingestion plan",
563 )
564 for field, allowed in (
565 ("rbac", {"assignments"}),
566 ("network", {"posture", "evidence"}),
567 ):
568 section = plan.get(field)
569 if section is not None:
570 if not isinstance(section, dict):
571 raise HelperFailure(
572 "input-schema-invalid",
573 f"{field} must be an object.",
574 blocked_at="input-resolution",
575 )
576 require_allowed_fields(section, allowed, label=field)
577 if plan.get("operation") != "ingest" or plan.get("cleanup_approved") is not False:
578 raise HelperFailure(
579 "operation-invalid",
580 "File ingestion requires operation ingest and cleanup_approved false.",
581 blocked_at="input-resolution",
582 )
583 if not isinstance(plan.get("name"), str) or not plan["name"]:
584 raise HelperFailure(
585 "name-invalid",
586 "Knowledge source name is required.",
587 blocked_at="input-resolution",
588 )
589 if not isinstance(plan.get("owner"), str) or not plan["owner"]:
590 raise HelperFailure(
591 "owner-invalid",
592 "File ingestion owner is required.",
593 blocked_at="input-resolution",
594 )
595 root = resolve_local_root(plan.get("local_root"))
596 records = plan.get("files")
597 if not isinstance(records, list) or not records or len(records) > 200:
598 raise HelperFailure(
599 "inventory-invalid",
600 "files must contain between 1 and 200 entries.",
601 blocked_at="input-resolution",
602 )
603 if not all(isinstance(record, dict) for record in records):
604 raise HelperFailure(
605 "inventory-invalid",
606 "Every file inventory entry must be an object.",
607 blocked_at="input-resolution",
608 )
609 for record in records:
610 require_allowed_fields(
611 record,
612 {"path", "size", "mtime_ns", "sha256", "media_type", "metadata"},
613 label="File inventory record",
614 )
615 service_tier = plan.get("service_tier")
616 if service_tier not in MAX_FILE_BYTES:
617 raise HelperFailure(
618 "service-tier-invalid",
619 "service_tier must be free, basic, dedicated, or serverless.",
620 blocked_at="input-resolution",
621 )
622 extraction_mode = plan.get("extraction_mode")
623 if extraction_mode not in {"minimal", "standard"}:
624 raise HelperFailure(
625 "extraction-mode-invalid",
626 "extraction_mode must be minimal or standard.",
627 blocked_at="input-resolution",
628 )
629 paths = [_validate_relative_path(record.get("path")) for record in records]
630 if paths != sorted(paths) or len(paths) != len(set(paths)):
631 raise HelperFailure(
632 "inventory-invalid",
633 "File inventory paths must be unique and byte-sorted.",
634 blocked_at="input-resolution",
635 )
636 if digest(records) != plan.get("inventory_digest"):
637 raise HelperFailure(
638 "inventory-drift",
639 "inventory_digest does not match the approved file records.",
640 blocked_at="confirmation",
641 )
642 for record in records:
643 size = record.get("size")
644 mtime_ns = record.get("mtime_ns")
645 media_type = record.get("media_type")
646 if (
647 not isinstance(size, int)
648 or size <= 0
649 or size > MAX_FILE_BYTES[service_tier]
650 or not isinstance(mtime_ns, int)
651 or mtime_ns <= 0
652 or not isinstance(media_type, str)
653 or MEDIA_TYPE.fullmatch(media_type) is None
654 ):
655 raise HelperFailure(
656 "inventory-invalid",
657 "Every file requires positive size/mtime, media type, and a tier-valid size.",
658 blocked_at="input-resolution",
659 )
660 _resolve_file(root, record)
661 _metadata(plan, record)
662 return root, records
663
664
665def _remaining_files(resources: list[dict[str, Any]]) -> list[dict[str, Any]]:
666 return [
667 {
668 "type": "knowledge-source-file",
669 "fileName": resource["fileName"],
670 "sha256": resource["sha256"],
671 }
672 for resource in resources
673 ]