lib/reticulum/src/node/outbound.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 /// Floor for every receipt deadline the node sets: six seconds, the wait the node allows for the
13 /// first hop of a packet it sent, which Reticulum@1.5.0 RNS/Reticulum.py:142 fixes. A receipt's
14 /// deadline adds six more seconds for each hop past the first.
15 pub const first_hop_timeout: node.Seconds = 6;
16
17 const Target = union(destination.Type) {
18 single: identity.KeyBytes,
19 group: *const [64]u8,
20 plain: void,
21 link: void,
22 };
23
24 fn target(node_owner: *node.Node, hash: [16]u8) ?Target {
25 if (node_owner.destinations.find(hash)) |entry| {
26 return switch (entry.kind) {
27 .single => if (entry.identity_index) |index|
28 if (node_owner.identityAt(index)) |private|
29 .{ .single = private.publicBytes() }
30 else
31 null
32 else
33 null,
34 .group => .{ .group = &entry.group_key },
35 .plain => .{ .plain = {} },
36 .link => .{ .link = {} },
37 };
38 }
39 const known = node_owner.known_identities.recall(hash) orelse return null;
40 return .{ .single = known.public_key };
41 }
42
43 fn packetDestinationType(value: Target) wire.DestinationType {
44 return switch (value) {
45 .single => .single,
46 .group => .group,
47 .plain => .plain,
48 .link => .link,
49 };
50 }
51
52 fn plaintextContext(context: wire.Context) bool {
53 const raw = context.encode();
54 if (raw >= wire.Context.resource.encode() and raw <= wire.Context.resource_rcl.encode()) {
55 return true;
56 }
57 return context == .keepalive or context == .cache_request;
58 }
59
60 fn encryptSingle(
61 node_owner: *node.Node,
62 send: node.ApplicationSend,
63 public_bytes: identity.KeyBytes,
64 out: []u8,
65 ) node.StepError![]u8 {
66 var public = identity.Public.fromBytes(public_bytes);
67 defer public.zero();
68 var rotating = node_owner.known_ratchets.get(send.destination, send.now);
69 const rotating_pointer = if (rotating) |*value| value else null;
70 const encrypted = destination.cipher.encrypt(.{ .single = .{
71 .public = &public,
72 .ratchet_public = rotating_pointer,
73 .ephemeral_private = &send.ephemeral_private,
74 .iv = send.iv,
75 } }, send.plaintext, out) catch return error.EncryptionFailed;
76 return encrypted.ciphertext;
77 }
78
79 fn encryptPayload(
80 node_owner: *node.Node,
81 send: node.ApplicationSend,
82 resolved: Target,
83 out: []u8,
84 ) node.StepError![]u8 {
85 if (plaintextContext(send.context)) {
86 const plain = destination.cipher.encrypt(.{ .plain = {} }, send.plaintext, out) catch
87 return error.PacketTooLarge;
88 return plain.ciphertext;
89 }
90 return switch (resolved) {
91 .single => |public| encryptSingle(node_owner, send, public, out),
92 .group => |key| blk: {
93 const encrypted = destination.cipher.encrypt(.{ .group = .{
94 .key = key,
95 .iv = send.iv,
96 } }, send.plaintext, out) catch return error.EncryptionFailed;
97 break :blk encrypted.ciphertext;
98 },
99 .plain => blk: {
100 const plain = destination.cipher.encrypt(.{ .plain = {} }, send.plaintext, out) catch
101 return error.PacketTooLarge;
102 break :blk plain.ciphertext;
103 },
104 .link => error.InvalidDestination,
105 };
106 }
107
108 fn encodeFrame(
109 node_owner: *node.Node,
110 send: node.ApplicationSend,
111 resolved: Target,
112 ) node.StepError![]const u8 {
113 const payload_start: usize = wire.header_one_bytes;
114 const payload = try encryptPayload(
115 node_owner,
116 send,
117 resolved,
118 node_owner.scratch[payload_start..],
119 );
120 const value = wire.Packet{
121 .ifac = 0,
122 .header = .one,
123 .context_flag = 0,
124 .transport = .broadcast,
125 .destination_type = packetDestinationType(resolved),
126 .packet_type = .data,
127 .hops = send.hops,
128 .transport_id = null,
129 .destination = send.destination,
130 .context = send.context,
131 .payload = payload,
132 };
133 return wire.encode(value, node_owner.scratch) catch |err| switch (err) {
134 error.InvalidHops => error.InvalidHops,
135 else => error.PacketTooLarge,
136 };
137 }
138
139 fn createsReceipt(send: node.ApplicationSend, resolved: Target) bool {
140 if (!send.create_receipt) return false;
141 switch (resolved) {
142 .plain, .link => return false,
143 .single, .group => {},
144 }
145 const context = send.context.encode();
146 if (context >= wire.Context.resource.encode() and
147 context <= wire.Context.resource_rcl.encode()) return false;
148 if (context >= wire.Context.keepalive.encode() and
149 context <= wire.Context.lrproof.encode()) return false;
150 return true;
151 }
152
153 /// Names the reach of one transmission for every send, so the node counts carriers from it: a
154 /// single carrier given by name, the whole set of outgoing carriers, or that set less one member.
155 pub const Fanout = union(enum) {
156 one: carrier.Index,
157 all: void,
158 all_except: carrier.Index,
159 };
160
161 fn selects(node_owner: *const node.Node, fanout: Fanout, index: usize) bool {
162 std.debug.assert(index < node_owner.interfaces.len);
163 const registration = node_owner.interfaces[index];
164 if (registration.registered == 0) return false;
165 if (registration.outgoing == 0) return false;
166 return switch (fanout) {
167 .one => |selected| index == selected,
168 .all => true,
169 .all_except => |excluded| index != excluded,
170 };
171 }
172
173 /// Counts the carriers one fanout reaches, so a step can reserve room before it appends frames. A
174 /// carrier counts only when the caller registered it and allowed sending on it.
175 pub fn frameCount(node_owner: *const node.Node, fanout: Fanout) usize {
176 var count: usize = 0;
177 for (0..node_owner.interfaces.len) |index| {
178 if (selects(node_owner, fanout, index)) count += 1;
179 }
180 std.debug.assert(count <= node_owner.interfaces.len);
181 return count;
182 }
183
184 pub fn reserve(node_owner: *node.Node, effects: usize, frames: usize) node.StepError!void {
185 std.debug.assert(effects <= node.effects_per_event_max);
186 std.debug.assert(frames <= node.effect_frames_per_event_max);
187 if (effects > node_owner.effects.capacity.effects_max - node_owner.effects.len) {
188 return error.EffectsFull;
189 }
190 if (frames > node_owner.effects.capacity.frames_max - node_owner.effects.frames_used) {
191 return error.EffectsFull;
192 }
193 }
194
195 fn protect(
196 registration: *const node.Interface,
197 frame: []const u8,
198 protected: *[carrier.frame_bytes_max]u8,
199 ) []const u8 {
200 std.debug.assert(frame.len >= 2);
201 std.debug.assert(frame.len <= wire.mtu);
202 const access_config = if (registration.access) |*value| value else return frame;
203 const outgoing = ifac.apply(&access_config.key, access_config.size, frame, protected) catch
204 unreachable;
205 std.debug.assert(outgoing.len == frame.len + access_config.size.byte());
206 return outgoing;
207 }
208
209 /// Turns a built frame into work for the caller to do by appending one carrier send for each
210 /// carrier the fanout reaches. Each frame leaves carrying that carrier's access code, which
211 /// Reticulum@1.5.0 RNS/Transport.py:1244-1278 applies to every transmitted frame. No room left in
212 /// the effect list returns `error.EffectsFull`.
213 pub fn transmit(node_owner: *node.Node, fanout: Fanout, frame: []const u8) node.StepError!void {
214 var protected: [carrier.frame_bytes_max]u8 = undefined;
215 for (node_owner.interfaces, 0..) |*registration, index| {
216 if (!selects(node_owner, fanout, index)) continue;
217 node_owner.effects.push(.{ .carrier_send = .{
218 .interface = @intCast(index),
219 .frame = protect(registration, frame, &protected),
220 } }) catch return error.EffectsFull;
221 }
222 }
223
224 /// Keeps one packet the node has sent, so the node can conclude its delivery later, and arms a
225 /// timer one second past the timeout the caller gave, as Reticulum@1.5.0 RNS/Transport.py:1307-1318
226 /// and Reticulum@1.5.0 RNS/Packet.py:540-548 time it. A full receipt table drops its oldest entry
227 /// and cancels that entry's timer. The node reports that dropped receipt to the application as a
228 /// `receipt_update` effect with status `culled` and no round trip time. No free timer returns
229 /// `error.TimerFull`, and no free effect slot returns `error.EffectsFull`.
230 pub fn insertReceipt(
231 node_owner: *node.Node,
232 destination_hash: [16]u8,
233 hash: packet.Hash,
234 now: node.Seconds,
235 timeout: node.Seconds,
236 ) node.StepError!void {
237 const timer_id = node.TimerId{ .receipt = hash };
238 const candidate = node_owner.receipts.cullCandidate(hash);
239 const cancel_id: ?node.TimerId = if (candidate) |value|
240 .{ .receipt = value.hash }
241 else
242 null;
243 if (!node_owner.timers.canScheduleAfterCancel(timer_id, cancel_id)) return error.TimerFull;
244 const value = packet.receipt.Receipt{
245 .hash = hash,
246 .truncated = hash[0..16].*,
247 .destination = destination_hash,
248 .sent_at = now,
249 .timeout = timeout,
250 };
251 if (node_owner.receipts.insert(value)) |culled| {
252 std.debug.assert(culled.status == .culled);
253 std.debug.assert(std.mem.eql(u8, &culled.hash, &candidate.?.hash));
254 _ = node_owner.timers.cancel(.{ .receipt = culled.hash });
255 node_owner.effects.push(.{ .receipt_update = .{
256 .packet_hash = culled.hash,
257 .status = .culled,
258 .rtt = null,
259 } }) catch return error.EffectsFull;
260 }
261 const deadline = now +| timeout +| 1;
262 node_owner.timers.schedule(timer_id, deadline) catch return error.TimerFull;
263 node_owner.effects.push(.{ .schedule_timer = .{
264 .id = timer_id,
265 .at = deadline,
266 } }) catch return error.EffectsFull;
267 }
268
269 fn route(
270 node_owner: *node.Node,
271 send: node.ApplicationSend,
272 resolved: Target,
273 ) ?*node.transport.path.Entry {
274 if (resolved != .single) return null;
275 return node_owner.transport.paths.find(send.destination, send.now);
276 }
277
278 /// Sends one application packet to a destination, path or no path, choosing its carriers from what
279 /// the node knows about reaching it, as Reticulum@1.5.0 RNS/Transport.py:1302-1402,1536-1552 routes
280 /// and sends a local packet. A hop count at or past the pathfinder hop maximum returns
281 /// `error.InvalidHops`. A destination the node has neither registered nor learned from an announce
282 /// returns `error.UnknownDestination`, and a link destination returns `error.InvalidDestination`.
283 /// With a known path the packet goes out on that path's carrier, and with none it goes out on every
284 /// outgoing carrier. A path longer than one hop gains a transport header naming the next hop. The
285 /// payload is encrypted to the destination's public key, to a group key, or left plain, by the kind
286 /// of destination and the packet's context. A receipt is created for single and group destinations
287 /// outside the resource and link contexts, with a deadline of six seconds plus six for each hop.
288 /// Without a path the node stores the packet's hash, so the copy that comes back is a duplicate. A
289 /// fanout that reaches no carrier returns `error.NoOutgoingCarrier`, and a packet that will not fit
290 /// returns `error.PacketTooLarge`.
291 pub fn run(node_owner: *node.Node, send: node.ApplicationSend) node.StepError!void {
292 if (send.hops >= wire.pathfinder_hops) return error.InvalidHops;
293 const resolved = target(node_owner, send.destination) orelse return error.UnknownDestination;
294 switch (resolved) {
295 .link => return error.InvalidDestination,
296 .single, .group, .plain => {},
297 }
298 const path = route(node_owner, send, resolved);
299 const fanout: Fanout = if (path) |entry| .{ .one = entry.carrier } else .all;
300 const carriers = frameCount(node_owner, fanout);
301 if (carriers == 0) return error.NoOutgoingCarrier;
302 const raw = try encodeFrame(node_owner, send, resolved);
303 const hash = wire.hash.full(raw) catch unreachable;
304 const inserts = if (path) |entry| entry.hops > 1 else false;
305 const frame = if (inserts)
306 try node.transport.rewrite.insert(raw, path.?.next_hop, node_owner.scratch)
307 else
308 raw;
309 const with_receipt = createsReceipt(send, resolved);
310 const culls = with_receipt and node_owner.receipts.cullCandidate(hash) != null;
311 try reserve(node_owner, carriers + @intFromBool(with_receipt) + @intFromBool(culls), carriers);
312 const hops = if (path) |entry| entry.hops else wire.pathfinder_hops;
313 if (with_receipt) {
314 const timeout = packet.receipt.timeoutFor(first_hop_timeout, hops);
315 try insertReceipt(node_owner, send.destination, hash, send.now, timeout);
316 }
317 if (path == null) _ = node_owner.duplicate_hashes.insert(hash);
318 if (inserts) path.?.timestamp = send.now;
319 try transmit(node_owner, fanout, frame);
320 }