fprobe.c

模仿cisco路由器

		if (cmpmtime(&flow->ctime, &flown->ctime) < 0)
			flown->ctime = flow->ctime;
		flown->tcp_flags |= flow->tcp_flags;
		flown->size += flow->size;
		flown->pkts += flow->pkts;
		if (flow->flags & FLOW_FRAG) {
			/* Fragmented flow require some additional work */
			if (flow->flags & FLOW_TL) {
				/*
				?FIXME?
				Several packets with FLOW_TL (attack)
				*/
				flown->sp = flow->sp;
				flown->dp = flow->dp;
			}
			if (flow->flags & FLOW_LASTFRAG) {
				/*
				?FIXME?
				Several packets with FLOW_LASTFRAG (attack)
				*/
				flown->sizeP = flow->sizeP;
			}
			flown->flags |= flow->flags;
			flown->sizeF += flow->sizeF;
			if ((flown->flags & FLOW_LASTFRAG)
				&& (flown->sizeF >= flown->sizeP)) {
				/* All fragments received - flow reassembled */
				*flowpp = flown->next;
				pthread_mutex_unlock(&flows_mutex[h]);
#if ((DEBUG) & DEBUG_I)
				flows_total--;
				flows_fragmented--;
#endif
				flown->id = 0;
				flown->flags &= ~FLOW_FRAG;
#if ((DEBUG) & (DEBUG_U | DEBUG_S))
				strcat(logbuf," R");
#endif
				ret = put_into(flown, MOVE_INTO
#if ((DEBUG) & (DEBUG_U | DEBUG_S))
						, logbuf
#endif
					);
			}
		}
		if (flag == MOVE_INTO) mem_free(flow);
	}
	pthread_mutex_unlock(&flows_mutex[h]);
	return ret;
}

