feat(protocol): define native session credentials
Verify Protocol / module (push) Successful in 1m13s
Verify Protocol / verify (push) Successful in 35s

This commit is contained in:
sechmachine
2026-08-11 18:38:19 +07:00
parent 79d9e49497
commit afcd5d99db
32 changed files with 1565 additions and 149 deletions
+63 -2
View File
@@ -24,7 +24,7 @@ def main() -> int:
assert set(definition["required"]).issubset(definition["properties"]), name
compatibility = json.loads((ROOT / "compatibility.json").read_text(encoding="utf-8"))
assert set([compatibility["current"], compatibility["n_minus_1"], compatibility["n_minus_2"]]) == {"1", "0", "-1"}
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"):
@@ -38,7 +38,7 @@ def main() -> int:
feature_registry = json.loads((ROOT / "registries/features.json").read_text(encoding="utf-8"))
registered_features = {entry["id"] for entry in feature_registry["features"]}
assert {"display.request.v1", "input.absolute.v1", "input.scroll.v1"}.issubset(registered_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"]
@@ -54,10 +54,57 @@ def main() -> int:
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()
@@ -87,6 +134,20 @@ def main() -> int:
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
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