mprobe_pbuffer.c

LInux BootLoader的说明文档

        p->trans_data = (unsigned char *)p + sizeof(packet_t) ;
        p->source_ip = IP_SADDR(ip_packet) ;
        p->target_ip = IP_DADDR(ip_packet) ;
		p->sport = TCP_SPORT(t_packet) ;
		p->dport = TCP_DPORT(t_packet) ;
        p->trans_len = t_len ;
        p->protocol_type= TCP_TYPE;
        memcpy(p->trans_data , t_packet , t_len) ;
        p->app_data = p->trans_data + TCP_HDR_LEN(t_packet) ;
		p->app_len = p->trans_len - TCP_HDR_LEN(t_packet) ;
		p->tv_sec = tm ;
    }
	else if(IP_PROTO_UDP(ip_packet)){/*    UDP数据 */
		struct ether_header * eth = (struct ether_header *)packet ;

		t_packet = ip_packet + IP_HDR_LEN(ip_packet) ;
		t_len = IP_TOTAL_LEN(ip_packet) - IP_HDR_LEN(ip_packet) ;
        p = malloc( t_len * sizeof(u_int8_t) + sizeof(packet_t)) ;
        if(p == NULL){
			fprintf(stdout , "malloc memory failed\n") ;
			*validate = 0 ;
            return p ;
        }
		memcpy(p->s_mac , eth->ether_shost , ETH_ALEN) ;
		memcpy(p->d_mac , eth->ether_dhost , ETH_ALEN) ;
        p->size = t_len * sizeof(u_int8_t) + sizeof(packet_t) ;
        p->trans_data = (unsigned char *)p + sizeof(packet_t) ;
        p->source_ip = IP_SADDR(ip_packet) ;
        p->target_ip = IP_DADDR(ip_packet) ;
	 p->sport = UDP_SPORT(t_packet) ;
	 p->dport = UDP_DPORT(t_packet) ;
        p->trans_len = t_len ;
        p->protocol_type= UDP_TYPE;
        memcpy(p->trans_data , t_packet , t_len) ;
        p->app_data = p->trans_data + sizeof(struct udphdr) ;    
		p->app_len = p->trans_len - sizeof(struct udphdr) ;
		p->tv_sec = tm ;
    }else{
    	*validate = 1 ;
        p = NULL ;
    }

    return p ;
}



/*
    _pbuffer_flush_cache    -   将一个队列的数据加入至另外一个队列
    ->@packets_t  * target_cache    :   被加入的队列
    ->@packets_t *  source_cache    :   加入的队列，该队列将被清空
    !void
*/

void
_pbuffer_flush_cache(packets_t * hdr1 , packets_t * hdr2)
{
/*
    packet_t * tmp = NULL ;
*/
    if(!hdr1 || !hdr2)
        return;
/*
    assert(hdr2->tqh_first) ;
	assert(hdr1->tqh_first == NULL ) ;

    if(hdr2->tqh_first == NULL)
        return  ;

    tmp = hdr2->tqh_first ;
    while(tmp){
        TAILQ_REMOVE(hdr2 , tmp , _list) ;
        TAILQ_INSERT_TAIL(hdr1 , tmp , _list) ;
        tmp = hdr2->tqh_first ;
    }
*/
	assert(hdr2->tqh_first) ;
	assert(hdr1->tqh_first == NULL) ;
	
	hdr1->tqh_first = hdr2->tqh_first;
	hdr1->tqh_first->_list.tqe_prev = &(hdr1->tqh_first);
//	*(hdr1->tqh_last) = (hdr2->tqh_first) ;
	hdr1->tqh_last = hdr2->tqh_last ;
	//*(hdr1->tqh_last) = hdr1->tqh_first;
	
	TAILQ_INIT(hdr2) ;
//	hdr2->tqh_last = NULL;
//	hdr2->tqh_first = NULL;

	return;
}

