nx_dht_announce.nx source
↩ module page · 139 lines · 9803 B
1// nx_dht_announce.nx -- SOVEREIGN DHT announce_peer (BEP-5): make OUR seeder DISCOVERABLE for an info_hash.
2//
3// The seeder (nx_torrent_seed) serves pieces, but a stranger can only reach it if the DHT knows we hold the
4// info_hash. This organ does the announce half: walk get_peers toward the info_hash (Kademlia BFS) capturing
5// each responding node's (ip, port, token) -- the token proves we queried that node -- then blast announce_peer
6// (ih, our_port, token) to those nodes. Each node that ACKs has STORED "peer <us> holds <ih>", so a later
7// get_peers(ih) reaching it returns us in its values list. Composes nx_dht (KRPC) + nx_dns_a + raw UDP.
8// v1 = BFS toward the hash (no strict XOR-closest sort yet; announce to reached responders) -- honest label.
9//
10// nx_dht_announce <ih_hex40> <our_port> -> announce we hold ih at our_port; prints #nodes acked
11// nx_dht_announce -> LIVE gate: announce the demo info_hash, acks>0 = GREEN
12// license_tier: ORIGINAL layer: peer-discovery (announce) module: nishi-core.torrent.dht_announce
13// depends: nishi-core.torrent.dht, nishi-core.dns_a
14import "nx_dht.nx"
15import "nx_dns_a.nx"
16const K_MAGIC_6881: i64 = 6881
17const K_MAGIC_25401: i64 = 25401
18const K_MAGIC_8192: i64 = 8192
19const K_MAGIC_4096: i64 = 4096
20
21func da_p(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
22// Digit scratch in module `static`s -- allocated ONCE for the process. The old per-call sys_mmap
23// pair leaked 8192 B per NUMBER PRINTED (sys_mmap is page-granular, no free): this lib runs inside
24// nx_seed_announce_all, the supervised daemon nx_resmon measured at 8.8 GB committed -- the leak
25// was in the LOGGING, not the work. Measured by nx_mmapleak_gate (laptop tree): 2000 unpaired
26// calls grow a process 4000 pages; static scratch grows it 2 pages TOTAL.
27static DAN_NT: i64
28static DAN_NO: i64
29func da_n(v: i64) -> i64 { if DAN_NT == 0 { DAN_NT = sys_mmap(64) as i64; DAN_NO = sys_mmap(64) as i64 } let t: *u8=DAN_NT as *u8; let o: *u8=DAN_NO as *u8; var m: i64=v; var k: i64=0; if m==0 {t[0]=48 as u8;k=1} while m>0 {t[k]=(48+(m%10)) as u8; m=m/10; k=k+1} var i: i64=0; while i<k {o[i]=t[k-1-i];i=i+1} sys_write(1,o,k); return 0 }
30// byte-exact copy of the PROVEN dgp_sockaddr (octets passed as args -- the packed-ipv variant misbehaved).
31func da_sock4(out: *u8, port: i64, a: i64, b: i64, c: i64, d: i64) -> i64 {
32 out[0]=2 as u8; out[1]=0 as u8; out[2]=((port>>8)&0xff) as u8; out[3]=(port&0xff) as u8
33 out[4]=a as u8; out[5]=b as u8; out[6]=c as u8; out[7]=d as u8
34 var i: i64=8; while i<16 { out[i]=0 as u8; i=i+1 }
35 return 16
36}
37func da_hexval(c: i64) -> i64 { if c>=48 { if c<=57 { return c-48 } } if c>=97 { if c<=102 { return c-87 } } if c>=65 { if c<=70 { return c-55 } } return 0 }
38func da_hex2bin(hex: *u8, out: *u8) -> i64 { var i: i64=0; while i<20 { let h: i64=hex[i*2] as i64; let l: i64=hex[i*2+1] as i64; if h==0 { return 0 } if l==0 { return 0 } out[i]=((da_hexval(h)*16)+da_hexval(l)) as u8; i=i+1 } return 1 }
39
40// Phase 1: BFS walk get_peers toward ih to converge a frontier of nodes NEAR the hash (parallel, null src).
41// Phase 2: SERIAL per frontier node -- get_peers(node) -> token (response is 1:1 with the node we just queried,
42// so no src capture needed) -> announce_peer(ih, our_port, token) -> count the ACK. Returns #nodes acked.
43func dan_announce(ih: *u8, nid: *u8, our_port: i64) -> i64 {
44 let cip: *i64 = sys_mmap(8*512) as *i64; let cpo: *i64 = sys_mmap(8*512) as *i64; var cn: i64 = 0
45 let xip: *i64 = sys_mmap(8*512) as *i64; let xpo: *i64 = sys_mmap(8*512) as *i64
46 let b: *u8 = sys_mmap(4)
47 if nx_dns_a_resolve("dht.transmissionbt.com" as *u8, b) == 1 { cip[cn]=((b[0] as i64)<<24)|((b[1] as i64)<<16)|((b[2] as i64)<<8)|(b[3] as i64); cpo[cn]=K_MAGIC_6881; cn=cn+1 }
48 if nx_dns_a_resolve("router.bittorrent.com" as *u8, b) == 1 { cip[cn]=((b[0] as i64)<<24)|((b[1] as i64)<<16)|((b[2] as i64)<<8)|(b[3] as i64); cpo[cn]=K_MAGIC_6881; cn=cn+1 }
49 if nx_dns_a_resolve("router.utorrent.com" as *u8, b) == 1 { cip[cn]=((b[0] as i64)<<24)|((b[1] as i64)<<16)|((b[2] as i64)<<8)|(b[3] as i64); cpo[cn]=K_MAGIC_6881; cn=cn+1 }
50 if nx_dns_a_resolve("dht.libtorrent.org" as *u8, b) == 1 { cip[cn]=((b[0] as i64)<<24)|((b[1] as i64)<<16)|((b[2] as i64)<<8)|(b[3] as i64); cpo[cn]=K_MAGIC_25401; cn=cn+1 }
51 let bip: *i64 = sys_mmap(8*8) as *i64; let bpo: *i64 = sys_mmap(8*8) as *i64; let bn: i64 = cn
52 var bb: i64 = 0; while bb < cn { bip[bb]=cip[bb]; bpo[bb]=cpo[bb]; bb=bb+1 }
53 let tx: *u8 = sys_mmap(2); tx[0]=97 as u8; tx[1]=97 as u8
54 let q: *u8 = sys_mmap(128); let r: *u8 = sys_mmap(K_MAGIC_8192); let dest: *u8 = sys_mmap(16)
55 let tnip: *i64 = sys_mmap(8*256) as *i64; let tnpo: *i64 = sys_mmap(8*256) as *i64
56 let QPR: i64 = 96
57 var round: i64 = 0
58 while round < 4 {
59 if cn == 0 { round = 4 } else {
60 let fd: i64 = sys_socket(AF_INET, SOCK_DGRAM, 0)
61 if fd < 0 { round = 4 } else {
62 sys_set_socket_timeout(fd, 3)
63 var sent: i64 = 0; var i: i64 = 0
64 while i < cn { if i < QPR {
65 da_sock4(dest, cpo[i], (cip[i]>>24)&0xff, (cip[i]>>16)&0xff, (cip[i]>>8)&0xff, cip[i]&0xff); let ql: i64 = dht_build_get_peers(nid, ih, tx, q)
66 sys_sendto(fd, q, ql, 0, dest, 16); sent = sent + 1
67 } i = i + 1 }
68 var xn: i64 = 0; var resp: i64 = 0; var go: i64 = 1
69 while go == 1 {
70 let n: i64 = sys_recvfrom(fd, r, K_MAGIC_8192, 0, 0 as *u8, 0 as *i64)
71 if n < 1 { go = 0 } else {
72 resp = resp + 1
73 var nn: i64 = dht_parse_nodes(r, n, tnip, tnpo, 256); if nn < 0 { nn = 0 }
74 var m: i64 = 0; while m < nn { if xn < 512 { xip[xn]=tnip[m]; xpo[xn]=tnpo[m]; xn=xn+1 } m=m+1 }
75 if resp >= sent { go = 0 }
76 }
77 }
78 sys_close(fd)
79 var ba: i64 = 0; while ba < bn { if xn < 512 { xip[xn]=bip[ba]; xpo[xn]=bpo[ba]; xn=xn+1 } ba=ba+1 }
80 da_p(" round " as *u8); da_n(round); da_p(": get_peers sent=" as *u8); da_n(sent); da_p(" replies=" as *u8); da_n(resp); da_p(" frontier=" as *u8); da_n(xn); da_p("\n" as *u8)
81 var c: i64 = 0; while c < xn { cip[c]=xip[c]; cpo[c]=xpo[c]; c=c+1 } cn = xn
82 round = round + 1
83 }
84 }
85 }
86 // ---- Phase 2: SERIAL announce to frontier nodes (get_peers -> token -> announce_peer -> ack) ----
87 if cn == 0 { da_p(" no frontier nodes -> cannot announce (DHT unreachable?)\n" as *u8); return 0 }
88 let afd: i64 = sys_socket(AF_INET, SOCK_DGRAM, 0); if afd < 0 { return 0 }
89 sys_set_socket_timeout(afd, 2)
90 let atx: *u8 = sys_mmap(2); atx[0]=98 as u8; atx[1]=98 as u8 // "bb"
91 let aq: *u8 = sys_mmap(160); let arb: *u8 = sys_mmap(K_MAGIC_4096); let tkb: *u8 = sys_mmap(64)
92 var acked: i64 = 0; var tried: i64 = 0; var attempts: i64 = 0; var i: i64 = 0
93 while i < cn {
94 if attempts >= 28 { i = cn } else { if acked >= 8 { i = cn } else {
95 da_sock4(dest, cpo[i], (cip[i]>>24)&0xff, (cip[i]>>16)&0xff, (cip[i]>>8)&0xff, cip[i]&0xff)
96 let ql: i64 = dht_build_get_peers(nid, ih, tx, q)
97 sys_sendto(afd, q, ql, 0, dest, 16)
98 let n: i64 = sys_recvfrom(afd, r, K_MAGIC_8192, 0, 0 as *u8, 0 as *i64)
99 attempts = attempts + 1
100 if n > 0 {
101 let tl: i64 = dht_parse_token(r, n, tkb, 48)
102 if tl > 0 {
103 tried = tried + 1
104 let al: i64 = dht_build_announce_peer(nid, ih, our_port, tkb, tl, atx, aq)
105 sys_sendto(afd, aq, al, 0, dest, 16)
106 let an: i64 = sys_recvfrom(afd, arb, K_MAGIC_4096, 0, 0 as *u8, 0 as *i64)
107 if an > 0 { if dht_is_response(arb, an) == 1 { acked = acked + 1 } }
108 }
109 }
110 i = i + 1
111 } }
112 }
113 sys_close(afd)
114 da_p(" announce_peer: nodes_tried=" as *u8); da_n(tried); da_p(" acked=" as *u8); da_n(acked); da_p("\n" as *u8)
115 return acked
116}
117
118func main(argc: i64, argv: *i64) -> i64 {
119 let nid: *u8 = sys_mmap(20); var i: i64 = 0; while i < 20 { nid[i] = (0x41 + i) as u8; i = i + 1 } // node id (0x41+i, as the proven get_peers walk uses)
120 let ih: *u8 = sys_mmap(20)
121 var our_port: i64 = K_MAGIC_6881
122 if argc >= 3 {
123 if da_hex2bin(argv[1] as *u8, ih) != 1 { da_p("bad info_hash hex (need 40 chars)\n" as *u8); sys_exit(2); return 2 }
124 our_port = 0; var pp: i64=0; let ps: *u8 = argv[2] as *u8
125 while ps[pp]!=(0 as u8) { let c: i64=ps[pp] as i64; if c>=48 { if c<=57 { our_port=our_port*10+(c-48) } } pp=pp+1 }
126 da_p("DHT announce: ih from arg, our_port=" as *u8); da_n(our_port); da_p("\n" as *u8)
127 let a: i64 = dan_announce(ih, nid, our_port)
128 if a > 0 { da_p("DHT-ANNOUNCE acked=" as *u8); da_n(a); da_p(" verdict=GREEN\n" as *u8); sys_exit(0); return 0 }
129 da_p("DHT-ANNOUNCE acked=0 verdict=RED\n" as *u8); sys_exit(1); return 1
130 }
131 // LIVE gate: announce the demo info_hash (70aed86c...) at :6881; acks>0 proves the announce works live.
132 let hx: *u8 = "70aed86cc665e340397a317fd0ea37848de90c66" as *u8
133 da_hex2bin(hx, ih)
134 da_p("DHT-ANNOUNCE live gate: announcing demo ih 70aed86c... at :6881\n" as *u8)
135 let a: i64 = dan_announce(ih, nid, K_MAGIC_6881)
136 da_p("DHT-ANNOUNCE authored=organ acked=" as *u8); da_n(a)
137 if a > 0 { da_p(" verdict=GREEN (live DHT accepted our announce_peer)\n" as *u8); sys_exit(0); return 0 }
138 da_p(" verdict=RED (no node acked -- DHT unreachable from here?)\n" as *u8); sys_exit(1); return 1
139}