1383 lines
56 KiB
Rust
1383 lines
56 KiB
Rust
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;
|
|
use std::time::{Duration, Instant};
|
|
|
|
use base64::engine::general_purpose::STANDARD;
|
|
use base64::Engine as _;
|
|
use rustls::pki_types::{PrivateKeyDer, PrivatePkcs8KeyDer};
|
|
use rustls::sign::SigningKey;
|
|
use rustls::SignatureScheme;
|
|
use serde::Deserialize;
|
|
use versevdi_core::error::CoreError;
|
|
use versevdi_core::transport::{
|
|
bounded_session_events, connect, connect_with_cancellation, AdmissionSigner, Cancellation,
|
|
SessionCommand, SessionEvent, Signers, TlsEd25519Signer,
|
|
};
|
|
use versevdi_core::wire::{ConnectionManifest, NativeTunnelCredential};
|
|
|
|
const ORACLE_SOURCE: &str = r#"
|
|
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/ed25519"
|
|
"crypto/rand"
|
|
"crypto/tls"
|
|
"crypto/x509"
|
|
"crypto/x509/pkix"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"encoding/pem"
|
|
"errors"
|
|
"math/big"
|
|
"os"
|
|
"strconv"
|
|
"sync"
|
|
"time"
|
|
|
|
gateway "git.sechmachine.io.vn/sechmachine/VerseVDI-Data-Plane/gateway"
|
|
protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol"
|
|
)
|
|
|
|
type admission struct {
|
|
mu sync.Mutex
|
|
lastTranscript []byte
|
|
reusable bool
|
|
failWorkOnce bool
|
|
releaseFails bool
|
|
mutationPath string
|
|
providerStarts int
|
|
authority protocol.SessionAuthority
|
|
work protocol.ProviderSessionWork
|
|
public ed25519.PublicKey
|
|
}
|
|
|
|
func (a *admission) Admit(_ context.Context, request protocol.TunnelAdmissionRequest) (protocol.SessionAuthority, error) {
|
|
signature, err := base64.RawURLEncoding.DecodeString(request.DeviceSignature)
|
|
transcript := request.DeviceAdmissionTranscript()
|
|
if err != nil || !ed25519.Verify(a.public, transcript, signature) {
|
|
return protocol.SessionAuthority{}, gateway.ErrAdmissionRejected
|
|
}
|
|
a.mu.Lock(); defer a.mu.Unlock()
|
|
if !a.reusable && bytes.Equal(a.lastTranscript, transcript) { return protocol.SessionAuthority{}, gateway.ErrAdmissionRejected }
|
|
a.lastTranscript = append(a.lastTranscript[:0], transcript...)
|
|
return a.authority, nil
|
|
}
|
|
func (a *admission) ProviderWork(context.Context, protocol.SessionAuthority) (protocol.ProviderSessionWork, error) {
|
|
a.mu.Lock(); defer a.mu.Unlock()
|
|
if a.failWorkOnce { a.failWorkOnce = false; return protocol.ProviderSessionWork{}, context.DeadlineExceeded }
|
|
return a.work, nil
|
|
}
|
|
func (a *admission) Release(context.Context, protocol.SessionAuthority) error {
|
|
if a.releaseFails { return errors.New("release failed") }
|
|
return nil
|
|
}
|
|
|
|
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++
|
|
count := p.admission.providerStarts
|
|
if err := os.WriteFile(p.admission.mutationPath, []byte(strconv.Itoa(count)), 0600); err != nil { panic(err) }
|
|
p.admission.mu.Unlock()
|
|
if p.failStart { return nil, context.DeadlineExceeded }
|
|
if p.delay > 0 { time.Sleep(p.delay) }
|
|
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, 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 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 }
|
|
|
|
func makeCertificate(parent *x509.Certificate, parentKey ed25519.PrivateKey, serial int64, dns string, usage x509.ExtKeyUsage, isCA bool) ([]byte, ed25519.PrivateKey) {
|
|
public, private, err := ed25519.GenerateKey(rand.Reader); if err != nil { panic(err) }
|
|
template := &x509.Certificate{SerialNumber:big.NewInt(serial), Subject:pkix.Name{CommonName:dns}, DNSNames:[]string{dns}, NotBefore:time.Now().Add(-time.Hour), NotAfter:time.Now().Add(time.Hour), IsCA:isCA, BasicConstraintsValid:true, KeyUsage:x509.KeyUsageDigitalSignature}
|
|
if isCA { template.KeyUsage |= x509.KeyUsageCertSign } else { template.ExtKeyUsage = []x509.ExtKeyUsage{usage} }
|
|
if parent == nil { parent = template; parentKey = private }
|
|
der, err := x509.CreateCertificate(rand.Reader, template, parent, public, parentKey); if err != nil { panic(err) }
|
|
return der, private
|
|
}
|
|
func pemCert(ders ...[]byte) string { var out []byte; for _, der := range ders { out = append(out, pem.EncodeToMemory(&pem.Block{Type:"CERTIFICATE",Bytes:der})...) }; return string(out) }
|
|
|
|
func main() {
|
|
readyPath, stopPath := os.Args[1], os.Args[2]
|
|
mode := os.Getenv("VERSEVDI_RUST_ORACLE_MODE")
|
|
caDER, caKey := makeCertificate(nil, nil, 1, "Verse Rust Oracle CA", 0, true)
|
|
ca, err := x509.ParseCertificate(caDER); if err != nil { panic(err) }
|
|
serverDER, serverKey := makeCertificate(ca, caKey, 2, "gateway.test", x509.ExtKeyUsageServerAuth, false)
|
|
clientDER, clientKey := makeCertificate(ca, caKey, 3, "client.test", x509.ExtKeyUsageClientAuth, false)
|
|
_, admissionKey := makeCertificate(ca, caKey, 6, "admission.test", x509.ExtKeyUsageClientAuth, false)
|
|
badCADER, _ := makeCertificate(nil, nil, 4, "Wrong CA", 0, true)
|
|
_, wrongClientKey := makeCertificate(ca, caKey, 5, "wrong-client.test", x509.ExtKeyUsageClientAuth, false)
|
|
pool := x509.NewCertPool(); pool.AddCert(ca)
|
|
server := &tls.Config{MinVersion:tls.VersionTLS13, MaxVersion:tls.VersionTLS13, Certificates:[]tls.Certificate{{Certificate:[][]byte{serverDER,caDER},PrivateKey:serverKey}}, ClientAuth:tls.RequireAndVerifyClientCert, ClientCAs:pool}
|
|
expiry := time.Now().Add(5*time.Minute).UTC().Truncate(time.Second).Format(time.RFC3339)
|
|
authority := protocol.SessionAuthority{Version:"1",SessionID:"session-1",GatewayID:"gateway-1",Audience:"versevdi-gateway",ExpiresAt:expiry,Capabilities:gateway.DefaultCapabilities(),ProviderProfile:gateway.ProviderProfileApollo,ProviderIdentity:"oracle#sha256:fixture"}
|
|
if mode == "session" { authority.SessionID = "other-session" }
|
|
if mode == "gateway" { authority.GatewayID = "other-gateway" }
|
|
if mode == "audience" { authority.Audience = "other-audience" }
|
|
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,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()))
|
|
_, firstReplayErr := replayGuard.Admit(context.Background(),replayRequest)
|
|
_, 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,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 }
|
|
service := newService(firstDelay)
|
|
services := []*gateway.Server{service}
|
|
addresses := []string{service.Addr().String()}
|
|
if mode == "retryable" {
|
|
draining := newService(0)
|
|
draining.BeginDrain()
|
|
services = []*gateway.Server{draining, service}
|
|
addresses = []string{draining.Addr().String(), service.Addr().String()}
|
|
}
|
|
if mode == "post-retryable" || mode == "provider-start-lost-response" || mode == "cleanup-release-failure" || mode == "lost-authority" {
|
|
second := newService(0)
|
|
services = []*gateway.Server{service, second}
|
|
addresses = []string{service.Addr().String(), second.Addr().String()}
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background()); defer cancel()
|
|
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) }
|
|
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) }
|
|
serverKeyDER, err := x509.MarshalPKCS8PrivateKey(serverKey); if err != nil { panic(err) }
|
|
ready,_ := json.Marshal(map[string]any{"manifest":string(manifestJSON),"credential":string(credentialJSON),"admission_key":base64.StdEncoding.EncodeToString(admissionKeyDER),"client_key":base64.StdEncoding.EncodeToString(keyDER),"wrong_client_key":base64.StdEncoding.EncodeToString(wrongKeyDER),"server_key":base64.StdEncoding.EncodeToString(serverKeyDER),"server_chain":pemCert(serverDER,caDER),"bad_trust":pemCert(badCADER),"replay_rejected":replayRejected})
|
|
if err := os.WriteFile(readyPath,ready,0600); err != nil { panic(err) }
|
|
for { if _, err := os.Stat(stopPath); err == nil { break }; time.Sleep(10*time.Millisecond) }
|
|
cancel(); for _, current := range services { _ = current.Close() }; for range services { <-done }
|
|
}
|
|
"#;
|
|
|
|
#[derive(Deserialize)]
|
|
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,
|
|
}
|
|
|
|
pub(crate) struct Oracle {
|
|
child: Child,
|
|
directory: std::path::PathBuf,
|
|
pub(crate) ready: Ready,
|
|
}
|
|
|
|
struct OracleStartup {
|
|
child: Option<Child>,
|
|
directory: Option<std::path::PathBuf>,
|
|
}
|
|
|
|
impl OracleStartup {
|
|
fn new(directory: std::path::PathBuf) -> Self {
|
|
Self {
|
|
child: None,
|
|
directory: Some(directory),
|
|
}
|
|
}
|
|
|
|
fn finish(mut self, ready: Ready) -> Oracle {
|
|
Oracle {
|
|
child: self.child.take().expect("started oracle child"),
|
|
directory: self.directory.take().expect("oracle directory"),
|
|
ready,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Drop for OracleStartup {
|
|
fn drop(&mut self) {
|
|
if let Some(child) = self.child.as_mut() {
|
|
stop_child(child, self.directory.as_deref());
|
|
}
|
|
if let Some(directory) = self.directory.as_ref() {
|
|
let _ = fs::remove_dir_all(directory);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn production_oracles_can_run_concurrently_with_isolated_state() {
|
|
let (first_ready_tx, first_ready_rx) = std::sync::mpsc::channel();
|
|
let (release_tx, release_rx) = std::sync::mpsc::channel();
|
|
let first = thread::spawn(move || {
|
|
let oracle = Oracle::start("");
|
|
first_ready_tx.send(()).expect("first ready");
|
|
release_rx.recv().expect("release first oracle");
|
|
drop(oracle);
|
|
});
|
|
first_ready_rx
|
|
.recv_timeout(Duration::from_secs(20))
|
|
.expect("first oracle ready");
|
|
let (second_ready_tx, second_ready_rx) = std::sync::mpsc::channel();
|
|
let second = thread::spawn(move || {
|
|
let oracle = Oracle::start("");
|
|
second_ready_tx.send(()).expect("second ready");
|
|
drop(oracle);
|
|
});
|
|
let concurrent = second_ready_rx.recv_timeout(Duration::from_secs(5)).is_ok();
|
|
release_tx.send(()).expect("release first");
|
|
first.join().expect("first oracle");
|
|
second.join().expect("second oracle");
|
|
assert!(concurrent, "second isolated oracle was globally serialized");
|
|
}
|
|
|
|
#[test]
|
|
fn oracle_startup_timeout_kills_child_and_removes_directory() {
|
|
let directory = Oracle::unique_directory();
|
|
let mut command = Command::new("sh");
|
|
command.args(["-c", "while :; do sleep 1; done"]);
|
|
let started = Instant::now();
|
|
|
|
let result = Oracle::start_in("", directory.clone(), command, Duration::from_millis(20));
|
|
|
|
assert!(result.is_err(), "non-ready child unexpectedly became ready");
|
|
assert!(started.elapsed() < Duration::from_secs(2));
|
|
assert!(!directory.exists(), "timed-out oracle directory leaked");
|
|
}
|
|
|
|
#[test]
|
|
fn oracle_source_failure_removes_directory_before_spawn() {
|
|
let directory = Oracle::unique_directory();
|
|
fs::create_dir(directory.join("main.go")).expect("block source file creation");
|
|
|
|
let result = Oracle::start_in(
|
|
"",
|
|
directory.clone(),
|
|
Command::new("go"),
|
|
Duration::from_millis(20),
|
|
);
|
|
|
|
assert!(result.is_err(), "invalid source path unexpectedly started");
|
|
assert!(!directory.exists(), "failed oracle directory leaked");
|
|
}
|
|
|
|
#[test]
|
|
fn oracle_spawn_failure_removes_directory() {
|
|
let directory = Oracle::unique_directory();
|
|
|
|
let result = Oracle::start_in(
|
|
"",
|
|
directory.clone(),
|
|
Command::new("/definitely/missing/versevdi-go"),
|
|
Duration::from_millis(20),
|
|
);
|
|
|
|
assert!(result.is_err(), "missing executable unexpectedly started");
|
|
assert!(!directory.exists(), "spawn-failed oracle directory leaked");
|
|
}
|
|
|
|
impl Oracle {
|
|
pub(crate) fn start(mode: &str) -> Self {
|
|
let directory = Self::unique_directory();
|
|
Self::start_in(mode, directory, Command::new("go"), Duration::from_secs(20))
|
|
.unwrap_or_else(|error| panic!("{error}"))
|
|
}
|
|
|
|
fn unique_directory() -> std::path::PathBuf {
|
|
static NEXT: AtomicUsize = AtomicUsize::new(1);
|
|
loop {
|
|
let candidate = std::env::temp_dir().join(format!(
|
|
"versevdi-rust-gateway-oracle-{}-{}",
|
|
std::process::id(),
|
|
NEXT.fetch_add(1, Ordering::Relaxed)
|
|
));
|
|
match fs::create_dir(&candidate) {
|
|
Ok(()) => break candidate,
|
|
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
|
|
Err(error) => panic!("create oracle directory: {error}"),
|
|
}
|
|
}
|
|
}
|
|
|
|
fn start_in(
|
|
mode: &str,
|
|
directory: std::path::PathBuf,
|
|
mut command: Command,
|
|
timeout: Duration,
|
|
) -> Result<Self, String> {
|
|
let mut startup = OracleStartup::new(directory.clone());
|
|
let source = directory.join("main.go");
|
|
let ready_path = directory.join("ready.json");
|
|
let stop_path = directory.join("stop");
|
|
fs::write(&source, ORACLE_SOURCE)
|
|
.map_err(|error| format!("write Go oracle source: {error}"))?;
|
|
let source = source
|
|
.to_str()
|
|
.ok_or_else(|| "Go oracle source path is not UTF-8".to_owned())?;
|
|
let child = command
|
|
.args(["run", source])
|
|
.arg(&ready_path)
|
|
.arg(&stop_path)
|
|
.env("VERSEVDI_RUST_ORACLE_MODE", mode)
|
|
.current_dir(env!("CARGO_MANIFEST_DIR"))
|
|
.stdout(Stdio::piped())
|
|
.stderr(Stdio::piped())
|
|
.spawn()
|
|
.map_err(|error| format!("start Go gateway oracle: {error}"))?;
|
|
startup.child = Some(child);
|
|
let deadline = Instant::now() + timeout;
|
|
while Instant::now() < deadline {
|
|
if let Ok(bytes) = fs::read(&ready_path) {
|
|
if let Ok(ready) = serde_json::from_slice(&bytes) {
|
|
return Ok(startup.finish(ready));
|
|
}
|
|
}
|
|
if let Some(status) = startup
|
|
.child
|
|
.as_mut()
|
|
.expect("started oracle child")
|
|
.try_wait()
|
|
.map_err(|error| format!("poll Go gateway oracle: {error}"))?
|
|
{
|
|
return Err(format!(
|
|
"Go gateway oracle exited before becoming ready: status={status}"
|
|
));
|
|
}
|
|
thread::sleep(Duration::from_millis(10));
|
|
}
|
|
Err("Go gateway oracle did not become ready before the startup deadline".to_owned())
|
|
}
|
|
|
|
fn provider_starts(&self) -> usize {
|
|
fs::read_to_string(self.directory.join("ready.json.provider-starts"))
|
|
.expect("read provider mutation count")
|
|
.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 {
|
|
fn drop(&mut self) {
|
|
stop_child(&mut self.child, Some(&self.directory));
|
|
let _ = fs::remove_dir_all(&self.directory);
|
|
}
|
|
}
|
|
|
|
fn stop_child(child: &mut Child, directory: Option<&std::path::Path>) {
|
|
if let Some(directory) = directory {
|
|
let _ = fs::write(directory.join("stop"), []);
|
|
}
|
|
if wait_for_child(child, Duration::from_millis(250)) {
|
|
return;
|
|
}
|
|
if child.kill().is_ok() {
|
|
let _ = wait_for_child(child, Duration::from_millis(500));
|
|
}
|
|
}
|
|
|
|
fn wait_for_child(child: &mut Child, timeout: Duration) -> bool {
|
|
let deadline = Instant::now() + timeout;
|
|
loop {
|
|
match child.try_wait() {
|
|
Ok(Some(_)) => return true,
|
|
Ok(None) if Instant::now() < deadline => thread::sleep(Duration::from_millis(5)),
|
|
Ok(None) | Err(_) => return false,
|
|
}
|
|
}
|
|
}
|
|
|
|
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
|
|
.load_private_key(PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(der)))
|
|
.expect("load ephemeral Ed25519 test key")
|
|
}
|
|
|
|
#[test]
|
|
fn callback_ed25519_signer_completes_tls13_quic_admission_without_private_key_input() {
|
|
let oracle = Oracle::start("");
|
|
let manifest = ConnectionManifest::decode(oracle.ready.manifest.as_bytes()).expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let admission_key = test_key(&oracle.ready.admission_key);
|
|
let tls_key = test_key(&oracle.ready.client_key);
|
|
let admission_inputs = Arc::new(Mutex::new(Vec::new()));
|
|
let tls_inputs = Arc::new(Mutex::new(Vec::new()));
|
|
let admission_calls = Arc::new(AtomicUsize::new(0));
|
|
let tls_calls = Arc::new(AtomicUsize::new(0));
|
|
|
|
let sign = |key: Arc<dyn SigningKey>, input: &[u8]| -> Result<[u8; 64], CoreError> {
|
|
let signer = key
|
|
.choose_scheme(&[SignatureScheme::ED25519])
|
|
.ok_or(CoreError::Tls)?;
|
|
signer
|
|
.sign(input)
|
|
.map_err(|_| CoreError::Tls)?
|
|
.try_into()
|
|
.map_err(|_| CoreError::Tls)
|
|
};
|
|
let admission = {
|
|
let key = Arc::clone(&admission_key);
|
|
let inputs = Arc::clone(&admission_inputs);
|
|
let calls = Arc::clone(&admission_calls);
|
|
AdmissionSigner::new(move |input| {
|
|
calls.fetch_add(1, Ordering::SeqCst);
|
|
*inputs.lock().expect("admission inputs") = input.to_vec();
|
|
sign(Arc::clone(&key), input)
|
|
})
|
|
};
|
|
let tls = {
|
|
let key = Arc::clone(&tls_key);
|
|
let inputs = Arc::clone(&tls_inputs);
|
|
let calls = Arc::clone(&tls_calls);
|
|
TlsEd25519Signer::new(move |input| {
|
|
calls.fetch_add(1, Ordering::SeqCst);
|
|
*inputs.lock().expect("TLS inputs") = input.to_vec();
|
|
sign(Arc::clone(&key), input)
|
|
})
|
|
};
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
let session = runtime
|
|
.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
Signers::new(admission, tls),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(10),
|
|
))
|
|
.expect("production Go gateway admission");
|
|
|
|
assert_eq!(session.authority().session_id(), "session-1");
|
|
assert_eq!(admission_calls.load(Ordering::SeqCst), 1);
|
|
assert_eq!(tls_calls.load(Ordering::SeqCst), 1);
|
|
assert!(admission_inputs
|
|
.lock()
|
|
.expect("admission transcript")
|
|
.starts_with(b"versevdi/tunnel-admission/v1"));
|
|
assert_ne!(
|
|
*admission_inputs.lock().expect("admission transcript"),
|
|
*tls_inputs.lock().expect("TLS transcript")
|
|
);
|
|
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>,
|
|
admission_threads: Mutex<Vec<thread::ThreadId>>,
|
|
tls_threads: Mutex<Vec<thread::ThreadId>>,
|
|
}
|
|
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 threads = if admission {
|
|
&context.admission_threads
|
|
} else {
|
|
&context.tls_threads
|
|
};
|
|
threads
|
|
.lock()
|
|
.expect("signer threads")
|
|
.push(thread::current().id());
|
|
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),
|
|
admission_threads: Mutex::new(Vec::new()),
|
|
tls_threads: Mutex::new(Vec::new()),
|
|
});
|
|
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(),
|
|
},
|
|
};
|
|
let calling_thread = thread::current().id();
|
|
assert_eq!(unsafe { verse_core_connect_v1(core, &request) }, 0);
|
|
assert_eq!(
|
|
signers
|
|
.admission_threads
|
|
.lock()
|
|
.expect("admission threads")
|
|
.as_slice(),
|
|
[calling_thread]
|
|
);
|
|
assert_eq!(
|
|
signers.tls_threads.lock().expect("TLS threads").as_slice(),
|
|
[calling_thread]
|
|
);
|
|
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])
|
|
.ok_or(CoreError::Tls)?
|
|
.sign(input)
|
|
.map_err(|_| CoreError::Tls)?
|
|
.try_into()
|
|
.map_err(|_| CoreError::Tls)
|
|
}
|
|
}
|
|
|
|
fn connect_oracle(
|
|
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> {
|
|
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");
|
|
let mut credential: serde_json::Value =
|
|
serde_json::from_str(&oracle.ready.credential).expect("credential JSON");
|
|
mutate_manifest(&mut manifest);
|
|
mutate_credential(&mut credential);
|
|
let manifest =
|
|
ConnectionManifest::decode(&serde_json::to_vec(&manifest).expect("encode manifest"))?;
|
|
let credential = NativeTunnelCredential::decode(
|
|
&serde_json::to_vec(&credential).expect("encode credential"),
|
|
)?;
|
|
connect(
|
|
&manifest,
|
|
&credential,
|
|
Signers::new(
|
|
AdmissionSigner::new(signer(test_key(admission_key))),
|
|
TlsEd25519Signer::new(signer(test_key(tls_key))),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(5),
|
|
)
|
|
.await
|
|
}
|
|
|
|
#[test]
|
|
fn production_gateway_rejects_tls_and_admission_identity_mismatches() {
|
|
for case in [
|
|
"wrong-key",
|
|
"wrong-signature",
|
|
"wrong-leaf",
|
|
"wrong-sni",
|
|
"wrong-root",
|
|
"wrong-alpn",
|
|
] {
|
|
let oracle = Oracle::start("");
|
|
let admission_key = &oracle.ready.admission_key;
|
|
let tls_key = &oracle.ready.client_key;
|
|
let wrong = &oracle.ready.wrong_client_key;
|
|
let result = match case {
|
|
"wrong-key" => connect_oracle(&oracle, admission_key, wrong, |_| {}, |_| {}),
|
|
"wrong-signature" => connect_oracle(&oracle, wrong, tls_key, |_| {}, |_| {}),
|
|
"wrong-leaf" => connect_oracle(
|
|
&oracle,
|
|
admission_key,
|
|
&oracle.ready.server_key,
|
|
|_| {},
|
|
|credential| {
|
|
credential["certificate_chain_pem"] = oracle.ready.server_chain.clone().into();
|
|
},
|
|
),
|
|
"wrong-sni" => connect_oracle(
|
|
&oracle,
|
|
admission_key,
|
|
tls_key,
|
|
|manifest| manifest["gateway"]["public_identity"] = "other.test".into(),
|
|
|_| {},
|
|
),
|
|
"wrong-root" => connect_oracle(
|
|
&oracle,
|
|
admission_key,
|
|
tls_key,
|
|
|_| {},
|
|
|credential| credential["trust_bundle_pem"] = oracle.ready.bad_trust.clone().into(),
|
|
),
|
|
"wrong-alpn" => {
|
|
drop(oracle);
|
|
let oracle = Oracle::start("alpn");
|
|
connect_oracle(
|
|
&oracle,
|
|
&oracle.ready.admission_key,
|
|
&oracle.ready.client_key,
|
|
|_| {},
|
|
|_| {},
|
|
)
|
|
}
|
|
_ => unreachable!(),
|
|
};
|
|
let expected = if case == "wrong-signature" {
|
|
CoreError::AuthorityRejected
|
|
} else {
|
|
CoreError::Tls
|
|
};
|
|
assert_eq!(result.err(), Some(expected), "wrong error for {case}");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn production_gateway_and_rust_reject_authority_binding_and_capability_mismatches() {
|
|
for mode in [
|
|
"session",
|
|
"gateway",
|
|
"audience",
|
|
"reconnect",
|
|
"capability",
|
|
"expiry",
|
|
] {
|
|
let oracle = Oracle::start(mode);
|
|
let result = connect_oracle(
|
|
&oracle,
|
|
&oracle.ready.admission_key,
|
|
&oracle.ready.client_key,
|
|
|_| {},
|
|
|_| {},
|
|
);
|
|
assert!(result.is_err(), "{mode} mismatch unexpectedly connected");
|
|
}
|
|
}
|
|
|
|
#[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("");
|
|
let admission_key = test_key(&oracle.ready.admission_key);
|
|
let tls_key = test_key(&oracle.ready.client_key);
|
|
let manifest = ConnectionManifest::decode(oracle.ready.manifest.as_bytes()).expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let admission = AdmissionSigner::new(move |message| {
|
|
if !message.starts_with(b"versevdi/tunnel-admission/v1") {
|
|
return Err(CoreError::AuthorityRejected);
|
|
}
|
|
signer(Arc::clone(&admission_key))(message)
|
|
});
|
|
let tls = TlsEd25519Signer::new(move |message| {
|
|
if message.starts_with(b"versevdi/tunnel-admission/v1") {
|
|
return Err(CoreError::Tls);
|
|
}
|
|
signer(Arc::clone(&tls_key))(message)
|
|
});
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
let session = runtime
|
|
.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
Signers::new(admission, tls),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(5),
|
|
))
|
|
.expect("purpose-separated signers");
|
|
runtime.block_on(session.close());
|
|
}
|
|
|
|
#[test]
|
|
fn swapping_admission_and_tls_signer_purposes_is_rejected() {
|
|
let oracle = Oracle::start("");
|
|
let result = connect_oracle(
|
|
&oracle,
|
|
&oracle.ready.client_key,
|
|
&oracle.ready.admission_key,
|
|
|_| {},
|
|
|_| {},
|
|
);
|
|
assert!(
|
|
result.is_err(),
|
|
"one interchangeable key was accepted for both signer purposes"
|
|
);
|
|
}
|
|
|
|
fn recording_signers(
|
|
admission_key: Arc<dyn SigningKey>,
|
|
tls_key: Arc<dyn SigningKey>,
|
|
transcripts: Arc<Mutex<Vec<Vec<u8>>>>,
|
|
) -> Signers {
|
|
Signers::new(
|
|
AdmissionSigner::new(move |message| {
|
|
transcripts
|
|
.lock()
|
|
.expect("admission transcripts")
|
|
.push(message.to_vec());
|
|
signer(Arc::clone(&admission_key))(message)
|
|
}),
|
|
TlsEd25519Signer::new(signer(tls_key)),
|
|
)
|
|
}
|
|
|
|
#[test]
|
|
fn fresh_nonce_changes_transcript_and_oracle_rejects_identical_replay() {
|
|
let transcripts = Arc::new(Mutex::new(Vec::new()));
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
for _ in 0..2 {
|
|
let oracle = Oracle::start("");
|
|
let manifest =
|
|
ConnectionManifest::decode(oracle.ready.manifest.as_bytes()).expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let session = runtime
|
|
.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
recording_signers(
|
|
test_key(&oracle.ready.admission_key),
|
|
test_key(&oracle.ready.client_key),
|
|
Arc::clone(&transcripts),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(5),
|
|
))
|
|
.expect("fresh nonce admission");
|
|
runtime.block_on(session.close());
|
|
assert!(
|
|
oracle.ready.replay_rejected,
|
|
"oracle accepted identical replay"
|
|
);
|
|
}
|
|
let transcripts = transcripts.lock().expect("transcripts");
|
|
assert_eq!(transcripts.len(), 2);
|
|
assert_ne!(transcripts[0], transcripts[1], "nonce was reused");
|
|
}
|
|
|
|
#[test]
|
|
fn overall_deadline_cancels_an_unresponsive_manifest_address() {
|
|
let oracle = Oracle::start("");
|
|
let socket = std::net::UdpSocket::bind("127.0.0.1:0").expect("bind blackhole");
|
|
let mut manifest: serde_json::Value =
|
|
serde_json::from_str(&oracle.ready.manifest).expect("manifest JSON");
|
|
manifest["gateway"]["addresses"] =
|
|
serde_json::json!([socket.local_addr().expect("blackhole address").to_string()]);
|
|
let manifest =
|
|
ConnectionManifest::decode(&serde_json::to_vec(&manifest).expect("encode manifest"))
|
|
.expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
let started = Instant::now();
|
|
let result = runtime.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
recording_signers(
|
|
test_key(&oracle.ready.admission_key),
|
|
test_key(&oracle.ready.client_key),
|
|
Arc::new(Mutex::new(Vec::new())),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_millis(150),
|
|
));
|
|
assert_eq!(result.err(), Some(CoreError::Cancelled));
|
|
assert!(started.elapsed() < Duration::from_secs(2));
|
|
}
|
|
|
|
#[test]
|
|
fn overall_deadline_includes_synchronous_admission_signing() {
|
|
let oracle = Oracle::start("");
|
|
let manifest = ConnectionManifest::decode(oracle.ready.manifest.as_bytes()).expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let admission_key = test_key(&oracle.ready.admission_key);
|
|
let tls_key = test_key(&oracle.ready.client_key);
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
let started = Instant::now();
|
|
let result = runtime.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
Signers::new(
|
|
AdmissionSigner::new(move |message| {
|
|
thread::sleep(Duration::from_millis(100));
|
|
signer(Arc::clone(&admission_key))(message)
|
|
}),
|
|
TlsEd25519Signer::new(signer(tls_key)),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_millis(25),
|
|
));
|
|
assert_eq!(result.err(), Some(CoreError::Cancelled));
|
|
assert!(started.elapsed() < Duration::from_secs(2));
|
|
}
|
|
|
|
#[test]
|
|
fn tls_signer_finite_deadline_overrun_is_cancelled_not_tls() {
|
|
let oracle = Oracle::start("");
|
|
let manifest = ConnectionManifest::decode(oracle.ready.manifest.as_bytes()).expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let admission_key = test_key(&oracle.ready.admission_key);
|
|
let tls_key = test_key(&oracle.ready.client_key);
|
|
let cancellation = Cancellation::new();
|
|
let trigger = cancellation.clone();
|
|
let canceller = thread::spawn(move || {
|
|
thread::sleep(Duration::from_millis(25));
|
|
trigger.cancel();
|
|
});
|
|
let result =
|
|
tokio::runtime::Runtime::new()
|
|
.expect("runtime")
|
|
.block_on(connect_with_cancellation(
|
|
&manifest,
|
|
&credential,
|
|
Signers::new(
|
|
AdmissionSigner::new(signer(admission_key)),
|
|
TlsEd25519Signer::new(move |message| {
|
|
thread::sleep(Duration::from_millis(100));
|
|
signer(Arc::clone(&tls_key))(message)
|
|
}),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(5),
|
|
&cancellation,
|
|
));
|
|
canceller.join().expect("canceller");
|
|
assert_eq!(result.err(), Some(CoreError::Cancelled));
|
|
}
|
|
|
|
#[test]
|
|
fn explicit_cancellation_interrupts_network_wait() {
|
|
let oracle = Oracle::start("");
|
|
let socket = std::net::UdpSocket::bind("127.0.0.1:0").expect("bind blackhole");
|
|
let mut manifest: serde_json::Value =
|
|
serde_json::from_str(&oracle.ready.manifest).expect("manifest JSON");
|
|
manifest["gateway"]["addresses"] =
|
|
serde_json::json!([socket.local_addr().expect("blackhole address").to_string()]);
|
|
let manifest =
|
|
ConnectionManifest::decode(&serde_json::to_vec(&manifest).expect("encode manifest"))
|
|
.expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let cancellation = Cancellation::new();
|
|
let trigger = cancellation.clone();
|
|
let canceller = thread::spawn(move || {
|
|
thread::sleep(Duration::from_millis(25));
|
|
trigger.cancel();
|
|
});
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
let started = Instant::now();
|
|
let result = runtime.block_on(connect_with_cancellation(
|
|
&manifest,
|
|
&credential,
|
|
recording_signers(
|
|
test_key(&oracle.ready.admission_key),
|
|
test_key(&oracle.ready.client_key),
|
|
Arc::new(Mutex::new(Vec::new())),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(5),
|
|
&cancellation,
|
|
));
|
|
canceller.join().expect("canceller");
|
|
assert_eq!(result.err(), Some(CoreError::Cancelled));
|
|
assert!(started.elapsed() < Duration::from_secs(2));
|
|
}
|
|
|
|
#[test]
|
|
fn blackholed_first_manifest_address_does_not_starve_live_second_address() {
|
|
let oracle = Oracle::start("");
|
|
let socket = std::net::UdpSocket::bind("127.0.0.1:0").expect("bind blackhole");
|
|
let mut manifest: serde_json::Value =
|
|
serde_json::from_str(&oracle.ready.manifest).expect("manifest JSON");
|
|
let live = manifest["gateway"]["addresses"][0].clone();
|
|
manifest["gateway"]["addresses"] = serde_json::json!([
|
|
socket.local_addr().expect("blackhole address").to_string(),
|
|
live,
|
|
]);
|
|
let manifest =
|
|
ConnectionManifest::decode(&serde_json::to_vec(&manifest).expect("encode manifest"))
|
|
.expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
let session = runtime
|
|
.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
recording_signers(
|
|
test_key(&oracle.ready.admission_key),
|
|
test_key(&oracle.ready.client_key),
|
|
Arc::new(Mutex::new(Vec::new())),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(5),
|
|
))
|
|
.expect("live second manifest address");
|
|
runtime.block_on(session.close());
|
|
}
|
|
|
|
#[test]
|
|
fn multi_record_first_manifest_address_does_not_starve_live_second_address() {
|
|
let oracle = Oracle::start("");
|
|
let socket = std::net::UdpSocket::bind("127.0.0.1:0").expect("bind blackhole");
|
|
let port = socket.local_addr().expect("blackhole address").port();
|
|
let mut manifest: serde_json::Value =
|
|
serde_json::from_str(&oracle.ready.manifest).expect("manifest JSON");
|
|
let live = manifest["gateway"]["addresses"][0].clone();
|
|
manifest["gateway"]["addresses"] = serde_json::json!([format!("localhost:{port}"), live]);
|
|
let manifest =
|
|
ConnectionManifest::decode(&serde_json::to_vec(&manifest).expect("encode manifest"))
|
|
.expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
let session = runtime
|
|
.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
recording_signers(
|
|
test_key(&oracle.ready.admission_key),
|
|
test_key(&oracle.ready.client_key),
|
|
Arc::new(Mutex::new(Vec::new())),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(5),
|
|
))
|
|
.expect("live second address after multi-record first address");
|
|
runtime.block_on(session.close());
|
|
}
|
|
|
|
#[test]
|
|
fn invalid_first_manifest_address_does_not_block_live_second_address() {
|
|
let oracle = Oracle::start("");
|
|
let mut manifest: serde_json::Value =
|
|
serde_json::from_str(&oracle.ready.manifest).expect("manifest JSON");
|
|
let live = manifest["gateway"]["addresses"][0].clone();
|
|
manifest["gateway"]["addresses"] = serde_json::json!(["127.0.0.1:0", live]);
|
|
let manifest =
|
|
ConnectionManifest::decode(&serde_json::to_vec(&manifest).expect("encode manifest"))
|
|
.expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
let session = runtime
|
|
.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
recording_signers(
|
|
test_key(&oracle.ready.admission_key),
|
|
test_key(&oracle.ready.client_key),
|
|
Arc::new(Mutex::new(Vec::new())),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(5),
|
|
))
|
|
.expect("live address after resolution failure");
|
|
runtime.block_on(session.close());
|
|
}
|
|
|
|
#[test]
|
|
fn retryable_stable_error_advances_to_live_manifest_address() {
|
|
let oracle = Oracle::start("retryable");
|
|
let manifest = ConnectionManifest::decode(oracle.ready.manifest.as_bytes()).expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let transcripts = Arc::new(Mutex::new(Vec::new()));
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
let session = runtime
|
|
.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
recording_signers(
|
|
test_key(&oracle.ready.admission_key),
|
|
test_key(&oracle.ready.client_key),
|
|
Arc::clone(&transcripts),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(5),
|
|
))
|
|
.expect("retryable draining response advanced to live address");
|
|
runtime.block_on(session.close());
|
|
let transcripts = transcripts.lock().expect("transcripts");
|
|
assert_eq!(transcripts.len(), 2);
|
|
assert_ne!(transcripts[0], transcripts[1]);
|
|
}
|
|
|
|
#[test]
|
|
fn post_admission_failure_is_terminal_and_does_not_mutate_twice() {
|
|
for mode in [
|
|
"post-retryable",
|
|
"provider-start-lost-response",
|
|
"cleanup-release-failure",
|
|
] {
|
|
let oracle = Oracle::start(mode);
|
|
let manifest =
|
|
ConnectionManifest::decode(oracle.ready.manifest.as_bytes()).expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let transcripts = Arc::new(Mutex::new(Vec::new()));
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
let result = runtime.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
recording_signers(
|
|
test_key(&oracle.ready.admission_key),
|
|
test_key(&oracle.ready.client_key),
|
|
Arc::clone(&transcripts),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(5),
|
|
));
|
|
assert!(result.is_err(), "{mode} was retried to success");
|
|
let transcripts = transcripts.lock().expect("transcripts");
|
|
assert_eq!(
|
|
transcripts.len(),
|
|
1,
|
|
"{mode} attempted admission mutation twice"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn lost_authority_after_provider_mutation_is_terminal_without_failover() {
|
|
let oracle = Oracle::start("lost-authority");
|
|
let manifest = ConnectionManifest::decode(oracle.ready.manifest.as_bytes()).expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let transcripts = Arc::new(Mutex::new(Vec::new()));
|
|
let result = tokio::runtime::Runtime::new()
|
|
.expect("runtime")
|
|
.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
recording_signers(
|
|
test_key(&oracle.ready.admission_key),
|
|
test_key(&oracle.ready.client_key),
|
|
Arc::clone(&transcripts),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_millis(400),
|
|
));
|
|
|
|
assert!(
|
|
result.is_err(),
|
|
"late first authority triggered live-address success"
|
|
);
|
|
assert_eq!(
|
|
transcripts.lock().expect("transcripts").len(),
|
|
1,
|
|
"client sent a second signed admission"
|
|
);
|
|
assert_eq!(oracle.provider_starts(), 1, "provider mutated twice");
|
|
}
|
|
|
|
#[test]
|
|
fn repeated_production_gateway_admissions_remain_bounded() {
|
|
let runtime = tokio::runtime::Runtime::new().expect("runtime");
|
|
for index in 0..10 {
|
|
let oracle = Oracle::start("reusable");
|
|
let manifest =
|
|
ConnectionManifest::decode(oracle.ready.manifest.as_bytes()).expect("manifest");
|
|
let credential =
|
|
NativeTunnelCredential::decode(oracle.ready.credential.as_bytes()).expect("credential");
|
|
let session = runtime
|
|
.block_on(connect(
|
|
&manifest,
|
|
&credential,
|
|
recording_signers(
|
|
test_key(&oracle.ready.admission_key),
|
|
test_key(&oracle.ready.client_key),
|
|
Arc::new(Mutex::new(Vec::new())),
|
|
),
|
|
"2026-08-12T00:00:00Z",
|
|
Duration::from_secs(5),
|
|
))
|
|
.unwrap_or_else(|error| panic!("stress admission {index}: {error:?}"));
|
|
runtime.block_on(session.close());
|
|
}
|
|
}
|