tcphashmain.cpp

linux 上http email 协议分析程序 主要能够处理大数据量的主干网的应用

	g_p_memAdm = new CMem();
	if( !g_p_memAdm->init(C_ARR_MEM_STREAM_COUNT,C_ARR_MEM_STREAM_SIZE )  )
	{
		writelog("init g_memAdm fail!");
		return  ;
	}
	printf("combine tcp stream use:");
	g_p_memAdm ->showTotalBytes();
	////////////////////////

	//初始化共享内存相关变量
	g_p_csa_in  = new CShmApply;
	g_p_csa_out_http = new CShmApply;
	g_p_csa_out_email = new CShmApply;
	if(g_p_csa_in->init(_rkey,_rsize,0666)!=0)
	{
		writelog("init share memory (g_p_csa_in) fail!\n");
		return ;
	}
	if(g_p_csa_out_http->init(_wkey_http,_wsize_http,0666)!=0)
	{
		writelog("init share memory (g_p_csa_in) fail!\n");
		return ;
	}
	if(g_p_csa_out_email->init(_wkey_email,_wsize_email,0666)!=0)
	{
		writelog("init share memory (g_p_csa_in) fail!\n");
		return ;
	}

	while(true)
	{
		//get IP Packet
		for (buff_counter= 0;buff_counter < C_MAX_PACKET_BUFFER;buff_counter++)
		{
			result = g_p_csa_in->read(rbuffer[buff_counter]);
			if(result <=0)
				break;
		}
		if(buff_counter == 0)
		{
			usleep(1);
			continue;
		}
		for(k = 0;k< buff_counter;k++)
		{

			iPacketCounter++;
			pip = (IP_PACKET*)rbuffer[k];
			ptcp = (TCP_PACKET*)((char*)(rbuffer[k])+get_ip_headlen(pip));

			if (pip == NULL)
				continue;
			li.reinit(get_ip_src_ip(pip),get_tcp_src_port(ptcp),get_ip_dst_ip(pip),get_tcp_dst_port(ptcp));
			ps = (CTCPStream*)hashtable->get(li);
			if (ps == NULL )//对应流不存在
			{
				if(get_tcp_syn(ptcp) )//新流
				{
					if(hashtable->size() == C_MAX_STREAM_COUNT - 10)//同时存在的流超过最大数，
						continue;//放弃
					sport = get_tcp_src_port(ptcp);
					dport = get_tcp_dst_port (ptcp);
					printf ("sip[%u],dip[%u]\n",get_ip_src_ip(pip),get_ip_dst_ip(pip));
					/*if ( !(
								(sport == 20480 && watchip.isFankongIP(get_ip_dst_ip(pip))) || 
								(dport==20480 &&  watchip.isFankongIP(get_ip_src_ip(pip))) ||
								watchip.isInWatched(get_ip_src_ip(pip)) || 
								watchip.isInWatched(get_ip_dst_ip(pip))  ||
								dport  == 6400 || sport == 6400 ||
								dport  == 47873||
								dport == 28160 || sport == 28160)
					   )
					{
						//非监控对象,而且不是EMail,也不是请求网页动作
						//丢弃
						continue;
					}*/
					/*
					   if (dport == 20480)
					   {
					   struct in_addr inaddr;
					   inaddr.s_addr = get_ip_dst_ip(pip);
					   printf("fankong:%s:port[%u]\n",inet_ntoa(inaddr),ntohs(dport));

					   }
					 */
#ifdef _DEBUG
					if (watchip.isInWatched(get_ip_src_ip(pip)))
					{
						struct in_addr inaddr;
						inaddr.s_addr = get_ip_src_ip(pip);
						printf("tekong src ip:[%s:%d]\n",inet_ntoa(inaddr),ntohs(sport));
					}
					if (watchip.isInWatched(get_ip_dst_ip(pip)))
					{
						struct in_addr inaddr;
						inaddr.s_addr = get_ip_dst_ip(pip);
						printf("tekong src ip:[%s:%d]\n",inet_ntoa(inaddr),ntohs(dport));
					}
#endif

					ptmp =  g_p_memAdm->maloc(sizeof(CTCPStream));
					if(ptmp == NULL)
					{
						writelog("apply space for a new stream fail!");
						return;
					}
					ps = new(ptmp) CTCPStream();
					if(ps ==NULL)
					{
						writelog("cast char* %p to class fail!\n",ptmp);
						g_p_memAdm->free(ptmp);
						return;
					}

					if(ps->init(C_TCP_STREAM_INIT_SIZE,pip,g_p_memAdm) < 0)//流初始化失败
					{
						writelog("init stream fail!hash count:%d",hashtable->size());
						ps->clear();
						if(!g_p_memAdm->free((char*)ps) )
							writelog("%s %d:free space fail!",__FILE__,__LINE__);
						ps = NULL;
						continue;
					}
					//保存到哈希表
					hashtable->put(li,ps);
				}
				else
				{
					//		printf("无主包\n");
					continue;//丢包
				}
				continue;

			}
			//对应流存在，如果这条流是重新建立的,例如某人按了浏览器刷新按钮
			if(get_tcp_syn(ptcp))
			{
				//printf("reinit stream!\n");
				//从新初始化
				if(tcp_checksum(pip) != 0)
					continue;//错包，不需要reinit
				if(ps->reinit(pip) < 0)
				{
					ps->clear();
					if(!g_p_memAdm->free((char*)ps) )
						writelog("%s %d:free space fail!",__FILE__,__LINE__);
					hashtable->remove(li);
				}
				continue;
			}
			result = ps->addPacket(pip);
			if(result < 0)//加包失败
			{
#ifdef _DEBUG
				printf("add packet fail\n");
#endif
				if(result < -1)
				{//流太长，或者内存不足，丢弃该流
#ifdef _DEBUG
					printf("one stream abort!(result:%d)\n",result);
					writelog("one stream abort!(result:%d)\n",result);
#endif
					ps->clear();
					if(!g_p_memAdm->free((char*)ps) )
						writelog("%s %d:free space fail!",__FILE__,__LINE__);
					hashtable->remove(li);
				}
				continue;
			}
			G_TIME_NOW = time(NULL);//更新当前时间
			if(ps->isStreamFinish() || ps->isStreamAbort())
			{
				if(ps->getReceivedRate() >= RECEIVEDRATE && ps->length() > 1)
				{
					//丢包率在某个范围一下的
					//扔给王治宇处理
					outputstream(ps);
				}//end if(ps->getReceivedRate() >= RECEIVEDRATE && ps->length() > 1)
				else
					outputstream2File(ps);
				//删除当前流
				hashtable->remove(li);
				ps->clear();
				if(!g_p_memAdm->free((char*)ps) )
					writelog("%s %d:free space fail!",__FILE__,__LINE__);
				continue;
			}//end if(ps->isStreamFinish() || ps->isStreamAbort())
			//debug

			//每隔一段时间(增加C_CHECK_TIMEOUT个包检查一次),论询所有TCP流,看有没有超时的流
			//if(iPacketCounter > C_CHECK_TIMEOUT*hashtable->size()*0.8)
			if (G_TIME_NOW - time_last_check_timeout > C_TIMEOUT_SECONDS)
			{
				time_last_check_timeout = G_TIME_NOW;
				writelog("readBuffer Usage:%.2f%%",g_p_csa_in->userate()*100);
				writelog("writeBuffer Usage:(HTTP)%.2f%%,(EMAIL)%.2f%%", g_p_csa_out_http->userate()*100,
						g_p_csa_out_email->userate()*100);
				g_p_memAdm->usage();
				iPacketCounter = 0;//复位计数器
				hashtable->remove(oughtDel);//传入函数
				writelog("readBuffer Usage:%.2f%%",g_p_csa_in->userate()*100);

			}
		}//end for(k = 0;k<buff_counter;k++)
	}//end while(true)
}

