Line data Source code
1 : #define _GNU_SOURCE
2 : #include "fd_event_client.h"
3 :
4 : #include "../../waltz/resolv/fd_netdb.h"
5 : #include "../../waltz/http/fd_url.h"
6 : #include "../../waltz/grpc/fd_grpc_client.h"
7 : #include "../../waltz/grpc/fd_grpc_client_private.h"
8 : #include "../../ballet/pb/fd_pb_tokenize.h"
9 : #include "../../ballet/pb/fd_pb_encode.h"
10 : #include "../../ballet/hex/fd_hex.h"
11 : #include "../../tango/tempo/fd_tempo.h"
12 : #include "../../util/net/fd_ip4.h"
13 : #include "../../util/log/fd_log.h"
14 : #include "../keyguard/fd_keyguard.h"
15 :
16 : #if FD_HAS_OPENSSL
17 : #include "../../waltz/openssl/fd_openssl.h"
18 : #include <openssl/ssl.h>
19 : #include <openssl/err.h>
20 : #endif
21 :
22 : #include <netinet/tcp.h>
23 : #include <unistd.h>
24 : #include <errno.h>
25 : #include <sys/socket.h>
26 : #include <netinet/in.h>
27 :
28 0 : #define DISCONNECT_REASON_IDENTITY_CHANGED (0)
29 0 : #define DISCONNECT_REASON_CONNECT_FAILED (1)
30 0 : #define DISCONNECT_REASON_DNS_RESOLVE_FAILED (2)
31 0 : #define DISCONNECT_REASON_TIMEOUT (3)
32 0 : #define DISCONNECT_REASON_TRANSPORT_FAILED (4)
33 0 : #define DISCONNECT_REASON_PEER_CLOSED (5)
34 0 : #define DISCONNECT_REASON_INVALID_CURSOR (6)
35 0 : #define DISCONNECT_REASON_AUTH_FAILED (7)
36 0 : #define DISCONNECT_REASON_INVALID_PROTOBUF (8)
37 :
38 0 : #define FD_EVENT_CLIENT_REQ_CTX_AUTHENTICATE (1UL)
39 : #define FD_EVENT_CLIENT_REQ_CTX_CONFIRM_AUTH (2UL)
40 0 : #define FD_EVENT_CLIENT_REQ_CTX_STREAM_EVENTS (3UL)
41 :
42 0 : #define FD_EVENT_CLIENT_HEARTBEAT_NANOS (15L*(long)1e9)
43 :
44 0 : #define FD_EVENT_CLIENT_TOKEN_SZ (217UL)
45 :
46 : struct fd_event_client {
47 : fd_grpc_client_t * grpc_client;
48 : fd_grpc_client_metrics_t grpc_metrics[1];
49 : fd_grpc_h2_stream_t * event_stream;
50 :
51 : char client_version[ 10UL ];
52 : char commit_hash[ 41UL ];
53 : char action[ 16UL ];
54 : uchar identity_pubkey[ 32UL ];
55 :
56 : int has_genesis_hash;
57 : fd_hash_t genesis_hash[1];
58 :
59 : int connect_fail_logged;
60 :
61 : ushort has_shred_version;
62 : ushort shred_version;
63 :
64 : ulong event_id;
65 :
66 : ulong instance_id;
67 : ulong boot_id;
68 : ulong machine_id;
69 :
70 : int defer_disconnect;
71 : ulong consecutive_failure_count;
72 :
73 : long last_stream_send_ticks;
74 : long heartbeat_ticks;
75 :
76 : int auth_send_pending;
77 :
78 : ulong state;
79 : union {
80 : struct {
81 : long reconnect_deadline;
82 : } disconnected;
83 :
84 : struct {
85 : long connect_deadline;
86 : } connecting;
87 :
88 : struct {
89 : long connected_timestamp;
90 : } connected;
91 : };
92 :
93 : int so_sndbuf;
94 : int sockfd;
95 :
96 : int use_tls;
97 : #if FD_HAS_OPENSSL
98 : SSL_CTX * ssl_ctx;
99 : SSL * ssl;
100 : #endif
101 :
102 : /* wallclock deadline for auth handshake, LONG_MAX if not
103 : authenticating. */
104 : long auth_deadline;
105 :
106 : char server_fqdn[ 256 ]; /* cstr */
107 : ulong server_fqdn_len;
108 : uint server_ip4_addr;
109 : ushort server_tcp_port;
110 :
111 : fd_rng_t * rng;
112 : fd_circq_t * circq;
113 : fd_keyguard_client_t * keyguard_client;
114 :
115 : /* Stateless-auth bearer value: "hex(challenge_token).hex(signature)",
116 : built from the challenge token returned by Authenticate and our
117 : ed25519 signature over it. Presented as the `authorization: Bearer
118 : <...>` header on the StreamEvents request. */
119 : char auth_bearer[ 2UL*FD_EVENT_CLIENT_TOKEN_SZ + 1UL + 2UL*64UL + 1UL ];
120 : ulong auth_bearer_len;
121 :
122 : fd_event_client_metrics_t metrics;
123 : };
124 :
125 : FD_FN_CONST ulong
126 0 : fd_event_client_align( void ) {
127 0 : return alignof( fd_event_client_t );
128 0 : }
129 :
130 : FD_FN_CONST ulong
131 0 : fd_event_client_footprint( ulong buf_max ) {
132 0 : ulong l;
133 0 : l = FD_LAYOUT_INIT;
134 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_event_client_t), sizeof(fd_event_client_t) );
135 0 : l = FD_LAYOUT_APPEND( l, fd_grpc_client_align(), fd_grpc_client_footprint( buf_max ) );
136 0 : return FD_LAYOUT_FINI( l, alignof(fd_event_client_t) );
137 0 : }
138 :
139 : void *
140 : fd_event_client_new( void * shmem,
141 : fd_keyguard_client_t * keyguard_client,
142 : fd_rng_t * rng,
143 : fd_circq_t * circq,
144 : int so_sndbuf,
145 : char const * _url,
146 : uchar const * identity_pubkey,
147 : char const * client_version,
148 : char const * commit_hash,
149 : char const * action,
150 : ulong instance_id,
151 : ulong boot_id,
152 : ulong machine_id,
153 : ulong buf_max,
154 : int use_tls,
155 0 : void * ssl_ctx ) {
156 0 : if( FD_UNLIKELY( !shmem ) ) {
157 0 : FD_LOG_WARNING(( "NULL shmem" ));
158 0 : return NULL;
159 0 : }
160 :
161 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_event_client_align() ) ) ) {
162 0 : FD_LOG_WARNING(( "misaligned shmem" ));
163 0 : return NULL;
164 0 : }
165 :
166 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
167 0 : fd_event_client_t * client = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_event_client_t), sizeof(fd_event_client_t) );
168 0 : void * grpc_client_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_grpc_client_align(), fd_grpc_client_footprint( buf_max ) );
169 :
170 0 : fd_url_t url[1];
171 0 : _Bool _is_ssl = 0;
172 0 : if( FD_UNLIKELY( fd_url_parse_endpoint( url,
173 0 : _url,
174 0 : strlen( _url ),
175 0 : &client->server_tcp_port,
176 0 : &_is_ssl,
177 0 : "[tiles.event.url]" ) ) ) {
178 0 : FD_LOG_ERR(( "Could not parse [tiles.event.url]" ));
179 0 : }
180 0 : if( FD_UNLIKELY( url->host_len > 255 ) ) {
181 0 : FD_LOG_CRIT(( "Invalid url->host_len" )); /* unreachable */
182 0 : }
183 0 : fd_cstr_fini( fd_cstr_append_text( fd_cstr_init( client->server_fqdn ), url->host, url->host_len ) );
184 0 : client->server_fqdn_len = url->host_len;
185 :
186 0 : fd_memcpy( client->identity_pubkey, identity_pubkey, 32UL );
187 0 : fd_cstr_ncpy( client->client_version, client_version, sizeof( client->client_version ) );
188 0 : fd_cstr_fini( fd_cstr_append_text( fd_cstr_init( client->commit_hash ), commit_hash, fd_ulong_min( strlen( commit_hash ), sizeof( client->commit_hash )-1UL ) ) );
189 0 : fd_cstr_ncpy( client->action, action, sizeof( client->action ) );
190 :
191 0 : client->event_id = 0UL;
192 :
193 0 : client->instance_id = instance_id;
194 0 : client->boot_id = boot_id;
195 0 : client->machine_id = machine_id;
196 :
197 0 : client->has_genesis_hash = 0;
198 0 : client->has_shred_version = 0;
199 0 : client->connect_fail_logged = 0;
200 :
201 0 : client->so_sndbuf = so_sndbuf;
202 0 : client->sockfd = -1;
203 0 : client->use_tls = use_tls;
204 0 : #if FD_HAS_OPENSSL
205 0 : client->ssl_ctx = (SSL_CTX *)ssl_ctx;
206 0 : client->ssl = NULL;
207 : #else
208 : (void)ssl_ctx;
209 : if( FD_UNLIKELY( use_tls ) ) {
210 : FD_LOG_ERR(( "TLS requested for event service but this build does not include OpenSSL. "
211 : "To install OpenSSL, re-run ./deps.sh and do a clean rebuild." ));
212 : }
213 : #endif
214 0 : client->auth_deadline = LONG_MAX;
215 0 : client->auth_send_pending = 0;
216 0 : client->state = FD_EVENT_CLIENT_STATE_DISCONNECTED;
217 0 : client->disconnected.reconnect_deadline = 0L;
218 :
219 0 : client->defer_disconnect = INT_MAX;
220 0 : client->consecutive_failure_count = 7UL; /* Start high, so if server is down we don't keep retrying on boot */
221 0 : client->last_stream_send_ticks = 0L;
222 0 : client->heartbeat_ticks = (long)(fd_tempo_tick_per_ns( NULL )*(double)FD_EVENT_CLIENT_HEARTBEAT_NANOS);
223 :
224 0 : client->circq = circq;
225 0 : client->rng = rng;
226 0 : client->keyguard_client = keyguard_client;
227 :
228 0 : extern fd_grpc_client_callbacks_t fd_event_client_grpc_callbacks;
229 0 : client->grpc_client = fd_grpc_client_new( grpc_client_mem, &fd_event_client_grpc_callbacks, client->grpc_metrics, client, buf_max, fd_rng_ulong( rng ) );
230 0 : FD_TEST( client->grpc_client );
231 :
232 0 : memset( &client->metrics, 0, sizeof(client->metrics) );
233 0 : memset( client->grpc_metrics, 0, sizeof(fd_grpc_client_metrics_t) );
234 :
235 0 : fd_grpc_client_set_version( client->grpc_client, client->client_version, strlen( client->client_version ) );
236 0 : fd_grpc_client_set_authority( client->grpc_client, client->server_fqdn, client->server_fqdn_len, client->server_tcp_port );
237 :
238 0 : return (void *)client;
239 0 : }
240 :
241 : fd_event_client_t *
242 0 : fd_event_client_join( void * shec ) {
243 0 : if( FD_UNLIKELY( !shec ) ) {
244 0 : FD_LOG_WARNING(( "NULL shec" ));
245 0 : return NULL;
246 0 : }
247 :
248 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shec, fd_event_client_align() ) ) ) {
249 0 : FD_LOG_WARNING(( "misaligned shec" ));
250 0 : return NULL;
251 0 : }
252 :
253 0 : fd_event_client_t * client = (fd_event_client_t *)shec;
254 :
255 0 : return client;
256 0 : }
257 :
258 : fd_event_client_metrics_t const *
259 0 : fd_event_client_metrics( fd_event_client_t const * client ) {
260 : /* Update bytes from grpc metrics */
261 0 : ((fd_event_client_t *)client)->metrics.bytes_written = client->grpc_metrics->stream_chunks_tx_bytes;
262 0 : ((fd_event_client_t *)client)->metrics.bytes_read = client->grpc_metrics->stream_chunks_rx_bytes;
263 0 : return &client->metrics;
264 0 : }
265 :
266 : ulong
267 0 : fd_event_client_state( fd_event_client_t const * client ) {
268 0 : return client->state;
269 0 : }
270 :
271 : ulong
272 0 : fd_event_client_id_reserve( fd_event_client_t * client ) {
273 0 : return client->event_id++;
274 0 : }
275 :
276 : void
277 : fd_event_client_init_genesis( fd_event_client_t * client,
278 0 : fd_genesis_meta_t const * meta ) {
279 0 : *client->genesis_hash = meta->genesis_hash;
280 0 : client->has_genesis_hash = 1;
281 0 : }
282 :
283 : void
284 : fd_event_client_init_shred_version( fd_event_client_t * client,
285 0 : ushort shred_version ) {
286 0 : client->shred_version = shred_version;
287 0 : client->has_shred_version = 1;
288 0 : }
289 :
290 : static void
291 0 : backoff( fd_event_client_t * client ) {
292 0 : long now = fd_log_wallclock();
293 0 : ulong backoff_base = 1UL << fd_ulong_min( client->consecutive_failure_count, 7UL ); /* max 4 mins */
294 0 : ulong backoff_jitter = fd_rng_ulong_roll( client->rng, backoff_base );
295 0 : client->disconnected.reconnect_deadline = now + (long)( backoff_base + backoff_jitter )*(long)1e9;
296 0 : if( FD_UNLIKELY( client->consecutive_failure_count < 8UL ) ) client->consecutive_failure_count++;
297 0 : }
298 :
299 : static void
300 : disconnect( fd_event_client_t * client,
301 : int reason,
302 : int err,
303 0 : int _backoff ) {
304 0 : #if FD_HAS_OPENSSL
305 0 : if( FD_UNLIKELY( client->ssl ) ) {
306 0 : SSL_free( client->ssl );
307 0 : client->ssl = NULL;
308 0 : }
309 0 : #endif
310 0 : if( FD_LIKELY( -1!=client->sockfd ) ) {
311 0 : if( FD_UNLIKELY( -1==close( client->sockfd ) ) ) FD_LOG_ERR(( "close() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
312 0 : client->sockfd = -1;
313 0 : client->state = FD_EVENT_CLIENT_STATE_DISCONNECTED;
314 0 : fd_circq_reset_cursor( client->circq );
315 0 : }
316 :
317 0 : client->event_stream = NULL;
318 0 : client->auth_deadline = LONG_MAX;
319 0 : client->auth_send_pending = 0;
320 :
321 0 : client->auth_bearer[ 0 ] = '\0';
322 0 : client->auth_bearer_len = 0UL;
323 :
324 0 : switch( reason ) {
325 0 : case DISCONNECT_REASON_IDENTITY_CHANGED:
326 0 : FD_LOG_INFO(( "disconnected: identity changed" ));
327 0 : break;
328 0 : case DISCONNECT_REASON_CONNECT_FAILED:
329 0 : if( FD_UNLIKELY( !client->connect_fail_logged ) ) FD_LOG_WARNING(( "connecting to telemetry server " FD_IP4_ADDR_FMT ":%u failed %s(%i-%s)%s", FD_IP4_ADDR_FMT_ARGS( client->server_ip4_addr ), client->server_tcp_port, fd_log_style_dim(), errno, fd_io_strerror( errno ), fd_log_style_normal() ));
330 0 : else FD_LOG_INFO(( "connecting to telemetry server " FD_IP4_ADDR_FMT ":%u failed (%i-%s)", FD_IP4_ADDR_FMT_ARGS( client->server_ip4_addr ), client->server_tcp_port, errno, fd_io_strerror( errno ) ));
331 0 : client->connect_fail_logged = 1;
332 0 : client->metrics.transport_fail_cnt++;
333 0 : break;
334 0 : case DISCONNECT_REASON_DNS_RESOLVE_FAILED:
335 0 : if( FD_UNLIKELY( !client->connect_fail_logged ) ) FD_LOG_WARNING(( "failed to resolve telemetry server host %.*s %s(%d-%s)%s", (int)client->server_fqdn_len, client->server_fqdn, fd_log_style_dim(), err, fd_gai_strerror( err ), fd_log_style_normal() ));
336 0 : else FD_LOG_INFO(( "failed to resolve telemetry server host %.*s (%d-%s)", (int)client->server_fqdn_len, client->server_fqdn, err, fd_gai_strerror( err ) ));
337 0 : client->connect_fail_logged = 1;
338 0 : client->metrics.transport_fail_cnt++;
339 0 : break;
340 0 : case DISCONNECT_REASON_TIMEOUT:
341 0 : FD_LOG_INFO(( "connection failed: timeout" ));
342 0 : client->metrics.transport_fail_cnt++;
343 0 : break;
344 0 : case DISCONNECT_REASON_TRANSPORT_FAILED:
345 0 : FD_LOG_WARNING(( "disconnected from telemetry server: transport failed %s(%d-%s)%s", fd_log_style_dim(), err, fd_io_strerror( err ), fd_log_style_normal() ));
346 0 : client->metrics.transport_fail_cnt++;
347 0 : break;
348 0 : case DISCONNECT_REASON_PEER_CLOSED:
349 0 : FD_LOG_WARNING(( "disconnected from telemetry server: peer closed connection" ));
350 0 : client->metrics.transport_fail_cnt++;
351 0 : break;
352 0 : case DISCONNECT_REASON_INVALID_CURSOR:
353 0 : FD_LOG_WARNING(( "disconnected from telemetry server: invalid cursor" ));
354 0 : client->metrics.transport_fail_cnt++;
355 0 : break;
356 0 : case DISCONNECT_REASON_AUTH_FAILED:
357 0 : FD_LOG_WARNING(( "disconnected from telemetry server: authentication failed" ));
358 0 : client->metrics.transport_fail_cnt++;
359 0 : break;
360 0 : case DISCONNECT_REASON_INVALID_PROTOBUF:
361 0 : FD_LOG_WARNING(( "disconnected from telemetry server: invalid protobuf message received" ));
362 0 : client->metrics.transport_fail_cnt++;
363 0 : break;
364 0 : default:
365 0 : FD_LOG_WARNING(( "disconnected from telemetry server: unknown reason %d", reason ));
366 0 : client->metrics.transport_fail_cnt++;
367 0 : break;
368 0 : }
369 :
370 0 : if( FD_LIKELY( _backoff ) ) backoff( client );
371 0 : }
372 :
373 : void
374 : fd_event_client_set_identity( fd_event_client_t * client,
375 0 : uchar const * identity_pubkey ) {
376 0 : fd_memcpy( client->identity_pubkey, identity_pubkey, 32UL );
377 0 : disconnect( client, DISCONNECT_REASON_IDENTITY_CHANGED, 0, 0 );
378 0 : }
379 :
380 : static void
381 : reconnect( fd_event_client_t * client,
382 0 : int * charge_busy ) {
383 0 : FD_TEST( client->state==FD_EVENT_CLIENT_STATE_DISCONNECTED );
384 :
385 0 : long now = fd_log_wallclock();
386 0 : if( FD_UNLIKELY( now<client->disconnected.reconnect_deadline ) ) return;
387 :
388 0 : *charge_busy = 1;
389 0 : client->metrics.connect_attempt_cnt++;
390 :
391 0 : FD_LOG_INFO(( "connecting to event server %s://%.*s:%u", client->use_tls ? "https" : "http", (int)client->server_fqdn_len, client->server_fqdn, client->server_tcp_port ));
392 :
393 : /* FIXME IPv6 support */
394 0 : fd_addrinfo_t hints = {0};
395 0 : hints.ai_family = AF_INET;
396 0 : fd_addrinfo_t * res = NULL;
397 0 : uchar scratch[ 4096 ];
398 0 : void * pscratch = scratch;
399 0 : int err = fd_getaddrinfo( client->server_fqdn, &hints, &res, &pscratch, sizeof(scratch) );
400 0 : if( FD_UNLIKELY( err ) ) {
401 0 : disconnect( client, DISCONNECT_REASON_DNS_RESOLVE_FAILED, err, 1 );
402 0 : return;
403 0 : }
404 :
405 0 : if( FD_UNLIKELY( !res || !res->ai_addr ) ) {
406 0 : disconnect( client, DISCONNECT_REASON_DNS_RESOLVE_FAILED, 0, 1 );
407 0 : return;
408 0 : }
409 :
410 0 : uint const ip4_addr = ((struct sockaddr_in *)res->ai_addr)->sin_addr.s_addr;
411 0 : client->server_ip4_addr = ip4_addr;
412 :
413 0 : client->sockfd = socket( AF_INET, SOCK_STREAM|SOCK_NONBLOCK, 0 );
414 0 : if( FD_UNLIKELY( -1==client->sockfd ) ) FD_LOG_ERR(( "socket() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
415 :
416 0 : struct sockaddr_in addr;
417 0 : fd_memset( &addr, 0, sizeof( addr ) );
418 0 : addr.sin_family = AF_INET;
419 0 : addr.sin_port = fd_ushort_bswap( client->server_tcp_port );
420 0 : addr.sin_addr.s_addr = ip4_addr;
421 :
422 0 : int tcp_nodelay = 1;
423 0 : if( FD_UNLIKELY( -1==setsockopt( client->sockfd, SOL_TCP, TCP_NODELAY, &tcp_nodelay, sizeof(int) ) ) ) FD_LOG_ERR(( "setsockopt failed (%d-%s)", errno, fd_io_strerror( errno ) ));
424 0 : if( FD_UNLIKELY( -1==setsockopt( client->sockfd, SOL_SOCKET, SO_SNDBUF, &client->so_sndbuf, sizeof(int) ) ) ) FD_LOG_ERR(( "setsockopt(SOL_SOCKET,SO_SNDBUF,%i) failed (%i-%s)", client->so_sndbuf, errno, fd_io_strerror( errno ) ));
425 :
426 0 : if( FD_UNLIKELY( -1==connect( client->sockfd, fd_type_pun_const( &addr ), sizeof(struct sockaddr_in) ) && errno!=EINPROGRESS ) ) {
427 0 : disconnect( client, DISCONNECT_REASON_CONNECT_FAILED, errno, 1 );
428 0 : return;
429 0 : }
430 :
431 0 : # if FD_HAS_OPENSSL
432 0 : if( client->use_tls ) {
433 0 : BIO * bio = fd_openssl_bio_new_socket( client->sockfd, BIO_NOCLOSE );
434 0 : if( FD_UNLIKELY( !bio ) ) {
435 0 : FD_LOG_WARNING(( "fd_openssl_bio_new_socket failed" ));
436 0 : disconnect( client, DISCONNECT_REASON_CONNECT_FAILED, 0, 1 );
437 0 : return;
438 0 : }
439 :
440 0 : SSL * ssl = SSL_new( client->ssl_ctx );
441 0 : if( FD_UNLIKELY( !ssl ) ) {
442 0 : FD_LOG_WARNING(( "SSL_new failed" ));
443 0 : BIO_free( bio );
444 0 : disconnect( client, DISCONNECT_REASON_CONNECT_FAILED, 0, 1 );
445 0 : return;
446 0 : }
447 :
448 0 : SSL_set_bio( ssl, bio, bio ); /* moves ownership of bio */
449 0 : SSL_set_connect_state( ssl );
450 :
451 : /* SNI and hostname verification */
452 0 : if( FD_UNLIKELY( !SSL_set_tlsext_host_name( ssl, client->server_fqdn ) ) ) {
453 0 : FD_LOG_WARNING(( "SSL_set_tlsext_host_name failed" ));
454 0 : SSL_free( ssl );
455 0 : disconnect( client, DISCONNECT_REASON_CONNECT_FAILED, 0, 1 );
456 0 : return;
457 0 : }
458 0 : if( FD_UNLIKELY( !SSL_set1_host( ssl, client->server_fqdn ) ) ) {
459 0 : FD_LOG_WARNING(( "SSL_set1_host failed" ));
460 0 : SSL_free( ssl );
461 0 : disconnect( client, DISCONNECT_REASON_CONNECT_FAILED, 0, 1 );
462 0 : return;
463 0 : }
464 :
465 0 : client->ssl = ssl;
466 0 : }
467 0 : # endif /* FD_HAS_OPENSSL */
468 :
469 0 : fd_grpc_client_reset( client->grpc_client );
470 :
471 0 : client->state = FD_EVENT_CLIENT_STATE_CONNECTING;
472 0 : client->connecting.connect_deadline = now+(long)1L*(long)1e9; /* 1 second to connect */
473 0 : }
474 :
475 : static int
476 0 : fd_event_client_try_send_authenticate( fd_event_client_t * client ) {
477 0 : if( FD_UNLIKELY( fd_grpc_client_request_is_blocked( client->grpc_client ) ) ) return 0;
478 0 : if( FD_UNLIKELY( fd_grpc_client_request_stream_busy( client->grpc_client ) ) ) return 0;
479 :
480 0 : fd_pb_encoder_t auth_req[1];
481 0 : uchar buffer[ 256UL ];
482 0 : fd_pb_encoder_init( auth_req, buffer, sizeof(buffer) );
483 :
484 0 : fd_pb_push_bytes( auth_req, 1U, client->identity_pubkey, 32UL );
485 0 : fd_pb_push_string( auth_req, 2U, client->client_version, strlen( client->client_version ) );
486 0 : fd_pb_push_string( auth_req, 3U, client->commit_hash, strlen( client->commit_hash ) );
487 0 : fd_pb_push_bytes( auth_req, 4U, client->genesis_hash, 32UL );
488 0 : fd_pb_push_uint64( auth_req, 5U, client->shred_version );
489 0 : fd_pb_push_uint64( auth_req, 6U, client->instance_id );
490 0 : fd_pb_push_uint64( auth_req, 7U, client->machine_id );
491 0 : fd_pb_push_uint64( auth_req, 8U, client->boot_id );
492 0 : fd_pb_push_string( auth_req, 9U, client->action, strlen( client->action ) );
493 :
494 0 : fd_grpc_h2_stream_t * stream = fd_grpc_client_request_start1(
495 0 : client->grpc_client,
496 0 : "/events.v1.EventService/Authenticate", strlen("/events.v1.EventService/Authenticate"),
497 0 : FD_EVENT_CLIENT_REQ_CTX_AUTHENTICATE,
498 0 : buffer, fd_pb_encoder_out_sz( auth_req ),
499 0 : NULL, 0UL,
500 0 : 0 /* not streaming */ );
501 :
502 0 : if( FD_UNLIKELY( !stream ) ) return 0;
503 :
504 0 : long now = fd_log_wallclock();
505 0 : fd_grpc_client_deadline_set( stream, FD_GRPC_DEADLINE_HEADER, now+(long)2e9 );
506 0 : fd_grpc_client_deadline_set( stream, FD_GRPC_DEADLINE_RX_END, now+(long)2e9 );
507 :
508 0 : client->auth_send_pending = 0;
509 0 : FD_LOG_INFO(( "Requesting auth challenge from event server " FD_IP4_ADDR_FMT ":%u (%.*s)",
510 0 : FD_IP4_ADDR_FMT_ARGS( client->server_ip4_addr ), client->server_tcp_port,
511 0 : (int)client->server_fqdn_len, client->server_fqdn ));
512 0 : return 1;
513 0 : }
514 :
515 : static void
516 0 : fd_event_client_grpc_conn_established( void * app_ctx ) {
517 0 : fd_event_client_t * client = app_ctx;
518 :
519 0 : long now = fd_log_wallclock();
520 0 : client->state = FD_EVENT_CLIENT_STATE_AUTHENTICATING;
521 0 : client->auth_deadline = now + (long)2e9;
522 0 : client->auth_send_pending = 1;
523 :
524 0 : fd_event_client_try_send_authenticate( client );
525 0 : }
526 :
527 : static void
528 : fd_event_client_handle_auth_challenge_resp( fd_event_client_t * client,
529 : void const * protobuf,
530 0 : ulong protobuf_sz ) {
531 0 : fd_pb_inbuf_t inbuf[1];
532 0 : fd_pb_inbuf_init( inbuf, protobuf, protobuf_sz );
533 :
534 0 : if( FD_UNLIKELY( protobuf_sz==0UL ) ) {
535 0 : FD_LOG_WARNING(( "Empty auth challenge response" ));
536 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
537 0 : return;
538 0 : }
539 :
540 0 : fd_pb_tlv_t challenge_tlv;
541 0 : if( FD_UNLIKELY( !fd_pb_read_tlv( inbuf, &challenge_tlv ) ) ) {
542 0 : FD_LOG_WARNING(( "Failed to parse auth challenge response" ));
543 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
544 0 : return;
545 0 : }
546 :
547 0 : if( FD_UNLIKELY( challenge_tlv.field_id!=1U || challenge_tlv.wire_type!=FD_PB_WIRE_TYPE_LEN ) ) {
548 0 : FD_LOG_WARNING(( "Unexpected field in auth challenge response" ));
549 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
550 0 : return;
551 0 : }
552 :
553 0 : ulong challenge_len = challenge_tlv.len;
554 0 : if( FD_UNLIKELY( challenge_len!=FD_EVENT_CLIENT_TOKEN_SZ ) ) {
555 0 : FD_LOG_WARNING(( "Invalid challenge token size: %lu bytes (expected %lu)", challenge_len, FD_EVENT_CLIENT_TOKEN_SZ ));
556 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
557 0 : return;
558 0 : }
559 :
560 0 : if( FD_UNLIKELY( fd_pb_inbuf_sz( inbuf )<challenge_len ) ) {
561 0 : FD_LOG_WARNING(( "Truncated auth challenge response" ));
562 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
563 0 : return;
564 0 : }
565 :
566 0 : uchar challenge_token[ FD_EVENT_CLIENT_TOKEN_SZ ];
567 0 : memcpy( challenge_token, inbuf->cur, challenge_len );
568 0 : inbuf->cur += challenge_len;
569 :
570 0 : if( FD_UNLIKELY( fd_pb_inbuf_sz( inbuf ) ) ) {
571 0 : FD_LOG_WARNING(( "Trailing data in auth challenge response" ));
572 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
573 0 : return;
574 0 : }
575 :
576 0 : uchar sign_request[ 100UL + FD_EVENT_CLIENT_TOKEN_SZ ];
577 0 : static char const sign_prefix[ 100 ] =
578 0 : " " /* 32 spaces */
579 0 : " " /* 32 spaces */
580 0 : "Firedancer event challenge-response";
581 0 : memcpy( sign_request, sign_prefix, sizeof(sign_prefix) );
582 0 : memcpy( sign_request+100, challenge_token, challenge_len );
583 :
584 0 : uchar signature[ 64UL ];
585 0 : fd_keyguard_client_sign( client->keyguard_client,
586 0 : signature,
587 0 : sign_request, 100UL+challenge_len,
588 0 : FD_KEYGUARD_SIGN_TYPE_ED25519 );
589 :
590 : /* Build "hex(challenge_token).hex(signature)" for the bearer token. */
591 0 : fd_hex_encode( client->auth_bearer, challenge_token, FD_EVENT_CLIENT_TOKEN_SZ );
592 0 : client->auth_bearer[ 2UL*FD_EVENT_CLIENT_TOKEN_SZ ] = '.';
593 0 : fd_hex_encode( client->auth_bearer + 2UL*FD_EVENT_CLIENT_TOKEN_SZ+1UL, signature, 64UL );
594 0 : client->auth_bearer_len = 2UL*FD_EVENT_CLIENT_TOKEN_SZ + 1UL + 2UL*64UL;
595 0 : client->auth_bearer[ client->auth_bearer_len ] = '\0';
596 :
597 0 : client->event_stream = NULL;
598 0 : client->metrics.transport_success_cnt++;
599 0 : client->state = FD_EVENT_CLIENT_STATE_CONNECTED;
600 0 : client->connected.connected_timestamp = fd_log_wallclock();
601 0 : client->connect_fail_logged = 0;
602 0 : FD_LOG_NOTICE(( "connected to telemetry server %s%s://%.*s:%u%s",
603 0 : fd_log_style_bold(), client->use_tls ? "https" : "http",
604 0 : (int)client->server_fqdn_len, client->server_fqdn, client->server_tcp_port,
605 0 : fd_log_style_normal() ));
606 0 : }
607 :
608 : static void
609 : fd_event_client_grpc_conn_dead( void * app_ctx,
610 : uint h2_err,
611 0 : int closed_by ) {
612 0 : fd_event_client_t * client = app_ctx;
613 0 : FD_LOG_WARNING(( "Event gRPC connection closed %s (%u-%s)",
614 0 : closed_by ? "by peer" : "due to error",
615 0 : h2_err, fd_h2_strerror( h2_err ) ));
616 0 : client->defer_disconnect = DISCONNECT_REASON_PEER_CLOSED;
617 0 : }
618 :
619 : static void
620 : fd_event_client_grpc_tx_complete( void * app_ctx,
621 0 : ulong request_ctx ) {
622 0 : (void)app_ctx; (void)request_ctx;
623 0 : }
624 :
625 : void
626 : fd_event_client_grpc_rx_start( void * app_ctx,
627 0 : ulong request_ctx ) {
628 0 : (void)app_ctx; (void)request_ctx;
629 0 : }
630 :
631 : static void
632 : fd_event_client_handle_stream_events_resp( fd_event_client_t * client,
633 : void const * protobuf,
634 0 : ulong protobuf_sz ) {
635 0 : fd_pb_inbuf_t inbuf[1];
636 0 : fd_pb_inbuf_init( inbuf, protobuf, protobuf_sz );
637 :
638 0 : ulong nonce_ack = 0UL;
639 0 : if( FD_LIKELY( protobuf_sz ) ) {
640 0 : fd_pb_tlv_t event_id;
641 0 : if( FD_UNLIKELY( !fd_pb_read_tlv( inbuf, &event_id ) ||
642 0 : event_id.field_id!=1U /* event_id */ ||
643 0 : event_id.wire_type!=FD_PB_WIRE_TYPE_VARINT ) ) {
644 0 : FD_LOG_WARNING(( "Event gRPC rx msg: invalid Protobuf" ));
645 0 : client->defer_disconnect = DISCONNECT_REASON_INVALID_PROTOBUF;
646 0 : return;
647 0 : }
648 0 : nonce_ack = event_id.varint;
649 :
650 0 : if( FD_UNLIKELY( fd_pb_inbuf_sz( inbuf ) ) ) {
651 0 : FD_LOG_WARNING(( "Event gRPC rx msg: trailing data in StreamEventsResponse" ));
652 0 : client->defer_disconnect = DISCONNECT_REASON_INVALID_PROTOBUF;
653 0 : return;
654 0 : }
655 0 : }
656 :
657 0 : client->metrics.events_acked++;
658 0 : if( FD_UNLIKELY( nonce_ack==ULONG_MAX ) ) return;
659 :
660 0 : client->metrics.last_acked_id = nonce_ack;
661 :
662 0 : int err = fd_circq_pop_until( client->circq, nonce_ack );
663 0 : if( FD_UNLIKELY( -1==err ) ) {
664 0 : FD_LOG_WARNING(( "Event gRPC rx msg: invalid cursor ack %lu", nonce_ack ));
665 0 : client->defer_disconnect = DISCONNECT_REASON_INVALID_CURSOR;
666 0 : }
667 0 : }
668 :
669 : void
670 : fd_event_client_grpc_rx_msg( void * app_ctx,
671 : void const * protobuf,
672 : ulong protobuf_sz,
673 0 : ulong request_ctx ) {
674 0 : fd_event_client_t * client = app_ctx;
675 :
676 0 : switch( request_ctx ) {
677 0 : case FD_EVENT_CLIENT_REQ_CTX_AUTHENTICATE:
678 0 : fd_event_client_handle_auth_challenge_resp( client, protobuf, protobuf_sz );
679 0 : break;
680 0 : case FD_EVENT_CLIENT_REQ_CTX_STREAM_EVENTS:
681 0 : fd_event_client_handle_stream_events_resp( client, protobuf, protobuf_sz );
682 0 : break;
683 0 : default:
684 0 : FD_LOG_WARNING(( "Unknown request_ctx: %lu, disconnecting", request_ctx ));
685 0 : client->defer_disconnect = DISCONNECT_REASON_INVALID_PROTOBUF;
686 0 : break;
687 0 : }
688 0 : }
689 :
690 : void
691 : fd_event_client_grpc_rx_end( void * app_ctx,
692 : ulong request_ctx,
693 0 : fd_grpc_resp_hdrs_t * resp ) {
694 0 : fd_event_client_t * client = app_ctx;
695 :
696 0 : if( FD_UNLIKELY( resp->h2_status!=200 ) ) {
697 0 : FD_LOG_WARNING(( "telemetry server request failed %s(HTTP status %u)%s", fd_log_style_dim(), resp->h2_status, fd_log_style_normal() ));
698 0 : client->defer_disconnect = DISCONNECT_REASON_TRANSPORT_FAILED;
699 0 : return;
700 0 : }
701 :
702 0 : resp->grpc_msg_len = (uint)fd_url_unescape( resp->grpc_msg, resp->grpc_msg_len );
703 0 : if( !resp->grpc_msg_len ) {
704 0 : fd_memcpy( resp->grpc_msg, "unknown error", 13 );
705 0 : resp->grpc_msg_len = 13;
706 0 : }
707 :
708 0 : if( FD_UNLIKELY( resp->grpc_status!=FD_GRPC_STATUS_OK ) ) {
709 0 : switch( request_ctx ) {
710 0 : case FD_EVENT_CLIENT_REQ_CTX_AUTHENTICATE:
711 0 : FD_LOG_WARNING(( "telemetry server authentication failed: %.*s %s(%u-%s)%s",
712 0 : (int)resp->grpc_msg_len, resp->grpc_msg,
713 0 : fd_log_style_dim(), resp->grpc_status, fd_grpc_status_cstr( resp->grpc_status ), fd_log_style_normal() ));
714 0 : client->defer_disconnect = DISCONNECT_REASON_AUTH_FAILED;
715 0 : return;
716 0 : case FD_EVENT_CLIENT_REQ_CTX_STREAM_EVENTS:
717 0 : FD_LOG_WARNING(( "telemetry server event stream failed: %.*s %s(%u-%s)%s",
718 0 : (int)resp->grpc_msg_len, resp->grpc_msg,
719 0 : fd_log_style_dim(), resp->grpc_status, fd_grpc_status_cstr( resp->grpc_status ), fd_log_style_normal() ));
720 0 : client->defer_disconnect = DISCONNECT_REASON_PEER_CLOSED;
721 0 : return;
722 0 : default:
723 0 : FD_LOG_WARNING(( "telemetry server request failed: %.*s %s(%u-%s)%s",
724 0 : (int)resp->grpc_msg_len, resp->grpc_msg,
725 0 : fd_log_style_dim(), resp->grpc_status, fd_grpc_status_cstr( resp->grpc_status ), fd_log_style_normal() ));
726 0 : client->defer_disconnect = DISCONNECT_REASON_TRANSPORT_FAILED;
727 0 : return;
728 0 : }
729 0 : }
730 :
731 0 : if( request_ctx==FD_EVENT_CLIENT_REQ_CTX_STREAM_EVENTS ) {
732 0 : FD_LOG_INFO(( "telemetry server event stream ended gracefully" ));
733 0 : client->defer_disconnect = DISCONNECT_REASON_PEER_CLOSED;
734 0 : }
735 0 : }
736 :
737 : void
738 : fd_event_client_grpc_rx_timeout( void * app_ctx,
739 : ulong request_ctx FD_PARAM_UNUSED,
740 0 : int deadline_kind FD_PARAM_UNUSED ) {
741 0 : FD_LOG_WARNING(( "Event gRPC rx timeout" ));
742 0 : fd_event_client_t * client = (fd_event_client_t *)app_ctx;
743 0 : client->defer_disconnect = DISCONNECT_REASON_TRANSPORT_FAILED;
744 0 : client->event_stream = NULL;
745 0 : }
746 :
747 : static void
748 0 : fd_event_client_grpc_ping_ack( void * app_ctx ) {
749 0 : (void)app_ctx;
750 0 : FD_LOG_WARNING(( "Event gRPC ping ack" ));
751 0 : }
752 :
753 : static void
754 : tx( fd_event_client_t * client,
755 0 : int * charge_busy ) {
756 0 : FD_TEST( client->state==FD_EVENT_CLIENT_STATE_CONNECTED );
757 :
758 0 : if( FD_UNLIKELY( fd_grpc_client_request_is_blocked( client->grpc_client ) ) ) return;
759 0 : if( FD_UNLIKELY( client->event_stream && client->grpc_client->request_stream != NULL && client->grpc_client->request_stream!=client->event_stream ) ) return;
760 :
761 0 : if( FD_UNLIKELY( !client->event_stream ) ) {
762 0 : client->event_stream = fd_grpc_client_request_start1(
763 0 : client->grpc_client,
764 0 : "/events.v1.EventService/StreamEvents", strlen("/events.v1.EventService/StreamEvents"),
765 0 : FD_EVENT_CLIENT_REQ_CTX_STREAM_EVENTS,
766 0 : NULL, 0UL, /* headers only; first message sent later */
767 0 : client->auth_bearer, client->auth_bearer_len,
768 0 : 1 /* streaming */ );
769 0 : if( FD_UNLIKELY( !client->event_stream ) ) return; /* transient; retry next poll */
770 0 : fd_grpc_client_deadline_set( client->event_stream, FD_GRPC_DEADLINE_HEADER, fd_log_wallclock()+(long)10e9 /* 10s */ );
771 0 : client->last_stream_send_ticks = fd_tickcount();
772 0 : *charge_busy = 1;
773 0 : return;
774 0 : }
775 :
776 0 : ulong msg_sz;
777 0 : uchar const * msg = fd_circq_cursor_advance( client->circq, &msg_sz );
778 0 : if( FD_LIKELY( !msg ) ) {
779 : /* Nothing to send. If the stream has been quiet long enough that an
780 : intermediary proxy might kill it, send a zero-length
781 : StreamEventsRequest to heartbeat. */
782 0 : long now_ticks = fd_tickcount();
783 0 : if( FD_UNLIKELY( now_ticks-client->last_stream_send_ticks>client->heartbeat_ticks ) ) {
784 0 : if( FD_LIKELY( fd_grpc_client_stream_send_msg1( client->grpc_client, client->event_stream, (uchar const *)"", 0UL ) ) ) {
785 0 : client->last_stream_send_ticks = now_ticks;
786 0 : *charge_busy = 1;
787 0 : }
788 0 : }
789 0 : return;
790 0 : }
791 :
792 0 : int result = fd_grpc_client_stream_send_msg1( client->grpc_client, client->event_stream, msg, msg_sz );
793 0 : if( FD_UNLIKELY( !result ) ) return; /* Only reason for failure is too big message, so just skip it */
794 :
795 0 : client->metrics.events_sent++;
796 0 : client->last_stream_send_ticks = fd_tickcount();
797 0 : *charge_busy = 1;
798 0 : }
799 :
800 : void
801 : fd_event_client_poll( fd_event_client_t * client,
802 0 : int * charge_busy ) {
803 0 : if( FD_UNLIKELY( !client->has_genesis_hash || !client->has_shred_version ) ) return;
804 :
805 0 : long now = fd_log_wallclock();
806 :
807 0 : if( FD_UNLIKELY( client->state==FD_EVENT_CLIENT_STATE_DISCONNECTED ) ) reconnect( client, charge_busy );
808 0 : if( FD_UNLIKELY( client->state==FD_EVENT_CLIENT_STATE_CONNECTING ) ) {
809 0 : if( FD_UNLIKELY( now>client->connecting.connect_deadline ) ) {
810 0 : disconnect( client, DISCONNECT_REASON_TIMEOUT, 0, 1 );
811 0 : return;
812 0 : }
813 0 : }
814 : /* Check auth handshake timeout */
815 0 : if( FD_UNLIKELY( client->state==FD_EVENT_CLIENT_STATE_AUTHENTICATING && now>client->auth_deadline ) ) {
816 0 : FD_LOG_WARNING(( "auth handshake timed out" ));
817 0 : client->metrics.handshake_timeout_cnt++;
818 0 : disconnect( client, DISCONNECT_REASON_TIMEOUT, 0, 1 );
819 0 : return;
820 0 : }
821 0 : if( FD_LIKELY( client->state!=FD_EVENT_CLIENT_STATE_DISCONNECTED ) ) {
822 0 : int rxtx_err;
823 0 : # if FD_HAS_OPENSSL
824 0 : if( client->use_tls )
825 0 : rxtx_err = fd_grpc_client_rxtx_ossl( client->grpc_client, client->ssl, charge_busy );
826 0 : else
827 0 : # endif
828 0 : rxtx_err = fd_grpc_client_rxtx_socket( client->grpc_client, client->sockfd, charge_busy );
829 0 : if( FD_UNLIKELY( -1==rxtx_err ) ) {
830 0 : disconnect( client, DISCONNECT_REASON_TRANSPORT_FAILED, errno, 1 );
831 0 : return;
832 0 : }
833 0 : }
834 :
835 0 : if( FD_UNLIKELY( client->defer_disconnect!=INT_MAX ) ) {
836 0 : int reason = client->defer_disconnect;
837 0 : client->defer_disconnect = INT_MAX;
838 0 : if( reason==DISCONNECT_REASON_AUTH_FAILED ) client->metrics.auth_fail_cnt++;
839 0 : if( reason==DISCONNECT_REASON_INVALID_PROTOBUF ) client->metrics.invalid_msg_cnt++;
840 0 : disconnect( client, reason, 0, 1 );
841 0 : return;
842 0 : }
843 :
844 0 : if( FD_UNLIKELY( client->state==FD_EVENT_CLIENT_STATE_AUTHENTICATING && client->auth_send_pending ) ) {
845 0 : fd_event_client_try_send_authenticate( client );
846 0 : }
847 :
848 0 : if( FD_LIKELY( client->state==FD_EVENT_CLIENT_STATE_CONNECTED ) ) {
849 0 : if( FD_UNLIKELY( client->consecutive_failure_count && (now-client->connected.connected_timestamp>10L*(long)1e9 ) ) ) client->consecutive_failure_count = 0UL;
850 0 : tx( client, charge_busy );
851 0 : }
852 0 : }
853 :
854 : fd_grpc_client_callbacks_t fd_event_client_grpc_callbacks = {
855 : .conn_established = fd_event_client_grpc_conn_established,
856 : .conn_dead = fd_event_client_grpc_conn_dead,
857 : .tx_complete = fd_event_client_grpc_tx_complete,
858 : .rx_start = fd_event_client_grpc_rx_start,
859 : .rx_msg = fd_event_client_grpc_rx_msg,
860 : .rx_end = fd_event_client_grpc_rx_end,
861 : .rx_timeout = fd_event_client_grpc_rx_timeout,
862 : .ping_ack = fd_event_client_grpc_ping_ack,
863 : };
|