void *fill(int fields, uint16_t *format, struct Flow *flow, void *p)
{
	int i;

	for (i = 0; i < fields; i++) {
#if ((DEBUG) & DEBUG_F)
		my_log(LOG_DEBUG, "F: field %04d at %x", format[i], (unsigned) p);
#endif
		switch (format[i]) {
			case NETFLOW_IPV4_SRC_ADDR:
				((struct in_addr *) p)->s_addr = flow->sip.s_addr;
				p += NETFLOW_IPV4_SRC_ADDR_SIZE;
				break;

			case NETFLOW_IPV4_DST_ADDR:
				((struct in_addr *) p)->s_addr = flow->dip.s_addr;
				p += NETFLOW_IPV4_DST_ADDR_SIZE;
				break;

			case NETFLOW_INPUT_SNMP:
				*((uint16_t *) p) = snmp_input_index;
				p += NETFLOW_INPUT_SNMP_SIZE;
				break;

			case NETFLOW_OUTPUT_SNMP:
				*((uint16_t *) p) = snmp_output_index;
				p += NETFLOW_OUTPUT_SNMP_SIZE;
				break;

			case NETFLOW_PKTS_32:
				*((uint32_t *) p) = htonl(flow->pkts);
				p += NETFLOW_PKTS_32_SIZE;
				break;

			case NETFLOW_BYTES_32:
				*((uint32_t *) p) = htonl(flow->size);
				p += NETFLOW_BYTES_32_SIZE;
				break;

			case NETFLOW_FIRST_SWITCHED:
				*((uint32_t *) p) = htonl(getuptime(&flow->ctime));
				p += NETFLOW_FIRST_SWITCHED_SIZE;
				break;

			case NETFLOW_LAST_SWITCHED:
				*((uint32_t *) p) = htonl(getuptime(&flow->mtime));
				p += NETFLOW_LAST_SWITCHED_SIZE;
				break;

			case NETFLOW_L4_SRC_PORT:
				*((uint16_t *) p) = flow->sp;
				p += NETFLOW_L4_SRC_PORT_SIZE;
				break;

			case NETFLOW_L4_DST_PORT:
				*((uint16_t *) p) = flow->dp;
				p += NETFLOW_L4_DST_PORT_SIZE;
				break;

			case NETFLOW_PROT:
				*((uint8_t *) p) = flow->proto;
				p += NETFLOW_PROT_SIZE;
				break;

			case NETFLOW_SRC_TOS:
				*((uint8_t *) p) = flow->tos;
				p += NETFLOW_SRC_TOS_SIZE;
				break;

			case NETFLOW_TCP_FLAGS:
				*((uint8_t *) p) = flow->tcp_flags;
				p += NETFLOW_TCP_FLAGS_SIZE;
				break;

			case NETFLOW_VERSION:
				*((uint16_t *) p) = htons(netflow->Version);
				p += NETFLOW_VERSION_SIZE;
				break;

			case NETFLOW_COUNT:
				*((uint16_t *) p) = htons(emit_count);
				p += NETFLOW_COUNT_SIZE;
				break;

			case NETFLOW_UPTIME:
				*((uint32_t *) p) = htonl(getuptime(&emit_time));
				p += NETFLOW_UPTIME_SIZE;
				break;

			case NETFLOW_UNIX_SECS:
				*((uint32_t *) p) = htonl(emit_time.sec);
				p += NETFLOW_UNIX_SECS_SIZE;
				break;

			case NETFLOW_UNIX_NSECS:
				*((uint32_t *) p) = htonl(emit_time.usec * 1000);
				p += NETFLOW_UNIX_NSECS_SIZE;
				break;

			case NETFLOW_FLOW_SEQUENCE:
				*((uint32_t *) p) = htonl(emit_sequence);
				p += NETFLOW_FLOW_SEQUENCE_SIZE;
				break;

			case NETFLOW_PAD8:
			/* Unsupported (uint8_t) */
			case NETFLOW_ENGINE_TYPE:
			case NETFLOW_ENGINE_ID:
			case NETFLOW_FLAGS7_1:
			case NETFLOW_SRC_MASK:
			case NETFLOW_DST_MASK:
				*((uint8_t *) p) = 0;
				p += NETFLOW_PAD8_SIZE;
				break;

			case NETFLOW_PAD16:
			/* Unsupported (uint16_t) */
			case NETFLOW_SRC_AS:
			case NETFLOW_DST_AS:
			case NETFLOW_FLAGS7_2:
				*((uint16_t *) p) = 0;
				p += NETFLOW_PAD16_SIZE;
				break;

			case NETFLOW_PAD32:
			/* Unsupported (uint32_t) */
			case NETFLOW_IPV4_NEXT_HOP:
			case NETFLOW_ROUTER_SC:
				*((uint32_t *) p) = 0;
				p += NETFLOW_PAD32_SIZE;
				break;

			default:
				my_log(LOG_CRIT, "fill(): Unknown format at %x[%d]: %d",
					format, i, format[i]);
				exit(1);
		}
	}
#if ((DEBUG) & DEBUG_F)
	my_log(LOG_DEBUG, "F: return %x", (unsigned) p);
#endif
	return p;
}

