<commit_before>#include <time.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
#include <errno.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include "util.h"

/* Max time to wait between batching up frames to send to voltaire */
#define BATCH_PERIOD	0.1
#define DEFAULT_LIBMARQUISE_ORIGIN	"BENHUR"


#define __STRINGIZE(x) #x
#define _STRINGIZE(x) __STRINGIZE(x)
#define __FILEPOS__ __FILE__ ":" _STRINGIZE(__LINE__)

#ifdef DEBUG
#define DEBUG_PRINTF(formatstr, ...) fprintf(stderr, __FILEPOS__ ", %s() - " formatstr, __func__,  __VA_ARGS__)
#else
#define DEBUG_PRINTF(...)
#endif

#if _POSIX_C_SOURCE >= 200809L
#define SCANF_ALLOCATE_STRING_FLAG "m"
#elif defined(__GLIBC__) && (__GLIBC__ >= 2)
#define SCANF_ALLOCATE_STRING_FLAG "a"
#else
#error "Please let us have POSIX.1-2008, glibc, or a puppy"
#endif
#define SCANF_ALLOCATE_STRING "%" SCANF_ALLOCATE_STRING_FLAG "s"

#define SOURCE_KEY_COLLECTION_POINT "collection_point"
#define SOURCE_KEY_IP "ip"
#define SOURCE_KEY_BYTES "bytes"

/**
 * This is only going to fly if we are getting data in on the fly
 * and we have no other timestamp source
 */
uint64_t timestamp_now() {
	struct timespec ts;
	if (clock_gettime(CLOCK_REALTIME, &ts)) { perror("clock_gettime"); exit(2); }
	return (ts.tv_sec*1000000000) + ts.tv_nsec;
}

typedef struct {
	in_addr_t network;
	in_addr_t netmask;
	void * next;
} networkaddr_ll_t;

/* returns 1 if it is in the whitelist, 0 if it is not, or -1
 * if it is not a well-formed address
 */
static inline int is_address_in_whitelist(char *ipaddr , networkaddr_ll_t *whitelist) {
	struct in_addr addr;
	if (inet_aton(ipaddr, &addr) == 0) return -1;
	/* everything is whitelisted by default if there is no whitelist */
	if (whitelist == NULL) return 1;
	for (; whitelist != NULL; whitelist = whitelist->next) {
		if ((addr.s_addr & whitelist->netmask) == whitelist->network)
			return 1;
	}
	return 0;
}
static void free_whitelist(networkaddr_ll_t *whitelist) {
	while (whitelist != NULL) {
		networkaddr_ll_t *next = whitelist->next;
		free(whitelist);
		whitelist = next;
	}
}

networkaddr_ll_t * read_ip_whitelist(char *pathname) {
	char *network = NULL;
	char *netmask = NULL;
	networkaddr_ll_t *head = NULL;

	FILE *infile = fopen(pathname, "r");
	if (infile == NULL)
		return perror(pathname), NULL;

	while (fscanf(infile,
			" %" SCANF_ALLOCATE_STRING_FLAG "[0123456789.]"
			"/" SCANF_ALLOCATE_STRING, &network,&netmask) == 2) {
		networkaddr_ll_t * entry = malloc(sizeof(networkaddr_ll_t));
		if (entry == NULL)
			return perror("malloc"),NULL;
		struct in_addr a;
		if (inet_aton(network, &a) == 0) {
			fprintf(stderr,"Invalid whitelisted network '%s'. Skipping\n",network);
			free(entry); free(network); free(netmask); continue;
		}
		entry->network = a.s_addr;
		struct in_addr b;
		if (inet_aton(netmask, &b) == 0) {
			fprintf(stderr,"Invalid whitelisted network '%s'. Skipping\n",network);
			free(entry); free(network); free(netmask); continue;
		}
		entry->netmask = b.s_addr;
		if (ntohl( entry->netmask ) <= 24) {
			/* Prefix length not netmask */
			entry->netmask = ntohl(entry->netmask);
			entry->netmask = ~((1<<(32- entry->netmask )) - 1);
			entry->netmask = htonl(entry->netmask);
		}
		a.s_addr = entry->network;
		b.s_addr = entry->netmask;
		fprintf(stderr,"Added to whitelist: %s/",
				inet_ntoa(a));
		fprintf(stderr,"%s\n",inet_ntoa(b));

		entry->next = head;
		head = entry;
		free(network); free(netmask);
	}
	fclose(infile);
	return head;
}

/*K
 * parse a pmacct record line.
 *
 * *source_ip and *dest_ip are allocated by libc
 * caller must free() *source_ip and *dest_ip iff the call was successful
 * (as per sscanf)
 *
 * pre: instring is zero terminated
 *
 * returns >= 1 on success
 */
