Context Navigation

source: lib/format_rt.c @ 9231fe5

4.0.1-hotfixescachetimestampsdevelopdpdk-ndagetsilivegetfragoffhelplibtrace4ndag_formatpfringrc-4.0.1rc-4.0.2rc-4.0.3rc-4.0.4ringdecrementfixringperformanceringtimestampfixes

Last change on this file since 9231fe5 was 9231fe5, checked in by Perry Lorier <perry@…>, 15 years ago
Minor cleanups in format_rt Shuffled protocols.c around for better vlan support Cleaned up IPv6 support to be more generic (less duplicated code) added trace_get_{source,destination}_address
Property mode set to `100644`
File size: 17.1 KB

Line
1	/*
2	* This file is part of libtrace
3	*
4	* Copyright (c) 2004 The University of Waikato, Hamilton, New Zealand.
5	* Authors: Daniel Lawson
6	* Perry Lorier
7	* Shane Alcock
8	*
9	* All rights reserved.
10	*
11	* This code has been developed by the University of Waikato WAND
12	* research group. For further information please see http://www.wand.net.nz/
13	*
14	* libtrace is free software; you can redistribute it and/or modify
15	* it under the terms of the GNU General Public License as published by
16	* the Free Software Foundation; either version 2 of the License, or
17	* (at your option) any later version.
18	*
19	* libtrace is distributed in the hope that it will be useful,
20	* but WITHOUT ANY WARRANTY; without even the implied warranty of
21	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
22	* GNU General Public License for more details.
23	*
24	* You should have received a copy of the GNU General Public License
25	* along with libtrace; if not, write to the Free Software
26	* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
27	*
28	* $Id$
29	*
30	*/
31
32	#define _GNU_SOURCE
33
34	#include "config.h"
35	#include "common.h"
36	#include "libtrace.h"
37	#include "libtrace_int.h"
38	#include "format_helper.h"
39	#include "parse_cmd.h"
40	#include "rt_protocol.h"
41
42	#ifdef HAVE_INTTYPES_H
43	# include <inttypes.h>
44	#else
45	# error "Can't find inttypes.h - this needs to be fixed"
46	#endif
47
48	#ifdef HAVE_STDDEF_H
49	# include <stddef.h>
50	#else
51	# error "Can't find stddef.h - do you define ptrdiff_t elsewhere?"
52	#endif
53	#include <sys/types.h>
54	#include <sys/socket.h>
55	#include <sys/un.h>
56	#include <sys/mman.h>
57	#include <sys/stat.h>
58	#include <unistd.h>
59	#include <assert.h>
60	#include <errno.h>
61	#include <netdb.h>
62	#include <fcntl.h>
63	#include <getopt.h>
64	#include <stdio.h>
65	#include <string.h>
66	#include <stdlib.h>
67
68	#define RT_INFO ((struct rt_format_data_t*)libtrace->format_data)
69
70	int reliability = 0;
71
72	char *rt_deny_reason(uint8_t reason) {
73	char *string = 0;
74
75	switch(reason) {
76	case RT_DENY_WRAPPER:
77	string = "Rejected by TCP Wrappers";
78	break;
79	case RT_DENY_FULL:
80	string = "Max connections reached on server";
81	break;
82	case RT_DENY_AUTH:
83	string = "Authentication failed";
84	break;
85	default:
86	string = "Unknown reason";
87	}
88
89	return string;
90	}
91
92
93	struct rt_format_data_t {
94	char *hostname;
95	int port;
96	int input_fd;
97	int reliable;
98	char *pkt_buffer;
99	char *buf_current;
100	int buf_left;
101
102
103	struct libtrace_t *dummy_erf;
104	struct libtrace_t *dummy_pcap;
105	struct libtrace_t *dummy_wag;
106	};
107
108	static struct libtrace_format_t rt;
109
110	static int rt_connect(struct libtrace_t *libtrace) {
111	struct hostent *he;
112	struct sockaddr_in remote;
113	rt_header_t connect_msg;
114	rt_deny_conn_t deny_hdr;
115	rt_hello_t hello_opts;
116	uint8_t reason;
117	int oldflags;
118
119
120	if ((he=gethostbyname(RT_INFO->hostname)) == NULL) {
121	perror("gethostbyname");
122	return -1;
123	}
124	if ((RT_INFO->input_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
125	perror("socket");
126	return -1;
127	}
128
129	remote.sin_family = AF_INET;
130	remote.sin_port = htons(RT_INFO->port);
131	remote.sin_addr = ((struct in_addr )he->h_addr);
132	bzero(&(remote.sin_zero), 8);
133
134	if (connect(RT_INFO->input_fd, (struct sockaddr *)&remote,
135	sizeof(struct sockaddr)) == -1) {
136	trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
137	"Could not connect to host %s on port %d",
138	RT_INFO->hostname, RT_INFO->port);
139	return -1;
140	}
141
142
143	#if 0
144	oldflags = fcntl(RT_INFO->input_fd, F_GETFL, 0);
145	if (oldflags == -1) {
146	trace_set_err(libtrace, errno,
147	"Could not get fd flags from fd %d\n",
148	RT_INFO->input_fd);
149	return -1;
150	}
151	oldflags \|= O_NONBLOCK;
152	if (fcntl(RT_INFO->input_fd, F_SETFL, oldflags) == -1) {
153	trace_set_err(libtrace, errno,
154	"Could not set fd flags for fd %d\n",
155	RT_INFO->input_fd);
156	return -1;
157	}
158	#endif
159
160
161	/* We are connected, now receive message from server */
162
163	if (recv(RT_INFO->input_fd, &connect_msg, sizeof(rt_header_t), 0) != sizeof(rt_header_t) ) {
164	trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
165	"Could not receive connection message from %s",
166	RT_INFO->hostname);
167	return -1;
168	}
169
170	switch (connect_msg.type) {
171	case RT_DENY_CONN:
172
173	if (recv(RT_INFO->input_fd, &deny_hdr,
174	sizeof(rt_deny_conn_t),
175	0) != sizeof(rt_deny_conn_t)) {
176	reason = 0;
177	}
178	reason = deny_hdr.reason;
179	trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
180	"Connection attempt is denied: %s",
181	rt_deny_reason(reason));
182	return -1;
183	case RT_HELLO:
184	/* do something with options */
185	if (recv(RT_INFO->input_fd, &hello_opts,
186	sizeof(rt_hello_t), 0)
187	!= sizeof(rt_hello_t)) {
188	trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
189	"Failed to receive RT_HELLO options");
190	return -1;
191	}
192	reliability = hello_opts.reliable;
193
194	return 0;
195	default:
196	trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
197	"Unknown message type received: %d",
198	connect_msg.type);
199	return -1;
200	}
201	trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
202	"Somehow you managed to reach this unreachable code");
203	return -1;
204	}
205
206
207	static int rt_init_input(struct libtrace_t *libtrace) {
208	char *scan;
209	char *uridata = libtrace->uridata;
210	libtrace->format_data = malloc(sizeof(struct rt_format_data_t));
211
212	RT_INFO->dummy_erf = NULL;
213	RT_INFO->dummy_pcap = NULL;
214	RT_INFO->dummy_wag = NULL;
215	RT_INFO->pkt_buffer = NULL;
216	RT_INFO->buf_current = NULL;
217	RT_INFO->buf_left = 0;
218
219	if (strlen(uridata) == 0) {
220	RT_INFO->hostname =
221	strdup("localhost");
222	RT_INFO->port =
223	COLLECTOR_PORT;
224	} else {
225	if ((scan = strchr(uridata,':')) == NULL) {
226	RT_INFO->hostname =
227	strdup(uridata);
228	RT_INFO->port =
229	COLLECTOR_PORT;
230	} else {
231	RT_INFO->hostname =
232	(char *)strndup(uridata,
233	(scan - uridata));
234	RT_INFO->port =
235	atoi(++scan);
236	}
237	}
238
239	return rt_connect(libtrace);
240	}
241
242	static int rt_start_input(struct libtrace_t *libtrace) {
243	rt_header_t start_msg;
244
245	start_msg.type = RT_START;
246	start_msg.length = sizeof(rt_start_t);
247
248
249	/* Need to send start message to server */
250	if (send(RT_INFO->input_fd, &start_msg, sizeof(rt_header_t) +
251	start_msg.length, 0) != sizeof(rt_header_t)) {
252	printf("Failed to send start message to server\n");
253	return -1;
254	}
255
256	return 0;
257	}
258
259	static int rt_fin_input(struct libtrace_t *libtrace) {
260	rt_header_t close_msg;
261
262	close_msg.type = RT_CLOSE;
263	close_msg.length = sizeof(rt_close_t);
264
265	/* Send a close message to the server */
266	if (send(RT_INFO->input_fd, &close_msg, sizeof(rt_header_t) +
267	close_msg.length, 0) != sizeof(rt_header_t)
268	+ close_msg.length) {
269	printf("Failed to send close message to server\n");
270
271	}
272	if (RT_INFO->dummy_erf)
273	trace_destroy_dead(RT_INFO->dummy_erf);
274
275	if (RT_INFO->dummy_pcap)
276	trace_destroy_dead(RT_INFO->dummy_pcap);
277
278	if (RT_INFO->dummy_wag)
279	trace_destroy_dead(RT_INFO->dummy_wag);
280	close(RT_INFO->input_fd);
281	free(libtrace->format_data);
282	return 0;
283	}
284
285	#define RT_BUF_SIZE 4000
286
287	static int rt_read(struct libtrace_t libtrace, void *buffer, size_t len, int block) {
288	int numbytes;
289	int i;
290	char *buf_ptr;
291
292	assert(len <= RT_BUF_SIZE);
293
294	if (!RT_INFO->pkt_buffer) {
295	RT_INFO->pkt_buffer = malloc(RT_BUF_SIZE);
296	RT_INFO->buf_current = RT_INFO->pkt_buffer;
297	RT_INFO->buf_left = 0;
298	}
299
300	if (block)
301	block=0;
302	else
303	block=MSG_DONTWAIT;
304
305
306	if (len > RT_INFO->buf_left) {
307	memcpy(RT_INFO->pkt_buffer, RT_INFO->buf_current,
308	RT_INFO->buf_left);
309	RT_INFO->buf_current = RT_INFO->pkt_buffer;
310
311	#ifndef MSG_NOSIGNAL
312	# define MSG_NOSIGNAL 0
313	#endif
314	while (len > RT_INFO->buf_left) {
315	if ((numbytes = recv(RT_INFO->input_fd,
316	RT_INFO->pkt_buffer +
317	RT_INFO->buf_left,
318	RT_BUF_SIZE-RT_INFO->buf_left,
319	MSG_NOSIGNAL\|block)) <= 0) {
320	if (numbytes == 0) {
321	trace_set_err(libtrace, TRACE_ERR_BAD_PACKET,
322	"No data received");
323	return -1;
324	}
325
326	if (errno == EINTR) {
327	/* ignore EINTR in case
328	* a caller is using signals
329	*/
330	continue;
331	}
332	if (errno == EAGAIN) {
333	trace_set_err(libtrace,
334	EAGAIN,
335	"EAGAIN");
336	return -1;
337	}
338
339	perror("recv");
340	trace_set_err(libtrace, TRACE_ERR_RECV_FAILED,
341	"Failed to read data into rt recv buffer");
342	return -1;
343	}
344	/*
345	buf_ptr = RT_INFO->pkt_buffer;
346	for (i = 0; i < RT_BUF_SIZE ; i++) {
347
348	printf("%02x", (unsigned char)*buf_ptr);
349	buf_ptr ++;
350	}
351	printf("\n");
352	*/
353	RT_INFO->buf_left+=numbytes;
354	}
355
356	}
357	*buffer = RT_INFO->buf_current;
358	RT_INFO->buf_current += len;
359	RT_INFO->buf_left -= len;
360	assert(RT_INFO->buf_left >= 0);
361	return len;
362	}
363
364
365	static int rt_set_format(libtrace_t libtrace, libtrace_packet_t packet)
366	{
367
368	if (packet->type >= RT_DATA_PCAP) {
369	if (!RT_INFO->dummy_pcap) {
370	RT_INFO->dummy_pcap = trace_create_dead("pcap:-");
371	}
372	packet->trace = RT_INFO->dummy_pcap;
373	return 0;
374	}
375
376	switch (packet->type) {
377	case RT_DATA_ERF:
378	if (!RT_INFO->dummy_erf) {
379	RT_INFO->dummy_erf = trace_create_dead("erf:-");
380	}
381	packet->trace = RT_INFO->dummy_erf;
382	break;
383	case RT_DATA_WAG:
384	if (!RT_INFO->dummy_wag) {
385	RT_INFO->dummy_wag = trace_create_dead("wtf:-");
386	}
387	packet->trace = RT_INFO->dummy_wag;
388	break;
389	case RT_DATA_LEGACY_ETH:
390	case RT_DATA_LEGACY_ATM:
391	case RT_DATA_LEGACY_POS:
392	printf("Sending legacy over RT is currently not supported\n");
393	trace_set_err(libtrace, TRACE_ERR_BAD_PACKET, "Legacy packet cannot be sent over rt");
394	return -1;
395	default:
396	printf("Unrecognised format: %d\n", packet->type);
397	trace_set_err(libtrace, TRACE_ERR_BAD_PACKET, "Unrecognised packet format");
398	return -1;
399	}
400	return 0; /* success */
401	}
402
403	static void rt_set_payload(struct libtrace_packet_t *packet) {
404	dag_record_t *erfptr;
405
406	switch (packet->type) {
407	case RT_DATA_ERF:
408	erfptr = (dag_record_t *)packet->header;
409
410	if (erfptr->flags.rxerror == 1) {
411	packet->payload = NULL;
412	break;
413	}
414	/* else drop into the default case */
415	default:
416	packet->payload = (char *)packet->buffer +
417	trace_get_framing_length(packet);
418	break;
419	}
420	}
421
422	static int rt_send_ack(struct libtrace_t *libtrace,
423	uint32_t seqno) {
424
425	static char *ack_buffer = 0;
426	char *buf_ptr;
427	int numbytes = 0;
428	int to_write = 0;
429	rt_header_t *hdr;
430	rt_ack_t *ack_hdr;
431
432	if (!ack_buffer) {
433	ack_buffer = malloc(sizeof(rt_header_t) + sizeof(rt_ack_t));
434	}
435
436	hdr = (rt_header_t *) ack_buffer;
437	ack_hdr = (rt_ack_t *) (ack_buffer + sizeof(rt_header_t));
438
439	hdr->type = RT_ACK;
440	hdr->length = sizeof(rt_ack_t);
441
442	ack_hdr->sequence = seqno;
443
444	to_write = hdr->length + sizeof(rt_header_t);
445	buf_ptr = ack_buffer;
446
447
448	while (to_write > 0) {
449	numbytes = send(RT_INFO->input_fd, buf_ptr, to_write, 0);
450	if (numbytes == -1) {
451	if (errno == EINTR \|\| errno == EAGAIN) {
452	continue;
453	}
454	else {
455	printf("Error sending ack\n");
456	trace_set_err(libtrace, TRACE_ERR_BAD_PACKET,
457	"Error sending ack");
458	return -1;
459	}
460	}
461	to_write = to_write - numbytes;
462	buf_ptr = buf_ptr + to_write;
463
464	}
465
466	return 1;
467	}
468
469
470	static int rt_read_packet_versatile(libtrace_t *libtrace,
471	libtrace_packet_t *packet,int blocking) {
472	rt_header_t rt_hdr;
473	rt_header_t *pkt_hdr = &rt_hdr;
474	int pkt_size = 0;
475
476
477	if (packet->buf_control == TRACE_CTRL_EXTERNAL \|\| !packet->buffer) {
478	packet->buf_control = TRACE_CTRL_PACKET;
479	packet->buffer = malloc(LIBTRACE_PACKET_BUFSIZE);
480	}
481
482
483	/* FIXME: Better error handling required */
484	if (rt_read(libtrace, (void **)&pkt_hdr, sizeof(rt_header_t),blocking) !=
485	sizeof(rt_header_t)) {
486	return -1;
487	}
488
489	packet->type = pkt_hdr->type;
490	pkt_size = pkt_hdr->length;
491	packet->size = pkt_hdr->length;
492
493	if (packet->type >= RT_DATA_SIMPLE) {
494	if (rt_read(libtrace, &packet->buffer, pkt_size,1) != pkt_size) {
495	printf("Error receiving packet\n");
496	return -1;
497	}
498	packet->header = packet->buffer;
499
500	if (rt_set_format(libtrace, packet) < 0) {
501	return -1;
502	}
503	rt_set_payload(packet);
504
505	if (reliability > 0) {
506
507	if (rt_send_ack(libtrace, pkt_hdr->sequence)
508	== -1)
509	{
510	return -1;
511	}
512	}
513	} else {
514	switch(packet->type) {
515	case RT_STATUS:
516	case RT_DUCK:
517	if (rt_read(libtrace, &packet->buffer,
518	pkt_size,1) !=
519	pkt_size) {
520	printf("Error receiving status packet\n");
521	return -1;
522	}
523	packet->header = 0;
524	packet->payload = packet->buffer;
525	break;
526	case RT_END_DATA:
527	return 0;
528	case RT_PAUSE_ACK:
529	/* FIXME: Do something useful */
530	break;
531	case RT_OPTION:
532	/* FIXME: Do something useful here as well */
533	break;
534	case RT_KEYCHANGE:
535	break;
536	default:
537	printf("Bad rt type for client receipt: %d\n",
538	pkt_hdr->type);
539	}
540	}
541	/* Return the number of bytes read from the stream */
542	return packet->size;
543	}
544
545	static int rt_read_packet(libtrace_t *libtrace,
546	libtrace_packet_t *packet) {
547	return rt_read_packet_versatile(libtrace,packet,1);
548	}
549
550
551	static int rt_get_capture_length(const struct libtrace_packet_t *packet) {
552	switch (packet->type) {
553	case RT_DUCK:
554	return sizeof(rt_duck_t);
555	case RT_STATUS:
556	return sizeof(rt_status_t);
557	case RT_HELLO:
558	return sizeof(rt_hello_t);
559	case RT_START:
560	return sizeof(rt_start_t);
561	case RT_ACK:
562	return sizeof(rt_ack_t);
563	case RT_END_DATA:
564	return sizeof(rt_end_data_t);
565	case RT_CLOSE:
566	return sizeof(rt_close_t);
567	case RT_DENY_CONN:
568	return sizeof(rt_deny_conn_t);
569	case RT_PAUSE:
570	return sizeof(rt_pause_t);
571	case RT_PAUSE_ACK:
572	return sizeof(rt_pause_ack_t);
573	case RT_OPTION:
574	return sizeof(rt_option_t);
575	case RT_KEYCHANGE:
576	return sizeof(rt_keychange_t);
577	}
578	printf("Unknown type: %d\n", packet->type);
579	return 0;
580	}
581
582	static int rt_get_wire_length(const libtrace_packet_t *packet) {
583	return 0;
584	}
585
586	static int rt_get_framing_length(const libtrace_packet_t *packet) {
587	return 0;
588	}
589
590	static int rt_get_fd(const libtrace_t *trace) {
591	return ((struct rt_format_data_t *)trace->format_data)->input_fd;
592	}
593
594	struct libtrace_eventobj_t trace_event_rt(struct libtrace_t trace, struct libtrace_packet_t packet) {
595	struct libtrace_eventobj_t event = {0,0,0.0,0};
596	libtrace_err_t read_err;
597	int data;
598
599	assert(trace);
600	assert(packet);
601
602	if (trace->format->get_fd) {
603	event.fd = trace->format->get_fd(trace);
604	} else {
605	event.fd = 0;
606	}
607
608	event.size = rt_read_packet_versatile(trace, packet, 0);
609	if (event.size == -1) {
610	read_err = trace_get_err(trace);
611	if (read_err.err_num == EAGAIN) {
612	event.type = TRACE_EVENT_IOWAIT;
613	}
614	else {
615	printf("packet error\n");
616	event.type = TRACE_EVENT_PACKET;
617	}
618	} else if (event.size == 0) {
619	event.type = TRACE_EVENT_TERMINATE;
620
621	}
622	else {
623	event.type = TRACE_EVENT_PACKET;
624	}
625
626	return event;
627	}
628
629	static void rt_help() {
630	printf("rt format module\n");
631	printf("Supported input URIs:\n");
632	printf("\trt:hostname:port\n");
633	printf("\trt:hostname (connects on default port)\n");
634	printf("\n");
635	printf("\te.g.: rt:localhost\n");
636	printf("\te.g.: rt:localhost:32500\n");
637	printf("\n");
638
639	}
640
641
642	static struct libtrace_format_t rt = {
643	"rt",
644	"$Id$",
645	TRACE_FORMAT_RT,
646	rt_init_input, /* init_input */
647	NULL, /* config_input */
648	rt_start_input, /* start_input */
649	NULL, /* init_output */
650	NULL, /* config_output */
651	NULL, /* start_output */
652	NULL, /* pause_output */
653	rt_fin_input, /* fin_input */
654	NULL, /* fin_output */
655	rt_read_packet, /* read_packet */
656	NULL, /* fin_packet */
657	NULL, /* write_packet */
658	NULL, /* get_link_type */
659	NULL, /* get_direction */
660	NULL, /* set_direction */
661	NULL, /* get_erf_timestamp */
662	NULL, /* get_timeval */
663	NULL, /* get_seconds */
664	NULL, /* seek_erf */
665	NULL, /* seek_timeval */
666	NULL, /* seek_seconds */
667	rt_get_capture_length, /* get_capture_length */
668	rt_get_wire_length, /* get_wire_length */
669	rt_get_framing_length, /* get_framing_length */
670	NULL, /* set_capture_length */
671	rt_get_fd, /* get_fd */
672	trace_event_rt, /* trace_event */
673	rt_help, /* help */
674	NULL /* next pointer */
675	};
676
677	void __attribute__((constructor)) rt_constructor() {
678	register_format(&rt);
679	}

Note: See TracBrowser for help on using the repository browser.

Download in other formats: