108 lines
4.4 KiB
Python
108 lines
4.4 KiB
Python
"""Fixed private loopback probes; does not load auth, fixtures, or patient data."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import socket
|
|
import struct
|
|
import time
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
SCRIPTS = {
|
|
"config": "private_ai_runtime_config_20261027.php",
|
|
"status-original": "private_ai_runtime_status_20261027.php",
|
|
"synthetic": "private_ai_runtime_synthetic_20261027.php",
|
|
"info": "private_ai_app_identity_20261027.php",
|
|
"status-synthetic": "private_ai_synthetic_status_20261027.php",
|
|
}
|
|
|
|
|
|
def record(kind: int, body: bytes = b"") -> bytes:
|
|
pad = (-len(body)) % 8
|
|
return struct.pack("!BBHHBB", 1, kind, 1, len(body), pad, 0) + body + b"\0" * pad
|
|
|
|
|
|
def pair(key: str, value: str) -> bytes:
|
|
key_b, value_b = key.encode(), value.encode()
|
|
def size(n: int) -> bytes:
|
|
return bytes([n]) if n < 128 else struct.pack("!I", n | 0x80000000)
|
|
return size(len(key_b)) + size(len(value_b)) + key_b + value_b
|
|
|
|
|
|
def call(action: str, profile: str, mode: str) -> dict:
|
|
script = ROOT / SCRIPTS[action]
|
|
body = (json.dumps({"profile": profile, "mode": mode}).encode() if action == "synthetic"
|
|
else json.dumps({"profile": profile}).encode() if action == "info" else b"")
|
|
params = {
|
|
"GATEWAY_INTERFACE": "CGI/1.1", "SERVER_SOFTWARE": "private-runtime-diagnostic",
|
|
"SERVER_PROTOCOL": "HTTP/1.1", "REQUEST_METHOD": "POST", "QUERY_STRING": "",
|
|
"SCRIPT_FILENAME": script.as_posix(), "SCRIPT_NAME": "/private-runtime-diagnostic.php",
|
|
"REQUEST_URI": "/private-runtime-diagnostic.php", "DOCUMENT_ROOT": ROOT.as_posix(),
|
|
"SERVER_NAME": "localhost", "SERVER_ADDR": "127.0.0.1", "SERVER_PORT": "80",
|
|
"REMOTE_ADDR": "127.0.0.1", "REMOTE_PORT": "19001", "REDIRECT_STATUS": "200",
|
|
"CONTENT_TYPE": "application/json", "CONTENT_LENGTH": str(len(body)),
|
|
}
|
|
output, stderr = bytearray(), False
|
|
started = time.monotonic()
|
|
deadline = started + 42
|
|
with socket.create_connection(("127.0.0.1", 9002), timeout=3) as sock:
|
|
def receive(n: int) -> bytes:
|
|
value = bytearray()
|
|
while len(value) < n:
|
|
left = deadline - time.monotonic()
|
|
if left <= 0:
|
|
raise TimeoutError
|
|
sock.settimeout(left)
|
|
data = sock.recv(n - len(value))
|
|
if not data:
|
|
raise ValueError("eof")
|
|
value.extend(data)
|
|
return bytes(value)
|
|
sock.sendall(record(1, struct.pack("!HB5x", 1, 0)))
|
|
sock.sendall(record(4, b"".join(pair(k, v) for k, v in params.items())) + record(4))
|
|
if body:
|
|
sock.sendall(record(5, body))
|
|
sock.sendall(record(5))
|
|
while True:
|
|
version, kind, rid, length, pad, _ = struct.unpack("!BBHHBB", receive(8))
|
|
if version != 1 or rid != 1:
|
|
raise ValueError("protocol")
|
|
content = receive(length)
|
|
if pad:
|
|
receive(pad)
|
|
if kind == 6:
|
|
if len(output) + len(content) > 65536:
|
|
raise ValueError("oversize")
|
|
output.extend(content)
|
|
elif kind == 7:
|
|
stderr = stderr or bool(content)
|
|
elif kind == 3:
|
|
break
|
|
_, separator, data = bytes(output).partition(b"\r\n\r\n")
|
|
if not separator:
|
|
_, separator, data = bytes(output).partition(b"\n\n")
|
|
parsed = json.loads(data.decode("utf-8"))
|
|
if not isinstance(parsed, dict):
|
|
raise ValueError("shape")
|
|
return {"action": action, "wall_ms": round((time.monotonic() - started) * 1000, 2),
|
|
"stderr_present": stderr, "result": parsed}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("action", choices=SCRIPTS)
|
|
parser.add_argument("--profile", choices=["qwen", "openai"], default="qwen")
|
|
parser.add_argument("--mode", choices=["complete", "cancel-before", "cancel-after", "timeout"], default="complete")
|
|
args = parser.parse_args()
|
|
try:
|
|
result = call(args.action, args.profile, args.mode)
|
|
except Exception:
|
|
# No raw stderr, PHP errors, response pages, exception prose, or identities.
|
|
result = {"action": args.action, "result": {"ok": False, "error_code": "PRIVATE_FASTCGI_CALL_FAILED"}}
|
|
print(json.dumps(result, ensure_ascii=True))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|