195 lines
11 KiB
Python
195 lines
11 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 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)
|
|
|
|
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"
|
|
]
|
|
|
|
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"])
|
|
|
|
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()
|
|
|
|
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
|
|
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: []
|
|
"""
|
|
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)
|