Line data Source code
1 : #define _GNU_SOURCE
2 : #include "fd_failover_channel.h"
3 : #include "fd_failover_tls.h"
4 : #include "../../util/fd_util.h"
5 : #include "../../util/net/fd_ip4.h"
6 : #include "../../ballet/ed25519/fd_ed25519.h"
7 : #include "../../waltz/tls/fd_tls_proto.h"
8 : #include "../../waltz/fd_token_bucket.h"
9 : #include "../../disco/bundle/fd_keepalive.h"
10 :
11 : #include <errno.h>
12 : #include <string.h>
13 : #include <unistd.h>
14 : #include <sys/socket.h>
15 : #include <netinet/in.h>
16 : #include <netinet/tcp.h>
17 :
18 : /* Candidate slots. */
19 306 : #define OUTBOUND_IDX (0UL)
20 726 : #define ADMITTED_IDX (1UL)
21 135492 : #define UNAUTHENTICATED_IDX (2UL)
22 :
23 : /* TLS admission phase FSM states. */
24 156 : #define PHASE_CONNECT (0)
25 27 : #define PHASE_START_SEND (1)
26 15 : #define PHASE_START_RECV (2)
27 21 : #define PHASE_CHALLENGE_SEND (3)
28 27 : #define PHASE_CHALLENGE_RECV (4)
29 6 : #define PHASE_REQUEST_SIGN (5)
30 21 : #define PHASE_REQUEST_SEND (6)
31 90 : #define PHASE_REQUEST_RECV (7)
32 6 : #define PHASE_RESPONSE_SIGN (8)
33 18 : #define PHASE_RESPONSE_SEND (9)
34 18 : #define PHASE_RESPONSE_RECV (10)
35 21 : #define PHASE_TLS_START (11)
36 3186 : #define PHASE_TLS (12)
37 15 : #define PHASE_HELLO (13)
38 117 : #define PHASE_READY (14)
39 :
40 : /* Bounds. */
41 : FD_STATIC_ASSERT( FD_FAILOVER_CHANNEL_CANDIDATE_MAX>UNAUTHENTICATED_IDX, admission_pool_capacity );
42 : FD_STATIC_ASSERT( FD_FAILOVER_CHANNEL_SESSION_MAX>=2UL, session_pool_capacity );
43 : FD_STATIC_ASSERT( FD_FAILOVER_CHANNEL_KEEPALIVE_NANOS>=2L && FD_FAILOVER_CHANNEL_IDLE_NANOS>0L, keepalive_params );
44 :
45 : FD_STATIC_ASSERT( FD_KEYGUARD_FAILOVER_PROOF_SZ<=FD_TLS_CV_SIGN_SZ, admission_sign_mtu );
46 : FD_STATIC_ASSERT( FD_FAILOVER_CLIENT_HELLO_MAX+sizeof(fd_tlsrec_hdr_t)+sizeof(fd_failover_admission_start_t)<=FD_FAILOVER_ADMISSION_MAX, admission_buffer );
47 :
48 : /* A HELLO whose identity does not match the key that was TLS authenticated */
49 0 : #define HELLO_ERR_TLS_KEY (-1)
50 :
51 : /* Logging: What the dialer specifically waits for during each phase. */
52 : static char const *
53 3 : awaited_cstr( int phase ) {
54 3 : switch( phase ) {
55 0 : case PHASE_CONNECT: return "its TCP answer";
56 0 : case PHASE_START_SEND: return "our admission start to drain";
57 3 : case PHASE_CHALLENGE_RECV: return "its admission challenge";
58 0 : case PHASE_REQUEST_SIGN: return "the sign tile to sign our admission proof";
59 0 : case PHASE_REQUEST_SEND: return "our admission proof to drain";
60 0 : case PHASE_RESPONSE_RECV: return "its answer to our admission proof";
61 0 : case PHASE_TLS: return "the TLS handshake to finish";
62 0 : default: return "its HELLO";
63 3 : }
64 3 : }
65 :
66 : struct candidate {
67 : int fd; /* -1 for an unused candidate. */
68 : int phase; /* PHASE_* */
69 : int dialed; /* we dialed it, else the listener accepted it */
70 : int session; /* ch->sessions index once TLS starts, else -1 */
71 : uint address;
72 : long deadline;
73 : long admission_until;
74 : fd_failover_hello_t local_hello;
75 : fd_failover_admission_start_t start;
76 : fd_failover_admission_challenge_t challenge;
77 : uchar client_hello[ FD_FAILOVER_CLIENT_HELLO_MAX+sizeof(fd_tlsrec_hdr_t) ];
78 : uchar transcript[ 32 ];
79 : uchar client_proof[ 64 ];
80 : fd_failover_hello_t hello;
81 : uchar rx[ FD_FAILOVER_ADMISSION_MAX ];
82 : uchar tx[ FD_FAILOVER_ADMISSION_MAX ];
83 : ulong rx_used;
84 : ulong tx_used;
85 : ulong tx_sent;
86 : };
87 :
88 : /* Expensive TLS state only for admitted candidates. */
89 : struct session {
90 : int owner; /* the candidate index that holds this session, -1 when free */
91 : fd_failover_tls_t tls;
92 : uchar rx[ FD_FAILOVER_FRAME_MAX ];
93 : uchar tx[ FD_FAILOVER_FRAME_MAX ];
94 : };
95 :
96 : struct fd_failover_channel {
97 : /* Session */
98 : ulong magic; /* FD_FAILOVER_CHANNEL_MAGIC once initialized */
99 : int dial_peer; /* the session machine runs as the dialer */
100 : uint peer_addr; /* the dial destination, zero when we do not dial */
101 : ushort peer_port; /* the dial destination port */
102 : int listen_fd; /* listening socket, -1 until it is open */
103 : int paired_idx; /* the paired candidate, -1 while none is paired */
104 : ulong cursor; /* the next unauthenticated candidate idx to service */
105 :
106 : /* Credentials */
107 : fd_failover_tls_ctx_t tls_ctx; /* TLS template every connection starts from */
108 : fd_failover_hello_t self_hello; /* the HELLO we send */
109 : fd_failover_hello_t peer_hello; /* the paired peer's HELLO, zero while unpaired */
110 : fd_failover_proof_signer_t proof_signer; /* signs our admission proofs in the sign tile */
111 : fd_sha512_t sha[ 1 ]; /* scratch hash for checking admission proofs */
112 :
113 : /* Connections */
114 : struct candidate candidates[ FD_FAILOVER_CHANNEL_CANDIDATE_MAX ]; /* connections that are being admitted or paired */
115 : struct session sessions[ FD_FAILOVER_CHANNEL_SESSION_MAX ]; /* TLS state, only for the dialer and verified candidates */
116 : ulong live; /* candidates with an open socket */
117 : uint known_addr; /* the last address that authenticated, gets the ADMITTED slot ahead of admission */
118 : long known_until; /* until when */
119 :
120 : /* Rate limits */
121 : fd_token_bucket_t accept_bucket; /* accept syscalls */
122 : fd_token_bucket_t verify_bucket; /* Ed25519 proof checks and signs, at most one per poll */
123 : int verify_budget; /* proof checks left this poll */
124 : long refused_log_at; /* when a refusal of an unproven peer is logged next */
125 : ulong refused_cnt; /* refusals of unproven peers not logged since */
126 :
127 : /* Timers */
128 : fd_keepalive_t keepalive[ 1 ]; /* when the paired session last heard from the peer / pings it next */
129 : fd_rng_t rng[ 1 ]; /* spreads redials and KEEPALIVEs, not for secure RNG */
130 : long retry_at; /* when the dialer dials again */
131 : long paired_at; /* when the current session paired */
132 : long backoff; /* the current dial backoff, doubles up to BACKOFF_MAX */
133 :
134 : /* The paired session */
135 : int expected_close; /* RESULT was sent, peer may close normally */
136 : ulong generation; /* identifies the current authenticated session */
137 : };
138 :
139 336 : FD_FN_CONST ulong fd_failover_channel_align ( void ) { return alignof(fd_failover_channel_t); }
140 6 : FD_FN_CONST ulong fd_failover_channel_footprint( void ) { return sizeof (fd_failover_channel_t); }
141 :
142 : /* A bucket of burst tokens refilling at `rate` per second. */
143 : static fd_token_bucket_t
144 : token_bucket_new( ulong rate,
145 336 : ulong burst ) {
146 336 : return (fd_token_bucket_t){ .rate=(float)rate/(float)FD_FAILOVER_NANOS_PER_SECOND, .burst=(float)burst, .balance=(float)burst };
147 336 : }
148 :
149 : void *
150 168 : fd_failover_channel_new( void * shmem ) {
151 168 : if( !shmem || !fd_ulong_is_aligned( (ulong)shmem, fd_failover_channel_align() ) ) return NULL;
152 168 : fd_failover_channel_t * ch = shmem;
153 :
154 : /* Initialize the listening state. */
155 168 : fd_memset( ch, 0, sizeof(*ch) );
156 168 : ch->listen_fd = -1;
157 168 : ch->paired_idx = -1;
158 172200 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) { ch->candidates[i].fd = -1; ch->candidates[i].session = -1; }
159 840 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_SESSION_MAX; i++ ) ch->sessions[i].owner = -1;
160 168 : ch->backoff = FD_FAILOVER_CHANNEL_BACKOFF_MIN_NANOS;
161 :
162 : /* Rate limiting */
163 168 : ch->accept_bucket = token_bucket_new( FD_FAILOVER_ACCEPT_RATE, FD_FAILOVER_ACCEPT_BURST );
164 168 : ch->verify_bucket = token_bucket_new( FD_FAILOVER_VERIFY_RATE, FD_FAILOVER_VERIFY_BURST );
165 :
166 : /* Set up secure randomness */
167 168 : uint seed;
168 168 : FD_TEST( fd_rng_secure( &seed, sizeof(seed) ) );
169 168 : FD_TEST( fd_rng_join( fd_rng_new( ch->rng, seed, 0UL ) ) );
170 168 : FD_TEST( fd_sha512_join( fd_sha512_new( ch->sha ) ) );
171 :
172 168 : ch->magic = FD_FAILOVER_CHANNEL_MAGIC;
173 168 : return shmem;
174 168 : }
175 :
176 : fd_failover_channel_t *
177 168 : fd_failover_channel_join( void * shmem ) {
178 168 : if( !shmem || !fd_ulong_is_aligned( (ulong)shmem, fd_failover_channel_align() ) ) return NULL;
179 168 : fd_failover_channel_t * ch = shmem;
180 168 : return ch->magic==FD_FAILOVER_CHANNEL_MAGIC ? ch : NULL;
181 168 : }
182 :
183 : /* Gives candidate idx a TLS session. Returns 1 once it owns one,
184 : 0 when every session is busy so the caller should try again later,
185 : and -1 when starting TLS failed. */
186 : static int
187 15 : acquire_session( fd_failover_channel_t * ch, ulong idx, int dial ) {
188 15 : struct candidate * c = &ch->candidates[idx];
189 15 : if( c->session!=-1 ) return 1;
190 15 : ulong lo = idx==OUTBOUND_IDX ? OUTBOUND_IDX : OUTBOUND_IDX+1UL;
191 15 : ulong hi = idx==OUTBOUND_IDX ? OUTBOUND_IDX+1UL : FD_FAILOVER_CHANNEL_SESSION_MAX;
192 33 : for( ulong i=lo; i<hi; i++ ) {
193 27 : if( ch->sessions[i].owner!=-1 ) continue;
194 9 : if( fd_failover_tls_new( &ch->sessions[i].tls, &ch->tls_ctx, c->fd, dial ) ) return -1;
195 9 : ch->sessions[i].owner = (int)idx;
196 9 : c->session = (int)i;
197 9 : return 1;
198 9 : }
199 6 : return 0;
200 15 : }
201 :
202 : static void
203 273432 : close_candidate( fd_failover_channel_t * ch, struct candidate * c ) {
204 273432 : if( c->session!=-1 ) {
205 108 : fd_failover_tls_fini( &ch->sessions[ c->session ].tls );
206 108 : ch->sessions[ c->session ].owner = -1;
207 108 : c->session = -1;
208 108 : }
209 : /* Closing a candidate frees its slot. */
210 273432 : if( c->fd!=-1 ) { close( c->fd ); c->fd = -1; ch->live--; }
211 273432 : c->dialed = 0;
212 273432 : c->rx_used = c->tx_used = c->tx_sent = 0UL;
213 273432 : }
214 :
215 : /* Discards every candidate and the pairing. */
216 : static void
217 210 : discard( fd_failover_channel_t * ch ) {
218 215250 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) close_candidate( ch, &ch->candidates[i] );
219 210 : ch->paired_idx = -1;
220 210 : fd_memset( &ch->peer_hello, 0, sizeof(ch->peer_hello) );
221 210 : }
222 :
223 : static int
224 180 : want_dial( fd_failover_channel_t const * ch ) {
225 180 : return !!ch->peer_addr;
226 180 : }
227 :
228 : /* Resets an unpaired channel: drops any dial in flight and goes back
229 : to listening, or to an immediate redial when a dial address is set. */
230 : static void
231 150 : reset( fd_failover_channel_t * ch ) {
232 150 : if( ch->paired_idx>=0 ) return;
233 144525 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) {
234 144384 : if( ch->candidates[i].dialed ) {
235 0 : close_candidate( ch, &ch->candidates[i] );
236 0 : }
237 144384 : }
238 141 : ch->dial_peer = want_dial( ch );
239 141 : ch->retry_at = 0L;
240 141 : }
241 :
242 : void
243 : fd_failover_channel_init_listener( fd_failover_channel_t * ch,
244 : uint address,
245 12 : ushort port ) {
246 12 : discard( ch );
247 12 : reset( ch );
248 12 : if( ch->listen_fd!=-1 ) return;
249 12 : ch->listen_fd = socket( AF_INET, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0 );
250 12 : if( ch->listen_fd==-1 ) {
251 0 : FD_LOG_ERR(( "failover socket failed (%i-%s)", errno, fd_io_strerror( errno ) ));
252 0 : }
253 12 : int yes = 1;
254 12 : if( setsockopt( ch->listen_fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes) ) ) {
255 0 : FD_LOG_ERR(( "failover setsockopt failed" ));
256 0 : }
257 12 : if( setsockopt( ch->listen_fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes) ) ) {
258 0 : FD_LOG_ERR(( "failover setsockopt(TCP_NODELAY) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
259 0 : }
260 12 : int defer = (int)(FD_FAILOVER_ADMISSION_START_NANOS/(long)FD_FAILOVER_NANOS_PER_SECOND);
261 12 : if( setsockopt( ch->listen_fd, IPPROTO_TCP, TCP_DEFER_ACCEPT, &defer, sizeof(defer) ) ) {
262 0 : FD_LOG_ERR(( "failover setsockopt(TCP_DEFER_ACCEPT) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
263 0 : }
264 12 : struct sockaddr_in addr = { .sin_family=AF_INET, .sin_port=fd_ushort_bswap( port ), .sin_addr.s_addr=address };
265 12 : if( bind( ch->listen_fd, fd_type_pun( &addr ), sizeof(addr) ) || listen( ch->listen_fd, FD_FAILOVER_LISTEN_BACKLOG ) ) {
266 0 : FD_LOG_ERR(( "could not listen on `" FD_IP4_ADDR_FMT ":%hu` (%i-%s)",
267 0 : FD_IP4_ADDR_FMT_ARGS( address ), port, errno, fd_io_strerror( errno ) ));
268 0 : }
269 12 : }
270 :
271 : void
272 : fd_failover_channel_init_dialer( fd_failover_channel_t * ch,
273 : uint address,
274 177 : ushort port ) {
275 177 : if( address==ch->peer_addr && port==ch->peer_port ) return;
276 111 : if( address && ch->paired_idx>=0 ) discard( ch );
277 111 : ch->peer_addr = address;
278 111 : ch->peer_port = port;
279 111 : if( address ) {
280 60 : ch->backoff = FD_FAILOVER_CHANNEL_BACKOFF_MIN_NANOS;
281 60 : }
282 111 : reset( ch );
283 111 : }
284 :
285 : int
286 : fd_failover_channel_set_identity( fd_failover_channel_t * ch,
287 : fd_tls_sign_t signer,
288 : fd_failover_proof_signer_t proof_signer,
289 30 : fd_failover_hello_t const * hello ) {
290 30 : if( !proof_signer.sign_fn || !signer.sign_fn ) return -1;
291 30 : discard( ch );
292 : /* An identity change requires a fresh session. */
293 30 : ch->known_addr = 0U;
294 30 : ch->known_until = 0L;
295 30 : ch->self_hello = *hello;
296 30 : ch->proof_signer = proof_signer;
297 30 : return fd_failover_tls_ctx_init( &ch->tls_ctx, hello->staked_pubkey, signer );
298 30 : }
299 :
300 : void
301 168 : fd_failover_channel_fini( fd_failover_channel_t * ch ) {
302 168 : discard( ch );
303 168 : if( ch->listen_fd!=-1 ) close( ch->listen_fd );
304 168 : ch->listen_fd = -1;
305 168 : fd_failover_tls_ctx_fini( &ch->tls_ctx );
306 168 : }
307 :
308 : void
309 : fd_failover_channel_set_holds( fd_failover_channel_t * ch,
310 57 : int holds ) {
311 57 : if( ch->self_hello.holds==(uchar)!!holds ) return;
312 51 : ch->self_hello.holds = (uchar)!!holds;
313 52275 : for( ulong i=0UL; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ )
314 52224 : if( (int)i!=ch->paired_idx ) close_candidate( ch, &ch->candidates[i] );
315 51 : if( ch->paired_idx<0 ) reset( ch );
316 51 : }
317 :
318 : /* Derive the session state from the socket state. */
319 : FD_FN_PURE ulong
320 858 : fd_failover_channel_state( fd_failover_channel_t const * ch ) {
321 858 : if( ch->paired_idx>=0 ) return FD_FAILOVER_SESSION_PAIRED;
322 477 : if( !ch->dial_peer ) return FD_FAILOVER_SESSION_LISTENING;
323 33 : struct candidate const * c = &ch->candidates[OUTBOUND_IDX];
324 33 : if( c->fd==-1 ) return FD_FAILOVER_SESSION_BACKOFF;
325 21 : if( c->phase==PHASE_CONNECT ) return FD_FAILOVER_SESSION_DIALING;
326 18 : return FD_FAILOVER_SESSION_HELLO;
327 21 : }
328 :
329 0 : FD_FN_PURE int fd_failover_channel_listen_fd( fd_failover_channel_t const * ch ) { return ch->listen_fd; }
330 :
331 : FD_FN_PURE int
332 60 : fd_failover_channel_tx_pending( fd_failover_channel_t const * ch ) {
333 60 : if( ch->paired_idx<0 ) return 0;
334 60 : struct candidate const * c = &ch->candidates[ch->paired_idx];
335 60 : return !!c->tx_used || fd_tlsrec_sock_tx_pending( &ch->sessions[ c->session ].tls.sock );
336 60 : }
337 :
338 441 : FD_FN_PURE fd_failover_hello_t const * fd_failover_channel_peer_hello( fd_failover_channel_t const * ch ) { return &ch->peer_hello; }
339 444 : FD_FN_PURE ulong fd_failover_channel_generation( fd_failover_channel_t const * ch ) { return ch->generation; }
340 39 : FD_FN_PURE long fd_failover_channel_peer_heard_at( fd_failover_channel_t const * ch ) { return ch->paired_idx<0 ? 0L : ch->keepalive->ts_last_rx; }
341 :
342 : ushort
343 6 : fd_failover_channel_listen_port( fd_failover_channel_t const * ch ) {
344 6 : if( FD_UNLIKELY( ch->listen_fd==-1 ) ) return (ushort)0;
345 6 : struct sockaddr_in addr;
346 6 : socklen_t len = sizeof(addr);
347 6 : if( getsockname( ch->listen_fd, fd_type_pun( &addr ), &len ) ) FD_LOG_ERR(( "failover getsockname failed" ));
348 6 : return fd_ushort_bswap( addr.sin_port );
349 6 : }
350 :
351 : static void
352 : drop( fd_failover_channel_t * ch,
353 : ulong idx,
354 48 : long now ) {
355 48 : struct candidate * c = &ch->candidates[idx];
356 48 : int paired = ch->paired_idx==(int)idx;
357 48 : int accepted = c->fd!=-1 && !c->dialed;
358 48 : close_candidate( ch, c );
359 48 : if( paired ) {
360 39 : ch->paired_idx = -1;
361 39 : fd_memset( &ch->peer_hello, 0, sizeof(ch->peer_hello) );
362 39 : }
363 : /* A candidate from the listener does not get a session until it pairs. */
364 48 : if( !paired && accepted ) return;
365 : /* After loss, use the current handoff address to choose listener or dialer. */
366 42 : if( paired && want_dial( ch )!=ch->dial_peer ) reset( ch );
367 42 : if( ch->dial_peer ) {
368 : /* Recompute backoff parameters */
369 3 : int established = paired && now>=ch->paired_at && now-ch->paired_at>=FD_FAILOVER_CHANNEL_HELLO_TIMEOUT_NANOS;
370 3 : long jitter = (long)fd_rng_ulong_roll( ch->rng, (ulong)FD_FAILOVER_CHANNEL_BACKOFF_MIN_NANOS );
371 3 : ch->retry_at = fd_long_sat_add( now, fd_long_sat_add( established ? 0L : ch->backoff, jitter ) );
372 3 : ch->backoff = established ? FD_FAILOVER_CHANNEL_BACKOFF_MIN_NANOS
373 3 : : fd_long_min( fd_long_sat_add( ch->backoff, ch->backoff ), FD_FAILOVER_CHANNEL_BACKOFF_MAX_NANOS );
374 3 : }
375 42 : }
376 :
377 : /* Same as drop, and if idx is the paired session we log why it ended. */
378 : static void
379 : lose( fd_failover_channel_t * ch,
380 : ulong idx,
381 : long now,
382 0 : char const * why ) {
383 0 : if( ch->paired_idx==(int)idx ) {
384 0 : FD_LOG_WARNING(( "lost the failover session with `" FD_IP4_ADDR_FMT "`, %s",
385 0 : FD_IP4_ADDR_FMT_ARGS( ch->candidates[idx].address ), why ));
386 0 : } else if( ch->candidates[idx].dialed ) {
387 0 : FD_LOG_WARNING(( "the handshake with the failover peer at `" FD_IP4_ADDR_FMT "` ended, %s",
388 0 : FD_IP4_ADDR_FMT_ARGS( ch->candidates[idx].address ), why ));
389 0 : }
390 0 : drop( ch, idx, now );
391 0 : }
392 :
393 : void
394 : fd_failover_channel_hangup( fd_failover_channel_t * ch,
395 111 : long now ) {
396 111 : if( ch->paired_idx>=0 ) drop( ch, (ulong)ch->paired_idx, now );
397 113775 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) if( ch->candidates[i].fd!=-1 ) drop( ch, i, now );
398 111 : }
399 :
400 : void
401 : fd_failover_channel_protocol_error( fd_failover_channel_t * ch,
402 0 : long now ) {
403 0 : if( FD_LIKELY( ch->paired_idx>=0 ) ) {
404 0 : lose( ch, (ulong)ch->paired_idx, now, "it sent a message that breaks the protocol" );
405 0 : }
406 0 : fd_failover_channel_hangup( ch, now );
407 0 : }
408 :
409 : /* Admit at most one live inbound candidate per source address. */
410 : static int
411 : admit( fd_failover_channel_t const * ch,
412 6 : uint address ) {
413 6144 : for( ulong i=ADMITTED_IDX; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ )
414 6138 : if( ch->candidates[i].fd!=-1 && ch->candidates[i].address==address ) return 0;
415 6 : return 1;
416 6 : }
417 :
418 : static int
419 39 : has_proof( struct candidate const * c ) {
420 39 : return c->phase==PHASE_REQUEST_RECV && c->rx_used==sizeof(c->client_proof);
421 39 : }
422 :
423 : static int
424 27 : known_source( fd_failover_channel_t const * ch, uint address, long now ) {
425 27 : return address==ch->known_addr && now<ch->known_until;
426 27 : }
427 :
428 : static void
429 12 : remember_source( fd_failover_channel_t * ch, uint address, long now ) {
430 12 : ch->known_addr = address;
431 12 : ch->known_until = fd_long_sat_add( now, FD_FAILOVER_CHANNEL_IDLE_NANOS );
432 12 : }
433 :
434 : static ulong
435 : listen_slot( fd_failover_channel_t const * ch,
436 : uint address,
437 27 : long now ) {
438 27 : if( known_source( ch, address, now ) ) {
439 9 : return ch->candidates[ADMITTED_IDX].fd==-1 ? ADMITTED_IDX : FD_FAILOVER_CHANNEL_CANDIDATE_MAX;
440 9 : }
441 3084 : for( ulong i=UNAUTHENTICATED_IDX; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) {
442 3081 : if( ch->candidates[i].fd==-1 ) return i;
443 3081 : }
444 3 : return FD_FAILOVER_CHANNEL_CANDIDATE_MAX;
445 18 : }
446 :
447 : static void
448 : start( fd_failover_channel_t * ch,
449 : ulong idx,
450 : int fd,
451 : uint address,
452 : long now,
453 27 : int phase ) {
454 27 : struct candidate * c = &ch->candidates[idx];
455 : /* Inbound sockets do not acquire a TLS session until admission succeeds. */
456 27 : if( c->fd==-1 ) ch->live++;
457 27 : c->fd = fd;
458 27 : c->address = address;
459 27 : c->phase = phase;
460 27 : c->dialed = phase==PHASE_CONNECT;
461 27 : c->session = -1;
462 27 : c->deadline = fd_long_sat_add( now, FD_FAILOVER_CHANNEL_HELLO_TIMEOUT_NANOS );
463 27 : c->admission_until = c->dialed ? c->deadline : fd_long_min( c->deadline, fd_long_sat_add( now, FD_FAILOVER_ADMISSION_START_NANOS ) );
464 27 : c->local_hello = ch->self_hello;
465 27 : c->rx_used = c->tx_used = c->tx_sent = 0UL;
466 27 : }
467 :
468 : static void
469 : accept_candidates( fd_failover_channel_t * ch,
470 : long now,
471 120 : int * busy ) {
472 126 : for( ulong n=0; n<FD_FAILOVER_ACCEPT_PER_POLL; n++ ) {
473 126 : if( !fd_token_bucket_consume( &ch->accept_bucket, 1.0f, now ) ) return;
474 126 : struct sockaddr_in addr;
475 126 : socklen_t len = sizeof(addr);
476 126 : int fd = accept4( ch->listen_fd, fd_type_pun( &addr ), &len, SOCK_NONBLOCK|SOCK_CLOEXEC );
477 126 : if( fd==-1 ) return;
478 6 : *busy = 1;
479 6 : uint address = addr.sin_addr.s_addr;
480 6 : ulong idx = listen_slot( ch, address, now );
481 6 : if( ch->paired_idx>=0 || len!=sizeof(addr) || addr.sin_family!=AF_INET ||
482 6 : idx==FD_FAILOVER_CHANNEL_CANDIDATE_MAX || !admit( ch, address ) ) {
483 0 : close( fd );
484 0 : continue;
485 0 : }
486 6 : start( ch, idx, fd, address, now, PHASE_START_RECV );
487 6 : }
488 120 : }
489 :
490 : static int
491 : flush( fd_failover_channel_t * ch,
492 : ulong idx,
493 : long now,
494 96 : int * busy ) {
495 96 : struct candidate * c = &ch->candidates[idx];
496 96 : struct session * s = &ch->sessions[ c->session ];
497 96 : if( !c->tx_used ) {
498 66 : int rc = fd_tlsrec_sock_flush( &s->tls.sock, s->tls.fd );
499 66 : if( rc<0 ) { lose( ch, idx, now, "flushing TLS output failed" ); return -1; }
500 66 : return rc;
501 66 : }
502 30 : long n = fd_failover_tls_write( &s->tls, s->tx+c->tx_sent, c->tx_used-c->tx_sent );
503 30 : if( n<0L ) { lose( ch, idx, now, "writing to it failed" ); return -1; }
504 30 : if( !n ) return 1;
505 30 : *busy = 1;
506 30 : c->tx_sent += (ulong)n;
507 30 : if( c->tx_sent<c->tx_used ) return 1;
508 30 : c->tx_used = c->tx_sent = 0UL;
509 30 : return fd_tlsrec_sock_tx_pending( &s->tls.sock ) ? 1 : 0;
510 30 : }
511 :
512 : static int
513 : decode( fd_failover_channel_t * ch,
514 : ulong idx,
515 : long now,
516 : int * busy,
517 : ushort * type,
518 : uchar * payload,
519 138 : ulong * payload_sz ) {
520 138 : struct candidate * c = &ch->candidates[idx];
521 138 : uchar * rx = ch->sessions[ c->session ].rx;
522 138 : uchar const * data;
523 138 : ulong frame_sz;
524 138 : int err = fd_failover_wire_decode( rx, c->rx_used, type, &data, payload_sz, &frame_sz );
525 138 : if( err==FD_FAILOVER_WIRE_ERR_AGAIN ) return 0;
526 9 : if( err ) { lose( ch, idx, now, "it sent a malformed frame" ); return -1; }
527 9 : fd_memcpy( payload, data, *payload_sz );
528 9 : memmove( rx, rx+frame_sz, c->rx_used-frame_sz );
529 9 : c->rx_used -= frame_sz;
530 9 : *busy = 1;
531 9 : return 1;
532 9 : }
533 :
534 : static char const *
535 3 : hello_err_cstr( int err ) {
536 3 : switch( err ) {
537 0 : case HELLO_ERR_TLS_KEY: return "the key TLS authenticated is not the failover identity";
538 0 : case FD_FAILOVER_HELLO_ERR_VERSION: return "the protocol version differs";
539 3 : case FD_FAILOVER_HELLO_ERR_STAKED: return "the failover identity differs";
540 0 : case FD_FAILOVER_HELLO_ERR_SELF: return "its boot id is ours, this machine dialed itself";
541 0 : case FD_FAILOVER_HELLO_ERR_BOTH_HOLD: return "both machines run the identity";
542 0 : case FD_FAILOVER_HELLO_ERR_HOLDS: return "it sent an unknown value";
543 0 : case FD_FAILOVER_HELLO_ERR_BOOT_ID: return "the boot id is zero";
544 0 : case FD_FAILOVER_HELLO_ERR_MODE: return "the consensus mode differs";
545 0 : case FD_FAILOVER_HELLO_ERR_PROOF: return "its admission proof is not signed by the failover identity";
546 0 : default: return "unknown";
547 3 : }
548 3 : }
549 :
550 : static void
551 : refuse_hello( fd_failover_channel_t * ch,
552 : struct candidate * c,
553 : int err,
554 3 : long now ) {
555 3 : int proven = c->dialed || c->phase>=PHASE_TLS_START;
556 3 : if( !proven && now<ch->refused_log_at ) { ch->refused_cnt++; return; }
557 3 : FD_LOG_WARNING(( "rejected the failover peer at `" FD_IP4_ADDR_FMT "`, %s%s%lu%s",
558 3 : FD_IP4_ADDR_FMT_ARGS( c->address ), hello_err_cstr( err ),
559 3 : ch->refused_cnt ? ", and " : "", ch->refused_cnt, ch->refused_cnt ? " more unproven peers since the last such line" : "" ));
560 3 : if( !proven ) {
561 3 : ch->refused_cnt = 0UL;
562 3 : ch->refused_log_at = fd_long_sat_add( now, FD_FAILOVER_LOG_INTERVAL_NANOS );
563 3 : }
564 3 : }
565 :
566 : static int
567 : read_frame( fd_failover_channel_t * ch,
568 : ulong idx,
569 : long now,
570 : int * busy,
571 : ushort * type,
572 : uchar * payload,
573 69 : ulong * payload_sz ) {
574 69 : struct candidate * c = &ch->candidates[idx];
575 69 : struct session * s = &ch->sessions[ c->session ];
576 69 : int rc = decode( ch, idx, now, busy, type, payload, payload_sz );
577 69 : if( rc ) return rc;
578 69 : ulong max = ch->paired_idx==(int)idx ? sizeof(s->rx) : FD_FAILOVER_FRAME_HDR_SZ+sizeof(fd_failover_hello_t);
579 69 : if( c->rx_used>=max ) { drop( ch, idx, now ); return -1; }
580 69 : long n = fd_failover_tls_read( &s->tls, s->rx+c->rx_used, max-c->rx_used );
581 69 : if( n<0L ) {
582 0 : if( ch->paired_idx==(int)idx && ch->expected_close && s->tls.peer_closed && !fd_failover_channel_tx_pending( ch ) ) {
583 0 : drop( ch, idx, now );
584 0 : return -1;
585 0 : }
586 0 : if( fd_tlsrec_conn_is_failed( &s->tls.conn ) ) {
587 0 : char why[ 192 ];
588 0 : fd_cstr_printf( why, sizeof(why), NULL, "TLS record rejected (%s)",
589 0 : fd_tls_reason_cstr( s->tls.conn.hs.base.reason ) );
590 0 : lose( ch, idx, now, why );
591 0 : } else if( s->tls.peer_closed ) {
592 0 : if( ch->paired_idx==(int)idx ) {
593 0 : FD_LOG_NOTICE(( "the failover peer at `" FD_IP4_ADDR_FMT "` closed the session", FD_IP4_ADDR_FMT_ARGS( c->address ) ));
594 0 : } else if( c->dialed && c->phase==PHASE_HELLO ) {
595 0 : FD_LOG_WARNING(( "the failover peer at `" FD_IP4_ADDR_FMT "` closed the connection before answering our HELLO",
596 0 : FD_IP4_ADDR_FMT_ARGS( c->address ) ));
597 0 : }
598 0 : drop( ch, idx, now );
599 0 : } else {
600 0 : lose( ch, idx, now, "the connection closed" );
601 0 : }
602 0 : return -1;
603 0 : }
604 69 : if( n>0L ) { c->rx_used += (ulong)n; *busy = 1; }
605 69 : return decode( ch, idx, now, busy, type, payload, payload_sz );
606 69 : }
607 :
608 : /* Return one to continue this poll, zero to stop after a wait or failure. */
609 : static int
610 : connect_candidate( fd_failover_channel_t * ch,
611 : ulong idx,
612 12 : long now ) {
613 12 : struct candidate * c = &ch->candidates[idx];
614 12 : struct sockaddr_in addr = {
615 12 : .sin_family=AF_INET,
616 12 : .sin_port=fd_ushort_bswap( ch->peer_port ),
617 12 : .sin_addr.s_addr=ch->peer_addr
618 12 : };
619 12 : if( connect( c->fd, fd_type_pun( &addr ), sizeof(addr) ) && errno!=EISCONN ) {
620 6 : if( errno==EINPROGRESS || errno==EALREADY || errno==EINTR ) return 0; /* TCP is still connecting. */
621 0 : FD_LOG_WARNING(( "could not connect to the failover peer at `" FD_IP4_ADDR_FMT ":%hu` (%i-%s)",
622 0 : FD_IP4_ADDR_FMT_ARGS( ch->peer_addr ), ch->peer_port, errno, fd_io_strerror( errno ) ));
623 0 : drop( ch, idx, now );
624 0 : return 0;
625 6 : }
626 :
627 : /* Only operator-initiated outbound work prepares a TLS key share here. */
628 6 : ulong hello_sz = sizeof(c->client_hello);
629 6 : if( acquire_session( ch, idx, 1 )<=0 ) {
630 0 : drop( ch, idx, now );
631 0 : return 0;
632 0 : }
633 :
634 6 : if( fd_failover_tls_client_hello( &ch->sessions[ c->session ].tls, c->client_hello, &hello_sz ) ||
635 6 : hello_sz>sizeof(c->client_hello) ) {
636 0 : drop( ch, idx, now );
637 0 : return 0;
638 0 : }
639 :
640 6 : c->start = (fd_failover_admission_start_t){
641 6 : .magic=FD_FAILOVER_ADMISSION_MAGIC,
642 6 : .hello=c->local_hello,
643 6 : .client_hello_sz=(ushort)hello_sz
644 6 : };
645 6 : if( !fd_rng_secure( c->start.nonce, sizeof(c->start.nonce) ) ) {
646 0 : drop( ch, idx, now );
647 0 : return 0;
648 0 : }
649 6 : fd_memcpy( c->tx, &c->start, sizeof(c->start) );
650 6 : fd_memcpy( c->tx+sizeof(c->start), c->client_hello, hello_sz );
651 6 : c->tx_used = sizeof(c->start)+hello_sz;
652 6 : c->phase = PHASE_START_SEND;
653 6 : return 1;
654 6 : }
655 :
656 : static int
657 : handshake_candidate( fd_failover_channel_t * ch,
658 : ulong idx,
659 : long now,
660 12 : int * busy ) {
661 12 : struct candidate * c = &ch->candidates[idx];
662 12 : struct session * s = &ch->sessions[ c->session ];
663 12 : int rc = fd_failover_tls_handshake( &s->tls );
664 12 : if( rc<0 ) {
665 0 : uint reason = s->tls.conn.hs.base.reason;
666 0 : FD_LOG_WARNING(( "TLS handshake with `" FD_IP4_ADDR_FMT "` failed (%s)",
667 0 : FD_IP4_ADDR_FMT_ARGS( c->address ), reason ? fd_tls_reason_cstr( reason ) : "connection closed or socket error" ));
668 0 : drop( ch, idx, now );
669 0 : return 0;
670 0 : }
671 12 : if( !rc ) return 0; /* Keep waiting for TLS. */
672 6 : *busy = 1;
673 :
674 : /* TLS authenticated a key, send the HELLO! */
675 6 : c->phase = PHASE_HELLO;
676 6 : if( c->dialed ) {
677 3 : c->tx_used = fd_failover_wire_encode( s->tx, FD_FAILOVER_MSG_HELLO,
678 3 : (uchar const *)&c->local_hello, sizeof(c->local_hello) );
679 3 : }
680 6 : return 1;
681 12 : }
682 :
683 : static int
684 : read_candidate_hello( fd_failover_channel_t * ch,
685 : ulong idx,
686 : long now,
687 : int * busy,
688 9 : uchar * payload ) {
689 9 : struct candidate * c = &ch->candidates[idx];
690 9 : ushort type;
691 9 : ulong sz;
692 9 : if( read_frame( ch, idx, now, busy, &type, payload, &sz )<=0 ) return 0;
693 6 : if( type!=FD_FAILOVER_MSG_HELLO || sz!=sizeof(c->hello) ) {
694 0 : FD_LOG_WARNING(( "rejected failover HELLO from `" FD_IP4_ADDR_FMT "`", FD_IP4_ADDR_FMT_ARGS( c->address ) ));
695 0 : drop( ch, idx, now );
696 0 : return 0;
697 0 : }
698 :
699 : /* HELLO and TLS must match admission. */
700 6 : int err = fd_memeq( payload, &c->hello, sizeof(c->hello) ) &&
701 6 : fd_memeq( c->hello.staked_pubkey, ch->sessions[ c->session ].tls.peer_pubkey, 32UL )
702 6 : ? fd_failover_hello_check( &c->local_hello, &c->hello ) : HELLO_ERR_TLS_KEY;
703 6 : if( err!=FD_FAILOVER_HELLO_OK ) {
704 0 : refuse_hello( ch, c, err, now );
705 0 : drop( ch, idx, now );
706 0 : return 0;
707 0 : }
708 :
709 : /* The peer is authenticated. */
710 6 : c->phase = PHASE_READY;
711 6 : if( !c->dialed ) {
712 3 : c->tx_used = fd_failover_wire_encode( ch->sessions[ c->session ].tx, FD_FAILOVER_MSG_HELLO,
713 3 : (uchar const *)&c->local_hello, sizeof(c->local_hello) );
714 3 : if( flush( ch, idx, now, busy )!=0 ) return 0; /* Remain READY until output drains, unless the write fails. */
715 3 : }
716 6 : return 1;
717 6 : }
718 :
719 : static void
720 : pair_candidate( fd_failover_channel_t * ch,
721 : ulong idx,
722 : long now,
723 6 : int * busy ) {
724 6 : struct candidate * c = &ch->candidates[idx];
725 : /* The first session that authenticates wins. */
726 6150 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) if( i!=idx ) close_candidate( ch, &ch->candidates[i] );
727 :
728 6 : if( !c->dialed ) ch->dial_peer = 0;
729 :
730 6 : ch->paired_idx = (int)idx;
731 6 : ch->sessions[ c->session ].tls.paired = 1;
732 6 : ch->peer_hello = c->hello;
733 6 : ch->generation++;
734 6 : remember_source( ch, c->address, now );
735 :
736 6 : fd_keepalive_init( ch->keepalive, ch->rng, FD_FAILOVER_CHANNEL_KEEPALIVE_NANOS, FD_FAILOVER_CHANNEL_IDLE_NANOS, now );
737 6 : fd_keepalive_rx( ch->keepalive, now );
738 6 : ch->paired_at = now;
739 6 : ch->expected_close = 0;
740 6 : *busy = 1;
741 :
742 6 : FD_LOG_NOTICE(( "paired with the failover peer at `" FD_IP4_ADDR_FMT "` via %s, %s, boot id %016lx",
743 6 : FD_IP4_ADDR_FMT_ARGS( c->address ), c->dialed ? "outbound dial" : "listener",
744 6 : c->hello.holds ? "it runs the identity" : "it does not run the identity", c->hello.boot_id ));
745 6 : }
746 :
747 : static int
748 51 : read_admission( struct candidate * c, ulong goal, int * busy ) {
749 51 : if( FD_UNLIKELY( goal>FD_FAILOVER_ADMISSION_MAX || c->rx_used>goal ) ) return -1;
750 51 : if( c->rx_used==goal ) return 1;
751 51 : long n = recv( c->fd, c->rx+c->rx_used, goal-c->rx_used, MSG_DONTWAIT|MSG_NOSIGNAL );
752 51 : if( n<0L ) return ( errno==EAGAIN || errno==EWOULDBLOCK || errno==EINTR ) ? 0 : -1;
753 21 : if( !n ) return -1;
754 :
755 21 : c->rx_used += (ulong)n;
756 21 : *busy = 1;
757 21 : return c->rx_used==goal;
758 21 : }
759 :
760 : static int
761 15 : write_admission( struct candidate * c, int * busy ) {
762 15 : long n = send( c->fd, c->tx+c->tx_sent, c->tx_used-c->tx_sent, MSG_DONTWAIT|MSG_NOSIGNAL );
763 15 : if( n<0L ) return ( errno==EAGAIN || errno==EWOULDBLOCK || errno==EINTR ) ? 0 : -1;
764 15 : if( !n ) return -1;
765 15 : c->tx_sent += (ulong)n;
766 15 : *busy = 1;
767 15 : if( c->tx_sent<c->tx_used ) return 0;
768 15 : c->tx_sent = c->tx_used = 0UL;
769 15 : return 1;
770 15 : }
771 :
772 : static int
773 6 : verify_admission_proof( fd_failover_channel_t * ch, struct candidate * c, int server, uchar const sig[ 64 ] ) {
774 6 : uchar msg[ FD_KEYGUARD_FAILOVER_PROOF_SZ ];
775 6 : fd_failover_proof_msg( msg, server, c->transcript );
776 6 : return fd_ed25519_verify( msg, sizeof(msg), sig, c->local_hello.staked_pubkey, ch->sha )==FD_ED25519_SUCCESS;
777 6 : }
778 :
779 : static int
780 6 : sign_admission_proof( fd_failover_channel_t * ch, struct candidate * c, int server ) {
781 6 : uchar msg[ FD_KEYGUARD_FAILOVER_PROOF_SZ ];
782 6 : fd_failover_proof_msg( msg, server, c->transcript );
783 6 : ch->proof_signer.sign_fn( ch->proof_signer.ctx, c->tx, msg );
784 6 : if( fd_ed25519_verify( msg, sizeof(msg), c->tx, c->local_hello.staked_pubkey, ch->sha )!=FD_ED25519_SUCCESS ) return 0;
785 6 : c->tx_used = 64UL;
786 6 : c->tx_sent = 0UL;
787 6 : return 1;
788 6 : }
789 :
790 : /* One Ed25519 operation per poll, within the global rate. */
791 : static int
792 12 : verify_credit( fd_failover_channel_t * ch, long now ) {
793 12 : if( !ch->verify_budget || !fd_token_bucket_consume( &ch->verify_bucket, 1.0f, now ) ) return 0;
794 12 : ch->verify_budget--;
795 12 : return 1;
796 12 : }
797 :
798 : /* Service the candidate admission state machine. */
799 : static void
800 75 : service_admission( fd_failover_channel_t * ch, ulong idx, long now, int * busy ) {
801 75 : struct candidate * c = &ch->candidates[idx];
802 75 : int rc = 1;
803 75 : int refused = 0;
804 75 : char const * why = NULL;
805 75 : switch( c->phase ) {
806 6 : case PHASE_START_SEND:
807 9 : case PHASE_CHALLENGE_SEND:
808 12 : case PHASE_REQUEST_SEND:
809 15 : case PHASE_RESPONSE_SEND:
810 15 : rc = write_admission( c, busy );
811 15 : if( rc<0 ) break;
812 15 : if( !rc ) return;
813 15 : c->rx_used = 0UL;
814 15 : if( c->phase==PHASE_START_SEND ) {
815 6 : c->phase = PHASE_CHALLENGE_RECV;
816 6 : }
817 9 : else if( c->phase==PHASE_CHALLENGE_SEND ) {
818 3 : c->admission_until = fd_long_min( c->deadline, fd_long_sat_add( now, FD_FAILOVER_ADMISSION_PROOF_NANOS ) );
819 3 : c->phase = PHASE_REQUEST_RECV;
820 3 : }
821 6 : else if( c->phase==PHASE_REQUEST_SEND ) {
822 3 : c->phase = PHASE_RESPONSE_RECV;
823 3 : }
824 3 : else {
825 3 : c->phase = PHASE_TLS;
826 3 : }
827 15 : return;
828 9 : case PHASE_START_RECV: {
829 : /* Read only the fixed header until its length is checked. */
830 9 : ulong goal = c->rx_used<sizeof(c->start) ? sizeof(c->start) : sizeof(c->start)+c->start.client_hello_sz;
831 9 : rc = read_admission( c, goal, busy );
832 9 : if( rc<0 ) break;
833 9 : if( !rc ) return;
834 9 : if( goal==sizeof(c->start) ) {
835 6 : fd_memcpy( &c->start, c->rx, sizeof(c->start) );
836 6 : if( c->start.magic!=FD_FAILOVER_ADMISSION_MAGIC ||
837 6 : c->start.client_hello_sz<=sizeof(fd_tlsrec_hdr_t) ||
838 6 : c->start.client_hello_sz>sizeof(c->client_hello) ) break;
839 6 : int err = fd_failover_hello_check( &c->local_hello, &c->start.hello );
840 6 : if( err ) { refuse_hello( ch, c, err, now ); refused = 1; break; }
841 3 : return;
842 6 : }
843 3 : fd_memcpy( c->client_hello, c->rx+sizeof(c->start), c->start.client_hello_sz );
844 3 : fd_tlsrec_hdr_t hdr;
845 3 : fd_memcpy( &hdr, c->client_hello, sizeof(hdr) );
846 3 : if( hdr.content_type!=FD_TLS_REC_HANDSHAKE ||
847 3 : (ulong)fd_ushort_bswap( hdr.length )+sizeof(hdr)!=c->start.client_hello_sz ||
848 3 : c->client_hello[ sizeof(hdr) ]!=FD_TLS_MSG_CLIENT_HELLO ) break;
849 3 : c->hello = c->start.hello;
850 3 : c->challenge = (fd_failover_admission_challenge_t){ .magic=FD_FAILOVER_ADMISSION_MAGIC, .hello=c->local_hello };
851 3 : if( !fd_rng_secure( c->challenge.nonce, sizeof(c->challenge.nonce) ) ) break;
852 3 : fd_failover_admission_hash( &c->start, c->client_hello, &c->challenge, c->transcript );
853 3 : fd_memcpy( c->tx, &c->challenge, sizeof(c->challenge) );
854 3 : c->tx_used = sizeof(c->challenge);
855 3 : c->rx_used = 0UL;
856 3 : c->admission_until = fd_long_min( c->deadline, fd_long_sat_add( now, FD_FAILOVER_ADMISSION_START_NANOS ) );
857 3 : c->phase = PHASE_CHALLENGE_SEND;
858 3 : return;
859 3 : }
860 18 : case PHASE_CHALLENGE_RECV:
861 18 : rc = read_admission( c, sizeof(c->challenge), busy );
862 18 : if( rc<0 ) break;
863 15 : if( !rc ) return;
864 3 : fd_memcpy( &c->challenge, c->rx, sizeof(c->challenge) );
865 3 : if( c->challenge.magic!=FD_FAILOVER_ADMISSION_MAGIC ) break;
866 3 : int err = fd_failover_hello_check( &c->local_hello, &c->challenge.hello );
867 3 : if( err ) { refuse_hello( ch, c, err, now ); refused = 1; break; }
868 3 : c->hello = c->challenge.hello;
869 3 : fd_failover_admission_hash( &c->start, c->client_hello, &c->challenge, c->transcript );
870 3 : c->rx_used = 0UL;
871 3 : c->phase = PHASE_REQUEST_SIGN;
872 3 : *busy = 1;
873 3 : return;
874 3 : case PHASE_REQUEST_SIGN:
875 3 : if( !verify_credit( ch, now ) ) return;
876 3 : if( !sign_admission_proof( ch, c, 0 ) ) { why = "this machine could not sign the admission proof"; break; }
877 3 : fd_memcpy( c->client_proof, c->tx, sizeof(c->client_proof) );
878 3 : c->phase = PHASE_REQUEST_SEND;
879 3 : *busy = 1;
880 3 : return;
881 15 : case PHASE_RESPONSE_RECV: {
882 15 : fd_failover_admission_response_t response;
883 15 : if( c->rx_used<sizeof(response) ) {
884 12 : rc = read_admission( c, sizeof(response), busy );
885 12 : if( rc<0 ) break;
886 12 : if( !rc ) return;
887 3 : fd_memcpy( &response, c->rx, sizeof(response) );
888 3 : if( !response.server_flight_sz || response.server_flight_sz>FD_FAILOVER_SERVER_FLIGHT_MAX ) break;
889 3 : return;
890 3 : }
891 3 : fd_memcpy( &response, c->rx, sizeof(response) );
892 3 : rc = read_admission( c, sizeof(response)+response.server_flight_sz, busy );
893 3 : if( rc<0 ) break;
894 3 : if( !rc || !verify_credit( ch, now ) ) return;
895 3 : fd_failover_response_hash( c->transcript, c->client_proof, c->rx+sizeof(response),
896 3 : response.server_flight_sz, c->transcript );
897 3 : if( !verify_admission_proof( ch, c, 1, response.proof ) ) {
898 0 : refuse_hello( ch, c, FD_FAILOVER_HELLO_ERR_PROOF, now );
899 0 : refused = 1;
900 0 : break;
901 0 : }
902 3 : if( fd_failover_tls_accept_flight( &ch->sessions[ c->session ].tls, c->rx+sizeof(response), response.server_flight_sz ) ) { why = "its TLS server flight was rejected"; break; }
903 3 : c->phase = PHASE_TLS;
904 3 : c->rx_used = 0UL;
905 3 : *busy = 1;
906 3 : return;
907 3 : }
908 9 : case PHASE_REQUEST_RECV:
909 9 : rc = read_admission( c, 64UL, busy );
910 9 : if( rc<0 ) break;
911 9 : if( !rc || !verify_credit( ch, now ) ) return;
912 3 : if( !verify_admission_proof( ch, c, 0, c->rx ) ) {
913 0 : refuse_hello( ch, c, FD_FAILOVER_HELLO_ERR_PROOF, now );
914 0 : refused = 1;
915 0 : break;
916 0 : }
917 3 : fd_memcpy( c->client_proof, c->rx, sizeof(c->client_proof) );
918 3 : c->rx_used = 0UL;
919 3 : c->admission_until = c->deadline;
920 3 : c->phase = PHASE_TLS_START;
921 3 : *busy = 1;
922 3 : return;
923 3 : case PHASE_RESPONSE_SIGN: {
924 3 : if( !verify_credit( ch, now ) ) return;
925 3 : fd_failover_admission_response_t response = { .server_flight_sz=(ushort)(c->tx_used-sizeof(fd_failover_admission_response_t)) };
926 3 : if( !sign_admission_proof( ch, c, 1 ) ) break;
927 3 : fd_memcpy( response.proof, c->tx, sizeof(response.proof) );
928 3 : fd_memcpy( c->tx, &response, sizeof(response) );
929 3 : c->tx_used = sizeof(response)+response.server_flight_sz;
930 3 : c->phase = PHASE_RESPONSE_SEND;
931 3 : *busy = 1;
932 3 : return;
933 3 : }
934 3 : case PHASE_TLS_START: {
935 3 : int got = acquire_session( ch, idx, 0 );
936 3 : if( !got ) return;
937 3 : if( got<0 ) break;
938 3 : struct session * s = &ch->sessions[ c->session ];
939 3 : if( fd_failover_tls_accept_flight( &s->tls, c->client_hello, c->start.client_hello_sz ) ) break;
940 3 : if( !s->tls.sock.tx_sz || s->tls.sock.tx_sz>FD_FAILOVER_SERVER_FLIGHT_MAX ) break;
941 3 : fd_failover_response_hash( c->transcript, c->client_proof, s->tls.sock.tx_buf,
942 3 : (ushort)s->tls.sock.tx_sz, c->transcript );
943 3 : fd_memcpy( c->tx+sizeof(fd_failover_admission_response_t), s->tls.sock.tx_buf, s->tls.sock.tx_sz );
944 3 : c->tx_used = sizeof(fd_failover_admission_response_t)+s->tls.sock.tx_sz;
945 3 : s->tls.sock.tx_off = s->tls.sock.tx_sz = 0UL;
946 3 : c->phase = PHASE_RESPONSE_SIGN;
947 3 : *busy = 1;
948 3 : return;
949 3 : }
950 0 : default:
951 0 : break;
952 75 : }
953 6 : if( c->dialed && !refused ) {
954 3 : if( why ) {
955 0 : FD_LOG_WARNING(( "the failover peer at `" FD_IP4_ADDR_FMT "` did not complete admission, %s", FD_IP4_ADDR_FMT_ARGS( c->address ), why ));
956 3 : } else if( rc<0 ) {
957 3 : FD_LOG_WARNING(( "the failover peer at `" FD_IP4_ADDR_FMT "` did not complete admission, the connection closed or failed while we waited for %s",
958 3 : FD_IP4_ADDR_FMT_ARGS( c->address ), awaited_cstr( c->phase ) ));
959 3 : } else {
960 0 : FD_LOG_WARNING(( "the failover peer at `" FD_IP4_ADDR_FMT "` did not complete admission, it sent an invalid message while we waited for %s",
961 0 : FD_IP4_ADDR_FMT_ARGS( c->address ), awaited_cstr( c->phase ) ));
962 0 : }
963 3 : }
964 6 : drop( ch, idx, now );
965 6 : }
966 :
967 : /* Drive the candidate lifecycle state machine. */
968 : static void
969 : service_candidate( fd_failover_channel_t * ch,
970 : ulong idx,
971 : long now,
972 : int * busy,
973 102 : uchar * payload ) {
974 102 : struct candidate * c = &ch->candidates[idx];
975 102 : if( c->phase==PHASE_CONNECT ) {
976 12 : if( connect_candidate( ch, idx, now ) ) *busy = 1;
977 12 : return;
978 12 : }
979 90 : if( c->phase<PHASE_TLS ) {
980 75 : service_admission( ch, idx, now, busy );
981 75 : return;
982 75 : }
983 15 : if( c->phase==PHASE_TLS ) {
984 12 : if( !handshake_candidate( ch, idx, now, busy ) ) {
985 6 : return;
986 6 : }
987 12 : }
988 9 : if( flush( ch, idx, now, busy )!=0 ) return;
989 9 : if( c->phase==PHASE_HELLO ) {
990 9 : if( !read_candidate_hello( ch, idx, now, busy, payload ) ) {
991 3 : return;
992 3 : }
993 9 : }
994 6 : if( c->phase==PHASE_READY && !c->tx_used ) {
995 6 : pair_candidate( ch, idx, now, busy );
996 6 : }
997 6 : }
998 :
999 : /* The paired session went silent. */
1000 : static int
1001 57 : silent( fd_failover_channel_t const * ch, long now ) {
1002 57 : long last_rx = ch->keepalive->ts_last_rx;
1003 57 : return now>=last_rx && now-last_rx>FD_FAILOVER_CHANNEL_IDLE_NANOS;
1004 57 : }
1005 :
1006 : /* Services the paired session. Returns 1 when a post-HELLO frame was delivered. */
1007 : static int
1008 : service_paired( fd_failover_channel_t * ch,
1009 : long now,
1010 : int * busy,
1011 : ushort * type,
1012 : uchar * payload,
1013 60 : ulong * payload_sz ) {
1014 60 : ulong idx = (ulong)ch->paired_idx;
1015 60 : int delivered = 0;
1016 60 : int flushed = flush( ch, idx, now, busy );
1017 60 : if( !flushed ) {
1018 60 : int rc = read_frame( ch, idx, now, busy, type, payload, payload_sz );
1019 60 : if( rc>0 ) {
1020 3 : if( *type==FD_FAILOVER_MSG_HELLO ) {
1021 0 : lose( ch, idx, now, "it sent a second HELLO" );
1022 0 : }
1023 3 : else {
1024 3 : fd_keepalive_rx( ch->keepalive, now );
1025 3 : remember_source( ch, ch->candidates[idx].address, now );
1026 3 : delivered = *type!=FD_FAILOVER_MSG_KEEPALIVE;
1027 3 : }
1028 57 : } else if( !rc && silent( ch, now ) ) {
1029 0 : lose( ch, idx, now, "it was idle past its deadline" );
1030 0 : }
1031 60 : }
1032 60 : if( flushed>0 && silent( ch, now ) ) {
1033 0 : lose( ch, idx, now, "it stopped making progress" );
1034 0 : }
1035 60 : if( ch->paired_idx>=0 &&
1036 60 : !fd_failover_channel_tx_pending( ch ) &&
1037 60 : fd_keepalive_should_tx( ch->keepalive, now ) )
1038 6 : {
1039 6 : if( !fd_failover_channel_send( ch, now, FD_FAILOVER_MSG_KEEPALIVE, NULL, 0UL ) ) {
1040 6 : fd_keepalive_tx( ch->keepalive, ch->rng, now );
1041 6 : *busy = 1;
1042 6 : }
1043 6 : }
1044 60 : return delivered;
1045 60 : }
1046 :
1047 : static void
1048 : dial( fd_failover_channel_t * ch,
1049 : long now,
1050 126 : int * busy ) {
1051 126 : if( !ch->dial_peer || !ch->peer_addr || ch->candidates[OUTBOUND_IDX].fd!=-1 || now<ch->retry_at ) {
1052 120 : return;
1053 120 : }
1054 :
1055 6 : *busy = 1;
1056 6 : int fd = socket( AF_INET, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0 );
1057 6 : if( fd==-1 ) {
1058 0 : drop( ch, OUTBOUND_IDX, now );
1059 0 : return;
1060 0 : }
1061 :
1062 6 : int yes = 1;
1063 6 : if( setsockopt( fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes) ) ) {
1064 0 : close( fd );
1065 0 : drop( ch, OUTBOUND_IDX, now );
1066 0 : return;
1067 0 : }
1068 :
1069 6 : start( ch, OUTBOUND_IDX, fd, ch->peer_addr, now, PHASE_CONNECT );
1070 6 : }
1071 :
1072 : int
1073 : fd_failover_channel_poll( fd_failover_channel_t * ch,
1074 : long now,
1075 : int * busy,
1076 : ushort * type,
1077 : uchar * payload,
1078 192 : ulong * payload_sz ) {
1079 192 : int delivered = 0;
1080 192 : ch->verify_budget = 1;
1081 192 : if( ch->paired_idx>=0 ) {
1082 60 : delivered = service_paired( ch, now, busy, type, payload, payload_sz );
1083 60 : }
1084 :
1085 : /* Expire every slot each poll, independently of round-robin service. */
1086 192 : if( ch->live > (ulong)( ch->paired_idx>=0 ) ) {
1087 98400 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) {
1088 98304 : struct candidate * c = &ch->candidates[i];
1089 98304 : if( c->fd!=-1 && (int)i!=ch->paired_idx && ( now>=c->deadline || now>=c->admission_until ) ) {
1090 0 : int dialed = c->dialed;
1091 0 : int phase = c->phase;
1092 0 : uint address = c->address;
1093 0 : drop( ch, i, now );
1094 0 : *busy = 1;
1095 0 : if( dialed ) {
1096 0 : long tenths = ch->retry_at>now ? (ch->retry_at-now)/((long)FD_FAILOVER_NANOS_PER_SECOND/10L) : 0L;
1097 0 : FD_LOG_WARNING(( "the failover peer at `" FD_IP4_ADDR_FMT "` did not finish the handshake within %ld s, it was still waiting for %s, retrying in %ld.%ld s",
1098 0 : FD_IP4_ADDR_FMT_ARGS( address ), FD_FAILOVER_CHANNEL_HELLO_TIMEOUT_NANOS/(long)FD_FAILOVER_NANOS_PER_SECOND,
1099 0 : awaited_cstr( phase ), tenths/10L, tenths%10L ));
1100 0 : }
1101 0 : }
1102 98304 : }
1103 96 : }
1104 :
1105 192 : if( FD_UNLIKELY( !ch->tls_ctx.ready ) || ch->paired_idx>=0 ) {
1106 66 : return delivered;
1107 66 : }
1108 126 : dial( ch, now, busy );
1109 :
1110 126 : ulong serviced = 0UL;
1111 372 : for( ulong idx=0UL; idx<UNAUTHENTICATED_IDX; idx++ ) {
1112 249 : if( ch->candidates[idx].fd==-1 ) continue;
1113 63 : service_candidate( ch, idx, now, busy, payload );
1114 63 : serviced++;
1115 63 : if( ch->paired_idx>=0 ) return delivered;
1116 63 : }
1117 :
1118 : /* Round-robin the shared pool. */
1119 123 : ulong reserved = (ulong)( ch->candidates[OUTBOUND_IDX].fd!=-1 ) + (ulong)( ch->candidates[ADMITTED_IDX].fd!=-1 );
1120 33852 : for( ulong n=0UL; ch->live>reserved && n<FD_FAILOVER_CHANNEL_CANDIDATE_MAX-UNAUTHENTICATED_IDX && serviced<FD_FAILOVER_SERVICE_PER_POLL; n++ ) {
1121 33732 : ulong idx = UNAUTHENTICATED_IDX+ch->cursor;
1122 33732 : ch->cursor = (ch->cursor+1UL)%(FD_FAILOVER_CHANNEL_CANDIDATE_MAX-UNAUTHENTICATED_IDX);
1123 33732 : if( ch->candidates[idx].fd==-1 ) continue;
1124 39 : if( has_proof( &ch->candidates[idx] ) && !ch->verify_budget ) continue;
1125 :
1126 39 : service_candidate( ch, idx, now, busy, payload );
1127 39 : serviced++;
1128 39 : if( ch->paired_idx>=0 ) {
1129 3 : return delivered;
1130 3 : }
1131 39 : }
1132 120 : if( ch->listen_fd!=-1 ) {
1133 120 : accept_candidates( ch, now, busy );
1134 120 : }
1135 120 : return delivered;
1136 123 : }
1137 :
1138 : int
1139 : fd_failover_channel_send( fd_failover_channel_t * ch,
1140 : long now,
1141 : ushort type,
1142 : uchar const * payload,
1143 24 : ulong sz ) {
1144 24 : if( ch->paired_idx<0 ) return -1;
1145 :
1146 24 : ulong idx = (ulong)ch->paired_idx;
1147 24 : struct candidate * c = &ch->candidates[idx];
1148 :
1149 24 : if( c->tx_used ) return -1;
1150 24 : c->tx_used = fd_failover_wire_encode( ch->sessions[ c->session ].tx, type, payload, sz );
1151 24 : if( !c->tx_used ) {
1152 0 : lose( ch, idx, now, "we could not encode a frame for it" );
1153 0 : return -1;
1154 0 : }
1155 :
1156 24 : int busy = 0;
1157 24 : int rc = flush( ch, idx, now, &busy )<0 ? -1 : 0;
1158 24 : if( !rc && type==FD_FAILOVER_MSG_HANDOFF_RESULT ) {
1159 0 : ch->expected_close = 1;
1160 0 : }
1161 24 : return rc;
1162 24 : }
|
