b866a48b01
실측으로 찾았다. 공개 런타임은 119 paths 구버전이고 /admin/voice-runtime 이 배포돼 있지 않다. capture-g7-runtime-evidence.py 는 정확히 그 경로만 부르므로 artifact 2 를 만들 수 없다. 즉 마이크와 사람을 다 준비해도 오늘 실행하면 50분을 버리고 실패한다. 현재 소스에는 있다(admin.py:1699, 프리뷰 126 paths). 오케스트레이터가 시작 전에 이걸 검사하고 admin_voice_runtime_not_deployed 로 즉시 멈춘다. 공개 런타임 대상 --rehearse 가 exit 2 로 몇 초 만에 차단됐다. 판정 방식을 한 번 고쳤다. 처음엔 미인증 404/401 상태로 짰는데, Cloudflare 앞단이 자동화 클라이언트에게 존재하지 않는 경로까지 포함해 모든 경로를 403(error code 1010)으로 돌려주는 것을 확인했다. 그 신호로는 "배포 누락"과 "edge 차단"을 구분할 수 없어 작동하는 것처럼 보이지만 무의미한 검사였다. OpenAPI spec 의 paths 만 authoritative 하게 쓰도록 바꾸고 그 이유를 테스트로 남겼다. G7 실행 순서가 이렇게 확정된다. 1. current source 를 공개 런타임에 승격(비-secure origin crypto.randomUUID 수정 포함) 2. --rehearse 로 배관 확인 3. 동의 하 물리 마이크 50분 + human pack 으로 실제 실행 검증: 오케스트레이터 25/25, SSOT FAIL 0, dashboard E2E 10/10, ruff clean.
397 lines
14 KiB
Python
397 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""G7 외부 종료 증거 3종을 하나의 겹치는 시간창에서 동시에 잡는 오케스트레이터.
|
|
|
|
`check-g7-external-proof.py` 는 네 artifact 를 요구하고, 그 중 세 개는 **같은 public host 의
|
|
겹치는 시간창**이어야 한다. 지금까지는 운영자가 세 명령을 따로 띄우고 시계를 손으로 맞춰야 했다.
|
|
50분짜리 실행에서 그건 한 번 어긋나면 처음부터 다시 해야 하는 종류의 취약점이다.
|
|
|
|
이 스크립트가 하는 일:
|
|
|
|
1. soak / runtime / topology 캡처를 **동시에** 시작하고 전부 끝날 때까지 기다린다.
|
|
2. 세 캡처가 같은 public host 를 보는지 시작 전에 확인한다(fail-closed).
|
|
3. 끝나면 human voice-gain pack 을 더해 `check-g7-external-proof.py` 를 그대로 돌린다.
|
|
|
|
이 스크립트는 게이트를 **약화시키지 않는다**. 물리 마이크 캡처는 `--confirm-physical-capture` 와
|
|
`--microphone-device` 가 둘 다 있어야 시작하고, human pack 은 만들어 주지 않는다. 네 번째 artifact 는
|
|
실제 참가자와 독립 평가자가 있어야 한다.
|
|
|
|
`--rehearse` 는 마이크·사람 없이 배관만 확인하는 모드다. 짧은 시간창으로 세 러너를 실제로 띄워
|
|
인자·경로·산출 파일까지 검증하되, soak 은 `--preflight-only` 로 돌려 장치를 열지 않는다. 실제 50분
|
|
실행 전에 이걸로 먼저 실패를 뽑아내라.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Iterable, Sequence
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPTS = REPO_ROOT / "scripts"
|
|
|
|
SOAK_SCRIPT = SCRIPTS / "soak-public-voice-websocket.py"
|
|
RUNTIME_SCRIPT = SCRIPTS / "capture-g7-runtime-evidence.py"
|
|
TOPOLOGY_SCRIPT = SCRIPTS / "capture-g7-topology-evidence.py"
|
|
CHECKER_SCRIPT = SCRIPTS / "check-g7-external-proof.py"
|
|
|
|
MIN_PRODUCTION_SECONDS = 3_000.0
|
|
RUNTIME_SAMPLE_MARGIN = 1
|
|
|
|
|
|
class WindowError(RuntimeError):
|
|
"""Fail-closed 오케스트레이션 오류."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Leg:
|
|
name: str
|
|
argv: list[str]
|
|
output: Path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LegResult:
|
|
name: str
|
|
returncode: int
|
|
output: Path
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return self.returncode == 0
|
|
|
|
|
|
def host_of(url: str) -> str:
|
|
from urllib.parse import urlsplit
|
|
|
|
return (urlsplit(url).hostname or "").lower()
|
|
|
|
|
|
def assert_single_host(*urls: str) -> str:
|
|
"""세 캡처가 같은 public host 를 보는지 확인한다."""
|
|
|
|
hosts = {host_of(url) for url in urls if url}
|
|
hosts.discard("")
|
|
if len(hosts) != 1:
|
|
raise WindowError(f"hosts_must_match:{sorted(hosts)}")
|
|
return hosts.pop()
|
|
|
|
|
|
def sample_plan(duration_seconds: float, interval_seconds: float) -> int:
|
|
"""시간창을 덮기에 충분한 샘플 수. 모자라면 겹침 검증이 깨진다."""
|
|
|
|
if duration_seconds <= 0 or interval_seconds <= 0:
|
|
raise WindowError("invalid_sampling_plan")
|
|
return int(duration_seconds // interval_seconds) + RUNTIME_SAMPLE_MARGIN
|
|
|
|
|
|
def build_soak_leg(args: argparse.Namespace, output: Path) -> Leg:
|
|
argv = [
|
|
sys.executable,
|
|
"-X",
|
|
"utf8",
|
|
str(SOAK_SCRIPT),
|
|
"--wss-url",
|
|
args.wss_url,
|
|
"--origin",
|
|
args.origin,
|
|
"--expected-stt-provider",
|
|
args.expected_stt_provider,
|
|
"--expected-stt-model",
|
|
args.expected_stt_model,
|
|
"--expected-tts-provider",
|
|
args.expected_tts_provider,
|
|
"--expected-tts-model",
|
|
args.expected_tts_model,
|
|
"--evidence-output",
|
|
str(output),
|
|
]
|
|
if args.rehearse:
|
|
# 배관 확인 전용. 장치를 열지 않는다.
|
|
argv.append("--preflight-only")
|
|
return Leg("voice_soak_preflight", argv, output)
|
|
if not args.microphone_device or not args.confirm_physical_capture:
|
|
raise WindowError("physical_microphone_consent_required")
|
|
argv += [
|
|
"--microphone-device",
|
|
args.microphone_device,
|
|
"--confirm-physical-capture",
|
|
"--duration-seconds",
|
|
str(args.duration_seconds),
|
|
]
|
|
return Leg("voice_soak", argv, output)
|
|
|
|
|
|
def build_runtime_leg(args: argparse.Namespace, output: Path) -> Leg:
|
|
samples = sample_plan(args.duration_seconds, args.runtime_interval_seconds)
|
|
argv = [
|
|
sys.executable,
|
|
"-X",
|
|
"utf8",
|
|
str(RUNTIME_SCRIPT),
|
|
"--url",
|
|
args.admin_runtime_url,
|
|
"--samples",
|
|
str(samples),
|
|
"--interval-seconds",
|
|
str(args.runtime_interval_seconds),
|
|
"--evidence-output",
|
|
str(output),
|
|
]
|
|
return Leg("runtime", argv, output)
|
|
|
|
|
|
def build_topology_leg(args: argparse.Namespace, output: Path) -> Leg:
|
|
samples = sample_plan(args.duration_seconds, args.topology_interval_seconds)
|
|
argv = [
|
|
sys.executable,
|
|
"-X",
|
|
"utf8",
|
|
str(TOPOLOGY_SCRIPT),
|
|
"--compose-project",
|
|
args.compose_project,
|
|
"--public-host",
|
|
host_of(args.wss_url),
|
|
"--api-container",
|
|
args.api_container,
|
|
"--api-image-digest",
|
|
args.api_image_digest,
|
|
"--caddy-container",
|
|
args.caddy_container,
|
|
"--caddy-image-digest",
|
|
args.caddy_image_digest,
|
|
"--samples",
|
|
str(samples),
|
|
"--interval-seconds",
|
|
str(args.topology_interval_seconds),
|
|
"--evidence-output",
|
|
str(output),
|
|
]
|
|
return Leg("topology", argv, output)
|
|
|
|
|
|
ADMIN_RUNTIME_PATH = "/admin/voice-runtime"
|
|
# Cloudflare 앞단은 자동화 클라이언트에 error code 1010 으로 **모든 경로**에 403 을 준다.
|
|
# 그래서 미인증 상태 코드로는 "endpoint 없음"과 "edge 차단"을 구분할 수 없다.
|
|
# 배포 여부는 OpenAPI spec 의 paths 로만 판정한다.
|
|
_BROWSER_UA = (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/140.0 Safari/537.36"
|
|
)
|
|
|
|
|
|
def assert_admin_endpoint_deployed(
|
|
url: str, *, paths_probe: Callable[[str], Sequence[str]] | None = None
|
|
) -> str:
|
|
"""관리자 runtime endpoint 가 그 배포에 실제로 존재하는지 먼저 본다.
|
|
|
|
2026-08-08 확인: 공개 런타임은 119 paths 구버전이라 `/admin/voice-runtime` 이
|
|
아예 없었다. 이 상태로 50분 창을 돌리면 마이크와 사람을 다 준비하고도 artifact 2
|
|
에서 실패한다. 그래서 시작 전에 막는다.
|
|
|
|
미인증 HTTP 상태로 판정하지 않는다. Cloudflare 가 자동화 클라이언트에 모든 경로를
|
|
403(error code 1010)으로 돌려주므로 그 신호로는 배포 누락을 알 수 없다. OpenAPI
|
|
spec 의 `paths` 만 authoritative 하게 쓴다.
|
|
"""
|
|
|
|
from urllib.parse import urlsplit
|
|
|
|
parsed = urlsplit(url)
|
|
if parsed.path.rstrip("/") != ADMIN_RUNTIME_PATH:
|
|
raise WindowError("admin_runtime_url_unexpected_path")
|
|
origin = f"{parsed.scheme}://{parsed.netloc}"
|
|
paths = (paths_probe or _openapi_paths)(origin)
|
|
if ADMIN_RUNTIME_PATH not in set(paths):
|
|
raise WindowError("admin_voice_runtime_not_deployed")
|
|
return origin
|
|
|
|
|
|
def _openapi_paths(origin: str) -> Sequence[str]: # pragma: no cover - 네트워크 I/O
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
request = urllib.request.Request(
|
|
f"{origin}/openapi.json",
|
|
headers={"User-Agent": _BROWSER_UA, "Accept": "application/json"},
|
|
method="GET",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
document = json.loads(response.read(8 * 1024 * 1024))
|
|
except (urllib.error.URLError, OSError, ValueError) as exc:
|
|
raise WindowError("openapi_unreachable") from exc
|
|
paths = document.get("paths") if isinstance(document, dict) else None
|
|
if not isinstance(paths, dict):
|
|
raise WindowError("openapi_shape_invalid")
|
|
return list(paths)
|
|
|
|
|
|
def plan_legs(
|
|
args: argparse.Namespace,
|
|
out_dir: Path,
|
|
*,
|
|
admin_probe: Callable[[str], Sequence[str]] | None = None,
|
|
) -> list[Leg]:
|
|
assert_single_host(args.wss_url, args.origin, args.admin_runtime_url)
|
|
assert_admin_endpoint_deployed(args.admin_runtime_url, paths_probe=admin_probe)
|
|
return [
|
|
build_soak_leg(args, out_dir / "voice-soak.json"),
|
|
build_runtime_leg(args, out_dir / "runtime.json"),
|
|
build_topology_leg(args, out_dir / "topology.json"),
|
|
]
|
|
|
|
|
|
def _run(leg: Leg, *, timeout: float, runner: Callable[..., Any]) -> LegResult:
|
|
completed = runner(
|
|
leg.argv,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=timeout,
|
|
shell=False,
|
|
)
|
|
return LegResult(leg.name, int(completed.returncode), leg.output)
|
|
|
|
|
|
def run_window(
|
|
legs: Sequence[Leg],
|
|
*,
|
|
timeout: float,
|
|
runner: Callable[..., Any] | None = None,
|
|
) -> list[LegResult]:
|
|
"""세 캡처를 동시에 시작한다. 하나가 죽어도 나머지를 끝까지 돌린다."""
|
|
|
|
call = runner or subprocess.run
|
|
with ThreadPoolExecutor(max_workers=len(legs)) as pool:
|
|
futures = [
|
|
pool.submit(_run, leg, timeout=timeout, runner=call) for leg in legs
|
|
]
|
|
return [future.result() for future in futures]
|
|
|
|
|
|
def build_checker_argv(
|
|
results: Sequence[LegResult], human_pack: Path
|
|
) -> list[str]:
|
|
by_name = {result.name: result.output for result in results}
|
|
soak = by_name.get("voice_soak") or by_name.get("voice_soak_preflight")
|
|
if soak is None:
|
|
raise WindowError("soak_evidence_missing")
|
|
return [
|
|
sys.executable,
|
|
"-X",
|
|
"utf8",
|
|
str(CHECKER_SCRIPT),
|
|
"--voice-soak",
|
|
str(soak),
|
|
"--runtime",
|
|
str(by_name["runtime"]),
|
|
"--topology",
|
|
str(by_name["topology"]),
|
|
"--human-voice-gain",
|
|
str(human_pack),
|
|
"--json",
|
|
]
|
|
|
|
|
|
def summarize(
|
|
results: Sequence[LegResult],
|
|
*,
|
|
rehearse: bool,
|
|
checker_returncode: int | None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"schema_version": "vignette.g7-external-proof-window.v1",
|
|
"mode": "rehearse" if rehearse else "production",
|
|
"legs": [
|
|
{"name": r.name, "returncode": r.returncode, "evidence": str(r.output)}
|
|
for r in results
|
|
],
|
|
"all_legs_passed": all(r.ok for r in results),
|
|
"checker_returncode": checker_returncode,
|
|
"gate_closed": checker_returncode == 0 and not rehearse,
|
|
"note": (
|
|
"rehearse 는 배관만 확인한다. 게이트는 실제 물리 마이크 50분 실행과 "
|
|
"독립 human voice-gain pack 이 있어야 닫힌다."
|
|
if rehearse
|
|
else "네 artifact 가 모두 통과해야 게이트가 닫힌다."
|
|
),
|
|
}
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--wss-url", required=True)
|
|
parser.add_argument("--origin", required=True)
|
|
parser.add_argument("--admin-runtime-url", required=True)
|
|
parser.add_argument("--compose-project", required=True)
|
|
parser.add_argument("--api-container", required=True)
|
|
parser.add_argument("--api-image-digest", required=True)
|
|
parser.add_argument("--caddy-container", required=True)
|
|
parser.add_argument("--caddy-image-digest", required=True)
|
|
parser.add_argument("--expected-stt-provider", default="local_whisper")
|
|
parser.add_argument("--expected-stt-model", default="large-v3")
|
|
parser.add_argument("--expected-tts-provider", default="melotts")
|
|
parser.add_argument("--expected-tts-model", default="melotts-korean")
|
|
parser.add_argument("--microphone-device", default="")
|
|
parser.add_argument("--confirm-physical-capture", action="store_true")
|
|
parser.add_argument("--duration-seconds", type=float, default=MIN_PRODUCTION_SECONDS)
|
|
parser.add_argument("--runtime-interval-seconds", type=float, default=100.0)
|
|
parser.add_argument("--topology-interval-seconds", type=float, default=100.0)
|
|
parser.add_argument("--human-voice-gain", type=Path)
|
|
parser.add_argument("--out-dir", type=Path, required=True)
|
|
parser.add_argument(
|
|
"--rehearse",
|
|
action="store_true",
|
|
help="마이크 없이 배관만 확인한다. 게이트를 닫지 않는다.",
|
|
)
|
|
return parser
|
|
|
|
|
|
def validate(args: argparse.Namespace) -> None:
|
|
if not args.rehearse and args.duration_seconds < MIN_PRODUCTION_SECONDS:
|
|
raise WindowError("production_window_too_short")
|
|
if not args.rehearse and args.human_voice_gain is None:
|
|
raise WindowError("human_voice_gain_pack_required")
|
|
|
|
|
|
def main(argv: Iterable[str] | None = None) -> int:
|
|
args = build_parser().parse_args(list(argv) if argv is not None else None)
|
|
try:
|
|
validate(args)
|
|
args.out_dir.mkdir(parents=True, exist_ok=True)
|
|
legs = plan_legs(args, args.out_dir)
|
|
except WindowError as exc:
|
|
print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False), file=sys.stderr)
|
|
return 2
|
|
|
|
results = run_window(legs, timeout=args.duration_seconds + 900)
|
|
|
|
checker_returncode: int | None = None
|
|
if all(result.ok for result in results) and args.human_voice_gain is not None:
|
|
checker = subprocess.run(
|
|
build_checker_argv(results, args.human_voice_gain),
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
shell=False,
|
|
)
|
|
checker_returncode = checker.returncode
|
|
|
|
report = summarize(
|
|
results, rehearse=args.rehearse, checker_returncode=checker_returncode
|
|
)
|
|
print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
|
|
return 0 if report["all_legs_passed"] else 1
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover - CLI
|
|
raise SystemExit(main())
|