void *emit_thread()
{
	struct Flow *flow;
	void *p;
	struct timeval now;
	struct timespec timeout;
	unsigned sent = 0, size, ret;
#if ((DEBUG) & DEBUG_E)
	void *pp;
	int cnt, seq = 0;
	char sip[16], dip[16];
#endif

	p = (void *) &emit_packet + netflow->HeaderSize;
	timeout.tv_nsec = 0;

	for (;;) {
		pthread_mutex_lock(&emit_mutex);
		while (!flows_emit) {
			gettimeofday(&now, 0);
			timeout.tv_sec = now.tv_sec + emit_timeout;
			/* Do not wait until emit_packet will filled - it may be too long */
			if (pthread_cond_timedwait(&emit_cond, &emit_mutex, &timeout) && emit_count) {
				pthread_mutex_unlock(&emit_mutex);
				goto sendit;
			}
		}
		flow = flows_emit;
		flows_emit = flows_emit->next;
#if ((DEBUG) & DEBUG_I)
		emit_queue--;
#endif		
		pthread_mutex_unlock(&emit_mutex);

#ifdef UPTIME_TRICK
		if (!emit_count) {
			gettime(&start_time);
			start_time.sec -= start_time_offset;
		}
#endif
#if ((DEBUG) & DEBUG_E)
		sprintf(sip, "%s", inet_ntoa(flow->sip));
		sprintf(dip, "%s", inet_ntoa(flow->dip));
		my_log(LOG_DEBUG,"E: %d(%d) %d/%d %s>%s P:%x TCP:%x %d>%d",
			flow->ctime.sec, flow->mtime.sec - flow->ctime.sec,
			flow->size, flow->pkts, sip, dip, flow->proto, flow->tcp_flags,
			ntohs(flow->sp), ntohs(flow->dp));
#endif		
		p = fill(netflow->FlowFields, netflow->FlowFormat, flow, p);
		mem_free(flow);
		emit_count++;
		if (emit_count == netflow->MaxFlows) {
		sendit:
			gettime(&emit_time);
			p = fill(netflow->HeaderFields, netflow->HeaderFormat, 0, &emit_packet);

#if ((DEBUG) & DEBUG_E)
			pp = (void *) &emit_packet;
			my_log(LOG_DEBUG, "E: Hdr ver:%d cnt:%d uptime:%d secs:%d nsecs:%d",
				ntohs(*(uint16_t*)(pp + 0)),
				ntohs(*(uint16_t*)(pp + 2)),
				ntohl(*(uint32_t*)(pp + 4)),
				ntohl(*(uint32_t*)(pp + 8)),
				ntohl(*(uint32_t*)(pp + 12))
			);
			cnt= ntohs(*(uint16_t *)(pp + 2));
			pp = (void *) &emit_packet + netflow->HeaderSize;
			for (; cnt--;) {
				sprintf(sip, "%s", inet_ntoa(*(struct in_addr*)(pp + 0)));
				sprintf(dip, "%s", inet_ntoa(*(struct in_addr*)(pp + 4)));
				my_log(LOG_DEBUG, "E: #%d %d-%d %d/%d %s>%s P:%x TCP:%x %d>%d",
					seq++,
					ntohl(*(uint32_t*)(pp + 24)),
					ntohl(*(uint32_t*)(pp + 28)),
					ntohl(*(uint32_t*)(pp + 20)),
					ntohl(*(uint32_t*)(pp + 16)),
					sip,
					dip,
					*(uint8_t *)(pp + 38),
					*(uint8_t *)(pp + 37),
					ntohs(*(uint16_t*)(pp + 32)),
					ntohs(*(uint16_t*)(pp + 34))
				);
				pp += netflow->FlowSize;
			}
#endif

			size = netflow->HeaderSize + emit_count * netflow->FlowSize;
			ret = sendto(sock, emit_packet, size, 0,
					(struct sockaddr *) &server, sizeof(server));
			if ( ret < size) {
#if ((DEBUG) & DEBUG_E) || defined MESSAGES
				my_log(LOG_ERR, "sendto() == %d: %s", ret, strerror(errno));
#endif
			}
#if ((DEBUG) & DEBUG_E)
			else {
				my_log(LOG_DEBUG, "E: Emitted %d flow(s)", emit_count);
			}
#endif
			emit_sequence += emit_count;
			emit_count = 0;
#if ((DEBUG) & DEBUG_I)
			emit_pkts++;
#endif

			/* Rate limit */
			if (emit_rate_bytes) {
				sent += size;
				size = sent / emit_rate_bytes;
				if (size) {
					sent %= emit_rate_bytes;
					timeout.tv_sec = 0;
					timeout.tv_nsec = emit_rate_delay * size;
					while (nanosleep(&timeout, &timeout) == -1 && errno == EINTR);
				}
			}
		}
	}
}	

void *unpending_thread()
{
	struct timeval now;
	struct timespec timeout;
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
	char logbuf[256];
#endif

	timeout.tv_nsec = 0;
	pthread_mutex_lock(&unpending_mutex);

	for (;;) {
		while (!(pending_tail->flags & FLOW_PENDING)) {
			gettimeofday(&now, 0);
			timeout.tv_sec = now.tv_sec + unpending_timeout;
			pthread_cond_timedwait(&unpending_cond, &unpending_mutex, &timeout);
		}

#if ((DEBUG) & (DEBUG_S | DEBUG_U))
		*logbuf = 0;
#endif
		if (put_into(pending_tail, COPY_INTO
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
				, logbuf
#endif
			) < 0) {
#if ((DEBUG) & DEBUG_I)
			pkts_lost_unpending++;
#endif				
		}

#if ((DEBUG) & DEBUG_U)
		my_log(LOG_DEBUG, "%s%s", "U:", logbuf);
#endif

		pending_tail->flags = 0;
		pending_tail = pending_tail->next;
#if ((DEBUG) & DEBUG_I)
		pkts_pending_done++;
#endif
	}
}

