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 0 : #define OUTBOUND_IDX (0UL)
20 0 : #define ADMITTED_IDX (1UL)
21 0 : #define UNAUTHENTICATED_IDX (2UL)
22 :
23 : /* TLS admission phase FSM states. */
24 0 : #define PHASE_CONNECT (0)
25 0 : #define PHASE_START_SEND (1)
26 0 : #define PHASE_START_RECV (2)
27 0 : #define PHASE_CHALLENGE_SEND (3)
28 0 : #define PHASE_CHALLENGE_RECV (4)
29 0 : #define PHASE_REQUEST_SIGN (5)
30 0 : #define PHASE_REQUEST_SEND (6)
31 0 : #define PHASE_REQUEST_RECV (7)
32 0 : #define PHASE_RESPONSE_SIGN (8)
33 0 : #define PHASE_RESPONSE_SEND (9)
34 0 : #define PHASE_RESPONSE_RECV (10)
35 0 : #define PHASE_TLS_START (11)
36 0 : #define PHASE_TLS (12)
37 0 : #define PHASE_HELLO (13)
38 0 : #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 0 : awaited_cstr( int phase ) {
54 0 : switch( phase ) {
55 0 : case PHASE_CONNECT: return "its TCP answer";
56 0 : case PHASE_START_SEND: return "our admission start to drain";
57 0 : 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 0 : }
64 0 : }
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 0 : FD_FN_CONST ulong fd_failover_channel_align ( void ) { return alignof(fd_failover_channel_t); }
140 0 : 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 0 : ulong burst ) {
146 0 : return (fd_token_bucket_t){ .rate=(float)rate/(float)FD_FAILOVER_NANOS_PER_SECOND, .burst=(float)burst, .balance=(float)burst };
147 0 : }
148 :
149 : void *
150 0 : fd_failover_channel_new( void * shmem ) {
151 0 : if( !shmem || !fd_ulong_is_aligned( (ulong)shmem, fd_failover_channel_align() ) ) return NULL;
152 0 : fd_failover_channel_t * ch = shmem;
153 :
154 : /* Initialize the listening state. */
155 0 : fd_memset( ch, 0, sizeof(*ch) );
156 0 : ch->listen_fd = -1;
157 0 : ch->paired_idx = -1;
158 0 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) { ch->candidates[i].fd = -1; ch->candidates[i].session = -1; }
159 0 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_SESSION_MAX; i++ ) ch->sessions[i].owner = -1;
160 0 : ch->backoff = FD_FAILOVER_CHANNEL_BACKOFF_MIN_NANOS;
161 :
162 : /* Rate limiting */
163 0 : ch->accept_bucket = token_bucket_new( FD_FAILOVER_ACCEPT_RATE, FD_FAILOVER_ACCEPT_BURST );
164 0 : ch->verify_bucket = token_bucket_new( FD_FAILOVER_VERIFY_RATE, FD_FAILOVER_VERIFY_BURST );
165 :
166 : /* Set up secure randomness */
167 0 : uint seed;
168 0 : FD_TEST( fd_rng_secure( &seed, sizeof(seed) ) );
169 0 : FD_TEST( fd_rng_join( fd_rng_new( ch->rng, seed, 0UL ) ) );
170 0 : FD_TEST( fd_sha512_join( fd_sha512_new( ch->sha ) ) );
171 :
172 0 : ch->magic = FD_FAILOVER_CHANNEL_MAGIC;
173 0 : return shmem;
174 0 : }
175 :
176 : fd_failover_channel_t *
177 0 : fd_failover_channel_join( void * shmem ) {
178 0 : if( !shmem || !fd_ulong_is_aligned( (ulong)shmem, fd_failover_channel_align() ) ) return NULL;
179 0 : fd_failover_channel_t * ch = shmem;
180 0 : return ch->magic==FD_FAILOVER_CHANNEL_MAGIC ? ch : NULL;
181 0 : }
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 0 : acquire_session( fd_failover_channel_t * ch, ulong idx, int dial ) {
188 0 : struct candidate * c = &ch->candidates[idx];
189 0 : if( c->session!=-1 ) return 1;
190 0 : ulong lo = idx==OUTBOUND_IDX ? OUTBOUND_IDX : OUTBOUND_IDX+1UL;
191 0 : ulong hi = idx==OUTBOUND_IDX ? OUTBOUND_IDX+1UL : FD_FAILOVER_CHANNEL_SESSION_MAX;
192 0 : for( ulong i=lo; i<hi; i++ ) {
193 0 : if( ch->sessions[i].owner!=-1 ) continue;
194 0 : if( fd_failover_tls_new( &ch->sessions[i].tls, &ch->tls_ctx, c->fd, dial ) ) return -1;
195 0 : ch->sessions[i].owner = (int)idx;
196 0 : c->session = (int)i;
197 0 : return 1;
198 0 : }
199 0 : return 0;
200 0 : }
201 :
202 : static void
203 0 : close_candidate( fd_failover_channel_t * ch, struct candidate * c ) {
204 0 : if( c->session!=-1 ) {
205 0 : fd_failover_tls_fini( &ch->sessions[ c->session ].tls );
206 0 : ch->sessions[ c->session ].owner = -1;
207 0 : c->session = -1;
208 0 : }
209 : /* Closing a candidate frees its slot. */
210 0 : if( c->fd!=-1 ) { close( c->fd ); c->fd = -1; ch->live--; }
211 0 : c->dialed = 0;
212 0 : c->rx_used = c->tx_used = c->tx_sent = 0UL;
213 0 : }
214 :
215 : /* Discards every candidate and the pairing. */
216 : static void
217 0 : discard( fd_failover_channel_t * ch ) {
218 0 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) close_candidate( ch, &ch->candidates[i] );
219 0 : ch->paired_idx = -1;
220 0 : fd_memset( &ch->peer_hello, 0, sizeof(ch->peer_hello) );
221 0 : }
222 :
223 : static int
224 0 : want_dial( fd_failover_channel_t const * ch ) {
225 0 : return !!ch->peer_addr;
226 0 : }
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 0 : reset( fd_failover_channel_t * ch ) {
232 0 : if( ch->paired_idx>=0 ) return;
233 0 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) {
234 0 : if( ch->candidates[i].dialed ) {
235 0 : close_candidate( ch, &ch->candidates[i] );
236 0 : }
237 0 : }
238 0 : ch->dial_peer = want_dial( ch );
239 0 : ch->retry_at = 0L;
240 0 : }
241 :
242 : void
243 : fd_failover_channel_init_listener( fd_failover_channel_t * ch,
244 : uint address,
245 0 : ushort port ) {
246 0 : discard( ch );
247 0 : reset( ch );
248 0 : if( ch->listen_fd!=-1 ) return;
249 0 : ch->listen_fd = socket( AF_INET, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0 );
250 0 : if( ch->listen_fd==-1 ) {
251 0 : FD_LOG_ERR(( "failover socket failed (%i-%s)", errno, fd_io_strerror( errno ) ));
252 0 : }
253 0 : int yes = 1;
254 0 : if( setsockopt( ch->listen_fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes) ) ) {
255 0 : FD_LOG_ERR(( "failover setsockopt failed" ));
256 0 : }
257 0 : 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 0 : int defer = (int)(FD_FAILOVER_ADMISSION_START_NANOS/(long)FD_FAILOVER_NANOS_PER_SECOND);
261 0 : 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 0 : struct sockaddr_in addr = { .sin_family=AF_INET, .sin_port=fd_ushort_bswap( port ), .sin_addr.s_addr=address };
265 0 : 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 0 : }
270 :
271 : void
272 : fd_failover_channel_init_dialer( fd_failover_channel_t * ch,
273 : uint address,
274 0 : ushort port ) {
275 0 : if( address==ch->peer_addr && port==ch->peer_port ) return;
276 0 : if( address && ch->paired_idx>=0 ) discard( ch );
277 0 : ch->peer_addr = address;
278 0 : ch->peer_port = port;
279 0 : if( address ) {
280 0 : ch->backoff = FD_FAILOVER_CHANNEL_BACKOFF_MIN_NANOS;
281 0 : }
282 0 : reset( ch );
283 0 : }
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 0 : fd_failover_hello_t const * hello ) {
290 0 : if( !proof_signer.sign_fn || !signer.sign_fn ) return -1;
291 0 : discard( ch );
292 : /* An identity change requires a fresh session. */
293 0 : ch->known_addr = 0U;
294 0 : ch->known_until = 0L;
295 0 : ch->self_hello = *hello;
296 0 : ch->proof_signer = proof_signer;
297 0 : return fd_failover_tls_ctx_init( &ch->tls_ctx, hello->staked_pubkey, signer );
298 0 : }
299 :
300 : void
301 0 : fd_failover_channel_fini( fd_failover_channel_t * ch ) {
302 0 : discard( ch );
303 0 : if( ch->listen_fd!=-1 ) close( ch->listen_fd );
304 0 : ch->listen_fd = -1;
305 0 : fd_failover_tls_ctx_fini( &ch->tls_ctx );
306 0 : }
307 :
308 : void
309 : fd_failover_channel_set_holds( fd_failover_channel_t * ch,
310 0 : int holds ) {
311 0 : if( ch->self_hello.holds==(uchar)!!holds ) return;
312 0 : ch->self_hello.holds = (uchar)!!holds;
313 0 : for( ulong i=0UL; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ )
314 0 : if( (int)i!=ch->paired_idx ) close_candidate( ch, &ch->candidates[i] );
315 0 : if( ch->paired_idx<0 ) reset( ch );
316 0 : }
317 :
318 : /* Derive the session state from the socket state. */
319 : FD_FN_PURE ulong
320 0 : fd_failover_channel_state( fd_failover_channel_t const * ch ) {
321 0 : if( ch->paired_idx>=0 ) return FD_FAILOVER_SESSION_PAIRED;
322 0 : if( !ch->dial_peer ) return FD_FAILOVER_SESSION_LISTENING;
323 0 : struct candidate const * c = &ch->candidates[OUTBOUND_IDX];
324 0 : if( c->fd==-1 ) return FD_FAILOVER_SESSION_BACKOFF;
325 0 : if( c->phase==PHASE_CONNECT ) return FD_FAILOVER_SESSION_DIALING;
326 0 : return FD_FAILOVER_SESSION_HELLO;
327 0 : }
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 0 : fd_failover_channel_tx_pending( fd_failover_channel_t const * ch ) {
333 0 : if( ch->paired_idx<0 ) return 0;
334 0 : struct candidate const * c = &ch->candidates[ch->paired_idx];
335 0 : return !!c->tx_used || fd_tlsrec_sock_tx_pending( &ch->sessions[ c->session ].tls.sock );
336 0 : }
337 :
338 0 : FD_FN_PURE fd_failover_hello_t const * fd_failover_channel_peer_hello( fd_failover_channel_t const * ch ) { return &ch->peer_hello; }
339 0 : FD_FN_PURE ulong fd_failover_channel_generation( fd_failover_channel_t const * ch ) { return ch->generation; }
340 0 : 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 0 : fd_failover_channel_listen_port( fd_failover_channel_t const * ch ) {
344 0 : if( FD_UNLIKELY( ch->listen_fd==-1 ) ) return (ushort)0;
345 0 : struct sockaddr_in addr;
346 0 : socklen_t len = sizeof(addr);
347 0 : if( getsockname( ch->listen_fd, fd_type_pun( &addr ), &len ) ) FD_LOG_ERR(( "failover getsockname failed" ));
348 0 : return fd_ushort_bswap( addr.sin_port );
349 0 : }
350 :
351 : static void
352 : drop( fd_failover_channel_t * ch,
353 : ulong idx,
354 0 : long now ) {
355 0 : struct candidate * c = &ch->candidates[idx];
356 0 : int paired = ch->paired_idx==(int)idx;
357 0 : int accepted = c->fd!=-1 && !c->dialed;
358 0 : close_candidate( ch, c );
359 0 : if( paired ) {
360 0 : ch->paired_idx = -1;
361 0 : fd_memset( &ch->peer_hello, 0, sizeof(ch->peer_hello) );
362 0 : }
363 : /* A candidate from the listener does not get a session until it pairs. */
364 0 : if( !paired && accepted ) return;
365 : /* After loss, use the current handoff address to choose listener or dialer. */
366 0 : if( paired && want_dial( ch )!=ch->dial_peer ) reset( ch );
367 0 : if( ch->dial_peer ) {
368 : /* Recompute backoff parameters */
369 0 : int established = paired && now>=ch->paired_at && now-ch->paired_at>=FD_FAILOVER_CHANNEL_HELLO_TIMEOUT_NANOS;
370 0 : long jitter = (long)fd_rng_ulong_roll( ch->rng, (ulong)FD_FAILOVER_CHANNEL_BACKOFF_MIN_NANOS );
371 0 : ch->retry_at = fd_long_sat_add( now, fd_long_sat_add( established ? 0L : ch->backoff, jitter ) );
372 0 : ch->backoff = established ? FD_FAILOVER_CHANNEL_BACKOFF_MIN_NANOS
373 0 : : fd_long_min( fd_long_sat_add( ch->backoff, ch->backoff ), FD_FAILOVER_CHANNEL_BACKOFF_MAX_NANOS );
374 0 : }
375 0 : }
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 0 : long now ) {
396 0 : if( ch->paired_idx>=0 ) drop( ch, (ulong)ch->paired_idx, now );
397 0 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) if( ch->candidates[i].fd!=-1 ) drop( ch, i, now );
398 0 : }
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 0 : uint address ) {
413 0 : for( ulong i=ADMITTED_IDX; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ )
414 0 : if( ch->candidates[i].fd!=-1 && ch->candidates[i].address==address ) return 0;
415 0 : return 1;
416 0 : }
417 :
418 : static int
419 0 : has_proof( struct candidate const * c ) {
420 0 : return c->phase==PHASE_REQUEST_RECV && c->rx_used==sizeof(c->client_proof);
421 0 : }
422 :
423 : static int
424 0 : known_source( fd_failover_channel_t const * ch, uint address, long now ) {
425 0 : return address==ch->known_addr && now<ch->known_until;
426 0 : }
427 :
428 : static void
429 0 : remember_source( fd_failover_channel_t * ch, uint address, long now ) {
430 0 : ch->known_addr = address;
431 0 : ch->known_until = fd_long_sat_add( now, FD_FAILOVER_CHANNEL_IDLE_NANOS );
432 0 : }
433 :
434 : static ulong
435 : listen_slot( fd_failover_channel_t const * ch,
436 : uint address,
437 0 : long now ) {
438 0 : if( known_source( ch, address, now ) ) {
439 0 : return ch->candidates[ADMITTED_IDX].fd==-1 ? ADMITTED_IDX : FD_FAILOVER_CHANNEL_CANDIDATE_MAX;
440 0 : }
441 0 : for( ulong i=UNAUTHENTICATED_IDX; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) {
442 0 : if( ch->candidates[i].fd==-1 ) return i;
443 0 : }
444 0 : return FD_FAILOVER_CHANNEL_CANDIDATE_MAX;
445 0 : }
446 :
447 : static void
448 : start( fd_failover_channel_t * ch,
449 : ulong idx,
450 : int fd,
451 : uint address,
452 : long now,
453 0 : int phase ) {
454 0 : struct candidate * c = &ch->candidates[idx];
455 : /* Inbound sockets do not acquire a TLS session until admission succeeds. */
456 0 : if( c->fd==-1 ) ch->live++;
457 0 : c->fd = fd;
458 0 : c->address = address;
459 0 : c->phase = phase;
460 0 : c->dialed = phase==PHASE_CONNECT;
461 0 : c->session = -1;
462 0 : c->deadline = fd_long_sat_add( now, FD_FAILOVER_CHANNEL_HELLO_TIMEOUT_NANOS );
463 0 : c->admission_until = c->dialed ? c->deadline : fd_long_min( c->deadline, fd_long_sat_add( now, FD_FAILOVER_ADMISSION_START_NANOS ) );
464 0 : c->local_hello = ch->self_hello;
465 0 : c->rx_used = c->tx_used = c->tx_sent = 0UL;
466 0 : }
467 :
468 : static void
469 : accept_candidates( fd_failover_channel_t * ch,
470 : long now,
471 0 : int * busy ) {
472 0 : for( ulong n=0; n<FD_FAILOVER_ACCEPT_PER_POLL; n++ ) {
473 0 : if( !fd_token_bucket_consume( &ch->accept_bucket, 1.0f, now ) ) return;
474 0 : struct sockaddr_in addr;
475 0 : socklen_t len = sizeof(addr);
476 0 : int fd = accept4( ch->listen_fd, fd_type_pun( &addr ), &len, SOCK_NONBLOCK|SOCK_CLOEXEC );
477 0 : if( fd==-1 ) return;
478 0 : *busy = 1;
479 0 : uint address = addr.sin_addr.s_addr;
480 0 : ulong idx = listen_slot( ch, address, now );
481 0 : if( ch->paired_idx>=0 || len!=sizeof(addr) || addr.sin_family!=AF_INET ||
482 0 : idx==FD_FAILOVER_CHANNEL_CANDIDATE_MAX || !admit( ch, address ) ) {
483 0 : close( fd );
484 0 : continue;
485 0 : }
486 0 : start( ch, idx, fd, address, now, PHASE_START_RECV );
487 0 : }
488 0 : }
489 :
490 : static int
491 : flush( fd_failover_channel_t * ch,
492 : ulong idx,
493 : long now,
494 0 : int * busy ) {
495 0 : struct candidate * c = &ch->candidates[idx];
496 0 : struct session * s = &ch->sessions[ c->session ];
497 0 : if( !c->tx_used ) {
498 0 : int rc = fd_tlsrec_sock_flush( &s->tls.sock, s->tls.fd );
499 0 : if( rc<0 ) { lose( ch, idx, now, "flushing TLS output failed" ); return -1; }
500 0 : return rc;
501 0 : }
502 0 : long n = fd_failover_tls_write( &s->tls, s->tx+c->tx_sent, c->tx_used-c->tx_sent );
503 0 : if( n<0L ) { lose( ch, idx, now, "writing to it failed" ); return -1; }
504 0 : if( !n ) return 1;
505 0 : *busy = 1;
506 0 : c->tx_sent += (ulong)n;
507 0 : if( c->tx_sent<c->tx_used ) return 1;
508 0 : c->tx_used = c->tx_sent = 0UL;
509 0 : return fd_tlsrec_sock_tx_pending( &s->tls.sock ) ? 1 : 0;
510 0 : }
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 0 : ulong * payload_sz ) {
520 0 : struct candidate * c = &ch->candidates[idx];
521 0 : uchar * rx = ch->sessions[ c->session ].rx;
522 0 : uchar const * data;
523 0 : ulong frame_sz;
524 0 : int err = fd_failover_wire_decode( rx, c->rx_used, type, &data, payload_sz, &frame_sz );
525 0 : if( err==FD_FAILOVER_WIRE_ERR_AGAIN ) return 0;
526 0 : if( err ) { lose( ch, idx, now, "it sent a malformed frame" ); return -1; }
527 0 : fd_memcpy( payload, data, *payload_sz );
528 0 : memmove( rx, rx+frame_sz, c->rx_used-frame_sz );
529 0 : c->rx_used -= frame_sz;
530 0 : *busy = 1;
531 0 : return 1;
532 0 : }
533 :
534 : static char const *
535 0 : hello_err_cstr( int err ) {
536 0 : 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 0 : 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 0 : }
548 0 : }
549 :
550 : static void
551 : refuse_hello( fd_failover_channel_t * ch,
552 : struct candidate * c,
553 : int err,
554 0 : long now ) {
555 0 : int proven = c->dialed || c->phase>=PHASE_TLS_START;
556 0 : if( !proven && now<ch->refused_log_at ) { ch->refused_cnt++; return; }
557 0 : FD_LOG_WARNING(( "rejected the failover peer at `" FD_IP4_ADDR_FMT "`, %s%s%lu%s",
558 0 : FD_IP4_ADDR_FMT_ARGS( c->address ), hello_err_cstr( err ),
559 0 : ch->refused_cnt ? ", and " : "", ch->refused_cnt, ch->refused_cnt ? " more unproven peers since the last such line" : "" ));
560 0 : if( !proven ) {
561 0 : ch->refused_cnt = 0UL;
562 0 : ch->refused_log_at = fd_long_sat_add( now, FD_FAILOVER_LOG_INTERVAL_NANOS );
563 0 : }
564 0 : }
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 0 : ulong * payload_sz ) {
574 0 : struct candidate * c = &ch->candidates[idx];
575 0 : struct session * s = &ch->sessions[ c->session ];
576 0 : int rc = decode( ch, idx, now, busy, type, payload, payload_sz );
577 0 : if( rc ) return rc;
578 0 : ulong max = ch->paired_idx==(int)idx ? sizeof(s->rx) : FD_FAILOVER_FRAME_HDR_SZ+sizeof(fd_failover_hello_t);
579 0 : if( c->rx_used>=max ) { drop( ch, idx, now ); return -1; }
580 0 : long n = fd_failover_tls_read( &s->tls, s->rx+c->rx_used, max-c->rx_used );
581 0 : 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 0 : if( n>0L ) { c->rx_used += (ulong)n; *busy = 1; }
605 0 : return decode( ch, idx, now, busy, type, payload, payload_sz );
606 0 : }
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 0 : long now ) {
613 0 : struct candidate * c = &ch->candidates[idx];
614 0 : struct sockaddr_in addr = {
615 0 : .sin_family=AF_INET,
616 0 : .sin_port=fd_ushort_bswap( ch->peer_port ),
617 0 : .sin_addr.s_addr=ch->peer_addr
618 0 : };
619 0 : if( connect( c->fd, fd_type_pun( &addr ), sizeof(addr) ) && errno!=EISCONN ) {
620 0 : 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 0 : }
626 :
627 : /* Only operator-initiated outbound work prepares a TLS key share here. */
628 0 : ulong hello_sz = sizeof(c->client_hello);
629 0 : if( acquire_session( ch, idx, 1 )<=0 ) {
630 0 : drop( ch, idx, now );
631 0 : return 0;
632 0 : }
633 :
634 0 : if( fd_failover_tls_client_hello( &ch->sessions[ c->session ].tls, c->client_hello, &hello_sz ) ||
635 0 : hello_sz>sizeof(c->client_hello) ) {
636 0 : drop( ch, idx, now );
637 0 : return 0;
638 0 : }
639 :
640 0 : c->start = (fd_failover_admission_start_t){
641 0 : .magic=FD_FAILOVER_ADMISSION_MAGIC,
642 0 : .hello=c->local_hello,
643 0 : .client_hello_sz=(ushort)hello_sz
644 0 : };
645 0 : if( !fd_rng_secure( c->start.nonce, sizeof(c->start.nonce) ) ) {
646 0 : drop( ch, idx, now );
647 0 : return 0;
648 0 : }
649 0 : fd_memcpy( c->tx, &c->start, sizeof(c->start) );
650 0 : fd_memcpy( c->tx+sizeof(c->start), c->client_hello, hello_sz );
651 0 : c->tx_used = sizeof(c->start)+hello_sz;
652 0 : c->phase = PHASE_START_SEND;
653 0 : return 1;
654 0 : }
655 :
656 : static int
657 : handshake_candidate( fd_failover_channel_t * ch,
658 : ulong idx,
659 : long now,
660 0 : int * busy ) {
661 0 : struct candidate * c = &ch->candidates[idx];
662 0 : struct session * s = &ch->sessions[ c->session ];
663 0 : int rc = fd_failover_tls_handshake( &s->tls );
664 0 : 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 0 : if( !rc ) return 0; /* Keep waiting for TLS. */
672 0 : *busy = 1;
673 :
674 : /* TLS authenticated a key, send the HELLO! */
675 0 : c->phase = PHASE_HELLO;
676 0 : if( c->dialed ) {
677 0 : c->tx_used = fd_failover_wire_encode( s->tx, FD_FAILOVER_MSG_HELLO,
678 0 : (uchar const *)&c->local_hello, sizeof(c->local_hello) );
679 0 : }
680 0 : return 1;
681 0 : }
682 :
683 : static int
684 : read_candidate_hello( fd_failover_channel_t * ch,
685 : ulong idx,
686 : long now,
687 : int * busy,
688 0 : uchar * payload ) {
689 0 : struct candidate * c = &ch->candidates[idx];
690 0 : ushort type;
691 0 : ulong sz;
692 0 : if( read_frame( ch, idx, now, busy, &type, payload, &sz )<=0 ) return 0;
693 0 : 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 0 : int err = fd_memeq( payload, &c->hello, sizeof(c->hello) ) &&
701 0 : fd_memeq( c->hello.staked_pubkey, ch->sessions[ c->session ].tls.peer_pubkey, 32UL )
702 0 : ? fd_failover_hello_check( &c->local_hello, &c->hello ) : HELLO_ERR_TLS_KEY;
703 0 : 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 0 : c->phase = PHASE_READY;
711 0 : if( !c->dialed ) {
712 0 : c->tx_used = fd_failover_wire_encode( ch->sessions[ c->session ].tx, FD_FAILOVER_MSG_HELLO,
713 0 : (uchar const *)&c->local_hello, sizeof(c->local_hello) );
714 0 : if( flush( ch, idx, now, busy )!=0 ) return 0; /* Remain READY until output drains, unless the write fails. */
715 0 : }
716 0 : return 1;
717 0 : }
718 :
719 : static void
720 : pair_candidate( fd_failover_channel_t * ch,
721 : ulong idx,
722 : long now,
723 0 : int * busy ) {
724 0 : struct candidate * c = &ch->candidates[idx];
725 : /* The first session that authenticates wins. */
726 0 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) if( i!=idx ) close_candidate( ch, &ch->candidates[i] );
727 :
728 0 : if( !c->dialed ) ch->dial_peer = 0;
729 :
730 0 : ch->paired_idx = (int)idx;
731 0 : ch->sessions[ c->session ].tls.paired = 1;
732 0 : ch->peer_hello = c->hello;
733 0 : ch->generation++;
734 0 : remember_source( ch, c->address, now );
735 :
736 0 : fd_keepalive_init( ch->keepalive, ch->rng, FD_FAILOVER_CHANNEL_KEEPALIVE_NANOS, FD_FAILOVER_CHANNEL_IDLE_NANOS, now );
737 0 : fd_keepalive_rx( ch->keepalive, now );
738 0 : ch->paired_at = now;
739 0 : ch->expected_close = 0;
740 0 : *busy = 1;
741 :
742 0 : FD_LOG_NOTICE(( "paired with the failover peer at `" FD_IP4_ADDR_FMT "` via %s, %s, boot id %016lx",
743 0 : FD_IP4_ADDR_FMT_ARGS( c->address ), c->dialed ? "outbound dial" : "listener",
744 0 : c->hello.holds ? "it runs the identity" : "it does not run the identity", c->hello.boot_id ));
745 0 : }
746 :
747 : static int
748 0 : read_admission( struct candidate * c, ulong goal, int * busy ) {
749 0 : if( FD_UNLIKELY( goal>FD_FAILOVER_ADMISSION_MAX || c->rx_used>goal ) ) return -1;
750 0 : if( c->rx_used==goal ) return 1;
751 0 : long n = recv( c->fd, c->rx+c->rx_used, goal-c->rx_used, MSG_DONTWAIT|MSG_NOSIGNAL );
752 0 : if( n<0L ) return ( errno==EAGAIN || errno==EWOULDBLOCK || errno==EINTR ) ? 0 : -1;
753 0 : if( !n ) return -1;
754 :
755 0 : c->rx_used += (ulong)n;
756 0 : *busy = 1;
757 0 : return c->rx_used==goal;
758 0 : }
759 :
760 : static int
761 0 : write_admission( struct candidate * c, int * busy ) {
762 0 : long n = send( c->fd, c->tx+c->tx_sent, c->tx_used-c->tx_sent, MSG_DONTWAIT|MSG_NOSIGNAL );
763 0 : if( n<0L ) return ( errno==EAGAIN || errno==EWOULDBLOCK || errno==EINTR ) ? 0 : -1;
764 0 : if( !n ) return -1;
765 0 : c->tx_sent += (ulong)n;
766 0 : *busy = 1;
767 0 : if( c->tx_sent<c->tx_used ) return 0;
768 0 : c->tx_sent = c->tx_used = 0UL;
769 0 : return 1;
770 0 : }
771 :
772 : static int
773 0 : verify_admission_proof( fd_failover_channel_t * ch, struct candidate * c, int server, uchar const sig[ 64 ] ) {
774 0 : uchar msg[ FD_KEYGUARD_FAILOVER_PROOF_SZ ];
775 0 : fd_failover_proof_msg( msg, server, c->transcript );
776 0 : return fd_ed25519_verify( msg, sizeof(msg), sig, c->local_hello.staked_pubkey, ch->sha )==FD_ED25519_SUCCESS;
777 0 : }
778 :
779 : static int
780 0 : sign_admission_proof( fd_failover_channel_t * ch, struct candidate * c, int server ) {
781 0 : uchar msg[ FD_KEYGUARD_FAILOVER_PROOF_SZ ];
782 0 : fd_failover_proof_msg( msg, server, c->transcript );
783 0 : ch->proof_signer.sign_fn( ch->proof_signer.ctx, c->tx, msg );
784 0 : if( fd_ed25519_verify( msg, sizeof(msg), c->tx, c->local_hello.staked_pubkey, ch->sha )!=FD_ED25519_SUCCESS ) return 0;
785 0 : c->tx_used = 64UL;
786 0 : c->tx_sent = 0UL;
787 0 : return 1;
788 0 : }
789 :
790 : /* One Ed25519 operation per poll, within the global rate. */
791 : static int
792 0 : verify_credit( fd_failover_channel_t * ch, long now ) {
793 0 : if( !ch->verify_budget || !fd_token_bucket_consume( &ch->verify_bucket, 1.0f, now ) ) return 0;
794 0 : ch->verify_budget--;
795 0 : return 1;
796 0 : }
797 :
798 : /* Service the candidate admission state machine. */
799 : static void
800 0 : service_admission( fd_failover_channel_t * ch, ulong idx, long now, int * busy ) {
801 0 : struct candidate * c = &ch->candidates[idx];
802 0 : int rc = 1;
803 0 : int refused = 0;
804 0 : char const * why = NULL;
805 0 : switch( c->phase ) {
806 0 : case PHASE_START_SEND:
807 0 : case PHASE_CHALLENGE_SEND:
808 0 : case PHASE_REQUEST_SEND:
809 0 : case PHASE_RESPONSE_SEND:
810 0 : rc = write_admission( c, busy );
811 0 : if( rc<0 ) break;
812 0 : if( !rc ) return;
813 0 : c->rx_used = 0UL;
814 0 : if( c->phase==PHASE_START_SEND ) {
815 0 : c->phase = PHASE_CHALLENGE_RECV;
816 0 : }
817 0 : else if( c->phase==PHASE_CHALLENGE_SEND ) {
818 0 : c->admission_until = fd_long_min( c->deadline, fd_long_sat_add( now, FD_FAILOVER_ADMISSION_PROOF_NANOS ) );
819 0 : c->phase = PHASE_REQUEST_RECV;
820 0 : }
821 0 : else if( c->phase==PHASE_REQUEST_SEND ) {
822 0 : c->phase = PHASE_RESPONSE_RECV;
823 0 : }
824 0 : else {
825 0 : c->phase = PHASE_TLS;
826 0 : }
827 0 : return;
828 0 : case PHASE_START_RECV: {
829 : /* Read only the fixed header until its length is checked. */
830 0 : ulong goal = c->rx_used<sizeof(c->start) ? sizeof(c->start) : sizeof(c->start)+c->start.client_hello_sz;
831 0 : rc = read_admission( c, goal, busy );
832 0 : if( rc<0 ) break;
833 0 : if( !rc ) return;
834 0 : if( goal==sizeof(c->start) ) {
835 0 : fd_memcpy( &c->start, c->rx, sizeof(c->start) );
836 0 : if( c->start.magic!=FD_FAILOVER_ADMISSION_MAGIC ||
837 0 : c->start.client_hello_sz<=sizeof(fd_tlsrec_hdr_t) ||
838 0 : c->start.client_hello_sz>sizeof(c->client_hello) ) break;
839 0 : int err = fd_failover_hello_check( &c->local_hello, &c->start.hello );
840 0 : if( err ) { refuse_hello( ch, c, err, now ); refused = 1; break; }
841 0 : return;
842 0 : }
843 0 : fd_memcpy( c->client_hello, c->rx+sizeof(c->start), c->start.client_hello_sz );
844 0 : fd_tlsrec_hdr_t hdr;
845 0 : fd_memcpy( &hdr, c->client_hello, sizeof(hdr) );
846 0 : if( hdr.content_type!=FD_TLS_REC_HANDSHAKE ||
847 0 : (ulong)fd_ushort_bswap( hdr.length )+sizeof(hdr)!=c->start.client_hello_sz ||
848 0 : c->client_hello[ sizeof(hdr) ]!=FD_TLS_MSG_CLIENT_HELLO ) break;
849 0 : c->hello = c->start.hello;
850 0 : c->challenge = (fd_failover_admission_challenge_t){ .magic=FD_FAILOVER_ADMISSION_MAGIC, .hello=c->local_hello };
851 0 : if( !fd_rng_secure( c->challenge.nonce, sizeof(c->challenge.nonce) ) ) break;
852 0 : fd_failover_admission_hash( &c->start, c->client_hello, &c->challenge, c->transcript );
853 0 : fd_memcpy( c->tx, &c->challenge, sizeof(c->challenge) );
854 0 : c->tx_used = sizeof(c->challenge);
855 0 : c->rx_used = 0UL;
856 0 : c->admission_until = fd_long_min( c->deadline, fd_long_sat_add( now, FD_FAILOVER_ADMISSION_START_NANOS ) );
857 0 : c->phase = PHASE_CHALLENGE_SEND;
858 0 : return;
859 0 : }
860 0 : case PHASE_CHALLENGE_RECV:
861 0 : rc = read_admission( c, sizeof(c->challenge), busy );
862 0 : if( rc<0 ) break;
863 0 : if( !rc ) return;
864 0 : fd_memcpy( &c->challenge, c->rx, sizeof(c->challenge) );
865 0 : if( c->challenge.magic!=FD_FAILOVER_ADMISSION_MAGIC ) break;
866 0 : int err = fd_failover_hello_check( &c->local_hello, &c->challenge.hello );
867 0 : if( err ) { refuse_hello( ch, c, err, now ); refused = 1; break; }
868 0 : c->hello = c->challenge.hello;
869 0 : fd_failover_admission_hash( &c->start, c->client_hello, &c->challenge, c->transcript );
870 0 : c->rx_used = 0UL;
871 0 : c->phase = PHASE_REQUEST_SIGN;
872 0 : *busy = 1;
873 0 : return;
874 0 : case PHASE_REQUEST_SIGN:
875 0 : if( !verify_credit( ch, now ) ) return;
876 0 : if( !sign_admission_proof( ch, c, 0 ) ) { why = "this machine could not sign the admission proof"; break; }
877 0 : fd_memcpy( c->client_proof, c->tx, sizeof(c->client_proof) );
878 0 : c->phase = PHASE_REQUEST_SEND;
879 0 : *busy = 1;
880 0 : return;
881 0 : case PHASE_RESPONSE_RECV: {
882 0 : fd_failover_admission_response_t response;
883 0 : if( c->rx_used<sizeof(response) ) {
884 0 : rc = read_admission( c, sizeof(response), busy );
885 0 : if( rc<0 ) break;
886 0 : if( !rc ) return;
887 0 : fd_memcpy( &response, c->rx, sizeof(response) );
888 0 : if( !response.server_flight_sz || response.server_flight_sz>FD_FAILOVER_SERVER_FLIGHT_MAX ) break;
889 0 : return;
890 0 : }
891 0 : fd_memcpy( &response, c->rx, sizeof(response) );
892 0 : rc = read_admission( c, sizeof(response)+response.server_flight_sz, busy );
893 0 : if( rc<0 ) break;
894 0 : if( !rc || !verify_credit( ch, now ) ) return;
895 0 : fd_failover_response_hash( c->transcript, c->client_proof, c->rx+sizeof(response),
896 0 : response.server_flight_sz, c->transcript );
897 0 : 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 0 : 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 0 : c->phase = PHASE_TLS;
904 0 : c->rx_used = 0UL;
905 0 : *busy = 1;
906 0 : return;
907 0 : }
908 0 : case PHASE_REQUEST_RECV:
909 0 : rc = read_admission( c, 64UL, busy );
910 0 : if( rc<0 ) break;
911 0 : if( !rc || !verify_credit( ch, now ) ) return;
912 0 : 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 0 : fd_memcpy( c->client_proof, c->rx, sizeof(c->client_proof) );
918 0 : c->rx_used = 0UL;
919 0 : c->admission_until = c->deadline;
920 0 : c->phase = PHASE_TLS_START;
921 0 : *busy = 1;
922 0 : return;
923 0 : case PHASE_RESPONSE_SIGN: {
924 0 : if( !verify_credit( ch, now ) ) return;
925 0 : fd_failover_admission_response_t response = { .server_flight_sz=(ushort)(c->tx_used-sizeof(fd_failover_admission_response_t)) };
926 0 : if( !sign_admission_proof( ch, c, 1 ) ) break;
927 0 : fd_memcpy( response.proof, c->tx, sizeof(response.proof) );
928 0 : fd_memcpy( c->tx, &response, sizeof(response) );
929 0 : c->tx_used = sizeof(response)+response.server_flight_sz;
930 0 : c->phase = PHASE_RESPONSE_SEND;
931 0 : *busy = 1;
932 0 : return;
933 0 : }
934 0 : case PHASE_TLS_START: {
935 0 : int got = acquire_session( ch, idx, 0 );
936 0 : if( !got ) return;
937 0 : if( got<0 ) break;
938 0 : struct session * s = &ch->sessions[ c->session ];
939 0 : if( fd_failover_tls_accept_flight( &s->tls, c->client_hello, c->start.client_hello_sz ) ) break;
940 0 : if( !s->tls.sock.tx_sz || s->tls.sock.tx_sz>FD_FAILOVER_SERVER_FLIGHT_MAX ) break;
941 0 : fd_failover_response_hash( c->transcript, c->client_proof, s->tls.sock.tx_buf,
942 0 : (ushort)s->tls.sock.tx_sz, c->transcript );
943 0 : fd_memcpy( c->tx+sizeof(fd_failover_admission_response_t), s->tls.sock.tx_buf, s->tls.sock.tx_sz );
944 0 : c->tx_used = sizeof(fd_failover_admission_response_t)+s->tls.sock.tx_sz;
945 0 : s->tls.sock.tx_off = s->tls.sock.tx_sz = 0UL;
946 0 : c->phase = PHASE_RESPONSE_SIGN;
947 0 : *busy = 1;
948 0 : return;
949 0 : }
950 0 : default:
951 0 : break;
952 0 : }
953 0 : if( c->dialed && !refused ) {
954 0 : 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 0 : } else if( rc<0 ) {
957 0 : 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 0 : FD_IP4_ADDR_FMT_ARGS( c->address ), awaited_cstr( c->phase ) ));
959 0 : } 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 0 : }
964 0 : drop( ch, idx, now );
965 0 : }
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 0 : uchar * payload ) {
974 0 : struct candidate * c = &ch->candidates[idx];
975 0 : if( c->phase==PHASE_CONNECT ) {
976 0 : if( connect_candidate( ch, idx, now ) ) *busy = 1;
977 0 : return;
978 0 : }
979 0 : if( c->phase<PHASE_TLS ) {
980 0 : service_admission( ch, idx, now, busy );
981 0 : return;
982 0 : }
983 0 : if( c->phase==PHASE_TLS ) {
984 0 : if( !handshake_candidate( ch, idx, now, busy ) ) {
985 0 : return;
986 0 : }
987 0 : }
988 0 : if( flush( ch, idx, now, busy )!=0 ) return;
989 0 : if( c->phase==PHASE_HELLO ) {
990 0 : if( !read_candidate_hello( ch, idx, now, busy, payload ) ) {
991 0 : return;
992 0 : }
993 0 : }
994 0 : if( c->phase==PHASE_READY && !c->tx_used ) {
995 0 : pair_candidate( ch, idx, now, busy );
996 0 : }
997 0 : }
998 :
999 : /* The paired session went silent. */
1000 : static int
1001 0 : silent( fd_failover_channel_t const * ch, long now ) {
1002 0 : long last_rx = ch->keepalive->ts_last_rx;
1003 0 : return now>=last_rx && now-last_rx>FD_FAILOVER_CHANNEL_IDLE_NANOS;
1004 0 : }
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 0 : ulong * payload_sz ) {
1014 0 : ulong idx = (ulong)ch->paired_idx;
1015 0 : int delivered = 0;
1016 0 : int flushed = flush( ch, idx, now, busy );
1017 0 : if( !flushed ) {
1018 0 : int rc = read_frame( ch, idx, now, busy, type, payload, payload_sz );
1019 0 : if( rc>0 ) {
1020 0 : if( *type==FD_FAILOVER_MSG_HELLO ) {
1021 0 : lose( ch, idx, now, "it sent a second HELLO" );
1022 0 : }
1023 0 : else {
1024 0 : fd_keepalive_rx( ch->keepalive, now );
1025 0 : remember_source( ch, ch->candidates[idx].address, now );
1026 0 : delivered = *type!=FD_FAILOVER_MSG_KEEPALIVE;
1027 0 : }
1028 0 : } else if( !rc && silent( ch, now ) ) {
1029 0 : lose( ch, idx, now, "it was idle past its deadline" );
1030 0 : }
1031 0 : }
1032 0 : if( flushed>0 && silent( ch, now ) ) {
1033 0 : lose( ch, idx, now, "it stopped making progress" );
1034 0 : }
1035 0 : if( ch->paired_idx>=0 &&
1036 0 : !fd_failover_channel_tx_pending( ch ) &&
1037 0 : fd_keepalive_should_tx( ch->keepalive, now ) )
1038 0 : {
1039 0 : if( !fd_failover_channel_send( ch, now, FD_FAILOVER_MSG_KEEPALIVE, NULL, 0UL ) ) {
1040 0 : fd_keepalive_tx( ch->keepalive, ch->rng, now );
1041 0 : *busy = 1;
1042 0 : }
1043 0 : }
1044 0 : return delivered;
1045 0 : }
1046 :
1047 : static void
1048 : dial( fd_failover_channel_t * ch,
1049 : long now,
1050 0 : int * busy ) {
1051 0 : if( !ch->dial_peer || !ch->peer_addr || ch->candidates[OUTBOUND_IDX].fd!=-1 || now<ch->retry_at ) {
1052 0 : return;
1053 0 : }
1054 :
1055 0 : *busy = 1;
1056 0 : int fd = socket( AF_INET, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0 );
1057 0 : if( fd==-1 ) {
1058 0 : drop( ch, OUTBOUND_IDX, now );
1059 0 : return;
1060 0 : }
1061 :
1062 0 : int yes = 1;
1063 0 : 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 0 : start( ch, OUTBOUND_IDX, fd, ch->peer_addr, now, PHASE_CONNECT );
1070 0 : }
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 0 : ulong * payload_sz ) {
1079 0 : int delivered = 0;
1080 0 : ch->verify_budget = 1;
1081 0 : if( ch->paired_idx>=0 ) {
1082 0 : delivered = service_paired( ch, now, busy, type, payload, payload_sz );
1083 0 : }
1084 :
1085 : /* Expire every slot each poll, independently of round-robin service. */
1086 0 : if( ch->live > (ulong)( ch->paired_idx>=0 ) ) {
1087 0 : for( ulong i=0; i<FD_FAILOVER_CHANNEL_CANDIDATE_MAX; i++ ) {
1088 0 : struct candidate * c = &ch->candidates[i];
1089 0 : 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 0 : }
1103 0 : }
1104 :
1105 0 : if( FD_UNLIKELY( !ch->tls_ctx.ready ) || ch->paired_idx>=0 ) {
1106 0 : return delivered;
1107 0 : }
1108 0 : dial( ch, now, busy );
1109 :
1110 0 : ulong serviced = 0UL;
1111 0 : for( ulong idx=0UL; idx<UNAUTHENTICATED_IDX; idx++ ) {
1112 0 : if( ch->candidates[idx].fd==-1 ) continue;
1113 0 : service_candidate( ch, idx, now, busy, payload );
1114 0 : serviced++;
1115 0 : if( ch->paired_idx>=0 ) return delivered;
1116 0 : }
1117 :
1118 : /* Round-robin the shared pool. */
1119 0 : ulong reserved = (ulong)( ch->candidates[OUTBOUND_IDX].fd!=-1 ) + (ulong)( ch->candidates[ADMITTED_IDX].fd!=-1 );
1120 0 : for( ulong n=0UL; ch->live>reserved && n<FD_FAILOVER_CHANNEL_CANDIDATE_MAX-UNAUTHENTICATED_IDX && serviced<FD_FAILOVER_SERVICE_PER_POLL; n++ ) {
1121 0 : ulong idx = UNAUTHENTICATED_IDX+ch->cursor;
1122 0 : ch->cursor = (ch->cursor+1UL)%(FD_FAILOVER_CHANNEL_CANDIDATE_MAX-UNAUTHENTICATED_IDX);
1123 0 : if( ch->candidates[idx].fd==-1 ) continue;
1124 0 : if( has_proof( &ch->candidates[idx] ) && !ch->verify_budget ) continue;
1125 :
1126 0 : service_candidate( ch, idx, now, busy, payload );
1127 0 : serviced++;
1128 0 : if( ch->paired_idx>=0 ) {
1129 0 : return delivered;
1130 0 : }
1131 0 : }
1132 0 : if( ch->listen_fd!=-1 ) {
1133 0 : accept_candidates( ch, now, busy );
1134 0 : }
1135 0 : return delivered;
1136 0 : }
1137 :
1138 : int
1139 : fd_failover_channel_send( fd_failover_channel_t * ch,
1140 : long now,
1141 : ushort type,
1142 : uchar const * payload,
1143 0 : ulong sz ) {
1144 0 : if( ch->paired_idx<0 ) return -1;
1145 :
1146 0 : ulong idx = (ulong)ch->paired_idx;
1147 0 : struct candidate * c = &ch->candidates[idx];
1148 :
1149 0 : if( c->tx_used ) return -1;
1150 0 : c->tx_used = fd_failover_wire_encode( ch->sessions[ c->session ].tx, type, payload, sz );
1151 0 : 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 0 : int busy = 0;
1157 0 : int rc = flush( ch, idx, now, &busy )<0 ? -1 : 0;
1158 0 : if( !rc && type==FD_FAILOVER_MSG_HANDOFF_RESULT ) {
1159 0 : ch->expected_close = 1;
1160 0 : }
1161 0 : return rc;
1162 0 : }
|
