353 lines
12 KiB
Python
353 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Capture authenticated, metadata-only G7 voice runtime snapshots.
|
|
|
|
The sampler calls the admin-only ``/admin/voice-runtime`` endpoint. It never
|
|
writes the session cookie, audio, transcripts, session IDs, or provider payloads.
|
|
Every sample must come from the same process-local worker instance.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import os
|
|
import ssl
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
from urllib.parse import urlsplit
|
|
|
|
|
|
SCHEMA_VERSION = "vignette.g7-runtime-sampling.v1"
|
|
SNAPSHOT_SCHEMA_VERSION = "vignette.voice-runtime.v1"
|
|
MAX_SAMPLES = 7_200
|
|
MAX_INTERVAL_SECONDS = 60.0
|
|
MAX_CAPTURE_WINDOW_SECONDS = 7_200.0
|
|
_BROWSER_UA = (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/140.0 Safari/537.36"
|
|
)
|
|
|
|
|
|
class EvidenceFailure(RuntimeError):
|
|
"""A runtime sampling or privacy contract failed."""
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(UTC).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
def validate_target(url: str) -> tuple[str, str]:
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise EvidenceFailure("runtime_url_invalid")
|
|
if parsed.username or parsed.password or parsed.fragment:
|
|
raise EvidenceFailure("runtime_url_contains_credentials_or_fragment")
|
|
if parsed.path.rstrip("/") != "/admin/voice-runtime" or parsed.query:
|
|
raise EvidenceFailure("runtime_url_path_invalid")
|
|
return parsed.netloc, parsed.path
|
|
|
|
|
|
def validate_capture_bounds(samples: int, interval_seconds: float) -> None:
|
|
if not 1 <= samples <= MAX_SAMPLES:
|
|
raise EvidenceFailure("sample_count_out_of_bounds")
|
|
if not math.isfinite(interval_seconds) or not 0.05 <= interval_seconds <= 60:
|
|
raise EvidenceFailure("sample_interval_out_of_bounds")
|
|
if (samples - 1) * interval_seconds > MAX_CAPTURE_WINDOW_SECONDS:
|
|
raise EvidenceFailure("capture_window_out_of_bounds")
|
|
|
|
|
|
def _nonnegative_number(value: object, *, code: str) -> int | float:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise EvidenceFailure(code)
|
|
if not math.isfinite(float(value)) or value < 0:
|
|
raise EvidenceFailure(code)
|
|
return value
|
|
|
|
|
|
def validate_snapshot(payload: object) -> dict[str, Any]:
|
|
if not isinstance(payload, dict):
|
|
raise EvidenceFailure("runtime_snapshot_invalid_shape")
|
|
if payload.get("schema_version") != SNAPSHOT_SCHEMA_VERSION:
|
|
raise EvidenceFailure("runtime_snapshot_schema_drift")
|
|
if payload.get("scope") != "single_api_worker":
|
|
raise EvidenceFailure("runtime_snapshot_scope_drift")
|
|
if payload.get("privacy_boundary") != (
|
|
"metadata_only_no_audio_transcript_or_session_ids"
|
|
):
|
|
raise EvidenceFailure("runtime_snapshot_privacy_boundary_drift")
|
|
if payload.get("reset_supported") is not False:
|
|
raise EvidenceFailure("runtime_snapshot_reset_contract_drift")
|
|
|
|
limits = payload.get("limits")
|
|
process = payload.get("process")
|
|
counters = payload.get("counters")
|
|
if not all(isinstance(item, dict) for item in (limits, process, counters)):
|
|
raise EvidenceFailure("runtime_snapshot_sections_missing")
|
|
assert isinstance(limits, dict) and isinstance(process, dict)
|
|
assert isinstance(counters, dict)
|
|
for field in (
|
|
"max_utterance_audio_bytes",
|
|
"streaming_event_queue_max_items",
|
|
"uvicorn_ws_max_queue",
|
|
):
|
|
if (
|
|
_nonnegative_number(
|
|
limits.get(field), code=f"runtime_limit_invalid:{field}"
|
|
)
|
|
<= 0
|
|
):
|
|
raise EvidenceFailure(f"runtime_limit_invalid:{field}")
|
|
worker_id = process.get("worker_instance_id")
|
|
started_at = process.get("started_at_utc")
|
|
platform = process.get("platform")
|
|
if not isinstance(worker_id, str) or len(worker_id) < 16:
|
|
raise EvidenceFailure("runtime_worker_identity_invalid")
|
|
if not isinstance(started_at, str) or not started_at:
|
|
raise EvidenceFailure("runtime_worker_start_invalid")
|
|
if not isinstance(platform, str) or not platform:
|
|
raise EvidenceFailure("runtime_platform_invalid")
|
|
for field in (
|
|
"pid",
|
|
"uptime_seconds",
|
|
"rss_bytes",
|
|
"peak_rss_bytes",
|
|
"cpu_user_seconds",
|
|
"cpu_system_seconds",
|
|
"threads",
|
|
):
|
|
_nonnegative_number(process.get(field), code=f"runtime_process_invalid:{field}")
|
|
for field, value in counters.items():
|
|
_nonnegative_number(value, code=f"runtime_counter_invalid:{field}")
|
|
|
|
forbidden = {
|
|
"session_id",
|
|
"transcript",
|
|
"transcript_text",
|
|
"raw_audio",
|
|
"provider_payload",
|
|
}
|
|
|
|
def walk(value: object) -> None:
|
|
if isinstance(value, dict):
|
|
if forbidden.intersection(str(key) for key in value):
|
|
raise EvidenceFailure("runtime_snapshot_forbidden_field")
|
|
for child in value.values():
|
|
walk(child)
|
|
elif isinstance(value, list):
|
|
for child in value:
|
|
walk(child)
|
|
|
|
walk(payload)
|
|
return payload
|
|
|
|
|
|
def fetch_snapshot(
|
|
*,
|
|
url: str,
|
|
cookie_name: str,
|
|
cookie_value: str,
|
|
timeout_seconds: float,
|
|
) -> dict[str, Any]:
|
|
if not cookie_name or any(char in cookie_name for char in "\r\n;="):
|
|
raise EvidenceFailure("cookie_name_invalid")
|
|
if not cookie_value or any(char in cookie_value for char in "\r\n;"):
|
|
raise EvidenceFailure("admin_session_cookie_invalid")
|
|
request = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Cookie": f"{cookie_name}={cookie_value}",
|
|
# Cloudflare는 Python urllib 기본 User-Agent를 공개 API 도달 전에
|
|
# 거부한다. 공개 배포 probe와 같은 브라우저 호환 전송 경계를 쓴다.
|
|
"User-Agent": _BROWSER_UA,
|
|
},
|
|
method="GET",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(
|
|
request,
|
|
timeout=timeout_seconds,
|
|
context=ssl.create_default_context(),
|
|
) as response:
|
|
if response.status != 200:
|
|
raise EvidenceFailure(f"runtime_http_status:{response.status}")
|
|
raw = response.read(1_048_577)
|
|
except (OSError, urllib.error.URLError) as exc:
|
|
raise EvidenceFailure("runtime_request_failed") from exc
|
|
if len(raw) > 1_048_576:
|
|
raise EvidenceFailure("runtime_response_too_large")
|
|
try:
|
|
return validate_snapshot(json.loads(raw.decode("utf-8")))
|
|
except (UnicodeError, json.JSONDecodeError) as exc:
|
|
raise EvidenceFailure("runtime_response_invalid_json") from exc
|
|
|
|
|
|
def base_evidence(
|
|
*, url: str, samples: int, interval_seconds: float, cookie_env: str
|
|
) -> dict[str, object]:
|
|
host, path = validate_target(url)
|
|
return {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"status": "running",
|
|
"started_at_utc": utc_now(),
|
|
"ended_at_utc": None,
|
|
"target_host": host,
|
|
"target_path": path,
|
|
"cookie_env": cookie_env,
|
|
"cookie_present": False,
|
|
"cookie_value_logged": False,
|
|
"privacy_boundary": "metadata_only_no_audio_transcript_session_or_cookie_values",
|
|
"requested_samples": samples,
|
|
"interval_seconds": interval_seconds,
|
|
"samples_completed": 0,
|
|
"worker_instance_id": None,
|
|
"worker_pid": None,
|
|
"worker_started_at_utc": None,
|
|
"samples": [],
|
|
"high_water": {},
|
|
"failure_type": None,
|
|
}
|
|
|
|
|
|
def capture_evidence(
|
|
*,
|
|
url: str,
|
|
cookie_name: str,
|
|
cookie_value: str,
|
|
cookie_env: str,
|
|
samples: int,
|
|
interval_seconds: float,
|
|
timeout_seconds: float,
|
|
fetch: Callable[..., dict[str, Any]] = fetch_snapshot,
|
|
sleep: Callable[[float], None] = time.sleep,
|
|
) -> dict[str, object]:
|
|
validate_capture_bounds(samples, interval_seconds)
|
|
evidence = base_evidence(
|
|
url=url,
|
|
samples=samples,
|
|
interval_seconds=interval_seconds,
|
|
cookie_env=cookie_env,
|
|
)
|
|
evidence["cookie_present"] = bool(cookie_value)
|
|
snapshots: list[dict[str, Any]] = []
|
|
identity: tuple[str, int, str] | None = None
|
|
for index in range(samples):
|
|
if index:
|
|
sleep(interval_seconds)
|
|
snapshot = fetch(
|
|
url=url,
|
|
cookie_name=cookie_name,
|
|
cookie_value=cookie_value,
|
|
timeout_seconds=timeout_seconds,
|
|
)
|
|
process = snapshot["process"]
|
|
current_identity = (
|
|
str(process["worker_instance_id"]),
|
|
int(process["pid"]),
|
|
str(process["started_at_utc"]),
|
|
)
|
|
if identity is None:
|
|
identity = current_identity
|
|
elif current_identity != identity:
|
|
raise EvidenceFailure("runtime_worker_drift")
|
|
snapshots.append(
|
|
{"sequence": index + 1, "observed_at_utc": utc_now(), "snapshot": snapshot}
|
|
)
|
|
|
|
assert identity is not None
|
|
(
|
|
evidence["worker_instance_id"],
|
|
evidence["worker_pid"],
|
|
evidence["worker_started_at_utc"],
|
|
) = identity
|
|
evidence["samples"] = snapshots
|
|
evidence["samples_completed"] = len(snapshots)
|
|
final_snapshot = snapshots[-1]["snapshot"]
|
|
evidence["high_water"] = {
|
|
"process_peak_rss_bytes": max(
|
|
item["snapshot"]["process"]["peak_rss_bytes"] for item in snapshots
|
|
),
|
|
"process_threads_max": max(
|
|
item["snapshot"]["process"]["threads"] for item in snapshots
|
|
),
|
|
"websocket_high_water": final_snapshot["counters"]["websocket_high_water"],
|
|
"streaming_provider_session_high_water": final_snapshot["counters"][
|
|
"streaming_provider_session_high_water"
|
|
],
|
|
"route_audio_buffer_high_water_bytes": final_snapshot["counters"][
|
|
"route_audio_buffer_high_water_bytes"
|
|
],
|
|
"streaming_event_queue_high_water_items": final_snapshot["counters"][
|
|
"streaming_event_queue_high_water_items"
|
|
],
|
|
"streaming_event_queue_saturation_total": final_snapshot["counters"][
|
|
"streaming_event_queue_saturation_total"
|
|
],
|
|
"streaming_event_queue_wait_seconds_total": final_snapshot["counters"][
|
|
"streaming_event_queue_wait_seconds_total"
|
|
],
|
|
}
|
|
evidence["status"] = "passed"
|
|
evidence["ended_at_utc"] = utc_now()
|
|
return evidence
|
|
|
|
|
|
def emit(evidence: dict[str, object], output: Path | None) -> None:
|
|
payload = json.dumps(evidence, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
|
if output is not None:
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(payload, encoding="utf-8")
|
|
print(payload, end="")
|
|
|
|
|
|
def parser() -> argparse.ArgumentParser:
|
|
result = argparse.ArgumentParser(description=__doc__)
|
|
result.add_argument("--url", required=True)
|
|
result.add_argument("--cookie-env", default="VIGNETTE_ADMIN_SESSION_COOKIE")
|
|
result.add_argument("--cookie-name", default="__Host-vignette_sid")
|
|
result.add_argument("--samples", type=int, default=10)
|
|
result.add_argument("--interval-seconds", type=float, default=1.0)
|
|
result.add_argument("--timeout-seconds", type=float, default=15.0)
|
|
result.add_argument("--evidence-output", type=Path)
|
|
return result
|
|
|
|
|
|
def main() -> int:
|
|
args = parser().parse_args()
|
|
evidence = base_evidence(
|
|
url=args.url,
|
|
samples=args.samples,
|
|
interval_seconds=args.interval_seconds,
|
|
cookie_env=args.cookie_env,
|
|
)
|
|
try:
|
|
validate_capture_bounds(args.samples, args.interval_seconds)
|
|
if not math.isfinite(args.timeout_seconds) or args.timeout_seconds <= 0:
|
|
raise EvidenceFailure("timeout_invalid")
|
|
cookie_value = os.getenv(args.cookie_env, "").strip()
|
|
evidence = capture_evidence(
|
|
url=args.url,
|
|
cookie_name=args.cookie_name,
|
|
cookie_value=cookie_value,
|
|
cookie_env=args.cookie_env,
|
|
samples=args.samples,
|
|
interval_seconds=args.interval_seconds,
|
|
timeout_seconds=args.timeout_seconds,
|
|
)
|
|
emit(evidence, args.evidence_output)
|
|
return 0
|
|
except EvidenceFailure as exc:
|
|
evidence["status"] = "failed"
|
|
evidence["failure_type"] = str(exc)
|
|
evidence["ended_at_utc"] = utc_now()
|
|
emit(evidence, args.evidence_output)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|