int main(int argc,char* argv[])
{
	if(argc < 2 )
	{
		printf("参数不足.\n用法:%s 配置文件\n",argv[0]);
		exit(1);
	}
	CConfig cfg;
	if(cfg.ReadConfig(argv[1]) !=0)
	{
		printf("读取配置文件%s失败!\n",argv[1]);
		exit(1);
	};
	char tmp[255];
	if(cfg.GetItemValue("COLLSHM","key_tcp",tmp)!=0)
	{
		printf("配置文件格式错误  [COLLSHM]:key_tcp\n");
		exit(1);
	}
	key_t rkey;
	rkey = (key_t)atoi(tmp);
	if(rkey <=0)
	{
		printf("配置文件[COLLSHM]:key_tcp必须为整数\n");
		exit(1);
	}
	if(cfg.GetItemValue("COLLSHM","size_tcp",tmp)!=0)
	{
		printf("配置文件格式错误  [COLLSHM]:size_tcp\n");
		exit(1);
	}
	int rsize;
	rsize = atoi(tmp)*1024*1024;
	if(rsize <=0)
	{
		printf("配置文件[COLLSHM]:size_tcp必须为整数,单位Mbytes\n");
		exit(1);
	}
	//写共享的配置
	if(cfg.GetItemValue("COMBINESHM","key_http",tmp)!=0)
	{
		printf("配置文件格式错误  [COMBINESHM]:key_http");
		exit(1);
	}
	key_t wkey_http;
	wkey_http = (key_t)atoi(tmp);
	if(wkey_http <=0)
	{
		printf("配置文件[COMBINESHM]:key_http必须为整数\n");
		exit(1);
	}
	if(cfg.GetItemValue("COMBINESHM","size_http",tmp)!=0)
	{
		printf("配置文件格式错误  [COMBINESHM]:size_http\n");
		exit(1);
	}
	int wsize_http;
	wsize_http = atoi(tmp)*1024*1024;
	if(wsize_http <=0)
	{
		printf("配置文件[COMBINESHM]:size_http必须为整数,单位Mbytes\n");
		exit(1);
	}

	//email
	if(cfg.GetItemValue("COMBINESHM","key_email",tmp)!=0)
	{
		printf("配置文件格式错误  [COMBINESHM]:key_email");
		exit(1);
	}
	key_t wkey_email;
	wkey_email= (key_t)atoi(tmp);
	if(wkey_email<=0)
	{
		printf("配置文件[COMBINESHM]:key_email必须为整数\n");
		exit(1);
	}
	if(cfg.GetItemValue("COMBINESHM","size_email",tmp)!=0)
	{
		printf("配置文件格式错误  [COMBINESHM]:size_email\n");
		exit(1);
	}
	int wsize_email;
	wsize_email = atoi(tmp)*1024*1024;
	if(wsize_email <=0)
	{
		printf("配置文件[COMBINESHM]:size_email必须为整数,单位Mbytes\n");
		exit(1);
	}

	//radius
	if(cfg.GetItemValue("IPWATCH","key",tmp)!=0)
	{
		printf("配置文件格式错误  [IPWATCH]:key");
		exit(1);
	}
	key_t key_ipwatch;
	key_ipwatch= (key_t)atoi(tmp);
	if(key_ipwatch<=0)
	{
		printf("配置文件[IPWATCH]:key必须为整数\n");
		exit(1);
	}
	if(cfg.GetItemValue("IPWATCH","size",tmp)!=0)
	{
		printf("配置文件格式错误  [IPWATCH]:size\n");
		exit(1);
	}
	int size_ipwatch;
	size_ipwatch = atoi(tmp)*1024*1024;
	if(size_ipwatch <=0)
	{
		printf("配置文件[IPWATCH]:size必须为整数,单位Mbytes\n");
		exit(1);
	}

	if(openlog("../log/cbtcphash.log")!=0)
	{
		exit(1);
	}

	writelog("init ok,running....");

	signal(SIGTERM,SIG_IGN);
	signal(SIGINT,SIG_IGN);
	signal(SIGQUIT,SIG_IGN);
	signal(SIGCHLD,SIG_IGN);
	signal(SIGPIPE,SIG_IGN);

	run( rkey,wkey_http,wkey_email,key_ipwatch,rsize,wsize_http,wsize_email,size_ipwatch);
}