122 lines
4.4 KiB
Python
Executable File
122 lines
4.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
|
|
def fail(msg: str) -> int:
|
|
print(f"[transport_platform_compat] ERROR: {msg}")
|
|
return 1
|
|
|
|
|
|
def request_json(api_url: str, method: str, path: str, payload: dict | None = None) -> tuple[int, dict]:
|
|
data = None
|
|
headers = {"Accept": "application/json"}
|
|
if payload is not None:
|
|
data = json.dumps(payload).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
|
|
req = urllib.request.Request(
|
|
f"{api_url.rstrip('/')}{path}",
|
|
data=data,
|
|
method=method.upper(),
|
|
headers=headers,
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=20.0) as resp:
|
|
raw = resp.read().decode("utf-8", errors="replace")
|
|
status = int(resp.getcode() or 200)
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read().decode("utf-8", errors="replace")
|
|
status = int(e.code or 500)
|
|
except Exception:
|
|
return 0, {}
|
|
|
|
try:
|
|
body = json.loads(raw) if raw else {}
|
|
except Exception:
|
|
body = {}
|
|
if not isinstance(body, dict):
|
|
body = {}
|
|
return status, body
|
|
|
|
|
|
def assert_capability_true(caps: dict, section: str, key: str) -> tuple[bool, str]:
|
|
obj = caps.get(section) or {}
|
|
if not isinstance(obj, dict):
|
|
return False, f"section `{section}` is missing in capabilities payload"
|
|
if key not in obj:
|
|
return False, f"`{section}.{key}` is missing in capabilities payload"
|
|
if not bool(obj.get(key)):
|
|
return False, f"`{section}.{key}` must be true for cross-platform contract"
|
|
return True, ""
|
|
|
|
|
|
def main() -> int:
|
|
api_url = os.environ.get("API_URL", "http://127.0.0.1:8080").strip()
|
|
if not api_url:
|
|
return fail("empty API_URL")
|
|
|
|
print(f"[transport_platform_compat] API_URL={api_url}")
|
|
|
|
status, caps = request_json(api_url, "GET", "/api/v1/transport/capabilities")
|
|
if status == 404:
|
|
print("[transport_platform_compat] SKIP: /api/v1/transport/* is unavailable on current backend build")
|
|
return 0
|
|
if status != 200 or not bool(caps.get("ok", False)):
|
|
return fail(f"capabilities failed status={status} payload={caps}")
|
|
|
|
clients = caps.get("clients") or {}
|
|
if not isinstance(clients, dict):
|
|
return fail(f"clients map is invalid: {caps}")
|
|
required_clients = ("singbox", "dnstt", "phoenix")
|
|
for kind in required_clients:
|
|
if kind not in clients:
|
|
return fail(f"missing transport client `{kind}` in capabilities")
|
|
if not isinstance(clients.get(kind), dict):
|
|
return fail(f"client capability `{kind}` must be an object")
|
|
|
|
ok, msg = assert_capability_true(caps, "runtime_modes", "exec")
|
|
if not ok:
|
|
return fail(msg)
|
|
ok, msg = assert_capability_true(caps, "packaging_profiles", "system")
|
|
if not ok:
|
|
return fail(msg)
|
|
ok, msg = assert_capability_true(caps, "packaging_profiles", "bundled")
|
|
if not ok:
|
|
return fail(msg)
|
|
|
|
# Базовый policy-контракт должен быть одинаково доступен для web/iOS/Android клиентов.
|
|
status, policy = request_json(api_url, "GET", "/api/v1/transport/policies")
|
|
if status != 200 or not bool(policy.get("ok", False)):
|
|
return fail(f"transport/policies failed status={status} payload={policy}")
|
|
revision = int(policy.get("policy_revision") or 0)
|
|
intents = policy.get("intents") or []
|
|
if not isinstance(intents, list):
|
|
return fail(f"policy intents must be array: {policy}")
|
|
|
|
status, validated = request_json(
|
|
api_url,
|
|
"POST",
|
|
"/api/v1/transport/policies/validate",
|
|
{"base_revision": revision, "intents": intents},
|
|
)
|
|
if status != 200 or not bool(validated.get("ok", False)):
|
|
return fail(f"transport/policies/validate failed status={status} payload={validated}")
|
|
if int(validated.get("base_revision") or 0) <= 0:
|
|
return fail(f"validate response has invalid base_revision: {validated}")
|
|
|
|
status, conflicts = request_json(api_url, "GET", "/api/v1/transport/conflicts")
|
|
if status != 200 or not bool(conflicts.get("ok", False)):
|
|
return fail(f"transport/conflicts failed status={status} payload={conflicts}")
|
|
|
|
print("[transport_platform_compat] capabilities + policy contract are compatible with web/iOS/Android clients")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|