sechmachine/VerseVDI-Protocol
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(protocol): fence gateway work recovery

Browse Source
This commit is contained in:
sechmachine
2026-08-13 07:43:42 +07:00
parent 8eacc4fda9
commit 6e18bc9ee6
21 changed files with 572 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/generate.py
+36 -4
View File
@@ -183,6 +183,16 @@ def go_validation(definition: dict[str, Any]) -> list[str]:
lines.append("\tif v.BitrateTargetKbps > v.BitrateMaximumKbps { violations = append(violations, FieldViolation{Field: \"bitrate_bounds\", Code: \"invalid_order\"}) }")
if name == "BitratePreference":
lines.append("\tif v.Mode == \"auto\" && v.TargetKbps != nil || v.Mode == \"explicit\" && v.TargetKbps == nil { violations = append(violations, FieldViolation{Field: \"target_kbps\", Code: \"invalid_tagged_value\"}) }")
if name == "GatewayQualityWorkRequest":
lines.append("\tif v.Acquisition == \"poll\" && (v.OperationID != \"\" || v.Revision != nil || v.LeaseGeneration != nil || v.CurrentAppliedRevision != nil) { violations = append(violations, FieldViolation{Field: \"acquisition\", Code: \"invalid_tagged_value\"}) }")
lines.append("\tif v.Acquisition == \"prompt\" && (v.OperationID == \"\" || v.Revision == nil || v.LeaseGeneration != nil || v.CurrentAppliedRevision != nil) { violations = append(violations, FieldViolation{Field: \"acquisition\", Code: \"invalid_tagged_value\"}) }")
lines.append("\tif v.Acquisition == \"observation\" && (v.OperationID == \"\" || v.Revision == nil || v.LeaseGeneration == nil || v.CurrentAppliedRevision == nil) { violations = append(violations, FieldViolation{Field: \"acquisition\", Code: \"invalid_tagged_value\"}) }")
if name == "GatewayStopWorkRequest":
lines.append("\tif v.Acquisition == \"poll\" && v.OperationID != \"\" || v.Acquisition == \"prompt\" && v.OperationID == \"\" { violations = append(violations, FieldViolation{Field: \"acquisition\", Code: \"invalid_tagged_value\"}) }")
if name == "GatewayQualityAck":
lines.append("\tif v.Outcome == \"applied\" && (v.CurrentAppliedRevision == nil || *v.CurrentAppliedRevision != v.Revision) { violations = append(violations, FieldViolation{Field: \"current_applied_revision\", Code: \"invalid_tagged_value\"}) }")
lines.append("\tif v.Outcome == \"proven_prior\" && (v.CurrentAppliedRevision == nil || *v.CurrentAppliedRevision >= v.Revision) { violations = append(violations, FieldViolation{Field: \"current_applied_revision\", Code: \"invalid_tagged_value\"}) }")
lines.append("\tif v.Outcome == \"unknown\" && v.CurrentAppliedRevision != nil { violations = append(violations, FieldViolation{Field: \"current_applied_revision\", Code: \"invalid_tagged_value\"}) }")
if name == "GatewayRegistration":
lines.append("\tif v.ProtocolMinVersion > v.ProtocolMaxVersion { violations = append(violations, FieldViolation{Field: \"protocol_version\", Code: \"invalid_order\"}) }")
if name == "ChannelFrame":
@@ -471,6 +481,16 @@ def rust_validation(definition: dict[str, Any]) -> list[str]:
lines.append(" if self.bitrateTargetKbps > self.bitrateMaximumKbps { return Err(ValidationError::new(\"bitrate_bounds\", \"invalid_order\")); }")
if name == "BitratePreference":
lines.append(" if self.mode == \"auto\" && self.targetKbps.is_some() || self.mode == \"explicit\" && self.targetKbps.is_none() { return Err(ValidationError::new(\"target_kbps\", \"invalid_tagged_value\")); }")
if name == "GatewayQualityWorkRequest":
lines.append(" if self.acquisition == \"poll\" && (self.operationId.is_some() || self.revision.is_some() || self.leaseGeneration.is_some() || self.currentAppliedRevision.is_some()) { return Err(ValidationError::new(\"acquisition\", \"invalid_tagged_value\")); }")
lines.append(" if self.acquisition == \"prompt\" && (self.operationId.is_none() || self.revision.is_none() || self.leaseGeneration.is_some() || self.currentAppliedRevision.is_some()) { return Err(ValidationError::new(\"acquisition\", \"invalid_tagged_value\")); }")
lines.append(" if self.acquisition == \"observation\" && (self.operationId.is_none() || self.revision.is_none() || self.leaseGeneration.is_none() || self.currentAppliedRevision.is_none()) { return Err(ValidationError::new(\"acquisition\", \"invalid_tagged_value\")); }")
if name == "GatewayStopWorkRequest":
lines.append(" if self.acquisition == \"poll\" && self.operationId.is_some() || self.acquisition == \"prompt\" && self.operationId.is_none() { return Err(ValidationError::new(\"acquisition\", \"invalid_tagged_value\")); }")
if name == "GatewayQualityAck":
lines.append(" if self.outcome == \"applied\" && self.currentAppliedRevision != Some(self.revision) { return Err(ValidationError::new(\"current_applied_revision\", \"invalid_tagged_value\")); }")
lines.append(" if self.outcome == \"proven_prior\" && self.currentAppliedRevision.map_or(true, |current| current >= self.revision) { return Err(ValidationError::new(\"current_applied_revision\", \"invalid_tagged_value\")); }")
lines.append(" if self.outcome == \"unknown\" && self.currentAppliedRevision.is_some() { return Err(ValidationError::new(\"current_applied_revision\", \"invalid_tagged_value\")); }")
if name == "GatewayRegistration":
lines.append(" if self.protocolMinVersion > self.protocolMaxVersion { return Err(ValidationError::new(\"protocol_version\", \"invalid_order\")); }")
if name == "ChannelFrame":
@@ -687,6 +707,16 @@ def swift_validation(definition: dict[str, Any]) -> list[str]:
lines.append(" if bitrateTargetKbps > bitrateMaximumKbps { throw ContractValidationError(field: \"bitrate_bounds\", code: \"invalid_order\") }")
if name == "BitratePreference":
lines.append(" if mode == \"auto\" && targetKbps != nil || mode == \"explicit\" && targetKbps == nil { throw ContractValidationError(field: \"target_kbps\", code: \"invalid_tagged_value\") }")
if name == "GatewayQualityWorkRequest":
lines.append(" if acquisition == \"poll\" && (operationId != nil || revision != nil || leaseGeneration != nil || currentAppliedRevision != nil) { throw ContractValidationError(field: \"acquisition\", code: \"invalid_tagged_value\") }")
lines.append(" if acquisition == \"prompt\" && (operationId == nil || revision == nil || leaseGeneration != nil || currentAppliedRevision != nil) { throw ContractValidationError(field: \"acquisition\", code: \"invalid_tagged_value\") }")
lines.append(" if acquisition == \"observation\" && (operationId == nil || revision == nil || leaseGeneration == nil || currentAppliedRevision == nil) { throw ContractValidationError(field: \"acquisition\", code: \"invalid_tagged_value\") }")
if name == "GatewayStopWorkRequest":
lines.append(" if acquisition == \"poll\" && operationId != nil || acquisition == \"prompt\" && operationId == nil { throw ContractValidationError(field: \"acquisition\", code: \"invalid_tagged_value\") }")
if name == "GatewayQualityAck":
lines.append(" if outcome == \"applied\" && currentAppliedRevision != revision { throw ContractValidationError(field: \"current_applied_revision\", code: \"invalid_tagged_value\") }")
lines.append(" if outcome == \"proven_prior\" && (currentAppliedRevision == nil || currentAppliedRevision! >= revision) { throw ContractValidationError(field: \"current_applied_revision\", code: \"invalid_tagged_value\") }")
lines.append(" if outcome == \"unknown\" && currentAppliedRevision != nil { throw ContractValidationError(field: \"current_applied_revision\", code: \"invalid_tagged_value\") }")
if name == "GatewayRegistration":
lines.append(" if protocolMinVersion > protocolMaxVersion { throw ContractValidationError(field: \"protocol_version\", code: \"invalid_order\") }")
if name == "ChannelFrame":
@@ -706,6 +736,7 @@ def generate_swift(defs: dict[str, dict[str, Any]], schema_hash: str, compatibil
"public struct ContractValidationError: Error, Equatable { public let field: String; public let code: String }",
"private struct AnyCodingKey: CodingKey { let stringValue: String; let intValue: Int?; init?(stringValue: String) { self.stringValue = stringValue; self.intValue = nil }; init?(intValue: Int) { self.stringValue = String(intValue); self.intValue = intValue } }",
"private func rejectDuplicateJSONKeys(_ data: Data) throws {",
" guard data.count <= 1_048_576 else { throw ContractValidationError(field: \"json\", code: \"payload_too_large\") }",
" var index = 0",
" func skipWhitespace() { while index < data.count && [9, 10, 13, 32].contains(data[index]) { index += 1 } }",
" func parseString() throws -> String {",
@@ -719,7 +750,8 @@ def generate_swift(defs: dict[str, dict[str, Any]], schema_hash: str, compatibil
" }",
" throw ContractValidationError(field: \"json\", code: \"invalid_json\")",
" }",
" func parseValue() throws {",
" func parseValue(_ depth: Int) throws {",
" guard depth <= 64 else { throw ContractValidationError(field: \"json\", code: \"nesting_too_deep\") }",
" skipWhitespace()",
" guard index < data.count else { throw ContractValidationError(field: \"json\", code: \"invalid_json\") }",
" if data[index] == 123 {",
@@ -734,7 +766,7 @@ def generate_swift(defs: dict[str, dict[str, Any]], schema_hash: str, compatibil
" skipWhitespace()",
" guard index < data.count, data[index] == 58 else { throw ContractValidationError(field: \"json\", code: \"invalid_json\") }",
" index += 1",
" try parseValue()",
" try parseValue(depth + 1)",
" skipWhitespace()",
" guard index < data.count else { throw ContractValidationError(field: \"json\", code: \"invalid_json\") }",
" if data[index] == 125 { index += 1; return }",
@@ -747,7 +779,7 @@ def generate_swift(defs: dict[str, dict[str, Any]], schema_hash: str, compatibil
" skipWhitespace()",
" if index < data.count, data[index] == 93 { index += 1; return }",
" while true {",
" try parseValue()",
" try parseValue(depth + 1)",
" skipWhitespace()",
" guard index < data.count else { throw ContractValidationError(field: \"json\", code: \"invalid_json\") }",
" if data[index] == 93 { index += 1; return }",
@@ -760,7 +792,7 @@ def generate_swift(defs: dict[str, dict[str, Any]], schema_hash: str, compatibil
" while index < data.count && ![9, 10, 13, 32, 44, 93, 125].contains(data[index]) { index += 1 }",
" guard index > start else { throw ContractValidationError(field: \"json\", code: \"invalid_json\") }",
" }",
" try parseValue()",
" try parseValue(0)",
" skipWhitespace()",
" guard index == data.count else { throw ContractValidationError(field: \"json\", code: \"trailing_json\") }",
"}",
tools/test_generated_contracts.py
+99 -16
View File
@@ -69,6 +69,21 @@ let audio = try AudioProfile(codec: "opus", sampleRateHz: 48000, channels: 2, ch
let display = try DisplayMode(resolutionWidth: 2560, resolutionHeight: 1440, fps: 120)
let adjustment = try SessionAdjustment(displayReason: "none", bitrateReason: "none")
let descriptor = try SelectedSessionDescriptor(videoProfile: video, audioProfile: audio, displayMode: display, bitrateTargetKbps: 40000, bitrateMaximumKbps: 50000, adjustment: adjustment, mediaTimestampBasis: "gateway-send-wall-clock-ms")
let operationId = "12345678-1234-1234-1234-123456789abc"
_ = try GatewayQualityWorkRequest(version: "1", sessionId: "session", gatewayId: "gateway", reconnectSequence: 2, acquisition: "poll", operationId: nil, revision: nil, leaseGeneration: nil, currentAppliedRevision: nil)
_ = try GatewayQualityWorkRequest(version: "1", sessionId: "session", gatewayId: "gateway", reconnectSequence: 2, acquisition: "prompt", operationId: operationId, revision: 7, leaseGeneration: nil, currentAppliedRevision: nil)
_ = try GatewayQualityWorkRequest(version: "1", sessionId: "session", gatewayId: "gateway", reconnectSequence: 2, acquisition: "observation", operationId: operationId, revision: 7, leaseGeneration: 3, currentAppliedRevision: 6)
do {
_ = try GatewayQualityWorkRequest(version: "1", sessionId: "session", gatewayId: "gateway", reconnectSequence: 2, acquisition: "poll", operationId: operationId, revision: 7, leaseGeneration: nil, currentAppliedRevision: nil)
fatalError("poll accepted unknown operation coordinates")
} catch { }
_ = try GatewayQualityAck(version: "1", sessionId: "session", gatewayId: "gateway", reconnectSequence: 2, operationId: operationId, revision: 7, leaseGeneration: 3, outcome: "applied", currentAppliedRevision: 7, failureCode: nil)
_ = try GatewayQualityAck(version: "1", sessionId: "session", gatewayId: "gateway", reconnectSequence: 2, operationId: operationId, revision: 7, leaseGeneration: 3, outcome: "proven_prior", currentAppliedRevision: 6, failureCode: nil)
_ = try GatewayQualityAck(version: "1", sessionId: "session", gatewayId: "gateway", reconnectSequence: 2, operationId: operationId, revision: 7, leaseGeneration: 3, outcome: "unknown", currentAppliedRevision: nil, failureCode: nil)
do {
_ = try GatewayQualityAck(version: "1", sessionId: "session", gatewayId: "gateway", reconnectSequence: 2, operationId: operationId, revision: 7, leaseGeneration: 3, outcome: "applied", currentAppliedRevision: 6, failureCode: nil)
fatalError("applied ack accepted a contradictory revision")
} catch { }
let capability = try CapabilityProfile(
transport: "quic-tls13", framing: "datagram-v1", media: "encoded",
sourceRateControl: "server", videoProfiles: [video], audioProfiles: [audio]
@@ -191,6 +206,19 @@ do {
_ = try CapabilityProfile.decodeJSON(duplicateCapability)
fatalError("capability accepted duplicate JSON keys")
} catch { }
do {
_ = try CapabilityProfile.decodeJSON(Data(repeating: 32, count: 1_048_577))
fatalError("capability accepted oversized JSON")
} catch let error as ContractValidationError {
guard error.code == "payload_too_large" else { fatalError("oversized JSON was not rejected before parsing") }
}
let deeplyNested = Data((String(repeating: "[", count: 65) + "null" + String(repeating: "]", count: 65)).utf8)
do {
_ = try CapabilityProfile.decodeJSON(deeplyNested)
fatalError("capability accepted over-deep JSON")
} catch let error as ContractValidationError {
guard error.code == "nesting_too_deep" else { fatalError("over-deep JSON was not rejected before decoding") }
}
for field in ["version", "session_id", "gateway_id", "audience", "reconnect_sequence", "expires_at", "capabilities", "selected_descriptor"] {
var missing = clientAuthorityObject
missing.removeValue(forKey: field)
@@ -385,6 +413,15 @@ fn main() {
let display = DisplayMode::new(2560, 1440, 120).unwrap();
let adjustment = SessionAdjustment::new("none".into(), "none".into()).unwrap();
let descriptor = SelectedSessionDescriptor::new(video.clone(), audio.clone(), display.clone(), 40000, 50000, adjustment, "gateway-send-wall-clock-ms".into()).unwrap();
let operation_id = "12345678-1234-1234-1234-123456789abc".to_string();
assert!(GatewayQualityWorkRequest::new("1".into(), "session".into(), "gateway".into(), 2, "poll".into(), None, None, None, None).is_ok());
assert!(GatewayQualityWorkRequest::new("1".into(), "session".into(), "gateway".into(), 2, "prompt".into(), Some(operation_id.clone()), Some(7), None, None).is_ok());
assert!(GatewayQualityWorkRequest::new("1".into(), "session".into(), "gateway".into(), 2, "observation".into(), Some(operation_id.clone()), Some(7), Some(3), Some(6)).is_ok());
assert!(GatewayQualityWorkRequest::new("1".into(), "session".into(), "gateway".into(), 2, "poll".into(), Some(operation_id.clone()), Some(7), None, None).is_err());
assert!(GatewayQualityAck::new("1".into(), "session".into(), "gateway".into(), 2, operation_id.clone(), 7, 3, "applied".into(), Some(7), None).is_ok());
assert!(GatewayQualityAck::new("1".into(), "session".into(), "gateway".into(), 2, operation_id.clone(), 7, 3, "proven_prior".into(), Some(6), None).is_ok());
assert!(GatewayQualityAck::new("1".into(), "session".into(), "gateway".into(), 2, operation_id.clone(), 7, 3, "unknown".into(), None, None).is_ok());
assert!(GatewayQualityAck::new("1".into(), "session".into(), "gateway".into(), 2, operation_id, 7, 3, "applied".into(), Some(6), None).is_err());
let capabilities = CapabilityProfile::new(
"quic-tls13".into(), "datagram-v1".into(), "encoded".into(),
"server".into(), vec![video.clone()], vec![audio.clone()],
@@ -544,22 +581,68 @@ fn main() {
)
run(["rustc", str(rust), "-o", str(workspace / "rust-contracts")], ROOT)
run([str(workspace / "rust-contracts")], ROOT)
expected_protobuf_fields = [
("version", 1),
("session_id", 2),
("gateway_id", 3),
("audience", 4),
("reconnect_sequence", 5),
("expires_at", 6),
("capabilities", 7),
("selected_descriptor", 8),
]
actual_protobuf_fields = protobuf_message_fields("ClientSessionAuthority")
if actual_protobuf_fields != expected_protobuf_fields:
raise RuntimeError(
f"ClientSessionAuthority protobuf fields = {actual_protobuf_fields}; "
f"want {expected_protobuf_fields}"
)
expected_protobuf_messages = {
"ClientSessionAuthority": [
("version", 1),
("session_id", 2),
("gateway_id", 3),
("audience", 4),
("reconnect_sequence", 5),
("expires_at", 6),
("capabilities", 7),
("selected_descriptor", 8),
],
"GatewayQualityWorkRequest": [
("version", 1),
("session_id", 2),
("gateway_id", 3),
("reconnect_sequence", 4),
("operation_id", 5),
("revision", 6),
("current_applied_revision", 7),
("acquisition", 8),
("lease_generation", 9),
],
"GatewayQualityWork": [
("version", 1),
("session_id", 2),
("gateway_id", 3),
("reconnect_sequence", 4),
("operation_id", 5),
("revision", 6),
("lease_expires_at", 7),
("selected_descriptor", 8),
("current_applied_revision", 9),
("lease_generation", 10),
],
"GatewayQualityAck": [
("version", 1),
("session_id", 2),
("gateway_id", 3),
("reconnect_sequence", 4),
("operation_id", 5),
("revision", 6),
("outcome", 7),
("current_applied_revision", 8),
("failure_code", 9),
("lease_generation", 10),
],
"GatewayStopWorkRequest": [
("version", 1),
("session_id", 2),
("gateway_id", 3),
("reconnect_sequence", 4),
("operation_id", 5),
("acquisition", 6),
],
}
for message, expected_protobuf_fields in expected_protobuf_messages.items():
actual_protobuf_fields = protobuf_message_fields(message)
if actual_protobuf_fields != expected_protobuf_fields:
raise RuntimeError(
f"{message} protobuf fields = {actual_protobuf_fields}; "
f"want {expected_protobuf_fields}"
)
rust_unknown = workspace / "unknown.rs"
shutil.copyfile(ROOT / "gen/rust/protocol.rs", rust_unknown)
with rust_unknown.open("a", encoding="utf-8") as output:
tools/validate.py
+34
View File
@@ -53,6 +53,27 @@ def main() -> int:
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"]
assert defs["GatewayQualityAck"]["required"][-3:] == ["revision", "lease_generation", "outcome"]
assert defs["GatewayQualityAck"]["properties"]["outcome"]["enum"] == ["applied", "proven_prior", "unknown"]
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"]
@@ -207,6 +228,18 @@ def main() -> int:
):
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
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
@@ -238,6 +271,7 @@ def main() -> int:
"""
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",