int parse_pmacct_record(char *cs, char **source_ip, char **dest_ip, uint64_t *bytes) {
	/* Format (with more whitespace in actual input):
	 *
	 * ID CLASS SRC_MAC DST_MAC VLAN SRC_AS DST_AS SRC_IP DST_IP SRC_PORT DST_PORT TCP_FLAGS PROTOCOL TOS PACKETS FLOWS BYTES
	 * 0 unknown 00:00:00:00:00:00 00:00:00:00:00:00 0 0 0 172.16.58.3 172.16.31.10 0 0 0 ip 0 24 0 34954
	 *
	 * We ignore everything other than source IP, destination IP, and bytes
	 */
	*source_ip = NULL;
	*dest_ip = NULL;
	return sscanf(cs,
		"%*s%*s%*s%*s%*s%*s%*s"
		SCANF_ALLOCATE_STRING
		SCANF_ALLOCATE_STRING
		"%*s%*s%*s%*s%*s%*s%*s"
		"%lu",
		source_ip, dest_ip, bytes
		) == 3;
}

static inline int emit_bytes(marquise_ctx *ctx, char *address_string, 
		marquise_source *marq_source, uint64_t timestamp, uint64_t bytes) {
	uint64_t address = marquise_hash_identifier(address_string, strlen(address_string));
	int success;
	success = marquise_send_simple(ctx, address, timestamp, bytes);
	if (success == 0) {
		DEBUG_PRINTF("successfully queued packet\n");
	} else {
		DEBUG_PRINTF("failed to send packet\n");
		return success;
	}	
	success = marquise_update_source(ctx, address, marq_source);
	if (success == 0) {
		DEBUG_PRINTF("successfully queued source packet\n");
	} else {
		DEBUG_PRINTF("failed to send source packet\n");
	return success;
	}		
	return 0;
}

static inline int emit_tx_bytes(marquise_ctx *ctx,
		char *collection_point, char *ip, uint64_t timestamp,
		uint64_t bytes) {
	char *address_string = build_address_string(collection_point, ip, "tx");
	marquise_source *marq_source = build_marquise_source(collection_point, ip, "tx");
	int ret = emit_bytes(ctx, address_string, marq_source, timestamp, bytes);
	free(address_string);
	return ret;
}

