finish backchannel message delivery
[oweals/gnunet.git] / src / transport / gnunet-service-tng.c
1 /*
2  This file is part of GNUnet.
3  Copyright (C) 2010-2016, 2018, 2019 GNUnet e.V.
4
5  GNUnet is free software: you can redistribute it and/or modify it
6  under the terms of the GNU Affero General Public License as published
7  by the Free Software Foundation, either version 3 of the License,
8  or (at your option) any later version.
9
10  GNUnet is distributed in the hope that it will be useful, but
11  WITHOUT ANY WARRANTY; without even the implied warranty of
12  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  Affero General Public License for more details.
14
15  You should have received a copy of the GNU Affero General Public License
16  along with this program.  If not, see <http://www.gnu.org/licenses/>.
17
18      SPDX-License-Identifier: AGPL3.0-or-later
19  */
20 /**
21  * @file transport/gnunet-service-tng.c
22  * @brief main for gnunet-service-tng
23  * @author Christian Grothoff
24  *
25  * TODO:
26  * Implement next:
27  * - track RTT, distance, loss, etc. => requires extra data structures!
28  * - proper use/initialization of timestamps in messages exchanged
29  *   during DV learning
30  * - persistence of monotonic time obtained from other peers
31  *   in PEERSTORE (by message type)
32  * - change transport-core API to provide proper flow control in both
33  *   directions, allow multiple messages per peer simultaneously (tag
34  *   confirmations with unique message ID), and replace quota-out with
35  *   proper flow control; specify transmission preferences (latency,
36  *   reliability, etc.) per message!
37  * - add logging
38  *
39  * Later:
40  * - review retransmission logic, right now there is no smartness there!
41  *   => congestion control, flow control, etc (requires RTT, loss, etc.)
42  *
43  * Optimizations:
44  * - use shorthashmap on msg_uuid's when matching reliability/fragment ACKs
45  *   against our pending message queue (requires additional per neighbour
46  *   hash map to be maintained, avoids possible linear scan on pending msgs)
47  * - queue_send_msg and route_message both by API design have to make copies
48  *   of the payload, and route_message on top of that requires a malloc/free.
49  *   Change design to approximate "zero" copy better...
50  * - could avoid copying body of message into each fragment and keep
51  *   fragments as just pointers into the original message and only
52  *   fully build fragments just before transmission (optimization, should
53  *   reduce CPU and memory use)
54  * - if messages are below MTU, consider adding ACKs and other stuff
55  *   (requires planning at receiver, and additional MST-style demultiplex
56  *    at receiver!)
57  * - When we passively learned DV (with unconfirmed freshness), we
58  *   right now add the path to our list but with a zero path_valid_until
59  *   time and only use it for unconfirmed routes.  However, we could consider
60  *   triggering an explicit validation mechansim ourselves, specifically routing
61  *   a challenge-response message over the path (OPTIMIZATION-FIXME).
62  *
63  * Design realizations / discussion:
64  * - communicators do flow control by calling MQ "notify sent"
65  *   when 'ready'. They determine flow implicitly (i.e. TCP blocking)
66  *   or explicitly via backchannel FC ACKs.  As long as the
67  *   channel is not full, they may 'notify sent' even if the other
68  *   peer has not yet confirmed receipt. The other peer confirming
69  *   is _only_ for FC, not for more reliable transmission; reliable
70  *   transmission (i.e. of fragments) is left to _transport_.
71  * - ACKs sent back in uni-directional communicators are done via
72  *   the background channel API; here transport _may_ initially
73  *   broadcast (with bounded # hops) if no path is known;
74  * - transport should _integrate_ DV-routing and build a view of
75  *   the network; then background channel traffic can be
76  *   routed via DV as well as explicit "DV" traffic.
77  * - background channel is also used for ACKs and NAT traversal support
78  * - transport service is responsible for AEAD'ing the background
79  *   channel, timestamps and monotonic time are used against replay
80  *   of old messages -> peerstore needs to be supplied with
81  *   "latest timestamps seen" data
82  * - if transport implements DV, we likely need a 3rd peermap
83  *   in addition to ephemerals and (direct) neighbours
84  *   ==> check if stuff needs to be moved out of "Neighbour"
85  * - transport should encapsualte core-level messages and do its
86  *   own ACKing for RTT/goodput/loss measurements _and_ fragment
87  *   for retransmission
88  */
89 #include "platform.h"
90 #include "gnunet_util_lib.h"
91 #include "gnunet_statistics_service.h"
92 #include "gnunet_transport_monitor_service.h"
93 #include "gnunet_peerstore_service.h"
94 #include "gnunet_hello_lib.h"
95 #include "gnunet_signatures.h"
96 #include "transport.h"
97
98
99 /**
100  * What is the size we assume for a read operation in the
101  * absence of an MTU for the purpose of flow control?
102  */
103 #define IN_PACKET_SIZE_WITHOUT_MTU 128
104
105 /**
106  * Minimum number of hops we should forward DV learn messages
107  * even if they are NOT useful for us in hope of looping
108  * back to the initiator?
109  *
110  * FIXME: allow initiator some control here instead?
111  */
112 #define MIN_DV_PATH_LENGTH_FOR_INITIATOR 3
113
114 /**
115  * Maximum DV distance allowed ever.
116  */
117 #define MAX_DV_HOPS_ALLOWED 16
118
119 /**
120  * Maximum number of DV learning activities we may
121  * have pending at the same time.
122  */
123 #define MAX_DV_LEARN_PENDING 64
124
125 /**
126  * Maximum number of DV paths we keep simultaneously to the same target.
127  */
128 #define MAX_DV_PATHS_TO_TARGET 3
129
130 /**
131  * If a queue delays the next message by more than this number
132  * of seconds we log a warning. Note: this is for testing,
133  * the value chosen here might be too aggressively low!
134  */
135 #define DELAY_WARN_THRESHOLD \
136   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
137
138 /**
139  * We only consider queues as "quality" connections when
140  * suppressing the generation of DV initiation messages if
141  * the latency of the queue is below this threshold.
142  */
143 #define DV_QUALITY_RTT_THRESHOLD \
144   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1)
145
146 /**
147  * How long do we consider a DV path valid if we see no
148  * further updates on it? Note: the value chosen here might be too low!
149  */
150 #define DV_PATH_VALIDITY_TIMEOUT \
151   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5)
152
153 /**
154  * How long do we cache backchannel (struct Backtalker) information
155  * after a backchannel goes inactive?
156  */
157 #define BACKCHANNEL_INACTIVITY_TIMEOUT \
158   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5)
159
160 /**
161  * How long before paths expire would we like to (re)discover DV paths? Should
162  * be below #DV_PATH_VALIDITY_TIMEOUT.
163  */
164 #define DV_PATH_DISCOVERY_FREQUENCY \
165   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 4)
166
167 /**
168  * How long are ephemeral keys valid?
169  */
170 #define EPHEMERAL_VALIDITY \
171   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_HOURS, 4)
172
173 /**
174  * How long do we keep partially reassembled messages around before giving up?
175  */
176 #define REASSEMBLY_EXPIRATION \
177   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 4)
178
179 /**
180  * What is the fastest rate at which we send challenges *if* we keep learning
181  * an address (gossip, DHT, etc.)?
182  */
183 #define FAST_VALIDATION_CHALLENGE_FREQ \
184   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 1)
185
186 /**
187  * What is the slowest rate at which we send challenges?
188  */
189 #define MAX_VALIDATION_CHALLENGE_FREQ \
190   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_DAYS, 1)
191
192 /**
193  * What is the non-randomized base frequency at which we
194  * would initiate DV learn messages?
195  */
196 #define DV_LEARN_BASE_FREQUENCY GNUNET_TIME_UNIT_MINUTES
197
198 /**
199  * How many good connections (confirmed, bi-directional, not DV)
200  * do we need to have to suppress initiating DV learn messages?
201  */
202 #define DV_LEARN_QUALITY_THRESHOLD 100
203
204 /**
205  * When do we forget an invalid address for sure?
206  */
207 #define MAX_ADDRESS_VALID_UNTIL \
208   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MONTHS, 1)
209
210 /**
211  * How long do we consider an address valid if we just checked?
212  */
213 #define ADDRESS_VALIDATION_LIFETIME \
214   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_HOURS, 4)
215
216 /**
217  * What is the maximum frequency at which we do address validation?
218  * A random value between 0 and this value is added when scheduling
219  * the #validation_task (both to ensure we do not validate too often,
220  * and to randomize a bit).
221  */
222 #define MIN_DELAY_ADDRESS_VALIDATION GNUNET_TIME_UNIT_MILLISECONDS
223
224 /**
225  * How many network RTTs before an address validation expires should we begin
226  * trying to revalidate? (Note that the RTT used here is the one that we
227  * experienced during the last validation, not necessarily the latest RTT
228  * observed).
229  */
230 #define VALIDATION_RTT_BUFFER_FACTOR 3
231
232 /**
233  * How many messages can we have pending for a given communicator
234  * process before we start to throttle that communicator?
235  *
236  * Used if a communicator might be CPU-bound and cannot handle the traffic.
237  */
238 #define COMMUNICATOR_TOTAL_QUEUE_LIMIT 512
239
240 /**
241  * How many messages can we have pending for a given queue (queue to
242  * a particular peer via a communicator) process before we start to
243  * throttle that queue?
244  */
245 #define QUEUE_LENGTH_LIMIT 32
246
247
248 GNUNET_NETWORK_STRUCT_BEGIN
249
250 /**
251  * Outer layer of an encapsulated backchannel message.
252  */
253 struct TransportBackchannelEncapsulationMessage
254 {
255   /**
256    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_BACKCHANNEL_ENCAPSULATION.
257    */
258   struct GNUNET_MessageHeader header;
259
260   /**
261    * Reserved, always zero.
262    */
263   uint32_t reserved GNUNET_PACKED;
264
265   /**
266    * Target's peer identity (as backchannels may be transmitted
267    * indirectly, or even be broadcast).
268    */
269   struct GNUNET_PeerIdentity target;
270
271   /**
272    * Ephemeral key setup by the sender for @e target, used
273    * to encrypt the payload.
274    */
275   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral_key;
276
277   /**
278    * We use an IV here as the @e ephemeral_key is re-used for
279    * #EPHEMERAL_VALIDITY time to avoid re-signing it all the time.
280    */
281   struct GNUNET_ShortHashCode iv;
282
283   /**
284    * HMAC over the ciphertext of the encrypted, variable-size
285    * body that follows.  Verified via DH of @e target and
286    * @e ephemeral_key
287    */
288   struct GNUNET_HashCode hmac;
289
290   /* Followed by encrypted, variable-size payload */
291 };
292
293
294 /**
295  * Body by which a peer confirms that it is using an ephemeral key.
296  */
297 struct EphemeralConfirmationPS
298 {
299
300   /**
301    * Purpose is #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL
302    */
303   struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
304
305   /**
306    * How long is this signature over the ephemeral key valid?
307    *
308    * Note that the receiver MUST IGNORE the absolute time, and only interpret
309    * the value as a mononic time and reject "older" values than the last one
310    * observed.  This is necessary as we do not want to require synchronized
311    * clocks and may not have a bidirectional communication channel.
312    *
313    * Even with this, there is no real guarantee against replay achieved here,
314    * unless the latest timestamp is persisted.  While persistence should be
315    * provided via PEERSTORE, we do not consider the mechanism reliable!  Thus,
316    * communicators must protect against replay attacks when using backchannel
317    * communication!
318    */
319   struct GNUNET_TIME_AbsoluteNBO ephemeral_validity;
320
321   /**
322    * Target's peer identity.
323    */
324   struct GNUNET_PeerIdentity target;
325
326   /**
327    * Ephemeral key setup by the sender for @e target, used
328    * to encrypt the payload.
329    */
330   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral_key;
331 };
332
333
334 /**
335  * Plaintext of the variable-size payload that is encrypted
336  * within a `struct TransportBackchannelEncapsulationMessage`
337  */
338 struct TransportBackchannelRequestPayload
339 {
340
341   /**
342    * Sender's peer identity.
343    */
344   struct GNUNET_PeerIdentity sender;
345
346   /**
347    * Signature of the sender over an
348    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL.
349    */
350   struct GNUNET_CRYPTO_EddsaSignature sender_sig;
351
352   /**
353    * How long is this signature over the ephemeral key valid?
354    *
355    * Note that the receiver MUST IGNORE the absolute time, and only interpret
356    * the value as a mononic time and reject "older" values than the last one
357    * observed.  This is necessary as we do not want to require synchronized
358    * clocks and may not have a bidirectional communication channel.
359    *
360    * Even with this, there is no real guarantee against replay achieved here,
361    * unless the latest timestamp is persisted.  While persistence should be
362    * provided via PEERSTORE, we do not consider the mechanism reliable!  Thus,
363    * communicators must protect against replay attacks when using backchannel
364    * communication!
365    */
366   struct GNUNET_TIME_AbsoluteNBO ephemeral_validity;
367
368   /**
369    * Current monotonic time of the sending transport service.  Used to
370    * detect replayed messages.  Note that the receiver should remember
371    * a list of the recently seen timestamps and only reject messages
372    * if the timestamp is in the list, or the list is "full" and the
373    * timestamp is smaller than the lowest in the list.
374    *
375    * Like the @e ephemeral_validity, the list of timestamps per peer should be
376    * persisted to guard against replays after restarts.
377    */
378   struct GNUNET_TIME_AbsoluteNBO monotonic_time;
379
380   /* Followed by a `struct GNUNET_MessageHeader` with a message
381      for a communicator */
382
383   /* Followed by a 0-termianted string specifying the name of
384      the communicator which is to receive the message */
385 };
386
387
388 /**
389  * Outer layer of an encapsulated unfragmented application message sent
390  * over an unreliable channel.
391  */
392 struct TransportReliabilityBox
393 {
394   /**
395    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_BOX
396    */
397   struct GNUNET_MessageHeader header;
398
399   /**
400    * Number of messages still to be sent before a commulative
401    * ACK is requested.  Zero if an ACK is requested immediately.
402    * In NBO.  Note that the receiver may send the ACK faster
403    * if it believes that is reasonable.
404    */
405   uint32_t ack_countdown GNUNET_PACKED;
406
407   /**
408    * Unique ID of the message used for signalling receipt of
409    * messages sent over possibly unreliable channels.  Should
410    * be a random.
411    */
412   struct GNUNET_ShortHashCode msg_uuid;
413 };
414
415
416 /**
417  * Confirmation that the receiver got a
418  * #GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_BOX. Note that the
419  * confirmation may be transmitted over a completely different queue,
420  * so ACKs are identified by a combination of PID of sender and
421  * message UUID, without the queue playing any role!
422  */
423 struct TransportReliabilityAckMessage
424 {
425   /**
426    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_ACK
427    */
428   struct GNUNET_MessageHeader header;
429
430   /**
431    * Reserved. Zero.
432    */
433   uint32_t reserved GNUNET_PACKED;
434
435   /**
436    * How long was the ACK delayed relative to the average time of
437    * receipt of the messages being acknowledged?  Used to calculate
438    * the average RTT by taking the receipt time of the ack minus the
439    * average transmission time of the sender minus this value.
440    */
441   struct GNUNET_TIME_RelativeNBO avg_ack_delay;
442
443   /* followed by any number of `struct GNUNET_ShortHashCode`
444      messages providing ACKs */
445 };
446
447
448 /**
449  * Outer layer of an encapsulated fragmented application message.
450  */
451 struct TransportFragmentBox
452 {
453   /**
454    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT
455    */
456   struct GNUNET_MessageHeader header;
457
458   /**
459    * Unique ID of this fragment (and fragment transmission!). Will
460    * change even if a fragement is retransmitted to make each
461    * transmission attempt unique! Should be incremented by one for
462    * each fragment transmission. If a client receives a duplicate
463    * fragment (same @e frag_off), it must send
464    * #GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT_ACK immediately.
465    */
466   uint32_t frag_uuid GNUNET_PACKED;
467
468   /**
469    * Original message ID for of the message that all the1
470    * fragments belong to.  Must be the same for all fragments.
471    */
472   struct GNUNET_ShortHashCode msg_uuid;
473
474   /**
475    * Offset of this fragment in the overall message.
476    */
477   uint16_t frag_off GNUNET_PACKED;
478
479   /**
480    * Total size of the message that is being fragmented.
481    */
482   uint16_t msg_size GNUNET_PACKED;
483 };
484
485
486 /**
487  * Outer layer of an fragmented application message sent over a queue
488  * with finite MTU.  When a #GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT is
489  * received, the receiver has two RTTs or 64 further fragments with
490  * the same basic message time to send an acknowledgement, possibly
491  * acknowledging up to 65 fragments in one ACK.  ACKs must also be
492  * sent immediately once all fragments were sent.
493  */
494 struct TransportFragmentAckMessage
495 {
496   /**
497    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT_ACK
498    */
499   struct GNUNET_MessageHeader header;
500
501   /**
502    * Unique ID of the lowest fragment UUID being acknowledged.
503    */
504   uint32_t frag_uuid GNUNET_PACKED;
505
506   /**
507    * Bitfield of up to 64 additional fragments following the
508    * @e msg_uuid being acknowledged by this message.
509    */
510   uint64_t extra_acks GNUNET_PACKED;
511
512   /**
513    * Original message ID for of the message that all the
514    * fragments belong to.
515    */
516   struct GNUNET_ShortHashCode msg_uuid;
517
518   /**
519    * How long was the ACK delayed relative to the average time of
520    * receipt of the fragments being acknowledged?  Used to calculate
521    * the average RTT by taking the receipt time of the ack minus the
522    * average transmission time of the sender minus this value.
523    */
524   struct GNUNET_TIME_RelativeNBO avg_ack_delay;
525
526   /**
527    * How long until the receiver will stop trying reassembly
528    * of this message?
529    */
530   struct GNUNET_TIME_RelativeNBO reassembly_timeout;
531 };
532
533
534 /**
535  * Content signed by the initator during DV learning.
536  *
537  * The signature is required to prevent DDoS attacks. A peer sending out this
538  * message is potentially generating a lot of traffic that will go back to the
539  * initator, as peers receiving this message will try to let the initiator
540  * know that they got the message.
541  *
542  * Without this signature, an attacker could abuse this mechanism for traffic
543  * amplification, sending a lot of traffic to a peer by putting out this type
544  * of message with the victim's peer identity.
545  *
546  * Even with just a signature, traffic amplification would be possible via
547  * replay attacks. The @e monotonic_time limits such replay attacks, as every
548  * potential amplificator will check the @e monotonic_time and only respond
549  * (at most) once per message.
550  */
551 struct DvInitPS
552 {
553   /**
554    * Purpose is #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR
555    */
556   struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
557
558   /**
559    * Time at the initiator when generating the signature.
560    *
561    * Note that the receiver MUST IGNORE the absolute time, and only interpret
562    * the value as a mononic time and reject "older" values than the last one
563    * observed.  This is necessary as we do not want to require synchronized
564    * clocks and may not have a bidirectional communication channel.
565    *
566    * Even with this, there is no real guarantee against replay achieved here,
567    * unless the latest timestamp is persisted.  Persistence should be
568    * provided via PEERSTORE if possible.
569    */
570   struct GNUNET_TIME_AbsoluteNBO monotonic_time;
571
572   /**
573    * Challenge value used by the initiator to re-identify the path.
574    */
575   struct GNUNET_ShortHashCode challenge;
576 };
577
578
579 /**
580  * Content signed by each peer during DV learning.
581  *
582  * This assues the initiator of the DV learning operation that the hop from @e
583  * pred via the signing peer to @e succ actually exists.  This makes it
584  * impossible for an adversary to supply the network with bogus routes.
585  *
586  * The @e challenge is included to provide replay protection for the
587  * initiator. This way, the initiator knows that the hop existed after the
588  * original @e challenge was first transmitted, providing a freshness metric.
589  *
590  * Peers other than the initiator that passively learn paths by observing
591  * these messages do NOT benefit from this. Here, an adversary may indeed
592  * replay old messages.  Thus, passively learned paths should always be
593  * immediately marked as "potentially stale".
594  */
595 struct DvHopPS
596 {
597   /**
598    * Purpose is #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_HOP
599    */
600   struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
601
602   /**
603    * Identity of the previous peer on the path.
604    */
605   struct GNUNET_PeerIdentity pred;
606
607   /**
608    * Identity of the next peer on the path.
609    */
610   struct GNUNET_PeerIdentity succ;
611
612   /**
613    * Challenge value used by the initiator to re-identify the path.
614    */
615   struct GNUNET_ShortHashCode challenge;
616 };
617
618
619 /**
620  * An entry describing a peer on a path in a
621  * `struct TransportDVLearn` message.
622  */
623 struct DVPathEntryP
624 {
625   /**
626    * Identity of a peer on the path.
627    */
628   struct GNUNET_PeerIdentity hop;
629
630   /**
631    * Signature of this hop over the path, of purpose
632    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_HOP
633    */
634   struct GNUNET_CRYPTO_EddsaSignature hop_sig;
635 };
636
637
638 /**
639  * Internal message used by transport for distance vector learning.
640  * If @e num_hops does not exceed the threshold, peers should append
641  * themselves to the peer list and flood the message (possibly only
642  * to a subset of their neighbours to limit discoverability of the
643  * network topology).  To the extend that the @e bidirectional bits
644  * are set, peers may learn the inverse paths even if they did not
645  * initiate.
646  *
647  * Unless received on a bidirectional queue and @e num_hops just
648  * zero, peers that can forward to the initator should always try to
649  * forward to the initiator.
650  */
651 struct TransportDVLearn
652 {
653   /**
654    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN
655    */
656   struct GNUNET_MessageHeader header;
657
658   /**
659    * Number of hops this messages has travelled, in NBO. Zero if
660    * sent by initiator.
661    */
662   uint16_t num_hops GNUNET_PACKED;
663
664   /**
665    * Bitmask of the last 16 hops indicating whether they are confirmed
666    * available (without DV) in both directions or not, in NBO.  Used
667    * to possibly instantly learn a path in both directions.  Each peer
668    * should shift this value by one to the left, and then set the
669    * lowest bit IF the current sender can be reached from it (without
670    * DV routing).
671    */
672   uint16_t bidirectional GNUNET_PACKED;
673
674   /**
675    * Peers receiving this message and delaying forwarding to other
676    * peers for any reason should increment this value by the non-network
677    * delay created by the peer.
678    */
679   struct GNUNET_TIME_RelativeNBO non_network_delay;
680
681   /**
682    * Signature of this hop over the path, of purpose
683    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR
684    */
685   struct GNUNET_CRYPTO_EddsaSignature init_sig;
686
687   /**
688    * Identity of the peer that started this learning activity.
689    */
690   struct GNUNET_PeerIdentity initiator;
691
692   /**
693    * Challenge value used by the initiator to re-identify the path.
694    */
695   struct GNUNET_ShortHashCode challenge;
696
697   /* Followed by @e num_hops `struct DVPathEntryP` values,
698      excluding the initiator of the DV trace; the last entry is the
699      current sender; the current peer must not be included. */
700 };
701
702
703 /**
704  * Outer layer of an encapsulated message send over multiple hops.
705  * The path given only includes the identities of the subsequent
706  * peers, i.e. it will be empty if we are the receiver. Each
707  * forwarding peer should scan the list from the end, and if it can,
708  * forward to the respective peer. The list should then be shortened
709  * by all the entries up to and including that peer.  Each hop should
710  * also increment @e total_hops to allow the receiver to get a precise
711  * estimate on the number of hops the message travelled.  Senders must
712  * provide a learned path that thus should work, but intermediaries
713  * know of a shortcut, they are allowed to send the message via that
714  * shortcut.
715  *
716  * If a peer finds itself still on the list, it must drop the message.
717  */
718 struct TransportDVBox
719 {
720   /**
721    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX
722    */
723   struct GNUNET_MessageHeader header;
724
725   /**
726    * Number of total hops this messages travelled. In NBO.
727    * @e origin sets this to zero, to be incremented at
728    * each hop.
729    */
730   uint16_t total_hops GNUNET_PACKED;
731
732   /**
733    * Number of hops this messages includes. In NBO.
734    */
735   uint16_t num_hops GNUNET_PACKED;
736
737   /**
738    * Identity of the peer that originated the message.
739    */
740   struct GNUNET_PeerIdentity origin;
741
742   /* Followed by @e num_hops `struct GNUNET_PeerIdentity` values;
743      excluding the @e origin and the current peer, the last must be
744      the ultimate target; if @e num_hops is zero, the receiver of this
745      message is the ultimate target. */
746
747   /* Followed by the actual message, which itself may be
748      another box, but not a DV_LEARN or DV_BOX message! */
749 };
750
751
752 /**
753  * Message send to another peer to validate that it can indeed
754  * receive messages at a particular address.
755  */
756 struct TransportValidationChallenge
757 {
758
759   /**
760    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_CHALLENGE
761    */
762   struct GNUNET_MessageHeader header;
763
764   /**
765    * Zero.
766    */
767   uint32_t reserved GNUNET_PACKED;
768
769   /**
770    * Challenge to be signed by the receiving peer.
771    */
772   struct GNUNET_ShortHashCode challenge;
773
774   /**
775    * Timestamp of the sender, to be copied into the reply
776    * to allow sender to calculate RTT.
777    */
778   struct GNUNET_TIME_AbsoluteNBO sender_time;
779 };
780
781
782 /**
783  * Message signed by a peer to confirm that it can indeed
784  * receive messages at a particular address.
785  */
786 struct TransportValidationPS
787 {
788
789   /**
790    * Purpose is #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_CHALLENGE
791    */
792   struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
793
794   /**
795    * How long does the sender believe the address on
796    * which the challenge was received to remain valid?
797    */
798   struct GNUNET_TIME_RelativeNBO validity_duration;
799
800   /**
801    * Challenge signed by the receiving peer.
802    */
803   struct GNUNET_ShortHashCode challenge;
804 };
805
806
807 /**
808  * Message send to a peer to respond to a
809  * #GNUNET_MESSAGE_TYPE_ADDRESS_VALIDATION_CHALLENGE
810  */
811 struct TransportValidationResponse
812 {
813
814   /**
815    * Type is #GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_RESPONSE
816    */
817   struct GNUNET_MessageHeader header;
818
819   /**
820    * Zero.
821    */
822   uint32_t reserved GNUNET_PACKED;
823
824   /**
825    * The peer's signature matching the
826    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_CHALLENGE purpose.
827    */
828   struct GNUNET_CRYPTO_EddsaSignature signature;
829
830   /**
831    * The challenge that was signed by the receiving peer.
832    */
833   struct GNUNET_ShortHashCode challenge;
834
835   /**
836    * Original timestamp of the sender (was @code{sender_time}),
837    * copied into the reply to allow sender to calculate RTT.
838    */
839   struct GNUNET_TIME_AbsoluteNBO origin_time;
840
841   /**
842    * How long does the sender believe this address to remain
843    * valid?
844    */
845   struct GNUNET_TIME_RelativeNBO validity_duration;
846 };
847
848
849 GNUNET_NETWORK_STRUCT_END
850
851
852 /**
853  * What type of client is the `struct TransportClient` about?
854  */
855 enum ClientType
856 {
857   /**
858    * We do not know yet (client is fresh).
859    */
860   CT_NONE = 0,
861
862   /**
863    * Is the CORE service, we need to forward traffic to it.
864    */
865   CT_CORE = 1,
866
867   /**
868    * It is a monitor, forward monitor data.
869    */
870   CT_MONITOR = 2,
871
872   /**
873    * It is a communicator, use for communication.
874    */
875   CT_COMMUNICATOR = 3,
876
877   /**
878    * "Application" telling us where to connect (i.e. TOPOLOGY, DHT or CADET).
879    */
880   CT_APPLICATION = 4
881 };
882
883
884 /**
885  * When did we launch this DV learning activity?
886  */
887 struct LearnLaunchEntry
888 {
889
890   /**
891    * Kept (also) in a DLL sorted by launch time.
892    */
893   struct LearnLaunchEntry *prev;
894
895   /**
896    * Kept (also) in a DLL sorted by launch time.
897    */
898   struct LearnLaunchEntry *next;
899
900   /**
901    * Challenge that uniquely identifies this activity.
902    */
903   struct GNUNET_ShortHashCode challenge;
904
905   /**
906    * When did we transmit the DV learn message (used to calculate RTT) and
907    * determine freshness of paths learned via this operation.
908    */
909   struct GNUNET_TIME_Absolute launch_time;
910 };
911
912
913 /**
914  * Entry in our cache of ephemeral keys we currently use.  This way, we only
915  * sign an ephemeral once per @e target, and then can re-use it over multiple
916  * #GNUNET_MESSAGE_TYPE_TRANSPORT_BACKCHANNEL_ENCAPSULATION messages (as
917  * signing is expensive and in some cases we may use backchannel messages a
918  * lot).
919  */
920 struct EphemeralCacheEntry
921 {
922
923   /**
924    * Target's peer identity (we don't re-use ephemerals
925    * to limit linkability of messages).
926    */
927   struct GNUNET_PeerIdentity target;
928
929   /**
930    * Signature affirming @e ephemeral_key of type
931    * #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL
932    */
933   struct GNUNET_CRYPTO_EddsaSignature sender_sig;
934
935   /**
936    * How long is @e sender_sig valid
937    */
938   struct GNUNET_TIME_Absolute ephemeral_validity;
939
940   /**
941    * Our ephemeral key.
942    */
943   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral_key;
944
945   /**
946    * Our private ephemeral key.
947    */
948   struct GNUNET_CRYPTO_EcdhePrivateKey private_key;
949
950   /**
951    * Node in the ephemeral cache for this entry.
952    * Used for expiration.
953    */
954   struct GNUNET_CONTAINER_HeapNode *hn;
955 };
956
957
958 /**
959  * Client connected to the transport service.
960  */
961 struct TransportClient;
962
963
964 /**
965  * A neighbour that at least one communicator is connected to.
966  */
967 struct Neighbour;
968
969
970 /**
971  * Entry in our #dv_routes table, representing a (set of) distance
972  * vector routes to a particular peer.
973  */
974 struct DistanceVector;
975
976 /**
977  * One possible hop towards a DV target.
978  */
979 struct DistanceVectorHop
980 {
981
982   /**
983    * Kept in a MDLL, sorted by @e timeout.
984    */
985   struct DistanceVectorHop *next_dv;
986
987   /**
988    * Kept in a MDLL, sorted by @e timeout.
989    */
990   struct DistanceVectorHop *prev_dv;
991
992   /**
993    * Kept in a MDLL.
994    */
995   struct DistanceVectorHop *next_neighbour;
996
997   /**
998    * Kept in a MDLL.
999    */
1000   struct DistanceVectorHop *prev_neighbour;
1001
1002   /**
1003    * What would be the next hop to @e target?
1004    */
1005   struct Neighbour *next_hop;
1006
1007   /**
1008    * Distance vector entry this hop belongs with.
1009    */
1010   struct DistanceVector *dv;
1011
1012   /**
1013    * Array of @e distance hops to the target, excluding @e next_hop.
1014    * NULL if the entire path is us to @e next_hop to `target`. Allocated
1015    * at the end of this struct. Excludes the target itself!
1016    */
1017   const struct GNUNET_PeerIdentity *path;
1018
1019   /**
1020    * At what time do we forget about this path unless we see it again
1021    * while learning?
1022    */
1023   struct GNUNET_TIME_Absolute timeout;
1024
1025   /**
1026    * For how long is the validation of this path considered
1027    * valid?
1028    * Set to ZERO if the path is learned by snooping on DV learn messages
1029    * initiated by other peers, and to the time at which we generated the
1030    * challenge for DV learn operations this peer initiated.
1031    */
1032   struct GNUNET_TIME_Absolute path_valid_until;
1033
1034   /**
1035    * Number of hops in total to the `target` (excluding @e next_hop and `target`
1036    * itself). Thus 0 still means a distance of 2 hops (to @e next_hop and then
1037    * to `target`).
1038    */
1039   unsigned int distance;
1040 };
1041
1042
1043 /**
1044  * Entry in our #dv_routes table, representing a (set of) distance
1045  * vector routes to a particular peer.
1046  */
1047 struct DistanceVector
1048 {
1049
1050   /**
1051    * To which peer is this a route?
1052    */
1053   struct GNUNET_PeerIdentity target;
1054
1055   /**
1056    * Known paths to @e target.
1057    */
1058   struct DistanceVectorHop *dv_head;
1059
1060   /**
1061    * Known paths to @e target.
1062    */
1063   struct DistanceVectorHop *dv_tail;
1064
1065   /**
1066    * Task scheduled to purge expired paths from @e dv_head MDLL.
1067    */
1068   struct GNUNET_SCHEDULER_Task *timeout_task;
1069
1070   /**
1071    * Task scheduled to possibly notfiy core that this queue is no longer
1072    * counting as confirmed.  Runs the #core_queue_visibility_check().
1073    */
1074   struct GNUNET_SCHEDULER_Task *visibility_task;
1075
1076   /**
1077    * Quota at which CORE is allowed to transmit to this peer
1078    * (note that the value CORE should actually be told is this
1079    *  value plus the respective value in `struct Neighbour`).
1080    * Should match the sum of the quotas of all of the paths.
1081    *
1082    * FIXME: not yet set, tricky to get right given multiple paths,
1083    *        many of which may be inactive! (=> Idea: measure???)
1084    * FIXME: how do we set this value initially when we tell CORE?
1085    *    Options: start at a minimum value or at literally zero?
1086    *         (=> Current thought: clean would be zero!)
1087    */
1088   struct GNUNET_BANDWIDTH_Value32NBO quota_out;
1089
1090   /**
1091    * Is one of the DV paths in this struct 'confirmed' and thus
1092    * the cause for CORE to see this peer as connected? (Note that
1093    * the same may apply to a `struct Neighbour` at the same time.)
1094    */
1095   int core_visible;
1096 };
1097
1098
1099 /**
1100  * A queue is a message queue provided by a communicator
1101  * via which we can reach a particular neighbour.
1102  */
1103 struct Queue;
1104
1105 /**
1106  * Message awaiting transmission. See detailed comments below.
1107  */
1108 struct PendingMessage;
1109
1110 /**
1111  * Entry identifying transmission in one of our `struct
1112  * Queue` which still awaits an ACK.  This is used to
1113  * ensure we do not overwhelm a communicator and limit the number of
1114  * messages outstanding per communicator (say in case communicator is
1115  * CPU bound) and per queue (in case bandwidth allocation exceeds
1116  * what the communicator can actually provide towards a particular
1117  * peer/target).
1118  */
1119 struct QueueEntry
1120 {
1121
1122   /**
1123    * Kept as a DLL.
1124    */
1125   struct QueueEntry *next;
1126
1127   /**
1128    * Kept as a DLL.
1129    */
1130   struct QueueEntry *prev;
1131
1132   /**
1133    * Queue this entry is queued with.
1134    */
1135   struct Queue *queue;
1136
1137   /**
1138    * Pending message this entry is for, or NULL for none.
1139    */
1140   struct PendingMessage *pm;
1141
1142   /**
1143    * Message ID used for this message with the queue used for transmission.
1144    */
1145   uint64_t mid;
1146 };
1147
1148
1149 /**
1150  * A queue is a message queue provided by a communicator
1151  * via which we can reach a particular neighbour.
1152  */
1153 struct Queue
1154 {
1155   /**
1156    * Kept in a MDLL.
1157    */
1158   struct Queue *next_neighbour;
1159
1160   /**
1161    * Kept in a MDLL.
1162    */
1163   struct Queue *prev_neighbour;
1164
1165   /**
1166    * Kept in a MDLL.
1167    */
1168   struct Queue *prev_client;
1169
1170   /**
1171    * Kept in a MDLL.
1172    */
1173   struct Queue *next_client;
1174
1175   /**
1176    * Head of DLL of unacked transmission requests.
1177    */
1178   struct QueueEntry *queue_head;
1179
1180   /**
1181    * End of DLL of unacked transmission requests.
1182    */
1183   struct QueueEntry *queue_tail;
1184
1185   /**
1186    * Which neighbour is this queue for?
1187    */
1188   struct Neighbour *neighbour;
1189
1190   /**
1191    * Which communicator offers this queue?
1192    */
1193   struct TransportClient *tc;
1194
1195   /**
1196    * Address served by the queue.
1197    */
1198   const char *address;
1199
1200   /**
1201    * Task scheduled for the time when this queue can (likely) transmit the
1202    * next message. Still needs to check with the @e tracker_out to be sure.
1203    */
1204   struct GNUNET_SCHEDULER_Task *transmit_task;
1205
1206   /**
1207    * Task scheduled to possibly notfiy core that this queue is no longer
1208    * counting as confirmed.  Runs the #core_queue_visibility_check().
1209    */
1210   struct GNUNET_SCHEDULER_Task *visibility_task;
1211
1212   /**
1213    * Our current RTT estimate for this queue.
1214    */
1215   struct GNUNET_TIME_Relative rtt;
1216
1217   /**
1218    * How long do *we* consider this @e address to be valid?  In the past or
1219    * zero if we have not yet validated it.  Can be updated based on
1220    * challenge-response validations (via address validation logic), or when we
1221    * receive ACKs that we can definitively map to transmissions via this
1222    * queue.
1223    */
1224   struct GNUNET_TIME_Absolute validated_until;
1225
1226   /**
1227    * Message ID generator for transmissions on this queue.
1228    */
1229   uint64_t mid_gen;
1230
1231   /**
1232    * Unique identifier of this queue with the communicator.
1233    */
1234   uint32_t qid;
1235
1236   /**
1237    * Maximum transmission unit supported by this queue.
1238    */
1239   uint32_t mtu;
1240
1241   /**
1242    * Messages pending.
1243    */
1244   uint32_t num_msg_pending;
1245
1246   /**
1247    * Bytes pending.
1248    */
1249   uint32_t num_bytes_pending;
1250
1251   /**
1252    * Length of the DLL starting at @e queue_head.
1253    */
1254   unsigned int queue_length;
1255
1256   /**
1257    * Network type offered by this queue.
1258    */
1259   enum GNUNET_NetworkType nt;
1260
1261   /**
1262    * Connection status for this queue.
1263    */
1264   enum GNUNET_TRANSPORT_ConnectionStatus cs;
1265
1266   /**
1267    * How much outbound bandwidth do we have available for this queue?
1268    */
1269   struct GNUNET_BANDWIDTH_Tracker tracker_out;
1270
1271   /**
1272    * How much inbound bandwidth do we have available for this queue?
1273    */
1274   struct GNUNET_BANDWIDTH_Tracker tracker_in;
1275 };
1276
1277
1278 /**
1279  * Information we keep for a message that we are reassembling.
1280  */
1281 struct ReassemblyContext
1282 {
1283
1284   /**
1285    * Original message ID for of the message that all the
1286    * fragments belong to.
1287    */
1288   struct GNUNET_ShortHashCode msg_uuid;
1289
1290   /**
1291    * Which neighbour is this context for?
1292    */
1293   struct Neighbour *neighbour;
1294
1295   /**
1296    * Entry in the reassembly heap (sorted by expiration).
1297    */
1298   struct GNUNET_CONTAINER_HeapNode *hn;
1299
1300   /**
1301    * Bitfield with @e msg_size bits representing the positions
1302    * where we have received fragments.  When we receive a fragment,
1303    * we check the bits in @e bitfield before incrementing @e msg_missing.
1304    *
1305    * Allocated after the reassembled message.
1306    */
1307   uint8_t *bitfield;
1308
1309   /**
1310    * Task for sending ACK. We may send ACKs either because of hitting
1311    * the @e extra_acks limit, or based on time and @e num_acks.  This
1312    * task is for the latter case.
1313    */
1314   struct GNUNET_SCHEDULER_Task *ack_task;
1315
1316   /**
1317    * At what time will we give up reassembly of this message?
1318    */
1319   struct GNUNET_TIME_Absolute reassembly_timeout;
1320
1321   /**
1322    * Average delay of all acks in @e extra_acks and @e frag_uuid.
1323    * Should be reset to zero when @e num_acks is set to 0.
1324    */
1325   struct GNUNET_TIME_Relative avg_ack_delay;
1326
1327   /**
1328    * Time we received the last fragment.  @e avg_ack_delay must be
1329    * incremented by now - @e last_frag multiplied by @e num_acks.
1330    */
1331   struct GNUNET_TIME_Absolute last_frag;
1332
1333   /**
1334    * Bitfield of up to 64 additional fragments following @e frag_uuid
1335    * to be acknowledged in the next cummulative ACK.
1336    */
1337   uint64_t extra_acks;
1338
1339   /**
1340    * Unique ID of the lowest fragment UUID to be acknowledged in the
1341    * next cummulative ACK.  Only valid if @e num_acks > 0.
1342    */
1343   uint32_t frag_uuid;
1344
1345   /**
1346    * Number of ACKs we have accumulated so far.  Reset to 0
1347    * whenever we send a #GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT_ACK.
1348    */
1349   unsigned int num_acks;
1350
1351   /**
1352    * How big is the message we are reassembling in total?
1353    */
1354   uint16_t msg_size;
1355
1356   /**
1357    * How many bytes of the message are still missing?  Defragmentation
1358    * is complete when @e msg_missing == 0.
1359    */
1360   uint16_t msg_missing;
1361
1362   /* Followed by @e msg_size bytes of the (partially) defragmented original
1363    * message */
1364
1365   /* Followed by @e bitfield data */
1366 };
1367
1368
1369 /**
1370  * A neighbour that at least one communicator is connected to.
1371  */
1372 struct Neighbour
1373 {
1374
1375   /**
1376    * Which peer is this about?
1377    */
1378   struct GNUNET_PeerIdentity pid;
1379
1380   /**
1381    * Map with `struct ReassemblyContext` structs for fragments under
1382    * reassembly. May be NULL if we currently have no fragments from
1383    * this @e pid (lazy initialization).
1384    */
1385   struct GNUNET_CONTAINER_MultiShortmap *reassembly_map;
1386
1387   /**
1388    * Heap with `struct ReassemblyContext` structs for fragments under
1389    * reassembly. May be NULL if we currently have no fragments from
1390    * this @e pid (lazy initialization).
1391    */
1392   struct GNUNET_CONTAINER_Heap *reassembly_heap;
1393
1394   /**
1395    * Task to free old entries from the @e reassembly_heap and @e reassembly_map.
1396    */
1397   struct GNUNET_SCHEDULER_Task *reassembly_timeout_task;
1398
1399   /**
1400    * Head of list of messages pending for this neighbour.
1401    */
1402   struct PendingMessage *pending_msg_head;
1403
1404   /**
1405    * Tail of list of messages pending for this neighbour.
1406    */
1407   struct PendingMessage *pending_msg_tail;
1408
1409   /**
1410    * Head of MDLL of DV hops that have this neighbour as next hop. Must be
1411    * purged if this neighbour goes down.
1412    */
1413   struct DistanceVectorHop *dv_head;
1414
1415   /**
1416    * Tail of MDLL of DV hops that have this neighbour as next hop. Must be
1417    * purged if this neighbour goes down.
1418    */
1419   struct DistanceVectorHop *dv_tail;
1420
1421   /**
1422    * Head of DLL of queues to this peer.
1423    */
1424   struct Queue *queue_head;
1425
1426   /**
1427    * Tail of DLL of queues to this peer.
1428    */
1429   struct Queue *queue_tail;
1430
1431   /**
1432    * Task run to cleanup pending messages that have exceeded their timeout.
1433    */
1434   struct GNUNET_SCHEDULER_Task *timeout_task;
1435
1436   /**
1437    * Quota at which CORE is allowed to transmit to this peer
1438    * (note that the value CORE should actually be told is this
1439    *  value plus the respective value in `struct DistanceVector`).
1440    * Should match the sum of the quotas of all of the queues.
1441    *
1442    * FIXME: not yet set, tricky to get right given multiple queues!
1443    *        (=> Idea: measure???)
1444    * FIXME: how do we set this value initially when we tell CORE?
1445    *    Options: start at a minimum value or at literally zero?
1446    *         (=> Current thought: clean would be zero!)
1447    */
1448   struct GNUNET_BANDWIDTH_Value32NBO quota_out;
1449
1450   /**
1451    * What is the earliest timeout of any message in @e pending_msg_tail?
1452    */
1453   struct GNUNET_TIME_Absolute earliest_timeout;
1454
1455   /**
1456    * Do we have a confirmed working queue and are thus visible to
1457    * CORE?
1458    */
1459   int core_visible;
1460 };
1461
1462
1463 /**
1464  * A peer that an application (client) would like us to talk to directly.
1465  */
1466 struct PeerRequest
1467 {
1468
1469   /**
1470    * Which peer is this about?
1471    */
1472   struct GNUNET_PeerIdentity pid;
1473
1474   /**
1475    * Client responsible for the request.
1476    */
1477   struct TransportClient *tc;
1478
1479   /**
1480    * Handle for watching the peerstore for HELLOs for this peer.
1481    */
1482   struct GNUNET_PEERSTORE_WatchContext *wc;
1483
1484   /**
1485    * What kind of performance preference does this @e tc have?
1486    */
1487   enum GNUNET_MQ_PreferenceKind pk;
1488
1489   /**
1490    * How much bandwidth would this @e tc like to see?
1491    */
1492   struct GNUNET_BANDWIDTH_Value32NBO bw;
1493 };
1494
1495
1496 /**
1497  * Types of different pending messages.
1498  */
1499 enum PendingMessageType
1500 {
1501
1502   /**
1503    * Ordinary message received from the CORE service.
1504    */
1505   PMT_CORE = 0,
1506
1507   /**
1508    * Fragment box.
1509    */
1510   PMT_FRAGMENT_BOX = 1,
1511
1512   /**
1513    * Reliability box.
1514    */
1515   PMT_RELIABILITY_BOX = 2,
1516
1517   /**
1518    * Any type of acknowledgement.
1519    */
1520   PMT_ACKNOWLEDGEMENT = 3,
1521
1522   /**
1523    * Control traffic generated by the TRANSPORT service itself.
1524    */
1525   PMT_CONTROL = 4
1526
1527 };
1528
1529
1530 /**
1531  * Transmission request that is awaiting delivery.  The original
1532  * transmission requests from CORE may be too big for some queues.
1533  * In this case, a *tree* of fragments is created.  At each
1534  * level of the tree, fragments are kept in a DLL ordered by which
1535  * fragment should be sent next (at the head).  The tree is searched
1536  * top-down, with the original message at the root.
1537  *
1538  * To select a node for transmission, first it is checked if the
1539  * current node's message fits with the MTU.  If it does not, we
1540  * either calculate the next fragment (based on @e frag_off) from the
1541  * current node, or, if all fragments have already been created,
1542  * descend to the @e head_frag.  Even though the node was already
1543  * fragmented, the fragment may be too big if the fragment was
1544  * generated for a queue with a larger MTU. In this case, the node
1545  * may be fragmented again, thus creating a tree.
1546  *
1547  * When acknowledgements for fragments are received, the tree
1548  * must be pruned, removing those parts that were already
1549  * acknowledged.  When fragments are sent over a reliable
1550  * channel, they can be immediately removed.
1551  *
1552  * If a message is ever fragmented, then the original "full" message
1553  * is never again transmitted (even if it fits below the MTU), and
1554  * only (remaining) fragments are sent.
1555  */
1556 struct PendingMessage
1557 {
1558   /**
1559    * Kept in a MDLL of messages for this @a target.
1560    */
1561   struct PendingMessage *next_neighbour;
1562
1563   /**
1564    * Kept in a MDLL of messages for this @a target.
1565    */
1566   struct PendingMessage *prev_neighbour;
1567
1568   /**
1569    * Kept in a MDLL of messages from this @a client (if @e pmt is #PMT_CORE)
1570    */
1571   struct PendingMessage *next_client;
1572
1573   /**
1574    * Kept in a MDLL of messages from this @a client  (if @e pmt is #PMT_CORE)
1575    */
1576   struct PendingMessage *prev_client;
1577
1578   /**
1579    * Kept in a MDLL of messages from this @a cpm (if @e pmt is
1580    * #PMT_FRAGMENT_BOx)
1581    */
1582   struct PendingMessage *next_frag;
1583
1584   /**
1585    * Kept in a MDLL of messages from this @a cpm  (if @e pmt is
1586    * #PMT_FRAGMENT_BOX)
1587    */
1588   struct PendingMessage *prev_frag;
1589
1590   /**
1591    * This message, reliability boxed. Only possibly available if @e pmt is
1592    * #PMT_CORE.
1593    */
1594   struct PendingMessage *bpm;
1595
1596   /**
1597    * Target of the request.
1598    */
1599   struct Neighbour *target;
1600
1601   /**
1602    * Set to non-NULL value if this message is currently being given to a
1603    * communicator and we are awaiting that communicator's acknowledgement.
1604    * Note that we must not retransmit a pending message while we're still
1605    * in the process of giving it to a communicator. If a pending message
1606    * is free'd while this entry is non-NULL, the @e qe reference to us
1607    * should simply be set to NULL.
1608    */
1609   struct QueueEntry *qe;
1610
1611   /**
1612    * Client that issued the transmission request, if @e pmt is #PMT_CORE.
1613    */
1614   struct TransportClient *client;
1615
1616   /**
1617    * Head of a MDLL of fragments created for this core message.
1618    */
1619   struct PendingMessage *head_frag;
1620
1621   /**
1622    * Tail of a MDLL of fragments created for this core message.
1623    */
1624   struct PendingMessage *tail_frag;
1625
1626   /**
1627    * Our parent in the fragmentation tree.
1628    */
1629   struct PendingMessage *frag_parent;
1630
1631   /**
1632    * At what time should we give up on the transmission (and no longer retry)?
1633    */
1634   struct GNUNET_TIME_Absolute timeout;
1635
1636   /**
1637    * What is the earliest time for us to retry transmission of this message?
1638    */
1639   struct GNUNET_TIME_Absolute next_attempt;
1640
1641   /**
1642    * UUID to use for this message (used for reassembly of fragments, only
1643    * initialized if @e msg_uuid_set is #GNUNET_YES).
1644    */
1645   struct GNUNET_ShortHashCode msg_uuid;
1646
1647   /**
1648    * Counter incremented per generated fragment.
1649    */
1650   uint32_t frag_uuidgen;
1651
1652   /**
1653    * Type of the pending message.
1654    */
1655   enum PendingMessageType pmt;
1656
1657   /**
1658    * Size of the original message.
1659    */
1660   uint16_t bytes_msg;
1661
1662   /**
1663    * Offset at which we should generate the next fragment.
1664    */
1665   uint16_t frag_off;
1666
1667   /**
1668    * #GNUNET_YES once @e msg_uuid was initialized
1669    */
1670   int16_t msg_uuid_set;
1671
1672   /* Followed by @e bytes_msg to transmit */
1673 };
1674
1675
1676 /**
1677  * One of the addresses of this peer.
1678  */
1679 struct AddressListEntry
1680 {
1681
1682   /**
1683    * Kept in a DLL.
1684    */
1685   struct AddressListEntry *next;
1686
1687   /**
1688    * Kept in a DLL.
1689    */
1690   struct AddressListEntry *prev;
1691
1692   /**
1693    * Which communicator provides this address?
1694    */
1695   struct TransportClient *tc;
1696
1697   /**
1698    * The actual address.
1699    */
1700   const char *address;
1701
1702   /**
1703    * Current context for storing this address in the peerstore.
1704    */
1705   struct GNUNET_PEERSTORE_StoreContext *sc;
1706
1707   /**
1708    * Task to periodically do @e st operation.
1709    */
1710   struct GNUNET_SCHEDULER_Task *st;
1711
1712   /**
1713    * What is a typical lifetime the communicator expects this
1714    * address to have? (Always from now.)
1715    */
1716   struct GNUNET_TIME_Relative expiration;
1717
1718   /**
1719    * Address identifier used by the communicator.
1720    */
1721   uint32_t aid;
1722
1723   /**
1724    * Network type offered by this address.
1725    */
1726   enum GNUNET_NetworkType nt;
1727 };
1728
1729
1730 /**
1731  * Client connected to the transport service.
1732  */
1733 struct TransportClient
1734 {
1735
1736   /**
1737    * Kept in a DLL.
1738    */
1739   struct TransportClient *next;
1740
1741   /**
1742    * Kept in a DLL.
1743    */
1744   struct TransportClient *prev;
1745
1746   /**
1747    * Handle to the client.
1748    */
1749   struct GNUNET_SERVICE_Client *client;
1750
1751   /**
1752    * Message queue to the client.
1753    */
1754   struct GNUNET_MQ_Handle *mq;
1755
1756   /**
1757    * What type of client is this?
1758    */
1759   enum ClientType type;
1760
1761   union
1762   {
1763
1764     /**
1765      * Information for @e type #CT_CORE.
1766      */
1767     struct
1768     {
1769
1770       /**
1771        * Head of list of messages pending for this client, sorted by
1772        * transmission time ("next_attempt" + possibly internal prioritization).
1773        */
1774       struct PendingMessage *pending_msg_head;
1775
1776       /**
1777        * Tail of list of messages pending for this client.
1778        */
1779       struct PendingMessage *pending_msg_tail;
1780
1781     } core;
1782
1783     /**
1784      * Information for @e type #CT_MONITOR.
1785      */
1786     struct
1787     {
1788
1789       /**
1790        * Peer identity to monitor the addresses of.
1791        * Zero to monitor all neighbours.  Valid if
1792        * @e type is #CT_MONITOR.
1793        */
1794       struct GNUNET_PeerIdentity peer;
1795
1796       /**
1797        * Is this a one-shot monitor?
1798        */
1799       int one_shot;
1800
1801     } monitor;
1802
1803
1804     /**
1805      * Information for @e type #CT_COMMUNICATOR.
1806      */
1807     struct
1808     {
1809       /**
1810        * If @e type is #CT_COMMUNICATOR, this communicator
1811        * supports communicating using these addresses.
1812        */
1813       char *address_prefix;
1814
1815       /**
1816        * Head of DLL of queues offered by this communicator.
1817        */
1818       struct Queue *queue_head;
1819
1820       /**
1821        * Tail of DLL of queues offered by this communicator.
1822        */
1823       struct Queue *queue_tail;
1824
1825       /**
1826        * Head of list of the addresses of this peer offered by this
1827        * communicator.
1828        */
1829       struct AddressListEntry *addr_head;
1830
1831       /**
1832        * Tail of list of the addresses of this peer offered by this
1833        * communicator.
1834        */
1835       struct AddressListEntry *addr_tail;
1836
1837       /**
1838        * Number of queue entries in all queues to this communicator. Used
1839        * throttle sending to a communicator if we see that the communicator
1840        * is globally unable to keep up.
1841        */
1842       unsigned int total_queue_length;
1843
1844       /**
1845        * Characteristics of this communicator.
1846        */
1847       enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc;
1848
1849     } communicator;
1850
1851     /**
1852      * Information for @e type #CT_APPLICATION
1853      */
1854     struct
1855     {
1856
1857       /**
1858        * Map of requests for peers the given client application would like to
1859        * see connections for.  Maps from PIDs to `struct PeerRequest`.
1860        */
1861       struct GNUNET_CONTAINER_MultiPeerMap *requests;
1862
1863     } application;
1864
1865   } details;
1866 };
1867
1868
1869 /**
1870  * State we keep for validation activities.  Each of these
1871  * is both in the #validation_heap and the #validation_map.
1872  */
1873 struct ValidationState
1874 {
1875
1876   /**
1877    * For which peer is @a address to be validated (or possibly valid)?
1878    * Serves as key in the #validation_map.
1879    */
1880   struct GNUNET_PeerIdentity pid;
1881
1882   /**
1883    * How long did the peer claim this @e address to be valid? Capped at
1884    * minimum of #MAX_ADDRESS_VALID_UNTIL relative to the time where we last
1885    * were told about the address and the value claimed by the other peer at
1886    * that time.  May be updated similarly when validation succeeds.
1887    */
1888   struct GNUNET_TIME_Absolute valid_until;
1889
1890   /**
1891    * How long do *we* consider this @e address to be valid?
1892    * In the past or zero if we have not yet validated it.
1893    */
1894   struct GNUNET_TIME_Absolute validated_until;
1895
1896   /**
1897    * When did we FIRST use the current @e challenge in a message?
1898    * Used to sanity-check @code{origin_time} in the response when
1899    * calculating the RTT. If the @code{origin_time} is not in
1900    * the expected range, the response is discarded as malicious.
1901    */
1902   struct GNUNET_TIME_Absolute first_challenge_use;
1903
1904   /**
1905    * When did we LAST use the current @e challenge in a message?
1906    * Used to sanity-check @code{origin_time} in the response when
1907    * calculating the RTT.  If the @code{origin_time} is not in
1908    * the expected range, the response is discarded as malicious.
1909    */
1910   struct GNUNET_TIME_Absolute last_challenge_use;
1911
1912   /**
1913    * Next time we will send the @e challenge to the peer, if this time is past
1914    * @e valid_until, this validation state is released at this time.  If the
1915    * address is valid, @e next_challenge is set to @e validated_until MINUS @e
1916    * validation_delay * #VALIDATION_RTT_BUFFER_FACTOR, such that we will try
1917    * to re-validate before the validity actually expires.
1918    */
1919   struct GNUNET_TIME_Absolute next_challenge;
1920
1921   /**
1922    * Current backoff factor we're applying for sending the @a challenge.
1923    * Reset to 0 if the @a challenge is confirmed upon validation.
1924    * Reduced to minimum of #FAST_VALIDATION_CHALLENGE_FREQ and half of the
1925    * existing value if we receive an unvalidated address again over
1926    * another channel (and thus should consider the information "fresh").
1927    * Maximum is #MAX_VALIDATION_CHALLENGE_FREQ.
1928    */
1929   struct GNUNET_TIME_Relative challenge_backoff;
1930
1931   /**
1932    * Initially set to "forever". Once @e validated_until is set, this value is
1933    * set to the RTT that tells us how long it took to receive the validation.
1934    */
1935   struct GNUNET_TIME_Relative validation_rtt;
1936
1937   /**
1938    * The challenge we sent to the peer to get it to validate the address. Note
1939    * that we rotate the challenge whenever we update @e validated_until to
1940    * avoid attacks where a peer simply replays an old challenge in the future.
1941    * (We must not rotate more often as otherwise we may discard valid answers
1942    * due to packet losses, latency and reorderings on the network).
1943    */
1944   struct GNUNET_ShortHashCode challenge;
1945
1946   /**
1947    * Claimed address of the peer.
1948    */
1949   char *address;
1950
1951   /**
1952    * Entry in the #validation_heap, which is sorted by @e next_challenge. The
1953    * heap is used to figure out when the next validation activity should be
1954    * run.
1955    */
1956   struct GNUNET_CONTAINER_HeapNode *hn;
1957
1958   /**
1959    * Handle to a PEERSTORE store operation for this @e address.  NULL if
1960    * no PEERSTORE operation is pending.
1961    */
1962   struct GNUNET_PEERSTORE_StoreContext *sc;
1963
1964   /**
1965    * We are technically ready to send the challenge, but we are waiting for
1966    * the respective queue to become available for transmission.
1967    */
1968   int awaiting_queue;
1969 };
1970
1971
1972 /**
1973  * A Backtalker is a peer sending us backchannel messages. We use this
1974  * struct to detect monotonic time violations, cache ephemeral key
1975  * material (to avoid repeatedly checking signatures), and to synchronize
1976  * monotonic time with the PEERSTORE.
1977  */
1978 struct Backtalker
1979 {
1980   /**
1981    * Peer this is about.
1982    */
1983   struct GNUNET_PeerIdentity pid;
1984
1985   /**
1986    * Last (valid) monotonic time received from this sender.
1987    */
1988   struct GNUNET_TIME_Absolute monotonic_time;
1989
1990   /**
1991    * When will this entry time out?
1992    */
1993   struct GNUNET_TIME_Absolute timeout;
1994
1995   /**
1996    * Last (valid) ephemeral key received from this sender.
1997    */
1998   struct GNUNET_CRYPTO_EcdhePublicKey last_ephemeral;
1999
2000   /**
2001    * Task associated with this backtalker. Can be for timeout,
2002    * or other asynchronous operations.
2003    */
2004   struct GNUNET_SCHEDULER_Task *task;
2005
2006   /**
2007    * Communicator context waiting on this backchannel's @e get, or NULL.
2008    */
2009   struct CommunicatorMessageContext *cmc;
2010
2011   /**
2012    * Handle for an operation to fetch @e monotonic_time information from the
2013    * PEERSTORE, or NULL.
2014    */
2015   struct GNUNET_PEERSTORE_IterateContext *get;
2016
2017   /**
2018    * Handle to a PEERSTORE store operation for this @e pid's @e
2019    * monotonic_time.  NULL if no PEERSTORE operation is pending.
2020    */
2021   struct GNUNET_PEERSTORE_StoreContext *sc;
2022
2023   /**
2024    * Number of bytes of the original message body that follows after this
2025    * struct.
2026    */
2027   size_t body_size;
2028 };
2029
2030
2031 /**
2032  * Head of linked list of all clients to this service.
2033  */
2034 static struct TransportClient *clients_head;
2035
2036 /**
2037  * Tail of linked list of all clients to this service.
2038  */
2039 static struct TransportClient *clients_tail;
2040
2041 /**
2042  * Statistics handle.
2043  */
2044 static struct GNUNET_STATISTICS_Handle *GST_stats;
2045
2046 /**
2047  * Configuration handle.
2048  */
2049 static const struct GNUNET_CONFIGURATION_Handle *GST_cfg;
2050
2051 /**
2052  * Our public key.
2053  */
2054 static struct GNUNET_PeerIdentity GST_my_identity;
2055
2056 /**
2057  * Our private key.
2058  */
2059 static struct GNUNET_CRYPTO_EddsaPrivateKey *GST_my_private_key;
2060
2061 /**
2062  * Map from PIDs to `struct Neighbour` entries.  A peer is
2063  * a neighbour if we have an MQ to it from some communicator.
2064  */
2065 static struct GNUNET_CONTAINER_MultiPeerMap *neighbours;
2066
2067 /**
2068  * Map from PIDs to `struct Backtalker` entries.  A peer is
2069  * a backtalker if it recently send us backchannel messages.
2070  */
2071 static struct GNUNET_CONTAINER_MultiPeerMap *backtalkers;
2072
2073 /**
2074  * Map from PIDs to `struct DistanceVector` entries describing
2075  * known paths to the peer.
2076  */
2077 static struct GNUNET_CONTAINER_MultiPeerMap *dv_routes;
2078
2079 /**
2080  * Map from PIDs to `struct ValidationState` entries describing
2081  * addresses we are aware of and their validity state.
2082  */
2083 static struct GNUNET_CONTAINER_MultiPeerMap *validation_map;
2084
2085 /**
2086  * Map from challenges to `struct LearnLaunchEntry` values.
2087  */
2088 static struct GNUNET_CONTAINER_MultiShortmap *dvlearn_map;
2089
2090 /**
2091  * Head of a DLL sorted by launch time.
2092  */
2093 static struct LearnLaunchEntry *lle_head;
2094
2095 /**
2096  * Tail of a DLL sorted by launch time.
2097  */
2098 static struct LearnLaunchEntry *lle_tail;
2099
2100 /**
2101  * MIN Heap sorted by "next_challenge" to `struct ValidationState` entries
2102  * sorting addresses we are aware of by when we should next try to (re)validate
2103  * (or expire) them.
2104  */
2105 static struct GNUNET_CONTAINER_Heap *validation_heap;
2106
2107 /**
2108  * Database for peer's HELLOs.
2109  */
2110 static struct GNUNET_PEERSTORE_Handle *peerstore;
2111
2112 /**
2113  * Heap sorting `struct EphemeralCacheEntry` by their
2114  * key/signature validity.
2115  */
2116 static struct GNUNET_CONTAINER_Heap *ephemeral_heap;
2117
2118 /**
2119  * Hash map for looking up `struct EphemeralCacheEntry`s
2120  * by peer identity. (We may have ephemerals in our
2121  * cache for which we do not have a neighbour entry,
2122  * and similar many neighbours may not need ephemerals,
2123  * so we use a second map.)
2124  */
2125 static struct GNUNET_CONTAINER_MultiPeerMap *ephemeral_map;
2126
2127 /**
2128  * Task to free expired ephemerals.
2129  */
2130 static struct GNUNET_SCHEDULER_Task *ephemeral_task;
2131
2132 /**
2133  * Task run to initiate DV learning.
2134  */
2135 static struct GNUNET_SCHEDULER_Task *dvlearn_task;
2136
2137 /**
2138  * Task to run address validation.
2139  */
2140 static struct GNUNET_SCHEDULER_Task *validation_task;
2141
2142
2143 /**
2144  * Free cached ephemeral key.
2145  *
2146  * @param ece cached signature to free
2147  */
2148 static void
2149 free_ephemeral (struct EphemeralCacheEntry *ece)
2150 {
2151   GNUNET_CONTAINER_multipeermap_remove (ephemeral_map, &ece->target, ece);
2152   GNUNET_CONTAINER_heap_remove_node (ece->hn);
2153   GNUNET_free (ece);
2154 }
2155
2156
2157 /**
2158  * Free validation state.
2159  *
2160  * @param vs validation state to free
2161  */
2162 static void
2163 free_validation_state (struct ValidationState *vs)
2164 {
2165   GNUNET_CONTAINER_multipeermap_remove (validation_map, &vs->pid, vs);
2166   GNUNET_CONTAINER_heap_remove_node (vs->hn);
2167   vs->hn = NULL;
2168   if (NULL != vs->sc)
2169   {
2170     GNUNET_PEERSTORE_store_cancel (vs->sc);
2171     vs->sc = NULL;
2172   }
2173   GNUNET_free (vs->address);
2174   GNUNET_free (vs);
2175 }
2176
2177
2178 /**
2179  * Lookup neighbour record for peer @a pid.
2180  *
2181  * @param pid neighbour to look for
2182  * @return NULL if we do not have this peer as a neighbour
2183  */
2184 static struct Neighbour *
2185 lookup_neighbour (const struct GNUNET_PeerIdentity *pid)
2186 {
2187   return GNUNET_CONTAINER_multipeermap_get (neighbours, pid);
2188 }
2189
2190
2191 /**
2192  * Details about what to notify monitors about.
2193  */
2194 struct MonitorEvent
2195 {
2196   /**
2197    * @deprecated To be discussed if we keep these...
2198    */
2199   struct GNUNET_TIME_Absolute last_validation;
2200   struct GNUNET_TIME_Absolute valid_until;
2201   struct GNUNET_TIME_Absolute next_validation;
2202
2203   /**
2204    * Current round-trip time estimate.
2205    */
2206   struct GNUNET_TIME_Relative rtt;
2207
2208   /**
2209    * Connection status.
2210    */
2211   enum GNUNET_TRANSPORT_ConnectionStatus cs;
2212
2213   /**
2214    * Messages pending.
2215    */
2216   uint32_t num_msg_pending;
2217
2218   /**
2219    * Bytes pending.
2220    */
2221   uint32_t num_bytes_pending;
2222 };
2223
2224
2225 /**
2226  * Free a @dvh. Callers MAY want to check if this was the last path to the
2227  * `target`, and if so call #free_dv_route to also free the associated DV
2228  * entry in #dv_routes (if not, the associated scheduler job should eventually
2229  * take care of it).
2230  *
2231  * @param dvh hop to free
2232  */
2233 static void
2234 free_distance_vector_hop (struct DistanceVectorHop *dvh)
2235 {
2236   struct Neighbour *n = dvh->next_hop;
2237   struct DistanceVector *dv = dvh->dv;
2238
2239   GNUNET_CONTAINER_MDLL_remove (neighbour, n->dv_head, n->dv_tail, dvh);
2240   GNUNET_CONTAINER_MDLL_remove (dv, dv->dv_head, dv->dv_tail, dvh);
2241   GNUNET_free (dvh);
2242 }
2243
2244
2245 /**
2246  * Free entry in #dv_routes.  First frees all hops to the target, and
2247  * if there are no entries left, frees @a dv as well.
2248  *
2249  * @param dv route to free
2250  */
2251 static void
2252 free_dv_route (struct DistanceVector *dv)
2253 {
2254   struct DistanceVectorHop *dvh;
2255
2256   while (NULL != (dvh = dv->dv_head))
2257     free_distance_vector_hop (dvh);
2258   if (NULL == dv->dv_head)
2259   {
2260     GNUNET_assert (
2261       GNUNET_YES ==
2262       GNUNET_CONTAINER_multipeermap_remove (dv_routes, &dv->target, dv));
2263     if (NULL != dv->visibility_task)
2264       GNUNET_SCHEDULER_cancel (dv->visibility_task);
2265     if (NULL != dv->timeout_task)
2266       GNUNET_SCHEDULER_cancel (dv->timeout_task);
2267     GNUNET_free (dv);
2268   }
2269 }
2270
2271
2272 /**
2273  * Notify monitor @a tc about an event.  That @a tc
2274  * cares about the event has already been checked.
2275  *
2276  * Send @a tc information in @a me about a @a peer's status with
2277  * respect to some @a address to all monitors that care.
2278  *
2279  * @param tc monitor to inform
2280  * @param peer peer the information is about
2281  * @param address address the information is about
2282  * @param nt network type associated with @a address
2283  * @param me detailed information to transmit
2284  */
2285 static void
2286 notify_monitor (struct TransportClient *tc,
2287                 const struct GNUNET_PeerIdentity *peer,
2288                 const char *address,
2289                 enum GNUNET_NetworkType nt,
2290                 const struct MonitorEvent *me)
2291 {
2292   struct GNUNET_MQ_Envelope *env;
2293   struct GNUNET_TRANSPORT_MonitorData *md;
2294   size_t addr_len = strlen (address) + 1;
2295
2296   env = GNUNET_MQ_msg_extra (md,
2297                              addr_len,
2298                              GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_DATA);
2299   md->nt = htonl ((uint32_t) nt);
2300   md->peer = *peer;
2301   md->last_validation = GNUNET_TIME_absolute_hton (me->last_validation);
2302   md->valid_until = GNUNET_TIME_absolute_hton (me->valid_until);
2303   md->next_validation = GNUNET_TIME_absolute_hton (me->next_validation);
2304   md->rtt = GNUNET_TIME_relative_hton (me->rtt);
2305   md->cs = htonl ((uint32_t) me->cs);
2306   md->num_msg_pending = htonl (me->num_msg_pending);
2307   md->num_bytes_pending = htonl (me->num_bytes_pending);
2308   memcpy (&md[1], address, addr_len);
2309   GNUNET_MQ_send (tc->mq, env);
2310 }
2311
2312
2313 /**
2314  * Send information in @a me about a @a peer's status with respect
2315  * to some @a address to all monitors that care.
2316  *
2317  * @param peer peer the information is about
2318  * @param address address the information is about
2319  * @param nt network type associated with @a address
2320  * @param me detailed information to transmit
2321  */
2322 static void
2323 notify_monitors (const struct GNUNET_PeerIdentity *peer,
2324                  const char *address,
2325                  enum GNUNET_NetworkType nt,
2326                  const struct MonitorEvent *me)
2327 {
2328   for (struct TransportClient *tc = clients_head; NULL != tc; tc = tc->next)
2329   {
2330     if (CT_MONITOR != tc->type)
2331       continue;
2332     if (tc->details.monitor.one_shot)
2333       continue;
2334     if ((0 != GNUNET_is_zero (&tc->details.monitor.peer)) &&
2335         (0 != GNUNET_memcmp (&tc->details.monitor.peer, peer)))
2336       continue;
2337     notify_monitor (tc, peer, address, nt, me);
2338   }
2339 }
2340
2341
2342 /**
2343  * Called whenever a client connects.  Allocates our
2344  * data structures associated with that client.
2345  *
2346  * @param cls closure, NULL
2347  * @param client identification of the client
2348  * @param mq message queue for the client
2349  * @return our `struct TransportClient`
2350  */
2351 static void *
2352 client_connect_cb (void *cls,
2353                    struct GNUNET_SERVICE_Client *client,
2354                    struct GNUNET_MQ_Handle *mq)
2355 {
2356   struct TransportClient *tc;
2357
2358   (void) cls;
2359   tc = GNUNET_new (struct TransportClient);
2360   tc->client = client;
2361   tc->mq = mq;
2362   GNUNET_CONTAINER_DLL_insert (clients_head, clients_tail, tc);
2363   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client %p connected\n", tc);
2364   return tc;
2365 }
2366
2367
2368 /**
2369  * Free @a rc
2370  *
2371  * @param rc data structure to free
2372  */
2373 static void
2374 free_reassembly_context (struct ReassemblyContext *rc)
2375 {
2376   struct Neighbour *n = rc->neighbour;
2377
2378   GNUNET_assert (rc == GNUNET_CONTAINER_heap_remove_node (rc->hn));
2379   GNUNET_assert (GNUNET_OK ==
2380                  GNUNET_CONTAINER_multishortmap_remove (n->reassembly_map,
2381                                                         &rc->msg_uuid,
2382                                                         rc));
2383   GNUNET_free (rc);
2384 }
2385
2386
2387 /**
2388  * Task run to clean up reassembly context of a neighbour that have expired.
2389  *
2390  * @param cls a `struct Neighbour`
2391  */
2392 static void
2393 reassembly_cleanup_task (void *cls)
2394 {
2395   struct Neighbour *n = cls;
2396   struct ReassemblyContext *rc;
2397
2398   n->reassembly_timeout_task = NULL;
2399   while (NULL != (rc = GNUNET_CONTAINER_heap_peek (n->reassembly_heap)))
2400   {
2401     if (0 == GNUNET_TIME_absolute_get_remaining (rc->reassembly_timeout)
2402                .rel_value_us)
2403     {
2404       free_reassembly_context (rc);
2405       continue;
2406     }
2407     GNUNET_assert (NULL == n->reassembly_timeout_task);
2408     n->reassembly_timeout_task =
2409       GNUNET_SCHEDULER_add_at (rc->reassembly_timeout,
2410                                &reassembly_cleanup_task,
2411                                n);
2412     return;
2413   }
2414 }
2415
2416
2417 /**
2418  * function called to #free_reassembly_context().
2419  *
2420  * @param cls NULL
2421  * @param key unused
2422  * @param value a `struct ReassemblyContext` to free
2423  * @return #GNUNET_OK (continue iteration)
2424  */
2425 static int
2426 free_reassembly_cb (void *cls,
2427                     const struct GNUNET_ShortHashCode *key,
2428                     void *value)
2429 {
2430   struct ReassemblyContext *rc = value;
2431
2432   (void) cls;
2433   (void) key;
2434   free_reassembly_context (rc);
2435   return GNUNET_OK;
2436 }
2437
2438
2439 /**
2440  * Release memory used by @a neighbour.
2441  *
2442  * @param neighbour neighbour entry to free
2443  */
2444 static void
2445 free_neighbour (struct Neighbour *neighbour)
2446 {
2447   struct DistanceVectorHop *dvh;
2448
2449   GNUNET_assert (NULL == neighbour->queue_head);
2450   GNUNET_assert (GNUNET_YES ==
2451                  GNUNET_CONTAINER_multipeermap_remove (neighbours,
2452                                                        &neighbour->pid,
2453                                                        neighbour));
2454   if (NULL != neighbour->timeout_task)
2455     GNUNET_SCHEDULER_cancel (neighbour->timeout_task);
2456   if (NULL != neighbour->reassembly_map)
2457   {
2458     GNUNET_CONTAINER_multishortmap_iterate (neighbour->reassembly_map,
2459                                             &free_reassembly_cb,
2460                                             NULL);
2461     GNUNET_CONTAINER_multishortmap_destroy (neighbour->reassembly_map);
2462     neighbour->reassembly_map = NULL;
2463     GNUNET_CONTAINER_heap_destroy (neighbour->reassembly_heap);
2464     neighbour->reassembly_heap = NULL;
2465   }
2466   while (NULL != (dvh = neighbour->dv_head))
2467   {
2468     struct DistanceVector *dv = dvh->dv;
2469
2470     free_distance_vector_hop (dvh);
2471     if (NULL == dv->dv_head)
2472       free_dv_route (dv);
2473   }
2474   if (NULL != neighbour->reassembly_timeout_task)
2475     GNUNET_SCHEDULER_cancel (neighbour->reassembly_timeout_task);
2476   GNUNET_free (neighbour);
2477 }
2478
2479
2480 /**
2481  * Send message to CORE clients that we lost a connection.
2482  *
2483  * @param tc client to inform (must be CORE client)
2484  * @param pid peer the connection is for
2485  * @param quota_out current quota for the peer
2486  */
2487 static void
2488 core_send_connect_info (struct TransportClient *tc,
2489                         const struct GNUNET_PeerIdentity *pid,
2490                         struct GNUNET_BANDWIDTH_Value32NBO quota_out)
2491 {
2492   struct GNUNET_MQ_Envelope *env;
2493   struct ConnectInfoMessage *cim;
2494
2495   GNUNET_assert (CT_CORE == tc->type);
2496   env = GNUNET_MQ_msg (cim, GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT);
2497   cim->quota_out = quota_out;
2498   cim->id = *pid;
2499   GNUNET_MQ_send (tc->mq, env);
2500 }
2501
2502
2503 /**
2504  * Send message to CORE clients that we gained a connection
2505  *
2506  * @param pid peer the queue was for
2507  * @param quota_out current quota for the peer
2508  */
2509 static void
2510 cores_send_connect_info (const struct GNUNET_PeerIdentity *pid,
2511                          struct GNUNET_BANDWIDTH_Value32NBO quota_out)
2512 {
2513   for (struct TransportClient *tc = clients_head; NULL != tc; tc = tc->next)
2514   {
2515     if (CT_CORE != tc->type)
2516       continue;
2517     core_send_connect_info (tc, pid, quota_out);
2518   }
2519 }
2520
2521
2522 /**
2523  * Send message to CORE clients that we lost a connection.
2524  *
2525  * @param pid peer the connection was for
2526  */
2527 static void
2528 cores_send_disconnect_info (const struct GNUNET_PeerIdentity *pid)
2529 {
2530   for (struct TransportClient *tc = clients_head; NULL != tc; tc = tc->next)
2531   {
2532     struct GNUNET_MQ_Envelope *env;
2533     struct DisconnectInfoMessage *dim;
2534
2535     if (CT_CORE != tc->type)
2536       continue;
2537     env = GNUNET_MQ_msg (dim, GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT);
2538     dim->peer = *pid;
2539     GNUNET_MQ_send (tc->mq, env);
2540   }
2541 }
2542
2543
2544 /**
2545  * We believe we are ready to transmit a message on a queue. Double-checks
2546  * with the queue's "tracker_out" and then gives the message to the
2547  * communicator for transmission (updating the tracker, and re-scheduling
2548  * itself if applicable).
2549  *
2550  * @param cls the `struct Queue` to process transmissions for
2551  */
2552 static void
2553 transmit_on_queue (void *cls);
2554
2555
2556 /**
2557  * Schedule next run of #transmit_on_queue().  Does NOTHING if
2558  * we should run immediately or if the message queue is empty.
2559  * Test for no task being added AND queue not being empty to
2560  * transmit immediately afterwards!  This function must only
2561  * be called if the message queue is non-empty!
2562  *
2563  * @param queue the queue to do scheduling for
2564  * @param inside_job set to #GNUNET_YES if called from
2565  *            #transmit_on_queue() itself and NOT setting
2566  *            the task means running immediately
2567  */
2568 static void
2569 schedule_transmit_on_queue (struct Queue *queue, int inside_job)
2570 {
2571   struct Neighbour *n = queue->neighbour;
2572   struct PendingMessage *pm = n->pending_msg_head;
2573   struct GNUNET_TIME_Relative out_delay;
2574   unsigned int wsize;
2575
2576   GNUNET_assert (NULL != pm);
2577   if (queue->tc->details.communicator.total_queue_length >=
2578       COMMUNICATOR_TOTAL_QUEUE_LIMIT)
2579   {
2580     GNUNET_STATISTICS_update (
2581       GST_stats,
2582       "# Transmission throttled due to communicator queue limit",
2583       1,
2584       GNUNET_NO);
2585     return;
2586   }
2587   if (queue->queue_length >= QUEUE_LENGTH_LIMIT)
2588   {
2589     GNUNET_STATISTICS_update (GST_stats,
2590                               "# Transmission throttled due to queue queue limit",
2591                               1,
2592                               GNUNET_NO);
2593     return;
2594   }
2595
2596   wsize = (0 == queue->mtu) ? pm->bytes_msg /* FIXME: add overheads? */
2597                             : queue->mtu;
2598   out_delay = GNUNET_BANDWIDTH_tracker_get_delay (&queue->tracker_out, wsize);
2599   out_delay = GNUNET_TIME_relative_max (GNUNET_TIME_absolute_get_remaining (
2600                                           pm->next_attempt),
2601                                         out_delay);
2602   if ((GNUNET_YES == inside_job) && (0 == out_delay.rel_value_us))
2603     return; /* we should run immediately! */
2604   /* queue has changed since we were scheduled, reschedule again */
2605   queue->transmit_task =
2606     GNUNET_SCHEDULER_add_delayed (out_delay, &transmit_on_queue, queue);
2607   if (out_delay.rel_value_us > DELAY_WARN_THRESHOLD.rel_value_us)
2608     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2609                 "Next transmission on queue `%s' in %s (high delay)\n",
2610                 queue->address,
2611                 GNUNET_STRINGS_relative_time_to_string (out_delay, GNUNET_YES));
2612   else
2613     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2614                 "Next transmission on queue `%s' in %s\n",
2615                 queue->address,
2616                 GNUNET_STRINGS_relative_time_to_string (out_delay, GNUNET_YES));
2617 }
2618
2619
2620 /**
2621  * Check whether the CORE visibility of @a n changed. If so,
2622  * check whether we need to notify CORE.
2623  *
2624  * @param n neighbour to perform the check for
2625  */
2626 static void
2627 update_neighbour_core_visibility (struct Neighbour *n);
2628
2629
2630 /**
2631  * Free @a queue.
2632  *
2633  * @param queue the queue to free
2634  */
2635 static void
2636 free_queue (struct Queue *queue)
2637 {
2638   struct Neighbour *neighbour = queue->neighbour;
2639   struct TransportClient *tc = queue->tc;
2640   struct MonitorEvent me = {.cs = GNUNET_TRANSPORT_CS_DOWN,
2641                             .rtt = GNUNET_TIME_UNIT_FOREVER_REL};
2642   struct QueueEntry *qe;
2643   int maxxed;
2644
2645   if (NULL != queue->transmit_task)
2646   {
2647     GNUNET_SCHEDULER_cancel (queue->transmit_task);
2648     queue->transmit_task = NULL;
2649   }
2650   if (NULL != queue->visibility_task)
2651   {
2652     GNUNET_SCHEDULER_cancel (queue->visibility_task);
2653     queue->visibility_task = NULL;
2654   }
2655   GNUNET_CONTAINER_MDLL_remove (neighbour,
2656                                 neighbour->queue_head,
2657                                 neighbour->queue_tail,
2658                                 queue);
2659   GNUNET_CONTAINER_MDLL_remove (client,
2660                                 tc->details.communicator.queue_head,
2661                                 tc->details.communicator.queue_tail,
2662                                 queue);
2663   maxxed = (COMMUNICATOR_TOTAL_QUEUE_LIMIT >=
2664             tc->details.communicator.total_queue_length);
2665   while (NULL != (qe = queue->queue_head))
2666   {
2667     GNUNET_CONTAINER_DLL_remove (queue->queue_head, queue->queue_tail, qe);
2668     queue->queue_length--;
2669     tc->details.communicator.total_queue_length--;
2670     if (NULL != qe->pm)
2671     {
2672       GNUNET_assert (qe == qe->pm->qe);
2673       qe->pm->qe = NULL;
2674     }
2675     GNUNET_free (qe);
2676   }
2677   GNUNET_assert (0 == queue->queue_length);
2678   if ((maxxed) && (COMMUNICATOR_TOTAL_QUEUE_LIMIT <
2679                    tc->details.communicator.total_queue_length))
2680   {
2681     /* Communicator dropped below threshold, resume all queues */
2682     GNUNET_STATISTICS_update (
2683       GST_stats,
2684       "# Transmission throttled due to communicator queue limit",
2685       -1,
2686       GNUNET_NO);
2687     for (struct Queue *s = tc->details.communicator.queue_head; NULL != s;
2688          s = s->next_client)
2689       schedule_transmit_on_queue (s, GNUNET_NO);
2690   }
2691   notify_monitors (&neighbour->pid, queue->address, queue->nt, &me);
2692   GNUNET_BANDWIDTH_tracker_notification_stop (&queue->tracker_in);
2693   GNUNET_BANDWIDTH_tracker_notification_stop (&queue->tracker_out);
2694   GNUNET_free (queue);
2695
2696   update_neighbour_core_visibility (neighbour);
2697   cores_send_disconnect_info (&neighbour->pid);
2698
2699   if (NULL == neighbour->queue_head)
2700   {
2701     free_neighbour (neighbour);
2702   }
2703 }
2704
2705
2706 /**
2707  * Free @a ale
2708  *
2709  * @param ale address list entry to free
2710  */
2711 static void
2712 free_address_list_entry (struct AddressListEntry *ale)
2713 {
2714   struct TransportClient *tc = ale->tc;
2715
2716   GNUNET_CONTAINER_DLL_remove (tc->details.communicator.addr_head,
2717                                tc->details.communicator.addr_tail,
2718                                ale);
2719   if (NULL != ale->sc)
2720   {
2721     GNUNET_PEERSTORE_store_cancel (ale->sc);
2722     ale->sc = NULL;
2723   }
2724   if (NULL != ale->st)
2725   {
2726     GNUNET_SCHEDULER_cancel (ale->st);
2727     ale->st = NULL;
2728   }
2729   GNUNET_free (ale);
2730 }
2731
2732
2733 /**
2734  * Stop the peer request in @a value.
2735  *
2736  * @param cls a `struct TransportClient` that no longer makes the request
2737  * @param pid the peer's identity
2738  * @param value a `struct PeerRequest`
2739  * @return #GNUNET_YES (always)
2740  */
2741 static int
2742 stop_peer_request (void *cls,
2743                    const struct GNUNET_PeerIdentity *pid,
2744                    void *value)
2745 {
2746   struct TransportClient *tc = cls;
2747   struct PeerRequest *pr = value;
2748
2749   GNUNET_PEERSTORE_watch_cancel (pr->wc);
2750   GNUNET_assert (
2751     GNUNET_YES ==
2752     GNUNET_CONTAINER_multipeermap_remove (tc->details.application.requests,
2753                                           pid,
2754                                           pr));
2755   GNUNET_free (pr);
2756
2757   return GNUNET_OK;
2758 }
2759
2760
2761 /**
2762  * Called whenever a client is disconnected.  Frees our
2763  * resources associated with that client.
2764  *
2765  * @param cls closure, NULL
2766  * @param client identification of the client
2767  * @param app_ctx our `struct TransportClient`
2768  */
2769 static void
2770 client_disconnect_cb (void *cls,
2771                       struct GNUNET_SERVICE_Client *client,
2772                       void *app_ctx)
2773 {
2774   struct TransportClient *tc = app_ctx;
2775
2776   (void) cls;
2777   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2778               "Client %p disconnected, cleaning up.\n",
2779               tc);
2780   GNUNET_CONTAINER_DLL_remove (clients_head, clients_tail, tc);
2781   switch (tc->type)
2782   {
2783   case CT_NONE:
2784     break;
2785   case CT_CORE: {
2786     struct PendingMessage *pm;
2787
2788     while (NULL != (pm = tc->details.core.pending_msg_head))
2789     {
2790       GNUNET_CONTAINER_MDLL_remove (client,
2791                                     tc->details.core.pending_msg_head,
2792                                     tc->details.core.pending_msg_tail,
2793                                     pm);
2794       pm->client = NULL;
2795     }
2796   }
2797   break;
2798   case CT_MONITOR:
2799     break;
2800   case CT_COMMUNICATOR: {
2801     struct Queue *q;
2802     struct AddressListEntry *ale;
2803
2804     while (NULL != (q = tc->details.communicator.queue_head))
2805       free_queue (q);
2806     while (NULL != (ale = tc->details.communicator.addr_head))
2807       free_address_list_entry (ale);
2808     GNUNET_free (tc->details.communicator.address_prefix);
2809   }
2810   break;
2811   case CT_APPLICATION:
2812     GNUNET_CONTAINER_multipeermap_iterate (tc->details.application.requests,
2813                                            &stop_peer_request,
2814                                            tc);
2815     GNUNET_CONTAINER_multipeermap_destroy (tc->details.application.requests);
2816     break;
2817   }
2818   GNUNET_free (tc);
2819 }
2820
2821
2822 /**
2823  * Iterator telling new CORE client about all existing
2824  * connections to peers.
2825  *
2826  * @param cls the new `struct TransportClient`
2827  * @param pid a connected peer
2828  * @param value the `struct Neighbour` with more information
2829  * @return #GNUNET_OK (continue to iterate)
2830  */
2831 static int
2832 notify_client_connect_info (void *cls,
2833                             const struct GNUNET_PeerIdentity *pid,
2834                             void *value)
2835 {
2836   struct TransportClient *tc = cls;
2837   struct Neighbour *neighbour = value;
2838
2839   core_send_connect_info (tc, pid, neighbour->quota_out);
2840   return GNUNET_OK;
2841 }
2842
2843
2844 /**
2845  * Initialize a "CORE" client.  We got a start message from this
2846  * client, so add it to the list of clients for broadcasting of
2847  * inbound messages.
2848  *
2849  * @param cls the client
2850  * @param start the start message that was sent
2851  */
2852 static void
2853 handle_client_start (void *cls, const struct StartMessage *start)
2854 {
2855   struct TransportClient *tc = cls;
2856   uint32_t options;
2857
2858   options = ntohl (start->options);
2859   if ((0 != (1 & options)) &&
2860       (0 != GNUNET_memcmp (&start->self, &GST_my_identity)))
2861   {
2862     /* client thinks this is a different peer, reject */
2863     GNUNET_break (0);
2864     GNUNET_SERVICE_client_drop (tc->client);
2865     return;
2866   }
2867   if (CT_NONE != tc->type)
2868   {
2869     GNUNET_break (0);
2870     GNUNET_SERVICE_client_drop (tc->client);
2871     return;
2872   }
2873   tc->type = CT_CORE;
2874   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
2875                                          &notify_client_connect_info,
2876                                          tc);
2877   GNUNET_SERVICE_client_continue (tc->client);
2878 }
2879
2880
2881 /**
2882  * Client asked for transmission to a peer.  Process the request.
2883  *
2884  * @param cls the client
2885  * @param obm the send message that was sent
2886  */
2887 static int
2888 check_client_send (void *cls, const struct OutboundMessage *obm)
2889 {
2890   struct TransportClient *tc = cls;
2891   uint16_t size;
2892   const struct GNUNET_MessageHeader *obmm;
2893
2894   if (CT_CORE != tc->type)
2895   {
2896     GNUNET_break (0);
2897     return GNUNET_SYSERR;
2898   }
2899   size = ntohs (obm->header.size) - sizeof (struct OutboundMessage);
2900   if (size < sizeof (struct GNUNET_MessageHeader))
2901   {
2902     GNUNET_break (0);
2903     return GNUNET_SYSERR;
2904   }
2905   obmm = (const struct GNUNET_MessageHeader *) &obm[1];
2906   if (size != ntohs (obmm->size))
2907   {
2908     GNUNET_break (0);
2909     return GNUNET_SYSERR;
2910   }
2911   return GNUNET_OK;
2912 }
2913
2914
2915 /**
2916  * Free fragment tree below @e root, excluding @e root itself.
2917  *
2918  * @param root root of the tree to free
2919  */
2920 static void
2921 free_fragment_tree (struct PendingMessage *root)
2922 {
2923   struct PendingMessage *frag;
2924
2925   while (NULL != (frag = root->head_frag))
2926   {
2927     free_fragment_tree (frag);
2928     GNUNET_CONTAINER_MDLL_remove (frag, root->head_frag, root->tail_frag, frag);
2929     GNUNET_free (frag);
2930   }
2931 }
2932
2933
2934 /**
2935  * Release memory associated with @a pm and remove @a pm from associated
2936  * data structures.  @a pm must be a top-level pending message and not
2937  * a fragment in the tree.  The entire tree is freed (if applicable).
2938  *
2939  * @param pm the pending message to free
2940  */
2941 static void
2942 free_pending_message (struct PendingMessage *pm)
2943 {
2944   struct TransportClient *tc = pm->client;
2945   struct Neighbour *target = pm->target;
2946
2947   if (NULL != tc)
2948   {
2949     GNUNET_CONTAINER_MDLL_remove (client,
2950                                   tc->details.core.pending_msg_head,
2951                                   tc->details.core.pending_msg_tail,
2952                                   pm);
2953   }
2954   GNUNET_CONTAINER_MDLL_remove (neighbour,
2955                                 target->pending_msg_head,
2956                                 target->pending_msg_tail,
2957                                 pm);
2958   free_fragment_tree (pm);
2959   if (NULL != pm->qe)
2960   {
2961     GNUNET_assert (pm == pm->qe->pm);
2962     pm->qe->pm = NULL;
2963   }
2964   GNUNET_free_non_null (pm->bpm);
2965   GNUNET_free (pm);
2966 }
2967
2968
2969 /**
2970  * Send a response to the @a pm that we have processed a
2971  * "send" request with status @a success. We
2972  * transmitted @a bytes_physical on the actual wire.
2973  * Sends a confirmation to the "core" client responsible
2974  * for the original request and free's @a pm.
2975  *
2976  * @param pm handle to the original pending message
2977  * @param success status code, #GNUNET_OK on success, #GNUNET_SYSERR
2978  *          for transmission failure
2979  * @param bytes_physical amount of bandwidth consumed
2980  */
2981 static void
2982 client_send_response (struct PendingMessage *pm,
2983                       int success,
2984                       uint32_t bytes_physical)
2985 {
2986   struct TransportClient *tc = pm->client;
2987   struct Neighbour *target = pm->target;
2988   struct GNUNET_MQ_Envelope *env;
2989   struct SendOkMessage *som;
2990
2991   if (NULL != tc)
2992   {
2993     env = GNUNET_MQ_msg (som, GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
2994     som->success = htonl ((uint32_t) success);
2995     som->bytes_msg = htons (pm->bytes_msg);
2996     som->bytes_physical = htonl (bytes_physical);
2997     som->peer = target->pid;
2998     GNUNET_MQ_send (tc->mq, env);
2999   }
3000   free_pending_message (pm);
3001 }
3002
3003
3004 /**
3005  * Checks the message queue for a neighbour for messages that have timed
3006  * out and purges them.
3007  *
3008  * @param cls a `struct Neighbour`
3009  */
3010 static void
3011 check_queue_timeouts (void *cls)
3012 {
3013   struct Neighbour *n = cls;
3014   struct PendingMessage *pm;
3015   struct GNUNET_TIME_Absolute now;
3016   struct GNUNET_TIME_Absolute earliest_timeout;
3017
3018   n->timeout_task = NULL;
3019   earliest_timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
3020   now = GNUNET_TIME_absolute_get ();
3021   for (struct PendingMessage *pos = n->pending_msg_head; NULL != pos; pos = pm)
3022   {
3023     pm = pos->next_neighbour;
3024     if (pos->timeout.abs_value_us <= now.abs_value_us)
3025     {
3026       GNUNET_STATISTICS_update (GST_stats,
3027                                 "# messages dropped (timeout before confirmation)",
3028                                 1,
3029                                 GNUNET_NO);
3030       client_send_response (pm, GNUNET_NO, 0);
3031       continue;
3032     }
3033     earliest_timeout =
3034       GNUNET_TIME_absolute_min (earliest_timeout, pos->timeout);
3035   }
3036   n->earliest_timeout = earliest_timeout;
3037   if (NULL != n->pending_msg_head)
3038     n->timeout_task =
3039       GNUNET_SCHEDULER_add_at (earliest_timeout, &check_queue_timeouts, n);
3040 }
3041
3042
3043 /**
3044  * Client asked for transmission to a peer.  Process the request.
3045  *
3046  * @param cls the client
3047  * @param obm the send message that was sent
3048  */
3049 static void
3050 handle_client_send (void *cls, const struct OutboundMessage *obm)
3051 {
3052   struct TransportClient *tc = cls;
3053   struct PendingMessage *pm;
3054   const struct GNUNET_MessageHeader *obmm;
3055   struct Neighbour *target;
3056   uint32_t bytes_msg;
3057   int was_empty;
3058
3059   GNUNET_assert (CT_CORE == tc->type);
3060   obmm = (const struct GNUNET_MessageHeader *) &obm[1];
3061   bytes_msg = ntohs (obmm->size);
3062   target = lookup_neighbour (&obm->peer);
3063   if (NULL == target)
3064   {
3065     /* Failure: don't have this peer as a neighbour (anymore).
3066        Might have gone down asynchronously, so this is NOT
3067        a protocol violation by CORE. Still count the event,
3068        as this should be rare. */
3069     struct GNUNET_MQ_Envelope *env;
3070     struct SendOkMessage *som;
3071
3072     env = GNUNET_MQ_msg (som, GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK);
3073     som->success = htonl (GNUNET_SYSERR);
3074     som->bytes_msg = htonl (bytes_msg);
3075     som->bytes_physical = htonl (0);
3076     som->peer = obm->peer;
3077     GNUNET_MQ_send (tc->mq, env);
3078     GNUNET_SERVICE_client_continue (tc->client);
3079     GNUNET_STATISTICS_update (GST_stats,
3080                               "# messages dropped (neighbour unknown)",
3081                               1,
3082                               GNUNET_NO);
3083     return;
3084   }
3085   was_empty = (NULL == target->pending_msg_head);
3086   pm = GNUNET_malloc (sizeof (struct PendingMessage) + bytes_msg);
3087   pm->client = tc;
3088   pm->target = target;
3089   pm->bytes_msg = bytes_msg;
3090   pm->timeout =
3091     GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_ntoh (obm->timeout));
3092   memcpy (&pm[1], &obm[1], bytes_msg);
3093   GNUNET_CONTAINER_MDLL_insert (neighbour,
3094                                 target->pending_msg_head,
3095                                 target->pending_msg_tail,
3096                                 pm);
3097   GNUNET_CONTAINER_MDLL_insert (client,
3098                                 tc->details.core.pending_msg_head,
3099                                 tc->details.core.pending_msg_tail,
3100                                 pm);
3101   if (target->earliest_timeout.abs_value_us > pm->timeout.abs_value_us)
3102   {
3103     target->earliest_timeout.abs_value_us = pm->timeout.abs_value_us;
3104     if (NULL != target->timeout_task)
3105       GNUNET_SCHEDULER_cancel (target->timeout_task);
3106     target->timeout_task = GNUNET_SCHEDULER_add_at (target->earliest_timeout,
3107                                                     &check_queue_timeouts,
3108                                                     target);
3109   }
3110   if (! was_empty)
3111     return; /* all queues must already be busy */
3112   for (struct Queue *queue = target->queue_head; NULL != queue;
3113        queue = queue->next_neighbour)
3114   {
3115     /* try transmission on any queue that is idle */
3116     if (NULL == queue->transmit_task)
3117       queue->transmit_task =
3118         GNUNET_SCHEDULER_add_now (&transmit_on_queue, queue);
3119   }
3120 }
3121
3122
3123 /**
3124  * Communicator started.  Test message is well-formed.
3125  *
3126  * @param cls the client
3127  * @param cam the send message that was sent
3128  */
3129 static int
3130 check_communicator_available (
3131   void *cls,
3132   const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
3133 {
3134   struct TransportClient *tc = cls;
3135   uint16_t size;
3136
3137   if (CT_NONE != tc->type)
3138   {
3139     GNUNET_break (0);
3140     return GNUNET_SYSERR;
3141   }
3142   tc->type = CT_COMMUNICATOR;
3143   size = ntohs (cam->header.size) - sizeof (*cam);
3144   if (0 == size)
3145     return GNUNET_OK; /* receive-only communicator */
3146   GNUNET_MQ_check_zero_termination (cam);
3147   return GNUNET_OK;
3148 }
3149
3150
3151 /**
3152  * Communicator started.  Process the request.
3153  *
3154  * @param cls the client
3155  * @param cam the send message that was sent
3156  */
3157 static void
3158 handle_communicator_available (
3159   void *cls,
3160   const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam)
3161 {
3162   struct TransportClient *tc = cls;
3163   uint16_t size;
3164
3165   size = ntohs (cam->header.size) - sizeof (*cam);
3166   if (0 == size)
3167     return; /* receive-only communicator */
3168   tc->details.communicator.address_prefix =
3169     GNUNET_strdup ((const char *) &cam[1]);
3170   tc->details.communicator.cc =
3171     (enum GNUNET_TRANSPORT_CommunicatorCharacteristics) ntohl (cam->cc);
3172   GNUNET_SERVICE_client_continue (tc->client);
3173 }
3174
3175
3176 /**
3177  * Communicator requests backchannel transmission.  Check the request.
3178  *
3179  * @param cls the client
3180  * @param cb the send message that was sent
3181  * @return #GNUNET_OK if message is well-formed
3182  */
3183 static int
3184 check_communicator_backchannel (
3185   void *cls,
3186   const struct GNUNET_TRANSPORT_CommunicatorBackchannel *cb)
3187 {
3188   const struct GNUNET_MessageHeader *inbox;
3189   const char *is;
3190   uint16_t msize;
3191   uint16_t isize;
3192
3193   (void) cls;
3194   msize = ntohs (cb->header.size) - sizeof (*cb);
3195   if (UINT16_MAX - msize >
3196       sizeof (struct TransportBackchannelEncapsulationMessage) +
3197         sizeof (struct TransportBackchannelRequestPayload))
3198   {
3199     GNUNET_break (0);
3200     return GNUNET_SYSERR;
3201   }
3202   inbox = (const struct GNUNET_MessageHeader *) &cb[1];
3203   isize = ntohs (inbox->size);
3204   if (isize >= msize)
3205   {
3206     GNUNET_break (0);
3207     return GNUNET_SYSERR;
3208   }
3209   is = (const char *) inbox;
3210   is += isize;
3211   msize -= isize;
3212   GNUNET_assert (msize > 0);
3213   if ('\0' != is[msize - 1])
3214   {
3215     GNUNET_break (0);
3216     return GNUNET_SYSERR;
3217   }
3218   return GNUNET_OK;
3219 }
3220
3221
3222 /**
3223  * Remove memory used by expired ephemeral keys.
3224  *
3225  * @param cls NULL
3226  */
3227 static void
3228 expire_ephemerals (void *cls)
3229 {
3230   struct EphemeralCacheEntry *ece;
3231
3232   (void) cls;
3233   ephemeral_task = NULL;
3234   while (NULL != (ece = GNUNET_CONTAINER_heap_peek (ephemeral_heap)))
3235   {
3236     if (0 == GNUNET_TIME_absolute_get_remaining (ece->ephemeral_validity)
3237                .rel_value_us)
3238     {
3239       free_ephemeral (ece);
3240       continue;
3241     }
3242     ephemeral_task = GNUNET_SCHEDULER_add_at (ece->ephemeral_validity,
3243                                               &expire_ephemerals,
3244                                               NULL);
3245     return;
3246   }
3247 }
3248
3249
3250 /**
3251  * Lookup ephemeral key in our #ephemeral_map. If no valid one exists, generate
3252  * one, cache it and return it.
3253  *
3254  * @param pid peer to look up ephemeral for
3255  * @param private_key[out] set to the private key
3256  * @param ephemeral_key[out] set to the key
3257  * @param ephemeral_sender_sig[out] set to the signature
3258  * @param ephemeral_validity[out] set to the validity expiration time
3259  */
3260 static void
3261 lookup_ephemeral (const struct GNUNET_PeerIdentity *pid,
3262                   struct GNUNET_CRYPTO_EcdhePrivateKey *private_key,
3263                   struct GNUNET_CRYPTO_EcdhePublicKey *ephemeral_key,
3264                   struct GNUNET_CRYPTO_EddsaSignature *ephemeral_sender_sig,
3265                   struct GNUNET_TIME_Absolute *ephemeral_validity)
3266 {
3267   struct EphemeralCacheEntry *ece;
3268   struct EphemeralConfirmationPS ec;
3269
3270   ece = GNUNET_CONTAINER_multipeermap_get (ephemeral_map, pid);
3271   if ((NULL != ece) &&
3272       (0 == GNUNET_TIME_absolute_get_remaining (ece->ephemeral_validity)
3273               .rel_value_us))
3274   {
3275     free_ephemeral (ece);
3276     ece = NULL;
3277   }
3278   if (NULL == ece)
3279   {
3280     ece = GNUNET_new (struct EphemeralCacheEntry);
3281     ece->target = *pid;
3282     ece->ephemeral_validity =
3283       GNUNET_TIME_absolute_add (GNUNET_TIME_absolute_get_monotonic (GST_cfg),
3284                                 EPHEMERAL_VALIDITY);
3285     GNUNET_assert (GNUNET_OK ==
3286                    GNUNET_CRYPTO_ecdhe_key_create2 (&ece->private_key));
3287     GNUNET_CRYPTO_ecdhe_key_get_public (&ece->private_key, &ece->ephemeral_key);
3288     ec.purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL);
3289     ec.purpose.size = htonl (sizeof (ec));
3290     ec.target = *pid;
3291     ec.ephemeral_key = ece->ephemeral_key;
3292     GNUNET_assert (GNUNET_OK == GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
3293                                                           &ec.purpose,
3294                                                           &ece->sender_sig));
3295     ece->hn =
3296       GNUNET_CONTAINER_heap_insert (ephemeral_heap,
3297                                     ece,
3298                                     ece->ephemeral_validity.abs_value_us);
3299     GNUNET_assert (GNUNET_OK ==
3300                    GNUNET_CONTAINER_multipeermap_put (
3301                      ephemeral_map,
3302                      &ece->target,
3303                      ece,
3304                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
3305     if (NULL == ephemeral_task)
3306       ephemeral_task = GNUNET_SCHEDULER_add_at (ece->ephemeral_validity,
3307                                                 &expire_ephemerals,
3308                                                 NULL);
3309   }
3310   *private_key = ece->private_key;
3311   *ephemeral_key = ece->ephemeral_key;
3312   *ephemeral_sender_sig = ece->sender_sig;
3313   *ephemeral_validity = ece->ephemeral_validity;
3314 }
3315
3316
3317 /**
3318  * Send the control message @a payload on @a queue.
3319  *
3320  * @param queue the queue to use for transmission
3321  * @param pm pending message to update once transmission is done, may be NULL!
3322  * @param payload the payload to send (encapsulated in a
3323  *        #GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG).
3324  * @param payload_size number of bytes in @a payload
3325  */
3326 static void
3327 queue_send_msg (struct Queue *queue,
3328                 struct PendingMessage *pm,
3329                 const void *payload,
3330                 size_t payload_size)
3331 {
3332   struct Neighbour *n = queue->neighbour;
3333   struct GNUNET_TRANSPORT_SendMessageTo *smt;
3334   struct GNUNET_MQ_Envelope *env;
3335
3336   env = GNUNET_MQ_msg_extra (smt,
3337                              payload_size,
3338                              GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG);
3339   smt->qid = queue->qid;
3340   smt->mid = queue->mid_gen;
3341   smt->receiver = n->pid;
3342   memcpy (&smt[1], payload, payload_size);
3343   {
3344     /* Pass the env to the communicator of queue for transmission. */
3345     struct QueueEntry *qe;
3346
3347     qe = GNUNET_new (struct QueueEntry);
3348     qe->mid = queue->mid_gen++;
3349     qe->queue = queue;
3350     if (NULL != pm)
3351     {
3352       qe->pm = pm;
3353       GNUNET_assert (NULL == pm->qe);
3354       pm->qe = qe;
3355     }
3356     GNUNET_CONTAINER_DLL_insert (queue->queue_head, queue->queue_tail, qe);
3357     GNUNET_assert (CT_COMMUNICATOR == queue->tc->type);
3358     queue->queue_length++;
3359     queue->tc->details.communicator.total_queue_length++;
3360     GNUNET_MQ_send (queue->tc->mq, env);
3361   }
3362 }
3363
3364
3365 /**
3366  * Which transmission options are allowable for transmission?
3367  * Interpreted bit-wise!
3368  */
3369 enum RouteMessageOptions
3370 {
3371   /**
3372    * Only confirmed, non-DV direct neighbours.
3373    */
3374   RMO_NONE = 0,
3375
3376   /**
3377    * We are allowed to use DV routing for this @a hdr
3378    */
3379   RMO_DV_ALLOWED = 1,
3380
3381   /**
3382    * We are allowed to use unconfirmed queues or DV routes for this message
3383    */
3384   RMO_UNCONFIRMED_ALLOWED = 2,
3385
3386   /**
3387    * Reliable and unreliable, DV and non-DV are all acceptable.
3388    */
3389   RMO_ANYTHING_GOES = (RMO_DV_ALLOWED | RMO_UNCONFIRMED_ALLOWED),
3390
3391   /**
3392    * If we have multiple choices, it is OK to send this message
3393    * over multiple channels at the same time to improve loss tolerance.
3394    * (We do at most 2 transmissions.)
3395    */
3396   RMO_REDUNDANT = 4
3397 };
3398
3399
3400 /**
3401  * Pick a queue of @a n under constraints @a options and schedule
3402  * transmission of @a hdr.
3403  *
3404  * @param n neighbour to send to
3405  * @param hdr message to send as payload
3406  * @param options whether queues must be confirmed or not,
3407  *        and whether we may pick multiple (2) queues
3408  */
3409 static void
3410 route_via_neighbour (const struct Neighbour *n,
3411                      const struct GNUNET_MessageHeader *hdr,
3412                      enum RouteMessageOptions options)
3413 {
3414   struct GNUNET_TIME_Absolute now;
3415   unsigned int candidates;
3416   unsigned int sel1;
3417   unsigned int sel2;
3418
3419   /* Pick one or two 'random' queues from n (under constraints of options) */
3420   now = GNUNET_TIME_absolute_get ();
3421   /* FIXME-OPTIMIZE: give queues 'weights' and pick proportional to
3422      weight in the future; weight could be assigned by observed
3423      bandwidth (note: not sure if we should do this for this type
3424      of control traffic though). */
3425   candidates = 0;
3426   for (struct Queue *pos = n->queue_head; NULL != pos;
3427        pos = pos->next_neighbour)
3428   {
3429     /* Count the queue with the visibility task in all cases, as
3430        otherwise we may end up with no queues just because the
3431        time for the visibility task just expired but the scheduler
3432        just ran this task first */
3433     if ((0 == (options & RMO_UNCONFIRMED_ALLOWED)) ||
3434         (pos->validated_until.abs_value_us > now.abs_value_us) ||
3435         (NULL != pos->visibility_task))
3436       candidates++;
3437   }
3438   if (0 == candidates)
3439   {
3440     /* Given that we above check for pos->visibility task,
3441        this should be strictly impossible. */
3442     GNUNET_break (0);
3443     return;
3444   }
3445   sel1 = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, candidates);
3446   if (0 == (options & RMO_REDUNDANT))
3447     sel2 = candidates; /* picks none! */
3448   else
3449     sel2 = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, candidates);
3450   candidates = 0;
3451   for (struct Queue *pos = n->queue_head; NULL != pos;
3452        pos = pos->next_neighbour)
3453   {
3454     /* Count the queue with the visibility task in all cases, as
3455        otherwise we may end up with no queues just because the
3456        time for the visibility task just expired but the scheduler
3457        just ran this task first */
3458     if ((pos->validated_until.abs_value_us > now.abs_value_us) ||
3459         (NULL != pos->visibility_task))
3460     {
3461       if ((sel1 == candidates) || (sel2 == candidates))
3462         queue_send_msg (pos, NULL, hdr, ntohs (hdr->size));
3463       candidates++;
3464     }
3465   }
3466 }
3467
3468
3469 /**
3470  * Given a distance vector path @a dvh route @a payload to
3471  * the ultimate destination respecting @a options.
3472  * Sets up the boxed message and queues it at the next hop.
3473  *
3474  * @param dvh choice of the path for the message
3475  * @param payload body to transmit
3476  * @param options options to use for control
3477  */
3478 static void
3479 forward_via_dvh (const struct DistanceVectorHop *dvh,
3480                  const struct GNUNET_MessageHeader *payload,
3481                  enum RouteMessageOptions options)
3482 {
3483   uint16_t mlen = ntohs (payload->size);
3484   char boxram[sizeof (struct TransportDVBox) +
3485               (dvh->distance + 1) * sizeof (struct GNUNET_PeerIdentity) +
3486               mlen] GNUNET_ALIGN;
3487   struct TransportDVBox *box = (struct TransportDVBox *) boxram;
3488   struct GNUNET_PeerIdentity *path = (struct GNUNET_PeerIdentity *) &box[1];
3489
3490   box->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX);
3491   box->header.size = htons (sizeof (boxram));
3492   box->total_hops = htons (0);
3493   box->num_hops = htons (dvh->distance + 1);
3494   box->origin = GST_my_identity;
3495   memcpy (path, dvh->path, dvh->distance * sizeof (struct GNUNET_PeerIdentity));
3496   path[dvh->distance] = dvh->dv->target;
3497   memcpy (&path[dvh->distance + 1], payload, mlen);
3498   route_via_neighbour (dvh->next_hop, &box->header, options);
3499 }
3500
3501
3502 /**
3503  * Pick a path of @a dv under constraints @a options and schedule
3504  * transmission of @a hdr.
3505  *
3506  * @param n neighbour to send to
3507  * @param hdr message to send as payload
3508  * @param options whether path must be confirmed or not
3509  *        and whether we may pick multiple (2) paths
3510  */
3511 static void
3512 route_via_dv (const struct DistanceVector *dv,
3513               const struct GNUNET_MessageHeader *hdr,
3514               enum RouteMessageOptions options)
3515 {
3516   struct DistanceVectorHop *h1;
3517   struct DistanceVectorHop *h2;
3518   uint64_t num_dv;
3519   uint64_t choice1;
3520   uint64_t choice2;
3521
3522   /* Pick random vectors, but weighted by distance, giving more weight
3523      to shorter vectors */
3524   num_dv = 0;
3525   for (struct DistanceVectorHop *pos = dv->dv_head; NULL != pos;
3526        pos = pos->next_dv)
3527   {
3528     if ((0 == (options & RMO_UNCONFIRMED_ALLOWED)) &&
3529         (GNUNET_TIME_absolute_get_remaining (pos->path_valid_until)
3530            .rel_value_us == 0))
3531       continue; /* pos unconfirmed and confirmed required */
3532     num_dv += MAX_DV_HOPS_ALLOWED - pos->distance;
3533   }
3534   if (0 == num_dv)
3535   {
3536     GNUNET_break (0);
3537     return;
3538   }
3539   choice1 = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, num_dv);
3540   choice2 = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, num_dv);
3541   num_dv = 0;
3542   h1 = NULL;
3543   h2 = NULL;
3544   for (struct DistanceVectorHop *pos = dv->dv_head; NULL != pos;
3545        pos = pos->next_dv)
3546   {
3547     uint32_t delta = MAX_DV_HOPS_ALLOWED - pos->distance;
3548
3549     if ((0 == (options & RMO_UNCONFIRMED_ALLOWED)) &&
3550         (GNUNET_TIME_absolute_get_remaining (pos->path_valid_until)
3551            .rel_value_us == 0))
3552       continue; /* pos unconfirmed and confirmed required */
3553     if ((num_dv <= choice1) && (num_dv + delta > choice1))
3554       h1 = pos;
3555     if ((num_dv <= choice2) && (num_dv + delta > choice2))
3556       h2 = pos;
3557     num_dv += delta;
3558   }
3559   forward_via_dvh (h1, hdr, options & (~RMO_REDUNDANT));
3560   if (0 == (options & RMO_REDUNDANT))
3561     forward_via_dvh (h2, hdr, options & (~RMO_REDUNDANT));
3562 }
3563
3564
3565 /**
3566  * We need to transmit @a hdr to @a target.  If necessary, this may
3567  * involve DV routing.
3568  *
3569  * @param target peer to receive @a hdr
3570  * @param hdr header of the message to route and #GNUNET_free()
3571  * @param options which transmission channels are allowed
3572  */
3573 static void
3574 route_message (const struct GNUNET_PeerIdentity *target,
3575                struct GNUNET_MessageHeader *hdr,
3576                enum RouteMessageOptions options)
3577 {
3578   struct Neighbour *n;
3579   struct DistanceVector *dv;
3580
3581   n = GNUNET_CONTAINER_multipeermap_get (neighbours, target);
3582   dv = (0 != (options & RMO_DV_ALLOWED))
3583          ? GNUNET_CONTAINER_multipeermap_get (dv_routes, target)
3584          : NULL;
3585   if (0 == (options & RMO_UNCONFIRMED_ALLOWED))
3586   {
3587     /* if confirmed is required, and we do not have anything
3588        confirmed, drop respective options */
3589     if ((NULL != n) && (GNUNET_NO == n->core_visible))
3590       n = NULL;
3591     if ((NULL != dv) && (GNUNET_NO == dv->core_visible))
3592       dv = NULL;
3593   }
3594   if ((NULL == n) && (NULL == dv))
3595   {
3596     GNUNET_STATISTICS_update (GST_stats,
3597                               "# Messages dropped in routing: no acceptable method",
3598                               1,
3599                               GNUNET_NO);
3600     GNUNET_free (hdr);
3601     return;
3602   }
3603   /* If both dv and n are possible and we must choose:
3604      flip a coin for the choice between the two; for now 50/50 */
3605   if ((NULL != n) && (NULL != dv) && (0 == (options & RMO_REDUNDANT)))
3606   {
3607     if (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 2))
3608       n = NULL;
3609     else
3610       dv = NULL;
3611   }
3612   if ((NULL != n) && (NULL != dv))
3613     options &= ~RMO_REDUNDANT; /* We will do one DV and one direct, that's
3614                                   enough for redunancy, so clear the flag. */
3615   if (NULL != n)
3616   {
3617     route_via_neighbour (n, hdr, options);
3618   }
3619   if (NULL != dv)
3620   {
3621     route_via_dv (dv, hdr, options);
3622   }
3623   GNUNET_free (hdr);
3624 }
3625
3626
3627 /**
3628  * Structure of the key material used to encrypt backchannel messages.
3629  */
3630 struct BackchannelKeyState
3631 {
3632   /**
3633    * State of our block cipher.
3634    */
3635   gcry_cipher_hd_t cipher;
3636
3637   /**
3638    * Actual key material.
3639    */
3640   struct
3641   {
3642
3643     /**
3644      * Key used for HMAC calculations (via #GNUNET_CRYPTO_hmac()).
3645      */
3646     struct GNUNET_CRYPTO_AuthKey hmac_key;
3647
3648     /**
3649      * Symmetric key to use for encryption.
3650      */
3651     char aes_key[256 / 8];
3652
3653     /**
3654      * Counter value to use during setup.
3655      */
3656     char aes_ctr[128 / 8];
3657
3658   } material;
3659 };
3660
3661
3662 static void
3663 bc_setup_key_state_from_km (const struct GNUNET_HashCode *km,
3664                             const struct GNUNET_ShortHashCode *iv,
3665                             struct BackchannelKeyState *key)
3666 {
3667   /* must match #dh_key_derive_eph_pub */
3668   GNUNET_assert (GNUNET_YES ==
3669                  GNUNET_CRYPTO_kdf (&key->material,
3670                                     sizeof (key->material),
3671                                     "transport-backchannel-key",
3672                                     strlen ("transport-backchannel-key"),
3673                                     &km,
3674                                     sizeof (km),
3675                                     iv,
3676                                     sizeof (*iv)));
3677   gcry_cipher_open (&key->cipher,
3678                     GCRY_CIPHER_AES256 /* low level: go for speed */,
3679                     GCRY_CIPHER_MODE_CTR,
3680                     0 /* flags */);
3681   gcry_cipher_setkey (key->cipher,
3682                       &key->material.aes_key,
3683                       sizeof (key->material.aes_key));
3684   gcry_cipher_setctr (key->cipher,
3685                       &key->material.aes_ctr,
3686                       sizeof (key->material.aes_ctr));
3687 }
3688
3689
3690 /**
3691  * Derive backchannel encryption key material from @a priv_ephemeral
3692  * and @a target and @a iv.
3693  *
3694  * @param priv_ephemeral ephemeral private key to use
3695  * @param target the target peer to encrypt to
3696  * @param iv unique IV to use
3697  * @param key[out] set to the key material
3698  */
3699 static void
3700 dh_key_derive_eph_pid (
3701   const struct GNUNET_CRYPTO_EcdhePrivateKey *priv_ephemeral,
3702   const struct GNUNET_PeerIdentity *target,
3703   const struct GNUNET_ShortHashCode *iv,
3704   struct BackchannelKeyState *key)
3705 {
3706   struct GNUNET_HashCode km;
3707
3708   GNUNET_assert (GNUNET_YES == GNUNET_CRYPTO_ecdh_eddsa (priv_ephemeral,
3709                                                          &target->public_key,
3710                                                          &km));
3711   bc_setup_key_state_from_km (&km, iv, key);
3712 }
3713
3714
3715 /**
3716  * Derive backchannel encryption key material from #GST_my_private_key
3717  * and @a pub_ephemeral and @a iv.
3718  *
3719  * @param priv_ephemeral ephemeral private key to use
3720  * @param target the target peer to encrypt to
3721  * @param iv unique IV to use
3722  * @param key[out] set to the key material
3723  */
3724 static void
3725 dh_key_derive_eph_pub (const struct GNUNET_CRYPTO_EcdhePublicKey *pub_ephemeral,
3726                        const struct GNUNET_ShortHashCode *iv,
3727                        struct BackchannelKeyState *key)
3728 {
3729   struct GNUNET_HashCode km;
3730
3731   GNUNET_assert (GNUNET_YES == GNUNET_CRYPTO_eddsa_ecdh (GST_my_private_key,
3732                                                          pub_ephemeral,
3733                                                          &km));
3734   bc_setup_key_state_from_km (&km, iv, key);
3735 }
3736
3737
3738 /**
3739  * Do HMAC calculation for backchannel messages over @a data using key
3740  * material from @a key.
3741  *
3742  * @param key key material (from DH)
3743  * @param hmac[out] set to the HMAC
3744  * @param data data to perform HMAC calculation over
3745  * @param data_size number of bytes in @a data
3746  */
3747 static void
3748 bc_hmac (const struct BackchannelKeyState *key,
3749          struct GNUNET_HashCode *hmac,
3750          const void *data,
3751          size_t data_size)
3752 {
3753   GNUNET_CRYPTO_hmac (&key->material.hmac_key, data, data_size, hmac);
3754 }
3755
3756
3757 /**
3758  * Perform backchannel encryption using symmetric secret in @a key
3759  * to encrypt data from @a in to @a dst.
3760  *
3761  * @param key[in,out] key material to use
3762  * @param dst where to write the result
3763  * @param in input data to encrypt (plaintext)
3764  * @param in_size number of bytes of input in @a in and available at @a dst
3765  */
3766 static void
3767 bc_encrypt (struct BackchannelKeyState *key,
3768             const void *in,
3769             void *dst,
3770             size_t in_size)
3771 {
3772   GNUNET_assert (0 ==
3773                  gcry_cipher_encrypt (key->cipher, dst, in_size, in, in_size));
3774 }
3775
3776
3777 /**
3778  * Perform backchannel encryption using symmetric secret in @a key
3779  * to encrypt data from @a in to @a dst.
3780  *
3781  * @param key[in,out] key material to use
3782  * @param ciph cipher text to decrypt
3783  * @param out[out] output data to generate (plaintext)
3784  * @param out_size number of bytes of input in @a ciph and available in @a out
3785  */
3786 static void
3787 bc_decrypt (struct BackchannelKeyState *key,
3788             void *out,
3789             const void *ciph,
3790             size_t out_size)
3791 {
3792   GNUNET_assert (
3793     0 == gcry_cipher_decrypt (key->cipher, out, out_size, ciph, out_size));
3794 }
3795
3796
3797 /**
3798  * Clean up key material in @a key.
3799  *
3800  * @param key key material to clean up (memory must not be free'd!)
3801  */
3802 static void
3803 bc_key_clean (struct BackchannelKeyState *key)
3804 {
3805   gcry_cipher_close (key->cipher);
3806   GNUNET_CRYPTO_zero_keys (&key->material, sizeof (key->material));
3807 }
3808
3809
3810 /**
3811  * Communicator requests backchannel transmission.  Process the request.
3812  *
3813  * @param cls the client
3814  * @param cb the send message that was sent
3815  */
3816 static void
3817 handle_communicator_backchannel (
3818   void *cls,
3819   const struct GNUNET_TRANSPORT_CommunicatorBackchannel *cb)
3820 {
3821   struct TransportClient *tc = cls;
3822   struct GNUNET_CRYPTO_EcdhePrivateKey private_key;
3823   struct GNUNET_TIME_Absolute ephemeral_validity;
3824   struct TransportBackchannelEncapsulationMessage *enc;
3825   struct TransportBackchannelRequestPayload ppay;
3826   struct BackchannelKeyState key;
3827   char *mpos;
3828   uint16_t msize;
3829
3830   /* encapsulate and encrypt message */
3831   msize = ntohs (cb->header.size) - sizeof (*cb) +
3832           sizeof (struct TransportBackchannelRequestPayload);
3833   enc = GNUNET_malloc (sizeof (*enc) + msize);
3834   enc->header.type =
3835     htons (GNUNET_MESSAGE_TYPE_TRANSPORT_BACKCHANNEL_ENCAPSULATION);
3836   enc->header.size = htons (sizeof (*enc) + msize);
3837   enc->target = cb->pid;
3838   lookup_ephemeral (&cb->pid,
3839                     &private_key,
3840                     &enc->ephemeral_key,
3841                     &ppay.sender_sig,
3842                     &ephemeral_validity);
3843   GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
3844                               &enc->iv,
3845                               sizeof (enc->iv));
3846   dh_key_derive_eph_pid (&private_key, &cb->pid, &enc->iv, &key);
3847   ppay.ephemeral_validity = GNUNET_TIME_absolute_hton (ephemeral_validity);
3848   ppay.monotonic_time =
3849     GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get_monotonic (GST_cfg));
3850   mpos = (char *) &enc[1];
3851   bc_encrypt (&key, &ppay, mpos, sizeof (ppay));
3852   bc_encrypt (&key,
3853               &cb[1],
3854               &mpos[sizeof (ppay)],
3855               ntohs (cb->header.size) - sizeof (*cb));
3856   bc_hmac (&key,
3857            &enc->hmac,
3858            mpos,
3859            sizeof (ppay) + ntohs (cb->header.size) - sizeof (*cb));
3860   bc_key_clean (&key);
3861   route_message (&cb->pid, &enc->header, RMO_DV_ALLOWED);
3862   GNUNET_SERVICE_client_continue (tc->client);
3863 }
3864
3865
3866 /**
3867  * Address of our peer added.  Test message is well-formed.
3868  *
3869  * @param cls the client
3870  * @param aam the send message that was sent
3871  * @return #GNUNET_OK if message is well-formed
3872  */
3873 static int
3874 check_add_address (void *cls,
3875                    const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
3876 {
3877   struct TransportClient *tc = cls;
3878
3879   if (CT_COMMUNICATOR != tc->type)
3880   {
3881     GNUNET_break (0);
3882     return GNUNET_SYSERR;
3883   }
3884   GNUNET_MQ_check_zero_termination (aam);
3885   return GNUNET_OK;
3886 }
3887
3888
3889 /**
3890  * Ask peerstore to store our address.
3891  *
3892  * @param cls an `struct AddressListEntry *`
3893  */
3894 static void
3895 store_pi (void *cls);
3896
3897
3898 /**
3899  * Function called when peerstore is done storing our address.
3900  *
3901  * @param cls a `struct AddressListEntry`
3902  * @param success #GNUNET_YES if peerstore was successful
3903  */
3904 static void
3905 peerstore_store_own_cb (void *cls, int success)
3906 {
3907   struct AddressListEntry *ale = cls;
3908
3909   ale->sc = NULL;
3910   if (GNUNET_YES != success)
3911     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
3912                 "Failed to store our own address `%s' in peerstore!\n",
3913                 ale->address);
3914   /* refresh period is 1/4 of expiration time, that should be plenty
3915      without being excessive. */
3916   ale->st =
3917     GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_divide (ale->expiration,
3918                                                                4ULL),
3919                                   &store_pi,
3920                                   ale);
3921 }
3922
3923
3924 /**
3925  * Ask peerstore to store our address.
3926  *
3927  * @param cls an `struct AddressListEntry *`
3928  */
3929 static void
3930 store_pi (void *cls)
3931 {
3932   struct AddressListEntry *ale = cls;
3933   void *addr;
3934   size_t addr_len;
3935   struct GNUNET_TIME_Absolute expiration;
3936
3937   ale->st = NULL;
3938   expiration = GNUNET_TIME_relative_to_absolute (ale->expiration);
3939   GNUNET_HELLO_sign_address (ale->address,
3940                              ale->nt,
3941                              expiration,
3942                              GST_my_private_key,
3943                              &addr,
3944                              &addr_len);
3945   ale->sc = GNUNET_PEERSTORE_store (peerstore,
3946                                     "transport",
3947                                     &GST_my_identity,
3948                                     GNUNET_PEERSTORE_TRANSPORT_HELLO_KEY,
3949                                     addr,
3950                                     addr_len,
3951                                     expiration,
3952                                     GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
3953                                     &peerstore_store_own_cb,
3954                                     ale);
3955   GNUNET_free (addr);
3956   if (NULL == ale->sc)
3957   {
3958     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
3959                 "Failed to store our address `%s' with peerstore\n",
3960                 ale->address);
3961     ale->st =
3962       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS, &store_pi, ale);
3963   }
3964 }
3965
3966
3967 /**
3968  * Address of our peer added.  Process the request.
3969  *
3970  * @param cls the client
3971  * @param aam the send message that was sent
3972  */
3973 static void
3974 handle_add_address (void *cls,
3975                     const struct GNUNET_TRANSPORT_AddAddressMessage *aam)
3976 {
3977   struct TransportClient *tc = cls;
3978   struct AddressListEntry *ale;
3979   size_t slen;
3980
3981   slen = ntohs (aam->header.size) - sizeof (*aam);
3982   ale = GNUNET_malloc (sizeof (struct AddressListEntry) + slen);
3983   ale->tc = tc;
3984   ale->address = (const char *) &ale[1];
3985   ale->expiration = GNUNET_TIME_relative_ntoh (aam->expiration);
3986   ale->aid = aam->aid;
3987   ale->nt = (enum GNUNET_NetworkType) ntohl (aam->nt);
3988   memcpy (&ale[1], &aam[1], slen);
3989   GNUNET_CONTAINER_DLL_insert (tc->details.communicator.addr_head,
3990                                tc->details.communicator.addr_tail,
3991                                ale);
3992   ale->st = GNUNET_SCHEDULER_add_now (&store_pi, ale);
3993   GNUNET_SERVICE_client_continue (tc->client);
3994 }
3995
3996
3997 /**
3998  * Address of our peer deleted.  Process the request.
3999  *
4000  * @param cls the client
4001  * @param dam the send message that was sent
4002  */
4003 static void
4004 handle_del_address (void *cls,
4005                     const struct GNUNET_TRANSPORT_DelAddressMessage *dam)
4006 {
4007   struct TransportClient *tc = cls;
4008
4009   if (CT_COMMUNICATOR != tc->type)
4010   {
4011     GNUNET_break (0);
4012     GNUNET_SERVICE_client_drop (tc->client);
4013     return;
4014   }
4015   for (struct AddressListEntry *ale = tc->details.communicator.addr_head;
4016        NULL != ale;
4017        ale = ale->next)
4018   {
4019     if (dam->aid != ale->aid)
4020       continue;
4021     GNUNET_assert (ale->tc == tc);
4022     free_address_list_entry (ale);
4023     GNUNET_SERVICE_client_continue (tc->client);
4024   }
4025   GNUNET_break (0);
4026   GNUNET_SERVICE_client_drop (tc->client);
4027 }
4028
4029
4030 /**
4031  * Context from #handle_incoming_msg().  Closure for many
4032  * message handlers below.
4033  */
4034 struct CommunicatorMessageContext
4035 {
4036   /**
4037    * Which communicator provided us with the message.
4038    */
4039   struct TransportClient *tc;
4040
4041   /**
4042    * Additional information for flow control and about the sender.
4043    */
4044   struct GNUNET_TRANSPORT_IncomingMessage im;
4045
4046   /**
4047    * Number of hops the message has travelled (if DV-routed).
4048    * FIXME: make use of this in ACK handling!
4049    */
4050   uint16_t total_hops;
4051 };
4052
4053
4054 /**
4055  * Given an inbound message @a msg from a communicator @a cmc,
4056  * demultiplex it based on the type calling the right handler.
4057  *
4058  * @param cmc context for demultiplexing
4059  * @param msg message to demultiplex
4060  */
4061 static void
4062 demultiplex_with_cmc (struct CommunicatorMessageContext *cmc,
4063                       const struct GNUNET_MessageHeader *msg);
4064
4065
4066 /**
4067  * Send ACK to communicator (if requested) and free @a cmc.
4068  *
4069  * @param cmc context for which we are done handling the message
4070  */
4071 static void
4072 finish_cmc_handling (struct CommunicatorMessageContext *cmc)
4073 {
4074   if (0 != ntohl (cmc->im.fc_on))
4075   {
4076     /* send ACK when done to communicator for flow control! */
4077     struct GNUNET_MQ_Envelope *env;
4078     struct GNUNET_TRANSPORT_IncomingMessageAck *ack;
4079
4080     env = GNUNET_MQ_msg (ack, GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG_ACK);
4081     ack->reserved = htonl (0);
4082     ack->fc_id = cmc->im.fc_id;
4083     ack->sender = cmc->im.sender;
4084     GNUNET_MQ_send (cmc->tc->mq, env);
4085   }
4086   GNUNET_SERVICE_client_continue (cmc->tc->client);
4087   GNUNET_free (cmc);
4088 }
4089
4090
4091 /**
4092  * Communicator gave us an unencapsulated message to pass as-is to
4093  * CORE.  Process the request.
4094  *
4095  * @param cls a `struct CommunicatorMessageContext` (must call
4096  * #finish_cmc_handling() when done)
4097  * @param mh the message that was received
4098  */
4099 static void
4100 handle_raw_message (void *cls, const struct GNUNET_MessageHeader *mh)
4101 {
4102   struct CommunicatorMessageContext *cmc = cls;
4103   uint16_t size = ntohs (mh->size);
4104
4105   if ((size > UINT16_MAX - sizeof (struct InboundMessage)) ||
4106       (size < sizeof (struct GNUNET_MessageHeader)))
4107   {
4108     struct GNUNET_SERVICE_Client *client = cmc->tc->client;
4109
4110     GNUNET_break (0);
4111     finish_cmc_handling (cmc);
4112     GNUNET_SERVICE_client_drop (client);
4113     return;
4114   }
4115   /* Forward to all CORE clients */
4116   for (struct TransportClient *tc = clients_head; NULL != tc; tc = tc->next)
4117   {
4118     struct GNUNET_MQ_Envelope *env;
4119     struct InboundMessage *im;
4120
4121     if (CT_CORE != tc->type)
4122       continue;
4123     env = GNUNET_MQ_msg_extra (im, size, GNUNET_MESSAGE_TYPE_TRANSPORT_RECV);
4124     im->peer = cmc->im.sender;
4125     memcpy (&im[1], mh, size);
4126     GNUNET_MQ_send (tc->mq, env);
4127   }
4128   /* FIXME: consider doing this _only_ once the message
4129      was drained from the CORE MQs to extend flow control to CORE!
4130      (basically, increment counter in cmc, decrement on MQ send continuation! */
4131   finish_cmc_handling (cmc);
4132 }
4133
4134
4135 /**
4136  * Communicator gave us a fragment box.  Check the message.
4137  *
4138  * @param cls a `struct CommunicatorMessageContext`
4139  * @param fb the send message that was sent
4140  * @return #GNUNET_YES if message is well-formed
4141  */
4142 static int
4143 check_fragment_box (void *cls, const struct TransportFragmentBox *fb)
4144 {
4145   uint16_t size = ntohs (fb->header.size);
4146   uint16_t bsize = size - sizeof (*fb);
4147
4148   if (0 == bsize)
4149   {
4150     GNUNET_break_op (0);
4151     return GNUNET_SYSERR;
4152   }
4153   if (bsize + ntohs (fb->frag_off) > ntohs (fb->msg_size))
4154   {
4155     GNUNET_break_op (0);
4156     return GNUNET_SYSERR;
4157   }
4158   if (ntohs (fb->frag_off) >= ntohs (fb->msg_size))
4159   {
4160     GNUNET_break_op (0);
4161     return GNUNET_SYSERR;
4162   }
4163   return GNUNET_YES;
4164 }
4165
4166
4167 /**
4168  * Generate a fragment acknowledgement for an @a rc.
4169  *
4170  * @param rc context to generate ACK for, @a rc ACK state is reset
4171  */
4172 static void
4173 send_fragment_ack (struct ReassemblyContext *rc)
4174 {
4175   struct TransportFragmentAckMessage *ack;
4176
4177   ack = GNUNET_new (struct TransportFragmentAckMessage);
4178   ack->header.size = htons (sizeof (struct TransportFragmentAckMessage));
4179   ack->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT_ACK);
4180   ack->frag_uuid = htonl (rc->frag_uuid);
4181   ack->extra_acks = GNUNET_htonll (rc->extra_acks);
4182   ack->msg_uuid = rc->msg_uuid;
4183   ack->avg_ack_delay = GNUNET_TIME_relative_hton (rc->avg_ack_delay);
4184   if (0 == rc->msg_missing)
4185     ack->reassembly_timeout = GNUNET_TIME_relative_hton (
4186       GNUNET_TIME_UNIT_FOREVER_REL); /* signal completion */
4187   else
4188     ack->reassembly_timeout = GNUNET_TIME_relative_hton (
4189       GNUNET_TIME_absolute_get_remaining (rc->reassembly_timeout));
4190   route_message (&rc->neighbour->pid, &ack->header, RMO_DV_ALLOWED);
4191   rc->avg_ack_delay = GNUNET_TIME_UNIT_ZERO;
4192   rc->num_acks = 0;
4193   rc->extra_acks = 0LLU;
4194 }
4195
4196
4197 /**
4198  * Communicator gave us a fragment.  Process the request.
4199  *
4200  * @param cls a `struct CommunicatorMessageContext` (must call
4201  * #finish_cmc_handling() when done)
4202  * @param fb the message that was received
4203  */
4204 static void
4205 handle_fragment_box (void *cls, const struct TransportFragmentBox *fb)
4206 {
4207   struct CommunicatorMessageContext *cmc = cls;
4208   struct Neighbour *n;
4209   struct ReassemblyContext *rc;
4210   const struct GNUNET_MessageHeader *msg;
4211   uint16_t msize;
4212   uint16_t fsize;
4213   uint16_t frag_off;
4214   uint32_t frag_uuid;
4215   char *target;
4216   struct GNUNET_TIME_Relative cdelay;
4217   int ack_now;
4218
4219   n = GNUNET_CONTAINER_multipeermap_get (neighbours, &cmc->im.sender);
4220   if (NULL == n)
4221   {
4222     struct GNUNET_SERVICE_Client *client = cmc->tc->client;
4223
4224     GNUNET_break (0);
4225     finish_cmc_handling (cmc);
4226     GNUNET_SERVICE_client_drop (client);
4227     return;
4228   }
4229   if (NULL == n->reassembly_map)
4230   {
4231     n->reassembly_map = GNUNET_CONTAINER_multishortmap_create (8, GNUNET_YES);
4232     n->reassembly_heap =
4233       GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
4234     n->reassembly_timeout_task =
4235       GNUNET_SCHEDULER_add_delayed (REASSEMBLY_EXPIRATION,
4236                                     &reassembly_cleanup_task,
4237                                     n);
4238   }
4239   msize = ntohs (fb->msg_size);
4240   rc = GNUNET_CONTAINER_multishortmap_get (n->reassembly_map, &fb->msg_uuid);
4241   if (NULL == rc)
4242   {
4243     rc = GNUNET_malloc (sizeof (*rc) + msize + /* reassembly payload buffer */
4244                         (msize + 7) / 8 * sizeof (uint8_t) /* bitfield */);
4245     rc->msg_uuid = fb->msg_uuid;
4246     rc->neighbour = n;
4247     rc->msg_size = msize;
4248     rc->reassembly_timeout =
4249       GNUNET_TIME_relative_to_absolute (REASSEMBLY_EXPIRATION);
4250     rc->last_frag = GNUNET_TIME_absolute_get ();
4251     rc->hn = GNUNET_CONTAINER_heap_insert (n->reassembly_heap,
4252                                            rc,
4253                                            rc->reassembly_timeout.abs_value_us);
4254     GNUNET_assert (GNUNET_OK ==
4255                    GNUNET_CONTAINER_multishortmap_put (
4256                      n->reassembly_map,
4257                      &rc->msg_uuid,
4258                      rc,
4259                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
4260     target = (char *) &rc[1];
4261     rc->bitfield = (uint8_t *) (target + rc->msg_size);
4262     rc->msg_missing = rc->msg_size;
4263   }
4264   else
4265   {
4266     target = (char *) &rc[1];
4267   }
4268   if (msize != rc->msg_size)
4269   {
4270     GNUNET_break (0);
4271     finish_cmc_handling (cmc);
4272     return;
4273   }
4274
4275   /* reassemble */
4276   fsize = ntohs (fb->header.size) - sizeof (*fb);
4277   frag_off = ntohs (fb->frag_off);
4278   memcpy (&target[frag_off], &fb[1], fsize);
4279   /* update bitfield and msg_missing */
4280   for (unsigned int i = frag_off; i < frag_off + fsize; i++)
4281   {
4282     if (0 == (rc->bitfield[i / 8] & (1 << (i % 8))))
4283     {
4284       rc->bitfield[i / 8] |= (1 << (i % 8));
4285       rc->msg_missing--;
4286     }
4287   }
4288
4289   /* Compute cummulative ACK */
4290   frag_uuid = ntohl (fb->frag_uuid);
4291   cdelay = GNUNET_TIME_absolute_get_duration (rc->last_frag);
4292   cdelay = GNUNET_TIME_relative_multiply (cdelay, rc->num_acks);
4293   rc->last_frag = GNUNET_TIME_absolute_get ();
4294   rc->avg_ack_delay = GNUNET_TIME_relative_add (rc->avg_ack_delay, cdelay);
4295   ack_now = GNUNET_NO;
4296   if (0 == rc->num_acks)
4297   {
4298     /* case one: first ack */
4299     rc->frag_uuid = frag_uuid;
4300     rc->extra_acks = 0LLU;
4301     rc->num_acks = 1;
4302   }
4303   else if ((frag_uuid >= rc->frag_uuid) && (frag_uuid <= rc->frag_uuid + 64))
4304   {
4305     /* case two: ack fits after existing min UUID */
4306     if ((frag_uuid == rc->frag_uuid) ||
4307         (0 != (rc->extra_acks & (1LLU << (frag_uuid - rc->frag_uuid - 1)))))
4308     {
4309       /* duplicate fragment, ack now! */
4310       ack_now = GNUNET_YES;
4311     }
4312     else
4313     {
4314       rc->extra_acks |= (1LLU << (frag_uuid - rc->frag_uuid - 1));
4315       rc->num_acks++;
4316     }
4317   }
4318   else if ((rc->frag_uuid > frag_uuid) &&
4319            (((rc->frag_uuid == frag_uuid + 64) && (0 == rc->extra_acks)) ||
4320             ((rc->frag_uuid < frag_uuid + 64) &&
4321              (rc->extra_acks ==
4322               (rc->extra_acks &
4323                ~((1LLU << (64 - (rc->frag_uuid - frag_uuid))) - 1LLU))))))
4324   {
4325     /* can fit ack by shifting extra acks and starting at
4326        frag_uid, test above esured that the bits we will
4327        shift 'extra_acks' by are all zero. */
4328     rc->extra_acks <<= (rc->frag_uuid - frag_uuid);
4329     rc->extra_acks |= (1LLU << (rc->frag_uuid - frag_uuid - 1));
4330     rc->frag_uuid = frag_uuid;
4331     rc->num_acks++;
4332   }
4333   if (65 == rc->num_acks) /* OPTIMIZE-FIXME: maybe use smaller threshold? This
4334                              is very aggressive. */
4335     ack_now = GNUNET_YES; /* maximum acks received */
4336   // FIXME: possibly also ACK based on RTT (but for that we'd need to
4337   // determine the queue used for the ACK first!)
4338
4339   /* is reassembly complete? */
4340   if (0 != rc->msg_missing)
4341   {
4342     if (ack_now)
4343       send_fragment_ack (rc);
4344     finish_cmc_handling (cmc);
4345     return;
4346   }
4347   /* reassembly is complete, verify result */
4348   msg = (const struct GNUNET_MessageHeader *) &rc[1];
4349   if (ntohs (msg->size) != rc->msg_size)
4350   {
4351     GNUNET_break (0);
4352     free_reassembly_context (rc);
4353     finish_cmc_handling (cmc);
4354     return;
4355   }
4356   /* successful reassembly */
4357   send_fragment_ack (rc);
4358   demultiplex_with_cmc (cmc, msg);
4359   /* FIXME-OPTIMIZE: really free here? Might be bad if fragments are still
4360      en-route and we forget that we finished this reassembly immediately!
4361      -> keep around until timeout?
4362      -> shorten timeout based on ACK? */
4363   free_reassembly_context (rc);
4364 }
4365
4366
4367 /**
4368  * Check the @a fa against the fragments associated with @a pm.
4369  * If it matches, remove the matching fragments from the transmission
4370  * list.
4371  *
4372  * @param pm pending message to check against the ack
4373  * @param fa the ack that was received
4374  * @return #GNUNET_YES if @a fa matched, #GNUNET_NO if not
4375  */
4376 static int
4377 check_ack_against_pm (struct PendingMessage *pm,
4378                       const struct TransportFragmentAckMessage *fa)
4379 {
4380   int match;
4381   struct PendingMessage *nxt;
4382   uint32_t fs = ntohl (fa->frag_uuid);
4383   uint64_t xtra = GNUNET_ntohll (fa->extra_acks);
4384
4385   match = GNUNET_NO;
4386   for (struct PendingMessage *frag = pm->head_frag; NULL != frag; frag = nxt)
4387   {
4388     const struct TransportFragmentBox *tfb =
4389       (const struct TransportFragmentBox *) &pm[1];
4390     uint32_t fu = ntohl (tfb->frag_uuid);
4391
4392     GNUNET_assert (PMT_FRAGMENT_BOX == frag->pmt);
4393     nxt = frag->next_frag;
4394     /* Check for exact match or match in the 'xtra' bitmask */
4395     if ((fu == fs) ||
4396         ((fu > fs) && (fu <= fs + 64) && (0 != (1LLU << (fu - fs - 1) & xtra))))
4397     {
4398       match = GNUNET_YES;
4399       free_fragment_tree (frag);
4400     }
4401   }
4402   return match;
4403 }
4404
4405
4406 /**
4407  * Communicator gave us a fragment acknowledgement.  Process the request.
4408  *
4409  * @param cls a `struct CommunicatorMessageContext` (must call
4410  * #finish_cmc_handling() when done)
4411  * @param fa the message that was received
4412  */
4413 static void
4414 handle_fragment_ack (void *cls, const struct TransportFragmentAckMessage *fa)
4415 {
4416   struct CommunicatorMessageContext *cmc = cls;
4417   struct Neighbour *n;
4418   int matched;
4419
4420   n = GNUNET_CONTAINER_multipeermap_get (neighbours, &cmc->im.sender);
4421   if (NULL == n)
4422   {
4423     struct GNUNET_SERVICE_Client *client = cmc->tc->client;
4424
4425     GNUNET_break (0);
4426     finish_cmc_handling (cmc);
4427     GNUNET_SERVICE_client_drop (client);
4428     return;
4429   }
4430   /* FIXME-OPTIMIZE: maybe use another hash map here? */
4431   matched = GNUNET_NO;
4432   for (struct PendingMessage *pm = n->pending_msg_head; NULL != pm;
4433        pm = pm->prev_neighbour)
4434   {
4435     if (0 != GNUNET_memcmp (&fa->msg_uuid, &pm->msg_uuid))
4436       continue;
4437     matched = GNUNET_YES;
4438     if (GNUNET_YES == check_ack_against_pm (pm, fa))
4439     {
4440       struct GNUNET_TIME_Relative avg_ack_delay =
4441         GNUNET_TIME_relative_ntoh (fa->avg_ack_delay);
4442       // FIXME: update RTT and other reliability data!
4443       // ISSUE: we don't know which of n's queues the message(s)
4444       // took (and in fact the different messages might have gone
4445       // over different queues and possibly over multiple).
4446       // => track queues with PendingMessages, and update RTT only if
4447       //    the queue used is unique?
4448       //    -> how can we get loss rates?
4449       //    -> or, add extra state to Box and ACK to identify queue?
4450       // IDEA: generate MULTIPLE frag-uuids per fragment and track
4451       //    the queue with the fragment! (-> this logic must
4452       //    be moved into check_ack_against_pm!)
4453       (void) avg_ack_delay;
4454     }
4455     else
4456     {
4457       GNUNET_STATISTICS_update (GST_stats,
4458                                 "# FRAGMENT_ACKS dropped, no matching fragment",
4459                                 1,
4460                                 GNUNET_NO);
4461     }
4462     if (NULL == pm->head_frag)
4463     {
4464       // if entire message is ACKed, handle that as well.
4465       // => clean up PM, any post actions?
4466       free_pending_message (pm);
4467     }
4468     else
4469     {
4470       struct GNUNET_TIME_Relative reassembly_timeout =
4471         GNUNET_TIME_relative_ntoh (fa->reassembly_timeout);
4472       // OPTIMIZE-FIXME: adjust retransmission strategy based on
4473       // reassembly_timeout!
4474       (void) reassembly_timeout;
4475     }
4476     break;
4477   }
4478   if (GNUNET_NO == matched)
4479   {
4480     GNUNET_STATISTICS_update (GST_stats,
4481                               "# FRAGMENT_ACKS dropped, no matching pending message",
4482                               1,
4483                               GNUNET_NO);
4484   }
4485   finish_cmc_handling (cmc);
4486 }
4487
4488
4489 /**
4490  * Communicator gave us a reliability box.  Check the message.
4491  *
4492  * @param cls a `struct CommunicatorMessageContext`
4493  * @param rb the send message that was sent
4494  * @return #GNUNET_YES if message is well-formed
4495  */
4496 static int
4497 check_reliability_box (void *cls, const struct TransportReliabilityBox *rb)
4498 {
4499   GNUNET_MQ_check_boxed_message (rb);
4500   return GNUNET_YES;
4501 }
4502
4503
4504 /**
4505  * Communicator gave us a reliability box.  Process the request.
4506  *
4507  * @param cls a `struct CommunicatorMessageContext` (must call
4508  * #finish_cmc_handling() when done)
4509  * @param rb the message that was received
4510  */
4511 static void
4512 handle_reliability_box (void *cls, const struct TransportReliabilityBox *rb)
4513 {
4514   struct CommunicatorMessageContext *cmc = cls;
4515   const struct GNUNET_MessageHeader *inbox =
4516     (const struct GNUNET_MessageHeader *) &rb[1];
4517
4518   if (0 == ntohl (rb->ack_countdown))
4519   {
4520     struct TransportReliabilityAckMessage *ack;
4521
4522     /* FIXME-OPTIMIZE: implement cummulative ACKs and ack_countdown,
4523        then setting the avg_ack_delay field below: */
4524     ack = GNUNET_malloc (sizeof (*ack) + sizeof (struct GNUNET_ShortHashCode));
4525     ack->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_ACK);
4526     ack->header.size =
4527       htons (sizeof (*ack) + sizeof (struct GNUNET_ShortHashCode));
4528     memcpy (&ack[1], &rb->msg_uuid, sizeof (struct GNUNET_ShortHashCode));
4529     route_message (&cmc->im.sender, &ack->header, RMO_DV_ALLOWED);
4530   }
4531   /* continue with inner message */
4532   demultiplex_with_cmc (cmc, inbox);
4533 }
4534
4535
4536 /**
4537  * Communicator gave us a reliability ack.  Process the request.
4538  *
4539  * @param cls a `struct CommunicatorMessageContext` (must call
4540  * #finish_cmc_handling() when done)
4541  * @param ra the message that was received
4542  */
4543 static void
4544 handle_reliability_ack (void *cls,
4545                         const struct TransportReliabilityAckMessage *ra)
4546 {
4547   struct CommunicatorMessageContext *cmc = cls;
4548   struct Neighbour *n;
4549   unsigned int n_acks;
4550   const struct GNUNET_ShortHashCode *msg_uuids;
4551   struct PendingMessage *nxt;
4552   int matched;
4553
4554   n = GNUNET_CONTAINER_multipeermap_get (neighbours, &cmc->im.sender);
4555   if (NULL == n)
4556   {
4557     struct GNUNET_SERVICE_Client *client = cmc->tc->client;
4558
4559     GNUNET_break (0);
4560     finish_cmc_handling (cmc);
4561     GNUNET_SERVICE_client_drop (client);
4562     return;
4563   }
4564   n_acks = (ntohs (ra->header.size) - sizeof (*ra)) /
4565            sizeof (struct GNUNET_ShortHashCode);
4566   msg_uuids = (const struct GNUNET_ShortHashCode *) &ra[1];
4567
4568   /* FIXME-OPTIMIZE: maybe use another hash map here? */
4569   matched = GNUNET_NO;
4570   for (struct PendingMessage *pm = n->pending_msg_head; NULL != pm; pm = nxt)
4571   {
4572     int in_list;
4573
4574     nxt = pm->next_neighbour;
4575     in_list = GNUNET_NO;
4576     for (unsigned int i = 0; i < n_acks; i++)
4577     {
4578       if (0 != GNUNET_memcmp (&msg_uuids[i], &pm->msg_uuid))
4579         continue;
4580       in_list = GNUNET_YES;
4581       break;
4582     }
4583     if (GNUNET_NO == in_list)
4584       continue;
4585
4586     /* this pm was acked! */
4587     matched = GNUNET_YES;
4588     free_pending_message (pm);
4589
4590     {
4591       struct GNUNET_TIME_Relative avg_ack_delay =
4592         GNUNET_TIME_relative_ntoh (ra->avg_ack_delay);
4593       // FIXME: update RTT and other reliability data!
4594       // ISSUE: we don't know which of n's queues the message(s)
4595       // took (and in fact the different messages might have gone
4596       // over different queues and possibly over multiple).
4597       // => track queues with PendingMessages, and update RTT only if
4598       //    the queue used is unique?
4599       //    -> how can we get loss rates?
4600       //    -> or, add extra state to MSG and ACKs to identify queue?
4601       //    -> if we do this, might just do the same for the avg_ack_delay!
4602       (void) avg_ack_delay;
4603     }
4604   }
4605   if (GNUNET_NO == matched)
4606   {
4607     GNUNET_STATISTICS_update (GST_stats,
4608                               "# FRAGMENT_ACKS dropped, no matching pending message",
4609                               1,
4610                               GNUNET_NO);
4611   }
4612   finish_cmc_handling (cmc);
4613 }
4614
4615
4616 /**
4617  * Communicator gave us a backchannel encapsulation.  Check the message.
4618  *
4619  * @param cls a `struct CommunicatorMessageContext`
4620  * @param be the send message that was sent
4621  * @return #GNUNET_YES if message is well-formed
4622  */
4623 static int
4624 check_backchannel_encapsulation (
4625   void *cls,
4626   const struct TransportBackchannelEncapsulationMessage *be)
4627 {
4628   uint16_t size = ntohs (be->header.size);
4629
4630   (void) cls;
4631   if ((size - sizeof (*be)) <
4632       (sizeof (struct TransportBackchannelRequestPayload) +
4633        sizeof (struct GNUNET_MessageHeader)))
4634   {
4635     GNUNET_break_op (0);
4636     return GNUNET_SYSERR;
4637   }
4638   return GNUNET_YES;
4639 }
4640
4641
4642 /**
4643  * We received the plaintext @a msg from backtalker @a b. Forward
4644  * it to the respective communicator.
4645  *
4646  * @param b a backtalker
4647  * @param msg a message, consisting of a `struct GNUNET_MessageHeader`
4648  *        followed by the target name of the communicator
4649  * @param msg_size number of bytes in @a msg
4650  */
4651 static void
4652 forward_backchannel_payload (struct Backtalker *b,
4653                              const void *msg,
4654                              size_t msg_size)
4655 {
4656   struct GNUNET_TRANSPORT_CommunicatorBackchannelIncoming *cbi;
4657   struct GNUNET_MQ_Envelope *env;
4658   struct TransportClient *tc;
4659   const struct GNUNET_MessageHeader *mh;
4660   const char *target_communicator;
4661   uint16_t mhs;
4662
4663   /* Determine target_communicator and check @a msg is well-formed */
4664   mh = msg;
4665   mhs = ntohs (mh->size);
4666   if (mhs <= msg_size)
4667   {
4668     GNUNET_break_op (0);
4669     return;
4670   }
4671   target_communicator = &((const char *) msg)[ntohs (mh->size)];
4672   if ('\0' != target_communicator[msg_size - mhs - 1])
4673   {
4674     GNUNET_break_op (0);
4675     return;
4676   }
4677   /* Find client providing this communicator */
4678   for (tc = clients_head; NULL != tc; tc = tc->next)
4679     if ((CT_COMMUNICATOR == tc->type) &&
4680         (0 ==
4681          strcmp (tc->details.communicator.address_prefix, target_communicator)))
4682       break;
4683   if (NULL == tc)
4684   {
4685     char *stastr;
4686
4687     GNUNET_asprintf (
4688       &stastr,
4689       "# Backchannel message dropped: target communicator `%s' unknown",
4690       target_communicator);
4691     GNUNET_STATISTICS_update (GST_stats, stastr, 1, GNUNET_NO);
4692     GNUNET_free (stastr);
4693     return;
4694   }
4695   /* Finally, deliver backchannel message to communicator */
4696   env = GNUNET_MQ_msg_extra (
4697     cbi,
4698     msg_size,
4699     GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL_INCOMING);
4700   cbi->pid = b->pid;
4701   memcpy (&cbi[1], msg, msg_size);
4702   GNUNET_MQ_send (tc->mq, env);
4703 }
4704
4705
4706 /**
4707  * Free data structures associated with @a b.
4708  *
4709  * @param b data structure to release
4710  */
4711 static void
4712 free_backtalker (struct Backtalker *b)
4713 {
4714   if (NULL != b->get)
4715   {
4716     GNUNET_PEERSTORE_iterate_cancel (b->get);
4717     b->get = NULL;
4718     GNUNET_assert (NULL != b->cmc);
4719     finish_cmc_handling (b->cmc);
4720     b->cmc = NULL;
4721   }
4722   if (NULL != b->task)
4723   {
4724     GNUNET_SCHEDULER_cancel (b->task);
4725     b->task = NULL;
4726   }
4727   if (NULL != b->sc)
4728   {
4729     GNUNET_PEERSTORE_store_cancel (b->sc);
4730     b->sc = NULL;
4731   }
4732   GNUNET_assert (
4733     GNUNET_YES ==
4734     GNUNET_CONTAINER_multipeermap_remove (backtalkers, &b->pid, b));
4735   GNUNET_free (b);
4736 }
4737
4738
4739 /**
4740  * Callback to free backtalker records.
4741  *
4742  * @param cls NULL
4743  * @param pid unused
4744  * @param value a `struct Backtalker`
4745  * @return #GNUNET_OK (always)
4746  */
4747 static int
4748 free_backtalker_cb (void *cls,
4749                     const struct GNUNET_PeerIdentity *pid,
4750                     void *value)
4751 {
4752   struct Backtalker *b = value;
4753
4754   (void) cls;
4755   (void) pid;
4756   free_backtalker (b);
4757   return GNUNET_OK;
4758 }
4759
4760
4761 /**
4762  * Function called when it is time to clean up a backtalker.
4763  *
4764  * @param cls a `struct Backtalker`
4765  */
4766 static void
4767 backtalker_timeout_cb (void *cls)
4768 {
4769   struct Backtalker *b = cls;
4770
4771   b->task = NULL;
4772   if (0 != GNUNET_TIME_absolute_get_remaining (b->timeout).rel_value_us)
4773   {
4774     b->task = GNUNET_SCHEDULER_add_at (b->timeout, &backtalker_timeout_cb, b);
4775     return;
4776   }
4777   GNUNET_assert (NULL == b->sc);
4778   free_backtalker (b);
4779 }
4780
4781
4782 /**
4783  * Function called with the monotonic time of a backtalker
4784  * by PEERSTORE. Updates the time and continues processing.
4785  *
4786  * @param cls a `struct Backtalker`
4787  * @param record the information found, NULL for the last call
4788  * @param emsg error message
4789  */
4790 static void
4791 backtalker_monotime_cb (void *cls,
4792                         const struct GNUNET_PEERSTORE_Record *record,
4793                         const char *emsg)
4794 {
4795   struct Backtalker *b = cls;
4796   struct GNUNET_TIME_AbsoluteNBO *mtbe;
4797   struct GNUNET_TIME_Absolute mt;
4798
4799   (void) emsg;
4800   if (NULL == record)
4801   {
4802     /* we're done with #backtalker_monotime_cb() invocations,
4803        continue normal processing */
4804     b->get = NULL;
4805     GNUNET_assert (NULL != b->cmc);
4806     finish_cmc_handling (b->cmc);
4807     b->cmc = NULL;
4808     if (0 != b->body_size)
4809       forward_backchannel_payload (b, &b[1], b->body_size);
4810     return;
4811   }
4812   if (sizeof (*mtbe) != record->value_size)
4813   {
4814     GNUNET_break (0);
4815     return;
4816   }
4817   mtbe = record->value;
4818   mt = GNUNET_TIME_absolute_ntoh (*mtbe);
4819   if (mt.abs_value_us > b->monotonic_time.abs_value_us)
4820   {
4821     GNUNET_STATISTICS_update (
4822       GST_stats,
4823       "# Backchannel messages dropped: monotonic time not increasing",
4824       1,
4825       GNUNET_NO);
4826     b->monotonic_time = mt;
4827     /* Setting body_size to 0 prevents call to #forward_backchannel_payload() */
4828     b->body_size = 0;
4829     return;
4830   }
4831 }
4832
4833
4834 /**
4835  * Function called by PEERSTORE when the store operation of
4836  * a backtalker's monotonic time is complete.
4837  *
4838  * @param cls the `struct Backtalker`
4839  * @param success #GNUNET_OK on success
4840  */
4841 static void
4842 backtalker_monotime_store_cb (void *cls, int success)
4843 {
4844   struct Backtalker *b = cls;
4845
4846   if (GNUNET_OK != success)
4847   {
4848     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
4849                 "Failed to store backtalker's monotonic time in PEERSTORE!\n");
4850   }
4851   b->sc = NULL;
4852   b->task = GNUNET_SCHEDULER_add_at (b->timeout, &backtalker_timeout_cb, b);
4853 }
4854
4855
4856 /**
4857  * The backtalker @a b monotonic time changed. Update PEERSTORE.
4858  *
4859  * @param b a backtalker with updated monotonic time
4860  */
4861 static void
4862 update_backtalker_monotime (struct Backtalker *b)
4863 {
4864   struct GNUNET_TIME_AbsoluteNBO mtbe;
4865
4866   if (NULL != b->sc)
4867   {
4868     GNUNET_PEERSTORE_store_cancel (b->sc);
4869     b->sc = NULL;
4870   }
4871   else
4872   {
4873     GNUNET_SCHEDULER_cancel (b->task);
4874     b->task = NULL;
4875   }
4876   mtbe = GNUNET_TIME_absolute_hton (b->monotonic_time);
4877   b->sc = GNUNET_PEERSTORE_store (peerstore,
4878                                   "transport",
4879                                   &b->pid,
4880                                   "transport-backchannel-monotonic-time",
4881                                   &mtbe,
4882                                   sizeof (mtbe),
4883                                   GNUNET_TIME_UNIT_FOREVER_ABS,
4884                                   GNUNET_PEERSTORE_STOREOPTION_REPLACE,
4885                                   &backtalker_monotime_store_cb,
4886                                   b);
4887 }
4888
4889
4890 /**
4891  * Communicator gave us a backchannel encapsulation.  Process the request.
4892  * (We are not the origin of the backchannel here, the communicator simply
4893  * received a backchannel message and we are expected to forward it.)
4894  *
4895  * @param cls a `struct CommunicatorMessageContext` (must call
4896  * #finish_cmc_handling() when done)
4897  * @param be the message that was received
4898  */
4899 static void
4900 handle_backchannel_encapsulation (
4901   void *cls,
4902   const struct TransportBackchannelEncapsulationMessage *be)
4903 {
4904   struct CommunicatorMessageContext *cmc = cls;
4905   struct BackchannelKeyState key;
4906   struct GNUNET_HashCode hmac;
4907   const char *hdr;
4908   size_t hdr_len;
4909
4910   if (0 != GNUNET_memcmp (&be->target, &GST_my_identity))
4911   {
4912     /* not for me, try to route to target */
4913     route_message (&be->target,
4914                    GNUNET_copy_message (&be->header),
4915                    RMO_DV_ALLOWED);
4916     finish_cmc_handling (cmc);
4917     return;
4918   }
4919   dh_key_derive_eph_pub (&be->ephemeral_key, &be->iv, &key);
4920   hdr = (const char *) &be[1];
4921   hdr_len = ntohs (be->header.size) - sizeof (*be);
4922   bc_hmac (&key, &hmac, hdr, hdr_len);
4923   if (0 != GNUNET_memcmp (&hmac, &be->hmac))
4924   {
4925     /* HMAC missmatch, disard! */
4926     GNUNET_break_op (0);
4927     finish_cmc_handling (cmc);
4928     return;
4929   }
4930   /* begin actual decryption */
4931   {
4932     struct Backtalker *b;
4933     struct GNUNET_TIME_Absolute monotime;
4934     struct TransportBackchannelRequestPayload ppay;
4935     char body[hdr_len - sizeof (ppay)];
4936
4937     GNUNET_assert (hdr_len >=
4938                    sizeof (ppay) + sizeof (struct GNUNET_MessageHeader));
4939     bc_decrypt (&key, &ppay, hdr, sizeof (ppay));
4940     bc_decrypt (&key, &body, &hdr[sizeof (ppay)], hdr_len - sizeof (ppay));
4941     bc_key_clean (&key);
4942     monotime = GNUNET_TIME_absolute_ntoh (ppay.monotonic_time);
4943     b = GNUNET_CONTAINER_multipeermap_get (backtalkers, &ppay.sender);
4944     if ((NULL != b) && (monotime.abs_value_us < b->monotonic_time.abs_value_us))
4945     {
4946       GNUNET_STATISTICS_update (
4947         GST_stats,
4948         "# Backchannel messages dropped: monotonic time not increasing",
4949         1,
4950         GNUNET_NO);
4951       finish_cmc_handling (cmc);
4952       return;
4953     }
4954     if ((NULL == b) ||
4955         (0 != GNUNET_memcmp (&b->last_ephemeral, &be->ephemeral_key)))
4956     {
4957       /* Check signature */
4958       struct EphemeralConfirmationPS ec;
4959
4960       ec.purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL);
4961       ec.purpose.size = htonl (sizeof (ec));
4962       ec.target = GST_my_identity;
4963       ec.ephemeral_key = be->ephemeral_key;
4964       if (
4965         GNUNET_OK !=
4966         GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_EPHEMERAL,
4967                                     &ec.purpose,
4968                                     &ppay.sender_sig,
4969                                     &ppay.sender.public_key))
4970       {
4971         /* Signature invalid, disard! */
4972         GNUNET_break_op (0);
4973         finish_cmc_handling (cmc);
4974         return;
4975       }
4976     }
4977     if (NULL != b)
4978     {
4979       /* update key cache and mono time */
4980       b->last_ephemeral = be->ephemeral_key;
4981       b->monotonic_time = monotime;
4982       update_backtalker_monotime (b);
4983       forward_backchannel_payload (b, body, sizeof (body));
4984       b->timeout =
4985         GNUNET_TIME_relative_to_absolute (BACKCHANNEL_INACTIVITY_TIMEOUT);
4986       finish_cmc_handling (cmc);
4987       return;
4988     }
4989     /* setup data structure to cache signature AND check
4990        monotonic time with PEERSTORE before forwarding backchannel payload */
4991     b = GNUNET_malloc (sizeof (struct Backtalker) + sizeof (body));
4992     b->pid = ppay.sender;
4993     b->body_size = sizeof (body);
4994     memcpy (&b[1], body, sizeof (body));
4995     GNUNET_assert (GNUNET_YES ==
4996                    GNUNET_CONTAINER_multipeermap_put (
4997                      backtalkers,
4998                      &b->pid,
4999                      b,
5000                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
5001     b->monotonic_time = monotime; /* NOTE: to be checked still! */
5002     b->cmc = cmc;
5003     b->timeout =
5004       GNUNET_TIME_relative_to_absolute (BACKCHANNEL_INACTIVITY_TIMEOUT);
5005     b->task = GNUNET_SCHEDULER_add_at (b->timeout, &backtalker_timeout_cb, b);
5006     b->get = GNUNET_PEERSTORE_iterate (peerstore,
5007                                        "transport",
5008                                        &b->pid,
5009                                        "transport-backchannel-monotonic-time",
5010                                        &backtalker_monotime_cb,
5011                                        b);
5012   }
5013 }
5014
5015
5016 /**
5017  * Task called when we should check if any of the DV paths
5018  * we have learned to a target are due for garbage collection.
5019  *
5020  * Collects stale paths, and possibly frees the entire DV
5021  * entry if no paths are left. Otherwise re-schedules itself.
5022  *
5023  * @param cls a `struct DistanceVector`
5024  */
5025 static void
5026 path_cleanup_cb (void *cls)
5027 {
5028   struct DistanceVector *dv = cls;
5029   struct DistanceVectorHop *pos;
5030
5031   dv->timeout_task = NULL;
5032   while (NULL != (pos = dv->dv_head))
5033   {
5034     GNUNET_assert (dv == pos->dv);
5035     if (GNUNET_TIME_absolute_get_remaining (pos->timeout).rel_value_us > 0)
5036       break;
5037     free_distance_vector_hop (pos);
5038   }
5039   if (NULL == pos)
5040   {
5041     free_dv_route (dv);
5042     return;
5043   }
5044   dv->timeout_task =
5045     GNUNET_SCHEDULER_add_at (pos->timeout, &path_cleanup_cb, dv);
5046 }
5047
5048 /**
5049  * Task run to check whether the hops of the @a cls still
5050  * are validated, or if we need to core about disconnection.
5051  *
5052  * @param cls a `struct DistanceVector` (with core_visible set!)
5053  */
5054 static void
5055 check_dv_path_down (void *cls)
5056 {
5057   struct DistanceVector *dv = cls;
5058   struct Neighbour *n;
5059
5060   dv->visibility_task = NULL;
5061   GNUNET_assert (GNUNET_YES == dv->core_visible);
5062   for (struct DistanceVectorHop *pos = dv->dv_head; NULL != pos;
5063        pos = pos->next_dv)
5064   {
5065     if (0 <
5066         GNUNET_TIME_absolute_get_remaining (pos->path_valid_until).rel_value_us)
5067     {
5068       dv->visibility_task = GNUNET_SCHEDULER_add_at (pos->path_valid_until,
5069                                                      &check_dv_path_down,
5070                                                      dv);
5071       return;
5072     }
5073   }
5074   /* all paths invalid, make dv core-invisible */
5075   dv->core_visible = GNUNET_NO;
5076   n = GNUNET_CONTAINER_multipeermap_get (neighbours, &dv->target);
5077   if ((NULL != n) && (GNUNET_YES == n->core_visible))
5078     return; /* no need to tell core, connection still up! */
5079   cores_send_disconnect_info (&dv->target);
5080 }
5081
5082
5083 /**
5084  * The @a hop is a validated path to the respective target
5085  * peer and we should tell core about it -- and schedule
5086  * a job to revoke the state.
5087  *
5088  * @param hop a path to some peer that is the reason for activation
5089  */
5090 static void
5091 activate_core_visible_dv_path (struct DistanceVectorHop *hop)
5092 {
5093   struct DistanceVector *dv = hop->dv;
5094   struct Neighbour *n;
5095
5096   GNUNET_assert (GNUNET_NO == dv->core_visible);
5097   GNUNET_assert (NULL == dv->visibility_task);
5098
5099   dv->core_visible = GNUNET_YES;
5100   dv->visibility_task =
5101     GNUNET_SCHEDULER_add_at (hop->path_valid_until, &check_dv_path_down, dv);
5102   n = GNUNET_CONTAINER_multipeermap_get (neighbours, &dv->target);
5103   if ((NULL != n) && (GNUNET_YES == n->core_visible))
5104     return; /* no need to tell core, connection already up! */
5105   cores_send_connect_info (&dv->target,
5106                            (NULL != n)
5107                              ? GNUNET_BANDWIDTH_value_sum (n->quota_out,
5108                                                            dv->quota_out)
5109                              : dv->quota_out);
5110 }
5111
5112
5113 /**
5114  * We have learned a @a path through the network to some other peer, add it to
5115  * our DV data structure (returning #GNUNET_YES on success).
5116  *
5117  * We do not add paths if we have a sufficient number of shorter
5118  * paths to this target already (returning #GNUNET_NO).
5119  *
5120  * We also do not add problematic paths, like those where we lack the first
5121  * hop in our neighbour list (i.e. due to a topology change) or where some
5122  * non-first hop is in our neighbour list (returning #GNUNET_SYSERR).
5123  *
5124  * @param path the path we learned, path[0] should be us,
5125  *             and then path contains a valid path from us to `path[path_len-1]`
5126  *             path[1] should be a direct neighbour (we should check!)
5127  * @param path_len number of entries on the @a path, at least three!
5128  * @param network_latency how long does the message take from us to
5129  * `path[path_len-1]`? set to "forever" if unknown
5130  * @param path_valid_until how long is this path considered validated? Maybe be
5131  * zero.
5132  * @return #GNUNET_YES on success,
5133  *         #GNUNET_NO if we have better path(s) to the target
5134  *         #GNUNET_SYSERR if the path is useless and/or invalid
5135  *                         (i.e. path[1] not a direct neighbour
5136  *                        or path[i+1] is a direct neighbour for i>0)
5137  */
5138 static int
5139 learn_dv_path (const struct GNUNET_PeerIdentity *path,
5140                unsigned int path_len,
5141                struct GNUNET_TIME_Relative network_latency,
5142                struct GNUNET_TIME_Absolute path_valid_until)
5143 {
5144   struct DistanceVectorHop *hop;
5145   struct DistanceVector *dv;
5146   struct Neighbour *next_hop;
5147   unsigned int shorter_distance;
5148
5149   if (path_len < 3)
5150   {
5151     /* what a boring path! not allowed! */
5152     GNUNET_break (0);
5153     return GNUNET_SYSERR;
5154   }
5155   GNUNET_assert (0 == GNUNET_memcmp (&GST_my_identity, &path[0]));
5156   next_hop = GNUNET_CONTAINER_multipeermap_get (neighbours, &path[1]);
5157   if (NULL == next_hop)
5158   {
5159     /* next hop must be a neighbour, otherwise this whole thing is useless! */
5160     GNUNET_break (0);
5161     return GNUNET_SYSERR;
5162   }
5163   for (unsigned int i = 2; i < path_len; i++)
5164     if (NULL != GNUNET_CONTAINER_multipeermap_get (neighbours, &path[i]))
5165     {
5166       /* Useless path, we have a direct connection to some hop
5167          in the middle of the path, so this one doesn't even
5168          seem terribly useful for redundancy */
5169       return GNUNET_SYSERR;
5170     }
5171   dv = GNUNET_CONTAINER_multipeermap_get (dv_routes, &path[path_len - 1]);
5172   if (NULL == dv)
5173   {
5174     dv = GNUNET_new (struct DistanceVector);
5175     dv->target = path[path_len - 1];
5176     dv->timeout_task = GNUNET_SCHEDULER_add_delayed (DV_PATH_VALIDITY_TIMEOUT,
5177                                                      &path_cleanup_cb,
5178                                                      dv);
5179     GNUNET_assert (GNUNET_OK ==
5180                    GNUNET_CONTAINER_multipeermap_put (
5181                      dv_routes,
5182                      &dv->target,
5183                      dv,
5184                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
5185   }
5186   /* Check if we have this path already! */
5187   shorter_distance = 0;
5188   for (struct DistanceVectorHop *pos = dv->dv_head; NULL != pos;
5189        pos = pos->next_dv)
5190   {
5191     if (pos->distance < path_len - 2)
5192       shorter_distance++;
5193     /* Note that the distances in 'pos' excludes us (path[0]) and
5194        the next_hop (path[1]), so we need to subtract two
5195        and check next_hop explicitly */
5196     if ((pos->distance == path_len - 2) && (pos->next_hop == next_hop))
5197     {
5198       int match = GNUNET_YES;
5199
5200       for (unsigned int i = 0; i < pos->distance; i++)
5201       {
5202         if (0 != GNUNET_memcmp (&pos->path[i], &path[i + 2]))
5203         {
5204           match = GNUNET_NO;
5205           break;
5206         }
5207       }
5208       if (GNUNET_YES == match)
5209       {
5210         struct GNUNET_TIME_Relative last_timeout;
5211
5212         /* Re-discovered known path, update timeout */
5213         GNUNET_STATISTICS_update (GST_stats,
5214                                   "# Known DV path refreshed",
5215                                   1,
5216                                   GNUNET_NO);
5217         last_timeout = GNUNET_TIME_absolute_get_remaining (pos->timeout);
5218         pos->timeout =
5219           GNUNET_TIME_relative_to_absolute (DV_PATH_VALIDITY_TIMEOUT);
5220         pos->path_valid_until =
5221           GNUNET_TIME_absolute_max (pos->path_valid_until, path_valid_until);
5222         GNUNET_CONTAINER_MDLL_remove (dv, dv->dv_head, dv->dv_tail, pos);
5223         GNUNET_CONTAINER_MDLL_insert (dv, dv->dv_head, dv->dv_tail, pos);
5224         if ((GNUNET_NO == dv->core_visible) &&
5225             (0 < GNUNET_TIME_absolute_get_remaining (path_valid_until)
5226                    .rel_value_us))
5227           activate_core_visible_dv_path (pos);
5228         if (last_timeout.rel_value_us <
5229             GNUNET_TIME_relative_subtract (DV_PATH_VALIDITY_TIMEOUT,
5230                                            DV_PATH_DISCOVERY_FREQUENCY)
5231               .rel_value_us)
5232         {
5233           /* Some peer send DV learn messages too often, we are learning
5234              the same path faster than it would be useful; do not forward! */
5235           return GNUNET_NO;
5236         }
5237         return GNUNET_YES;
5238       }
5239     }
5240   }
5241   /* Count how many shorter paths we have (incl. direct
5242      neighbours) before simply giving up on this one! */
5243   if (shorter_distance >= MAX_DV_PATHS_TO_TARGET)
5244   {
5245     /* We have a shorter path already! */
5246     return GNUNET_NO;
5247   }
5248   /* create new DV path entry */
5249   hop = GNUNET_malloc (sizeof (struct DistanceVectorHop) +
5250                        sizeof (struct GNUNET_PeerIdentity) * (path_len - 2));
5251   hop->next_hop = next_hop;
5252   hop->dv = dv;
5253   hop->path = (const struct GNUNET_PeerIdentity *) &hop[1];
5254   memcpy (&hop[1],
5255           &path[2],
5256           sizeof (struct GNUNET_PeerIdentity) * (path_len - 2));
5257   hop->timeout = GNUNET_TIME_relative_to_absolute (DV_PATH_VALIDITY_TIMEOUT);
5258   hop->path_valid_until = path_valid_until;
5259   hop->distance = path_len - 2;
5260   GNUNET_CONTAINER_MDLL_insert (dv, dv->dv_head, dv->dv_tail, hop);
5261   GNUNET_CONTAINER_MDLL_insert (neighbour,
5262                                 next_hop->dv_head,
5263                                 next_hop->dv_tail,
5264                                 hop);
5265   if ((GNUNET_NO == dv->core_visible) &&
5266       (0 < GNUNET_TIME_absolute_get_remaining (path_valid_until).rel_value_us))
5267     activate_core_visible_dv_path (hop);
5268   return GNUNET_YES;
5269 }
5270
5271
5272 /**
5273  * Communicator gave us a DV learn message.  Check the message.
5274  *
5275  * @param cls a `struct CommunicatorMessageContext`
5276  * @param dvl the send message that was sent
5277  * @return #GNUNET_YES if message is well-formed
5278  */
5279 static int
5280 check_dv_learn (void *cls, const struct TransportDVLearn *dvl)
5281 {
5282   uint16_t size = ntohs (dvl->header.size);
5283   uint16_t num_hops = ntohs (dvl->num_hops);
5284   const struct DVPathEntryP *hops = (const struct DVPathEntryP *) &dvl[1];
5285
5286   (void) cls;
5287   if (size != sizeof (*dvl) + num_hops * sizeof (struct DVPathEntryP))
5288   {
5289     GNUNET_break_op (0);
5290     return GNUNET_SYSERR;
5291   }
5292   if (num_hops > MAX_DV_HOPS_ALLOWED)
5293   {
5294     GNUNET_break_op (0);
5295     return GNUNET_SYSERR;
5296   }
5297   for (unsigned int i = 0; i < num_hops; i++)
5298   {
5299     if (0 == GNUNET_memcmp (&dvl->initiator, &hops[i].hop))
5300     {
5301       GNUNET_break_op (0);
5302       return GNUNET_SYSERR;
5303     }
5304     if (0 == GNUNET_memcmp (&GST_my_identity, &hops[i].hop))
5305     {
5306       GNUNET_break_op (0);
5307       return GNUNET_SYSERR;
5308     }
5309   }
5310   return GNUNET_YES;
5311 }
5312
5313
5314 /**
5315  * Build and forward a DV learn message to @a next_hop.
5316  *
5317  * @param next_hop peer to send the message to
5318  * @param msg message received
5319  * @param bi_history bitmask specifying hops on path that were bidirectional
5320  * @param nhops length of the @a hops array
5321  * @param hops path the message traversed so far
5322  * @param in_time when did we receive the message, used to calculate network
5323  * delay
5324  */
5325 static void
5326 forward_dv_learn (const struct GNUNET_PeerIdentity *next_hop,
5327                   const struct TransportDVLearn *msg,
5328                   uint16_t bi_history,
5329                   uint16_t nhops,
5330                   const struct DVPathEntryP *hops,
5331                   struct GNUNET_TIME_Absolute in_time)
5332 {
5333   struct DVPathEntryP *dhops;
5334   struct TransportDVLearn *fwd;
5335   struct GNUNET_TIME_Relative nnd;
5336
5337   /* compute message for forwarding */
5338   GNUNET_assert (nhops < MAX_DV_HOPS_ALLOWED);
5339   fwd = GNUNET_malloc (sizeof (struct TransportDVLearn) +
5340                        (nhops + 1) * sizeof (struct DVPathEntryP));
5341   fwd->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN);
5342   fwd->header.size = htons (sizeof (struct TransportDVLearn) +
5343                             (nhops + 1) * sizeof (struct DVPathEntryP));
5344   fwd->num_hops = htons (nhops + 1);
5345   fwd->bidirectional = htons (bi_history);
5346   nnd = GNUNET_TIME_relative_add (GNUNET_TIME_absolute_get_duration (in_time),
5347                                   GNUNET_TIME_relative_ntoh (
5348                                     msg->non_network_delay));
5349   fwd->non_network_delay = GNUNET_TIME_relative_hton (nnd);
5350   fwd->init_sig = msg->init_sig;
5351   fwd->initiator = msg->initiator;
5352   fwd->challenge = msg->challenge;
5353   dhops = (struct DVPathEntryP *) &fwd[1];
5354   GNUNET_memcpy (dhops, hops, sizeof (struct DVPathEntryP) * nhops);
5355   dhops[nhops].hop = GST_my_identity;
5356   {
5357     struct DvHopPS dhp = {.purpose.purpose =
5358                             htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_HOP),
5359                           .purpose.size = htonl (sizeof (dhp)),
5360                           .pred = dhops[nhops - 1].hop,
5361                           .succ = *next_hop,
5362                           .challenge = msg->challenge};
5363
5364     GNUNET_assert (GNUNET_OK ==
5365                    GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
5366                                              &dhp.purpose,
5367                                              &dhops[nhops].hop_sig));
5368   }
5369   route_message (next_hop, &fwd->header, RMO_UNCONFIRMED_ALLOWED);
5370 }
5371
5372
5373 /**
5374  * Check signature of type #GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR
5375  *
5376  * @param init the signer
5377  * @param challenge the challenge that was signed
5378  * @param init_sig signature presumably by @a init
5379  * @return #GNUNET_OK if the signature is valid
5380  */
5381 static int
5382 validate_dv_initiator_signature (
5383   const struct GNUNET_PeerIdentity *init,
5384   const struct GNUNET_ShortHashCode *challenge,
5385   const struct GNUNET_CRYPTO_EddsaSignature *init_sig)
5386 {
5387   struct DvInitPS ip = {.purpose.purpose = htonl (
5388                           GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR),
5389                         .purpose.size = htonl (sizeof (ip)),
5390                         .challenge = *challenge};
5391
5392   if (
5393     GNUNET_OK !=
5394     GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR,
5395                                 &ip.purpose,
5396                                 init_sig,
5397                                 &init->public_key))
5398   {
5399     GNUNET_break_op (0);
5400     return GNUNET_SYSERR;
5401   }
5402   return GNUNET_OK;
5403 }
5404
5405
5406 /**
5407  * Communicator gave us a DV learn message.  Process the request.
5408  *
5409  * @param cls a `struct CommunicatorMessageContext` (must call
5410  * #finish_cmc_handling() when done)
5411  * @param dvl the message that was received
5412  */
5413 static void
5414 handle_dv_learn (void *cls, const struct TransportDVLearn *dvl)
5415 {
5416   struct CommunicatorMessageContext *cmc = cls;
5417   enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc;
5418   int bi_hop;
5419   uint16_t nhops;
5420   uint16_t bi_history;
5421   const struct DVPathEntryP *hops;
5422   int do_fwd;
5423   int did_initiator;
5424   struct GNUNET_TIME_Absolute in_time;
5425
5426   nhops = ntohs (dvl->bidirectional); /* 0 = sender is initiator */
5427   bi_history = ntohs (dvl->bidirectional);
5428   hops = (const struct DVPathEntryP *) &dvl[1];
5429   if (0 == nhops)
5430   {
5431     /* sanity check */
5432     if (0 != GNUNET_memcmp (&dvl->initiator, &cmc->im.sender))
5433     {
5434       GNUNET_break (0);
5435       finish_cmc_handling (cmc);
5436       return;
5437     }
5438   }
5439   else
5440   {
5441     /* sanity check */
5442     if (0 != GNUNET_memcmp (&hops[nhops - 1].hop, &cmc->im.sender))
5443     {
5444       GNUNET_break (0);
5445       finish_cmc_handling (cmc);
5446       return;
5447     }
5448   }
5449
5450   GNUNET_assert (CT_COMMUNICATOR == cmc->tc->type);
5451   cc = cmc->tc->details.communicator.cc;
5452   bi_hop = (GNUNET_TRANSPORT_CC_RELIABLE ==
5453             cc); // FIXME: add bi-directional flag to cc?
5454   in_time = GNUNET_TIME_absolute_get ();
5455
5456   /* continue communicator here, everything else can happen asynchronous! */
5457   finish_cmc_handling (cmc);
5458
5459   /* OPTIMIZE-FIXME: Technically, we only need to bother checking
5460      the initiator signature if we send the message back to the initiator... */
5461   if (GNUNET_OK != validate_dv_initiator_signature (&dvl->initiator,
5462                                                     &dvl->challenge,
5463                                                     &dvl->init_sig))
5464   {
5465     GNUNET_break_op (0);
5466     return;
5467   }
5468   // FIXME: asynchronously (!) verify hop-by-hop signatures!
5469   // => if signature verification load too high, implement random drop
5470   // strategy!?
5471
5472   do_fwd = GNUNET_YES;
5473   if (0 == GNUNET_memcmp (&GST_my_identity, &dvl->initiator))
5474   {
5475     struct GNUNET_PeerIdentity path[nhops + 1];
5476     struct GNUNET_TIME_Relative host_latency_sum;
5477     struct GNUNET_TIME_Relative latency;
5478     struct GNUNET_TIME_Relative network_latency;
5479
5480     /* We initiated this, learn the forward path! */
5481     path[0] = GST_my_identity;
5482     path[1] = hops[0].hop;
5483     host_latency_sum = GNUNET_TIME_relative_ntoh (dvl->non_network_delay);
5484
5485     // Need also something to lookup initiation time
5486     // to compute RTT! -> add RTT argument here?
5487     latency = GNUNET_TIME_UNIT_FOREVER_REL; // FIXME: initialize properly
5488     // (based on dvl->challenge, we can identify time of origin!)
5489
5490     network_latency = GNUNET_TIME_relative_subtract (latency, host_latency_sum);
5491     /* assumption: latency on all links is the same */
5492     network_latency = GNUNET_TIME_relative_divide (network_latency, nhops);
5493
5494     for (unsigned int i = 2; i <= nhops; i++)
5495     {
5496       struct GNUNET_TIME_Relative ilat;
5497
5498       /* assumption: linear latency increase per hop */
5499       ilat = GNUNET_TIME_relative_multiply (network_latency, i);
5500       path[i] = hops[i - 1].hop;
5501       learn_dv_path (path,
5502                      i,
5503                      ilat,
5504                      GNUNET_TIME_relative_to_absolute (
5505                        ADDRESS_VALIDATION_LIFETIME));
5506     }
5507     /* as we initiated, do not forward again (would be circular!) */
5508     do_fwd = GNUNET_NO;
5509     return;
5510   }
5511   else if (bi_hop)
5512   {
5513     /* last hop was bi-directional, we could learn something here! */
5514     struct GNUNET_PeerIdentity path[nhops + 2];
5515
5516     path[0] = GST_my_identity;
5517     path[1] = hops[nhops - 1].hop; /* direct neighbour == predecessor! */
5518     for (unsigned int i = 0; i < nhops; i++)
5519     {
5520       int iret;
5521
5522       if (0 == (bi_history & (1 << i)))
5523         break; /* i-th hop not bi-directional, stop learning! */
5524       if (i == nhops)
5525       {
5526         path[i + 2] = dvl->initiator;
5527       }
5528       else
5529       {
5530         path[i + 2] = hops[nhops - i - 2].hop;
5531       }
5532
5533       iret = learn_dv_path (path,
5534                             i + 2,
5535                             GNUNET_TIME_UNIT_FOREVER_REL,
5536                             GNUNET_TIME_UNIT_ZERO_ABS);
5537       if (GNUNET_SYSERR == iret)
5538       {
5539         /* path invalid or too long to be interesting for US, thus should also
5540            not be interesting to our neighbours, cut path when forwarding to
5541            'i' hops, except of course for the one that goes back to the
5542            initiator */
5543         GNUNET_STATISTICS_update (GST_stats,
5544                                   "# DV learn not forwarded due invalidity of path",
5545                                   1,
5546                                   GNUNET_NO);
5547         do_fwd = GNUNET_NO;
5548         break;
5549       }
5550       if ((GNUNET_NO == iret) && (nhops == i + 1))
5551       {
5552         /* we have better paths, and this is the longest target,
5553            so there cannot be anything interesting later */
5554         GNUNET_STATISTICS_update (GST_stats,
5555                                   "# DV learn not forwarded, got better paths",
5556                                   1,
5557                                   GNUNET_NO);
5558         do_fwd = GNUNET_NO;
5559         break;
5560       }
5561     }
5562   }
5563
5564   if (MAX_DV_HOPS_ALLOWED == nhops)
5565   {
5566     /* At limit, we're out of here! */
5567     finish_cmc_handling (cmc);
5568     return;
5569   }
5570
5571   /* Forward to initiator, if path non-trivial and possible */
5572   bi_history = (bi_history << 1) | (bi_hop ? 1 : 0);
5573   did_initiator = GNUNET_NO;
5574   if ((1 < nhops) &&
5575       (GNUNET_YES ==
5576        GNUNET_CONTAINER_multipeermap_contains (neighbours, &dvl->initiator)))
5577   {
5578     /* send back to origin! */
5579     forward_dv_learn (&dvl->initiator, dvl, bi_history, nhops, hops, in_time);
5580     did_initiator = GNUNET_YES;
5581   }
5582   /* We forward under two conditions: either we still learned something
5583      ourselves (do_fwd), or the path was darn short and thus the initiator is
5584      likely to still be very interested in this (and we did NOT already
5585      send it back to the initiator) */
5586   if ((do_fwd) || ((nhops < MIN_DV_PATH_LENGTH_FOR_INITIATOR) &&
5587                    (GNUNET_NO == did_initiator)))
5588   {
5589     /* FIXME: loop over all neighbours, pick those with low
5590        queues AND that are not yet on the path; possibly
5591        adapt threshold to nhops! */
5592 #if FIXME
5593     forward_dv_learn (NULL, // fill in peer from iterator here!
5594                       dvl,
5595                       bi_history,
5596                       nhops,
5597                       hops,
5598                       in_time);
5599 #endif
5600   }
5601 }
5602
5603
5604 /**
5605  * Communicator gave us a DV box.  Check the message.
5606  *
5607  * @param cls a `struct CommunicatorMessageContext`
5608  * @param dvb the send message that was sent
5609  * @return #GNUNET_YES if message is well-formed
5610  */
5611 static int
5612 check_dv_box (void *cls, const struct TransportDVBox *dvb)
5613 {
5614   uint16_t size = ntohs (dvb->header.size);
5615   uint16_t num_hops = ntohs (dvb->num_hops);
5616   const struct GNUNET_PeerIdentity *hops =
5617     (const struct GNUNET_PeerIdentity *) &dvb[1];
5618   const struct GNUNET_MessageHeader *inbox =
5619     (const struct GNUNET_MessageHeader *) &hops[num_hops];
5620   uint16_t isize;
5621   uint16_t itype;
5622
5623   (void) cls;
5624   if (size < sizeof (*dvb) + num_hops * sizeof (struct GNUNET_PeerIdentity) +
5625                sizeof (struct GNUNET_MessageHeader))
5626   {
5627     GNUNET_break_op (0);
5628     return GNUNET_SYSERR;
5629   }
5630   isize = ntohs (inbox->size);
5631   if (size !=
5632       sizeof (*dvb) + num_hops * sizeof (struct GNUNET_PeerIdentity) + isize)
5633   {
5634     GNUNET_break_op (0);
5635     return GNUNET_SYSERR;
5636   }
5637   itype = ntohs (inbox->type);
5638   if ((GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX == itype) ||
5639       (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN == itype))
5640   {
5641     GNUNET_break_op (0);
5642     return GNUNET_SYSERR;
5643   }
5644   if (0 == GNUNET_memcmp (&dvb->origin, &GST_my_identity))
5645   {
5646     GNUNET_break_op (0);
5647     return GNUNET_SYSERR;
5648   }
5649   return GNUNET_YES;
5650 }
5651
5652
5653 /**
5654  * Create a DV Box message and queue it for transmission to
5655  * @ea next_hop.
5656  *
5657  * @param next_hop peer to receive the message next
5658  * @param total_hops how many hops did the message take so far
5659  * @param num_hops length of the @a hops array
5660  * @param origin origin of the message
5661  * @param hops next peer(s) to the destination, including destination
5662  * @param payload payload of the box
5663  * @param payload_size number of bytes in @a payload
5664  */
5665 static void
5666 forward_dv_box (struct Neighbour *next_hop,
5667                 uint16_t total_hops,
5668                 uint16_t num_hops,
5669                 const struct GNUNET_PeerIdentity *origin,
5670                 const struct GNUNET_PeerIdentity *hops,
5671                 const void *payload,
5672                 uint16_t payload_size)
5673 {
5674   struct TransportDVBox *dvb;
5675   struct GNUNET_PeerIdentity *dhops;
5676
5677   GNUNET_assert (UINT16_MAX < sizeof (struct TransportDVBox) +
5678                                 sizeof (struct GNUNET_PeerIdentity) * num_hops +
5679                                 payload_size);
5680   dvb = GNUNET_malloc (sizeof (struct TransportDVBox) +
5681                        sizeof (struct GNUNET_PeerIdentity) * num_hops +
5682                        payload_size);
5683   dvb->header.size =
5684     htons (sizeof (struct TransportDVBox) +
5685            sizeof (struct GNUNET_PeerIdentity) * num_hops + payload_size);
5686   dvb->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX);
5687   dvb->total_hops = htons (total_hops);
5688   dvb->num_hops = htons (num_hops);
5689   dvb->origin = *origin;
5690   dhops = (struct GNUNET_PeerIdentity *) &dvb[1];
5691   memcpy (dhops, hops, num_hops * sizeof (struct GNUNET_PeerIdentity));
5692   memcpy (&dhops[num_hops], payload, payload_size);
5693   route_message (&next_hop->pid, &dvb->header, RMO_NONE);
5694 }
5695
5696
5697 /**
5698  * Communicator gave us a DV box.  Process the request.
5699  *
5700  * @param cls a `struct CommunicatorMessageContext` (must call
5701  * #finish_cmc_handling() when done)
5702  * @param dvb the message that was received
5703  */
5704 static void
5705 handle_dv_box (void *cls, const struct TransportDVBox *dvb)
5706 {
5707   struct CommunicatorMessageContext *cmc = cls;
5708   uint16_t size = ntohs (dvb->header.size) - sizeof (*dvb);
5709   uint16_t num_hops = ntohs (dvb->num_hops);
5710   const struct GNUNET_PeerIdentity *hops =
5711     (const struct GNUNET_PeerIdentity *) &dvb[1];
5712   const struct GNUNET_MessageHeader *inbox =
5713     (const struct GNUNET_MessageHeader *) &hops[num_hops];
5714
5715   if (num_hops > 0)
5716   {
5717     /* We're trying from the end of the hops array, as we may be
5718        able to find a shortcut unknown to the origin that way */
5719     for (int i = num_hops - 1; i >= 0; i--)
5720     {
5721       struct Neighbour *n;
5722
5723       if (0 == GNUNET_memcmp (&hops[i], &GST_my_identity))
5724       {
5725         GNUNET_break_op (0);
5726         finish_cmc_handling (cmc);
5727         return;
5728       }
5729       n = GNUNET_CONTAINER_multipeermap_get (neighbours, &hops[i]);
5730       if (NULL == n)
5731         continue;
5732       forward_dv_box (n,
5733                       ntohs (dvb->total_hops) + 1,
5734                       num_hops - i - 1, /* number of hops left */
5735                       &dvb->origin,
5736                       &hops[i + 1], /* remaining hops */
5737                       (const void *) &dvb[1],
5738                       size);
5739       finish_cmc_handling (cmc);
5740       return;
5741     }
5742     /* Woopsie, next hop not in neighbours, drop! */
5743     GNUNET_STATISTICS_update (GST_stats,
5744                               "# DV Boxes dropped: next hop unknown",
5745                               1,
5746                               GNUNET_NO);
5747     finish_cmc_handling (cmc);
5748     return;
5749   }
5750   /* We are the target. Unbox and handle message. */
5751   cmc->im.sender = dvb->origin;
5752   cmc->total_hops = ntohs (dvb->total_hops);
5753   demultiplex_with_cmc (cmc, inbox);
5754 }
5755
5756
5757 /**
5758  * Client notified us about transmission from a peer.  Process the request.
5759  *
5760  * @param cls a `struct TransportClient` which sent us the message
5761  * @param obm the send message that was sent
5762  * @return #GNUNET_YES if message is well-formed
5763  */
5764 static int
5765 check_incoming_msg (void *cls,
5766                     const struct GNUNET_TRANSPORT_IncomingMessage *im)
5767 {
5768   struct TransportClient *tc = cls;
5769
5770   if (CT_COMMUNICATOR != tc->type)
5771   {
5772     GNUNET_break (0);
5773     return GNUNET_SYSERR;
5774   }
5775   GNUNET_MQ_check_boxed_message (im);
5776   return GNUNET_OK;
5777 }
5778
5779
5780 /**
5781  * Communicator gave us a transport address validation challenge.  Process the
5782  * request.
5783  *
5784  * @param cls a `struct CommunicatorMessageContext` (must call
5785  * #finish_cmc_handling() when done)
5786  * @param tvc the message that was received
5787  */
5788 static void
5789 handle_validation_challenge (void *cls,
5790                              const struct TransportValidationChallenge *tvc)
5791 {
5792   struct CommunicatorMessageContext *cmc = cls;
5793   struct TransportValidationResponse *tvr;
5794
5795   if (cmc->total_hops > 0)
5796   {
5797     /* DV routing is not allowed for validation challenges! */
5798     GNUNET_break_op (0);
5799     finish_cmc_handling (cmc);
5800     return;
5801   }
5802   tvr = GNUNET_new (struct TransportValidationResponse);
5803   tvr->header.type =
5804     htons (GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_RESPONSE);
5805   tvr->header.size = htons (sizeof (*tvr));
5806   tvr->challenge = tvc->challenge;
5807   tvr->origin_time = tvc->sender_time;
5808   tvr->validity_duration = cmc->im.expected_address_validity;
5809   {
5810     /* create signature */
5811     struct TransportValidationPS tvp =
5812       {.purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_CHALLENGE),
5813        .purpose.size = htonl (sizeof (tvp)),
5814        .validity_duration = tvr->validity_duration,
5815        .challenge = tvc->challenge};
5816
5817     GNUNET_assert (GNUNET_OK == GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
5818                                                           &tvp.purpose,
5819                                                           &tvr->signature));
5820   }
5821   route_message (&cmc->im.sender,
5822                  &tvr->header,
5823                  RMO_ANYTHING_GOES | RMO_REDUNDANT);
5824   finish_cmc_handling (cmc);
5825 }
5826
5827
5828 /**
5829  * Closure for #check_known_challenge.
5830  */
5831 struct CheckKnownChallengeContext
5832 {
5833   /**
5834    * Set to the challenge we are looking for.
5835    */
5836   const struct GNUNET_ShortHashCode *challenge;
5837
5838   /**
5839    * Set to a matching validation state, if one was found.
5840    */
5841   struct ValidationState *vs;
5842 };
5843
5844
5845 /**
5846  * Test if the validation state in @a value matches the
5847  * challenge from @a cls.
5848  *
5849  * @param cls a `struct CheckKnownChallengeContext`
5850  * @param pid unused (must match though)
5851  * @param value a `struct ValidationState`
5852  * @return #GNUNET_OK if not matching, #GNUNET_NO if match found
5853  */
5854 static int
5855 check_known_challenge (void *cls,
5856                        const struct GNUNET_PeerIdentity *pid,
5857                        void *value)
5858 {
5859   struct CheckKnownChallengeContext *ckac = cls;
5860   struct ValidationState *vs = value;
5861
5862   (void) pid;
5863   if (0 != GNUNET_memcmp (&vs->challenge, ckac->challenge))
5864     return GNUNET_OK;
5865   ckac->vs = vs;
5866   return GNUNET_NO;
5867 }
5868
5869
5870 /**
5871  * Function called when peerstore is done storing a
5872  * validated address.
5873  *
5874  * @param cls a `struct ValidationState`
5875  * @param success #GNUNET_YES on success
5876  */
5877 static void
5878 peerstore_store_validation_cb (void *cls, int success)
5879 {
5880   struct ValidationState *vs = cls;
5881
5882   vs->sc = NULL;
5883   if (GNUNET_YES == success)
5884     return;
5885   GNUNET_STATISTICS_update (GST_stats,
5886                             "# Peerstore failed to store foreign address",
5887                             1,
5888                             GNUNET_NO);
5889 }
5890
5891
5892 /**
5893  * Task run periodically to validate some address based on #validation_heap.
5894  *
5895  * @param cls NULL
5896  */
5897 static void
5898 validation_start_cb (void *cls);
5899
5900
5901 /**
5902  * Set the time for next_challenge of @a vs to @a new_time.
5903  * Updates the heap and if necessary reschedules the job.
5904  *
5905  * @param vs validation state to update
5906  * @param new_time new time for revalidation
5907  */
5908 static void
5909 update_next_challenge_time (struct ValidationState *vs,
5910                             struct GNUNET_TIME_Absolute new_time)
5911 {
5912   struct GNUNET_TIME_Relative delta;
5913
5914   if (new_time.abs_value_us == vs->next_challenge.abs_value_us)
5915     return; /* be lazy */
5916   vs->next_challenge = new_time;
5917   if (NULL == vs->hn)
5918     vs->hn =
5919       GNUNET_CONTAINER_heap_insert (validation_heap, vs, new_time.abs_value_us);
5920   else
5921     GNUNET_CONTAINER_heap_update_cost (vs->hn, new_time.abs_value_us);
5922   if ((vs != GNUNET_CONTAINER_heap_peek (validation_heap)) &&
5923       (NULL != validation_task))
5924     return;
5925   if (NULL != validation_task)
5926     GNUNET_SCHEDULER_cancel (validation_task);
5927   /* randomize a bit */
5928   delta.rel_value_us =
5929     GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
5930                               MIN_DELAY_ADDRESS_VALIDATION.rel_value_us);
5931   new_time = GNUNET_TIME_absolute_add (new_time, delta);
5932   validation_task =
5933     GNUNET_SCHEDULER_add_at (new_time, &validation_start_cb, NULL);
5934 }
5935
5936
5937 /**
5938  * Find the queue matching @a pid and @a address.
5939  *
5940  * @param pid peer the queue must go to
5941  * @param address address the queue must use
5942  * @return NULL if no such queue exists
5943  */
5944 static struct Queue *
5945 find_queue (const struct GNUNET_PeerIdentity *pid, const char *address)
5946 {
5947   struct Neighbour *n;
5948
5949   n = GNUNET_CONTAINER_multipeermap_get (neighbours, pid);
5950   if (NULL == n)
5951     return NULL;
5952   for (struct Queue *pos = n->queue_head; NULL != pos;
5953        pos = pos->next_neighbour)
5954   {
5955     if (0 == strcmp (pos->address, address))
5956       return pos;
5957   }
5958   return NULL;
5959 }
5960
5961
5962 /**
5963  * Task run periodically to check whether the validity of the given queue has
5964  * run its course. If so, finds either another queue to take over, or clears
5965  * the neighbour's `core_visible` flag. In the latter case, gives DV routes a
5966  * chance to take over, and if that fails, notifies CORE about the disconnect.
5967  *
5968  * @param cls a `struct Queue`
5969  */
5970 static void
5971 core_queue_visibility_check (void *cls)
5972 {
5973   struct Queue *q = cls;
5974
5975   q->visibility_task = NULL;
5976   if (0 != GNUNET_TIME_absolute_get_remaining (q->validated_until).rel_value_us)
5977   {
5978     q->visibility_task = GNUNET_SCHEDULER_add_at (q->validated_until,
5979                                                   &core_queue_visibility_check,
5980                                                   q);
5981     return;
5982   }
5983   update_neighbour_core_visibility (q->neighbour);
5984 }
5985
5986
5987 /**
5988  * Check whether the CORE visibility of @a n should change.  Finds either a
5989  * queue to preserve the visibility, or clears the neighbour's `core_visible`
5990  * flag. In the latter case, gives DV routes a chance to take over, and if
5991  * that fails, notifies CORE about the disconnect.  If so, check whether we
5992  * need to notify CORE.
5993  *
5994  * @param n neighbour to perform the check for
5995  */
5996 static void
5997 update_neighbour_core_visibility (struct Neighbour *n)
5998 {
5999   struct DistanceVector *dv;
6000
6001   GNUNET_assert (GNUNET_YES == n->core_visible);
6002   /* Check if _any_ queue of this neighbour is still valid, if so, schedule
6003      the #core_queue_visibility_check() task for that queue */
6004   for (struct Queue *q = n->queue_head; NULL != q; q = q->next_neighbour)
6005   {
6006     if (0 !=
6007         GNUNET_TIME_absolute_get_remaining (q->validated_until).rel_value_us)
6008     {
6009       /* found a valid queue, use this one */
6010       q->visibility_task =
6011         GNUNET_SCHEDULER_add_at (q->validated_until,
6012                                  &core_queue_visibility_check,
6013                                  q);
6014       return;
6015     }
6016   }
6017   n->core_visible = GNUNET_NO;
6018
6019   /* Check if _any_ DV route to this neighbour is currently
6020      valid, if so, do NOT tell core about the loss of direct
6021      connectivity (DV still counts!) */
6022   dv = GNUNET_CONTAINER_multipeermap_get (dv_routes, &n->pid);
6023   if (GNUNET_YES == dv->core_visible)
6024     return;
6025   /* Nothing works anymore, need to tell CORE about the loss of
6026      connectivity! */
6027   cores_send_disconnect_info (&n->pid);
6028 }
6029
6030
6031 /**
6032  * Communicator gave us a transport address validation response.  Process the
6033  * request.
6034  *
6035  * @param cls a `struct CommunicatorMessageContext` (must call
6036  * #finish_cmc_handling() when done)
6037  * @param tvr the message that was received
6038  */
6039 static void
6040 handle_validation_response (void *cls,
6041                             const struct TransportValidationResponse *tvr)
6042 {
6043   struct CommunicatorMessageContext *cmc = cls;
6044   struct ValidationState *vs;
6045   struct CheckKnownChallengeContext ckac = {.challenge = &tvr->challenge,
6046                                             .vs = NULL};
6047   struct GNUNET_TIME_Absolute origin_time;
6048   struct Queue *q;
6049   struct DistanceVector *dv;
6050   struct Neighbour *n;
6051
6052   /* check this is one of our challenges */
6053   (void) GNUNET_CONTAINER_multipeermap_get_multiple (validation_map,
6054                                                      &cmc->im.sender,
6055                                                      &check_known_challenge,
6056                                                      &ckac);
6057   if (NULL == (vs = ckac.vs))
6058   {
6059     /* This can happen simply if we 'forgot' the challenge by now,
6060        i.e. because we received the validation response twice */
6061     GNUNET_STATISTICS_update (GST_stats,
6062                               "# Validations dropped, challenge unknown",
6063                               1,
6064                               GNUNET_NO);
6065     finish_cmc_handling (cmc);
6066     return;
6067   }
6068
6069   /* sanity check on origin time */
6070   origin_time = GNUNET_TIME_absolute_ntoh (tvr->origin_time);
6071   if ((origin_time.abs_value_us < vs->first_challenge_use.abs_value_us) ||
6072       (origin_time.abs_value_us > vs->last_challenge_use.abs_value_us))
6073   {
6074     GNUNET_break_op (0);
6075     finish_cmc_handling (cmc);
6076     return;
6077   }
6078
6079   {
6080     /* check signature */
6081     struct TransportValidationPS tvp =
6082       {.purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_CHALLENGE),
6083        .purpose.size = htonl (sizeof (tvp)),
6084        .validity_duration = tvr->validity_duration,
6085        .challenge = tvr->challenge};
6086
6087     if (
6088       GNUNET_OK !=
6089       GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_TRANSPORT_CHALLENGE,
6090                                   &tvp.purpose,
6091                                   &tvr->signature,
6092                                   &cmc->im.sender.public_key))
6093     {
6094       GNUNET_break_op (0);
6095       finish_cmc_handling (cmc);
6096       return;
6097     }
6098   }
6099
6100   /* validity is capped by our willingness to keep track of the
6101      validation entry and the maximum the other peer allows */
6102   vs->valid_until = GNUNET_TIME_relative_to_absolute (
6103     GNUNET_TIME_relative_min (GNUNET_TIME_relative_ntoh (
6104                                 tvr->validity_duration),
6105                               MAX_ADDRESS_VALID_UNTIL));
6106   vs->validated_until =
6107     GNUNET_TIME_absolute_min (vs->valid_until,
6108                               GNUNET_TIME_relative_to_absolute (
6109                                 ADDRESS_VALIDATION_LIFETIME));
6110   vs->validation_rtt = GNUNET_TIME_absolute_get_duration (origin_time);
6111   vs->challenge_backoff = GNUNET_TIME_UNIT_ZERO;
6112   GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
6113                               &vs->challenge,
6114                               sizeof (vs->challenge));
6115   vs->first_challenge_use = GNUNET_TIME_absolute_subtract (
6116     vs->validated_until,
6117     GNUNET_TIME_relative_multiply (vs->validation_rtt,
6118                                    VALIDATION_RTT_BUFFER_FACTOR));
6119   vs->last_challenge_use =
6120     GNUNET_TIME_UNIT_ZERO_ABS; /* challenge was not yet used */
6121   update_next_challenge_time (vs, vs->first_challenge_use);
6122   vs->sc = GNUNET_PEERSTORE_store (peerstore,
6123                                    "transport",
6124                                    &cmc->im.sender,
6125                                    GNUNET_PEERSTORE_TRANSPORT_URLADDRESS_KEY,
6126                                    vs->address,
6127                                    strlen (vs->address) + 1,
6128                                    vs->valid_until,
6129                                    GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
6130                                    &peerstore_store_validation_cb,
6131                                    vs);
6132   finish_cmc_handling (cmc);
6133
6134   /* Finally, we now possibly have a confirmed (!) working queue,
6135      update queue status (if queue still is around) */
6136   q = find_queue (&vs->pid, vs->address);
6137   if (NULL == q)
6138   {
6139     GNUNET_STATISTICS_update (GST_stats,
6140                               "# Queues lost at time of successful validation",
6141                               1,
6142                               GNUNET_NO);
6143     return;
6144   }
6145   q->validated_until = vs->validated_until;
6146   q->rtt = vs->validation_rtt;
6147   n = q->neighbour;
6148   if (GNUNET_NO != n->core_visible)
6149     return; /* nothing changed, we are done here */
6150   n->core_visible = GNUNET_YES;
6151   q->visibility_task = GNUNET_SCHEDULER_add_at (q->validated_until,
6152                                                 &core_queue_visibility_check,
6153                                                 q);
6154   /* Check if _any_ DV route to this neighbour is
6155      currently valid, if so, do NOT tell core anything! */
6156   dv = GNUNET_CONTAINER_multipeermap_get (dv_routes, &n->pid);
6157   if ((NULL != dv) && (GNUNET_YES == dv->core_visible))
6158     return; /* nothing changed, done */
6159   /* We lacked a confirmed connection to the neighbour
6160      before, so tell CORE about it (finally!) */
6161   cores_send_connect_info (&n->pid,
6162                            (NULL != dv)
6163                              ? GNUNET_BANDWIDTH_value_sum (dv->quota_out,
6164                                                            n->quota_out)
6165                              : n->quota_out);
6166 }
6167
6168
6169 /**
6170  * Incoming meessage.  Process the request.
6171  *
6172  * @param im the send message that was received
6173  */
6174 static void
6175 handle_incoming_msg (void *cls,
6176                      const struct GNUNET_TRANSPORT_IncomingMessage *im)
6177 {
6178   struct TransportClient *tc = cls;
6179   struct CommunicatorMessageContext *cmc =
6180     GNUNET_new (struct CommunicatorMessageContext);
6181
6182   cmc->tc = tc;
6183   cmc->im = *im;
6184   demultiplex_with_cmc (cmc, (const struct GNUNET_MessageHeader *) &im[1]);
6185 }
6186
6187
6188 /**
6189  * Given an inbound message @a msg from a communicator @a cmc,
6190  * demultiplex it based on the type calling the right handler.
6191  *
6192  * @param cmc context for demultiplexing
6193  * @param msg message to demultiplex
6194  */
6195 static void
6196 demultiplex_with_cmc (struct CommunicatorMessageContext *cmc,
6197                       const struct GNUNET_MessageHeader *msg)
6198 {
6199   struct GNUNET_MQ_MessageHandler handlers[] =
6200     {GNUNET_MQ_hd_var_size (fragment_box,
6201                             GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT,
6202                             struct TransportFragmentBox,
6203                             &cmc),
6204      GNUNET_MQ_hd_fixed_size (fragment_ack,
6205                               GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT_ACK,
6206                               struct TransportFragmentAckMessage,
6207                               &cmc),
6208      GNUNET_MQ_hd_var_size (reliability_box,
6209                             GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_BOX,
6210                             struct TransportReliabilityBox,
6211                             &cmc),
6212      GNUNET_MQ_hd_fixed_size (reliability_ack,
6213                               GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_ACK,
6214                               struct TransportReliabilityAckMessage,
6215                               &cmc),
6216      GNUNET_MQ_hd_var_size (backchannel_encapsulation,
6217                             GNUNET_MESSAGE_TYPE_TRANSPORT_BACKCHANNEL_ENCAPSULATION,
6218                             struct TransportBackchannelEncapsulationMessage,
6219                             &cmc),
6220      GNUNET_MQ_hd_var_size (dv_learn,
6221                             GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN,
6222                             struct TransportDVLearn,
6223                             &cmc),
6224      GNUNET_MQ_hd_var_size (dv_box,
6225                             GNUNET_MESSAGE_TYPE_TRANSPORT_DV_BOX,
6226                             struct TransportDVBox,
6227                             &cmc),
6228      GNUNET_MQ_hd_fixed_size (
6229        validation_challenge,
6230        GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_CHALLENGE,
6231        struct TransportValidationChallenge,
6232        &cmc),
6233      GNUNET_MQ_hd_fixed_size (
6234        validation_response,
6235        GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_RESPONSE,
6236        struct TransportValidationResponse,
6237        &cmc),
6238      GNUNET_MQ_handler_end ()};
6239   int ret;
6240
6241   ret = GNUNET_MQ_handle_message (handlers, msg);
6242   if (GNUNET_SYSERR == ret)
6243   {
6244     GNUNET_break (0);
6245     GNUNET_SERVICE_client_drop (cmc->tc->client);
6246     GNUNET_free (cmc);
6247     return;
6248   }
6249   if (GNUNET_NO == ret)
6250   {
6251     /* unencapsulated 'raw' message */
6252     handle_raw_message (&cmc, msg);
6253   }
6254 }
6255
6256
6257 /**
6258  * New queue became available.  Check message.
6259  *
6260  * @param cls the client
6261  * @param aqm the send message that was sent
6262  */
6263 static int
6264 check_add_queue_message (void *cls,
6265                          const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
6266 {
6267   struct TransportClient *tc = cls;
6268
6269   if (CT_COMMUNICATOR != tc->type)
6270   {
6271     GNUNET_break (0);
6272     return GNUNET_SYSERR;
6273   }
6274   GNUNET_MQ_check_zero_termination (aqm);
6275   return GNUNET_OK;
6276 }
6277
6278
6279 /**
6280  * Bandwidth tracker informs us that the delay until we should receive
6281  * more has changed.
6282  *
6283  * @param cls a `struct Queue` for which the delay changed
6284  */
6285 static void
6286 tracker_update_in_cb (void *cls)
6287 {
6288   struct Queue *queue = cls;
6289   struct GNUNET_TIME_Relative in_delay;
6290   unsigned int rsize;
6291
6292   rsize = (0 == queue->mtu) ? IN_PACKET_SIZE_WITHOUT_MTU : queue->mtu;
6293   in_delay = GNUNET_BANDWIDTH_tracker_get_delay (&queue->tracker_in, rsize);
6294   // FIXME: how exactly do we do inbound flow control?
6295 }
6296
6297
6298 /**
6299  * If necessary, generates the UUID for a @a pm
6300  *
6301  * @param pm pending message to generate UUID for.
6302  */
6303 static void
6304 set_pending_message_uuid (struct PendingMessage *pm)
6305 {
6306   if (pm->msg_uuid_set)
6307     return;
6308   GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
6309                               &pm->msg_uuid,
6310                               sizeof (pm->msg_uuid));
6311   pm->msg_uuid_set = GNUNET_YES;
6312 }
6313
6314
6315 /**
6316  * Fragment the given @a pm to the given @a mtu.  Adds
6317  * additional fragments to the neighbour as well. If the
6318  * @a mtu is too small, generates and error for the @a pm
6319  * and returns NULL.
6320  *
6321  * @param pm pending message to fragment for transmission
6322  * @param mtu MTU to apply
6323  * @return new message to transmit
6324  */
6325 static struct PendingMessage *
6326 fragment_message (struct PendingMessage *pm, uint16_t mtu)
6327 {
6328   struct PendingMessage *ff;
6329
6330   set_pending_message_uuid (pm);
6331
6332   /* This invariant is established in #handle_add_queue_message() */
6333   GNUNET_assert (mtu > sizeof (struct TransportFragmentBox));
6334
6335   /* select fragment for transmission, descending the tree if it has
6336      been expanded until we are at a leaf or at a fragment that is small enough
6337    */
6338   ff = pm;
6339   while (((ff->bytes_msg > mtu) || (pm == ff)) &&
6340          (ff->frag_off == ff->bytes_msg) && (NULL != ff->head_frag))
6341   {
6342     ff = ff->head_frag; /* descent into fragmented fragments */
6343   }
6344
6345   if (((ff->bytes_msg > mtu) || (pm == ff)) && (pm->frag_off < pm->bytes_msg))
6346   {
6347     /* Did not yet calculate all fragments, calculate next fragment */
6348     struct PendingMessage *frag;
6349     struct TransportFragmentBox tfb;
6350     const char *orig;
6351     char *msg;
6352     uint16_t fragmax;
6353     uint16_t fragsize;
6354     uint16_t msize;
6355     uint16_t xoff = 0;
6356
6357     orig = (const char *) &ff[1];
6358     msize = ff->bytes_msg;
6359     if (pm != ff)
6360     {
6361       const struct TransportFragmentBox *tfbo;
6362
6363       tfbo = (const struct TransportFragmentBox *) orig;
6364       orig += sizeof (struct TransportFragmentBox);
6365       msize -= sizeof (struct TransportFragmentBox);
6366       xoff = ntohs (tfbo->frag_off);
6367     }
6368     fragmax = mtu - sizeof (struct TransportFragmentBox);
6369     fragsize = GNUNET_MIN (msize - ff->frag_off, fragmax);
6370     frag = GNUNET_malloc (sizeof (struct PendingMessage) +
6371                           sizeof (struct TransportFragmentBox) + fragsize);
6372     frag->target = pm->target;
6373     frag->frag_parent = ff;
6374     frag->timeout = pm->timeout;
6375     frag->bytes_msg = sizeof (struct TransportFragmentBox) + fragsize;
6376     frag->pmt = PMT_FRAGMENT_BOX;
6377     msg = (char *) &frag[1];
6378     tfb.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_FRAGMENT);
6379     tfb.header.size = htons (sizeof (struct TransportFragmentBox) + fragsize);
6380     tfb.frag_uuid = htonl (pm->frag_uuidgen++);
6381     tfb.msg_uuid = pm->msg_uuid;
6382     tfb.frag_off = htons (ff->frag_off + xoff);
6383     tfb.msg_size = htons (pm->bytes_msg);
6384     memcpy (msg, &tfb, sizeof (tfb));
6385     memcpy (&msg[sizeof (tfb)], &orig[ff->frag_off], fragsize);
6386     GNUNET_CONTAINER_MDLL_insert (frag, ff->head_frag, ff->tail_frag, frag);
6387     ff->frag_off += fragsize;
6388     ff = frag;
6389   }
6390
6391   /* Move head to the tail and return it */
6392   GNUNET_CONTAINER_MDLL_remove (frag,
6393                                 ff->frag_parent->head_frag,
6394                                 ff->frag_parent->tail_frag,
6395                                 ff);
6396   GNUNET_CONTAINER_MDLL_insert_tail (frag,
6397                                      ff->frag_parent->head_frag,
6398                                      ff->frag_parent->tail_frag,
6399                                      ff);
6400   return ff;
6401 }
6402
6403
6404 /**
6405  * Reliability-box the given @a pm. On error (can there be any), NULL
6406  * may be returned, otherwise the "replacement" for @a pm (which
6407  * should then be added to the respective neighbour's queue instead of
6408  * @a pm).  If the @a pm is already fragmented or reliability boxed,
6409  * or itself an ACK, this function simply returns @a pm.
6410  *
6411  * @param pm pending message to box for transmission over unreliabile queue
6412  * @return new message to transmit
6413  */
6414 static struct PendingMessage *
6415 reliability_box_message (struct PendingMessage *pm)
6416 {
6417   struct TransportReliabilityBox rbox;
6418   struct PendingMessage *bpm;
6419   char *msg;
6420
6421   if (PMT_CORE != pm->pmt)
6422     return pm; /* already fragmented or reliability boxed, or control message:
6423                   do nothing */
6424   if (NULL != pm->bpm)
6425     return pm->bpm; /* already computed earlier: do nothing */
6426   GNUNET_assert (NULL == pm->head_frag);
6427   if (pm->bytes_msg + sizeof (rbox) > UINT16_MAX)
6428   {
6429     /* failed hard */
6430     GNUNET_break (0);
6431     client_send_response (pm, GNUNET_NO, 0);
6432     return NULL;
6433   }
6434   bpm = GNUNET_malloc (sizeof (struct PendingMessage) + sizeof (rbox) +
6435                        pm->bytes_msg);
6436   bpm->target = pm->target;
6437   bpm->frag_parent = pm;
6438   GNUNET_CONTAINER_MDLL_insert (frag, pm->head_frag, pm->tail_frag, bpm);
6439   bpm->timeout = pm->timeout;
6440   bpm->pmt = PMT_RELIABILITY_BOX;
6441   bpm->bytes_msg = pm->bytes_msg + sizeof (rbox);
6442   set_pending_message_uuid (bpm);
6443   rbox.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_RELIABILITY_BOX);
6444   rbox.header.size = htons (sizeof (rbox) + pm->bytes_msg);
6445   rbox.ack_countdown = htonl (0); // FIXME: implement ACK countdown support
6446   rbox.msg_uuid = pm->msg_uuid;
6447   msg = (char *) &bpm[1];
6448   memcpy (msg, &rbox, sizeof (rbox));
6449   memcpy (&msg[sizeof (rbox)], &pm[1], pm->bytes_msg);
6450   pm->bpm = bpm;
6451   return bpm;
6452 }
6453
6454
6455 /**
6456  * Change the value of the `next_attempt` field of @a pm
6457  * to @a next_attempt and re-order @a pm in the transmission
6458  * list as required by the new timestmap.
6459  *
6460  * @param pm a pending message to update
6461  * @param next_attempt timestamp to use
6462  */
6463 static void
6464 update_pm_next_attempt (struct PendingMessage *pm,
6465                         struct GNUNET_TIME_Absolute next_attempt)
6466 {
6467   struct Neighbour *neighbour = pm->target;
6468
6469   pm->next_attempt = next_attempt;
6470   if (NULL == pm->frag_parent)
6471   {
6472     struct PendingMessage *pos;
6473
6474     /* re-insert sort in neighbour list */
6475     GNUNET_CONTAINER_MDLL_remove (neighbour,
6476                                   neighbour->pending_msg_head,
6477                                   neighbour->pending_msg_tail,
6478                                   pm);
6479     pos = neighbour->pending_msg_tail;
6480     while ((NULL != pos) &&
6481            (next_attempt.abs_value_us > pos->next_attempt.abs_value_us))
6482       pos = pos->prev_neighbour;
6483     GNUNET_CONTAINER_MDLL_insert_after (neighbour,
6484                                         neighbour->pending_msg_head,
6485                                         neighbour->pending_msg_tail,
6486                                         pos,
6487                                         pm);
6488   }
6489   else
6490   {
6491     /* re-insert sort in fragment list */
6492     struct PendingMessage *fp = pm->frag_parent;
6493     struct PendingMessage *pos;
6494
6495     GNUNET_CONTAINER_MDLL_remove (frag, fp->head_frag, fp->tail_frag, pm);
6496     pos = fp->tail_frag;
6497     while ((NULL != pos) &&
6498            (next_attempt.abs_value_us > pos->next_attempt.abs_value_us))
6499       pos = pos->prev_frag;
6500     GNUNET_CONTAINER_MDLL_insert_after (frag,
6501                                         fp->head_frag,
6502                                         fp->tail_frag,
6503                                         pos,
6504                                         pm);
6505   }
6506 }
6507
6508
6509 /**
6510  * We believe we are ready to transmit a message on a queue. Double-checks
6511  * with the queue's "tracker_out" and then gives the message to the
6512  * communicator for transmission (updating the tracker, and re-scheduling
6513  * itself if applicable).
6514  *
6515  * @param cls the `struct Queue` to process transmissions for
6516  */
6517 static void
6518 transmit_on_queue (void *cls)
6519 {
6520   struct Queue *queue = cls;
6521   struct Neighbour *n = queue->neighbour;
6522   struct PendingMessage *pm;
6523   struct PendingMessage *s;
6524   uint32_t overhead;
6525
6526   queue->transmit_task = NULL;
6527   if (NULL == (pm = n->pending_msg_head))
6528   {
6529     /* no message pending, nothing to do here! */
6530     return;
6531   }
6532   if (NULL != pm->qe)
6533   {
6534     /* message still pending with communciator!
6535        LOGGING-FIXME: Use stats? logging? Should this not be rare? */
6536     return;
6537   }
6538   schedule_transmit_on_queue (queue, GNUNET_YES);
6539   if (NULL != queue->transmit_task)
6540     return; /* do it later */
6541   overhead = 0;
6542   if (GNUNET_TRANSPORT_CC_RELIABLE != queue->tc->details.communicator.cc)
6543     overhead += sizeof (struct TransportReliabilityBox);
6544   s = pm;
6545   if ( ( (0 != queue->mtu) &&
6546          (pm->bytes_msg + overhead > queue->mtu) ) ||
6547        (pm->bytes_msg > UINT16_MAX - sizeof (struct GNUNET_TRANSPORT_SendMessageTo)) ||
6548        (NULL != pm->head_frag /* fragments already exist, should
6549                                  respect that even if MTU is 0 for
6550                                  this queue */) )
6551     s = fragment_message (s,
6552                           (0 == queue->mtu)
6553                             ? UINT16_MAX -
6554                                 sizeof (struct GNUNET_TRANSPORT_SendMessageTo)
6555                             : queue->mtu);
6556   if (NULL == s)
6557   {
6558     /* Fragmentation failed, try next message... */
6559     schedule_transmit_on_queue (queue, GNUNET_NO);
6560     return;
6561   }
6562   if (GNUNET_TRANSPORT_CC_RELIABLE != queue->tc->details.communicator.cc)
6563     s = reliability_box_message (s);
6564   if (NULL == s)
6565   {
6566     /* Reliability boxing failed, try next message... */
6567     schedule_transmit_on_queue (queue, GNUNET_NO);
6568     return;
6569   }
6570
6571   /* Pass 's' for transission to the communicator */
6572   queue_send_msg (queue, s, &s[1], s->bytes_msg);
6573   // FIXME: do something similar to the logic below
6574   // in defragmentation / reliability ACK handling!
6575
6576   /* Check if this transmission somehow conclusively finished handing 'pm'
6577      even without any explicit ACKs */
6578   if ((PMT_CORE == s->pmt) &&
6579       (GNUNET_TRANSPORT_CC_RELIABLE == queue->tc->details.communicator.cc))
6580   {
6581     /* Full message sent, and over reliabile channel */
6582     client_send_response (pm, GNUNET_YES, pm->bytes_msg);
6583   }
6584   else if ((GNUNET_TRANSPORT_CC_RELIABLE ==
6585             queue->tc->details.communicator.cc) &&
6586            (PMT_FRAGMENT_BOX == s->pmt))
6587   {
6588     struct PendingMessage *pos;
6589
6590     /* Fragment sent over reliabile channel */
6591     free_fragment_tree (s);
6592     pos = s->frag_parent;
6593     GNUNET_CONTAINER_MDLL_remove (frag, pos->head_frag, pos->tail_frag, s);
6594     GNUNET_free (s);
6595     /* check if subtree is done */
6596     while ((NULL == pos->head_frag) && (pos->frag_off == pos->bytes_msg) &&
6597            (pos != pm))
6598     {
6599       s = pos;
6600       pos = s->frag_parent;
6601       GNUNET_CONTAINER_MDLL_remove (frag, pos->head_frag, pos->tail_frag, s);
6602       GNUNET_free (s);
6603     }
6604
6605     /* Was this the last applicable fragmment? */
6606     if ((NULL == pm->head_frag) && (pm->frag_off == pm->bytes_msg))
6607       client_send_response (
6608         pm,
6609         GNUNET_YES,
6610         pm->bytes_msg /* FIXME: calculate and add overheads! */);
6611   }
6612   else if (PMT_CORE != pm->pmt)
6613   {
6614     /* This was an acknowledgement of some type, always free */
6615     free_pending_message (pm);
6616   }
6617   else
6618   {
6619     /* Message not finished, waiting for acknowledgement.
6620        Update time by which we might retransmit 's' based on queue
6621        characteristics (i.e. RTT); it takes one RTT for the message to
6622        arrive and the ACK to come back in the best case; but the other
6623        side is allowed to delay ACKs by 2 RTTs, so we use 4 RTT before
6624        retransmitting.  Note that in the future this heuristic should
6625        likely be improved further (measure RTT stability, consider
6626        message urgency and size when delaying ACKs, etc.) */
6627     update_pm_next_attempt (s,
6628                             GNUNET_TIME_relative_to_absolute (
6629                               GNUNET_TIME_relative_multiply (queue->rtt, 4)));
6630   }
6631
6632   /* finally, re-schedule queue transmission task itself */
6633   schedule_transmit_on_queue (queue, GNUNET_NO);
6634 }
6635
6636
6637 /**
6638  * Bandwidth tracker informs us that the delay until we
6639  * can transmit again changed.
6640  *
6641  * @param cls a `struct Queue` for which the delay changed
6642  */
6643 static void
6644 tracker_update_out_cb (void *cls)
6645 {
6646   struct Queue *queue = cls;
6647   struct Neighbour *n = queue->neighbour;
6648
6649   if (NULL == n->pending_msg_head)
6650   {
6651     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
6652                 "Bandwidth allocation updated for empty transmission queue `%s'\n",
6653                 queue->address);
6654     return; /* no message pending, nothing to do here! */
6655   }
6656   GNUNET_SCHEDULER_cancel (queue->transmit_task);
6657   queue->transmit_task = NULL;
6658   schedule_transmit_on_queue (queue, GNUNET_NO);
6659 }
6660
6661
6662 /**
6663  * Bandwidth tracker informs us that excessive outbound bandwidth was
6664  * allocated which is not being used.
6665  *
6666  * @param cls a `struct Queue` for which the excess was noted
6667  */
6668 static void
6669 tracker_excess_out_cb (void *cls)
6670 {
6671   (void) cls;
6672
6673   /* FIXME: trigger excess bandwidth report to core? Right now,
6674      this is done internally within transport_api2_core already,
6675      but we probably want to change the logic and trigger it
6676      from here via a message instead! */
6677   /* TODO: maybe inform someone at this point? */
6678   GNUNET_STATISTICS_update (GST_stats,
6679                             "# Excess outbound bandwidth reported",
6680                             1,
6681                             GNUNET_NO);
6682 }
6683
6684
6685 /**
6686  * Bandwidth tracker informs us that excessive inbound bandwidth was allocated
6687  * which is not being used.
6688  *
6689  * @param cls a `struct Queue` for which the excess was noted
6690  */
6691 static void
6692 tracker_excess_in_cb (void *cls)
6693 {
6694   (void) cls;
6695
6696   /* TODO: maybe inform somone at this point? */
6697   GNUNET_STATISTICS_update (GST_stats,
6698                             "# Excess inbound bandwidth reported",
6699                             1,
6700                             GNUNET_NO);
6701 }
6702
6703
6704 /**
6705  * Queue to a peer went down.  Process the request.
6706  *
6707  * @param cls the client
6708  * @param dqm the send message that was sent
6709  */
6710 static void
6711 handle_del_queue_message (void *cls,
6712                           const struct GNUNET_TRANSPORT_DelQueueMessage *dqm)
6713 {
6714   struct TransportClient *tc = cls;
6715
6716   if (CT_COMMUNICATOR != tc->type)
6717   {
6718     GNUNET_break (0);
6719     GNUNET_SERVICE_client_drop (tc->client);
6720     return;
6721   }
6722   for (struct Queue *queue = tc->details.communicator.queue_head; NULL != queue;
6723        queue = queue->next_client)
6724   {
6725     struct Neighbour *neighbour = queue->neighbour;
6726
6727     if ((dqm->qid != queue->qid) ||
6728         (0 != GNUNET_memcmp (&dqm->receiver, &neighbour->pid)))
6729       continue;
6730     free_queue (queue);
6731     GNUNET_SERVICE_client_continue (tc->client);
6732     return;
6733   }
6734   GNUNET_break (0);
6735   GNUNET_SERVICE_client_drop (tc->client);
6736 }
6737
6738
6739 /**
6740  * Message was transmitted.  Process the request.
6741  *
6742  * @param cls the client
6743  * @param sma the send message that was sent
6744  */
6745 static void
6746 handle_send_message_ack (void *cls,
6747                          const struct GNUNET_TRANSPORT_SendMessageToAck *sma)
6748 {
6749   struct TransportClient *tc = cls;
6750   struct QueueEntry *qe;
6751   struct PendingMessage *pm;
6752
6753   if (CT_COMMUNICATOR != tc->type)
6754   {
6755     GNUNET_break (0);
6756     GNUNET_SERVICE_client_drop (tc->client);
6757     return;
6758   }
6759
6760   /* find our queue entry matching the ACK */
6761   qe = NULL;
6762   for (struct Queue *queue = tc->details.communicator.queue_head; NULL != queue;
6763        queue = queue->next_client)
6764   {
6765     if (0 != GNUNET_memcmp (&queue->neighbour->pid, &sma->receiver))
6766       continue;
6767     for (struct QueueEntry *qep = queue->queue_head; NULL != qep;
6768          qep = qep->next)
6769     {
6770       if (qep->mid != sma->mid)
6771         continue;
6772       qe = qep;
6773       break;
6774     }
6775     break;
6776   }
6777   if (NULL == qe)
6778   {
6779     /* this should never happen */
6780     GNUNET_break (0);
6781     GNUNET_SERVICE_client_drop (tc->client);
6782     return;
6783   }
6784   GNUNET_CONTAINER_DLL_remove (qe->queue->queue_head,
6785                                qe->queue->queue_tail,
6786                                qe);
6787   qe->queue->queue_length--;
6788   tc->details.communicator.total_queue_length--;
6789   GNUNET_SERVICE_client_continue (tc->client);
6790
6791   /* if applicable, resume transmissions that waited on ACK */
6792   if (COMMUNICATOR_TOTAL_QUEUE_LIMIT - 1 ==
6793       tc->details.communicator.total_queue_length)
6794   {
6795     /* Communicator dropped below threshold, resume all queues
6796        incident with this client! */
6797     GNUNET_STATISTICS_update (
6798       GST_stats,
6799       "# Transmission throttled due to communicator queue limit",
6800       -1,
6801       GNUNET_NO);
6802     for (struct Queue *queue = tc->details.communicator.queue_head;
6803          NULL != queue;
6804          queue = queue->next_client)
6805       schedule_transmit_on_queue (queue, GNUNET_NO);
6806   }
6807   else if (QUEUE_LENGTH_LIMIT - 1 == qe->queue->queue_length)
6808   {
6809     /* queue dropped below threshold; only resume this one queue */
6810     GNUNET_STATISTICS_update (GST_stats,
6811                               "# Transmission throttled due to queue queue limit",
6812                               -1,
6813                               GNUNET_NO);
6814     schedule_transmit_on_queue (qe->queue, GNUNET_NO);
6815   }
6816
6817   if (NULL != (pm = qe->pm))
6818   {
6819     struct Neighbour *n;
6820
6821     GNUNET_assert (qe == pm->qe);
6822     pm->qe = NULL;
6823     /* If waiting for this communicator may have blocked transmission
6824        of pm on other queues for this neighbour, force schedule
6825        transmit on queue for queues of the neighbour */
6826     n = pm->target;
6827     if (n->pending_msg_head == pm)
6828     {
6829       for (struct Queue *queue = n->queue_head; NULL != queue;
6830            queue = queue->next_neighbour)
6831         schedule_transmit_on_queue (queue, GNUNET_NO);
6832     }
6833     if (GNUNET_OK != ntohl (sma->status))
6834     {
6835       GNUNET_log (
6836         GNUNET_ERROR_TYPE_INFO,
6837         "Queue failed in transmission, will try retransmission immediately\n");
6838       update_pm_next_attempt (pm, GNUNET_TIME_UNIT_ZERO_ABS);
6839     }
6840   }
6841   GNUNET_free (qe);
6842 }
6843
6844
6845 /**
6846  * Iterator telling new MONITOR client about all existing
6847  * queues to peers.
6848  *
6849  * @param cls the new `struct TransportClient`
6850  * @param pid a connected peer
6851  * @param value the `struct Neighbour` with more information
6852  * @return #GNUNET_OK (continue to iterate)
6853  */
6854 static int
6855 notify_client_queues (void *cls,
6856                       const struct GNUNET_PeerIdentity *pid,
6857                       void *value)
6858 {
6859   struct TransportClient *tc = cls;
6860   struct Neighbour *neighbour = value;
6861
6862   GNUNET_assert (CT_MONITOR == tc->type);
6863   for (struct Queue *q = neighbour->queue_head; NULL != q;
6864        q = q->next_neighbour)
6865   {
6866     struct MonitorEvent me = {.rtt = q->rtt,
6867                               .cs = q->cs,
6868                               .num_msg_pending = q->num_msg_pending,
6869                               .num_bytes_pending = q->num_bytes_pending};
6870
6871     notify_monitor (tc, pid, q->address, q->nt, &me);
6872   }
6873   return GNUNET_OK;
6874 }
6875
6876
6877 /**
6878  * Initialize a monitor client.
6879  *
6880  * @param cls the client
6881  * @param start the start message that was sent
6882  */
6883 static void
6884 handle_monitor_start (void *cls,
6885                       const struct GNUNET_TRANSPORT_MonitorStart *start)
6886 {
6887   struct TransportClient *tc = cls;
6888
6889   if (CT_NONE != tc->type)
6890   {
6891     GNUNET_break (0);
6892     GNUNET_SERVICE_client_drop (tc->client);
6893     return;
6894   }
6895   tc->type = CT_MONITOR;
6896   tc->details.monitor.peer = start->peer;
6897   tc->details.monitor.one_shot = ntohl (start->one_shot);
6898   GNUNET_CONTAINER_multipeermap_iterate (neighbours, &notify_client_queues, tc);
6899   GNUNET_SERVICE_client_mark_monitor (tc->client);
6900   GNUNET_SERVICE_client_continue (tc->client);
6901 }
6902
6903
6904 /**
6905  * Find transport client providing communication service
6906  * for the protocol @a prefix.
6907  *
6908  * @param prefix communicator name
6909  * @return NULL if no such transport client is available
6910  */
6911 static struct TransportClient *
6912 lookup_communicator (const char *prefix)
6913 {
6914   for (struct TransportClient *tc = clients_head; NULL != tc; tc = tc->next)
6915   {
6916     if (CT_COMMUNICATOR != tc->type)
6917       continue;
6918     if (0 == strcmp (prefix, tc->details.communicator.address_prefix))
6919       return tc;
6920   }
6921   GNUNET_log (
6922     GNUNET_ERROR_TYPE_WARNING,
6923     "Somone suggested use of communicator for `%s', but we do not have such a communicator!\n",
6924     prefix);
6925   return NULL;
6926 }
6927
6928
6929 /**
6930  * Signature of a function called with a communicator @a address of a peer
6931  * @a pid that an application wants us to connect to.
6932  *
6933  * @param pid target peer
6934  * @param address the address to try
6935  */
6936 static void
6937 suggest_to_connect (const struct GNUNET_PeerIdentity *pid, const char *address)
6938 {
6939   static uint32_t idgen;
6940   struct TransportClient *tc;
6941   char *prefix;
6942   struct GNUNET_TRANSPORT_CreateQueue *cqm;
6943   struct GNUNET_MQ_Envelope *env;
6944   size_t alen;
6945
6946   prefix = GNUNET_HELLO_address_to_prefix (address);
6947   if (NULL == prefix)
6948   {
6949     GNUNET_break (0); /* We got an invalid address!? */
6950     return;
6951   }
6952   tc = lookup_communicator (prefix);
6953   if (NULL == tc)
6954   {
6955     GNUNET_STATISTICS_update (GST_stats,
6956                               "# Suggestions ignored due to missing communicator",
6957                               1,
6958                               GNUNET_NO);
6959     return;
6960   }
6961   /* forward suggestion for queue creation to communicator */
6962   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
6963               "Request #%u for `%s' communicator to create queue to `%s'\n",
6964               (unsigned int) idgen,
6965               prefix,
6966               address);
6967   alen = strlen (address) + 1;
6968   env =
6969     GNUNET_MQ_msg_extra (cqm, alen, GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE);
6970   cqm->request_id = htonl (idgen++);
6971   cqm->receiver = *pid;
6972   memcpy (&cqm[1], address, alen);
6973   GNUNET_MQ_send (tc->mq, env);
6974 }
6975
6976
6977 /**
6978  * The queue @a q (which matches the peer and address in @a vs) is
6979  * ready for queueing. We should now queue the validation request.
6980  *
6981  * @param q queue to send on
6982  * @param vs state to derive validation challenge from
6983  */
6984 static void
6985 validation_transmit_on_queue (struct Queue *q, struct ValidationState *vs)
6986 {
6987   struct TransportValidationChallenge tvc;
6988
6989   vs->last_challenge_use = GNUNET_TIME_absolute_get ();
6990   tvc.header.type =
6991     htons (GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_VALIDATION_CHALLENGE);
6992   tvc.header.size = htons (sizeof (tvc));
6993   tvc.reserved = htonl (0);
6994   tvc.challenge = vs->challenge;
6995   tvc.sender_time = GNUNET_TIME_absolute_hton (vs->last_challenge_use);
6996   queue_send_msg (q, NULL, &tvc, sizeof (tvc));
6997 }
6998
6999
7000 /**
7001  * Task run periodically to validate some address based on #validation_heap.
7002  *
7003  * @param cls NULL
7004  */
7005 static void
7006 validation_start_cb (void *cls)
7007 {
7008   struct ValidationState *vs;
7009   struct Queue *q;
7010
7011   (void) cls;
7012   validation_task = NULL;
7013   vs = GNUNET_CONTAINER_heap_peek (validation_heap);
7014   /* drop validations past their expiration */
7015   while (
7016     (NULL != vs) &&
7017     (0 == GNUNET_TIME_absolute_get_remaining (vs->valid_until).rel_value_us))
7018   {
7019     free_validation_state (vs);
7020     vs = GNUNET_CONTAINER_heap_peek (validation_heap);
7021   }
7022   if (NULL == vs)
7023     return; /* woopsie, no more addresses known, should only
7024                happen if we're really a lonely peer */
7025   q = find_queue (&vs->pid, vs->address);
7026   if (NULL == q)
7027   {
7028     vs->awaiting_queue = GNUNET_YES;
7029     suggest_to_connect (&vs->pid, vs->address);
7030   }
7031   else
7032     validation_transmit_on_queue (q, vs);
7033   /* Finally, reschedule next attempt */
7034   vs->challenge_backoff =
7035     GNUNET_TIME_randomized_backoff (vs->challenge_backoff,
7036                                     MAX_VALIDATION_CHALLENGE_FREQ);
7037   update_next_challenge_time (vs,
7038                               GNUNET_TIME_relative_to_absolute (
7039                                 vs->challenge_backoff));
7040 }
7041
7042
7043 /**
7044  * Closure for #check_connection_quality.
7045  */
7046 struct QueueQualityContext
7047 {
7048   /**
7049    * Set to the @e k'th queue encountered.
7050    */
7051   struct Queue *q;
7052
7053   /**
7054    * Set to the number of quality queues encountered.
7055    */
7056   unsigned int quality_count;
7057
7058   /**
7059    * Set to the total number of queues encountered.
7060    */
7061   unsigned int num_queues;
7062
7063   /**
7064    * Decremented for each queue, for selection of the
7065    * k-th queue in @e q.
7066    */
7067   unsigned int k;
7068 };
7069
7070
7071 /**
7072  * Check whether any queue to the given neighbour is
7073  * of a good "quality" and if so, increment the counter.
7074  * Also counts the total number of queues, and returns
7075  * the k-th queue found.
7076  *
7077  * @param cls a `struct QueueQualityContext *` with counters
7078  * @param pid peer this is about
7079  * @param value a `struct Neighbour`
7080  * @return #GNUNET_OK (continue to iterate)
7081  */
7082 static int
7083 check_connection_quality (void *cls,
7084                           const struct GNUNET_PeerIdentity *pid,
7085                           void *value)
7086 {
7087   struct QueueQualityContext *ctx = cls;
7088   struct Neighbour *n = value;
7089   int do_inc;
7090
7091   (void) pid;
7092   do_inc = GNUNET_NO;
7093   for (struct Queue *q = n->queue_head; NULL != q; q = q->next_neighbour)
7094   {
7095     ctx->num_queues++;
7096     if (0 == ctx->k--)
7097       ctx->q = q;
7098     /* OPTIMIZE-FIXME: in the future, add reliability / goodput
7099        statistics and consider those as well here? */
7100     if (q->rtt.rel_value_us < DV_QUALITY_RTT_THRESHOLD.rel_value_us)
7101       do_inc = GNUNET_YES;
7102   }
7103   if (GNUNET_YES == do_inc)
7104     ctx->quality_count++;
7105   return GNUNET_OK;
7106 }
7107
7108
7109 /**
7110  * Task run when we CONSIDER initiating a DV learn
7111  * process. We first check that sending out a message is
7112  * even possible (queues exist), then that it is desirable
7113  * (if not, reschedule the task for later), and finally
7114  * we may then begin the job.  If there are too many
7115  * entries in the #dvlearn_map, we purge the oldest entry
7116  * using #lle_tail.
7117  *
7118  * @param cls NULL
7119  */
7120 static void
7121 start_dv_learn (void *cls)
7122 {
7123   struct LearnLaunchEntry *lle;
7124   struct QueueQualityContext qqc;
7125   struct TransportDVLearn dvl;
7126
7127   (void) cls;
7128   dvlearn_task = NULL;
7129   if (0 == GNUNET_CONTAINER_multipeermap_size (neighbours))
7130     return; /* lost all connectivity, cannot do learning */
7131   qqc.quality_count = 0;
7132   qqc.num_queues = 0;
7133   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
7134                                          &check_connection_quality,
7135                                          &qqc);
7136   if (qqc.quality_count > DV_LEARN_QUALITY_THRESHOLD)
7137   {
7138     struct GNUNET_TIME_Relative delay;
7139     unsigned int factor;
7140
7141     /* scale our retries by how far we are above the threshold */
7142     factor = qqc.quality_count / DV_LEARN_QUALITY_THRESHOLD;
7143     delay = GNUNET_TIME_relative_multiply (DV_LEARN_BASE_FREQUENCY, factor);
7144     dvlearn_task = GNUNET_SCHEDULER_add_delayed (delay, &start_dv_learn, NULL);
7145     return;
7146   }
7147   /* remove old entries in #dvlearn_map if it has grown too big */
7148   while (MAX_DV_LEARN_PENDING >=
7149          GNUNET_CONTAINER_multishortmap_size (dvlearn_map))
7150   {
7151     lle = lle_tail;
7152     GNUNET_assert (GNUNET_YES ==
7153                    GNUNET_CONTAINER_multishortmap_remove (dvlearn_map,
7154                                                           &lle->challenge,
7155                                                           lle));
7156     GNUNET_CONTAINER_DLL_remove (lle_head, lle_tail, lle);
7157     GNUNET_free (lle);
7158   }
7159   /* setup data structure for learning */
7160   lle = GNUNET_new (struct LearnLaunchEntry);
7161   GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
7162                               &lle->challenge,
7163                               sizeof (lle->challenge));
7164   GNUNET_CONTAINER_DLL_insert (lle_head, lle_tail, lle);
7165   GNUNET_break (GNUNET_YES ==
7166                 GNUNET_CONTAINER_multishortmap_put (
7167                   dvlearn_map,
7168                   &lle->challenge,
7169                   lle,
7170                   GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
7171   dvl.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_DV_LEARN);
7172   dvl.header.size = htons (sizeof (dvl));
7173   dvl.num_hops = htons (0);
7174   dvl.bidirectional = htons (0);
7175   dvl.non_network_delay = GNUNET_TIME_relative_hton (GNUNET_TIME_UNIT_ZERO);
7176   {
7177     struct DvInitPS dvip = {.purpose.purpose = htonl (
7178                               GNUNET_SIGNATURE_PURPOSE_TRANSPORT_DV_INITIATOR),
7179                             .purpose.size = htonl (sizeof (dvip)),
7180                             .challenge = lle->challenge};
7181
7182     GNUNET_assert (GNUNET_OK == GNUNET_CRYPTO_eddsa_sign (GST_my_private_key,
7183                                                           &dvip.purpose,
7184                                                           &dvl.init_sig));
7185   }
7186   dvl.initiator = GST_my_identity;
7187   dvl.challenge = lle->challenge;
7188
7189   qqc.quality_count = 0;
7190   qqc.k = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, qqc.num_queues);
7191   qqc.num_queues = 0;
7192   qqc.q = NULL;
7193   GNUNET_CONTAINER_multipeermap_iterate (neighbours,
7194                                          &check_connection_quality,
7195                                          &qqc);
7196   GNUNET_assert (NULL != qqc.q);
7197
7198   /* Do this as close to transmission time as possible! */
7199   lle->launch_time = GNUNET_TIME_absolute_get ();
7200
7201   queue_send_msg (qqc.q, NULL, &dvl, sizeof (dvl));
7202   /* reschedule this job, randomizing the time it runs (but no
7203      actual backoff!) */
7204   dvlearn_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_randomize (
7205                                                  DV_LEARN_BASE_FREQUENCY),
7206                                                &start_dv_learn,
7207                                                NULL);
7208 }
7209
7210
7211 /**
7212  * A new queue has been created, check if any address validation
7213  * requests have been waiting for it.
7214  *
7215  * @param cls a `struct Queue`
7216  * @param pid peer concerned (unused)
7217  * @param value a `struct ValidationState`
7218  * @return #GNUNET_NO if a match was found and we can stop looking
7219  */
7220 static int
7221 check_validation_request_pending (void *cls,
7222                                   const struct GNUNET_PeerIdentity *pid,
7223                                   void *value)
7224 {
7225   struct Queue *q = cls;
7226   struct ValidationState *vs = value;
7227
7228   (void) pid;
7229   if ((GNUNET_YES == vs->awaiting_queue) &&
7230       (0 == strcmp (vs->address, q->address)))
7231   {
7232     vs->awaiting_queue = GNUNET_NO;
7233     validation_transmit_on_queue (q, vs);
7234     return GNUNET_NO;
7235   }
7236   return GNUNET_OK;
7237 }
7238
7239
7240 /**
7241  * New queue became available.  Process the request.
7242  *
7243  * @param cls the client
7244  * @param aqm the send message that was sent
7245  */
7246 static void
7247 handle_add_queue_message (void *cls,
7248                           const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
7249 {
7250   struct TransportClient *tc = cls;
7251   struct Queue *queue;
7252   struct Neighbour *neighbour;
7253   const char *addr;
7254   uint16_t addr_len;
7255
7256   if (ntohl (aqm->mtu) <= sizeof (struct TransportFragmentBox))
7257   {
7258     /* MTU so small as to be useless for transmissions,
7259        required for #fragment_message()! */
7260     GNUNET_break_op (0);
7261     GNUNET_SERVICE_client_drop (tc->client);
7262     return;
7263   }
7264   neighbour = lookup_neighbour (&aqm->receiver);
7265   if (NULL == neighbour)
7266   {
7267     neighbour = GNUNET_new (struct Neighbour);
7268     neighbour->earliest_timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
7269     neighbour->pid = aqm->receiver;
7270     GNUNET_assert (GNUNET_OK ==
7271                    GNUNET_CONTAINER_multipeermap_put (
7272                      neighbours,
7273                      &neighbour->pid,
7274                      neighbour,
7275                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
7276   }
7277   addr_len = ntohs (aqm->header.size) - sizeof (*aqm);
7278   addr = (const char *) &aqm[1];
7279
7280   queue = GNUNET_malloc (sizeof (struct Queue) + addr_len);
7281   queue->tc = tc;
7282   queue->address = (const char *) &queue[1];
7283   queue->rtt = GNUNET_TIME_UNIT_FOREVER_REL;
7284   queue->qid = aqm->qid;
7285   queue->mtu = ntohl (aqm->mtu);
7286   queue->nt = (enum GNUNET_NetworkType) ntohl (aqm->nt);
7287   queue->cs = (enum GNUNET_TRANSPORT_ConnectionStatus) ntohl (aqm->cs);
7288   queue->neighbour = neighbour;
7289   GNUNET_BANDWIDTH_tracker_init2 (&queue->tracker_in,
7290                                   &tracker_update_in_cb,
7291                                   queue,
7292                                   GNUNET_BANDWIDTH_ZERO,
7293                                   GNUNET_CONSTANTS_MAX_BANDWIDTH_CARRY_S,
7294                                   &tracker_excess_in_cb,
7295                                   queue);
7296   GNUNET_BANDWIDTH_tracker_init2 (&queue->tracker_out,
7297                                   &tracker_update_out_cb,
7298                                   queue,
7299                                   GNUNET_BANDWIDTH_ZERO,
7300                                   GNUNET_CONSTANTS_MAX_BANDWIDTH_CARRY_S,
7301                                   &tracker_excess_out_cb,
7302                                   queue);
7303   memcpy (&queue[1], addr, addr_len);
7304   /* notify monitors about new queue */
7305   {
7306     struct MonitorEvent me = {.rtt = queue->rtt, .cs = queue->cs};
7307
7308     notify_monitors (&neighbour->pid, queue->address, queue->nt, &me);
7309   }
7310   GNUNET_CONTAINER_MDLL_insert (neighbour,
7311                                 neighbour->queue_head,
7312                                 neighbour->queue_tail,
7313                                 queue);
7314   GNUNET_CONTAINER_MDLL_insert (client,
7315                                 tc->details.communicator.queue_head,
7316                                 tc->details.communicator.queue_tail,
7317                                 queue);
7318   /* check if valdiations are waiting for the queue */
7319   (void)
7320     GNUNET_CONTAINER_multipeermap_get_multiple (validation_map,
7321                                                 &aqm->receiver,
7322                                                 &check_validation_request_pending,
7323                                                 queue);
7324   /* might be our first queue, try launching DV learning */
7325   if (NULL == dvlearn_task)
7326     dvlearn_task = GNUNET_SCHEDULER_add_now (&start_dv_learn, NULL);
7327   GNUNET_SERVICE_client_continue (tc->client);
7328 }
7329
7330
7331 /**
7332  * Communicator tells us that our request to create a queue "worked", that
7333  * is setting up the queue is now in process.
7334  *
7335  * @param cls the `struct TransportClient`
7336  * @param cqr confirmation message
7337  */
7338 static void
7339 handle_queue_create_ok (void *cls,
7340                         const struct GNUNET_TRANSPORT_CreateQueueResponse *cqr)
7341 {
7342   struct TransportClient *tc = cls;
7343
7344   if (CT_COMMUNICATOR != tc->type)
7345   {
7346     GNUNET_break (0);
7347     GNUNET_SERVICE_client_drop (tc->client);
7348     return;
7349   }
7350   GNUNET_STATISTICS_update (GST_stats,
7351                             "# Suggestions succeeded at communicator",
7352                             1,
7353                             GNUNET_NO);
7354   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
7355               "Request #%u for communicator to create queue succeeded\n",
7356               (unsigned int) ntohs (cqr->request_id));
7357   GNUNET_SERVICE_client_continue (tc->client);
7358 }
7359
7360
7361 /**
7362  * Communicator tells us that our request to create a queue failed. This usually
7363  * indicates that the provided address is simply invalid or that the
7364  * communicator's resources are exhausted.
7365  *
7366  * @param cls the `struct TransportClient`
7367  * @param cqr failure message
7368  */
7369 static void
7370 handle_queue_create_fail (
7371   void *cls,
7372   const struct GNUNET_TRANSPORT_CreateQueueResponse *cqr)
7373 {
7374   struct TransportClient *tc = cls;
7375
7376   if (CT_COMMUNICATOR != tc->type)
7377   {
7378     GNUNET_break (0);
7379     GNUNET_SERVICE_client_drop (tc->client);
7380     return;
7381   }
7382   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
7383               "Request #%u for communicator to create queue failed\n",
7384               (unsigned int) ntohs (cqr->request_id));
7385   GNUNET_STATISTICS_update (GST_stats,
7386                             "# Suggestions failed in queue creation at communicator",
7387                             1,
7388                             GNUNET_NO);
7389   GNUNET_SERVICE_client_continue (tc->client);
7390 }
7391
7392
7393 /**
7394  * We have received a `struct ExpressPreferenceMessage` from an application
7395  * client.
7396  *
7397  * @param cls handle to the client
7398  * @param msg the start message
7399  */
7400 static void
7401 handle_suggest_cancel (void *cls, const struct ExpressPreferenceMessage *msg)
7402 {
7403   struct TransportClient *tc = cls;
7404   struct PeerRequest *pr;
7405
7406   if (CT_APPLICATION != tc->type)
7407   {
7408     GNUNET_break (0);
7409     GNUNET_SERVICE_client_drop (tc->client);
7410     return;
7411   }
7412   pr = GNUNET_CONTAINER_multipeermap_get (tc->details.application.requests,
7413                                           &msg->peer);
7414   if (NULL == pr)
7415   {
7416     GNUNET_break (0);
7417     GNUNET_SERVICE_client_drop (tc->client);
7418     return;
7419   }
7420   (void) stop_peer_request (tc, &pr->pid, pr);
7421   GNUNET_SERVICE_client_continue (tc->client);
7422 }
7423
7424
7425 /**
7426  * Check #GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_CONSIDER_VERIFY
7427  * messages. We do nothing here, real verification is done later.
7428  *
7429  * @param cls a `struct TransportClient *`
7430  * @param msg message to verify
7431  * @return #GNUNET_OK
7432  */
7433 static int
7434 check_address_consider_verify (
7435   void *cls,
7436   const struct GNUNET_TRANSPORT_AddressToVerify *hdr)
7437 {
7438   (void) cls;
7439   (void) hdr;
7440   return GNUNET_OK;
7441 }
7442
7443
7444 /**
7445  * Closure for #check_known_address.
7446  */
7447 struct CheckKnownAddressContext
7448 {
7449   /**
7450    * Set to the address we are looking for.
7451    */
7452   const char *address;
7453
7454   /**
7455    * Set to a matching validation state, if one was found.
7456    */
7457   struct ValidationState *vs;
7458 };
7459
7460
7461 /**
7462  * Test if the validation state in @a value matches the
7463  * address from @a cls.
7464  *
7465  * @param cls a `struct CheckKnownAddressContext`
7466  * @param pid unused (must match though)
7467  * @param value a `struct ValidationState`
7468  * @return #GNUNET_OK if not matching, #GNUNET_NO if match found
7469  */
7470 static int
7471 check_known_address (void *cls,
7472                      const struct GNUNET_PeerIdentity *pid,
7473                      void *value)
7474 {
7475   struct CheckKnownAddressContext *ckac = cls;
7476   struct ValidationState *vs = value;
7477
7478   (void) pid;
7479   if (0 != strcmp (vs->address, ckac->address))
7480     return GNUNET_OK;
7481   ckac->vs = vs;
7482   return GNUNET_NO;
7483 }
7484
7485
7486 /**
7487  * Start address validation.
7488  *
7489  * @param pid peer the @a address is for
7490  * @param address an address to reach @a pid (presumably)
7491  * @param expiration when did @a pid claim @a address will become invalid
7492  */
7493 static void
7494 start_address_validation (const struct GNUNET_PeerIdentity *pid,
7495                           const char *address,
7496                           struct GNUNET_TIME_Absolute expiration)
7497 {
7498   struct GNUNET_TIME_Absolute now;
7499   struct ValidationState *vs;
7500   struct CheckKnownAddressContext ckac = {.address = address, .vs = NULL};
7501
7502   if (0 == GNUNET_TIME_absolute_get_remaining (expiration).rel_value_us)
7503     return; /* expired */
7504   (void) GNUNET_CONTAINER_multipeermap_get_multiple (validation_map,
7505                                                      pid,
7506                                                      &check_known_address,
7507                                                      &ckac);
7508   if (NULL != (vs = ckac.vs))
7509   {
7510     /* if 'vs' is not currently valid, we need to speed up retrying the
7511      * validation */
7512     if (vs->validated_until.abs_value_us < vs->next_challenge.abs_value_us)
7513     {
7514       /* reduce backoff as we got a fresh advertisement */
7515       vs->challenge_backoff =
7516         GNUNET_TIME_relative_min (FAST_VALIDATION_CHALLENGE_FREQ,
7517                                   GNUNET_TIME_relative_divide (vs->challenge_backoff,
7518                                                                2));
7519       update_next_challenge_time (vs,
7520                                   GNUNET_TIME_relative_to_absolute (
7521                                     vs->challenge_backoff));
7522     }
7523     return;
7524   }
7525   now = GNUNET_TIME_absolute_get ();
7526   vs = GNUNET_new (struct ValidationState);
7527   vs->pid = *pid;
7528   vs->valid_until = expiration;
7529   vs->first_challenge_use = now;
7530   vs->validation_rtt = GNUNET_TIME_UNIT_FOREVER_REL;
7531   GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
7532                               &vs->challenge,
7533                               sizeof (vs->challenge));
7534   vs->address = GNUNET_strdup (address);
7535   GNUNET_assert (GNUNET_YES ==
7536                  GNUNET_CONTAINER_multipeermap_put (
7537                    validation_map,
7538                    &vs->pid,
7539                    vs,
7540                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
7541   update_next_challenge_time (vs, now);
7542 }
7543
7544
7545 /**
7546  * Function called by PEERSTORE for each matching record.
7547  *
7548  * @param cls closure
7549  * @param record peerstore record information
7550  * @param emsg error message, or NULL if no errors
7551  */
7552 static void
7553 handle_hello (void *cls,
7554               const struct GNUNET_PEERSTORE_Record *record,
7555               const char *emsg)
7556 {
7557   struct PeerRequest *pr = cls;
7558   const char *val;
7559
7560   if (NULL != emsg)
7561   {
7562     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
7563                 "Got failure from PEERSTORE: %s\n",
7564                 emsg);
7565     return;
7566   }
7567   val = record->value;
7568   if ((0 == record->value_size) || ('\0' != val[record->value_size - 1]))
7569   {
7570     GNUNET_break (0);
7571     return;
7572   }
7573   start_address_validation (&pr->pid,
7574                             (const char *) record->value,
7575                             record->expiry);
7576 }
7577
7578
7579 /**
7580  * We have received a `struct ExpressPreferenceMessage` from an application
7581  * client.
7582  *
7583  * @param cls handle to the client
7584  * @param msg the start message
7585  */
7586 static void
7587 handle_suggest (void *cls, const struct ExpressPreferenceMessage *msg)
7588 {
7589   struct TransportClient *tc = cls;
7590   struct PeerRequest *pr;
7591
7592   if (CT_NONE == tc->type)
7593   {
7594     tc->type = CT_APPLICATION;
7595     tc->details.application.requests =
7596       GNUNET_CONTAINER_multipeermap_create (16, GNUNET_YES);
7597   }
7598   if (CT_APPLICATION != tc->type)
7599   {
7600     GNUNET_break (0);
7601     GNUNET_SERVICE_client_drop (tc->client);
7602     return;
7603   }
7604   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
7605               "Client suggested we talk to %s with preference %d at rate %u\n",
7606               GNUNET_i2s (&msg->peer),
7607               (int) ntohl (msg->pk),
7608               (int) ntohl (msg->bw.value__));
7609   pr = GNUNET_new (struct PeerRequest);
7610   pr->tc = tc;
7611   pr->pid = msg->peer;
7612   pr->bw = msg->bw;
7613   pr->pk = (enum GNUNET_MQ_PreferenceKind) ntohl (msg->pk);
7614   if (GNUNET_YES != GNUNET_CONTAINER_multipeermap_put (
7615                       tc->details.application.requests,
7616                       &pr->pid,
7617                       pr,
7618                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
7619   {
7620     GNUNET_break (0);
7621     GNUNET_free (pr);
7622     GNUNET_SERVICE_client_drop (tc->client);
7623     return;
7624   }
7625   pr->wc = GNUNET_PEERSTORE_watch (peerstore,
7626                                    "transport",
7627                                    &pr->pid,
7628                                    GNUNET_PEERSTORE_TRANSPORT_URLADDRESS_KEY,
7629                                    &handle_hello,
7630                                    pr);
7631   GNUNET_SERVICE_client_continue (tc->client);
7632 }
7633
7634
7635 /**
7636  * Given another peers address, consider checking it for validity
7637  * and then adding it to the Peerstore.
7638  *
7639  * @param cls a `struct TransportClient`
7640  * @param hdr message containing the raw address data and
7641  *        signature in the body, see #GNUNET_HELLO_extract_address()
7642  */
7643 static void
7644 handle_address_consider_verify (
7645   void *cls,
7646   const struct GNUNET_TRANSPORT_AddressToVerify *hdr)
7647 {
7648   struct TransportClient *tc = cls;
7649   char *address;
7650   enum GNUNET_NetworkType nt;
7651   struct GNUNET_TIME_Absolute expiration;
7652
7653   (void) cls;
7654   // OPTIMIZE-FIXME: checking that we know this address already should
7655   //        be done BEFORE checking the signature => HELLO API change!
7656   // OPTIMIZE-FIXME: pre-check: rate-limit signature verification / validation?!
7657   address =
7658     GNUNET_HELLO_extract_address (&hdr[1],
7659                                   ntohs (hdr->header.size) - sizeof (*hdr),
7660                                   &hdr->peer,
7661                                   &nt,
7662                                   &expiration);
7663   if (NULL == address)
7664   {
7665     GNUNET_break_op (0);
7666     return;
7667   }
7668   start_address_validation (&hdr->peer, address, expiration);
7669   GNUNET_free (address);
7670   GNUNET_SERVICE_client_continue (tc->client);
7671 }
7672
7673
7674 /**
7675  * Check #GNUNET_MESSAGE_TYPE_TRANSPORT_REQUEST_HELLO_VALIDATION
7676  * messages.
7677  *
7678  * @param cls a `struct TransportClient *`
7679  * @param m message to verify
7680  * @return #GNUNET_OK on success
7681  */
7682 static int
7683 check_request_hello_validation (void *cls,
7684                                 const struct RequestHelloValidationMessage *m)
7685 {
7686   (void) cls;
7687   GNUNET_MQ_check_zero_termination (m);
7688   return GNUNET_OK;
7689 }
7690
7691
7692 /**
7693  * A client encountered an address of another peer. Consider validating it,
7694  * and if validation succeeds, persist it to PEERSTORE.
7695  *
7696  * @param cls a `struct TransportClient *`
7697  * @param m message to verify
7698  */
7699 static void
7700 handle_request_hello_validation (void *cls,
7701                                  const struct RequestHelloValidationMessage *m)
7702 {
7703   struct TransportClient *tc = cls;
7704
7705   start_address_validation (&m->peer,
7706                             (const char *) &m[1],
7707                             GNUNET_TIME_absolute_ntoh (m->expiration));
7708   GNUNET_SERVICE_client_continue (tc->client);
7709 }
7710
7711
7712 /**
7713  * Free neighbour entry.
7714  *
7715  * @param cls NULL
7716  * @param pid unused
7717  * @param value a `struct Neighbour`
7718  * @return #GNUNET_OK (always)
7719  */
7720 static int
7721 free_neighbour_cb (void *cls,
7722                    const struct GNUNET_PeerIdentity *pid,
7723                    void *value)
7724 {
7725   struct Neighbour *neighbour = value;
7726
7727   (void) cls;
7728   (void) pid;
7729   GNUNET_break (0); // should this ever happen?
7730   free_neighbour (neighbour);
7731
7732   return GNUNET_OK;
7733 }
7734
7735
7736 /**
7737  * Free DV route entry.
7738  *
7739  * @param cls NULL
7740  * @param pid unused
7741  * @param value a `struct DistanceVector`
7742  * @return #GNUNET_OK (always)
7743  */
7744 static int
7745 free_dv_routes_cb (void *cls,
7746                    const struct GNUNET_PeerIdentity *pid,
7747                    void *value)
7748 {
7749   struct DistanceVector *dv = value;
7750
7751   (void) cls;
7752   (void) pid;
7753   free_dv_route (dv);
7754
7755   return GNUNET_OK;
7756 }
7757
7758
7759 /**
7760  * Free ephemeral entry.
7761  *
7762  * @param cls NULL
7763  * @param pid unused
7764  * @param value a `struct EphemeralCacheEntry`
7765  * @return #GNUNET_OK (always)
7766  */
7767 static int
7768 free_ephemeral_cb (void *cls,
7769                    const struct GNUNET_PeerIdentity *pid,
7770                    void *value)
7771 {
7772   struct EphemeralCacheEntry *ece = value;
7773
7774   (void) cls;
7775   (void) pid;
7776   free_ephemeral (ece);
7777   return GNUNET_OK;
7778 }
7779
7780
7781 /**
7782  * Free validation state.
7783  *
7784  * @param cls NULL
7785  * @param pid unused
7786  * @param value a `struct ValidationState`
7787  * @return #GNUNET_OK (always)
7788  */
7789 static int
7790 free_validation_state_cb (void *cls,
7791                           const struct GNUNET_PeerIdentity *pid,
7792                           void *value)
7793 {
7794   struct ValidationState *vs = value;
7795
7796   (void) cls;
7797   (void) pid;
7798   free_validation_state (vs);
7799   return GNUNET_OK;
7800 }
7801
7802
7803 /**
7804  * Function called when the service shuts down.  Unloads our plugins
7805  * and cancels pending validations.
7806  *
7807  * @param cls closure, unused
7808  */
7809 static void
7810 do_shutdown (void *cls)
7811 {
7812   struct LearnLaunchEntry *lle;
7813   (void) cls;
7814
7815   if (NULL != ephemeral_task)
7816   {
7817     GNUNET_SCHEDULER_cancel (ephemeral_task);
7818     ephemeral_task = NULL;
7819   }
7820   GNUNET_CONTAINER_multipeermap_iterate (neighbours, &free_neighbour_cb, NULL);
7821   if (NULL != peerstore)
7822   {
7823     GNUNET_PEERSTORE_disconnect (peerstore, GNUNET_NO);
7824     peerstore = NULL;
7825   }
7826   if (NULL != GST_stats)
7827   {
7828     GNUNET_STATISTICS_destroy (GST_stats, GNUNET_NO);
7829     GST_stats = NULL;
7830   }
7831   if (NULL != GST_my_private_key)
7832   {
7833     GNUNET_free (GST_my_private_key);
7834     GST_my_private_key = NULL;
7835   }
7836   GNUNET_CONTAINER_multipeermap_destroy (neighbours);
7837   neighbours = NULL;
7838   GNUNET_CONTAINER_multipeermap_iterate (backtalkers,
7839                                          &free_backtalker_cb,
7840                                          NULL);
7841   GNUNET_CONTAINER_multipeermap_destroy (backtalkers);
7842   backtalkers = NULL;
7843   GNUNET_CONTAINER_multipeermap_iterate (validation_map,
7844                                          &free_validation_state_cb,
7845                                          NULL);
7846   GNUNET_CONTAINER_multipeermap_destroy (validation_map);
7847   validation_map = NULL;
7848   while (NULL != (lle = lle_head))
7849   {
7850     GNUNET_CONTAINER_DLL_remove (lle_head, lle_tail, lle);
7851     GNUNET_free (lle);
7852   }
7853   GNUNET_CONTAINER_multishortmap_destroy (dvlearn_map);
7854   dvlearn_map = NULL;
7855   GNUNET_CONTAINER_heap_destroy (validation_heap);
7856   validation_heap = NULL;
7857   GNUNET_CONTAINER_multipeermap_iterate (dv_routes, &free_dv_routes_cb, NULL);
7858   GNUNET_CONTAINER_multipeermap_destroy (dv_routes);
7859   dv_routes = NULL;
7860   GNUNET_CONTAINER_multipeermap_iterate (ephemeral_map,
7861                                          &free_ephemeral_cb,
7862                                          NULL);
7863   GNUNET_CONTAINER_multipeermap_destroy (ephemeral_map);
7864   ephemeral_map = NULL;
7865   GNUNET_CONTAINER_heap_destroy (ephemeral_heap);
7866   ephemeral_heap = NULL;
7867 }
7868
7869
7870 /**
7871  * Initiate transport service.
7872  *
7873  * @param cls closure
7874  * @param c configuration to use
7875  * @param service the initialized service
7876  */
7877 static void
7878 run (void *cls,
7879      const struct GNUNET_CONFIGURATION_Handle *c,
7880      struct GNUNET_SERVICE_Handle *service)
7881 {
7882   (void) cls;
7883   (void) service;
7884   /* setup globals */
7885   GST_cfg = c;
7886   backtalkers = GNUNET_CONTAINER_multipeermap_create (16, GNUNET_YES);
7887   neighbours = GNUNET_CONTAINER_multipeermap_create (1024, GNUNET_YES);
7888   dv_routes = GNUNET_CONTAINER_multipeermap_create (1024, GNUNET_YES);
7889   ephemeral_map = GNUNET_CONTAINER_multipeermap_create (32, GNUNET_YES);
7890   ephemeral_heap =
7891     GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
7892   dvlearn_map = GNUNET_CONTAINER_multishortmap_create (2 * MAX_DV_LEARN_PENDING,
7893                                                        GNUNET_YES);
7894   validation_map = GNUNET_CONTAINER_multipeermap_create (1024, GNUNET_YES);
7895   validation_heap =
7896     GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
7897   GST_my_private_key =
7898     GNUNET_CRYPTO_eddsa_key_create_from_configuration (GST_cfg);
7899   if (NULL == GST_my_private_key)
7900   {
7901     GNUNET_log (
7902       GNUNET_ERROR_TYPE_ERROR,
7903       _ (
7904         "Transport service is lacking key configuration settings. Exiting.\n"));
7905     GNUNET_SCHEDULER_shutdown ();
7906     return;
7907   }
7908   GNUNET_CRYPTO_eddsa_key_get_public (GST_my_private_key,
7909                                       &GST_my_identity.public_key);
7910   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
7911               "My identity is `%s'\n",
7912               GNUNET_i2s_full (&GST_my_identity));
7913   GST_stats = GNUNET_STATISTICS_create ("transport", GST_cfg);
7914   GNUNET_SCHEDULER_add_shutdown (&do_shutdown, NULL);
7915   peerstore = GNUNET_PEERSTORE_connect (GST_cfg);
7916   if (NULL == peerstore)
7917   {
7918     GNUNET_break (0);
7919     GNUNET_SCHEDULER_shutdown ();
7920     return;
7921   }
7922 }
7923
7924
7925 /**
7926  * Define "main" method using service macro.
7927  */
7928 GNUNET_SERVICE_MAIN (
7929   "transport",
7930   GNUNET_SERVICE_OPTION_SOFT_SHUTDOWN,
7931   &run,
7932   &client_connect_cb,
7933   &client_disconnect_cb,
7934   NULL,
7935   /* communication with applications */
7936   GNUNET_MQ_hd_fixed_size (suggest,
7937                            GNUNET_MESSAGE_TYPE_TRANSPORT_SUGGEST,
7938                            struct ExpressPreferenceMessage,
7939                            NULL),
7940   GNUNET_MQ_hd_fixed_size (suggest_cancel,
7941                            GNUNET_MESSAGE_TYPE_TRANSPORT_SUGGEST_CANCEL,
7942                            struct ExpressPreferenceMessage,
7943                            NULL),
7944   GNUNET_MQ_hd_var_size (request_hello_validation,
7945                          GNUNET_MESSAGE_TYPE_TRANSPORT_REQUEST_HELLO_VALIDATION,
7946                          struct RequestHelloValidationMessage,
7947                          NULL),
7948   /* communication with core */
7949   GNUNET_MQ_hd_fixed_size (client_start,
7950                            GNUNET_MESSAGE_TYPE_TRANSPORT_START,
7951                            struct StartMessage,
7952                            NULL),
7953   GNUNET_MQ_hd_var_size (client_send,
7954                          GNUNET_MESSAGE_TYPE_TRANSPORT_SEND,
7955                          struct OutboundMessage,
7956                          NULL),
7957   /* communication with communicators */
7958   GNUNET_MQ_hd_var_size (communicator_available,
7959                          GNUNET_MESSAGE_TYPE_TRANSPORT_NEW_COMMUNICATOR,
7960                          struct GNUNET_TRANSPORT_CommunicatorAvailableMessage,
7961                          NULL),
7962   GNUNET_MQ_hd_var_size (communicator_backchannel,
7963                          GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL,
7964                          struct GNUNET_TRANSPORT_CommunicatorBackchannel,
7965                          NULL),
7966   GNUNET_MQ_hd_var_size (add_address,
7967                          GNUNET_MESSAGE_TYPE_TRANSPORT_ADD_ADDRESS,
7968                          struct GNUNET_TRANSPORT_AddAddressMessage,
7969                          NULL),
7970   GNUNET_MQ_hd_fixed_size (del_address,
7971                            GNUNET_MESSAGE_TYPE_TRANSPORT_DEL_ADDRESS,
7972                            struct GNUNET_TRANSPORT_DelAddressMessage,
7973                            NULL),
7974   GNUNET_MQ_hd_var_size (incoming_msg,
7975                          GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG,
7976                          struct GNUNET_TRANSPORT_IncomingMessage,
7977                          NULL),
7978   GNUNET_MQ_hd_fixed_size (queue_create_ok,
7979                            GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_OK,
7980                            struct GNUNET_TRANSPORT_CreateQueueResponse,
7981                            NULL),
7982   GNUNET_MQ_hd_fixed_size (queue_create_fail,
7983                            GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_FAIL,
7984                            struct GNUNET_TRANSPORT_CreateQueueResponse,
7985                            NULL),
7986   GNUNET_MQ_hd_var_size (add_queue_message,
7987                          GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_SETUP,
7988                          struct GNUNET_TRANSPORT_AddQueueMessage,
7989                          NULL),
7990   GNUNET_MQ_hd_var_size (address_consider_verify,
7991                          GNUNET_MESSAGE_TYPE_TRANSPORT_ADDRESS_CONSIDER_VERIFY,
7992                          struct GNUNET_TRANSPORT_AddressToVerify,
7993                          NULL),
7994   GNUNET_MQ_hd_fixed_size (del_queue_message,
7995                            GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_TEARDOWN,
7996                            struct GNUNET_TRANSPORT_DelQueueMessage,
7997                            NULL),
7998   GNUNET_MQ_hd_fixed_size (send_message_ack,
7999                            GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG_ACK,
8000                            struct GNUNET_TRANSPORT_SendMessageToAck,
8001                            NULL),
8002   /* communication with monitors */
8003   GNUNET_MQ_hd_fixed_size (monitor_start,
8004                            GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_START,
8005                            struct GNUNET_TRANSPORT_MonitorStart,
8006                            NULL),
8007   GNUNET_MQ_handler_end ());
8008
8009
8010 /* end of file gnunet-service-transport.c */