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

feat(core): run bounded real gateway sessions

Browse Source
This commit is contained in:
sechmachine
2026-08-12 22:03:10 +07:00
parent 86a95952b6
commit 519c04e18f
10 changed files with 1598 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/tests/abi_contract.rs
+153 -34
View File
@@ -2,17 +2,31 @@
unsafe_code,
clippy::borrow_as_ptr,
clippy::cast_possible_truncation,
clippy::items_after_statements
clippy::items_after_statements,
clashing_extern_declarations
)]
use std::cell::RefCell;
use std::collections::HashMap;
use std::ffi::c_void;
use std::mem::{offset_of, size_of};
use std::ptr;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
use std::sync::{Condvar, Mutex};
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::{Duration, Instant};
use rustls::sign::SigningKey;
use rustls::SignatureScheme;
#[path = "gateway_oracle.rs"]
mod gateway_oracle;
thread_local! {
static CORE_CONTEXTS: RefCell<HashMap<usize, usize>> = RefCell::new(HashMap::new());
static ORACLES: RefCell<HashMap<usize, gateway_oracle::Oracle>> = RefCell::new(HashMap::new());
}
const ABI_V1: u32 = 1;
const OK: u32 = 0;
const INVALID_ARGUMENT: u32 = 1;
@@ -167,6 +181,8 @@ struct Context {
cancel_on_connecting: AtomicBool,
reentry_target: AtomicUsize,
reentry_results: Mutex<Vec<u32>>,
admission_key: Mutex<Option<Arc<dyn SigningKey>>>,
tls_key: Mutex<Option<Arc<dyn SigningKey>>>,
}
impl Default for Context {
@@ -191,6 +207,8 @@ impl Default for Context {
cancel_on_connecting: AtomicBool::new(false),
reentry_target: AtomicUsize::new(0),
reentry_results: Mutex::new(Vec::new()),
admission_key: Mutex::new(None),
tls_key: Mutex::new(None),
}
}
}
@@ -206,9 +224,22 @@ unsafe extern "C" fn sign_admission(raw: *mut c_void, input: BytesView, output:
let ctx = unsafe { context(raw) };
ctx.admission_calls.fetch_add(1, Ordering::SeqCst);
*ctx.admission_input.lock().expect("admission lock") = bytes.to_vec();
let status = ctx.admission_status.load(Ordering::SeqCst);
if status != OK {
return status;
}
let key = ctx.admission_key.lock().expect("admission key");
let Some(key) = key.as_ref() else {
return INTERNAL;
};
let signer = key
.choose_scheme(&[SignatureScheme::ED25519])
.expect("Ed25519 admission signer");
let signature = signer.sign(bytes).expect("admission signature");
assert_eq!(signature.len(), 64);
// Test invariant: ABI promises a writable 64-byte Rust-owned signature buffer.
unsafe { ptr::write_bytes(output, 0xA5, 64) };
ctx.admission_status.load(Ordering::SeqCst)
unsafe { ptr::copy_nonoverlapping(signature.as_ptr(), output, signature.len()) };
OK
}
unsafe extern "C" fn sign_tls(raw: *mut c_void, input: BytesView, output: *mut u8) -> u32 {
@@ -217,9 +248,22 @@ unsafe extern "C" fn sign_tls(raw: *mut c_void, input: BytesView, output: *mut u
let ctx = unsafe { context(raw) };
ctx.tls_calls.fetch_add(1, Ordering::SeqCst);
*ctx.tls_input.lock().expect("tls lock") = bytes.to_vec();
let status = ctx.tls_status.load(Ordering::SeqCst);
if status != OK {
return status;
}
let key = ctx.tls_key.lock().expect("TLS key");
let Some(key) = key.as_ref() else {
return INTERNAL;
};
let signer = key
.choose_scheme(&[SignatureScheme::ED25519])
.expect("Ed25519 TLS signer");
let signature = signer.sign(bytes).expect("TLS signature");
assert_eq!(signature.len(), 64);
// Test invariant: ABI promises a writable 64-byte Rust-owned signature buffer.
unsafe { ptr::write_bytes(output, 0x5A, 64) };
ctx.tls_status.load(Ordering::SeqCst)
unsafe { ptr::copy_nonoverlapping(signature.as_ptr(), output, signature.len()) };
OK
}
unsafe extern "C" fn sign_admission_probes_global_reentry(
@@ -391,14 +435,66 @@ fn create(ctx: &mut Context) -> *mut Core {
// Test invariant: config/out pointers remain valid for the synchronous create call.
assert_eq!(unsafe { verse_core_create_v1(&config, &mut core) }, OK);
assert!(!core.is_null());
ctx.core.store(core as usize, Ordering::SeqCst);
register(core, ctx);
core
}
fn register(core: *mut Core, ctx: &mut Context) {
ctx.core.store(core as usize, Ordering::SeqCst);
CORE_CONTEXTS.with(|contexts| {
contexts
.borrow_mut()
.insert(core as usize, ptr::from_mut(ctx) as usize);
});
}
fn connect(core: *mut Core, manifest: &[u8], credential: &[u8]) -> u32 {
let request = request(manifest, credential);
// Test invariant: request and backing byte slices remain valid for the synchronous call.
unsafe { verse_core_connect_v1(core, &request) }
connect_mode(core, "", manifest, credential)
}
fn connect_mode(core: *mut Core, mode: &str, manifest: &[u8], credential: &[u8]) -> u32 {
connect_with_oracle(core, mode, |oracle| {
let manifest = if manifest == MANIFEST {
oracle.ready.manifest.as_bytes()
} else {
manifest
};
let credential = if credential == CREDENTIAL {
oracle.ready.credential.as_bytes()
} else {
credential
};
let request = request(manifest, credential);
// Test invariant: request and backing byte slices remain valid for the synchronous call.
unsafe { verse_core_connect_v1(core, &request) }
})
}
fn connect_with_oracle(
core: *mut Core,
mode: &str,
action: impl FnOnce(&gateway_oracle::Oracle) -> u32,
) -> u32 {
let oracle = gateway_oracle::Oracle::start(mode);
CORE_CONTEXTS.with(|contexts| {
let raw = *contexts
.borrow()
.get(&(core as usize))
.expect("registered ABI context");
// Test invariant: context outlives the core and is removed only after destroy succeeds.
let ctx = unsafe { &*(raw as *const Context) };
*ctx.admission_key.lock().expect("admission key") =
Some(gateway_oracle::test_key(&oracle.ready.admission_key));
*ctx.tls_key.lock().expect("TLS key") =
Some(gateway_oracle::test_key(&oracle.ready.client_key));
});
let status = action(&oracle);
if status == OK {
ORACLES.with(|oracles| {
oracles.borrow_mut().insert(core as usize, oracle);
});
}
status
}
fn wait_for(ctx: &Context, predicate: impl Fn(&[u32]) -> bool) {
@@ -419,7 +515,16 @@ fn wait_for(ctx: &Context, predicate: impl Fn(&[u32]) -> bool) {
fn destroy(core: *mut Core) -> u32 {
// Test invariant: caller retains the handle until destroy reports success.
unsafe { verse_core_destroy_v1(core, 2_000) }
let status = unsafe { verse_core_destroy_v1(core, 2_000) };
if status == OK {
ORACLES.with(|oracles| {
oracles.borrow_mut().remove(&(core as usize));
});
CORE_CONTEXTS.with(|contexts| {
contexts.borrow_mut().remove(&(core as usize));
});
}
status
}
#[test]
@@ -500,8 +605,11 @@ fn connect_copies_inputs_and_calls_purpose_specific_signers_once() {
assert_eq!(ctx.admission_calls.load(Ordering::SeqCst), 1);
assert_eq!(ctx.tls_calls.load(Ordering::SeqCst), 1);
assert_eq!(&*ctx.admission_input.lock().expect("admission"), MANIFEST);
assert_eq!(&*ctx.tls_input.lock().expect("tls"), CREDENTIAL);
let admission = ctx.admission_input.lock().expect("admission");
let tls = ctx.tls_input.lock().expect("tls");
assert!(admission.starts_with(b"versevdi/tunnel-admission/v1"));
assert!(!tls.is_empty());
assert_ne!(&*admission, &*tls);
assert_eq!(destroy(core), OK);
}
@@ -564,13 +672,18 @@ fn tables_and_slices_reject_short_unsupported_null_and_oversized_inputs() {
base: ConnectRequest,
ignored: [u8; 24],
}
let mut trailing_request = ExtendedRequest {
base: request(MANIFEST, CREDENTIAL),
ignored: [0xEE; 24],
};
trailing_request.base.struct_size = size_of::<ExtendedRequest>() as u32;
assert_eq!(
unsafe { verse_core_connect_v1(core, &trailing_request.base) },
connect_with_oracle(core, "", |oracle| {
let mut trailing_request = ExtendedRequest {
base: request(
oracle.ready.manifest.as_bytes(),
oracle.ready.credential.as_bytes(),
),
ignored: [0xEE; 24],
};
trailing_request.base.struct_size = size_of::<ExtendedRequest>() as u32;
unsafe { verse_core_connect_v1(core, &trailing_request.base) }
}),
OK
);
@@ -616,7 +729,7 @@ fn signer_callbacks_cannot_reenter_even_cancel() {
cfg.sign_admission = Some(sign_admission_reenters);
let mut core = ptr::null_mut();
assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK);
ctx.core.store(core as usize, Ordering::SeqCst);
register(core, &mut ctx);
assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK);
assert_eq!(ctx.reentry_cancel.load(Ordering::SeqCst), REENTRANT);
assert_eq!(destroy(core), OK);
@@ -634,7 +747,7 @@ fn signer_callback_rejects_every_stateful_api_across_handles() {
cfg.sign_admission = Some(sign_admission_probes_global_reentry);
let mut core = ptr::null_mut();
assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK);
ctx.core.store(core as usize, Ordering::SeqCst);
register(core, &mut ctx);
assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK);
assert_eq!(
ctx.reentry_results
@@ -644,8 +757,8 @@ fn signer_callback_rejects_every_stateful_api_across_handles() {
[ABI_V1, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT]
);
assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK);
assert_eq!(destroy(core), OK);
assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK);
assert_eq!(destroy(other), OK);
}
@@ -661,7 +774,7 @@ fn event_callback_allows_only_originating_handle_cancel() {
cfg.on_state = Some(on_state_probes_global_reentry);
let mut core = ptr::null_mut();
assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK);
ctx.core.store(core as usize, Ordering::SeqCst);
register(core, &mut ctx);
assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK);
wait_for(&ctx, |states| states.contains(&STATE_CANCELLED));
assert_eq!(
@@ -672,8 +785,8 @@ fn event_callback_allows_only_originating_handle_cancel() {
[ABI_V1, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT, REENTRANT, OK]
);
assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK);
assert_eq!(destroy(core), OK);
assert_eq!(connect(other, MANIFEST, CREDENTIAL), OK);
assert_eq!(destroy(other), OK);
}
@@ -720,7 +833,7 @@ fn cancel_during_connect_preserves_state_order_and_stops_before_tls_signing() {
cfg.sign_admission = Some(sign_admission_waits_for_cancel);
let mut core = ptr::null_mut();
assert_eq!(unsafe { verse_core_create_v1(&cfg, &mut core) }, OK);
ctx.core.store(core as usize, Ordering::SeqCst);
register(core, &mut ctx);
assert_eq!(connect(core, MANIFEST, CREDENTIAL), CANCELLED);
wait_for(&ctx, |states| states.contains(&STATE_CANCELLED));
@@ -737,16 +850,22 @@ fn send_input_is_nonblocking_bounded_and_cancel_is_idempotent() {
let mut ctx = Context::default();
ctx.reentry_cancel.store(OK, Ordering::SeqCst);
let core = create(&mut ctx);
assert_eq!(connect(core, MANIFEST, CREDENTIAL), OK);
assert_eq!(connect_mode(core, "slow-input", MANIFEST, CREDENTIAL), OK);
let event = keyboard_event();
for _ in 0..64 {
assert_eq!(unsafe { verse_core_send_input_v1(core, &event) }, OK);
}
assert_eq!(
unsafe { verse_core_send_input_v1(core, &event) },
QUEUE_FULL
);
assert_eq!(unsafe { verse_core_request_idr_v1(core) }, OK);
let started = Instant::now();
let mut sent = 0;
let saturated = loop {
match unsafe { verse_core_send_input_v1(core, &event) } {
OK => sent += 1,
QUEUE_FULL => break true,
status => panic!("unexpected input status {status}"),
}
if sent == 10_000 || started.elapsed() > Duration::from_secs(1) {
break false;
}
};
assert!(saturated, "real slow consumer did not expose bounded queue");
assert!(started.elapsed() < Duration::from_secs(1));
assert_eq!(unsafe { verse_core_cancel_v1(core) }, OK);
assert_eq!(unsafe { verse_core_cancel_v1(core) }, OK);
assert_eq!(unsafe { verse_core_send_input_v1(core, &event) }, CANCELLED);
core/tests/ffi/abi_smoke.c
+53 -32
View File
@@ -8,7 +8,7 @@
static const char MANIFEST[] =
"{\"version\":\"1\",\"purpose\":\"launch\",\"session_id\":\"session\","
"\"reconnect_sequence\":0,\"gateway\":{\"id\":\"gateway\",\"addresses\":["
"\"gateway.test:443\"],\"public_identity\":\"gateway.test\"},\"tunnel\":{"
"\"127.0.0.1:9\"],\"public_identity\":\"gateway.test\"},\"tunnel\":{"
"\"versions\":[\"verse-gateway-v1/1\"],\"features\":[\"control.v1\"]},"
"\"profile\":{\"id\":\"standard\",\"bounds\":{\"minimum_kbps\":1000,"
"\"target_kbps\":5000,\"maximum_kbps\":10000},\"display_mode\":null},"
@@ -24,36 +24,28 @@ static const char CREDENTIAL[] =
typedef struct smoke_context {
atomic_uint admission_calls;
atomic_uint tls_calls;
atomic_uint state_calls;
} smoke_context_t;
static verse_status_t sign_admission(
static verse_status_t reject_admission(
void *raw,
verse_bytes_view_t input,
uint8_t signature[64]) {
smoke_context_t *context = raw;
assert(input.length == sizeof(MANIFEST) - 1U);
memset(signature, 0xA5, 64U);
(void)input;
(void)signature;
atomic_fetch_add(&context->admission_calls, 1U);
return VERSE_STATUS_OK;
return VERSE_STATUS_INTERNAL;
}
static verse_status_t sign_tls(
static verse_status_t reject_tls(
void *raw,
verse_bytes_view_t input,
uint8_t signature[64]) {
smoke_context_t *context = raw;
assert(input.length == sizeof(CREDENTIAL) - 1U);
memset(signature, 0x5A, 64U);
(void)input;
(void)signature;
atomic_fetch_add(&context->tls_calls, 1U);
return VERSE_STATUS_OK;
}
static void on_state(void *raw, const verse_state_event_v1_t *event) {
smoke_context_t *context = raw;
assert(event->struct_size == sizeof(*event));
assert(event->abi_version == VERSE_CORE_ABI_VERSION_1);
atomic_fetch_add(&context->state_calls, 1U);
return VERSE_STATUS_INTERNAL;
}
int main(void) {
@@ -62,35 +54,64 @@ int main(void) {
.struct_size = sizeof(config),
.abi_version = VERSE_CORE_ABI_VERSION_1,
.context = &context,
.sign_admission = sign_admission,
.sign_tls_ed25519 = sign_tls,
.on_state = on_state,
.sign_admission = reject_admission,
.sign_tls_ed25519 = reject_tls,
};
verse_core_t *core = NULL;
assert(verse_core_abi_version() == VERSE_CORE_ABI_VERSION_1);
assert(verse_core_create_v1(NULL, &core) == VERSE_STATUS_INVALID_ARGUMENT);
assert(verse_core_create_v1(&config, NULL) == VERSE_STATUS_INVALID_ARGUMENT);
config.struct_size = 8U;
assert(verse_core_create_v1(&config, &core) == VERSE_STATUS_INVALID_ARGUMENT);
config.struct_size = sizeof(config);
config.abi_version = VERSE_CORE_ABI_VERSION_1 + 1U;
assert(verse_core_create_v1(&config, &core) == VERSE_STATUS_UNSUPPORTED_ABI);
config.abi_version = VERSE_CORE_ABI_VERSION_1;
assert(verse_core_create_v1(&config, &core) == VERSE_STATUS_OK);
assert(core != NULL);
const verse_connect_request_v1_t request = {
.struct_size = sizeof(request),
.abi_version = VERSE_CORE_ABI_VERSION_1,
.manifest_json = {(const uint8_t *)MANIFEST, sizeof(MANIFEST) - 1U},
.tunnel_credential_json = {(const uint8_t *)CREDENTIAL, sizeof(CREDENTIAL) - 1U},
};
assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_OK);
verse_input_event_v1_t input = {
.struct_size = sizeof(input),
.abi_version = VERSE_CORE_ABI_VERSION_1,
.kind = VERSE_INPUT_KEYBOARD,
.values = {1, 0, 30},
};
assert(verse_core_send_input_v1(core, &input) == VERSE_STATUS_OK);
assert(verse_core_request_idr_v1(core) == VERSE_STATUS_OK);
assert(verse_core_send_input_v1(core, &input) == VERSE_STATUS_INVALID_STATE);
assert(verse_core_request_idr_v1(core) == VERSE_STATUS_INVALID_STATE);
char manifest[sizeof(MANIFEST)];
char credential[sizeof(CREDENTIAL)];
memcpy(manifest, MANIFEST, sizeof(manifest));
memcpy(credential, CREDENTIAL, sizeof(credential));
verse_connect_request_v1_t request = {
.struct_size = sizeof(request),
.abi_version = VERSE_CORE_ABI_VERSION_1,
.manifest_json = {(const uint8_t *)manifest, sizeof(manifest) - 1U},
.tunnel_credential_json = {(const uint8_t *)credential, sizeof(credential) - 1U},
};
assert(verse_core_connect_v1(NULL, &request) == VERSE_STATUS_INVALID_ARGUMENT);
request.struct_size = 8U;
assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_INVALID_ARGUMENT);
request.struct_size = sizeof(request);
request.abi_version = VERSE_CORE_ABI_VERSION_1 + 1U;
assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_UNSUPPORTED_ABI);
request.abi_version = VERSE_CORE_ABI_VERSION_1;
request.manifest_json.data = NULL;
assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_INVALID_ARGUMENT);
request.manifest_json.data = (const uint8_t *)manifest;
assert(verse_core_connect_v1(core, &request) == VERSE_STATUS_TLS);
memset(manifest, 0, sizeof(manifest));
memset(credential, 0, sizeof(credential));
assert(verse_core_send_input_v1(core, &input) == VERSE_STATUS_CANCELLED);
assert(verse_core_request_idr_v1(core) == VERSE_STATUS_CANCELLED);
assert(verse_core_cancel_v1(core) == VERSE_STATUS_OK);
assert(verse_core_cancel_v1(core) == VERSE_STATUS_OK);
assert(atomic_load(&context.admission_calls) == 0U);
assert(atomic_load(&context.tls_calls) == 0U);
assert(verse_core_destroy_v1(core, 2000U) == VERSE_STATUS_OK);
assert(atomic_load(&context.admission_calls) == 1U);
assert(atomic_load(&context.tls_calls) == 1U);
assert(atomic_load(&context.admission_calls) == 0U);
assert(atomic_load(&context.tls_calls) == 0U);
return EXIT_SUCCESS;
}
core/tests/gateway_oracle.rs
+326 -27
View File
@@ -1,5 +1,8 @@
use std::ffi::c_void;
use std::fs;
use std::mem::size_of;
use std::process::{Child, Command, Stdio};
use std::ptr;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
@@ -13,7 +16,8 @@ use rustls::SignatureScheme;
use serde::Deserialize;
use versevdi_core::error::CoreError;
use versevdi_core::transport::{
connect, connect_with_cancellation, AdmissionSigner, Cancellation, Signers, TlsEd25519Signer,
bounded_session_events, connect, connect_with_cancellation, AdmissionSigner, Cancellation,
SessionCommand, SessionEvent, Signers, TlsEd25519Signer,
};
use versevdi_core::wire::{ConnectionManifest, NativeTunnelCredential};
@@ -76,8 +80,8 @@ func (a *admission) Release(context.Context, protocol.SessionAuthority) error {
return nil
}
type provider struct { admission *admission; failStart bool; delay time.Duration }
type session struct { state protocol.ProviderState; video chan gateway.ProviderMedia; audio chan gateway.ProviderMedia; events chan gateway.ProviderEvent }
type provider struct { admission *admission; failStart bool; delay time.Duration; mode string; inputPath string; releasePath string }
type session struct { state protocol.ProviderState; video chan gateway.ProviderMedia; audio chan gateway.ProviderMedia; events chan gateway.ProviderEvent; mode string; inputPath string; releasePath string }
func (p provider) Start(_ context.Context, request gateway.LaunchRequest) (gateway.ProviderSession, error) {
p.admission.mu.Lock()
p.admission.providerStarts++
@@ -86,18 +90,24 @@ func (p provider) Start(_ context.Context, request gateway.LaunchRequest) (gatew
p.admission.mu.Unlock()
if p.failStart { return nil, context.DeadlineExceeded }
if p.delay > 0 { time.Sleep(p.delay) }
return &session{state: protocol.ProviderState{Version:"1", SessionID:request.SessionID, State:gateway.ProviderStateReady, Channels:[]string{"video","audio","input","feedback"}}, video:make(chan gateway.ProviderMedia), audio:make(chan gateway.ProviderMedia), events:make(chan gateway.ProviderEvent)}, nil
current := &session{state: protocol.ProviderState{Version:"1", SessionID:request.SessionID, State:gateway.ProviderStateReady, Channels:[]string{"video","audio","input","feedback"}}, video:make(chan gateway.ProviderMedia,2), audio:make(chan gateway.ProviderMedia,2), events:make(chan gateway.ProviderEvent,2), mode:p.mode, inputPath:p.inputPath, releasePath:p.releasePath}
if p.mode == "io" || p.mode == "input" {
current.video <- gateway.ProviderMedia{Payload:bytes.Repeat([]byte{0xA5}, 2000)}
current.audio <- gateway.ProviderMedia{Payload:[]byte{0x10,0x20,0x30}}
current.events <- gateway.ProviderEvent{Kind:gateway.ProviderEventRumble,Payload:[]byte{0,0,1,0,2}}
}
return current, nil
}
func (s *session) Ready(context.Context) error { return nil }
func (s *session) Video() <-chan gateway.ProviderMedia { return s.video }
func (s *session) Audio() <-chan gateway.ProviderMedia { return s.audio }
func (s *session) Events() <-chan gateway.ProviderEvent { return s.events }
func (s *session) Input(context.Context, gateway.InputEvent) error { return nil }
func (s *session) Input(_ context.Context, event gateway.InputEvent) error { if s.mode == "slow-input" { time.Sleep(100*time.Millisecond) }; data,_ := json.Marshal(event); if err := os.WriteFile(s.inputPath,data,0600); err != nil { panic(err) }; return nil }
func (s *session) Feedback(context.Context, gateway.Feedback) error { return nil }
func (s *session) ReadClipboard(context.Context) (string, error) { return "", errors.New("disabled") }
func (s *session) WriteClipboard(context.Context, string) error { return errors.New("disabled") }
func (s *session) Telemetry() gateway.ProviderTelemetry { return gateway.ProviderTelemetry{State:s.state.State} }
func (s *session) ReleaseAll(context.Context) error { return nil }
func (s *session) ReleaseAll(context.Context) error { return os.WriteFile(s.releasePath,[]byte("released"),0600) }
func (s *session) Terminate(context.Context) error { return nil }
func (s *session) State() protocol.ProviderState { return s.state }
@@ -128,11 +138,11 @@ func main() {
if mode == "session" { authority.SessionID = "other-session" }
if mode == "gateway" { authority.GatewayID = "other-gateway" }
if mode == "audience" { authority.Audience = "other-audience" }
if mode == "reconnect" { authority.ReconnectSequence = 1 }
if mode == "reconnect" || mode == "reconnect-success" { authority.ReconnectSequence = 1 }
if mode == "expiry" { authority.ExpiresAt = time.Now().Add(-time.Minute).UTC().Truncate(time.Second).Format(time.RFC3339) }
if mode == "capability" { authority.Capabilities.ClientDecode = []string{"hevc-opus"} }
if mode == "alpn" { server.NextProtos = []string{"wrong-alpn"} }
work := protocol.ProviderSessionWork{Version:"1",SessionID:authority.SessionID,GatewayID:authority.GatewayID,ExpiresAt:expiry,ProviderProfile:gateway.ProviderProfileApollo,ProviderIdentity:authority.ProviderIdentity,PolicyVersionID:"policy-1",StreamPolicy:protocol.ProviderStreamPolicy{ResolutionWidth:1920,ResolutionHeight:1080,Fps:60,Codec:"H264",BitrateKbps:8000,AudioEnabled:true},ApplicationID:"1",ClientID:"client",ManagementHost:"provider.invalid",ManagementPort:47990,StreamHost:"provider.invalid",StreamPort:47984,ClientCertificatePem:"certificate",ClientPrivateKeyPem:"private-key",ServerCertificatePem:"certificate",ClipboardPolicy:protocol.ClipboardPolicy{MaxTextBytes:65536,MaxUpdatesPerMinute:30}}
work := protocol.ProviderSessionWork{Version:"1",SessionID:authority.SessionID,GatewayID:authority.GatewayID,ReconnectSequence:authority.ReconnectSequence,ExpiresAt:expiry,ProviderProfile:gateway.ProviderProfileApollo,ProviderIdentity:authority.ProviderIdentity,PolicyVersionID:"policy-1",StreamPolicy:protocol.ProviderStreamPolicy{ResolutionWidth:1920,ResolutionHeight:1080,Fps:60,Codec:"H264",BitrateKbps:8000,AudioEnabled:true},ApplicationID:"1",ClientID:"client",ManagementHost:"provider.invalid",ManagementPort:47990,StreamHost:"provider.invalid",StreamPort:47984,ClientCertificatePem:"certificate",ClientPrivateKeyPem:"private-key",ServerCertificatePem:"certificate",ClipboardPolicy:protocol.ClipboardPolicy{MaxTextBytes:65536,MaxUpdatesPerMinute:30}}
replayGuard := &admission{authority:authority,work:work,public:admissionKey.Public().(ed25519.PublicKey)}
replayRequest := protocol.TunnelAdmissionRequest{Version:"1",SessionID:"session-1",GatewayID:"gateway-1",Audience:"versevdi-gateway",Grant:"ggggggggggggggggggggggggggggggggggggggggggg",ClientNonce:"bm9uY2UtZm9yLXJlcGxheQ",Capabilities:gateway.DefaultCapabilities()}
replayRequest.DeviceSignature = base64.RawURLEncoding.EncodeToString(ed25519.Sign(admissionKey,replayRequest.DeviceAdmissionTranscript()))
@@ -140,10 +150,12 @@ func main() {
_, secondReplayErr := replayGuard.Admit(context.Background(),replayRequest)
replayRejected := firstReplayErr == nil && errors.Is(secondReplayErr,gateway.ErrAdmissionRejected)
mutationPath := readyPath + ".provider-starts"
inputPath := readyPath + ".input"
releasePath := readyPath + ".release-all"
if err := os.WriteFile(mutationPath, []byte("0"), 0600); err != nil { panic(err) }
admissionService := &admission{authority:authority,work:work,public:admissionKey.Public().(ed25519.PublicKey),reusable:mode == "reusable",failWorkOnce:mode == "post-retryable",releaseFails:mode == "cleanup-release-failure",mutationPath:mutationPath}
newService := func(delay time.Duration) *gateway.Server {
service, err := gateway.NewServer(gateway.ServerConfig{ListenAddress:"127.0.0.1:0",TLSConfig:server,GatewayID:authority.GatewayID,Admission:admissionService,Provider:provider{admission:admissionService,failStart:mode == "provider-start-lost-response" || mode == "cleanup-release-failure",delay:delay}}); if err != nil { panic(err) }
service, err := gateway.NewServer(gateway.ServerConfig{ListenAddress:"127.0.0.1:0",TLSConfig:server,GatewayID:authority.GatewayID,Admission:admissionService,Provider:provider{admission:admissionService,failStart:mode == "provider-start-lost-response" || mode == "cleanup-release-failure",delay:delay,mode:mode,inputPath:inputPath,releasePath:releasePath}}); if err != nil { panic(err) }
return service
}
firstDelay := time.Duration(0); if mode == "lost-authority" { firstDelay = 300*time.Millisecond }
@@ -165,7 +177,8 @@ func main() {
done := make(chan error,len(services)); for _, current := range services { go func(server *gateway.Server){done <- server.Serve(ctx)}(current) }
keyDER, err := x509.MarshalPKCS8PrivateKey(clientKey); if err != nil { panic(err) }
admissionKeyDER, err := x509.MarshalPKCS8PrivateKey(admissionKey); if err != nil { panic(err) }
manifest := map[string]any{"version":"1","purpose":"launch","session_id":"session-1","reconnect_sequence":0,"gateway":map[string]any{"id":"gateway-1","addresses":addresses,"public_identity":"gateway.test"},"tunnel":map[string]any{"versions":[]string{"verse-gateway-v1/1"},"features":[]string{"control.v1"}},"profile":map[string]any{"id":"standard","bounds":map[string]any{"minimum_kbps":1000,"target_kbps":5000,"maximum_kbps":10000},"display_mode":map[string]any{"resolution_width":1920,"resolution_height":1080,"fps":60}},"grant":map[string]any{"opaque_value":"ggggggggggggggggggggggggggggggggggggggggggg","expires_at":"2099-01-01T00:00:00Z","audience":"versevdi-gateway"},"correlation_id":"oracle"}
purpose := "launch"; reconnectSequence := 0; if mode == "reconnect-success" { purpose = "reconnect"; reconnectSequence = 1 }
manifest := map[string]any{"version":"1","purpose":purpose,"session_id":"session-1","reconnect_sequence":reconnectSequence,"gateway":map[string]any{"id":"gateway-1","addresses":addresses,"public_identity":"gateway.test"},"tunnel":map[string]any{"versions":[]string{"verse-gateway-v1/1"},"features":[]string{"control.v1","input.absolute.v1","input.scroll.v1"}},"profile":map[string]any{"id":"standard","bounds":map[string]any{"minimum_kbps":1000,"target_kbps":5000,"maximum_kbps":10000},"display_mode":map[string]any{"resolution_width":1920,"resolution_height":1080,"fps":60}},"grant":map[string]any{"opaque_value":"ggggggggggggggggggggggggggggggggggggggggggg","expires_at":"2099-01-01T00:00:00Z","audience":"versevdi-gateway"},"correlation_id":"oracle"}
credential := map[string]any{"client_device_id":"device","device_key_id":"key","certificate_chain_pem":pemCert(clientDER,caDER),"trust_bundle_pem":pemCert(caDER),"expires_at":"2099-01-01T00:00:00Z"}
manifestJSON,_ := json.Marshal(manifest); credentialJSON,_ := json.Marshal(credential)
wrongKeyDER, err := x509.MarshalPKCS8PrivateKey(wrongClientKey); if err != nil { panic(err) }
@@ -178,27 +191,27 @@ func main() {
"#;
#[derive(Deserialize)]
struct Ready {
manifest: String,
credential: String,
admission_key: String,
client_key: String,
wrong_client_key: String,
server_key: String,
server_chain: String,
bad_trust: String,
replay_rejected: bool,
pub(crate) struct Ready {
pub(crate) manifest: String,
pub(crate) credential: String,
pub(crate) admission_key: String,
pub(crate) client_key: String,
pub(crate) wrong_client_key: String,
pub(crate) server_key: String,
pub(crate) server_chain: String,
pub(crate) bad_trust: String,
pub(crate) replay_rejected: bool,
}
struct Oracle {
pub(crate) struct Oracle {
_serial: std::sync::MutexGuard<'static, ()>,
child: Child,
directory: std::path::PathBuf,
ready: Ready,
pub(crate) ready: Ready,
}
impl Oracle {
fn start(mode: &str) -> Self {
pub(crate) fn start(mode: &str) -> Self {
static SERIAL: Mutex<()> = Mutex::new(());
static NEXT: AtomicUsize = AtomicUsize::new(1);
let serial = SERIAL
@@ -253,6 +266,27 @@ impl Oracle {
.parse()
.expect("provider mutation count")
}
fn input(&self) -> Option<Vec<u8>> {
fs::read(self.directory.join("ready.json.input")).ok()
}
fn release_all(&self) -> bool {
self.directory.join("ready.json.release-all").exists()
}
fn files(&self) -> Vec<String> {
fs::read_dir(&self.directory)
.expect("read oracle directory")
.map(|entry| {
entry
.expect("oracle entry")
.file_name()
.to_string_lossy()
.into_owned()
})
.collect()
}
}
impl Drop for Oracle {
@@ -263,7 +297,7 @@ impl Drop for Oracle {
}
}
fn test_key(encoded: &str) -> Arc<dyn SigningKey> {
pub(crate) fn test_key(encoded: &str) -> Arc<dyn SigningKey> {
let der = STANDARD.decode(encoded).expect("decode test key");
rustls::crypto::ring::default_provider()
.key_provider
@@ -339,6 +373,237 @@ fn callback_ed25519_signer_completes_tls13_quic_admission_without_private_key_in
runtime.block_on(session.close());
}
#[test]
fn production_gateway_session_delivers_bounded_media_control_and_ordered_input() {
let oracle = Oracle::start("input");
let runtime = tokio::runtime::Runtime::new().expect("runtime");
let session = runtime
.block_on(connect_oracle_async(
&oracle,
&oracle.ready.admission_key,
&oracle.ready.client_key,
|_| {},
|_| {},
))
.expect("production Go gateway admission");
let (command_tx, command_rx) = tokio::sync::mpsc::channel(64);
let (event_tx, _event_rx) = bounded_session_events();
let cancellation = Cancellation::new();
let running_cancellation = cancellation.clone();
let mut running = runtime.spawn(session.run(command_rx, event_tx, running_cancellation));
runtime
.block_on(command_tx.send(SessionCommand::Input(
b"VGI1\x01\x04\x01\x00\x00\x1e".to_vec(),
)))
.expect("bounded input");
runtime.block_on(async { tokio::time::sleep(Duration::from_millis(100)).await });
let input_deadline = Instant::now() + Duration::from_secs(2);
while oracle.input().is_none() && Instant::now() < input_deadline {
thread::sleep(Duration::from_millis(10));
}
let early = runtime
.block_on(async { tokio::time::timeout(Duration::from_millis(10), &mut running).await });
if let (None, Ok(completed)) = (oracle.input(), early) {
panic!(
"gateway did not receive input; session result={:?}",
completed.expect("session task")
);
}
assert!(
oracle.input().is_some(),
"gateway did not receive input: {:?}",
oracle.files()
);
cancellation.cancel();
assert_eq!(
runtime.block_on(running).expect("session task"),
Err(CoreError::Cancelled)
);
let release_deadline = Instant::now() + Duration::from_secs(2);
while !oracle.release_all() && Instant::now() < release_deadline {
thread::sleep(Duration::from_millis(10));
}
assert!(
oracle.release_all(),
"gateway did not release pressed input"
);
}
#[test]
fn production_gateway_session_delivers_bounded_media_and_control() {
let oracle = Oracle::start("io");
let runtime = tokio::runtime::Runtime::new().expect("runtime");
let session = runtime
.block_on(connect_oracle_async(
&oracle,
&oracle.ready.admission_key,
&oracle.ready.client_key,
|_| {},
|_| {},
))
.expect("production Go gateway admission");
let (_command_tx, command_rx) = tokio::sync::mpsc::channel(64);
let (event_tx, event_rx) = bounded_session_events();
let cancellation = Cancellation::new();
let running_cancellation = cancellation.clone();
let running = runtime.spawn(session.run(command_rx, event_tx, running_cancellation));
let mut media = Vec::new();
let mut control = Vec::new();
let deadline = Instant::now() + Duration::from_secs(5);
while (media.len() < 2 || control.is_empty()) && Instant::now() < deadline {
match event_rx.recv_timeout(Duration::from_millis(100)) {
Ok(SessionEvent::Media(unit)) => media.push(unit),
Ok(SessionEvent::Control { kind, payload }) => control.push((kind, payload)),
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
Err(error) => panic!("session event channel: {error}"),
}
}
media.sort_by_key(|unit| unit.channel as u8);
assert_eq!(media.len(), 2);
assert!(media.iter().any(|unit| unit.payload == vec![0xA5; 2000]));
assert!(media
.iter()
.any(|unit| unit.payload == vec![0x10, 0x20, 0x30]));
assert_eq!(control, vec![(0x11, vec![0, 0, 1, 0, 2])]);
cancellation.cancel();
assert_eq!(
runtime.block_on(running).expect("session task"),
Err(CoreError::Cancelled)
);
}
#[repr(C)]
struct AbiCore {
_private: [u8; 0],
}
#[repr(C)]
#[derive(Clone, Copy)]
struct AbiBytes {
data: *const u8,
length: usize,
}
#[repr(C)]
struct AbiConfig {
struct_size: u32,
abi_version: u32,
context: *mut c_void,
sign_admission: Option<unsafe extern "C" fn(*mut c_void, AbiBytes, *mut u8) -> u32>,
sign_tls_ed25519: Option<unsafe extern "C" fn(*mut c_void, AbiBytes, *mut u8) -> u32>,
on_state: Option<unsafe extern "C" fn(*mut c_void, *const u8)>,
on_error: Option<unsafe extern "C" fn(*mut c_void, *const u8)>,
on_stats: Option<unsafe extern "C" fn(*mut c_void, *const u8)>,
on_media: Option<unsafe extern "C" fn(*mut c_void, *const u8)>,
on_control: Option<unsafe extern "C" fn(*mut c_void, *const u8)>,
}
#[repr(C)]
struct AbiConnect {
struct_size: u32,
abi_version: u32,
manifest_json: AbiBytes,
tunnel_credential_json: AbiBytes,
}
#[repr(C)]
struct AbiInput {
struct_size: u32,
abi_version: u32,
kind: u32,
flags: u32,
values: [i32; 12],
}
unsafe extern "C" {
fn verse_core_create_v1(config: *const AbiConfig, out: *mut *mut AbiCore) -> u32;
fn verse_core_connect_v1(core: *mut AbiCore, request: *const AbiConnect) -> u32;
fn verse_core_send_input_v1(core: *mut AbiCore, event: *const AbiInput) -> u32;
fn verse_core_cancel_v1(core: *mut AbiCore) -> u32;
fn verse_core_destroy_v1(core: *mut AbiCore, timeout_ms: u32) -> u32;
}
struct AbiSigners {
admission: Arc<dyn SigningKey>,
tls: Arc<dyn SigningKey>,
}
unsafe extern "C" fn abi_admission(raw: *mut c_void, input: AbiBytes, output: *mut u8) -> u32 {
unsafe { abi_sign(raw, input, output, true) }
}
unsafe extern "C" fn abi_tls(raw: *mut c_void, input: AbiBytes, output: *mut u8) -> u32 {
unsafe { abi_sign(raw, input, output, false) }
}
unsafe fn abi_sign(raw: *mut c_void, input: AbiBytes, output: *mut u8, admission: bool) -> u32 {
let context = unsafe { &*raw.cast::<AbiSigners>() };
let bytes = unsafe { std::slice::from_raw_parts(input.data, input.length) };
let key = if admission {
&context.admission
} else {
&context.tls
};
let signature = key
.choose_scheme(&[SignatureScheme::ED25519])
.expect("Ed25519")
.sign(bytes)
.expect("sign");
unsafe { ptr::copy_nonoverlapping(signature.as_ptr(), output, 64) };
0
}
#[test]
fn real_abi_exports_connect_send_cancel_and_destroy_the_go_gateway_session() {
let oracle = Oracle::start("input");
let mut signers = Box::new(AbiSigners {
admission: test_key(&oracle.ready.admission_key),
tls: test_key(&oracle.ready.client_key),
});
let config = AbiConfig {
struct_size: size_of::<AbiConfig>() as u32,
abi_version: 1,
context: ptr::from_mut(&mut *signers).cast(),
sign_admission: Some(abi_admission),
sign_tls_ed25519: Some(abi_tls),
on_state: None,
on_error: None,
on_stats: None,
on_media: None,
on_control: None,
};
let mut core = ptr::null_mut();
assert_eq!(unsafe { verse_core_create_v1(&config, &mut core) }, 0);
let request = AbiConnect {
struct_size: size_of::<AbiConnect>() as u32,
abi_version: 1,
manifest_json: AbiBytes {
data: oracle.ready.manifest.as_ptr(),
length: oracle.ready.manifest.len(),
},
tunnel_credential_json: AbiBytes {
data: oracle.ready.credential.as_ptr(),
length: oracle.ready.credential.len(),
},
};
assert_eq!(unsafe { verse_core_connect_v1(core, &request) }, 0);
let mut values = [0; 12];
values[0] = 1;
values[2] = 30;
let input = AbiInput {
struct_size: size_of::<AbiInput>() as u32,
abi_version: 1,
kind: 1,
flags: 0,
values,
};
assert_eq!(unsafe { verse_core_send_input_v1(core, &input) }, 0);
let deadline = Instant::now() + Duration::from_secs(2);
while oracle.input().is_none() && Instant::now() < deadline {
thread::sleep(Duration::from_millis(10));
}
assert!(oracle.input().is_some(), "ABI input did not reach gateway");
assert_eq!(unsafe { verse_core_cancel_v1(core) }, 0);
assert_eq!(unsafe { verse_core_destroy_v1(core, 2_000) }, 0);
assert!(
oracle.release_all(),
"destroy did not release provider input"
);
}
fn signer(key: Arc<dyn SigningKey>) -> impl Fn(&[u8]) -> Result<[u8; 64], CoreError> {
move |input| {
key.choose_scheme(&[SignatureScheme::ED25519])
@@ -356,6 +621,24 @@ fn connect_oracle(
tls_key: &str,
mutate_manifest: impl FnOnce(&mut serde_json::Value),
mutate_credential: impl FnOnce(&mut serde_json::Value),
) -> Result<versevdi_core::transport::TransportSession, CoreError> {
tokio::runtime::Runtime::new()
.expect("runtime")
.block_on(connect_oracle_async(
oracle,
admission_key,
tls_key,
mutate_manifest,
mutate_credential,
))
}
async fn connect_oracle_async(
oracle: &Oracle,
admission_key: &str,
tls_key: &str,
mutate_manifest: impl FnOnce(&mut serde_json::Value),
mutate_credential: impl FnOnce(&mut serde_json::Value),
) -> Result<versevdi_core::transport::TransportSession, CoreError> {
let mut manifest: serde_json::Value =
serde_json::from_str(&oracle.ready.manifest).expect("manifest JSON");
@@ -368,8 +651,7 @@ fn connect_oracle(
let credential = NativeTunnelCredential::decode(
&serde_json::to_vec(&credential).expect("encode credential"),
)?;
let runtime = tokio::runtime::Runtime::new().expect("runtime");
runtime.block_on(connect(
connect(
&manifest,
&credential,
Signers::new(
@@ -378,7 +660,8 @@ fn connect_oracle(
),
"2026-08-12T00:00:00Z",
Duration::from_secs(5),
))
)
.await
}
#[test]
@@ -465,6 +748,22 @@ fn production_gateway_and_rust_reject_authority_binding_and_capability_mismatche
}
}
#[test]
fn production_gateway_accepts_reconnect_only_when_exact_sequence_is_bound() {
let oracle = Oracle::start("reconnect-success");
let session = connect_oracle(
&oracle,
&oracle.ready.admission_key,
&oracle.ready.client_key,
|_| {},
|_| {},
)
.expect("exact reconnect sequence");
tokio::runtime::Runtime::new()
.expect("runtime")
.block_on(session.close());
}
#[test]
fn signer_purposes_are_not_interchangeable() {
let oracle = Oracle::start("");
core/tests/protocol_fixtures.rs
+18
View File
@@ -533,6 +533,16 @@ fn reassembly_discards_conflicts_and_expires_after_250_ms() {
assert_eq!(reassembler.expired_units(), 1);
}
#[test]
fn reassembly_expires_at_exactly_250_ms() {
let mut reassembler = Reassembler::new();
let expiring = MediaFragment::new_video(2, 1, 0, 2, vec![1]).expect("fragment");
reassembler.push(expiring, 10).expect("first");
let next = MediaFragment::new_video(3, 1, 0, 2, vec![1]).expect("fragment");
reassembler.push(next, 260).expect("expiry sweep");
assert_eq!(reassembler.expired_units(), 1);
}
#[test]
fn reassembly_rejects_a_complete_unit_above_one_mebibyte() {
let mut reassembler = Reassembler::new();
@@ -565,6 +575,14 @@ fn input_features_and_feedback_booleans_fail_closed() {
);
}
#[test]
fn disconnected_gateway_feedback_is_provider_free_control() {
assert_eq!(
decode_feedback(&decode_hex("5647463101130000")).expect("disconnected feedback"),
FeedbackEvent::Disconnected
);
}
#[test]
fn fec_feedback_enforces_rc5_go_field_invariants_on_decode_and_encode() {
for invalid in [