code wiki / (root) / nx_mcp_edge_boundary_gate.nx

nx_mcp_edge_boundary_gate.nx source

↩ module page · 44 lines · 2996 B

1import "nx_api_edge.nx" 2import "nx_gate_verdict.nx" 3func meh_case(name: *u8, headers: *u8, status: i64, ctr: *i64) -> i64 { 4 let req: *u8=sys_mmap(EDGE_REQ_MAX) 5 let out: *u8=sys_mmap(EDGE_RESP_CAP) 6 let body: *u8="{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\"}" 7 var n: i64=ta_cat(req,0,"POST /mcp HTTP/1.1\r\nHost: nishifamily.com\r\n") 8 n=ta_cat(req,n,headers); n=ta_cat(req,n,"Content-Length: "); n=ta_catn(req,n,ta_slen(body)); n=ta_cat(req,n,"\r\n\r\n"); n=ta_cat(req,n,body) 9 let got: i64=nx_api_edge_handle(req,n,out) 10 let expected: *u8=sys_mmap(64) 11 var e: i64=ta_cat(expected,0,"HTTP/1.1 "); e=ta_catn(expected,e,status); expected[e]=0 as u8 12 gv_check_eq(name,gv_at(out,got,0,expected),1,ctr) 13 return 0 14} 15func main() -> i64 { 16 let ctr: *i64=gv_ctr() 17 gv_head("MCP origin and header boundary regression") 18 meh_case("approved origin","Origin: https://nishifamily.com\r\n",200,ctr) 19 meh_case("lowercase approved","origin: https://nishifamily.com\r\n",200,ctr) 20 meh_case("uppercase approved","ORIGIN:\thttps://nishifamily.com \t\r\n",200,ctr) 21 meh_case("hostile origin","Origin: https://untrusted.invalid\r\n",403,ctr) 22 meh_case("lowercase hostile","origin: https://untrusted.invalid\r\n",403,ctr) 23 meh_case("duplicate approved origin","Origin: https://nishifamily.com\r\norigin: https://nishifamily.com\r\n",403,ctr) 24 meh_case("mixed duplicate origin","Origin: https://nishifamily.com\r\norigin: https://untrusted.invalid\r\n",403,ctr) 25 meh_case("absent origin","",200,ctr) 26 meh_case("opaque origin","Origin: null\r\n",403,ctr) 27 let spans: *i64=sys_mmap(16) as *i64 28 let body_fake: *u8="POST /mcp HTTP/1.1\r\nHost: x\r\n\r\nOrigin: https://untrusted.invalid\r\n" 29 gv_check_eq("body cannot supply header",edge_find_header(body_fake,ta_slen(body_fake),"Origin",6,spans),0,ctr) 30 let dup: *u8="POST /mcp HTTP/1.1\r\nOrigin: one\r\norigin: two\r\n\r\n" 31 gv_check_eq("duplicate count",edge_find_header(dup,ta_slen(dup),"Origin",6,spans),2,ctr) 32 meh_case("legacy protocol","MCP-Protocol-Version: 2024-11-05\r\n",200,ctr) 33 meh_case("march protocol","mcp-protocol-version: 2025-03-26\r\n",200,ctr) 34 meh_case("june protocol","MCP-Protocol-Version: 2025-06-18\r\n",200,ctr) 35 meh_case("unknown protocol","MCP-Protocol-Version: unknown\r\n",400,ctr) 36 meh_case("duplicate protocol","MCP-Protocol-Version: 2025-06-18\r\nmcp-protocol-version: 2025-06-18\r\n",400,ctr) 37 let get_req: *u8="GET /mcp HTTP/1.1\r\nHost: nishifamily.com\r\nAccept: text/event-stream\r\n\r\n" 38 let response: *u8=sys_mmap(EDGE_RESP_CAP) 39 let response_n: i64=nx_api_edge_handle(get_req,ta_slen(get_req),response) 40 gv_check_eq("GET stream unsupported status",gv_at(response,response_n,0,"HTTP/1.1 405 Method Not Allowed"),1,ctr) 41 gv_check_eq("GET empty response",response_n-ta_body_off(response,response_n),0,ctr) 42 gv_check_eq("GET allowed methods",edge_has(response,response_n,"Allow: POST, OPTIONS"),1,ctr) 43 return gv_verdict("MCP-EDGE-BOUNDARY",ctr,"origin, protocol header, method contract; live deployment pending") 44}