fatalf("Data Set based on known but unhandled template type %d", bt->setID); } }
msg(MSG_FATAL, "Data Set based on known but unhandled template type %d", bt->setID); } }    static int processNetflowV9Packet(IpfixReceiver* ipfixReceiver, byte* message, uint16_t length) { NetflowV9Header* header = (NetflowV9Header*)message;   IpfixSetHeader* set = (IpfixSetHeader*)&header->data;  int i; for (i = 0; i < ntohs(header->setCount); i++) { if (ntohs(set->id) == NetflowV9_SetId_Template) { processTemplateSet(ipfixReceiver, ntohs(header->sourceId), set); } else if (ntohs(set->id) >= IPFIX_SetId_Data_Start) { processDataSet(ipfixReceiver, ntohs(header->sourceId), set); } else { msg(MSG_ERROR, "processNetflowV9: Unsupported Set ID - expected 0/256+, got %d", ntohs(set->id)); } set = (IpfixSetHeader*)((byte*)set + ntohs(set->length)); }  return 0; }   static int processIpfixPacket(IpfixReceiver* ipfixReceiver, byte* message, uint16_t length) { IpfixHeader* header = (IpfixHeader*)message;  if (ntohs(header->length) != length) { DPRINTF("Bad message length - expected %#06x, got %#06x\n", length, ntohs(header->length)); return -1; }   IpfixSetHeader* set = (IpfixSetHeader*)&header->data;   IpfixSetHeader* setX = (IpfixSetHeader*)((char*)message + length);  while (set < setX) { if (ntohs(set->id) == IPFIX_SetId_Template) { processTemplateSet(ipfixReceiver, ntohs(header->sourceId), set); } else if (ntohs(set->id) == IPFIX_SetId_OptionsTemplate) { processOptionsTemplateSet(ipfixReceiver, ntohs(header->sourceId), set); } else if (ntohs(set->id) == IPFIX_SetId_DataTemplate) { processDataTemplateSet(ipfixReceiver, ntohs(header->sourceId), set); } else if (ntohs(set->id) >= IPFIX_SetId_Data_Start) { processDataSet(ipfixReceiver, ntohs(header->sourceId), set); } else { msg(MSG_ERROR, "processIpfixPacket: Unsupported Set ID - expected 2/3/4/256+, got %d", ntohs(set->id)); } set = (IpfixSetHeader*)((byte*)set + ntohs(set->length)); }  return 0; }   static int processMessage(IpfixReceiver* ipfixReceiver, byte* message, uint16_t length) { IpfixHeader* header = (IpfixHeader*)message; if (ntohs(header->version) == 0x000a) { return processIpfixPacket(ipfixReceiver, message, length); } #ifdef SUPPORT_NETFLOWV9 if (ntohs(header->version) == 0x0009) { return processNetflowV9Packet(ipfixReceiver, message, length); } DPRINTF("Bad message version - expected 0x009 or 0x000a, got %#06x\n", ntohs(header->version)); return -1; #else DPRINTF("Bad message version - expected 0x000a, got %#06x\n", ntohs(header->version)); return -1; #endif }  static void printIPv4(FieldType type, FieldData* data) { int octet1 = 0; int octet2 = 0; int octet3 = 0; int octet4 = 0; int imask = 0; if (type.length >= 1) octet1 = data[0]; if (type.length >= 2) octet2 = data[1]; if (type.length >= 3) octet3 = data[2]; if (type.length >= 4) octet4 = data[3]; if (type.length >= 5) imask = data[4]; if (type.length > 5) { DPRINTF("IPv4 Address with length %d unparseable\n", type.length); return; }  if ((type.length == 5) ) { printf("%d.%d.%d.%d/%d", octet1, octet2, octet3, octet4, 32-imask); } else { printf("%d.%d.%d.%d", octet1, octet2, octet3, octet4); } }  static void printPort(FieldType type, FieldData* data) { if (type.length == 0) { printf("zero-length Port"); return; } if (type.length == 2) { int port = ((uint16_t)data[0] << 8)+data[1]; printf("%d", port); return; } if ((type.length >= 4) && ((type.length % 4) == 0)) { int i; for (i = 0; i < type.length; i+=4) { int starti = ((uint16_t)data[i+0] << 8)+data[i+1]; int endi = ((uint16_t)data[i+2] << 8)+data[i+3]; if (i > 0) printf(","); if (starti != endi) { printf("%d:%d", starti, endi); } else { printf("%d", starti); } } return; }  errorf("Port with length %d unparseable", type.length); }  void printProtocol(FieldType type, FieldData* data) { if (type.length != 1) { errorf("Protocol with length %d unparseable", type.length); return; } switch (data[0]) { case IPFIX_protocolIdentifier_ICMP: printf("ICMP"); return; case IPFIX_protocolIdentifier_TCP: printf("TCP"); return; case IPFIX_protocolIdentifier_UDP: printf("UDP"); return; case IPFIX_protocolIdentifier_RAW: printf("RAW"); return; default: printf("unknownProtocol"); return; } }  static void printUint(FieldType type, FieldData* data) { switch (type.length) { case 1: printf("%hhu",*(uint8_t*)data); return; case 2: printf("%hu",ntohs(*(uint16_t*)data)); return; case 4: printf("%u",ntohl(*(uint32_t*)data)); return; case 8: printf("%Lu",ntohll(*(uint64_t*)data)); return; default: errorf("Uint with length %d unparseable", type.length); return; } }   static void* listenerUdpIpv4(void* ipfixReceiver_) { IpfixReceiver* ipfixReceiver = (IpfixReceiver*)ipfixReceiver_;  struct sockaddr_in clientAddress; socklen_t clientAddressLen; byte* data = (byte*)malloc(sizeof(byte)*MAX_MSG_LEN); int n;  while(1) {     clientAddressLen = sizeof(struct sockaddr_in); n = recvfrom(ipfixReceiver->socket, data, MAX_MSG_LEN, 0, (struct sockaddr*)&clientAddress, &clientAddressLen);  if(n < 0) { msg(MSG_DEBUG, "recvfrom returned with error, terminating UDP/IPv4 listener thread"); break; }  pthread_mutex_lock(&ipfixReceiver->mutex); processMessage(ipfixReceiver, data, n); pthread_mutex_unlock(&ipfixReceiver->mutex); }  free(data);  return 0; }     int initializeIpfixReceivers() { return 0; }   int deinitializeIpfixReceivers() { return 0; }   void printFieldData(FieldType type, FieldData* pattern) { char* s;  switch (type.id) { case IPFIX_TYPEID_protocolIdentifier: printf("protocolIdentifier:"); printProtocol(type, pattern); break; case IPFIX_TYPEID_sourceIPv4Address: printf("sourceIPv4Address:"); printIPv4(type, pattern); break; case IPFIX_TYPEID_destinationIPv4Address: printf("destinationIPv4Address:"); printIPv4(type, pattern); break; case IPFIX_TYPEID_sourceTransportPort: printf("sourceTransportPort:"); printPort(type, pattern); break; case IPFIX_TYPEID_destinationtransportPort: printf("destinationtransportPort:"); printPort(type, pattern); break; default: s = typeid2string(type.id); if (s != NULL) { printf("%s:", s); printUint(type, pattern); } else { DPRINTF("Field with ID %d unparseable\n", type.id); } break; } }   FieldInfo* getTemplateFieldInfo(TemplateInfo* ti, FieldType* type) { int i;  for (i = 0; i < ti->fieldCount; i++) { if ((ti->fieldInfo[i].type.id == type->id) && (ti->fieldInfo[i].type.eid == type->eid)) { return &ti->fieldInfo[i]; } }  return NULL; }   FieldInfo* getDataTemplateFieldInfo(DataTemplateInfo* ti, FieldType* type) { int i;  for (i = 0; i < ti->fieldCount; i++) { if ((ti->fieldInfo[i].type.id == type->id) && (ti->fieldInfo[i].type.eid == type->eid)) { return &ti->fieldInfo[i]; } }  return NULL; }   FieldInfo* getDataTemplateDataInfo(DataTemplateInfo* ti, FieldType* type) { int i;  for (i = 0; i < ti->dataCount; i++) { if ((ti->dataInfo[i].type.id == type->id) && (ti->dataInfo[i].type.eid == type->eid)) { return &ti->dataInfo[i]; } }  return NULL; }   void addIpfixReceiverCallbacks(IpfixReceiver* ipfixReceiver, CallbackInfo handles) { int n = ++ipfixReceiver->callbackCount; ipfixReceiver->callbackInfo = (CallbackInfo*)realloc(ipfixReceiver->callbackInfo, n * sizeof(CallbackInfo)); memcpy(&ipfixReceiver->callbackInfo[n-1], &handles, sizeof(CallbackInfo)); }   IpfixReceiver* createIpfixReceiver(uint16_t port) { IpfixReceiver* ipfixReceiver; struct sockaddr_in serverAddress;  if(!(ipfixReceiver=(IpfixReceiver*)malloc(sizeof(IpfixReceiver)))) { msg(MSG_FATAL, "Ran out of memory"); goto out0; }  ipfixReceiver->callbackInfo = 0;  if(!(ipfixReceiver->templateBuffer = createTemplateBuffer(ipfixReceiver))) { msg(MSG_FATAL, "Could not create template Buffer"); goto out1; }  if (pthread_mutex_init(&ipfixReceiver->mutex, NULL) != 0) { msg(MSG_FATAL, "Could not init mutex"); goto out2; }  if (pthread_mutex_lock(&ipfixReceiver->mutex) != 0) { msg(MSG_FATAL, "Could not lock mutex"); goto out2; }  ipfixReceiver->socket = socket(AF_INET, SOCK_DGRAM, 0); if(ipfixReceiver->socket < 0) { msg(MSG_FATAL, "Could not create socket"); goto out2; }  serverAddress.sin_family = AF_INET; serverAddress.sin_addr.s_addr = htonl(INADDR_ANY); serverAddress.sin_port = htons(port); if(bind(ipfixReceiver->socket, (struct sockaddr*)&serverAddress, sizeof(struct sockaddr_in)) < 0) { msg(MSG_FATAL, "Could not bind socket"); goto out3; }  if(pthread_create(&(ipfixReceiver->thread), 0, listenerUdpIpv4, ipfixReceiver) != 0) { msg(MSG_FATAL, "Could not pthread_create listener thread"); goto out3; }   return ipfixReceiver;  out3: close(ipfixReceiver->socket); out2: destroyTemplateBuffer(ipfixReceiver->templateBuffer); out1: free(ipfixReceiver); out0: return NULL; }   int startIpfixReceiver(IpfixReceiver* ipfixReceiver) { if (pthread_mutex_unlock(&ipfixReceiver->mutex) != 0) { return -1; }  return 0; }   int stopIpfixReceiver(IpfixReceiver* ipfixReceiver) { if (pthread_mutex_lock(&ipfixReceiver->mutex) != 0) { return -1; }  return 0; }   void destroyIpfixReceiver(IpfixReceiver* ipfixReceiver) { close(ipfixReceiver->socket);  if (pthread_mutex_unlock(&ipfixReceiver->mutex) != 0) { DPRINTF("Could not unlock mutex\n"); }  destroyTemplateBuffer(ipfixReceiver->templateBuffer); free(ipfixReceiver->callbackInfo); free(ipfixReceiver); }
static void processDataSet(IpfixReceiver* ipfixReceiver, SourceID sourceId, IpfixSetHeader* set) {	BufferedTemplate* bt = getBufferedTemplate(ipfixReceiver->templateBuffer, sourceId, ntohs(set->id));	if (bt == 0) {		errorf("Template %d unknown to collecting process", ntohs(set->id));		return;		}		#ifdef SUPPORT_NETFLOWV9	if ((bt->setID == IPFIX_SetId_Template) || (bt->setID == NetflowV9_SetId_Template)) {	#else	if (bt->setID == IPFIX_SetId_Template) {	#endif		TemplateInfo* ti = bt->templateInfo;		uint16_t length = ntohs(set->length)-((byte*)(&set->data)-(byte*)set);		byte* record = &set->data;			if (bt->recordLength < 65535) {			byte* recordX = record+length;			if (record >= recordX - (bt->recordLength - 1)) {				error("Got a Data Set that contained not a single full record");				}			/* We stop processing when no full record is left */			while (record < recordX - (bt->recordLength - 1)) {				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->dataRecordCallbackFunction) ci->dataRecordCallbackFunction(ci->handle, sourceId, ti, bt->recordLength, record);					}				record = record + bt->recordLength;				}			} else {			byte* recordX = record+length;			if (record >= recordX - 3) {				error("Got a Data Set that contained not a single full record");				}			/* We assume that all variable-length records are >= 4 byte, so we stop processing when only 3 bytes are left */			while (record < recordX - 3) {				int recordLength=0;				int i;				for (i = 0; i < ti->fieldCount; i++) {					int fieldLength = 0;					if (ti->fieldInfo[i].type.length < 65535) {						fieldLength = ti->fieldInfo[i].type.length;						} else {						if (*(byte*)record < 255) {							fieldLength = *(byte*)record;							} else {							fieldLength = *(uint16_t*)(record+1);							}						}					ti->fieldInfo[i].type.length = fieldLength;					ti->fieldInfo[i].offset = recordLength;					recordLength += fieldLength;					}				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->dataRecordCallbackFunction) ci->dataRecordCallbackFunction(ci->handle, sourceId, ti, recordLength, record);					}				record = record + recordLength;				}			}		} else 	if (bt->setID == IPFIX_SetId_OptionsTemplate) {  			OptionsTemplateInfo* ti = bt->optionsTemplateInfo;		uint16_t length = ntohs(set->length)-((byte*)(&set->data)-(byte*)set);		byte* record = &set->data;			if (bt->recordLength < 65535) {			byte* recordX = record+length;			while (record < recordX) {				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->optionsRecordCallbackFunction) ci->optionsRecordCallbackFunction(ci->handle, sourceId, ti, bt->recordLength, record);					}				record = record + bt->recordLength;				}			} else {			byte* recordX = record+length;			while (record < recordX) {				int recordLength=0;				int i;				for (i = 0; i < ti->scopeCount; i++) {					int fieldLength = 0;					if (ti->scopeInfo[i].type.length < 65535) {						fieldLength = ti->scopeInfo[i].type.length;						} else {						if (*(byte*)record < 255) {							fieldLength = *(byte*)record;							} else {							fieldLength = *(uint16_t*)(record+1);							}						}					ti->scopeInfo[i].type.length = fieldLength;					ti->scopeInfo[i].offset = recordLength;					recordLength += fieldLength;					}				for (i = 0; i < ti->fieldCount; i++) {					int fieldLength = 0;					if (ti->fieldInfo[i].type.length < 65535) {						fieldLength = ti->fieldInfo[i].type.length;						} else {						if (*(byte*)record < 255) {							fieldLength = *(byte*)record;							} else {							fieldLength = *(uint16_t*)(record+1);							}						}					ti->fieldInfo[i].type.length = fieldLength;					ti->fieldInfo[i].offset = recordLength;					recordLength += fieldLength;					}				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->optionsRecordCallbackFunction) ci->optionsRecordCallbackFunction(ci->handle, sourceId, ti, recordLength, record);					}				record = record + recordLength;				}			}		} else   	if (bt->setID == IPFIX_SetId_DataTemplate) {		DataTemplateInfo* ti = bt->dataTemplateInfo;		uint16_t length = ntohs(set->length)-((byte*)(&set->data)-(byte*)set);		byte* record = &set->data;		if (bt->recordLength < 65535) {			byte* recordX = record+length;			while (record < recordX) {				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->dataDataRecordCallbackFunction) ci->dataDataRecordCallbackFunction(ci->handle, sourceId, ti, bt->recordLength, record);					}				record = record + bt->recordLength;				}			} else {			byte* recordX = record+length;			while (record < recordX) {				int recordLength=0;				int i;				for (i = 0; i < ti->fieldCount; i++) {					int fieldLength = 0;					if (ti->fieldInfo[i].type.length < 65535) {						fieldLength = ti->fieldInfo[i].type.length;						} else {						if (*(byte*)record < 255) {							fieldLength = *(byte*)record;							} else {							fieldLength = *(uint16_t*)(record+1);							}						}					ti->fieldInfo[i].type.length = fieldLength;					ti->fieldInfo[i].offset = recordLength;					recordLength += fieldLength;					}				int n;						for (n = 0; n < ipfixReceiver->callbackCount; n++) {					CallbackInfo* ci = &ipfixReceiver->callbackInfo[n];					if (ci->dataDataRecordCallbackFunction) ci->dataDataRecordCallbackFunction(ci->handle, sourceId, ti, recordLength, record);					}				record = record + recordLength;				}			}		} else {		fatalf("Data Set based on known but unhandled template type %d", bt->setID);		}	}