Files
VerseVDI-Protocol/tools/validate.py
T
sechmachine c3ea2f35c6
Verify Protocol / module (push) Successful in 1m21s
Verify Protocol / verify (push) Successful in 1m2s
fix(protocol): align quality ack schema
2026-08-13 07:53:35 +07:00

341 lines
22 KiB
Python

#!/usr/bin/env python3
"""Dependency-free structural and scope validation for the Protocol sources."""
from __future__ import annotations
import json
import pathlib
import hashlib
import re
import sys
ROOT = pathlib.Path(__file__).resolve().parents[1]
def matches_outcome_branch(instance: dict[str, object], branch: dict[str, object]) -> bool:
required = branch.get("required", [])
if not all(field in instance for field in required):
return False
outcome = branch.get("properties", {}).get("outcome", {}).get("const")
if instance.get("outcome") != outcome:
return False
forbidden = branch.get("not", {}).get("required", [])
return not all(field in instance for field in forbidden) if forbidden else True
def main() -> int:
schema_path = ROOT / "schemas/control-v1.schema.json"
schema = json.loads(schema_path.read_text(encoding="utf-8"))
assert schema["$schema"].endswith("2020-12/schema")
defs = schema["$defs"]
assert len(defs) >= 16
for name, definition in defs.items():
assert definition["type"] == "object", name
assert definition["additionalProperties"] is False, name
assert set(definition["required"]).issubset(definition["properties"]), name
compatibility = json.loads((ROOT / "compatibility.json").read_text(encoding="utf-8"))
assert [compatibility["current"], compatibility["n_minus_1"], compatibility["n_minus_2"]] == ["2", "1", "0"]
assert len(set(compatibility["unsupported"])) == len(compatibility["unsupported"])
for registry in ("registries/features.json", "registries/datagrams.json"):
value = json.loads((ROOT / registry).read_text(encoding="utf-8"))
entries = value.get("features", value.get("datagrams"))
assert entries and len({entry["id"] for entry in entries}) == len(entries)
maximum_bound = 1_048_576 if registry.endswith("datagrams.json") else 65_536
for entry in entries:
maximum = entry.get("max_frame_bytes", entry.get("max_payload_bytes"))
assert isinstance(maximum, int) and 1 <= maximum <= maximum_bound
feature_registry = json.loads((ROOT / "registries/features.json").read_text(encoding="utf-8"))
registered_features = {entry["id"] for entry in feature_registry["features"]}
assert {"control.v1", "control.v2", "display.request.v1", "input.absolute.v1", "input.scroll.v1"}.issubset(registered_features)
assert {"video.profile.v1", "session.quality.v1", "session.stop.v1", "controller.arrival.v1"}.issubset(registered_features)
assert defs["VideoProfile"]["required"] == ["codec", "bit_depth", "chroma_subsampling", "color_space", "transfer_function"]
assert defs["AudioProfile"]["required"] == ["codec", "sample_rate_hz", "channels", "channel_layout", "packet_duration_ms"]
assert defs["CapabilityProfile"]["required"] == ["transport", "framing", "media", "source_rate_control", "video_profiles", "audio_profiles"]
assert defs["CapabilityProfile"]["properties"]["video_profiles"] == {"type": "array", "minItems": 1, "maxItems": 12, "uniqueItems": True, "items": {"$ref": "#/$defs/VideoProfile"}}
assert defs["CapabilityProfile"]["properties"]["audio_profiles"] == {"type": "array", "minItems": 1, "maxItems": 1, "uniqueItems": True, "items": {"$ref": "#/$defs/AudioProfile"}}
assert defs["SessionRequest"]["required"][-2:] == ["video_profiles", "bitrate_preference"]
assert defs["ReconnectRequest"]["required"][-1] == "display_relaunch_confirmed"
assert defs["AssignedDesktop"]["required"][-1] == "quality_limits"
assert defs["EntitledPool"]["required"][-1] == "quality_limits"
for owner in ("SessionAuthority", "ClientSessionAuthority"):
assert defs[owner]["required"][-1] == "selected_descriptor"
assert defs["ConnectionManifest"]["required"][-1] == "selected_descriptor"
quality_request = defs["GatewayQualityWorkRequest"]
assert quality_request["required"] == ["version", "session_id", "gateway_id", "reconnect_sequence", "acquisition"]
assert quality_request["properties"]["acquisition"]["enum"] == ["poll", "prompt", "observation"]
assert quality_request["properties"]["lease_generation"]["minimum"] == 1
assert defs["GatewayStopWorkRequest"]["required"] == ["version", "session_id", "gateway_id", "reconnect_sequence", "acquisition"]
assert defs["GatewayQualityWork"]["required"][-3:] == ["lease_generation", "lease_expires_at", "selected_descriptor"]
quality_ack = defs["GatewayQualityAck"]
assert quality_ack["required"][-3:] == ["revision", "lease_generation", "outcome"]
assert quality_ack["properties"]["outcome"]["enum"] == ["applied", "proven_prior", "unknown"]
assert quality_ack["description"] == (
"Outcome invariants: applied requires current_applied_revision equal to revision; "
"proven_prior requires current_applied_revision strictly less than revision; "
"unknown forbids current_applied_revision and makes no applied-revision assertion."
)
assert quality_ack["oneOf"] == [
{"properties": {"outcome": {"const": "applied"}}, "required": ["current_applied_revision"]},
{"properties": {"outcome": {"const": "proven_prior"}}, "required": ["current_applied_revision"]},
{"properties": {"outcome": {"const": "unknown"}}, "not": {"required": ["current_applied_revision"]}},
]
ack_branches = quality_ack["oneOf"]
for valid_ack in (
{"outcome": "applied", "current_applied_revision": 7},
{"outcome": "proven_prior", "current_applied_revision": 6},
{"outcome": "unknown"},
):
assert sum(matches_outcome_branch(valid_ack, branch) for branch in ack_branches) == 1, valid_ack
for invalid_ack in (
{"outcome": "applied"},
{"outcome": "proven_prior"},
{"outcome": "unknown", "current_applied_revision": 6},
):
assert not any(matches_outcome_branch(invalid_ack, branch) for branch in ack_branches), invalid_ack
operation_id_pattern = r"^(?!00000000-0000-0000-0000-000000000000$)[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
canonical_time_pattern = r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]{0,8}[1-9])?Z$"
for definition in ("QualityChangeOperation", "StopOperation", "GatewayQualityWorkRequest", "GatewayQualityWork", "GatewayQualityAck", "GatewayStopWorkRequest", "GatewayStopWork", "GatewayStopAck"):
assert defs[definition]["properties"]["operation_id"]["pattern"] == operation_id_pattern, definition
assert defs[definition]["x-max-bytes"] == 16384, definition
for definition, fields in {
"QualityChangeOperation": ("created_at", "deadline_at", "updated_at"),
"StopOperation": ("created_at", "deadline_at", "updated_at"),
"GatewayQualityWork": ("lease_expires_at",),
}.items():
for field in fields:
assert defs[definition]["properties"][field]["pattern"] == canonical_time_pattern, (definition, field)
display_mode = defs["DisplayMode"]
assert display_mode["required"] == ["resolution_width", "resolution_height", "fps"]
assert display_mode["properties"]["resolution_width"] == {"type": "integer", "minimum": 320, "maximum": 16384}
assert display_mode["properties"]["resolution_height"] == {"type": "integer", "minimum": 200, "maximum": 8640}
assert display_mode["properties"]["fps"] == {"type": "integer", "minimum": 1, "maximum": 240}
for owner, field in (
("SessionRequest", "requested_display_mode"),
("BrokerSession", "requested_display_mode"),
("BrokerSession", "effective_display_mode"),
("ManifestProfile", "display_mode"),
):
assert field not in defs[owner]["required"]
assert defs[owner]["properties"][field] == {"$ref": "#/$defs/DisplayMode"}
session_request = defs["SessionRequest"]
assert "policy_snapshot" not in session_request["required"]
assert "policy_snapshot" not in session_request["properties"]
assert "policy_snapshot" in defs["BrokerSession"]["required"]
assert defs["BrokerSession"]["properties"]["policy_snapshot"] == {"$ref": "#/$defs/AllocationPolicy"}
native_identity = defs["NativeSessionIdentity"]
assert native_identity["required"] == ["client_device_id", "device_key_id"]
browser_session = defs["BrowserAuthenticatedSession"]
assert browser_session["required"] == ["username", "provider", "roles", "role"]
assert "native_identity" not in browser_session["properties"]
native_session = defs["NativeAuthenticatedSession"]
assert native_session["required"] == ["username", "provider", "roles", "role", "native_identity"]
assert native_session["properties"]["native_identity"] == {"$ref": "#/$defs/NativeSessionIdentity"}
for session_definition in (browser_session, native_session):
assert session_definition["properties"]["roles"]["items"] == {
"type": "string", "minLength": 1, "maxLength": 64, "x-max-bytes": 64
}
tunnel_credential = defs["NativeTunnelCredential"]
assert tunnel_credential["required"] == [
"client_device_id", "device_key_id", "certificate_chain_pem", "trust_bundle_pem", "expires_at"
]
client_authority_expiry = defs["ClientSessionAuthority"]["properties"]["expires_at"]
assert client_authority_expiry["format"] == "date-time"
client_authority_expiry_pattern = re.compile(client_authority_expiry.get("pattern", r"(?!)"))
assert client_authority_expiry_pattern.fullmatch("2099-01-01T00:00:00Z"), "client authority expiry must accept canonical UTC"
for noncanonical_expiry in ("2099-01-01T00:00:00+00:00", "2099-01-01T00:00:00.100Z"):
assert not client_authority_expiry_pattern.fullmatch(noncanonical_expiry), (
f"client authority expiry accepted noncanonical UTC {noncanonical_expiry}"
)
manifest = json.loads((ROOT / "fixtures/valid/manifest.json").read_text(encoding="utf-8"))
assert set(manifest).issubset(set(defs["ConnectionManifest"]["properties"]))
public_identity = manifest["gateway"]["public_identity"]
assert public_identity == "gateway.control.test"
assert public_identity not in {
manifest["gateway"]["id"],
*manifest["gateway"]["addresses"],
"sha256:" + "00" * 32,
"apollo-provider-1",
}
forbidden = json.loads((ROOT / "fixtures/invalid/manifest-provider-field.json").read_text(encoding="utf-8"))
assert "provider_url" not in defs["ConnectionManifest"]["properties"] and "provider_url" in forbidden
session_request_fixture = json.loads((ROOT / "fixtures/valid/session-request.json").read_text(encoding="utf-8"))
assert "policy_snapshot" not in session_request_fixture
rejected_policy_fixture = json.loads((ROOT / "fixtures/invalid/session-request-policy-snapshot.json").read_text(encoding="utf-8"))
assert "policy_snapshot" in rejected_policy_fixture
browser_session_fixture = json.loads((ROOT / "fixtures/valid/authenticated-browser-session.json").read_text(encoding="utf-8"))
assert "native_identity" not in browser_session_fixture
native_session_fixture = json.loads((ROOT / "fixtures/valid/authenticated-native-session.json").read_text(encoding="utf-8"))
assert set(native_session_fixture["native_identity"]) == {"client_device_id", "device_key_id"}
partial_identity_fixture = json.loads((ROOT / "fixtures/invalid/authenticated-session-partial-native-identity.json").read_text(encoding="utf-8"))
assert set(partial_identity_fixture["native_identity"]) != {"client_device_id", "device_key_id"}
browser_native_fixture = json.loads((ROOT / "fixtures/invalid/browser-session-native-identity.json").read_text(encoding="utf-8"))
assert "native_identity" in browser_native_fixture
native_missing_fixture = json.loads((ROOT / "fixtures/invalid/native-session-missing-identity.json").read_text(encoding="utf-8"))
assert "native_identity" not in native_missing_fixture
tunnel_credential_fixture = json.loads((ROOT / "fixtures/valid/native-tunnel-credential.json").read_text(encoding="utf-8"))
assert set(tunnel_credential_fixture) == set(tunnel_credential["required"])
valid_fixture_contracts = {
"fixtures/valid/session-request.json": "SessionRequest",
"fixtures/valid/selected-session-descriptor.json": "SelectedSessionDescriptor",
"fixtures/valid/session-quality-limits.json": "SessionQualityLimits",
}
for relative, definition in valid_fixture_contracts.items():
fixture = json.loads((ROOT / relative).read_text(encoding="utf-8"))
assert set(fixture) == set(defs[definition]["required"]), (relative, definition)
assert set(json.loads((ROOT / "fixtures/invalid/capability-rc5-opaque.json").read_text())) & {"audio", "client_decode"} == {"audio", "client_decode"}
assert "video_profiles" not in json.loads((ROOT / "fixtures/invalid/session-request-rc5.json").read_text())
assert set(json.loads((ROOT / "fixtures/invalid/provider-stream-policy-rc5.json").read_text())) == {"resolution_width", "resolution_height", "fps", "codec", "bitrate_kbps", "audio_enabled"}
expected_header = "id\tversion\tkind\tinput\texpected"
ids = set()
for fixture_path in sorted((ROOT / "fixtures/conformance").glob("*.tsv")):
lines = fixture_path.read_text(encoding="utf-8").splitlines()
assert lines and lines[0] == expected_header, fixture_path
for line in lines[1:]:
fields = line.split("\t")
assert len(fields) == 5, line
assert fields[0] not in ids, fields[0]
ids.add(fields[0])
assert fields[4] == "valid" or fields[4].startswith("invalid:") or (
fields[2] == "device_proof_transcript" and re.fullmatch(r"[0-9a-f]{224}", fields[4])
), line
fixture_manifest = json.loads((ROOT / "fixtures/manifest.json").read_text(encoding="utf-8"))
assert fixture_manifest["files"] == sorted(
path.relative_to(ROOT).as_posix() for path in (ROOT / "fixtures/conformance").glob("*.tsv")
)
fixture_hash = hashlib.sha256()
for relative in fixture_manifest["files"]:
fixture_hash.update(relative.encode("utf-8"))
fixture_hash.update(b"\0")
fixture_hash.update((ROOT / relative).read_bytes())
fixture_hash.update(b"\0")
assert fixture_manifest["corpus_sha256"] == fixture_hash.hexdigest()
json_fixture_paths = sorted(
path.relative_to(ROOT).as_posix()
for directory in (ROOT / "fixtures/valid", ROOT / "fixtures/invalid")
for path in directory.glob("*.json")
)
assert fixture_manifest["json_files"] == json_fixture_paths
json_fixture_hash = hashlib.sha256()
for relative in json_fixture_paths:
json_fixture_hash.update(relative.encode("utf-8"))
json_fixture_hash.update(b"\0")
json_fixture_hash.update((ROOT / relative).read_bytes())
json_fixture_hash.update(b"\0")
assert fixture_manifest["json_corpus_sha256"] == json_fixture_hash.hexdigest()
openapi = (ROOT / "openapi/control-v1.yaml").read_text(encoding="utf-8")
assert "openapi: 3.1.0" in openapi
assert "/api/v1/auth/refresh:" in openapi and "/api/v1/resources:" in openapi and "/api/v1/events:" in openapi
for route in (
"/api/v1/session-quality-limits:",
"/api/v1/session-quality-limits/assignments/{assignment_id}:",
"/api/v1/session-quality-limits/pools/{pool_id}:",
"/api/v1/admin/entitlements/{entitlement_id}/display-limit-override:",
"/api/v1/broker/sessions/{session_id}/quality-changes:",
"/api/v1/broker/sessions/{session_id}/quality-changes/{operation_id}:",
"/api/v1/broker/sessions/{session_id}/stop-operations:",
"/api/v1/broker/sessions/{session_id}/stop-operations/{operation_id}:",
"/api/v1/gateway/quality-work:",
"/api/v1/gateway/quality-ack:",
"/api/v1/gateway/stop-work:",
"/api/v1/gateway/stop-ack:",
):
assert route in openapi, route
for operation_id in (
"getSessionQualityLimits", "getAssignmentSessionQualityLimits", "getPoolSessionQualityLimits",
"createSessionQualityChange", "getSessionQualityChange", "createSessionStopOperation", "getSessionStopOperation",
):
operation = openapi.split(f" operationId: {operation_id}\n", 1)[1].split(" responses:\n", 1)[0]
assert "nativeBearer: []" in operation and "browserSession" not in operation, operation_id
for operation_id in (
"acquireGatewayQualityWork", "acknowledgeGatewayQualityWork", "acquireGatewayStopWork", "acknowledgeGatewayStopWork",
):
operation = openapi.split(f" operationId: {operation_id}\n", 1)[1].split(" responses:\n", 1)[0]
assert "gatewayMutualTLS: []" in operation and "nativeBearer" not in operation and "browserSession" not in operation, operation_id
assert "certificate identity MUST match" in operation, operation_id
assert "Maximum JSON body: 16384 bytes." in operation, operation_id
quality_acquisition = openapi.split(" operationId: acquireGatewayQualityWork\n", 1)[1].split(" responses:\n", 1)[0]
assert "`poll` acquisition omits unknown operation coordinates" in quality_acquisition
assert "coordinates MUST match exactly" in quality_acquisition
quality_acknowledgement = openapi.split(" operationId: acknowledgeGatewayQualityWork\n", 1)[1].split(" responses:\n", 1)[0]
assert "Stale lease generations MUST be rejected" in quality_acknowledgement
assert "`applied` requires `current_applied_revision == revision`" in quality_acknowledgement
assert "`proven_prior` requires `current_applied_revision < revision`" in quality_acknowledgement
assert "`unknown` forbids `current_applied_revision`" in quality_acknowledgement
for operation_id in ("createSessionQualityChange", "getSessionQualityChange", "createSessionStopOperation", "getSessionStopOperation"):
operation = openapi.split(f" operationId: {operation_id}\n", 1)[1].split(" responses:\n", 1)[0]
assert "owning principal and active device/key" in operation, operation_id
assert openapi.count("Maximum JSON body: 16384 bytes.") >= 8
assert operation_id_pattern in openapi
assert "provider_url" not in openapi and "vm_address" not in openapi
session_endpoint = openapi.split(" /api/v1/auth/session:", 1)[1].split("\n /api/", 1)[0]
assert "$defs/BrowserAuthenticatedSession" in session_endpoint
assert "$defs/NativeAuthenticatedSession" in session_endpoint
login_endpoint = openapi.split(" /api/v1/auth/login:", 1)[1].split("\n /api/", 1)[0]
assert "$defs/BrowserAuthenticatedSession" in login_endpoint
assert "$defs/NativeAuthenticatedSession" not in login_endpoint
tunnel_endpoint = openapi.split(" /api/v1/auth/tunnel-credentials:", 1)[1].split("\n /api/", 1)[0]
assert "- nativeBearer: []" in tunnel_endpoint
assert "browserSession" not in tunnel_endpoint and "requestBody:" not in tunnel_endpoint
assert "$defs/NativeTunnelCredential" in tunnel_endpoint
assert "Cache-Control:" in tunnel_endpoint and "const: no-store" in tunnel_endpoint
csrf_schemes = """ browserCsrfCookie:
type: apiKey
in: cookie
name: versevdi_csrf
description: Must be identical to X-CSRF-Token and is checked against Server session state.
browserCsrfHeader:
type: apiKey
in: header
name: X-CSRF-Token
description: Must be identical to the versevdi_csrf cookie and is checked against Server session state.
"""
assert csrf_schemes in openapi, "missing exact browser CSRF security schemes"
browser_requirement = """ security:
- browserSession: []
browserCsrfCookie: []
browserCsrfHeader: []
"""
admin_override = openapi.split(" operationId: updateEntitlementDisplayLimitOverride\n", 1)[1].split(" responses:\n", 1)[0]
assert browser_requirement.removeprefix(" ") in admin_override
assert "Maximum JSON body: 16384 bytes." in admin_override
for operation_id in (
"issueReauthenticationGrant", "logoutSession", "registerDevice", "proveDevice", "revokeDevice",
"requestBrokerSession", "allocateBrokerSession", "reconnectBrokerSession", "cancelBrokerSession",
):
operation = openapi.split(f" operationId: {operation_id}\n", 1)[1].split(" responses:\n", 1)[0]
assert browser_requirement.removeprefix(" ") in operation, f"{operation_id}: missing browser CSRF AND requirement"
for operation_id in ("logoutSession", "requestBrokerSession", "allocateBrokerSession", "reconnectBrokerSession", "cancelBrokerSession"):
operation = openapi.split(f" operationId: {operation_id}\n", 1)[1].split(" responses:\n", 1)[0]
assert " browserCsrfHeader: []\n - nativeBearer: []\n" in operation, f"{operation_id}: native bearer must remain a separate OR requirement"
for operation_id in ("loginBrowserSession", "rotateNativeCredential", "issueNativeTunnelCredential"):
operation = openapi.split(f" operationId: {operation_id}\n", 1)[1].split(" responses:\n", 1)[0]
assert "browserCsrf" not in operation, f"{operation_id}: excluded operation gained browser CSRF"
for operation_id in ("getAuthenticatedSession", "listResources", "getBrokerSession", "resumeUserEvents"):
operation = openapi.split(f" operationId: {operation_id}\n", 1)[1].split(" responses:\n", 1)[0]
assert "browserCsrf" not in operation, f"{operation_id}: safe GET gained browser CSRF"
assert defs["ManifestGateway"]["properties"]["public_identity"]["description"] == (
"Exact TLS server name; distinct from dial addresses, gateway UUIDs, certificate fingerprints, and provider identities."
)
print("Protocol source validation passed")
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except (AssertionError, OSError, json.JSONDecodeError) as exc:
print(f"validate: {exc}", file=sys.stderr)
raise SystemExit(1)