/*
    _pbuffer_insert_packet	-	将packet_t放入缓冲队列
    ->@pbuffer_t *	pbuffer	:   数据封包缓冲区
    ->@packet_t *	packet  :   用于存放数据封包的结构
    !int					:   放入成功，返回0，否则返回-1
*/
int
_pbuffer_insert_packet(pbuffer_t * pbuffer , packet_t * packet)
{
	int		cache_full = 0 ;

	assert(pbuffer && packet) ;

	//	print_num ++;

    cache_full = (packet->size + pbuffer->cache_memory >= pbuffer->max_memory) ;

	if(!cache_full){/*	预存队列仍然拥有空间,将数据包放入预存队列	*/
		TAILQ_INSERT_TAIL(&(pbuffer->cache) , packet , _list) ;
		pbuffer->cache_memory  += packet->size ;
		return 0 ;
	}


	/*	预存队列没有空间,等待缓冲区为空时将数据一次倒入	*/
	pthread_mutex_lock(&(pbuffer->mutex)) ;
	while((pbuffer->buffer).tqh_first != NULL)
	{
		printf("************pcap thread begin sleep\n");
		pthread_cond_wait(&(pbuffer->cond_empty) , &(pbuffer->mutex)) ; 
		printf("~~~~~~~~~~~~~~~~pcap thread wake up\n");
	}
	assert(pbuffer->cache.tqh_first) ;
	_pbuffer_flush_cache(&(pbuffer->buffer) , &(pbuffer->cache)) ;
	pbuffer->use_memory = pbuffer->cache_memory ;
	pbuffer->cache_memory = 0 ;
	pthread_mutex_unlock(&(pbuffer->mutex)) ;

	TAILQ_INSERT_TAIL(&(pbuffer->cache) , packet , _list) ;
	pbuffer->cache_memory += packet->size ;

	pthread_cond_broadcast(&(pbuffer->cond_full)) ;
	

	return 0 ;
}

/*
    _pbuffer_destroy_queue  -   释放缓存数据封包队列
    ->@packets_t    *   packets     :   数据封包队列
    !void                                   :   
*/    

void
_pbuffer_destroy_queue(packets_t  * packets)
{
    packet_t    *   tmp = NULL ;

    if(!packets)
        return;

    tmp = packets->tqh_first ;
    while(tmp){
        TAILQ_REMOVE(packets , tmp , _list) ;
        free( tmp) ;
        tmp = packets->tqh_first ;
    }

    return;
}

/*****************************************************************************/

#include <unistd.h>
#include <pcap.h>
#include <signal.h>

#ifdef MPROBE_PBUFFER_DEMO

#ifndef DEVICE
#define DEVICE "eth1"
#endif			/*	#ifndef DEVICE	*/

#ifndef SNAPLEN
#define SNAPLEN  1518
#endif			/*	#ifndef SNAPLEN	*/


void * __pcb = NULL ;


volatile sig_atomic_t done = 0;

void
ether_print(const char * packet) ;

void
ip_print(const char * ip_packet) ;

void
udp_print(const char * udp_packet) ;

void 
tcp_print(const char * tcp_packet) ;

void *
capture(void *) ;

void *
decoder(void *) ;

int
_set_filter(pcap_t * pcap , char * bpf) ;

static
void
_sig_handler(int signo) ;


int
main(int argc , char * argv[])
{
    pthread_t capture_id ;
    pthread_t decoder_id ;
 
    __pcb = pbuffer_open(2*KB) ;
    if(__pcb == NULL){
        fprintf(stderr , "获取缓冲区失败,程序退出\n") ;
        exit(1) ;
    }
    signal(SIGINT , _sig_handler) ;
    pthread_create(&capture_id , NULL , capture , NULL) ;
    pthread_create(&decoder_id , NULL , decoder , NULL) ;

    pause() ;
    pthread_cancel(capture_id) ;
    pthread_cancel(decoder_id) ;
    fprintf(stdout , "pbuffer->cache_memory = %u\n" , ((pbuffer_t *)__pcb)->cache_memory) ;
    fprintf(stdout , "pbuffer->use_memory = %u\n" , ((pbuffer_t *)__pcb)->use_memory) ;
    fprintf(stdout , "pbuffer->prefetch_memory = %u\n" , ((pbuffer_t *)__pcb)->prefetch_memory) ;
    pbuffer_close(__pcb) ;
    __pcb = NULL ;
    exit(0) ;

}