void *scan_thread()
{
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
	char logbuf[256];
#endif
	int i;
	struct Flow *flow, **flowpp;
	struct Time now;
	struct timespec timeout;

	timeout.tv_nsec = 0;
	pthread_mutex_lock(&scan_mutex);

	for (;;) {
		gettime(&now);
		timeout.tv_sec = now.sec + scan_interval;
		pthread_cond_timedwait(&scan_cond, &scan_mutex, &timeout);

		gettime(&now);
#if ((DEBUG) & DEBUG_S)
		my_log(LOG_DEBUG, "S: %d", now.sec);
#endif
		for (i = 0; i < 1 << HASH_BITS ; i++) {
			pthread_mutex_lock(&flows_mutex[i]);
			flow = flows[i];
			flowpp = &flows[i];
			while (flow) {
				if (flow->flags & FLOW_FRAG) {
					/* Process fragmented flow */
					if ((now.sec - flow->mtime.sec) > frag_lifetime) {
						/* Fragmented flow expired - put it into special chain */
#if ((DEBUG) & DEBUG_I)
						flows_fragmented--;
						flows_total--;
#endif
						*flowpp = flow->next;
						flow->id = 0;
						flow->flags &= ~FLOW_FRAG;
						flow->next = scan_frag_dreg;
						scan_frag_dreg = flow;
						flow = *flowpp;
						continue;
					}
				} else {
					/* Flow is not frgamented */
					if ((now.sec - flow->mtime.sec) > inactive_lifetime
						|| (flow->mtime.sec - flow->ctime.sec) > active_lifetime) {
						/* Flow expired */
#if ((DEBUG) & DEBUG_S)
						my_log(LOG_DEBUG, "S: E %x", flow);
#endif
#if ((DEBUG) & DEBUG_I)
						flows_total--;
#endif
						*flowpp = flow->next;
						pthread_mutex_lock(&emit_mutex);
						flow->next = flows_emit;
						flows_emit = flow;
#if ((DEBUG) & DEBUG_I)
						emit_queue++;
#endif				
						pthread_mutex_unlock(&emit_mutex);
						flow = *flowpp;
						continue;
					}
				}
				flowpp = &flow->next;
				flow = flow->next;
			} /* chain loop */
			pthread_mutex_unlock(&flows_mutex[i]);
		} /* hash loop */
		if (flows_emit) pthread_cond_signal(&emit_cond);

		while (scan_frag_dreg) {
			flow = scan_frag_dreg;
			scan_frag_dreg = flow->next;
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
			*logbuf = 0;
#endif
			put_into(flow, MOVE_INTO
#if ((DEBUG) & (DEBUG_S | DEBUG_U))
				, logbuf
#endif
			);
#if ((DEBUG) & DEBUG_S)
			my_log(LOG_DEBUG, "%s%s", "S: FE", logbuf);
#endif
		}
	}
}

void pcap_callback(u_char *useless,
	const struct pcap_pkthdr* pkthdr, const u_char *packet)
{
	struct ip *nl;
	void *tl;
	struct Flow *flow;
	int off_frag, psize;
#if ((DEBUG) & DEBUG_C)
	char buf[64];
	char logbuf[256];
#endif

	if (killed) return; /* SIGTERM received - stop capturing */

#if ((DEBUG) & DEBUG_C)
	sprintf(logbuf, "C: %d/%d", pkthdr->caplen, pkthdr->len);
#endif

	/* Offset (from begin of captured packet) to network layer header */
	nl = (void *) packet + off_nl;
	psize = pkthdr->caplen - off_nl;

	/* Sanity check */
	if (psize < (signed) sizeof(struct ip) || nl->ip_v != 4) {
#if ((DEBUG) & DEBUG_C)
		strcat(logbuf, " U");
		my_log(LOG_DEBUG, "%s", logbuf);
#endif
#if ((DEBUG) & DEBUG_I)
		pkts_ignored++;
#endif
		return;
	}

	if (pending_head->flags) {
#if ((DEBUG) & DEBUG_C) || defined MESSAGES
		my_log(LOG_ERR,
# if ((DEBUG) & DEBUG_C)
			"%s %s %s", logbuf,
# else
			"%s %s",
# endif
			"pending queue full:", "packet lost");
#endif
#if ((DEBUG) & DEBUG_I)
		pkts_lost_capture++;
#endif
		goto done;
	}