Context Navigation

source: trunk/lib/format_rt.c @ 743

Visit:

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

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

Download in other formats: