code wiki / _hdl_build / nx_signaling_v2_gate.nx
`nx_signaling_v2_gate.nx` source

↩ module page · 634 lines · 35023 B
1import "nx_gate_base.nx"
2import "nx_connect.nx"   // bounded connect: a raw sys_connect hangs ~127s on a black-holed host
3// nx_signaling_v2_gate.nx -- ENGINEER gate for the N-party signaling daemon.
4// Self-contained + re-runnable: forks /tmp/nx_signaling_v2.sov.elf, connects
5// THREE WebSocket clients to the same room over loopback, and proves:
6//   (1) all 3 complete the RFC 6455 client upgrade (101)
7//   (2) a data frame from A is relayed to BOTH B and C (broadcast)
8//   (3) ...and NOT echoed back to A
9//   (4) a frame from B reaches A and C (any-peer broadcast, not 1:1 v1)
10//   (5) ping from A -> pong back to A (control handled per-peer, not relayed)
11//   (6) A closes -> B<->C relay still works (leave doesn't kill the room)
12//   (7) a 4th..9th peer joins fine, the 9th is rejected w/ close (cap=8)
13// Composes nx_websocket_client_upgrade + nx_websocket_stream (KAT'd).
14// license_tier: ORIGINAL
15
16import "nx_syscalls.nx"
17import "nx_gate_emit_lib.nx"
18import "nx_room_key.nx"      // token computation for the keyed-room tests (same SSOT organ as the relay)
19import "nx_dgram_media.nx"   // R2: UDP-plane client side (pack/parse/reassemble) for the mixed-plane tests
20import "nx_websocket_client_upgrade.nx"
21import "nx_websocket_stream.nx"
22import "nx_websocket_frame.nx"
23
24func grow(name: *u8, ok: i64) -> i64 { if ok==1 { gw("  PASS " as *u8) } else { gw("  FAIL " as *u8) } gw(name); gw("
25" as *u8); return ok }
26func g_loopback_addr(addr: *u8, port: i64) -> i64 {
  addr[0]=2 as u8; addr[1]=0 as u8
  addr[2]=((port>>8)&0xff) as u8; addr[3]=(port&0xff) as u8
  addr[4]=127 as u8; addr[5]=0 as u8; addr[6]=0 as u8; addr[7]=1 as u8
  var i: i64=8
  while i<16 { addr[i]=0 as u8; i=i+1 }
  return 16
33}
34
35// connect + WS-upgrade one client into /signal/gateroom; returns fd or -1
36func g_join(req_buf: *u8, resp_buf: *u8) -> i64 {
  let fd: i64 = sys_socket(AF_INET, SOCK_STREAM, 0)
  if fd < 0 { return 0 - 1 }
  let addr: *u8 = sys_mmap(16)
  g_loopback_addr(addr, 8445)
  if nx_connect_bounded(fd, addr, 16, NX_CONN_DEFAULT_MS) < 0 { sys_close(fd); return 0 - 1 }
  let host: *u8 = "127.0.0.1" as *u8
  let path: *u8 = "/signal/gateroom" as *u8
  let v: i64 = nx_ws_client_upgrade(fd, host, 9, path, 16, 8445, 1,
                                    req_buf, 2048, resp_buf, 4096)
  if v != NX_WSCU_OK {
      g_puts("    join: client upgrade verdict=" as *u8); g_pn(v); g_puts("\n" as *u8)
      sys_close(fd)
      return 0 - 1
  }
  return fd
52}
53
54// read one data frame (skipping nothing) with the streaming reader; payload
55// copied into out; returns payload len or -1.
56func g_read_payload(fd: i64, buf: *u8, cap: i64, out: *u8, outcap: i64) -> i64 {
  let f_raw: *u8 = sys_mmap(128)
  let f: *WsFrame = f_raw as *WsFrame
  let v: i64 = nx_ws_read_frame_from_fd(fd, buf, cap, f)
  if v != NX_WSS_OK { return 0 - 1 }
  if f.payload_len > outcap { return 0 - 1 }
  var i: i64 = 0
  while i < f.payload_len { out[i] = buf[f.payload_off + i]; i = i + 1 }
  return f.payload_len
65}
66
67func g_eq(a: *u8, b: *u8, n: i64) -> i64 {
  var i: i64=0
  while i<n { if a[i]!=b[i] { return 0 } i=i+1 }
  return 1
71}
72func g_wr(path: *u8, data: *u8, n: i64) -> i64 {
  let fd: i64 = sys_openat_wr(path, 0x1a4)   // 0644
  if fd < 0 { return 0 - 1 }
  sys_write(fd, data, n)
  sys_close(fd)
  return 0 }
78// "/signal/keyroom?k=<tok16>"
79func g_mkkeypath(buf: *u8, tok: *u8) -> i64 {
  let pre: *u8 = "/signal/keyroom?k=" as *u8
  var i: i64 = 0
  while pre[i] != (0 as u8) { buf[i] = pre[i]; i = i + 1 }
  var j: i64 = 0
  while j < 16 { buf[i + j] = tok[j]; j = j + 1 }
  return i + 16 }
86
87
88// ---- SFU-test helpers (task 41) ----
89func g_join_room(path: *u8, plen: i64, req_buf: *u8, resp_buf: *u8) -> i64 {
  let fd: i64 = sys_socket(AF_INET, SOCK_STREAM, 0)
  if fd < 0 { return 0 - 1 }
  let addr: *u8 = sys_mmap(16)
  g_loopback_addr(addr, 8445)
  if nx_connect_bounded(fd, addr, 16, NX_CONN_DEFAULT_MS) < 0 { sys_close(fd); return 0 - 1 }
  let v: i64 = nx_ws_client_upgrade(fd, "127.0.0.1" as *u8, 9, path, plen, 8445, 1,
                                    req_buf, 2048, resp_buf, 4096)
  if v != NX_WSCU_OK { sys_close(fd); return 0 - 1 }
  return fd }
99func g_tmo(fd: i64, sec: i64) -> i64 {
  let tv: *i64 = sys_mmap(16) as *i64; tv[0]=sec; tv[1]=0
  sys_setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, tv as *u8, 16); return 0 }
102// 16B registration frame: [kind 0x43][id8][seq4=0][pad3] -- any binary frame teaches the relay this conn id
103func g_mkreg(b: *u8, id: *u8) -> i64 {
  b[0] = 0x43 as u8
  var i: i64 = 0
  while i < 8 { b[1+i] = id[i]; i = i + 1 }
  i = 9
  while i < 16 { b[i] = 0 as u8; i = i + 1 }
  return 0 }
110// 32B video frame: [0x57][id8][seq4 LE][flags][qp][w2][h2][13B junk] (flags bit0=key bit1=LO)
111func g_mk57(b: *u8, id: *u8, seq: i64, flags: i64) -> i64 {
  b[0] = 0x57 as u8
  var i: i64 = 0
  while i < 8 { b[1+i] = id[i]; i = i + 1 }
  b[9]  = (seq & 255) as u8
  b[10] = ((seq / 256) & 255) as u8
  b[11] = ((seq / 65536) & 255) as u8
  b[12] = ((seq / 16777216) & 255) as u8
  b[13] = flags as u8
  b[14] = 20 as u8
  b[15] = 64 as u8; b[16] = 1 as u8      // w=320
  b[17] = 240 as u8; b[18] = 0 as u8     // h=240
  i = 19
  while i < 32 { b[i] = ((i * 3) & 255) as u8; i = i + 1 }
  return 0 }
126func g_rdseq(b: *u8) -> i64 {
  return (b[9]&0xff) + ((b[10]&0xff)*256) + ((b[11]&0xff)*65536) + ((b[12]&0xff)*16777216) }
128
129func main() -> i64 {
  g_puts("nx_signaling_v2 gate (3-peer mesh over loopback)\n" as *u8)
  var pass: i64 = 0
  var total: i64 = 0
133
  // ---- spawn the daemon: sovereign build first, fast-runner build as
  // the iteration fallback (final re-gate runs against the .sov.elf) ----
  var dpath: *u8 = "/tmp/nx_signaling_v2.sov.elf" as *u8
  let probe: i64 = sys_openat_rd(dpath)
  if probe >= 0 { sys_close(probe) }
  else { dpath = "/tmp/nx_signaling_v2.elf" as *u8 }
  g_puts("daemon: " as *u8); g_puts(dpath); g_puts("\n" as *u8)
  // token-wall test config: secret + one protected room "keyroom". Written BEFORE the daemon reads them
  // at startup. All OTHER test rooms are unlisted -> unaffected (proves the wall is opt-in).
  g_wr("/tmp/rk_secret_gate.txt" as *u8, "gatesecret" as *u8, 10)
  g_wr("/tmp/rk_protected_gate.txt" as *u8, "keyroom\n" as *u8, 8)
  let pid: i64 = sys_fork()
  if pid == 0 {
      let av: *i64 = sys_mmap(64) as *i64
      av[0] = dpath as i64
      av[1] = "/tmp/rk_secret_gate.txt" as *u8 as i64      // token-wall test config (runner pre-writes;
      av[2] = "/tmp/rk_protected_gate.txt" as *u8 as i64   //  missing files = wall off = old behavior)
      av[3] = 0
      let ev: *i64 = sys_mmap(16) as *i64
      ev[0] = 0
      sys_execve(dpath, av, ev)
      sys_exit(127)
  }
  sys_sleep_ms(300)
158
  let req_buf: *u8 = sys_mmap(2048)
  let resp_buf: *u8 = sys_mmap(4096)
  let frame_buf: *u8 = sys_mmap(16384)
  let pay: *u8 = sys_mmap(4096)
163
  // (1) three peers join
  let fa: i64 = g_join(req_buf, resp_buf)
  let fb: i64 = g_join(req_buf, resp_buf)
  let fc: i64 = g_join(req_buf, resp_buf)
  pass = pass + g_check("A+B+C upgraded (101 x3)" as *u8, (fa >= 0) + (fb >= 0) + (fc >= 0) == 3); total = total + 1
  if fa < 0 { return 1 }
  if fb < 0 { return 1 }
  if fc < 0 { return 1 }
172
  // (2) A broadcasts -> B and C both receive
  let msg_a: *u8 = "{\"from\":\"A\",\"type\":\"hello\"}" as *u8
  let msg_a_n: i64 = g_slen(msg_a)
  nx_ws_send_frame_to_fd(fa, 1, WS_OP_TEXT, msg_a, msg_a_n)
  let nb: i64 = g_read_payload(fb, frame_buf, 16384, pay, 4096)
  pass = pass + g_check("B received A's hello" as *u8, (nb == msg_a_n) * g_eq(pay, msg_a, msg_a_n)); total = total + 1
  let nc: i64 = g_read_payload(fc, frame_buf, 16384, pay, 4096)
  pass = pass + g_check("C received A's hello (N-party broadcast)" as *u8, (nc == msg_a_n) * g_eq(pay, msg_a, msg_a_n)); total = total + 1
181
  // (4) B broadcasts -> A and C receive
  let msg_b: *u8 = "{\"from\":\"B\",\"type\":\"offer\",\"to\":\"A\"}" as *u8
  let msg_b_n: i64 = g_slen(msg_b)
  nx_ws_send_frame_to_fd(fb, 1, WS_OP_TEXT, msg_b, msg_b_n)
  let na2: i64 = g_read_payload(fa, frame_buf, 16384, pay, 4096)
  pass = pass + g_check("A received B's offer (verbatim envelope)" as *u8, (na2 == msg_b_n) * g_eq(pay, msg_b, msg_b_n)); total = total + 1
  let nc2: i64 = g_read_payload(fc, frame_buf, 16384, pay, 4096)
  pass = pass + g_check("C received B's offer too (broadcast relay)" as *u8, nc2 == msg_b_n); total = total + 1
190
  // (5) ping from A -> pong to A
  nx_ws_send_frame_to_fd(fa, 1, WS_OP_PING, "pp" as *u8, 2)
  let f_raw: *u8 = sys_mmap(128)
  let f: *WsFrame = f_raw as *WsFrame
  let vp: i64 = nx_ws_read_frame_from_fd(fa, frame_buf, 16384, f)
  pass = pass + g_check("A ping -> pong back to A" as *u8, (vp == NX_WSS_OK) * (f.opcode == WS_OP_PONG)); total = total + 1
197
  // (5b) BINARY media frame (the pure-Nishi video plane): 12KB JPEG-sized
  // payload from B relays to C bit-exact with opcode preserved.
  let big: *u8 = sys_mmap(12288)
  var bi: i64 = 0
  while bi < 12288 { big[bi] = ((bi * 31 + 7) & 0xff) as u8; bi = bi + 1 }
  nx_ws_send_frame_to_fd(fb, 1, WS_OP_BINARY, big, 12288)
  let fbin_raw: *u8 = sys_mmap(128)
  let fbin: *WsFrame = fbin_raw as *WsFrame
  let vbin: i64 = nx_ws_read_frame_from_fd(fc, frame_buf, 16384, fbin)
  var bin_ok: i64 = 0
  if vbin == NX_WSS_OK { if fbin.opcode == WS_OP_BINARY { if fbin.payload_len == 12288 {
      bin_ok = 1
      var bj: i64 = 0
      while bj < 12288 {
          if frame_buf[fbin.payload_off + bj] != big[bj] { bin_ok = 0; bj = 12288 } else { bj = bj + 1 }
      }
  } } }
  pass = pass + g_check("12KB BINARY media frame relays bit-exact" as *u8, bin_ok == 1); total = total + 1
  // drain A's copy of the binary frame so later reads on A stay aligned
  nx_ws_read_frame_from_fd(fa, frame_buf, 16384, fbin)
218
  // (6) A leaves; B <-> C still relay
  nx_ws_send_close(fa, 1000)
  sys_close(fa)
  sys_sleep_ms(100)
  let msg_c: *u8 = "{\"from\":\"C\",\"type\":\"bye-test\"}" as *u8
  let msg_c_n: i64 = g_slen(msg_c)
  nx_ws_send_frame_to_fd(fc, 1, WS_OP_TEXT, msg_c, msg_c_n)
  let nb3: i64 = g_read_payload(fb, frame_buf, 16384, pay, 4096)
  pass = pass + g_check("after A left, C->B relay still live" as *u8, nb3 == msg_c_n); total = total + 1
228
  // (7) fill to the 8-peer cap.  After A left the room holds B,C (2);
  // f4..f9 are six more = 8 total (at cap); f10 is the true 9th.
  let f4: i64 = g_join(req_buf, resp_buf)
  let f5: i64 = g_join(req_buf, resp_buf)
  let f6: i64 = g_join(req_buf, resp_buf)
  let f7: i64 = g_join(req_buf, resp_buf)
  let f8: i64 = g_join(req_buf, resp_buf)
  let f9: i64 = g_join(req_buf, resp_buf)
  pass = pass + g_check("peers 4..9 admitted (room grows to cap=8)" as *u8,
      (f4>=0) + (f5>=0) + (f6>=0) + (f7>=0) + (f8>=0) + (f9>=0) == 6); total = total + 1
  // the 9th joiner: upgrade succeeds at HTTP level then the daemon
  // closes w/ 1008 -- the next read on it sees CLOSE/EOF.
  let f10: i64 = g_join(req_buf, resp_buf)
  var ninth_rejected: i64 = 0
  if f10 < 0 { ninth_rejected = 1 }
  else {
      let v10: i64 = nx_ws_read_frame_from_fd(f10, frame_buf, 16384, f)
      if v10 != NX_WSS_OK { ninth_rejected = 1 }
      else { if f.opcode == WS_OP_CLOSE { ninth_rejected = 1 } }
  }
  pass = pass + g_check("9th peer rejected (cap=8 enforced)" as *u8, ninth_rejected == 1); total = total + 1
250
  // ---- R2 (task #15): the SOVEREIGN UDP PLANE + the MIXED-PLANE BRIDGE ----
  // Native clients speak nx_dgram_media datagrams to :8471; the relay raw-fans to UDP peers AND bridges
  // completed frames to WSS members (and vice versa). Isolated in "udproom" (fresh WS client, no backlog).
  let us_a: i64 = sys_socket(AF_INET, SOCK_DGRAM, 0)
  let us_b: i64 = sys_socket(AF_INET, SOCK_DGRAM, 0)
  let tv: *u8 = sys_mmap(16)                       // SO_RCVTIMEO 2s so a missing reply FAILS, never hangs
  let tvp: *i64 = tv as *i64
  tvp[0] = 2; tvp[1] = 0
  sys_setsockopt(us_a, SOL_SOCKET, 20, tv, 16)
  sys_setsockopt(us_b, SOL_SOCKET, 20, tv, 16)
  let uad: *u8 = sys_mmap(16)
  g_loopback_addr(uad, 8471)
  let upk: *u8 = sys_mmap(131072)
  let urx_a: *u8 = sys_mmap(DGM_RX_REGION)
  dgm_rx_init(urx_a)
  let urx_b: *u8 = sys_mmap(DGM_RX_REGION)
  dgm_rx_init(urx_b)
  let uinfo: *i64 = sys_mmap(8 * 12) as *i64
  let ufr: *u8 = sys_mmap(65536)
  let ubuf: *u8 = sys_mmap(2048)
  let ufrom2: *u8 = sys_mmap(16)
  let uflen: *i64 = sys_mmap(16) as *i64
  // (11) A joins over UDP -> ACK datagram comes back
  var np: i64 = dgm_pack_frame(0x4A, "AAAAAAA1" as *u8, 0, "udproom" as *u8, 7, upk, 131072)
  var pl: i64 = (upk[0] & 0xff) + ((upk[1] & 0xff) << 8)
  sys_sendto(us_a, ((upk as i64) + 2) as *u8, pl, 0, uad, 16)
  uflen[0] = 16
  var arn: i64 = sys_recvfrom(us_a, ubuf, 2048, 0, ufrom2, uflen)
  var ackok: i64 = 0
  if arn > 0 { if dgm_parse_pkt(ubuf, arn, uinfo) == 0 { if uinfo[0] == 0x4B { ackok = 1 } } }
  pass = pass + g_check("UDP JOIN -> ACK (sovereign dgram plane up on :8471)" as *u8, ackok == 1); total = total + 1
  // (12) B joins; A sends a fragmented 3KB room-frame -> raw passthrough -> B reassembles BIT-EXACT
  np = dgm_pack_frame(0x4A, "BBBBBBB2" as *u8, 0, "udproom" as *u8, 7, upk, 131072)
  pl = (upk[0] & 0xff) + ((upk[1] & 0xff) << 8)
  sys_sendto(us_b, ((upk as i64) + 2) as *u8, pl, 0, uad, 16)
  uflen[0] = 16
  sys_recvfrom(us_b, ubuf, 2048, 0, ufrom2, uflen)   // B's ACK
  let body3k: *u8 = sys_mmap(3000)
  var bi3: i64 = 0
  while bi3 < 3000 { body3k[bi3] = (((bi3 * 37) + 11) & 255) as u8; bi3 = bi3 + 1 }
  np = dgm_pack_frame(0x4D, "AAAAAAA1" as *u8, 1, body3k, 3000, upk, 131072)
  var o3: i64 = 0
  var s3: i64 = 0
  while s3 < np {
      pl = (upk[o3] & 0xff) + ((upk[o3+1] & 0xff) << 8)
      sys_sendto(us_a, ((upk as i64) + o3 + 2) as *u8, pl, 0, uad, 16)
      o3 = o3 + 2 + pl
      s3 = s3 + 1
  }
  var got3: i64 = 0
  var tries: i64 = 0
  while tries < 12 {
      if got3 == 0 {
          uflen[0] = 16
          let rn3: i64 = sys_recvfrom(us_b, ubuf, 2048, 0, ufrom2, uflen)
          if rn3 > 0 { let rr3: i64 = dgm_rx_add(urx_b, ubuf, rn3, ufr, 65536, uinfo)
              if rr3 == 3000 { if g_eq(ufr, body3k, 3000) == 1 { got3 = 1 } } }
      }
      tries = tries + 1
  }
  pass = pass + g_check("UDP->UDP: 3KB frame raw-fans + reassembles bit-exact" as *u8, got3 == 1); total = total + 1
  // (13) MIXED PLANE: a fresh WSS client in the SAME room; UDP frame -> arrives as ONE WS binary frame;
  //      WS binary -> arrives at A as datagrams that reassemble bit-exact. The bridge, both directions.
  let fdw: i64 = sys_socket(AF_INET, SOCK_STREAM, 0)
  var fu: i64 = 0 - 1
  if fdw >= 0 {
      let wad: *u8 = sys_mmap(16)
      g_loopback_addr(wad, 8445)
      if nx_connect_bounded(fdw, wad, 16, NX_CONN_DEFAULT_MS) >= 0 {
          let v13: i64 = nx_ws_client_upgrade(fdw, "127.0.0.1" as *u8, 9, "/signal/udproom" as *u8, 15, 8445, 1,
                                              req_buf, 2048, resp_buf, 4096)
          if v13 == NX_WSCU_OK { fu = fdw }
      }
  }
  var b13a: i64 = 0
  if fu >= 0 {
      np = dgm_pack_frame(0x4D, "AAAAAAA1" as *u8, 2, body3k, 3000, upk, 131072)
      o3 = 0
      s3 = 0
      while s3 < np {
          pl = (upk[o3] & 0xff) + ((upk[o3+1] & 0xff) << 8)
          sys_sendto(us_a, ((upk as i64) + o3 + 2) as *u8, pl, 0, uad, 16)
          o3 = o3 + 2 + pl
          s3 = s3 + 1
      }
      let wl: i64 = g_read_payload(fu, frame_buf, 16384, ufr, 65536)
      if wl == 3000 { if g_eq(ufr, body3k, 3000) == 1 { b13a = 1 } }
  }
  pass = pass + g_check("BRIDGE UDP->WSS: dgram frame lands as ONE WS frame, bit-exact" as *u8, b13a == 1); total = total + 1
  var b13b: i64 = 0
  if fu >= 0 {
      var bi4: i64 = 0
      while bi4 < 3000 { body3k[bi4] = (((bi4 * 53) + 7) & 255) as u8; bi4 = bi4 + 1 }
      nx_ws_send_frame_to_fd(fu, 1, WS_OP_BINARY, body3k, 3000)
      var tries2: i64 = 0
      while tries2 < 12 {
          if b13b == 0 {
              uflen[0] = 16
              let rn4: i64 = sys_recvfrom(us_a, ubuf, 2048, 0, ufrom2, uflen)
              if rn4 > 0 { let rr4: i64 = dgm_rx_add(urx_a, ubuf, rn4, ufr, 65536, uinfo)
                  if rr4 == 3000 { if uinfo[0] == 0x4D { if g_eq(ufr, body3k, 3000) == 1 { b13b = 1 } } } }
          }
          tries2 = tries2 + 1
      }
  }
  pass = pass + g_check("BRIDGE WSS->UDP: WS frame fragments to dgrams, reassembles bit-exact" as *u8, b13b == 1); total = total + 1
357
  // ================= SFU (task #41): selective forwarding, END-TO-END through the real relay =========
  // Fresh room "sfuroom": S(ender) + R1 + R2. R1 stays default (HI); R2 LSUBs to LO. S emits interleaved
  // HI/LO-marked 0x57 frames. Assert: R1 gets ONLY HI with rewritten contiguous seq 0,1,2...; R2 gets HI
  // until the LO KEY then ONLY LO, seq contiguous ACROSS the switch; LSUB itself is never broadcast.
  let fs2: i64 = g_join_room("/signal/sfuroom" as *u8, 15, req_buf, resp_buf)
  let fr1: i64 = g_join_room("/signal/sfuroom" as *u8, 15, req_buf, resp_buf)
  let fr2: i64 = g_join_room("/signal/sfuroom" as *u8, 15, req_buf, resp_buf)
  var sfu_ok: i64 = 0
  if fs2 >= 0 { if fr1 >= 0 { if fr2 >= 0 { sfu_ok = 1 } } }
  pass = pass + g_check("SFU: S+R1+R2 joined sfuroom" as *u8, sfu_ok == 1); total = total + 1
  if sfu_ok == 1 {
      g_tmo(fs2, 2); g_tmo(fr1, 2); g_tmo(fr2, 2)
      // registration: every member must have sent ONE binary frame (relay learns id -> memidx)
      let regf: *u8 = sys_mmap(64)
      g_mkreg(regf, "RCVR0001" as *u8)
      nx_ws_send_frame_to_fd(fr1, 1, WS_OP_BINARY, regf, 16)
      g_mkreg(regf, "RCVR0002" as *u8)
      nx_ws_send_frame_to_fd(fr2, 1, WS_OP_BINARY, regf, 16)
      g_mkreg(regf, "SENDER00" as *u8)
      nx_ws_send_frame_to_fd(fs2, 1, WS_OP_BINARY, regf, 16)
      // drain the reg cross-traffic (R1 sees R2+S regs, R2 sees R1+S regs; S sees R1+R2 but never reads)
      let dbuf: *u8 = sys_mmap(4096)
      g_read_payload(fr1, frame_buf, 16384, dbuf, 4096)
      g_read_payload(fr1, frame_buf, 16384, dbuf, 4096)
      g_read_payload(fr2, frame_buf, 16384, dbuf, 4096)
      g_read_payload(fr2, frame_buf, 16384, dbuf, 4096)
      // R2 subscribes LO for SENDER00 (kind 0x58; consumed by the relay, never broadcast)
      let lsub: *u8 = sys_mmap(64)
      lsub[0] = 0x58 as u8
      var lz: i64 = 0
      let r2id: *u8 = "RCVR0002" as *u8
      while lz < 8 { lsub[1+lz] = r2id[lz]; lz = lz + 1 }
      lsub[9]=0 as u8; lsub[10]=0 as u8; lsub[11]=0 as u8; lsub[12]=0 as u8
      let sid: *u8 = "SENDER00" as *u8
      lz = 0
      while lz < 8 { lsub[13+lz] = sid[lz]; lz = lz + 1 }
      lsub[21] = 1 as u8
      nx_ws_send_frame_to_fd(fr2, 1, WS_OP_BINARY, lsub, 22)
      sys_sleep_ms(80)                                     // LSUB rides R2's conn; let the relay apply it
      // S emits: HI key, HI P, LO key, HI P, LO P  (in_seq spaces deliberately disjoint per layer)
      let vf: *u8 = sys_mmap(256)
      g_mk57(vf, sid, 100, 1); nx_ws_send_frame_to_fd(fs2, 1, WS_OP_BINARY, vf, 32)   // HI KEY
      g_mk57(vf, sid, 101, 0); nx_ws_send_frame_to_fd(fs2, 1, WS_OP_BINARY, vf, 32)   // HI P
      g_mk57(vf, sid, 7,   3); nx_ws_send_frame_to_fd(fs2, 1, WS_OP_BINARY, vf, 32)   // LO KEY
      g_mk57(vf, sid, 102, 0); nx_ws_send_frame_to_fd(fs2, 1, WS_OP_BINARY, vf, 32)   // HI P
      g_mk57(vf, sid, 8,   2); nx_ws_send_frame_to_fd(fs2, 1, WS_OP_BINARY, vf, 32)   // LO P
      // R1 (default HI): must read exactly HI key(0), HI P(1), HI P(2) -- the LO frames never arrive
      var r1ok: i64 = 1
      let rb: *u8 = sys_mmap(4096)
      var n1: i64 = g_read_payload(fr1, frame_buf, 16384, rb, 4096)
      if n1 < 18 { r1ok = 0 } else { if (rb[13]&1) != 1 { r1ok = 0 } if (rb[13]&2) != 0 { r1ok = 0 } if g_rdseq(rb) != 0 { r1ok = 0 } }
      n1 = g_read_payload(fr1, frame_buf, 16384, rb, 4096)
      if n1 < 18 { r1ok = 0 } else { if (rb[13]&3) != 0 { r1ok = 0 } if g_rdseq(rb) != 1 { r1ok = 0 } }
      n1 = g_read_payload(fr1, frame_buf, 16384, rb, 4096)
      if n1 < 18 { r1ok = 0 } else { if (rb[13]&3) != 0 { r1ok = 0 } if g_rdseq(rb) != 2 { r1ok = 0 } }
      pass = pass + g_check("SFU: default receiver gets ONLY HI, seq rewritten 0,1,2" as *u8, r1ok == 1); total = total + 1
      // R2 (LSUB LO): HI key(0), HI P(1), LO KEY(2) = switch, LO P(3); the post-switch HI P never arrives
      var r2ok: i64 = 1
      var n2: i64 = g_read_payload(fr2, frame_buf, 16384, rb, 4096)
      if n2 < 18 { r2ok = 0 } else { if (rb[13]&3) != 1 { r2ok = 0 } if g_rdseq(rb) != 0 { r2ok = 0 } }
      n2 = g_read_payload(fr2, frame_buf, 16384, rb, 4096)
      if n2 < 18 { r2ok = 0 } else { if (rb[13]&3) != 0 { r2ok = 0 } if g_rdseq(rb) != 1 { r2ok = 0 } }
      n2 = g_read_payload(fr2, frame_buf, 16384, rb, 4096)
      if n2 < 18 { r2ok = 0 } else { if (rb[13]&3) != 3 { r2ok = 0 } if g_rdseq(rb) != 2 { r2ok = 0 } }
      n2 = g_read_payload(fr2, frame_buf, 16384, rb, 4096)
      if n2 < 18 { r2ok = 0 } else { if (rb[13]&3) != 2 { r2ok = 0 } if g_rdseq(rb) != 3 { r2ok = 0 } }
      pass = pass + g_check("SFU: subscriber switches to LO AT the LO key, seq contiguous across switch" as *u8, r2ok == 1); total = total + 1
      // starvation guard: R1 must still be flowing HI (one more HI P -> seq 3), proving pair isolation
      g_mk57(vf, sid, 103, 0); nx_ws_send_frame_to_fd(fs2, 1, WS_OP_BINARY, vf, 32)   // HI P
      var r3ok: i64 = 1
      n1 = g_read_payload(fr1, frame_buf, 16384, rb, 4096)
      if n1 < 18 { r3ok = 0 } else { if (rb[13]&3) != 0 { r3ok = 0 } if g_rdseq(rb) != 3 { r3ok = 0 } }
      // and R2 must NOT get that HI P: with a 1s timeout this read must FAIL (nothing pending for R2)
      g_tmo(fr2, 1)
      let nn: i64 = g_read_payload(fr2, frame_buf, 16384, rb, 4096)
      if nn >= 0 { r3ok = 0 }
      pass = pass + g_check("SFU: pair isolation (R1 flows on; R2 clean of dropped HI)" as *u8, r3ok == 1); total = total + 1
      sys_close(fs2); sys_close(fr1); sys_close(fr2)
  }
437
  // ============ R4: DOMINANT-SPEAKER (0x5A announce) + LAST-N (0x59 cap), E2E ============
  // domroom: S1 loud (fat LPC frames = speech), S2 silent, R receiver. Speech scoring is payload-length
  // (LPC compresses silence away), so a FAT 0x4C frame IS the speech signal.
  let ds1: i64 = g_join_room("/signal/domroom" as *u8, 15, req_buf, resp_buf)
  let ds2: i64 = g_join_room("/signal/domroom" as *u8, 15, req_buf, resp_buf)
  let dr: i64  = g_join_room("/signal/domroom" as *u8, 15, req_buf, resp_buf)
  var dok: i64 = 0
  if ds1 >= 0 { if ds2 >= 0 { if dr >= 0 { dok = 1 } } }
  if dok == 1 {
      g_tmo(ds1, 2); g_tmo(ds2, 2); g_tmo(dr, 2)
      let regf2: *u8 = sys_mmap(64)
      g_mkreg(regf2, "SPKLOUD1" as *u8); nx_ws_send_frame_to_fd(ds1, 1, WS_OP_BINARY, regf2, 16)
      g_mkreg(regf2, "SPKQUIET" as *u8); nx_ws_send_frame_to_fd(ds2, 1, WS_OP_BINARY, regf2, 16)
      g_mkreg(regf2, "RECVLASN" as *u8); nx_ws_send_frame_to_fd(dr, 1, WS_OP_BINARY, regf2, 16)
      let dbuf2: *u8 = sys_mmap(4096)
      g_read_payload(ds1, frame_buf, 16384, dbuf2, 4096); g_read_payload(ds1, frame_buf, 16384, dbuf2, 4096)
      g_read_payload(ds2, frame_buf, 16384, dbuf2, 4096); g_read_payload(ds2, frame_buf, 16384, dbuf2, 4096)
      g_read_payload(dr, frame_buf, 16384, dbuf2, 4096);  g_read_payload(dr, frame_buf, 16384, dbuf2, 4096)
      // T19: S1 speaks (one FAT 0x4C) -> relay announces 0x5A[SPKLOUD1] to the room BEFORE relaying audio
      let aud: *u8 = sys_mmap(1024)
      aud[0] = 0x4C as u8
      let s1id: *u8 = "SPKLOUD1" as *u8
      var az: i64 = 0
      while az < 8 { aud[1+az] = s1id[az]; az = az + 1 }
      az = 9
      while az < 813 { aud[az] = ((az * 7) & 255) as u8; az = az + 1 }   // 800B of "speech"
      nx_ws_send_frame_to_fd(ds1, 1, WS_OP_BINARY, aud, 813)
      var t19: i64 = 1
      let rb2: *u8 = sys_mmap(4096)
      var m1: i64 = g_read_payload(dr, frame_buf, 16384, rb2, 4096)
      if m1 < 9 { t19 = 0 } else {
          if (rb2[0]&0xff) != 0x5A { t19 = 0 }
          var qz: i64 = 0
          while qz < 8 { if rb2[1+qz] != s1id[qz] { t19 = 0; qz = 8 } else { qz = qz + 1 } }
      }
      m1 = g_read_payload(dr, frame_buf, 16384, rb2, 4096)          // then the relayed audio itself
      if m1 != 813 { t19 = 0 } else { if (rb2[0]&0xff) != 0x4C { t19 = 0 } }
      pass = pass + g_check("R4: dominant-speaker 0x5A announced on speech (LPC-length energy)" as *u8, t19 == 1); total = total + 1
      // drain S2's copies (0x5A + audio) so its socket is clean
      g_read_payload(ds2, frame_buf, 16384, dbuf2, 4096); g_read_payload(ds2, frame_buf, 16384, dbuf2, 4096)
      g_read_payload(ds1, frame_buf, 16384, dbuf2, 4096)            // S1 gets the 0x5A too (not its own audio)
      // T20: R caps last-N=1 -> only the ACTIVE speaker's video passes; silent S2's video blocked for R
      let lns: *u8 = sys_mmap(32)
      lns[0] = 0x59 as u8
      let rid2: *u8 = "RECVLASN" as *u8
      az = 0
      while az < 8 { lns[1+az] = rid2[az]; az = az + 1 }
      lns[9]=0 as u8; lns[10]=0 as u8; lns[11]=0 as u8; lns[12]=0 as u8
      lns[13] = 1 as u8
      nx_ws_send_frame_to_fd(dr, 1, WS_OP_BINARY, lns, 14)
      sys_sleep_ms(80)
      let vf2: *u8 = sys_mmap(256)
      g_mk57(vf2, s1id, 500, 1); nx_ws_send_frame_to_fd(ds1, 1, WS_OP_BINARY, vf2, 32)   // loud S1 video KEY
      g_mk57(vf2, "SPKQUIET" as *u8, 900, 1); nx_ws_send_frame_to_fd(ds2, 1, WS_OP_BINARY, vf2, 32)  // silent S2 video KEY
      var t20: i64 = 1
      m1 = g_read_payload(dr, frame_buf, 16384, rb2, 4096)          // R: S1's video passes (top-1 active)
      if m1 < 19 { t20 = 0 } else {
          if (rb2[0]&0xff) != 0x57 { t20 = 0 }
          var qz2: i64 = 0
          while qz2 < 8 { if rb2[1+qz2] != s1id[qz2] { t20 = 0; qz2 = 8 } else { qz2 = qz2 + 1 } }
      }
      g_tmo(dr, 1)
      let mnone: i64 = g_read_payload(dr, frame_buf, 16384, rb2, 4096)   // S2's video must NOT arrive at R
      if mnone >= 0 { t20 = 0 }
      // control: S1 (no cap) DOES receive silent S2's video (last-N is per-receiver, not global).
      // Skim 0x5B LPROD control chatter (R5 baseline announces ride the same socket).
      m1 = g_read_payload(ds1, frame_buf, 16384, rb2, 4096)
      var skim: i64 = 0
      while skim < 3 { if m1 >= 1 { if (rb2[0]&0xff) == 0x5B { m1 = g_read_payload(ds1, frame_buf, 16384, rb2, 4096); skim = skim + 1 } else { skim = 3 } } else { skim = 3 } }
      if m1 < 19 { t20 = 0 } else {
          if (rb2[0]&0xff) != 0x57 { t20 = 0 }
          let q2id: *u8 = "SPKQUIET" as *u8
          var qz3: i64 = 0
          while qz3 < 8 { if rb2[1+qz3] != q2id[qz3] { t20 = 0; qz3 = 8 } else { qz3 = qz3 + 1 } }
      }
      pass = pass + g_check("R4: last-N=1 per receiver (active passes, silent blocked, others unaffected)" as *u8, t20 == 1); total = total + 1
      // T21 R5 LAYER SUSPENSION: R re-subscribes S1's LO -> S1 gets LPROD 0x5B want=1; R back to HI +
      // the LO chain switches away -> S1's next video frame triggers LPROD want=0. (First-eval baseline
      // want=0 was already announced to S1 by its T20 video frames -- drain anything pending first.)
      g_tmo(ds1, 1)
      var drn: i64 = 0
      while drn == 0 { if g_read_payload(ds1, frame_buf, 16384, dbuf2, 4096) < 0 { drn = 1 } }
      var t21: i64 = 1
      let lsub2: *u8 = sys_mmap(64)
      lsub2[0] = 0x58 as u8
      az = 0
      while az < 8 { lsub2[1+az] = rid2[az]; az = az + 1 }
      lsub2[9]=0 as u8; lsub2[10]=0 as u8; lsub2[11]=0 as u8; lsub2[12]=0 as u8
      az = 0
      while az < 8 { lsub2[13+az] = s1id[az]; az = az + 1 }
      lsub2[21] = 1 as u8                                            // R wants S1's LO
      nx_ws_send_frame_to_fd(dr, 1, WS_OP_BINARY, lsub2, 22)
      g_tmo(ds1, 2)
      var m21: i64 = g_read_payload(ds1, frame_buf, 16384, rb2, 4096)   // S1 must get LPROD want=1
      if m21 < 14 { t21 = 0 } else {
          if (rb2[0]&0xff) != 0x5B { t21 = 0 }
          if (rb2[13]&0xff) != 1 { t21 = 0 }
      }
      // S1 ships a LO key (now wanted) -> R switches to LO; then R unsubscribes; then S1's HI key both
      // completes the switch-back AND (frame-driven re-eval) earns S1 the LPROD want=0.
      g_mk57(vf2, s1id, 600, 3); nx_ws_send_frame_to_fd(ds1, 1, WS_OP_BINARY, vf2, 32)   // LO KEY
      g_read_payload(dr, frame_buf, 16384, dbuf2, 4096)              // R receives it (switched to LO)
      lsub2[21] = 0 as u8
      nx_ws_send_frame_to_fd(dr, 1, WS_OP_BINARY, lsub2, 22)         // R back to HI (no LPROD yet: LO still flowing)
      sys_sleep_ms(60)
      g_mk57(vf2, s1id, 601, 1); nx_ws_send_frame_to_fd(ds1, 1, WS_OP_BINARY, vf2, 32)   // HI KEY -> switch-back
      g_read_payload(dr, frame_buf, 16384, dbuf2, 4096)              // R gets the HI key
      g_mk57(vf2, s1id, 602, 0); nx_ws_send_frame_to_fd(ds1, 1, WS_OP_BINARY, vf2, 32)   // next frame re-verdicts
      g_read_payload(dr, frame_buf, 16384, dbuf2, 4096)
      var m22: i64 = g_read_payload(ds1, frame_buf, 16384, rb2, 4096)   // S1 must get LPROD want=0
      if m22 < 14 { t21 = 0 } else {
          if (rb2[0]&0xff) != 0x5B { t21 = 0 }
          if (rb2[13]&0xff) != 0 { t21 = 0 }
      }
      pass = pass + g_check("R5: layer suspension LPROD (want=1 on subscribe, want=0 when unwatched)" as *u8, t21 == 1); total = total + 1
      // T22 RECEIVER REPORT routing: R sends RR(0x52) targeting S1 -> ONLY S1 receives it (S2 clean).
      var t22: i64 = 1
      g_tmo(ds2, 1)                                                  // S2 accumulated T20/T21 broadcast copies
      var drn2: i64 = 0                                              // + its own LPROD baseline -> drain to empty
      while drn2 == 0 { if g_read_payload(ds2, frame_buf, 16384, dbuf2, 4096) < 0 { drn2 = 1 } }
      let rrf: *u8 = sys_mmap(64)
      rrf[0] = 0x52 as u8
      az = 0
      while az < 8 { rrf[1+az] = rid2[az]; az = az + 1 }
      rrf[9]=0 as u8; rrf[10]=0 as u8; rrf[11]=0 as u8; rrf[12]=0 as u8
      az = 0
      while az < 8 { rrf[13+az] = s1id[az]; az = az + 1 }            // target = SPKLOUD1
      rrf[21] = 14 as u8                                             // rxfps=14
      nx_ws_send_frame_to_fd(dr, 1, WS_OP_BINARY, rrf, 22)
      g_tmo(ds1, 2)
      var m23: i64 = g_read_payload(ds1, frame_buf, 16384, rb2, 4096)
      var skim2: i64 = 0
      while skim2 < 3 { if m23 >= 1 { if (rb2[0]&0xff) == 0x5B { m23 = g_read_payload(ds1, frame_buf, 16384, rb2, 4096); skim2 = skim2 + 1 } else { skim2 = 3 } } else { skim2 = 3 } }
      if m23 != 22 { t22 = 0 } else {
          if (rb2[0]&0xff) != 0x52 { t22 = 0 }
          if (rb2[21]&0xff) != 14 { t22 = 0 }
      }
      g_tmo(ds2, 1)
      let mnone2: i64 = g_read_payload(ds2, frame_buf, 16384, rb2, 4096)   // S2 must get NOTHING
      if mnone2 >= 0 { t22 = 0 }
      pass = pass + g_check("RR: receiver report routed to its TARGET sender only (0x52)" as *u8, t22 == 1); total = total + 1
      sys_close(ds1); sys_close(ds2); sys_close(dr)
  } else {
      pass = pass + g_check("R4: domroom join" as *u8, 0); total = total + 1
  }
583
  // ============ ROOM TOKENS: capability-link wall (keyroom protected, freeroom open) ============
  let tok: *u8 = sys_mmap(32)
  rk_token("keyroom" as *u8, 7, "gatesecret" as *u8, 10, tok)   // same SSOT organ the relay uses
  let kp: *u8 = sys_mmap(64)
  let kplen: i64 = g_mkkeypath(kp, tok)
  // T23 ADMIT: two valid-token joins into keyroom relay to each other (proves admit + room functions).
  let ka: i64 = g_join_room(kp, kplen, req_buf, resp_buf)
  let kb: i64 = g_join_room(kp, kplen, req_buf, resp_buf)
  var t23: i64 = 0
  if ka >= 0 { if kb >= 0 {
      g_tmo(kb, 2)
      nx_ws_send_frame_to_fd(ka, 1, WS_OP_TEXT, "validok!" as *u8, 8)
      let nkb: i64 = g_read_payload(kb, frame_buf, 16384, pay, 4096)
      if nkb == 8 { if g_eq(pay, "validok!" as *u8, 8) == 1 { t23 = 1 } }
  } }
  pass = pass + g_check("TOKEN: valid token admits + keyed room relays" as *u8, t23 == 1); total = total + 1
  // T24 REJECT: no-token + wrong-token joins are closed pre-membership -> they never receive ka's frame,
  // but the valid kb does (proves rejection + no leakage in one shot).
  let knox: i64 = g_join_room("/signal/keyroom" as *u8, 15, req_buf, resp_buf)   // no ?k
  let kwp: *u8 = sys_mmap(64)
  let bad: *u8 = "00000000deadbeef" as *u8
  let kwlen: i64 = g_mkkeypath(kwp, bad)                                          // wrong token
  let kwrong: i64 = g_join_room(kwp, kwlen, req_buf, resp_buf)
  var t24: i64 = 1
  g_tmo(knox, 1); g_tmo(kwrong, 1); g_tmo(kb, 2)
  sys_sleep_ms(60)
  nx_ws_send_frame_to_fd(ka, 1, WS_OP_TEXT, "afterrej" as *u8, 8)
  let nbk: i64 = g_read_payload(kb, frame_buf, 16384, pay, 4096)                  // valid peer gets it
  if nbk != 8 { t24 = 0 } else { if g_eq(pay, "afterrej" as *u8, 8) != 1 { t24 = 0 } }
  if g_read_payload(knox, frame_buf, 16384, pay, 4096) == 8 { t24 = 0 }           // rejected: no frame
  if g_read_payload(kwrong, frame_buf, 16384, pay, 4096) == 8 { t24 = 0 }
  pass = pass + g_check("TOKEN: no-token + wrong-token REJECTED (closed, no leak)" as *u8, t24 == 1); total = total + 1
  // T25 OPEN unaffected: an UNLISTED room joins + relays normally even with the wall armed.
  let oa: i64 = g_join_room("/signal/freeroom" as *u8, 16, req_buf, resp_buf)
  let ob: i64 = g_join_room("/signal/freeroom" as *u8, 16, req_buf, resp_buf)
  var t25: i64 = 0
  if oa >= 0 { if ob >= 0 {
      g_tmo(ob, 2)
      nx_ws_send_frame_to_fd(oa, 1, WS_OP_TEXT, "freepass" as *u8, 8)
      let nob: i64 = g_read_payload(ob, frame_buf, 16384, pay, 4096)
      if nob == 8 { if g_eq(pay, "freepass" as *u8, 8) == 1 { t25 = 1 } }
  } }
  pass = pass + g_check("TOKEN: unlisted room UNAFFECTED by armed wall" as *u8, t25 == 1); total = total + 1
  sys_close(ka); sys_close(kb); sys_close(knox); sys_close(kwrong); sys_close(oa); sys_close(ob)
628
  // teardown (nx_kill: rv64 129 -> x86 62 via the __syscall translation table)
  nx_kill(pid, 9)
  g_puts("---- signaling_v2 gate: passed " as *u8); g_pn(pass); g_puts(" / " as *u8); g_pn(total); g_puts(" ----\n" as *u8)
  if pass == total { return 0 }
  return 1
634}
nx_signaling_v2_gate.nx source

`nx_signaling_v2_gate.nx` source
