lib/reticulum/src/node/inbound.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const reticulum = @import("../root.zig");
3
4 const destination = reticulum.destination;
5 const carrier = reticulum.carrier;
6 const identity = reticulum.identity;
7 const ifac = reticulum.interface.ifac;
8 const node = reticulum.node;
9 const packet = reticulum.packet;
10 const wire = reticulum.wire;
11
12 pub fn diagnostic(node_owner: *node.Node, code: node.Code, hash: ?packet.Hash) void {
13 node_owner.effects.push(.{ .diagnostic = .{
14 .code = code,
15 .packet_hash = hash,
16 } }) catch unreachable;
17 }
18
19 fn authenticatedFrame(
20 node_owner: *node.Node,
21 registration: *const node.Interface,
22 frame: []const u8,
23 scratch: *[carrier.frame_bytes_max]u8,
24 ) ?[]const u8 {
25 const access_config = if (registration.access) |*value| value else {
26 if (frame.len > wire.mtu) {
27 diagnostic(node_owner, .frame_too_large, null);
28 return null;
29 }
30 if (frame.len > 0 and ifac.hasFlag(frame[0])) {
31 diagnostic(node_owner, .ifac_unexpected_flag, null);
32 return null;
33 }
34 return frame;
35 };
36 const frame_max = @as(usize, wire.mtu) + access_config.size.byte();
37 std.debug.assert(frame_max <= carrier.frame_bytes_max);
38 if (frame.len > frame_max) {
39 diagnostic(node_owner, .frame_too_large, null);
40 return null;
41 }
42 if (frame.len == 0 or !ifac.hasFlag(frame[0])) {
43 diagnostic(node_owner, .ifac_missing_flag, null);
44 return null;
45 }
46 const raw = ifac.strip(&access_config.key, access_config.size, frame, scratch) catch |err| {
47 switch (err) {
48 error.InvalidCode => diagnostic(node_owner, .ifac_invalid_code, null),
49 error.Truncated => diagnostic(node_owner, .ifac_truncated, null),
50 error.MissingFlag,
51 error.OutputTooSmall,
52 error.OverlappingBuffers,
53 error.PacketTooLarge,
54 => unreachable,
55 }
56 return null;
57 };
58 std.debug.assert(raw.len + access_config.size.byte() == frame.len);
59 return raw;
60 }
61
62 fn localKind(kind: wire.DestinationType) destination.Type {
63 return @fromBackingInt(@intCast(@backingInt(kind)));
64 }
65
66 fn unsupportedContext(context: wire.Context) bool {
67 const raw = context.encode();
68 return raw >= wire.Context.resource.encode() and
69 raw <= wire.Context.resource_rcl.encode();
70 }
71
72 fn candidateFor(
73 value: wire.Packet,
74 interface: carrier.Index,
75 hash: packet.Hash,
76 blob: node.transport.path.Blob,
77 ) node.transport.path.Candidate {
78 std.debug.assert(value.packet_type == .announce);
79 return .{
80 .destination = value.destination,
81 .next_hop = value.transport_id orelse value.destination,
82 .announce_hash = hash,
83 .hops = value.hops,
84 .carrier = interface,
85 .context_flag = value.context_flag,
86 .blob = blob,
87 .payload = value.payload,
88 };
89 }
90
91 fn acceptAnnounce(
92 node_owner: *node.Node,
93 value: wire.Packet,
94 interface: carrier.Index,
95 hash: packet.Hash,
96 now: node.Seconds,
97 ) node.StepError!void {
98 const validated = destination.announce.validate(
99 value.destination,
100 value.header,
101 value.context_flag,
102 value.payload,
103 ) catch {
104 diagnostic(node_owner, .invalid_announce, hash);
105 return;
106 };
107 if (node_owner.destinations.find(value.destination) != null) {
108 diagnostic(node_owner, .own_announce, hash);
109 return;
110 }
111 std.debug.assert(value.hops <= wire.pathfinder_hops);
112 const public_key = validated.public_key.toBytes();
113 if (node_owner.known_identities.recall(value.destination)) |known| {
114 if (!std.mem.eql(u8, &known.public_key, &public_key)) {
115 diagnostic(node_owner, .announce_key_changed, hash);
116 return;
117 }
118 }
119 const candidate = candidateFor(value, interface, hash, validated.random_hash[0..10].*);
120 const known = node_owner.transport.paths.find(value.destination, now);
121 const admitted = node.transport.path.admits(known, candidate, now);
122 const decision = node.transport.retransmit.plan(node_owner, value, admitted);
123 var waiting: usize = 0;
124 if (admitted) waiting = node.transport.requests.waitingFrames(node_owner, value, now);
125 const accepted: usize = @intFromBool(admitted);
126 const rebroadcast: usize = @intFromBool(decision != .none);
127 try node.outbound.reserve(node_owner, accepted + rebroadcast + waiting, accepted + waiting);
128 if (decision == .write and !node.transport.retransmit.canSchedule(node_owner)) {
129 return error.TimerFull;
130 }
131 node_owner.known_identities.remember(.{
132 .destination_hash = value.destination,
133 .public_key = public_key,
134 .announce_packet_hash = hash,
135 .received = now,
136 .app_data = validated.app_data,
137 }) catch unreachable;
138 if (validated.rotating_public_key) |rotating| node_owner.known_ratchets.remember(.{
139 .destination_hash = value.destination,
140 .public_key = rotating[0..32].*,
141 .received = now,
142 });
143 node.transport.retransmit.detect(node_owner, value, now);
144 if (!admitted) return;
145 _ = node_owner.transport.paths.learn(candidate, now);
146 node.transport.retransmit.enter(node_owner, value, decision, hash, now);
147 try node.transport.requests.answerWaiting(node_owner, value, now);
148 _ = node_owner.transport.inflight_requests.remove(value.destination);
149 node_owner.effects.push(.{ .announce_received = .{
150 .destination_hash = value.destination,
151 .identity_hash = reticulum.hash.truncated(&public_key),
152 .public_key = public_key,
153 .app_data = validated.app_data,
154 .hops = value.hops,
155 .rotating_key_present = validated.rotating_public_key != null,
156 .path_response = value.context == .path_response,
157 } }) catch unreachable;
158 }
159
160 fn decryptData(
161 node_owner: *node.Node,
162 entry: *const destination.registry.Entry,
163 ciphertext: []const u8,
164 ) ?destination.cipher.Decrypted {
165 return switch (entry.kind) {
166 .single => blk: {
167 const index = entry.identity_index orelse break :blk null;
168 const private = node_owner.identityAt(index) orelse break :blk null;
169 const binding = node_owner.ratchet_bindings[index];
170 const empty: [0]identity.Ratchet = .{};
171 const rotating = if (binding.ring) |ring| ring.all() else empty[0..];
172 break :blk destination.cipher.decrypt(.{ .single = .{
173 .private = private,
174 .ratchets = rotating,
175 .enforce_ratchets = binding.enforce,
176 } }, ciphertext, node_owner.scratch) catch null;
177 },
178 .group => destination.cipher.decrypt(.{
179 .group = .{ .key = &entry.group_key },
180 }, ciphertext, node_owner.scratch) catch null,
181 .plain => destination.cipher.decrypt(.{
182 .plain = {},
183 }, ciphertext, node_owner.scratch) catch null,
184 .link => null,
185 };
186 }
187
188 fn proofFrame(
189 node_owner: *node.Node,
190 hash: packet.Hash,
191 private: *const identity.Private,
192 ) node.StepError![]const u8 {
193 const payload_start: usize = wire.header_one_bytes;
194 const payload = packet.proof.build(
195 hash,
196 private,
197 node_owner.proof_mode == .implicit,
198 node_owner.scratch[payload_start..],
199 ) catch return error.PacketTooLarge;
200 return wire.encode(
201 packet.proof.makePacket(hash, payload),
202 node_owner.scratch,
203 ) catch return error.PacketTooLarge;
204 }
205
206 fn sendProof(
207 node_owner: *node.Node,
208 destination_hash: [16]u8,
209 hash: packet.Hash,
210 interface: carrier.Index,
211 ) node.StepError!void {
212 const entry = node_owner.destinations.find(destination_hash) orelse
213 return error.UnknownDestination;
214 if (entry.kind == .plain or entry.kind == .link) return error.InvalidDestination;
215 const index = entry.identity_index orelse return error.InvalidDestination;
216 const private = node_owner.identityAt(index) orelse return error.InvalidDestination;
217 const frame = try proofFrame(node_owner, hash, private);
218 try node.outbound.transmit(node_owner, .{ .one = interface }, frame);
219 }
220
221 fn acceptData(
222 node_owner: *node.Node,
223 value: wire.Packet,
224 interface: carrier.Index,
225 hash: packet.Hash,
226 ) node.StepError!void {
227 const entry = node_owner.destinations.find(value.destination) orelse {
228 diagnostic(node_owner, .unknown_destination, hash);
229 return;
230 };
231 if (entry.kind != localKind(value.destination_type)) {
232 diagnostic(node_owner, .destination_type_mismatch, hash);
233 return;
234 }
235 const decrypted = decryptData(node_owner, entry, value.payload) orelse {
236 diagnostic(node_owner, .decryption_failed, hash);
237 return;
238 };
239 const carriers = node.outbound.frameCount(node_owner, .{ .one = interface });
240 std.debug.assert(carriers <= 1);
241 const proves_now = entry.proof_strategy == .all and
242 entry.identity_index != null and carriers > 0;
243 const effect_count = 1 + if (proves_now) carriers else 0;
244 try node.outbound.reserve(node_owner, effect_count, effect_count);
245 node_owner.effects.push(.{ .application_delivery = .{
246 .destination = value.destination,
247 .packet_hash = hash,
248 .plaintext = decrypted.plaintext,
249 .ratchet_id = decrypted.ratchet_id,
250 .proof_requested = entry.proof_strategy == .app,
251 .interface = interface,
252 } }) catch unreachable;
253 if (proves_now) try sendProof(node_owner, value.destination, hash, interface);
254 }
255
256 fn receiptPublic(node_owner: *node.Node, hash: [16]u8) ?identity.Public {
257 if (node_owner.known_identities.recall(hash)) |known| return known.public();
258 const entry = node_owner.destinations.find(hash) orelse return null;
259 if (entry.kind == .plain or entry.kind == .link) return null;
260 const index = entry.identity_index orelse return null;
261 const private = node_owner.identityAt(index) orelse return null;
262 return private.public();
263 }
264
265 fn validateReceipt(
266 node_owner: *node.Node,
267 receipt: *packet.receipt.Receipt,
268 proof: wire.proof.Proof,
269 now: node.Seconds,
270 ) bool {
271 if (receipt.status != .sent) return false;
272 var public = receiptPublic(node_owner, receipt.destination) orelse return false;
273 defer public.zero();
274 return receipt.validateProof(proof, &public, now);
275 }
276
277 /// Marks one receipt delivered and gives its round trip time, counted in seconds from the moment
278 /// the packet left to the current instant, as Reticulum@1.5.0 RNS/Packet.py:439-461,530-536
279 /// concludes it, so a caller learns that the bytes it sent arrived and how long the round trip
280 /// took. The call then releases the receipt together with its timer. The caller has already found
281 /// the receipt, and a hash with no receipt behind it is unreachable here.
282 pub fn concludeReceipt(node_owner: *node.Node, hash: packet.Hash, now: node.Seconds) void {
283 const receipt = node_owner.receipts.find(hash) orelse unreachable;
284 const rtt = now -| receipt.sent_at;
285 _ = node_owner.timers.cancel(.{ .receipt = hash });
286 _ = node_owner.receipts.remove(hash);
287 node_owner.effects.push(.{ .receipt_update = .{
288 .packet_hash = hash,
289 .status = .delivered,
290 .rtt = rtt,
291 } }) catch unreachable;
292 }
293
294 fn acceptProof(
295 node_owner: *node.Node,
296 value: wire.Packet,
297 raw: []const u8,
298 interface: carrier.Index,
299 frame_hash: packet.Hash,
300 now: node.Seconds,
301 ) node.StepError!void {
302 const relayed = try node.transport.relay.proof(
303 node_owner,
304 value,
305 raw,
306 interface,
307 frame_hash,
308 now,
309 );
310 try node.outbound.reserve(node_owner, 1, 0);
311 const decoded = wire.proof.decode(value.payload) catch {
312 if (!relayed) diagnostic(node_owner, .proof_rejected, frame_hash);
313 return;
314 };
315 switch (decoded) {
316 .explicit => |explicit| {
317 const receipt = node_owner.receipts.find(explicit.packet_hash) orelse {
318 if (!relayed) diagnostic(node_owner, .proof_rejected, frame_hash);
319 return;
320 };
321 if (!validateReceipt(node_owner, receipt, decoded, now)) {
322 if (!relayed) diagnostic(node_owner, .proof_rejected, frame_hash);
323 return;
324 }
325 concludeReceipt(node_owner, explicit.packet_hash, now);
326 },
327 .implicit => {
328 for (0..node_owner.receipts.capacity.receipts_max) |index| {
329 if (index == node_owner.receipts.len) break;
330 const receipt = &node_owner.receipts.entries[index];
331 const hash = receipt.hash;
332 if (!validateReceipt(node_owner, receipt, decoded, now)) continue;
333 concludeReceipt(node_owner, hash, now);
334 return;
335 }
336 if (!relayed) diagnostic(node_owner, .proof_rejected, frame_hash);
337 },
338 }
339 }
340
341 /// Handles one frame that arrived on a carrier, from authenticating it through to delivering,
342 /// relaying, or refusing the packet inside, so every byte the node hears from the network passes
343 /// through this call. A carrier index the caller has yet to register reports
344 /// `unregistered_interface`. The carrier's access code is checked and stripped, as Reticulum@1.5.0
345 /// RNS/Transport.py:1636-1701 authenticates carrier frames, and a frame that is too large, that
346 /// carries the wrong flag, or whose code fails reports which of those happened. Bytes that do not
347 /// decode as a packet report `malformed_packet`. The packet then passes the admission rules of
348 /// Reticulum@1.5.0 RNS/Transport.py:1700-1716,1868-1872: one the node has already handled reports
349 /// `duplicate_packet`, and any other refusal reports `packet_filtered`. The hop count rises by one
350 /// before anything else reads it. A path request goes to the path request code. Link traffic goes
351 /// to the link code, which says whether it handled the packet. A context this node carries no
352 /// support for reports `unsupported_context`. A transport node relays a packet that names it as the
353 /// next hop, as Reticulum@1.5.0 RNS/Transport.py:1907-2028 relays HEADER_2 packets, and one with no
354 /// path reports `no_path`. An announce is validated and its path learned, as Reticulum@1.5.0
355 /// RNS/Transport.py:2083-2135,2137-2213,2374-2375 does, and a transport node rebroadcasts it as
356 /// Reticulum@1.5.0 RNS/Transport.py:2401-2455 publishes it. Data addressed to one of the node's own
357 /// destinations is decrypted and delivered, and proved at once under the `.all` strategy, as
358 /// Reticulum@1.5.0 RNS/Transport.py:2490-2533 delivers and proves it. A proof is matched against an
359 /// outstanding receipt, as Reticulum@1.5.0 RNS/Transport.py:2658-2697 validates delivery proofs,
360 /// and one that matches none reports `proof_rejected`. Packets bound for endpoint links reach them
361 /// as Reticulum@1.5.0 RNS/Transport.py:2456-2487,2490-2516,2600-2650 hands them over.
362 pub fn run(node_owner: *node.Node, frame: node.CarrierFrame) node.StepError!void {
363 const interface_index: usize = frame.interface;
364 if (interface_index >= node_owner.interfaces.len or
365 node_owner.interfaces[interface_index].registered == 0)
366 {
367 diagnostic(node_owner, .unregistered_interface, null);
368 return;
369 }
370 var access_scratch: [carrier.frame_bytes_max]u8 = undefined;
371 const raw = authenticatedFrame(
372 node_owner,
373 &node_owner.interfaces[interface_index],
374 frame.bytes,
375 &access_scratch,
376 ) orelse return;
377 var value = wire.decode(raw) catch {
378 diagnostic(node_owner, .malformed_packet, null);
379 return;
380 };
381 const hash = wire.hash.full(raw) catch unreachable;
382 const seen = node_owner.duplicate_hashes.contains(hash);
383 if (packet.filter.admit(&value, seen, node_owner.transport.identity_hash) == .reject) {
384 diagnostic(node_owner, if (seen) .duplicate_packet else .packet_filtered, hash);
385 return;
386 }
387 value.hops += 1;
388 if (node.transport.requests.addressed(value)) {
389 return node.transport.requests.receive(node_owner, value, frame.interface, hash, frame.now);
390 }
391 const relayed_link = node.transport.link.relay.find(node_owner, value.destination) != null;
392 const lrproof = value.packet_type == .proof and value.context == .lrproof;
393 if (!relayed_link and !lrproof) _ = node_owner.duplicate_hashes.insert(hash);
394 if (try node.link.receive(node_owner, value, raw, frame, hash)) return;
395 if (unsupportedContext(value.context)) {
396 diagnostic(node_owner, .unsupported_context, hash);
397 return;
398 }
399 if (node.transport.relay.applies(node_owner, value)) {
400 const outcome = try node.transport.relay.run(
401 node_owner,
402 value,
403 raw,
404 frame.interface,
405 hash,
406 frame.now,
407 );
408 if (outcome == .relayed) return;
409 diagnostic(node_owner, .no_path, hash);
410 if (value.packet_type == .data) return;
411 }
412 switch (value.packet_type) {
413 .announce => try acceptAnnounce(node_owner, value, frame.interface, hash, frame.now),
414 .data => try acceptData(node_owner, value, frame.interface, hash),
415 .proof => try acceptProof(node_owner, value, raw, frame.interface, hash, frame.now),
416 .link_request => unreachable,
417 }
418 }
419
420 /// Sends a proof of one packet the node delivered earlier, which Reticulum@1.5.0
421 /// RNS/Transport.py:2527-2530 leaves to the application, so an application under the `.app`
422 /// strategy answers a delivery and the sender learns its packet arrived. The proof goes out on the
423 /// carrier the packet arrived on and on no other, as Reticulum@1.5.0 RNS/Identity.py:943-954
424 /// proves. A packet hash the node has already forgotten returns `error.ProofUnavailable`. A carrier
425 /// that carries no outgoing traffic returns `error.NoOutgoingCarrier`.
426 pub fn prove(node_owner: *node.Node, value: node.ApplicationProve) node.StepError!void {
427 if (!node_owner.duplicate_hashes.contains(value.packet_hash)) return error.ProofUnavailable;
428 const carriers = node.outbound.frameCount(node_owner, .{ .one = value.interface });
429 if (carriers == 0) return error.NoOutgoingCarrier;
430 try node.outbound.reserve(node_owner, carriers, carriers);
431 try sendProof(node_owner, value.destination, value.packet_hash, value.interface);
432 }
433
434 /// Handles one timer that came due, by what the timer names, so every deadline the node armed comes
435 /// back through here because the caller owns the clock. The duplicate-hash timer rotates that table
436 /// to its next generation. The announce timer sweeps the rebroadcast queue, a link timer runs that
437 /// link's watchdog, and the relayed-link timer runs the one a transport node keeps. A receipt timer
438 /// checks its deadline, as Reticulum@1.5.0 RNS/Packet.py:540-548 concludes receipt deadlines: a
439 /// receipt past its timeout is reported failed and dropped, and one still waiting is armed again
440 /// one second after its deadline. Arming it again with no free timer returns `error.TimerFull`, and
441 /// with no free effect slot returns `error.EffectsFull`.
442 pub fn timerExpired(node_owner: *node.Node, value: node.TimerExpired) node.StepError!void {
443 switch (value.id) {
444 .hashlist => {
445 _ = node_owner.timers.cancel(.hashlist);
446 node_owner.duplicate_hashes.rotate();
447 },
448 .announces => try node.transport.retransmit.sweep(node_owner, value.now),
449 .link => |link_id| try node.link.timerExpired(node_owner, link_id, value),
450 .link_entries => try node.transport.link.relay.timerExpired(node_owner, value),
451 .receipt => |hash| {
452 _ = node_owner.timers.cancel(value.id);
453 const receipt = node_owner.receipts.find(hash) orelse return;
454 receipt.checkTimeout(value.now);
455 if (receipt.status == .failed) {
456 try node.outbound.reserve(node_owner, 1, 0);
457 _ = node_owner.receipts.remove(hash);
458 node_owner.effects.push(.{ .receipt_update = .{
459 .packet_hash = hash,
460 .status = .failed,
461 .rtt = null,
462 } }) catch unreachable;
463 return;
464 }
465 const deadline = receipt.sent_at +| receipt.timeout +| 1;
466 node_owner.timers.schedule(value.id, deadline) catch return error.TimerFull;
467 node_owner.effects.push(.{ .schedule_timer = .{
468 .id = value.id,
469 .at = deadline,
470 } }) catch return error.EffectsFull;
471 },
472 }
473 }