from __future__ import annotations
|
|
import json
|
import errno
|
import struct
|
import sys
|
import threading
|
from dataclasses import dataclass
|
from pathlib import Path
|
from typing import Any, BinaryIO
|
|
from .constants import MAX_FRAME_BYTES
|
from .strict_json import canonical_bytes
|
|
|
PEER_CLOSED_CODE = "E_NATIVE_PEER_CLOSED"
|
_WINDOWS_PEER_CLOSED = {38, 109, 232, 233}
|
|
|
@dataclass(frozen=True)
|
class WriteResult:
|
written: bool
|
error_code: str | None
|
|
|
def _is_peer_closed(error: BaseException) -> bool:
|
if isinstance(error, (BrokenPipeError, EOFError)):
|
return True
|
return isinstance(error, OSError) and (
|
error.errno == errno.EPIPE or getattr(error, "winerror", None) in _WINDOWS_PEER_CLOSED
|
)
|
|
|
class NativeFrameWriter:
|
"""One-owner serialized native-messaging writer with bounded peer-close semantics."""
|
|
def __init__(self, stream: BinaryIO) -> None:
|
self._stream = stream
|
self._lock = threading.Lock()
|
self._closed = False
|
|
def write_frame(self, value: dict[str, Any]) -> WriteResult:
|
payload = canonical_bytes(value)
|
frame = struct.pack("<I", len(payload)) + payload
|
with self._lock:
|
if self._closed:
|
return WriteResult(False, PEER_CLOSED_CODE)
|
try:
|
# A single write narrows the Windows WriteFile/peer-exit race.
|
self._stream.write(frame)
|
self._stream.flush()
|
except BaseException as error:
|
if not _is_peer_closed(error):
|
raise
|
self._closed = True
|
return WriteResult(False, PEER_CLOSED_CODE)
|
return WriteResult(True, None)
|
|
|
def _write_frame(value: dict[str, Any]) -> WriteResult:
|
return NativeFrameWriter(sys.stdout.buffer).write_frame(value)
|
|
|
def main() -> int:
|
"""Fail closed until reviewed build/install and visible local-load approvals exist.
|
|
No argv, environment variable, stdin JSONL, fixture or transcript can supply
|
those identities. A later reviewed installer writes the adjacent immutable
|
runtime config consumed by the packaged host.
|
"""
|
config_path = Path(sys.executable).with_name("runtime-config.json")
|
if not config_path.is_file():
|
written = _write_frame({
|
"schema_version": 1,
|
"ok": False,
|
"status": "REFRESH_FAILED_PAGE_UNREADABLE",
|
"error_code": "E_TRUSTED_ADAPTER_UNAVAILABLE",
|
"authoritative": False,
|
"saved": False,
|
"no_new": False,
|
})
|
return 4 if written.written else 0
|
written = _write_frame({
|
"schema_version": 1,
|
"ok": False,
|
"status": "REFRESH_FAILED_PAGE_UNREADABLE",
|
"error_code": "E_TRUSTED_ADAPTER_UNAVAILABLE",
|
"authoritative": False,
|
"saved": False,
|
"no_new": False,
|
})
|
return 4 if written.written else 0
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
