Setting the file. One moment. File Cu Canary
def retry_after_metadata
— line 162
This file
- Number
- 10.27
- Position
- 27 of 77
- Type
- Python
- Size
- 37 KB
- Lines
- 949
helpers/_common.py
Python·949 lines·37 KB
13
import
uuid
14from collections.abc import Mapping
15from contextlib import suppress
16from dataclasses import dataclass
17from datetime import datetime, timedelta, timezone
18from email.utils import parsedate_to_datetime
19from http.client import HTTPException
20from pathlib import Path
21from typing import Any, Callable, NamedTuple
22from urllib.error import HTTPError, URLError
23from urllib.parse import quote, urlsplit
24from urllib.request import HTTPRedirectHandler, Request, build_opener, urlopen
25
26
27SEARCH_AUDIENCE = "https://search.azure.com"
28MANAGEMENT_AUDIENCE = "https://management.azure.com"
29SEARCH_HOST = re.compile(
30 r"^[a-z0-9](?:[a-z0-9-]{0,58}[a-z0-9])?\.search\.windows\.net$"
31)
32SECRET_FIELDS = {
33 "accesskey",
34 "accesstoken",
35 "accountkey",
36 "apikey",
37 "applicationsecret",
38 "authorization",
39 "clientsecret",
40 "credential",
41 "connectionsecret",
42 "key",
43 "password",
44 "privatekey",
45 "refreshtoken",
46 "sas",
47 "sastoken",
48 "secret",
49 "sharedaccesskey",
50 "storageaccountkey",
51 "token",
52}
53RESOURCE_ID_CONNECTION = re.compile(r"^ResourceId=/[^;\r\n]+;?$")
54
55
56def normalize_azure_location(value: Any) -> str | None:
57 """Normalize ASCII case/whitespace only; this does not validate region availability."""
58 if not isinstance(value, str) or len(value) > 128 or not value.isascii():
59 return None
60 compact = re.sub(r"\s+", "", value, flags=re.ASCII).lower()
61 return compact if re.fullmatch(r"[a-z][a-z0-9]{1,40}", compact) else None
62
63
64def _normalize_response_headers(headers: Mapping[str, str]) -> dict[str, str]:
65 normalized: dict[str, str] = {}
66 for name, value in headers.items():
67 key = name.lower()
68 if key in {"x-ms-request-id", "request-id"} and normalized.get(key):
69 continue
70 if key == "retry-after" and key in normalized:
71 normalized[key] = ""
72 continue
73 normalized[key] = value
74 return normalized
75
76
77def _request_id(headers: Mapping[str, str]) -> str | None:
78 normalized = _normalize_response_headers(headers)
79 return normalized.get("x-ms-request-id") or normalized.get("request-id") or None
80
81
82@dataclass(frozen=True)
83class HttpResult:
84 status: int
85 body: Any
86 headers: dict[str, str]
87 etag_values: tuple[str, ...] | None = None
88 recovery_deadline: float | None = None
89 ack_failure: HelperFailure | None = None
90
91 @property
92 def request_id(self) -> str | None:
93 return _request_id(self.headers)
94
95 @property
96 def retry_after(self) -> RetryAfter:
97 return retry_after_metadata(self.headers)
98
99
100class RetryAfter(NamedTuple):
101 kind: str
102 value: float = 0
103
104
105class RetryAfterTiming(NamedTuple):
106 received_at_utc: float | None
107 not_before_utc: float | None
108 server_delay_seconds: int | None = None
109
110
111def valid_utc_timestamp(value):
112 return type(value) in (int, float) and -62135596800 <= value < 253402300800 and math.isfinite(value)
113
114
115def retry_after_not_before(metadata, received_at):
116 """Resolve typed metadata against UTC, without shortening a server interval."""
117 if not valid_utc_timestamp(received_at) or not isinstance(metadata, RetryAfter):
118 return None
119 value = metadata.value
120 if not valid_utc_timestamp(value):
121 return None
122 if metadata.kind == "seconds":
123 value = received_at + value if value >= 0 else math.inf
124 elif metadata.kind == "date-rfc850":
125 try:
126 epoch = datetime(1970, 1, 1, tzinfo=timezone.utc)
127 parsed = epoch + timedelta(seconds=value)
128 current_year = (epoch + timedelta(seconds=received_at)).year
129 year = current_year // 100 * 100 + parsed.year % 100
130 if year > current_year + 50:
131 year -= 100
132 value = parsed.replace(year=year).timestamp()
133 except (ValueError, OverflowError):
134 return None
135 elif metadata.kind in {"missing", "invalid"}:
136 value = received_at + 1
137 elif metadata.kind != "date":
138 return None
139 return value if valid_utc_timestamp(value) else None
140
141
142def retry_after_timing(headers, *, received_at=None):
143 """Portable timing alongside the unchanged, capped RetryAfter protocol."""
144 received_at = time.time() if received_at is None else received_at
145 if not valid_utc_timestamp(received_at):
146 return RetryAfterTiming(None, None)
147 metadata = retry_after_metadata(headers)
148 deadline = retry_after_not_before(metadata, received_at)
149 server_delay = metadata.value if metadata.kind == "seconds" else None
150 if metadata.kind == "overlong":
151 values = [value for name, value in headers.items()
152 if isinstance(name, str) and name.lower() == "retry-after"]
153 if (len(values) == 1 and isinstance(values[0], str) and len(values[0]) <= 128
154 and re.fullmatch(r"[0-9]+", values[0].strip(" \t"))):
155 seconds = int(values[0].strip(" \t"))
156 if seconds < 253402300800 - received_at:
157 deadline = received_at + seconds
158 server_delay = seconds
159 return RetryAfterTiming(received_at, deadline, server_delay)
160
161
162def retry_after_metadata(headers: Mapping[str, str]) -> RetryAfter:
163 """Retain only a bounded delay/date, never raw server header text."""
164 values = []
165 for name, value in headers.items():
166 if isinstance(name, str) and name.lower() == "retry-after":
167 values.append(value)
168 if len(values) > 1:
169 return RetryAfter("invalid")
170 if not values:
171 return RetryAfter("missing")
172 value = values[0]
173 if not isinstance(value, str):
174 return RetryAfter("invalid")
175 # An unbounded field cannot safely authorize an early request.
176 if len(value) > 128:
177 return RetryAfter("overlong")
178 if not value.isascii():
179 return RetryAfter("invalid")
180 value = value.strip(" \t")
181 if re.fullmatch(r"[0-9]+", value):
182 seconds = int(value)
183 return RetryAfter("seconds", seconds) if seconds <= 30 else RetryAfter("overlong")
184 # HTTP-date includes obsolete RFC850/asctime forms, but not arbitrary email dates.
185 if not re.fullmatch(
186 r"(?:[A-Z][a-z]{2}, [0-9]{2} [A-Z][a-z]{2} [0-9]{4} [0-9:]{8} GMT"
187 r"|[A-Z][a-z]+, [0-9]{2}-[A-Z][a-z]{2}-[0-9]{2} [0-9:]{8} GMT"
188 r"|[A-Z][a-z]{2} [A-Z][a-z]{2} [ 0-9][0-9] [0-9:]{8} [0-9]{4})", value
189 ):
190 return RetryAfter("invalid")
191 try:
192 parsed = parsedate_to_datetime(value)
193 stamp = parsed.replace(tzinfo=timezone.utc).timestamp()
194 return RetryAfter("date-rfc850" if "-" in value else "date", stamp)
195 except (TypeError, ValueError, OverflowError):
196 return RetryAfter("invalid")
197
198
199class HelperFailure(RuntimeError):
200 def __init__(
201 self,
202 code: str,
203 message: str,
204 *,
205 blocked_at: str,
206 writes: list[dict[str, Any]] | None = None,
207 resources_remaining: list[dict[str, Any]] | None = None,
208 resources_reused: list[dict[str, Any]] | None = None,
209 resources_unverified: list[dict[str, Any]] | None = None,
210 request_id: str | None = None,
211 status: int | None = None,
212 partial: bool = False,
213 warnings: list[str] | None = None,
214 retry_after: RetryAfter | None = None,
215 recovery_deadline: float | None = None,
216 retry_after_timing: RetryAfterTiming | None = None,
217 ) -> None:
218 super().__init__(message)
219 self.code = code
220 self.message = message
221 self.blocked_at = blocked_at
222 self.writes = writes or []
223 self.resources_remaining = resources_remaining or []
224 self.resources_reused = resources_reused or []
225 self.resources_unverified = resources_unverified or []
226 self.request_id = request_id
227 self.http_status = status
228 self.partial = partial
229 self.warnings = warnings or []
230 self.retry_after = retry_after or RetryAfter("missing")
231 self.recovery_deadline = recovery_deadline
232 self.retry_after_timing = retry_after_timing
233 self.response_close_failed = False
234 self.file_batch = None
235
236
237class ReadRecovery:
238 """One 429 delay opportunity and one deadline across an explicit read sequence."""
239
240 def __init__(self, *, monotonic=None, wall_clock=None, sleeper=None, deadline=None, on_wait=None):
241 self.monotonic = monotonic or time.monotonic
242 self.wall_clock = wall_clock or time.time
243 self.sleeper = sleeper or time.sleep
244 self.deadline = min(self.monotonic() + 60, deadline if deadline is not None else math.inf)
245 self.delayed = False
246 self.request_ids: list[str] = []
247 self.warnings: list[str] = []
248 self.on_wait = on_wait
249
250 @staticmethod
251 def safe_id(request_id):
252 return request_id if isinstance(request_id, str) and re.fullmatch(
253 r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", request_id
254 ) else "[withheld]"
255
256 def record(self, request_id):
257 if request_id and len(self.request_ids) < 202:
258 self.request_ids.append(self.safe_id(request_id))
259
260 def diagnostics(self):
261 return [*self.warnings, *(
262 ["Read-only request IDs: " + ", ".join(self.request_ids)] if self.request_ids else []
263 )]
264
265 def annotate(self, failure):
266 for warning in self.diagnostics():
267 if warning not in failure.warnings:
268 failure.warnings.append(warning)
269 return failure
270
271 def _stop(self, code, message):
272 self.warnings.append(message)
273 raise self.annotate(HelperFailure(
274 code, message + " Stop; resume only read-only verification, never replay the write.",
275 blocked_at="verification",
276 request_id=self.request_ids[-1] if self.request_ids else None,
277 ))
278
279 def delay(self, failure):
280 if failure.blocked_at == "local-persistence":
281 self._stop("read-recovery-persistence-failed", "Required receipt persistence failed; recovery is terminal.")
282 if failure.response_close_failed:
283 self._stop("read-recovery-response-close-failed", "HTTP error response cleanup failed; delayed recovery is blocked.")
284 if failure.http_status != 429:
285 return
286 if failure.recovery_deadline is not None:
287 self.deadline = min(self.deadline, failure.recovery_deadline)
288 if self.delayed:
289 self._stop("read-recovery-exhausted", "The single HTTP 429 delay opportunity is exhausted.")
290 metadata = failure.retry_after
291 delay = metadata.value
292 if metadata.kind in {"date", "date-rfc850"}:
293 now = self.wall_clock()
294 deadline = retry_after_not_before(metadata, now)
295 if deadline is None:
296 self._stop("read-recovery-delay-exceeded", "Retry-After date cannot be represented safely.")
297 delay = max(0, deadline - now)
298 elif metadata.kind in {"missing", "invalid"}:
299 delay = 1
300 self.warnings.append("Retry-After missing/invalid; using the fixed 1-second fallback.")
301 if metadata.kind == "overlong" or not math.isfinite(delay) or delay > 30:
302 self._stop("read-recovery-delay-exceeded", "Retry-After exceeds the 30-second wait allowance; it was not shortened.")
303 remaining = self.deadline - self.monotonic()
304 if remaining <= delay:
305 self._stop("read-recovery-budget-exhausted", "Insufficient recovery read budget for Retry-After.")
306 self.delayed = True
307 start = self.monotonic()
308 if self.on_wait is not None:
309 self.on_wait(delay)
310 self.sleeper(delay)
311 elapsed = self.monotonic() - start
312 if elapsed < delay:
313 self._stop("read-recovery-wait-incomplete", "The required Retry-After delay did not elapse.")
314 if self.monotonic() >= self.deadline:
315 self._stop("read-recovery-budget-exhausted", "Recovery read deadline elapsed during the wait.")
316
317 def get(self, url, token, *, transport, max_requests=2, **kwargs):
318 self.deadline = min(self.deadline, kwargs.get("response_deadline", math.inf))
319 first = None
320 for attempt in range(2):
321 try:
322 remaining = self.deadline - self.monotonic()
323 if remaining <= 0 or attempt >= max_requests:
324 self._stop("read-recovery-budget-exhausted", "The recovery read deadline elapsed.")
325 options = {**kwargs, "follow_redirects": False, "response_deadline": self.deadline,
326 "timeout": min(kwargs.get("timeout", 180), remaining),
327 "max_response_bytes": kwargs.get("max_response_bytes", 1024 * 1024)}
328 result = transport("GET", url, token, **options)
329 if result.recovery_deadline is not None:
330 self.deadline = min(self.deadline, result.recovery_deadline)
331 self.record(result.request_id)
332 if self.monotonic() >= self.deadline:
333 self._stop("read-recovery-budget-exhausted", "The recovery response exceeded its read deadline.")
334 if result.status != 429:
335 return result
336 raise HelperFailure(
337 "azure-http-error", "Azure request failed with HTTP 429.",
338 blocked_at="verification", status=429, request_id=result.request_id,
339 retry_after=result.retry_after,
340 recovery_deadline=result.recovery_deadline,
341 )
342 except HelperFailure as failure:
343 if not self.request_ids or self.request_ids[-1] != failure.request_id:
344 self.record(failure.request_id)
345 if failure.http_status != 429 or attempt:
346 if first is not None:
347 self.warnings.append(
348 f"Recovery read stopped ({failure.code}); delay opportunity exhausted; "
349 "initial read failure retained. Resume read-only; never replay the write."
350 )
351 raise self.annotate(first) from failure
352 raise self.annotate(failure)
353 first = failure
354 try:
355 if attempt + 1 >= max_requests:
356 self._stop("read-recovery-exhausted", "No requests remain in the recovery read allowance.")
357 self.delay(failure)
358 except HelperFailure as stopped:
359 self.warnings.append(stopped.message)
360 raise self.annotate(first) from stopped
361 raise AssertionError("unreachable")
362
363
364def canonical_bytes(value: Any) -> bytes:
365 return json.dumps(
366 value,
367 ensure_ascii=True,
368 sort_keys=True,
369 separators=(",", ":"),
370 ).encode("utf-8")
371
372
373def digest(value: Any) -> str:
374 return f"sha256:{hashlib.sha256(canonical_bytes(value)).hexdigest()}"
375
376
377def file_digest(path: Path) -> str:
378 hasher = hashlib.sha256()
379 with path.open("rb") as handle:
380 for chunk in iter(lambda: handle.read(1024 * 1024), b""):
381 hasher.update(chunk)
382 return f"sha256:{hasher.hexdigest()}"
383
384
385def load_approved_input(path: Path) -> tuple[dict[str, Any], dict[str, Any], str]:
386 try:
387 document = json.loads(path.read_text(encoding="utf-8"))
388 except OSError as exc:
389 raise HelperFailure(
390 "input-unreadable",
391 "Input file cannot be read.",
392 blocked_at="input-resolution",
393 ) from exc
394 except json.JSONDecodeError as exc:
395 raise HelperFailure(
396 "input-invalid-json",
397 f"Input file is not valid JSON: {exc}",
398 blocked_at="input-resolution",
399 ) from exc
400 if not isinstance(document, dict) or document.get("schema_version") != "1.0":
401 raise HelperFailure(
402 "input-schema-invalid",
403 "Input must be an object with schema_version 1.0.",
404 blocked_at="input-resolution",
405 )
406 reject_secrets(document)
407 require_allowed_fields(
408 document,
409 {"schema_version", "plan", "approval"},
410 label="input envelope",
411 )
412 plan = document.get("plan")
413 approval = document.get("approval")
414 if not isinstance(plan, dict) or not isinstance(approval, dict):
415 raise HelperFailure(
416 "input-schema-invalid",
417 "Input must contain plan and approval objects.",
418 blocked_at="input-resolution",
419 )
420 require_allowed_fields(
421 approval,
422 {"confirmed", "fingerprint"},
423 label="approval",
424 )
425 computed = digest(plan)
426 if approval.get("confirmed") is not True:
427 raise HelperFailure(
428 "approval-missing",
429 "The exact plan has not been explicitly approved.",
430 blocked_at="confirmation",
431 )
432 if approval.get("fingerprint") != computed:
433 raise HelperFailure(
434 "approval-mismatch",
435 "The approved fingerprint does not match the canonical plan.",
436 blocked_at="confirmation",
437 )
438 return document, plan, computed
439
440
441def reject_secrets(value: Any, *, path: str = "") -> None:
442 if isinstance(value, dict):
443 for key, child in value.items():
444 normalized = re.sub(r"[^a-z0-9]", "", str(key).casefold())
445 child_path = f"{path}/{key}"
446 if normalized in SECRET_FIELDS and child not in (None, "", []):
447 raise HelperFailure(
448 "secret-input-forbidden",
449 f"Secret-bearing field is forbidden at {child_path}.",
450 blocked_at="input-resolution",
451 )
452 if normalized == "connectionstring" and child not in (None, ""):
453 if (
454 not isinstance(child, str)
455 or RESOURCE_ID_CONNECTION.fullmatch(child) is None
456 ):
457 raise HelperFailure(
458 "secret-input-forbidden",
459 "Only one ResourceId storage connectionString component is allowed.",
460 blocked_at="input-resolution",
461 )
462 reject_secrets(child, path=child_path)
463 elif isinstance(value, list):
464 for index, child in enumerate(value):
465 reject_secrets(child, path=f"{path}/{index}")
466
467
468def require_allowed_fields(
469 value: dict[str, Any],
470 allowed: set[str],
471 *,
472 label: str,
473) -> None:
474 if set(value) - allowed:
475 raise HelperFailure(
476 "input-schema-invalid",
477 f"{label} contains unsupported fields.",
478 blocked_at="input-resolution",
479 )
480
481
482def is_ambiguous_mutation_failure(failure: HelperFailure) -> bool:
483 return (
484 failure.partial
485 or failure.code == "azure-response-ambiguous"
486 or failure.http_status in {408, 429}
487 or (
488 isinstance(failure.http_status, int)
489 and failure.http_status >= 500
490 )
491 )
492
493
494def is_ambiguous_status(status: int) -> bool:
495 return status in {408, 429} or status >= 500
496
497
498def sdk_error_status(error: Exception) -> int | None:
499 for source in (error, getattr(error, "response", None)):
500 status = getattr(source, "status_code", None)
501 if type(status) is int and 100 <= status <= 599:
502 return status
503 return None
504
505
506def sdk_error_metadata(error: Exception, fallback_code: str | None = None) -> dict[str, Any]:
507 """Read only bounded identifier fields, never exception text or response bodies."""
508 def identifier(value: Any) -> str | None:
509 if isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}", value):
510 return value
511 return None
512
513 response = getattr(error, "response", None)
514 raw_headers = getattr(response, "headers", None)
515 headers = {}
516 if isinstance(raw_headers, Mapping):
517 headers = _normalize_response_headers({
518 key: identifier(value)
519 for key, value in raw_headers.items()
520 if isinstance(key, str) and key.lower() in {"x-ms-request-id", "request-id", "x-ms-error-code"}
521 })
522 result = {"status": sdk_error_status(error), "request_id": _request_id(headers)}
523 if fallback_code is not None:
524 detail = getattr(error, "error", None)
525 code = detail.get("code") if isinstance(detail, Mapping) else getattr(detail, "code", None)
526 result["code"] = (
527 identifier(code) or identifier(getattr(error, "code", None))
528 or headers.get("x-ms-error-code") or fallback_code
529 )
530 return result
531
532
533def is_ambiguous_sdk_error(error: Exception) -> bool:
534 status = sdk_error_status(error)
535 return status is None or is_ambiguous_status(status)
536
537
538def redact_sensitive(value: Any) -> Any:
539 if isinstance(value, dict):
540 result: dict[str, Any] = {}
541 for key, child in value.items():
542 normalized = re.sub(r"[^a-z0-9]", "", str(key).casefold())
543 if normalized in SECRET_FIELDS or normalized == "connectionstring":
544 result[str(key)] = "[REDACTED]"
545 else:
546 result[str(key)] = redact_sensitive(child)
547 return result
548 if isinstance(value, list):
549 return [redact_sensitive(child) for child in value]
550 return value
551
552
553def validate_search_endpoint(endpoint: Any) -> str:
554 if not isinstance(endpoint, str):
555 raise HelperFailure(
556 "endpoint-invalid",
557 "Search endpoint must be a string.",
558 blocked_at="input-resolution",
559 )
560 try:
561 endpoint.encode("utf-8")
562 parsed = urlsplit(endpoint)
563 port = parsed.port
564 except (ValueError, UnicodeEncodeError) as exc:
565 raise HelperFailure(
566 "endpoint-invalid",
567 "Search endpoint must be a valid UTF-8 HTTPS service root.",
568 blocked_at="input-resolution",
569 ) from exc
570 if (
571 parsed.scheme != "https"
572 or not parsed.hostname
573 or SEARCH_HOST.fullmatch(parsed.hostname) is None
574 or parsed.path not in {"", "/"}
575 or parsed.query
576 or parsed.fragment
577 or parsed.username
578 or parsed.password
579 or port not in {None, 443}
580 ):
581 raise HelperFailure(
582 "endpoint-invalid",
583 "Search endpoint must be an HTTPS search.windows.net service root.",
584 blocked_at="input-resolution",
585 )
586 return endpoint.rstrip("/")
587
588
589def odata_name(name: Any) -> str:
590 if not isinstance(name, str) or not name or len(name) > 128:
591 raise HelperFailure(
592 "name-invalid",
593 "Resource name must be a non-empty string no longer than 128 characters.",
594 blocked_at="input-resolution",
595 )
596 try:
597 return quote(name.replace("'", "''"), safe="")
598 except UnicodeEncodeError as exc:
599 raise HelperFailure(
600 "name-invalid",
601 "Resource name must be valid UTF-8 text.",
602 blocked_at="input-resolution",
603 ) from exc
604
605
606def azure_cli_token(resource: str) -> str:
607 try:
608 executable = shutil.which("az")
609 if executable is None:
610 raise FileNotFoundError("Azure CLI executable was not found on PATH.")
611 completed = subprocess.run(
612 [
613 executable,
614 "account",
615 "get-access-token",
616 "--resource",
617 resource,
618 "--query",
619 "accessToken",
620 "--output",
621 "tsv",
622 ],
623 check=True,
624 capture_output=True,
625 text=True,
626 timeout=60,
627 )
628 except FileNotFoundError as exc:
629 raise HelperFailure(
630 "azure-cli-unavailable",
631 "Azure CLI is required for keyless authentication.",
632 blocked_at="execution",
633 ) from exc
634 except subprocess.CalledProcessError as exc:
635 raise HelperFailure(
636 "azure-authentication-failed",
637 "Azure CLI could not acquire the required access token.",
638 blocked_at="execution",
639 ) from exc
640 except subprocess.TimeoutExpired as exc:
641 raise HelperFailure(
642 "azure-authentication-timeout",
643 "Azure CLI did not return an access token within 60 seconds.",
644 blocked_at="execution",
645 ) from exc
646 token = completed.stdout.strip()
647 if not token:
648 raise HelperFailure(
649 "azure-authentication-failed",
650 "Azure CLI returned an empty access token.",
651 blocked_at="execution",
652 )
653 return token
654
655
656class _NoRedirect(HTTPRedirectHandler):
657 def redirect_request(self, req: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> None:
658 return None
659
660
661def _read_response_with_deadline(
662 response: Any, deadline: float, max_bytes: int, *, method: str
663) -> bytes:
664 def timed_out() -> HelperFailure:
665 return HelperFailure(
666 "response-deadline-exceeded",
667 "HTTP response body exceeded its monotonic deadline.",
668 blocked_at="verification",
669 request_id=_request_id(response.headers),
670 status=response.status,
671 partial=method not in {"GET", "HEAD"},
672 )
673
674 remaining = deadline - time.monotonic()
675 if remaining <= 0:
676 raise timed_out()
677 connection = getattr(getattr(getattr(response, "fp", None), "raw", None), "_sock", None)
678 if not isinstance(connection, socket.socket):
679 raise HelperFailure(
680 "response-deadline-unsupported",
681 "Deadline reads require an interruptible urllib HTTP socket.",
682 blocked_at="verification",
683 request_id=_request_id(response.headers),
684 status=response.status,
685 partial=method not in {"GET", "HEAD"},
686 )
687 expired = threading.Event()
688
689 def interrupt() -> None:
690 expired.set()
691 # BufferedReader.close can wait on the active read lock; interrupt its socket instead.
692 with suppress(OSError):
693 connection.shutdown(socket.SHUT_RDWR)
694 with suppress(OSError):
695 connection.close()
696
697 watchdog = threading.Timer(max(0, deadline - time.monotonic()), interrupt)
698 watchdog.start()
699 try:
700 if expired.is_set() or time.monotonic() >= deadline:
701 raise timed_out()
702 try:
703 payload = response.read(max_bytes + 1)
704 except (OSError, HTTPException, ValueError) as exc:
705 if expired.is_set() or time.monotonic() >= deadline:
706 raise timed_out() from exc
707 raise HelperFailure(
708 "response-read-failed", "HTTP response body could not be read.",
709 blocked_at="verification",
710 request_id=_request_id(response.headers),
711 status=response.status,
712 partial=method not in {"GET", "HEAD"},
713 ) from exc
714 if expired.is_set() or time.monotonic() >= deadline:
715 raise timed_out()
716 return payload
717 finally:
718 watchdog.cancel()
719 watchdog.join()
720
721
722def http_request(
723 method: str,
724 url: str,
725 token: str,
726 *,
727 body: bytes | None = None,
728 headers: dict[str, str] | None = None,
729 timeout: int = 180,
730 raw_response: bool = False,
731 max_response_bytes: int | None = None,
732 follow_redirects: bool = True,
733 response_deadline: float | None = None,
734) -> HttpResult:
735 if response_deadline is not None:
736 if (
737 not isinstance(response_deadline, (int, float))
738 or not math.isfinite(response_deadline)
739 or response_deadline - time.monotonic() > threading.TIMEOUT_MAX
740 ):
741 raise HelperFailure(
742 "response-deadline-invalid", "Response deadline must be a supported finite monotonic time.",
743 blocked_at="input-resolution",
744 )
745 if max_response_bytes is None:
746 max_response_bytes = 1024 * 1024
747 if not isinstance(max_response_bytes, int) or max_response_bytes < 0:
748 raise HelperFailure(
749 "response-limit-invalid", "Response byte limit must be a nonnegative integer.",
750 blocked_at="input-resolution",
751 )
752 if response_deadline <= time.monotonic():
753 raise HelperFailure(
754 "response-deadline-exceeded", "HTTP deadline elapsed before the request.",
755 blocked_at="verification",
756 )
757 request_headers = {
758 "Accept": "application/json;odata.metadata=minimal",
759 "Authorization": f"Bearer {token}",
760 "x-ms-client-request-id": str(uuid.uuid4()),
761 }
762 request_headers.update(headers or {})
763 request = Request(url=url, data=body, headers=request_headers, method=method)
764 try:
765 open_request = urlopen if follow_redirects else build_opener(_NoRedirect).open
766 with open_request(request, timeout=timeout) as response:
767 if response_deadline is not None:
768 payload = _read_response_with_deadline(
769 response, response_deadline, max_response_bytes, method=method
770 )
771 else:
772 payload = (
773 response.read(max_response_bytes + 1)
774 if max_response_bytes is not None
775 else response.read()
776 )
777 etag_values = tuple(value for name, value in response.headers.items() if name.lower() == "etag")
778 response_headers = _normalize_response_headers(response.headers)
779 if max_response_bytes is not None and len(payload) > max_response_bytes:
780 raise HelperFailure(
781 "response-too-large",
782 "Azure response exceeded the bounded read limit.",
783 blocked_at="verification",
784 request_id=_request_id(response_headers),
785 status=response.status,
786 partial=method not in {"GET", "HEAD"},
787 )
788 if raw_response:
789 parsed: Any = payload
790 elif not payload:
791 parsed = None
792 else:
793 try:
794 parsed = json.loads(payload.decode("utf-8"))
795 except (UnicodeDecodeError, json.JSONDecodeError) as exc:
796 raise HelperFailure(
797 "response-invalid-json",
798 "Azure returned a non-JSON response where JSON was required.",
799 blocked_at="verification",
800 request_id=_request_id(response_headers),
801 status=response.status,
802 partial=method not in {"GET", "HEAD"},
803 ) from exc
804 return HttpResult(response.status, parsed, response_headers, etag_values, response_deadline)
805 except HTTPError as exc:
806 retry_after = retry_after_metadata(exc.headers)
807 timing = retry_after_timing(exc.headers)
808 response_headers = _normalize_response_headers(exc.headers)
809 status = int(exc.code)
810 close_failed = False
811 if status == 429:
812 try:
813 exc.close()
814 except (OSError, HTTPException):
815 close_failed = True
816 elif response_deadline is not None:
817 exc.close()
818 ambiguous = method not in {"GET", "HEAD"} and is_ambiguous_status(status)
819 failure = HelperFailure(
820 "azure-http-error",
821 f"Azure request failed with HTTP {status}.",
822 blocked_at="execution",
823 request_id=_request_id(response_headers),
824 status=status,
825 partial=ambiguous,
826 retry_after=retry_after,
827 recovery_deadline=response_deadline,
828 retry_after_timing=timing,
829 warnings=(["response-close-failed: HTTP 429 response cleanup failed; "
830 "original HTTP failure retained and delayed recovery blocked."] if close_failed else []),
831 )
832 failure.response_close_failed = close_failed
833 raise failure from exc
834 except (URLError, TimeoutError) as exc:
835 raise HelperFailure(
836 "azure-response-ambiguous",
837 "Azure request did not return an authoritative response.",
838 blocked_at="verification",
839 partial=method not in {"GET", "HEAD"},
840 ) from exc
841
842
843def blocked_result(
844 failure: HelperFailure,
845 *,
846 outcome: str,
847 fingerprint: str | None,
848 owner: Any = None,
849) -> dict[str, Any]:
850 result = _blocked_result(failure, outcome=outcome, fingerprint=fingerprint, owner=owner)
851 if failure.file_batch is not None:
852 result["file_batch"] = failure.file_batch
853 result["safe_next_decision"] = (
854 "Retain the source and original ACK/journal. Do not rerun creation, replay an uncertain upload, "
855 "or delete/reset resources. Plan upload-only continuation for never-attempted files with "
856 "file_upload.py --plan; missing original evidence blocks continuation, not retention."
857 )
858 return result
859
860
861def _blocked_result(failure, *, outcome, fingerprint, owner):
862 if failure.partial or failure.writes:
863 return {
864 "status": "partial",
865 "outcome": outcome,
866 "approved_plan": {
867 "fingerprint": fingerprint,
868 "confirmed": fingerprint is not None,
869 },
870 "first_failure": {
871 "code": failure.code,
872 "operation": failure.blocked_at,
873 "status": failure.http_status,
874 "message": failure.message,
875 "request_id": failure.request_id,
876 },
877 "completed_writes": failure.writes,
878 "failed_or_unverified_postconditions": [failure.code],
879 "resources_remaining": {
880 "run_owned": failure.resources_remaining,
881 "reused": failure.resources_reused,
882 **({"unverified": failure.resources_unverified} if failure.resources_unverified else {}),
883 },
884 "rollback": {"possible": False, "exact_plan": []},
885 "cleanup": {"status": "separate-plan-and-approval-required"},
886 "owner": owner,
887 "warnings": failure.warnings,
888 }
889 result = {
890 "status": "blocked",
891 "outcome": outcome,
892 "blocked_at": failure.blocked_at,
893 "first_blocker": {
894 "code": failure.code,
895 "message": failure.message,
896 "status": failure.http_status,
897 "request_id": failure.request_id,
898 },
899 "missing_or_conflicting_input": failure.code,
900 "read_only_evidence": [],
901 "writes_performed": [],
902 "safe_next_decision": "Resolve the first blocker, rebuild the plan, and obtain new approval.",
903 "ownership": {"run_owned": [], "reused_not_owned": []},
904 "cleanup": "not applicable",
905 }
906 if fingerprint is not None:
907 result["approved_plan"] = {
908 "fingerprint": fingerprint,
909 "confirmed": True,
910 }
911 if failure.warnings:
912 result["warnings"] = failure.warnings
913 return result
914
915
916def emit_result(
917 result: dict[str, Any], *, stream: Any = None, preserve_unapproved_input: bool = False
918) -> None:
919 if stream is None:
920 stream = sys.stdout
921 safe = redact_sensitive(result)
922 if preserve_unapproved_input:
923 document = result.get("execution_input")
924 if (
925 result.get("status") != "planned" or not isinstance(document, dict)
926 or document.get("schema_version") != "1.0" or not isinstance(document.get("plan"), dict)
927 or not isinstance(document.get("approval"), dict)
928 or document["approval"].get("confirmed") is not False
929 or document.get("approval") != {"confirmed": False, "fingerprint": digest(document["plan"])}
930 ):
931 raise HelperFailure(
932 "planning-output-invalid", "Only an exact unapproved execution input can retain ResourceId bindings.",
933 blocked_at="verification",
934 )
935 reject_secrets(document)
936 require_allowed_fields(document, {"schema_version", "plan", "approval"}, label="planning envelope")
937 safe["execution_input"] = document
938 stream.write(
939 json.dumps(
940 safe,
941 sort_keys=True,
942 separators=(",", ":"),
943 )
944 + "\n"
945 )
946
947
948TokenProvider = Callable[[str], str]
949Transport = Callable[..., HttpResult]