nx_websocket_stream.nx source
↩ module page · 189 lines · 7018 B
1// nx_websocket_stream.nx -- streaming WebSocket framer over a TCP fd.
2//
3// The missing piece between nx_websocket_frame (parse one frame from
4// a buffer) and a real per-connection daemon loop. Reads bytes from
5// the socket handling partial reads + assembles a complete frame
6// (header + payload), then unmasks the payload in place per RFC 6455
7// §5.3 (client-to-server frames MUST be masked).
8//
9// Composes:
10// nx_websocket_frame.ws_parse_frame -- header decode
11// nx_websocket_frame.ws_apply_mask -- in-place unmask
12// nx_websocket_frame.ws_build_header -- server-to-client frame build
13// sys_read / sys_write -- I/O
14//
15// Per Cardinals 9 (single-responsibility: ONE frame per call), 12
16// (defensive at boundary: bounds checks on payload_len + out_cap),
17// 22 (composition of shipped primitives, no inline frame parsing).
18//
19// Caller responsibilities:
20// - allocate out_buf with cap >= max expected frame (header + payload)
21// - check returned verdict before using out_frame fields
22// - for incoming client frames, payload is unmasked in place
23//
24// genealogy_id: rfc_6455 + nx_websocket_frame + nx_tls13_read_record_from_fd_q10
25// lineage_id: nishi_websocket_stream_q1
26
27import "nx_syscalls.nx"
28import "nx_websocket_frame.nx"
29
30const NX_WSS_OK: i64 = 1
31const NX_WSS_READ_FAIL: i64 = 2
32const NX_WSS_EOF: i64 = 3
33const NX_WSS_FRAME_TOO_BIG: i64 = 4
34const NX_WSS_PARSE_FAIL: i64 = 5
35const NX_WSS_WRITE_FAIL: i64 = 6
36const NX_WSS_VERDICT_N: i64 = 7
37
38func nx_wss_verdict_is_valid(v: i64) -> i64 {
39 if v < NX_WSS_OK { return 0 }
40 if v >= NX_WSS_VERDICT_N { return 0 }
41 return 1
42}
43
44// Read bytes from fd into buf[off..off+want), looping until exactly
45// `want` bytes arrive or EOF/error. Returns 0 on full read, 1 on
46// EOF before any bytes (clean close), -1 on read error or short EOF.
47func _wss_read_n(fd: i64, buf: *u8, off: i64, want: i64) -> i64 {
48 var got: i64 = 0
49 while got < want {
50 let r: i64 = sys_read(fd, (buf as i64 + off + got) as *u8, want - got)
51 if r == 0 {
52 if got == 0 { return 1 } // clean EOF at frame boundary
53 return 0 - 1 // short EOF mid-frame
54 }
55 if r < 0 { return 0 - 1 }
56 got = got + r
57 }
58 return 0
59}
60
61// Write all of buf[0..n) to fd via looping sys_write. Returns 0 on
62// success, -1 on any sys_write error.
63func _wss_write_n(fd: i64, buf: *u8, n: i64) -> i64 {
64 var off: i64 = 0
65 while off < n {
66 let w: i64 = sys_write(fd, (buf as i64 + off) as *u8, n - off)
67 if w <= 0 { return 0 - 1 }
68 off = off + w
69 }
70 return 0
71}
72
73// Read one complete WebSocket frame from fd into out_buf. On success
74// populates out_frame and unmasks payload in place if frame was
75// client-masked. Returns verdict.
76//
77// Reads in three phases:
78// 1. Read 2-byte minimal header (b0 + b1).
79// 2. Determine extended length bytes (0 / 2 / 8) + mask bytes (0 / 4)
80// from b1; read those.
81// 3. Read payload bytes.
82//
83// After phase 1+2 we re-parse the now-complete header via ws_parse_frame
84// to populate out_frame. Then phase 3 reads the payload. Then we
85// unmask in place if frame.masked.
86func nx_ws_read_frame_from_fd(fd: i64,
87 out_buf: *u8, out_cap: i64,
88 out_frame: *WsFrame) -> i64 {
89 if out_cap < 2 { return NX_WSS_FRAME_TOO_BIG }
90
91 // Phase 1: read 2 minimum bytes.
92 let r1: i64 = _wss_read_n(fd, out_buf, 0, 2)
93 if r1 == 1 { return NX_WSS_EOF }
94 if r1 < 0 { return NX_WSS_READ_FAIL }
95
96 let b1: i64 = out_buf[1] & 0xFF
97 let len7: i64 = b1 & 0x7F
98 let mask_bit: i64 = (b1 >> 7) & 0x1
99
100 var ext_len_bytes: i64 = 0
101 if len7 == 126 { ext_len_bytes = 2 }
102 if len7 == 127 { ext_len_bytes = 8 }
103
104 var mask_bytes: i64 = 0
105 if mask_bit == 1 { mask_bytes = 4 }
106
107 let header_extra: i64 = ext_len_bytes + mask_bytes
108 if 2 + header_extra > out_cap { return NX_WSS_FRAME_TOO_BIG }
109
110 // Phase 2: read extended-length + mask-key bytes.
111 if header_extra > 0 {
112 let r2: i64 = _wss_read_n(fd, out_buf, 2, header_extra)
113 if r2 != 0 { return NX_WSS_READ_FAIL }
114 }
115
116 // Parse the now-complete HEADER. ws_parse_frame's final bounds
117 // check tests n >= frame_len which includes the not-yet-read
118 // payload, so we lie with n=out_cap (the buffer we've already
119 // verified is big enough for the full frame). ws_parse_frame
120 // only READS bytes through (2 + header_extra), so the uninitialized
121 // payload region of out_buf is untouched.
122 let pv: i64 = ws_parse_frame(out_buf, out_cap, 0, out_frame)
123 if pv != 0 { return NX_WSS_PARSE_FAIL }
124
125 // Bounds: full frame must fit in out_cap (cheap re-check).
126 if out_frame.frame_len > out_cap { return NX_WSS_FRAME_TOO_BIG }
127 if out_frame.payload_len < 0 { return NX_WSS_PARSE_FAIL }
128
129 // Phase 3: read payload bytes.
130 if out_frame.payload_len > 0 {
131 let r3: i64 = _wss_read_n(fd, out_buf, out_frame.payload_off,
132 out_frame.payload_len)
133 if r3 != 0 { return NX_WSS_READ_FAIL }
134 }
135
136 // Unmask in place if client-masked.
137 if out_frame.masked == 1 {
138 ws_apply_mask(out_buf, out_frame.payload_off,
139 out_frame.payload_len, out_frame.mask_key)
140 }
141
142 return NX_WSS_OK
143}
144
145// Send a server-to-client frame on fd. Builds the header (mask=0 per
146// RFC 6455 §5.1 server side) then writes header + payload as two
147// sys_writes. fin=1 for non-fragmented messages.
148func nx_ws_send_frame_to_fd(fd: i64,
149 fin: i64, opcode: i64,
150 payload: *u8, payload_len: i64) -> i64 {
151 let hdr: *u8 = sys_mmap(16)
152 let hlen: i64 = ws_build_header(hdr, 16, fin, opcode, 0, 0, payload_len)
153 if hlen < 0 { return NX_WSS_PARSE_FAIL }
154
155 let w1: i64 = _wss_write_n(fd, hdr, hlen)
156 if w1 < 0 { return NX_WSS_WRITE_FAIL }
157
158 if payload_len > 0 {
159 let w2: i64 = _wss_write_n(fd, payload, payload_len)
160 if w2 < 0 { return NX_WSS_WRITE_FAIL }
161 }
162 return NX_WSS_OK
163}
164
165// Send a text frame (opcode 0x1) with fin=1 -- shortcut.
166func nx_ws_send_text(fd: i64, text: *u8, text_len: i64) -> i64 {
167 return nx_ws_send_frame_to_fd(fd, 1, WS_OP_TEXT, text, text_len)
168}
169
170// Send a pong frame (opcode 0xA, echoing the ping payload).
171func nx_ws_send_pong(fd: i64, payload: *u8, payload_len: i64) -> i64 {
172 return nx_ws_send_frame_to_fd(fd, 1, WS_OP_PONG, payload, payload_len)
173}
174
175// Send a close frame (opcode 0x8) with an optional status code + reason.
176// Empty payload = clean close without code.
177func nx_ws_send_close(fd: i64, code: i64) -> i64 {
178 if code == 0 {
179 return nx_ws_send_frame_to_fd(fd, 1, WS_OP_CLOSE, 0 as *u8, 0)
180 }
181 let p: *u8 = sys_mmap(8)
182 p[0] = ((code >> 8) & 0xFF) as u8
183 p[1] = (code & 0xFF) as u8
184 return nx_ws_send_frame_to_fd(fd, 1, WS_OP_CLOSE, p, 2)
185}
186
187func main() -> i64 {
188 return 0
189}