void *
capture(void * para)
{
	pcap_t * pcap ;
	char pcap_errbuf[PCAP_ERRBUF_SIZE] ;
	const char * packet = NULL ;
	struct pcap_pkthdr hdr ;
    
	pcap = pcap_open_live(DEVICE , SNAPLEN , 1 , 0 , pcap_errbuf) ;
	if(pcap == NULL){
		fprintf(stderr , "初始化网络接口%s失败(%s)\n" , 
						DEVICE , pcap_errbuf) ;
		exit(1) ;
	}

    _set_filter(pcap , "tcp port 25") ;
	fprintf(stdout , "开始在网络接口%s进行监听...\n" , DEVICE) ;
	fprintf(stdout , "链路层信息:%s(%s)\n",
					pcap_datalink_val_to_name(pcap_datalink(pcap)) ,
					pcap_datalink_val_to_description(pcap_datalink(pcap))) ;

	while(!done){
		packet = pcap_next(pcap , &hdr) ;
		if(packet == NULL) break ;
        pbuffer_insert_packet(__pcb , packet , hdr.caplen) ;
    }

    return NULL ;
}

void *
decoder(void * para)
{
	packet_t * packet = NULL ;
	static int count = 0 ;

    while(!done){
        packet = pbuffer_get_packet(__pcb) ;
      	fprintf(stdout , "源IP地址:%d.%d.%d.%d\n" , _IP_ADDR_FORMAT(packet->source_ip)) ;
       	fprintf(stdout , "目的IP地址:%d.%d.%d.%d\n" , _IP_ADDR_FORMAT(packet->target_ip)) ;
       	if(packet->tcp){
       		tcp_print(packet->data) ;
       	}else{
       		udp_print(packet->data) ;
       	}
        free(packet) ;
    }

    return NULL ;
}
 
void
udp_print(const char * udp_packet)
{
	fprintf(stdout , "源端口:%d\n" , UDP_SPORT(udp_packet)) ;
	fprintf(stdout , "目的端口:%d\n" , UDP_DPORT(udp_packet)) ;
	fprintf(stdout , "UDP包长度:%d\n" , UDP_LEN(udp_packet)) ;
}

void
tcp_print(const char * tcp_packet)
{
	fprintf(stdout , "源端口:%u\n" , TCP_SPORT(tcp_packet)) ;
	fprintf(stdout , "目的端口:%u\n" , TCP_DPORT(tcp_packet)) ;
	fprintf(stdout , "sequence号:%u\n" , TCP_SEQ(tcp_packet)) ;
	fprintf(stdout , "acknowledgement号:%u\n" , TCP_ACK(tcp_packet)) ;
	fprintf(stdout , "TCP头长度:%d\n" , TCP_HDR_LEN(tcp_packet)) ;
	fprintf(stdout , "TCP标志:ACK %s\tPSH %s\tRST %s\tSYN %s\tFIN %s\n" , 
					TCP_FLAG_ACK(tcp_packet)?"是":"否" ,
					TCP_FLAG_PSH(tcp_packet)?"是":"否" ,
					TCP_FLAG_RST(tcp_packet)?"是":"否" ,
					TCP_FLAG_SYN(tcp_packet)?"是":"否" ,
					TCP_FLAG_FIN(tcp_packet)?"是":"否") ;
}

int
_set_filter(pcap_t * pcap , char * bpf)
{
    struct bpf_program  fp ;
   
    if(pcap_compile(pcap , &fp , bpf , 1 , 0) < 0){
        return -1 ;
    }

    if(pcap_setfilter(pcap , &fp) < 0){
        return -1 ;
    }

    return 0 ;
}

void
_sig_handler(int signo)
{
	done = 1 ;
    return;
}

#endif      /*  #ifdef MPROBE_PBUFFER_DEMO  */