<commit_before>/*
 * Copyright (c) 2008, 2009 by Internet Systems Consortium, Inc. ("ISC")
 *
 * Permission to use, copy, modify, and/or distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
 * OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

#include <sys/socket.h>

#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <pcap.h>

#include "kickfile.h"
#include "nmsgtool.h"

static const int on = 1;

void
add_sock_input(nmsgtool_ctx *c, const char *ss) {
	char *t;
	int pa, pz, pn, pl;

	t = strchr(ss, '/');
	if (t == NULL)
		usage("argument to -l needs a /");
	if (sscanf(t + 1, "%d..%d", &pa, &pz) == 2) {
		if (pa > pz || pz - pa > 20)
			usage("bad port range in -l argument");
	} else if (sscanf(t + 1, "%d", &pa) == 1) {
		pz = pa;
	} else {
		usage("need a port number or range after /");
	}
	pl = t - ss;
	for (pn = pa; pn <= pz; pn++) {
		char *spec;
		int len, pf, s;
		nmsgtool_sockaddr su;
		nmsg_input_t input;
		nmsg_res res;

		nmsg_asprintf(&spec, "%*.*s/%d", pl, pl, ss, pn);
		pf = getsock(&su, spec, NULL, NULL);
		if (c->debug >= 2)
			fprintf(stderr, "%s: nmsg socket input: %s\n",
				argv_program, spec);
		free(spec);
		if (pf < 0)
			usage("bad -l socket");
		s = socket(pf, SOCK_DGRAM, 0);
		if (s < 0) {
			perror("socket");
			exit(1);
		}
		Setsockopt(s, SOL_SOCKET, SO_REUSEADDR, on);
#ifdef SO_REUSEPORT
		Setsockopt(s, SOL_SOCKET, SO_REUSEPORT, on);
#endif
		len = 32 * 1024;
		Setsockopt(s, SOL_SOCKET, SO_RCVBUF, len);
		if (bind(s, &su.sa, NMSGTOOL_SA_LEN(su.sa)) < 0) {
			perror("bind");
			exit(1);
		}
		input = nmsg_input_open_sock(s);
		if (input == NULL) {
			fprintf(stderr, "%s: nmsg_input_open_sock() failed\n",
				argv_program);
			exit(1);
		}
		setup_nmsg_input(c, input);
		res = nmsg_io_add_input(c->io, input, NULL);
		if (res != nmsg_res_success) {
			fprintf(stderr, "%s: nmsg_io_add_input() failed\n",
				argv_program);
			exit(1);
		}
		c->n_inputs += 1;
	}
}

void
add_sock_output(nmsgtool_ctx *c, const char *ss) {
	char *r, *t;
	int pa, pz, pn, pl;

	t = strchr(ss, '/');
	r = strchr(ss, ',');
	if (t == NULL)
		usage("argument to -s needs a /");
	if (sscanf(t + 1, "%d..%d", &pa, &pz) == 2) {
		if (pa > pz || pz - pa > 20)
			usage("bad port range in -s argument");
	} else if (sscanf(t + 1, "%d", &pa) == 1) {
		pz = pa;
	} else {
		usage("need a port number or range after /");
	}
	pl = t - ss;
	for (pn = pa; pn <= pz; pn++) {
		char *spec;
		int len, pf, s;
		nmsgtool_sockaddr su;
		nmsg_output_t output;
		nmsg_res res;
		unsigned rate = 0, freq = 0;

		nmsg_asprintf(&spec, "%*.*s/%d%s", pl, pl, ss, pn,
			      r != NULL ? r : "");
		pf = getsock(&su, spec, &rate, &freq);
		if (freq == 0)
			freq = DEFAULT_FREQ;
		if (c->debug >= 2)
			fprintf(stderr, "%s: nmsg socket output: %s\n",
				argv_program, spec);
		if (c->debug >= 2 && rate > 0)
			fprintf(stderr, "%s: nmsg socket rate: %u freq: %u\n",
				argv_program, rate, freq);
		free(spec);
		if (pf < 0)
			usage("bad -s socket");
		s = socket(pf, SOCK_DGRAM, 0);
		if (s < 0) {
			perror("socket");
			exit(1);
		}
		Setsockopt(s, SOL_SOCKET, SO_BROADCAST, on);
		len = 32 * 1024;
		Setsockopt(s, SOL_SOCKET, SO_SNDBUF, len);
		if (connect(s, &su.sa, NMSGTOOL_SA_LEN(su.sa)) < 0) {
			perror("connect");
			exit(1);
		}
		output = nmsg_output_open_sock(s, c->mtu);
		if (output == NULL) {
			fprintf(stderr, "%s: nmsg_output_open_sock() failed\n",
				argv_program);
			exit(1);
		}
		setup_nmsg_output(c, output);
		if (rate > 0 && freq > 0) {
			nmsg_rate_t nr;

			nr = nmsg_rate_init(rate, freq);
			assert(nr != NULL);
			nmsg_output_set_rate(output, nr);
		}
		if (c->kicker != NULL) {
			res = nmsg_io_add_output(c->io, output, (void *) -1);
		} else {
			res = nmsg_io_add_output(c->io, output, NULL);
		}
		if (res != nmsg_res_success) {
			fprintf(stderr, "%s: nmsg_io_add_output() failed\n",
				argv_program);
			exit(1);
		}
		c->n_outputs += 1;
	}
}

void
add_file_input(nmsgtool_ctx *c, const char *fname) {
	nmsg_input_t input;
	nmsg_res res;

	input = nmsg_input_open_file(open_rfile(fname));
	if (input == NULL) {
		fprintf(stderr, "%s: nmsg_input_open_file() failed\n",
			argv_program);
		exit(1);
	}
	setup_nmsg_input(c, input);
	res = nmsg_io_add_input(c->io, input, NULL);
	if (res != nmsg_res_success) {
		fprintf(stderr, "%s: nmsg_io_add_input() failed\n",<commit_msg>src/io: call nmsg_input_set_filter_msgtype() on sock / file inputs<commit_after>/*
 * Copyright (c) 2008, 2009 by Internet Systems Consortium, Inc. ("ISC")
 *
 * Permission to use, copy, modify, and/or distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
 * OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

#include <sys/socket.h>

#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <pcap.h>

#include "kickfile.h"
#include "nmsgtool.h"

static const int on = 1;

void
add_sock_input(nmsgtool_ctx *c, const char *ss) {
	char *t;
	int pa, pz, pn, pl;

	t = strchr(ss, '/');
	if (t == NULL)
		usage("argument to -l needs a /");
	if (sscanf(t + 1, "%d..%d", &pa, &pz) == 2) {
		if (pa > pz || pz - pa > 20)
			usage("bad port range in -l argument");
	} else if (sscanf(t + 1, "%d", &pa) == 1) {
		pz = pa;
	} else {
		usage("need a port number or range after /");
	}
	pl = t - ss;
	for (pn = pa; pn <= pz; pn++) {
		char *spec;
		int len, pf, s;
		nmsgtool_sockaddr su;
		nmsg_input_t input;
		nmsg_res res;

		nmsg_asprintf(&spec, "%*.*s/%d", pl, pl, ss, pn);
		pf = getsock(&su, spec, NULL, NULL);
		if (c->debug >= 2)
			fprintf(stderr, "%s: nmsg socket input: %s\n",
				argv_program, spec);
		free(spec);
		if (pf < 0)
			usage("bad -l socket");
		s = socket(pf, SOCK_DGRAM, 0);
		if (s < 0) {
			perror("socket");
			exit(1);
		}
		Setsockopt(s, SOL_SOCKET, SO_REUSEADDR, on);
#ifdef SO_REUSEPORT
		Setsockopt(s, SOL_SOCKET, SO_REUSEPORT, on);
#endif
		len = 32 * 1024;
		Setsockopt(s, SOL_SOCKET, SO_RCVBUF, len);
		if (bind(s, &su.sa, NMSGTOOL_SA_LEN(su.sa)) < 0) {
			perror("bind");
			exit(1);
		}
		input = nmsg_input_open_sock(s);
		if (input == NULL) {
			fprintf(stderr, "%s: nmsg_input_open_sock() failed\n",
				argv_program);
			exit(1);
		}
		if (c->vid != 0 && c->msgtype != 0)
			nmsg_input_set_filter_msgtype(input, c->vid, c->msgtype);
		setup_nmsg_input(c, input);
		res = nmsg_io_add_input(c->io, input, NULL);
		if (res != nmsg_res_success) {
			fprintf(stderr, "%s: nmsg_io_add_input() failed\n",
				argv_program);
			exit(1);
		}
		c->n_inputs += 1;
	}
}

void
add_sock_output(nmsgtool_ctx *c, const char *ss) {
	char *r, *t;
	int pa, pz, pn, pl;

	t = strchr(ss, '/');
	r = strchr(ss, ',');
	if (t == NULL)
		usage("argument to -s needs a /");
	if (sscanf(t + 1, "%d..%d", &pa, &pz) == 2) {
		if (pa > pz || pz - pa > 20)
			usage("bad port range in -s argument");
	} else if (sscanf(t + 1, "%d", &pa) == 1) {
		pz = pa;
	} else {
		usage("need a port number or range after /");
	}
	pl = t - ss;
	for (pn = pa; pn <= pz; pn++) {
		char *spec;
		int len, pf, s;
		nmsgtool_sockaddr su;
		nmsg_output_t output;
		nmsg_res res;
		unsigned rate = 0, freq = 0;

		nmsg_asprintf(&spec, "%*.*s/%d%s", pl, pl, ss, pn,
			      r != NULL ? r : "");
		pf = getsock(&su, spec, &rate, &freq);
		if (freq == 0)
			freq = DEFAULT_FREQ;
		if (c->debug >= 2)
			fprintf(stderr, "%s: nmsg socket output: %s\n",
				argv_program, spec);
		if (c->debug >= 2 && rate > 0)
			fprintf(stderr, "%s: nmsg socket rate: %u freq: %u\n",
				argv_program, rate, freq);
		free(spec);
		if (pf < 0)
			usage("bad -s socket");
		s = socket(pf, SOCK_DGRAM, 0);
		if (s < 0) {
			perror("socket");
			exit(1);
		}
		Setsockopt(s, SOL_SOCKET, SO_BROADCAST, on);
		len = 32 * 1024;
		Setsockopt(s, SOL_SOCKET, SO_SNDBUF, len);
		if (connect(s, &su.sa, NMSGTOOL_SA_LEN(su.sa)) < 0) {
			perror("connect");
			exit(1);
		}
		output = nmsg_output_open_sock(s, c->mtu);
		if (output == NULL) {
			fprintf(stderr, "%s: nmsg_output_open_sock() failed\n",
				argv_program);
			exit(1);
		}
		setup_nmsg_output(c, output);
		if (rate > 0 && freq > 0) {
			nmsg_rate_t nr;

			nr = nmsg_rate_init(rate, freq);
			assert(nr != NULL);
			nmsg_output_set_rate(output, nr);
		}
		if (c->kicker != NULL) {
			res = nmsg_io_add_output(c->io, output, (void *) -1);
		} else {
			res = nmsg_io_add_output(c->io, output, NULL);
		}
		if (res != nmsg_res_success) {
			fprintf(stderr, "%s: nmsg_io_add_output() failed\n",
				argv_program);
			exit(1);
		}
		c->n_outputs += 1;
	}
}

void
add_file_input(nmsgtool_ctx *c, const char *fname) {
	nmsg_input_t input;
	nmsg_res res;

	input = nmsg_input_open_file(open_rfile(fname));
	if (input == NULL) {
		fprintf(stderr, "%s: nmsg_input_open_file() failed\n",
			argv_program);
		exit(1);
	}
	if (c->vid != 0 && c->msgtype != 0)
		nmsg_input_set_filter_msgtype(input, c->vid, c->msgtype);
	setup_nmsg_input(c, input);
	res = nmsg_io_add_input(c->io, input, NULL);
	if (res != nmsg_res_success) {
		fprintf(stderr, "%s: nmsg_io_add_input() failed\n",
{'fraction_non_alphanumeric': 0.16836381611468115, 'fraction_numerical': 0.012654473554127533, 'mean_word_length': 3.4603174603174605, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 65, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '841264', 'n_tokens_mistral': 4642, 'n_tokens_neox': 4153, 'n_words': 1153}