static inline int emit_rx_bytes(marquise_ctx *ctx,
		char *collection_point, char *ip, uint64_t timestamp,
		uint64_t bytes) {
	char *address_string = build_address_string(collection_point, ip, "rx");
	marquise_source *marq_source = build_marquise_source(collection_point, ip, "rx");
	int ret = emit_bytes(ctx, address_string, marq_source, timestamp, bytes);<commit_msg>whitespace nazi<commit_after>#include <time.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
#include <errno.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include "util.h"

/* Max time to wait between batching up frames to send to voltaire */
#define BATCH_PERIOD	0.1
#define DEFAULT_LIBMARQUISE_ORIGIN	"BENHUR"


#define __STRINGIZE(x) #x
#define _STRINGIZE(x) __STRINGIZE(x)
#define __FILEPOS__ __FILE__ ":" _STRINGIZE(__LINE__)

#ifdef DEBUG
#define DEBUG_PRINTF(formatstr, ...) fprintf(stderr, __FILEPOS__ ", %s() - " formatstr, __func__,  __VA_ARGS__)
#else
#define DEBUG_PRINTF(...)
#endif

#if _POSIX_C_SOURCE >= 200809L
#define SCANF_ALLOCATE_STRING_FLAG "m"
#elif defined(__GLIBC__) && (__GLIBC__ >= 2)
#define SCANF_ALLOCATE_STRING_FLAG "a"
#else
#error "Please let us have POSIX.1-2008, glibc, or a puppy"
#endif
#define SCANF_ALLOCATE_STRING "%" SCANF_ALLOCATE_STRING_FLAG "s"

#define SOURCE_KEY_COLLECTION_POINT "collection_point"
#define SOURCE_KEY_IP "ip"
#define SOURCE_KEY_BYTES "bytes"

/**
 * This is only going to fly if we are getting data in on the fly
 * and we have no other timestamp source
 */
uint64_t timestamp_now() {
	struct timespec ts;
	if (clock_gettime(CLOCK_REALTIME, &ts)) { perror("clock_gettime"); exit(2); }
	return (ts.tv_sec*1000000000) + ts.tv_nsec;
}

typedef struct {
	in_addr_t network;
	in_addr_t netmask;
	void * next;
} networkaddr_ll_t;

/* returns 1 if it is in the whitelist, 0 if it is not, or -1
 * if it is not a well-formed address
 */
static inline int is_address_in_whitelist(char *ipaddr , networkaddr_ll_t *whitelist) {
	struct in_addr addr;
	if (inet_aton(ipaddr, &addr) == 0) return -1;
	/* everything is whitelisted by default if there is no whitelist */
	if (whitelist == NULL) return 1;
	for (; whitelist != NULL; whitelist = whitelist->next) {
		if ((addr.s_addr & whitelist->netmask) == whitelist->network)
			return 1;
	}
	return 0;
}
static void free_whitelist(networkaddr_ll_t *whitelist) {
	while (whitelist != NULL) {
		networkaddr_ll_t *next = whitelist->next;
		free(whitelist);
		whitelist = next;
	}
}

networkaddr_ll_t * read_ip_whitelist(char *pathname) {
	char *network = NULL;
	char *netmask = NULL;
	networkaddr_ll_t *head = NULL;

	FILE *infile = fopen(pathname, "r");
	if (infile == NULL)
		return perror(pathname), NULL;

	while (fscanf(infile,
			" %" SCANF_ALLOCATE_STRING_FLAG "[0123456789.]"
			"/" SCANF_ALLOCATE_STRING, &network,&netmask) == 2) {
		networkaddr_ll_t * entry = malloc(sizeof(networkaddr_ll_t));
		if (entry == NULL)
			return perror("malloc"),NULL;
		struct in_addr a;
		if (inet_aton(network, &a) == 0) {
			fprintf(stderr,"Invalid whitelisted network '%s'. Skipping\n",network);
			free(entry); free(network); free(netmask); continue;
		}
		entry->network = a.s_addr;
		struct in_addr b;
		if (inet_aton(netmask, &b) == 0) {
			fprintf(stderr,"Invalid whitelisted network '%s'. Skipping\n",network);
			free(entry); free(network); free(netmask); continue;
		}
		entry->netmask = b.s_addr;
		if (ntohl( entry->netmask ) <= 24) {
			/* Prefix length not netmask */
			entry->netmask = ntohl(entry->netmask);
			entry->netmask = ~((1<<(32- entry->netmask )) - 1);
			entry->netmask = htonl(entry->netmask);
		}
		a.s_addr = entry->network;
		b.s_addr = entry->netmask;
		fprintf(stderr,"Added to whitelist: %s/",
				inet_ntoa(a));
		fprintf(stderr,"%s\n",inet_ntoa(b));

		entry->next = head;
		head = entry;
		free(network); free(netmask);
	}
	fclose(infile);
	return head;
}

/*K
 * parse a pmacct record line.
 *
 * *source_ip and *dest_ip are allocated by libc
 * caller must free() *source_ip and *dest_ip iff the call was successful
 * (as per sscanf)
 *
 * pre: instring is zero terminated
 *
 * returns >= 1 on success
 */
int parse_pmacct_record(char *cs, char **source_ip, char **dest_ip, uint64_t *bytes) {
	/* Format (with more whitespace in actual input):
	 *
	 * ID CLASS SRC_MAC DST_MAC VLAN SRC_AS DST_AS SRC_IP DST_IP SRC_PORT DST_PORT TCP_FLAGS PROTOCOL TOS PACKETS FLOWS BYTES
	 * 0 unknown 00:00:00:00:00:00 00:00:00:00:00:00 0 0 0 172.16.58.3 172.16.31.10 0 0 0 ip 0 24 0 34954
	 *
	 * We ignore everything other than source IP, destination IP, and bytes
	 */
	*source_ip = NULL;
	*dest_ip = NULL;
	return sscanf(cs,
		"%*s%*s%*s%*s%*s%*s%*s"
		SCANF_ALLOCATE_STRING
		SCANF_ALLOCATE_STRING
		"%*s%*s%*s%*s%*s%*s%*s"
		"%lu",
		source_ip, dest_ip, bytes
		) == 3;
}

static inline int emit_bytes(marquise_ctx *ctx, char *address_string,
		marquise_source *marq_source, uint64_t timestamp, uint64_t bytes) {
	uint64_t address = marquise_hash_identifier(address_string, strlen(address_string));
	int success;
	success = marquise_send_simple(ctx, address, timestamp, bytes);
	if (success == 0) {
		DEBUG_PRINTF("successfully queued packet\n");
	} else {
		DEBUG_PRINTF("failed to send packet\n");
		return success;
	}
	success = marquise_update_source(ctx, address, marq_source);
	if (success == 0) {
		DEBUG_PRINTF("successfully queued source packet\n");
	} else {
		DEBUG_PRINTF("failed to send source packet\n");
	return success;
	}
	return 0;
}

static inline int emit_tx_bytes(marquise_ctx *ctx,
		char *collection_point, char *ip, uint64_t timestamp,
		uint64_t bytes) {
	char *address_string = build_address_string(collection_point, ip, "tx");
	marquise_source *marq_source = build_marquise_source(collection_point, ip, "tx");
	int ret = emit_bytes(ctx, address_string, marq_source, timestamp, bytes);
	free(address_string);
	return ret;
}

static inline int emit_rx_bytes(marquise_ctx *ctx,
		char *collection_point, char *ip, uint64_t timestamp,
		uint64_t bytes) {
	char *address_string = build_address_string(collection_point, ip, "rx");
	marquise_source *marq_source = build_marquise_source(collection_point, ip, "rx");
	int ret = emit_bytes(ctx, address_string, marq_source, timestamp, bytes);
{'fraction_non_alphanumeric': 0.13170897938319986, 'fraction_numerical': 0.022491054694155733, 'mean_word_length': 4.88716148445336, 'pattern_counts': {'":': 2, '<': 25, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3987202', 'n_tokens_mistral': 4803, 'n_tokens_neox': 4426, 'n_words': 1124}