nx_room_live_gate.nx source

1import "nx_gate_base.nx" 2import "nx_connect.nx" // bounded connect: a raw sys_connect hangs ~127s on a black-holed host 3// nx_room_live_gate.nx -- ENGINEER gate: MULTIPLE USERS IN A LIVE VIDEO ROOM. 4// 5// Operator 2026-06-10: "make sure that the team actually tests that multiple 6// users can be in a video room and interact." The loopback signaling gate 7// proves relay SEMANTICS; THIS gate proves the WHOLE USER PATH on the real 8// public stack: DNS -> TCP -> TLS 1.3 (cert chain validated against the real 9// Mozilla store) -> wss upgrade on nishifamily.com:443 -> RFC 6455 frames in 10// the room's ACTUAL wire format (the same 13-byte kind/id/seq header the 11// browser client speaks) -> N-party relay back to every other peer. 12// 13// Three sovereign clients (A, B, C) join the same room as three USERS: 14// - all three complete the wss upgrade (101) through the public domain 15// - A says hello (text) -> B AND C receive it verbatim 16// - B sends a VIDEO frame (binary, kind 'V' + id + seq + 8KB JPEG-shaped 17// payload) -> A AND C receive it bit-exact + parse it 18// - C sends an AUDIO frame (binary, kind 'A' + id + seq + rate + PCM) 19// -> A AND B receive it bit-exact 20// - A pings -> pong comes back to A only 21// - A leaves -> B->C still relays (room survives churn) 22// 23// Every media byte crosses the open internet through OUR daemon -- nothing 24// is loopback, nothing is mocked. Requires /tmp/mozilla_certdata.txt 25// (runner stages it from data/). 26// license_tier: ORIGINAL 27 28import "nx_syscalls.nx" 29import "nx_gate_emit_lib.nx" 30import "nx_x509_trust_store.nx" 31import "nx_trust_store_load_from_certdata.nx" 32import "nx_tls13_client_validate_certificate.nx" 33import "nx_tls13_client_session_run.nx" 34import "nx_dns_resolve_a_record.nx" 35import "nx_https_url_connect.nx" 36import "nx_websocket_frame.nx" 37import "nx_websocket_client_upgrade.nx" 38 39 40// Slot layout for the per-connection state array `c` (*i64, RG_SLOTS entries, allocated by this 41// file). The whole block was missing -- defined in NO file in the tree -- so nx_parse stopped at the 42// first use (RG_S) and both gates had been unbuildable. These are private indices, so any distinct 43// assignment is correct; RG_SLOTS is the allocation count. 44const RG_FD: i64 = 0 45const RG_S: i64 = 1 46const RG_ACC: i64 = 2 47const RG_ACC_CAP: i64 = 3 48const RG_ACC_LEN: i64 = 4 49const RG_ACC_OFF: i64 = 5 50const RG_SLOTS: i64 = 6 51 52 53func grow(name: *u8, ok: i64) -> i64 { if ok==1 { gw(" PASS " as *u8) } else { gw(" FAIL " as *u8) } gw(name); gw(" 54" as *u8); return ok } 55func rg_tls_send(c: *i64, buf: *u8, n: i64) -> i64 { 56 let s: *Tls13ClientSession = c[RG_S] as *Tls13ClientSession 57 let rec: *u8 = sys_mmap(n + 64) 58 let hdr: *u8 = rec 59 let ct: *u8 = rec + NX_TLS13_RECORD_HEADER_LEN 60 let tag: *u8 = rec + NX_TLS13_RECORD_HEADER_LEN + n + 1 61 let v: i64 = nx_tls13_record_encrypt_v2( 62 s.cipher_suite, s.client_app_traffic_key, s.client_app_iv, 63 s.client_app_seq, buf, n, NX_TLS13_CT_APPLICATION_DATA, 0, 64 hdr, ct, tag) 65 s.client_app_seq = s.client_app_seq + 1 66 if v != NX_TLS13_REC_VERDICT_OK { return 0 - 1 } 67 let total: i64 = NX_TLS13_RECORD_HEADER_LEN + n + 1 + NX_TLS13_RECORD_TAG_LEN 68 var off: i64 = 0 69 while off < total { 70 let w: i64 = sys_write(c[RG_FD], (rec as i64 + off) as *u8, total - off) 71 if w <= 0 { return 0 - 1 } 72 off = off + w 73 } 74 return 0 75} 76 77// pump ONE TLS record from the wire into the accumulator (app data only; 78// NewSessionTicket etc. skipped). Returns bytes appended, 0 if the record 79// was non-app, negative on error/close. 80func rg_pump(c: *i64) -> i64 { 81 let s: *Tls13ClientSession = c[RG_S] as *Tls13ClientSession 82 let rec: *u8 = sys_mmap(16700) 83 let total: i64 = nx_tls13_read_record_from_fd(c[RG_FD], rec, 16700) 84 if total < 0 { return 0 - 1 } 85 let ct_len: i64 = total - NX_TLS13_RECORD_HEADER_LEN - NX_TLS13_RECORD_TAG_LEN 86 if ct_len <= 0 { return 0 - 1 } 87 let plain: *u8 = sys_mmap(ct_len + 16) 88 let ctp: *i64 = sys_mmap(16) as *i64 89 let lnp: *i64 = sys_mmap(16) as *i64 90 let v: i64 = nx_tls13_record_decrypt_v2( 91 s.cipher_suite, s.server_app_traffic_key, s.server_app_iv, 92 s.server_app_seq, rec, rec + NX_TLS13_RECORD_HEADER_LEN, ct_len, 93 rec + total - NX_TLS13_RECORD_TAG_LEN, plain, ctp, lnp) 94 s.server_app_seq = s.server_app_seq + 1 95 if v != NX_TLS13_REC_VERDICT_OK { return 0 - 2 } 96 if *ctp == NX_TLS13_CT_ALERT { return 0 - 3 } 97 if *ctp != NX_TLS13_CT_APPLICATION_DATA { return 0 } 98 let acc: *u8 = c[RG_ACC] as *u8 99 var w: i64 = c[RG_ACC_LEN] 100 if w + *lnp > c[RG_ACC_CAP] { return 0 - 4 } 101 var i: i64 = 0 102 while i < *lnp { acc[w + i] = plain[i]; i = i + 1 } 103 c[RG_ACC_LEN] = w + *lnp 104 return *lnp 105} 106 107// parse the next complete WS frame out of the accumulator, pumping the 108// wire as needed (bounded). Returns 0 + fills f, or negative. 109func rg_next_frame(c: *i64, f: *WsFrame) -> i64 { 110 var tries: i64 = 0 111 while tries < 64 { 112 let acc: *u8 = c[RG_ACC] as *u8 113 let v: i64 = ws_parse_frame(acc, c[RG_ACC_LEN], c[RG_ACC_OFF], f) 114 if v == 0 { 115 c[RG_ACC_OFF] = c[RG_ACC_OFF] + f.frame_len 116 return 0 117 } 118 let p: i64 = rg_pump(c) 119 if p < 0 { return p } 120 tries = tries + 1 121 } 122 return 0 - 9 123} 124 125// masked client->server WS frame over TLS 126func rg_send_ws(c: *i64, opcode: i64, payload: *u8, n: i64, mask: i64) -> i64 { 127 let buf: *u8 = sys_mmap(n + 32) 128 let hl: i64 = ws_build_header(buf, n + 32, 1, opcode, 1, mask, n) 129 if hl < 0 { return 0 - 1 } 130 var i: i64 = 0 131 while i < n { buf[hl + i] = payload[i]; i = i + 1 } 132 ws_apply_mask(buf, hl, n, mask) 133 return rg_tls_send(c, buf, hl + n) 134} 135 136// connect + TLS handshake + wss upgrade one client into the room. 137// Returns 0 on success. 138func rg_join(c: *i64, ipv4: i64, sni: *u8, sni_n: i64, path: *u8, path_n: i64, 139 store: *TrustStore, seed: i64) -> i64 { 140 let fd: i64 = sys_socket(AF_INET, SOCK_STREAM, 0) 141 if fd < 0 { return 0 - 1 } 142 sys_set_socket_timeout(fd, 15) 143 let sa: *u8 = sys_mmap(16) 144 nx_https_build_sockaddr(sa, ipv4, 443) 145 if nx_connect_bounded(fd, sa, 16, NX_CONN_DEFAULT_MS) < 0 { sys_close(fd); return 0 - 2 } 146 147 let cr: *u8 = sys_mmap(32) 148 let priv: *u8 = sys_mmap(32) 149 var i: i64 = 0 150 while i < 32 { cr[i] = ((seed * 37 + i * 11 + 5) & 0xff) as u8; priv[i] = ((seed * 53 + i * 7 + 9) & 0xff) as u8; i = i + 1 } 151 152 let vraw: *u8 = sys_mmap(64) 153 let val_ctx: *TlsValidationContext = vraw as *TlsValidationContext 154 val_ctx.store = store 155 val_ctx.sni_host = sni 156 val_ctx.sni_host_len = sni_n 157 val_ctx.now_epoch = sys_now_realtime_sec() 158 159 let r: i64 = nx_tls13_client_session_run(fd, sni, sni_n, cr, priv, val_ctx) 160 if r < 0 { g_puts(" tls run verdict=" as *u8); g_pn(r); g_puts("\n" as *u8); sys_close(fd); return 0 - 3 } 161 c[RG_FD] = fd 162 c[RG_S] = r 163 c[RG_ACC] = sys_mmap(262144) as i64 164 c[RG_ACC_LEN] = 0 165 c[RG_ACC_OFF] = 0 166 c[RG_ACC_CAP] = 262144 167 168 // wss upgrade THROUGH the TLS session 169 let req: *u8 = sys_mmap(2048) 170 let key24: *u8 = sys_mmap(32) 171 let rn: i64 = nx_ws_client_build_request(req, 2048, sni, sni_n, path, path_n, 443, 0, key24) 172 if rn <= 0 { return 0 - 4 } 173 if rg_tls_send(c, req, rn) < 0 { return 0 - 5 } 174 // pump until the header terminator 175 var hend: i64 = 0 - 1 176 var tries: i64 = 0 177 while tries < 32 { 178 let acc: *u8 = c[RG_ACC] as *u8 179 var j: i64 = 0 180 while j + 3 < c[RG_ACC_LEN] { 181 var hit: i64 = 1 182 if acc[j] != (13 as u8) { hit = 0 } 183 if acc[j+1] != (10 as u8) { hit = 0 } 184 if acc[j+2] != (13 as u8) { hit = 0 } 185 if acc[j+3] != (10 as u8) { hit = 0 } 186 if hit == 1 { hend = j + 4; j = c[RG_ACC_LEN] } else { j = j + 1 } 187 } 188 if hend > 0 { tries = 32 } else { 189 if rg_pump(c) < 0 { return 0 - 6 } 190 tries = tries + 1 191 } 192 } 193 if hend < 0 { return 0 - 6 } 194 let acc2: *u8 = c[RG_ACC] as *u8 195 let uv: i64 = nx_ws_client_validate_response(acc2, hend, key24) 196 if uv != NX_WSCU_OK { g_puts(" upgrade verdict=" as *u8); g_pn(uv); g_puts("\n" as *u8); return 0 - 7 } 197 c[RG_ACC_OFF] = hend 198 return 0 199} 200 201// build the room's BINARY media payload: [kind][8-byte id][u32 seq LE][body] 202func rg_media(out: *u8, kind: i64, id: *u8, seq: i64, body: *u8, n: i64) -> i64 { 203 out[0] = kind as u8 204 var i: i64 = 0 205 while i < 8 { out[1 + i] = id[i]; i = i + 1 } 206 out[9] = (seq & 255) as u8 207 out[10] = ((seq / 256) & 255) as u8 208 out[11] = ((seq / 65536) & 255) as u8 209 out[12] = ((seq / 16777216) & 255) as u8 210 i = 0 211 while i < n { out[13 + i] = body[i]; i = i + 1 } 212 return 13 + n 213} 214 215func rg_eq(a: *u8, ao: i64, b: *u8, n: i64) -> i64 { 216 var i: i64 = 0 217 while i < n { if a[ao + i] != b[i] { return 0 } i = i + 1 } 218 return 1 219} 220 221// receive next frame on c and byte-compare against expected payload 222func rg_expect(c: *i64, f: *WsFrame, want: *u8, want_n: i64, want_op: i64) -> i64 { 223 if rg_next_frame(c, f) != 0 { return 0 } 224 if f.opcode != want_op { return 0 } 225 if f.payload_len != want_n { return 0 } 226 let acc: *u8 = c[RG_ACC] as *u8 227 return rg_eq(acc, f.payload_off, want, want_n) 228} 229 230func main() -> i64 { 231 g_puts("nx_room_live gate (3 USERS in one LIVE public video room)\n" as *u8) 232 var pass: i64 = 0 233 var total: i64 = 0 234 235 // ---- real Mozilla trust store ---- 236 let cpath: *u8 = "/tmp/mozilla_certdata.txt" as *u8 237 let lr: i64 = nx_trust_store_load_from_certdata(cpath, 300, 4194304) 238 if lr <= 0 { g_puts("FAIL trust store load (stage /tmp/mozilla_certdata.txt)\n" as *u8); return 2 } 239 let store: *TrustStore = lr as *TrustStore 240 241 // ---- DNS once ---- 242 let host: *u8 = "nishifamily.com" as *u8 243 let hn: i64 = 15 244 let dr: *DnsResolveResult = nx_dns_resolve_default(host, hn, sys_now_realtime_sec()) 245 if dr.verdict != NX_DNS_R_OK { g_puts("FAIL dns\n" as *u8); return 2 } 246 247 let path: *u8 = "/signal/liveroomgate" as *u8 248 let pn: i64 = g_slen(path) 249 250 // ---- three users join ---- 251 let ca: *i64 = sys_mmap(RG_SLOTS * 8) as *i64 252 let cb: *i64 = sys_mmap(RG_SLOTS * 8) as *i64 253 let cc: *i64 = sys_mmap(RG_SLOTS * 8) as *i64 254 let ja: i64 = rg_join(ca, dr.ipv4_packed, host, hn, path, pn, store, 1) 255 let jb: i64 = rg_join(cb, dr.ipv4_packed, host, hn, path, pn, store, 2) 256 let jc: i64 = rg_join(cc, dr.ipv4_packed, host, hn, path, pn, store, 3) 257 pass = pass + g_check("3 users joined: TLS1.3 (real-CA validated) + wss 101 x3, public domain" as *u8, 258 (ja == 0) + (jb == 0) + (jc == 0) == 3); total = total + 1 259 if ja != 0 { g_pn(ja); return 1 } 260 if jb != 0 { return 1 } 261 if jc != 0 { return 1 } 262 263 let fraw: *u8 = sys_mmap(128) 264 let f: *WsFrame = fraw as *WsFrame 265 266 // ---- A says hello -> B and C ---- 267 let hello: *u8 = "{\"from\":\"useraaaa\",\"type\":\"hello\",\"name\":\"alice\"}" as *u8 268 let hello_n: i64 = g_slen(hello) 269 rg_send_ws(ca, WS_OP_TEXT, hello, hello_n, 0x11223344) 270 pass = pass + g_check("user B received A's hello (text, verbatim)" as *u8, 271 rg_expect(cb, f, hello, hello_n, WS_OP_TEXT)); total = total + 1 272 pass = pass + g_check("user C received A's hello too (N-party)" as *u8, 273 rg_expect(cc, f, hello, hello_n, WS_OP_TEXT)); total = total + 1 274 275 // ---- B sends a VIDEO frame (room wire format, JPEG-shaped 8KB) ---- 276 let vbody: *u8 = sys_mmap(8000) 277 var i: i64 = 0 278 while i < 8000 { vbody[i] = ((i * 31 + 7) & 0xff) as u8; i = i + 1 } 279 let vfrm: *u8 = sys_mmap(8200) 280 let vn: i64 = rg_media(vfrm, 0x56, "userbbbb" as *u8, 42, vbody, 8000) 281 rg_send_ws(cb, WS_OP_BINARY, vfrm, vn, 0x55667788) 282 pass = pass + g_check("user A received B's VIDEO frame bit-exact (8KB binary)" as *u8, 283 rg_expect(ca, f, vfrm, vn, WS_OP_BINARY)); total = total + 1 284 pass = pass + g_check("user C received B's VIDEO frame bit-exact too" as *u8, 285 rg_expect(cc, f, vfrm, vn, WS_OP_BINARY)); total = total + 1 286 // parse the wire header out of C's copy (the contract the browser speaks) 287 let acc_c: *u8 = cc[RG_ACC] as *u8 288 let voff: i64 = f.payload_off 289 var hdr_ok: i64 = 1 290 if (acc_c[voff] & 0xff) != 0x56 { hdr_ok = 0 } 291 if rg_eq(acc_c, voff + 1, "userbbbb" as *u8, 8) != 1 { hdr_ok = 0 } 292 var rseq: i64 = acc_c[voff + 9] & 0xff 293 rseq = rseq + ((acc_c[voff + 10] & 0xff) * 256) 294 rseq = rseq + ((acc_c[voff + 11] & 0xff) * 65536) 295 rseq = rseq + ((acc_c[voff + 12] & 0xff) * 16777216) 296 if rseq != 42 { hdr_ok = 0 } 297 pass = pass + g_check("VIDEO wire header parses (kind=V id=userbbbb seq=42)" as *u8, hdr_ok); total = total + 1 298 299 // ---- C sends an AUDIO frame (rate in-band + PCM pattern) ---- 300 let abody: *u8 = sys_mmap(4100) 301 abody[0] = 0x80 as u8; abody[1] = 0xbb as u8; abody[2] = 0; abody[3] = 0 // 48000 LE 302 i = 0 303 while i < 4096 { abody[4 + i] = ((i * 13 + 5) & 0xff) as u8; i = i + 1 } 304 let afrm: *u8 = sys_mmap(4300) 305 let an: i64 = rg_media(afrm, 0x41, "usercccc" as *u8, 7, abody, 4100) 306 rg_send_ws(cc, WS_OP_BINARY, afrm, an, 0x99aabbcc) 307 pass = pass + g_check("user A received C's AUDIO frame bit-exact (PCM, rate in-band)" as *u8, 308 rg_expect(ca, f, afrm, an, WS_OP_BINARY)); total = total + 1 309 pass = pass + g_check("user B received C's AUDIO frame bit-exact too" as *u8, 310 rg_expect(cb, f, afrm, an, WS_OP_BINARY)); total = total + 1 311 312 // ---- ping/pong stays per-user ---- 313 rg_send_ws(ca, WS_OP_PING, "pp" as *u8, 2, 0x13579bdf) 314 var pong_ok: i64 = 0 315 if rg_next_frame(ca, f) == 0 { if f.opcode == WS_OP_PONG { pong_ok = 1 } } 316 pass = pass + g_check("A ping -> pong back to A (control not relayed)" as *u8, pong_ok); total = total + 1 317 318 // ---- A leaves; B<->C still interact ---- 319 let cbuf: *u8 = sys_mmap(8) 320 cbuf[0] = 3; cbuf[1] = 0xe8 as u8 // 1000 BE 321 rg_send_ws(ca, WS_OP_CLOSE, cbuf, 2, 0x2468ace0) 322 sys_close(ca[RG_FD]) 323 sys_sleep_ms(200) 324 let bye2: *u8 = "{\"from\":\"userbbbb\",\"type\":\"chat\",\"text\":\"still here?\"}" as *u8 325 let bye2_n: i64 = g_slen(bye2) 326 rg_send_ws(cb, WS_OP_TEXT, bye2, bye2_n, 0x0f1e2d3c) 327 pass = pass + g_check("after A left, B's message still reaches C (room survives churn)" as *u8, 328 rg_expect(cc, f, bye2, bye2_n, WS_OP_TEXT)); total = total + 1 329 330 sys_close(cb[RG_FD]) 331 sys_close(cc[RG_FD]) 332 333 g_puts("---- room_live gate: passed " as *u8); g_pn(pass); g_puts(" / " as *u8); g_pn(total); g_puts(" ----\n" as *u8) 334 if pass == total { return 0 } 335 return 1 336}