gnutella.cpp.svn-base

这是和p2p相关的一份源码

// ------------------------------------------------
// File : gnutella.cpp
// Date: 4-apr-2002
// Author: giles
// Desc: 
//		GnuPacket is a Gnutella protocol packet.
//		GnuStream is a Stream that reads/writes GnuPackets
//
//
// (c) 2002 peercast.org
// 
// ------------------------------------------------
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.

// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
// ------------------------------------------------

#include "gnutella.h"
#include "stream.h"
#include "common.h"
#include "servent.h"
#include "servmgr.h"
#include "stats.h"
#include <stdlib.h>

// ---------------------------
const char *GNU_FUNC_STR(int func)
{
	switch (func)
	{
		case GNU_FUNC_PING: return "PING"; break;
		case GNU_FUNC_PONG: return "PONG"; break;
		case GNU_FUNC_QUERY: return "QUERY"; break;
		case GNU_FUNC_HIT: return "HIT"; break;
		case GNU_FUNC_PUSH: return "PUSH"; break;
		default: return "UNKNOWN"; break;
	}
}

// ---------------------------
const char *GnuStream::getRouteStr(R_TYPE r)
{
	switch(r)
	{
		case R_PROCESS: return "PROCESS"; break;
		case R_DEAD: return "DEAD"; break;
		case R_DISCARD: return "DISCARD"; break;
		case R_ACCEPTED: return "ACCEPTED"; break;
		case R_BROADCAST: return "BROADCAST"; break;
		case R_ROUTE: return "ROUTE"; break;
		case R_DUPLICATE: return "DUPLICATE"; break;
		case R_BADVERSION: return "BADVERSION"; break;
		default: return "UNKNOWN"; break;
	}
}
// ---------------------------
void GnuPacket::makeChecksumID()
{
	for(unsigned int i=0; i<len; i++)
		id.id[i%16] += data[i];	
}

// ---------------------------
void GnuPacket::initPing(int t)
{
	func = GNU_FUNC_PING;
	ttl = t;
	hops = 0;
	len = 0;

	id.generate();
}
// ---------------------------
void GnuPacket::initPong(Host &h, bool ownPong, GnuPacket &ping)
{
	func = GNU_FUNC_PONG;
	ttl = ping.hops;
	hops = 0;
	len = 14;
	id = ping.id;

	MemoryStream data(data,len);

	data.writeShort(h.port);		// port
	data.writeLong(SWAP4(h.ip));	// ip
	if (ownPong)
	{
		data.writeLong(chanMgr->numConnected());	// cnt
		data.writeLong(servMgr->totalOutput(false));	// total
	}else{
		data.writeLong(0);				// cnt
		data.writeLong(0);				// total
	}


}
// ---------------------------
void GnuPacket::initPush(ChanHit &ch, Host &sh)
{
	func = GNU_FUNC_PUSH;
	ttl = ch.hops;
	hops = 0;
	len = 26;
	id.generate();

	MemoryStream data(data,len);

	// ID of Hit packet
	data.write(ch.packetID.id,16);

	// index of channel
	data.writeLong(ch.index);

	data.writeLong(SWAP4(sh.ip));	// ip
	data.writeShort(sh.port);		// port
}


// ---------------------------
bool GnuPacket::initHit(Host &h, int cnt, Channel **ch, GnuPacket *query, int ver, bool push, bool busy, bool stable, int maxttl)
{
	int i;

	func = GNU_FUNC_HIT;
	hops = 0;
	id.generate();

	ttl = maxttl;


	MemoryStream mem(data,MAX_DATA);

	mem.writeChar(cnt);			// num hits
	mem.writeShort(h.port);		// port
	mem.writeLong(SWAP4(h.ip));	// ip

	if (query)
		mem.writeLong(0);			// speed - route
	else
		mem.writeLong(1);			// broadcast

	
	for(i=0; i<cnt; i++)
	{
		mem.writeLong(ch[i]->index);				// index
		mem.writeShort(ch[i]->getBitrate());			// bitrate
		mem.writeShort(ch[i]->numListeners);			// num listeners

		if ((ch[i]->status == Channel::S_BROADCASTING) || ((sys->rndSeed % chanMgr->fullHitFrequency)==0))
		{
			int rl = strlen(ch[i]->getName());
			mem.write((void *)ch[i]->getName(),rl+1);

			XML xml;
			XML::Node *cn = ch[i]->info.createChannelXML();
			cn->add(ch[i]->createRelayXML(false));
			cn->add(ch[i]->info.createTrackXML());
			cn->add(ch[i]->info.createServentXML());
			xml.setRoot(cn);
			xml.write(mem);

		}else{

			mem.writeChar(0);	// no name

			XML xml;
			XML::Node *cn = ch[i]->info.createRelayChannelXML();
			xml.setRoot(cn);
			xml.writeShort(mem);
		}

		mem.writeChar(0);							// extra null 
	}



	// QHD
	mem.writeLong('PCST');				// vendor ID		
	mem.writeChar(2);					// public sector length 

	int f1 = 0, f2 = 0;

	f1 = 1 | 4 | 8 | 32;	// use push | busy | stable | broadcast  

	if (push) f2 |= 1;			
	if (busy) f2 |= 4;
	if (stable) f2 |= 8;
	if (!query) f2 |= 32;
	
	mem.writeChar(f1);
	mem.writeChar(f2);

	mem.writeChar(0);			// length of private sector: peercast specific

	// queryID/not used
	if (query)
		mem.write(query->id.id,16);					
	else
		mem.write(id.id,16);					

	len = mem.pos;

	if (len >= MAX_DATA)
		return false;

	servMgr->addReplyID(id);
	return true;
}


// ---------------------------
void GnuPacket::initFind(const char *str, XML *xml)
{

	func = GNU_FUNC_QUERY;
	ttl = 7;
	hops = 0;
	id.generate();

	MemoryStream mem(data,MAX_DATA);

	mem.writeShort(0);		// min speed

	if (str)
	{
		int slen = strlen(str);
		mem.write((void *)str,slen+1);	// string
	}else
		mem.writeChar(0);		// null string

	
	if (xml)
		xml->writeShort(mem);

	len = mem.pos;
}

// ---------------------------
void GnuStream::ping(int ttl)
{
	GnuPacket ping;
	ping.initPing(ttl);
	servMgr->addReplyID(ping.id);
	sendPacket(ping);
	LOG_NETWORK("ping out %02x%02x%02x%02x",ping.id.id[0],ping.id.id[1],ping.id.id[2],ping.id.id[3]);
}

// ---------------------------
void GnuStream::sendPacket(GnuPacket &p)
{
	try 
	{
		lock.on();
		packetsOut++;
		stats.add(Stats::NUMPACKETSOUT);

		switch(p.func)
		{
			case GNU_FUNC_PING: stats.add(Stats::NUMPINGOUT); break;
			case GNU_FUNC_PONG: stats.add(Stats::NUMPONGOUT); break;
			case GNU_FUNC_QUERY: stats.add(Stats::NUMQUERYOUT); break;
			case GNU_FUNC_HIT: stats.add(Stats::NUMHITOUT); break;
			case GNU_FUNC_PUSH: stats.add(Stats::NUMPUSHOUT); break;
			default: stats.add(Stats::NUMOTHEROUT); break;
		}


		write(p.id.id,16);
		writeChar(p.func);	// ping func
		writeChar(p.ttl);	// ttl
		writeChar(p.hops);	// hops
		writeLong(p.len);	// len

		if (p.len)
			write(p.data,p.len);

		stats.add(Stats::PACKETDATAOUT,23+p.len);

		lock.off();
	}catch(StreamException &e)
	{		
		lock.off();
		throw e;
	}
}
// ---------------------------
bool GnuStream::readPacket(GnuPacket &p)
{
	try 
	{
		lock.on();
		packetsIn++;
		stats.add(Stats::NUMPACKETSIN);

		read(p.id.id,16);
		p.func = readChar();
		p.ttl = readChar();
		p.hops = readChar();
		p.len = readLong();


		if ((p.hops >= 1) && (p.hops <= 10))
			stats.add((Stats::STAT)((int)Stats::NUMHOPS1+p.hops));

		stats.add(Stats::PACKETDATAIN,23+p.len);

		switch(p.func)
		{
			case GNU_FUNC_PING: stats.add(Stats::NUMPINGIN); break;
			case GNU_FUNC_PONG: stats.add(Stats::NUMPONGIN); break;
			case GNU_FUNC_QUERY: stats.add(Stats::NUMQUERYIN); break;
			case GNU_FUNC_HIT: stats.add(Stats::NUMHITIN); break;
			case GNU_FUNC_PUSH: stats.add(Stats::NUMPUSHIN); break;
			default: stats.add(Stats::NUMOTHERIN); break;
		}


		if (p.len)
		{
			if (p.len > GnuPacket::MAX_DATA)
			{
				while (p.len--)
					readChar();
				lock.off();
				return false;
			}
			read(p.data,p.len);
		}

		lock.off();
		return true;
	}catch(StreamException &e)
	{		
		lock.off();
		throw e;
	}
}

// ---------------------------
GnuStream::R_TYPE GnuStream::processPacket(GnuPacket &in, GnuPacket &out, Servent *serv, GnuID &routeID)
{

	R_TYPE ret = R_DISCARD;

	MemoryStream data(in.data,in.len);

	Host remoteHost = serv->getHost();

	in.ttl--;
	in.hops++;

	routeID = in.id;


	int ver = in.getVersion();

//	LOG("packet %d %x",in.func,ver);

	switch(in.func)
	{
		case GNU_FUNC_PING: // ping
			{
				LOG_NETWORK("ping: from %d.%d.%d.%d : %02x%02x%02x%02x",
							remoteHost.ip>>24&0xff,remoteHost.ip>>16&0xff,remoteHost.ip>>8&0xff,remoteHost.ip&0xff,
							in.id.id[0],in.id.id[1],in.id.id[2],in.id.id[3]
					);
				Host sh = servMgr->serverHost;
				if (sh.isValid())
				{

					if ((servMgr->getFirewall() != ServMgr::FW_ON) && servMgr->needConnections())
					{
						LOG_NETWORK("pong out");
						out.initPong(sh,true,in);
					}
				}
				ret = R_BROADCAST;			
			}
			break;
		case GNU_FUNC_PONG: // pong
			{
				{
					int ip,port,cnt,total;
					port = data.readShort();
					ip = data.readLong();
					cnt = data.readLong();
					total = data.readLong();

					ip = SWAP4(ip);


					Host h;
					h.ip = ip;
					h.port = port;

					char sIP[64],rIP[64];
					h.toStr(sIP);
					remoteHost.toStr(rIP);
					
					LOG_NETWORK("pong: %s via %s",sIP,rIP);
				

					ret = R_DISCARD;

					if (h.isValid())
					{

						// accept if this pong is a reply from one of our own pings, otherwise route back
						if (servMgr->isReplyID(in.id))
						{
							servMgr->addHost(h);
							ret = R_ACCEPTED;
						}else
							ret = R_ROUTE;
						
					}

				}
			}
			break;
		case GNU_FUNC_QUERY: // query
			ret = R_BROADCAST;

			{
				Host sh = servMgr->serverHost;
				if (!sh.isValid())
					sh.ip = 127<<24|1;

				char words[256];
				short spd = data.readShort();
				data.readString(words,sizeof(words));
				words[sizeof(words)-1] = 0;

				MemoryStream xm(&data.buf[data.pos],data.len-data.pos);
				xm.buf[xm.len] = 0;

				Channel *hits[16];
				int cnt=0;

				if (strncmp(xm.buf,"<?xml",5)==0)
				{
					XML xml;
					xml.read(xm);
					XML::Node *cn = xml.findNode("channel");
					if (cn)
					{
						ChanInfo info;
						info.init(cn);
						info.status = ChanInfo::S_PLAY;
						cnt = chanMgr->findChannels(info,hits,16);
					}
					LOG_NETWORK("query XML: %s : found %d",xm.buf,cnt);
				}else{
					ChanInfo info;
					info.name.set(words);
					info.genre.set(words);
					info.id.fromStr(words);
					info.status = ChanInfo::S_PLAY;
					cnt = chanMgr->findChannels(info,hits,16);
					LOG_NETWORK("query STR: %s : found %d",words,cnt);
				}

			

				if (cnt)
				{
					bool push = (servMgr->getFirewall()!=ServMgr::FW_OFF);
					bool busy = (servMgr->inFull() && servMgr->outFull()) || servMgr->streamFull();
					bool stable = servMgr->totalStreams>0;
					if (!out.initHit(sh,cnt,hits,&in,ver,push,busy,stable,in.hops))
					{
						out.func = 255;
						LOG_ERROR("query reply overflow");
					}
				}
			}
			break;
		case GNU_FUNC_PUSH:	// push
			{

				GnuID pid;
				data.read(pid.id,16);
				
				//LOG("push serv= %02x%02x%02x%02x",servMgr->id[0],servMgr->id[1],servMgr->id[2],servMgr->id[3]);
				//LOG("pack = %02x%02x%02x%02x",id[0],id[1],id[2],id[3]);


				int index = data.readLong();
				int ip = data.readLong();
				int port = data.readShort();

			
				ip = SWAP4(ip);

				Host h(ip,port);
				char hostName[64];
				h.toStr(hostName);

				if (servMgr->isReplyID(pid))
				{

					Channel *c = chanMgr->findChannelByIndex(index);

					if (!c)
					{
						LOG_NETWORK("push 0x%x to %s: Not found",index,hostName);
					}else
					{
						if (!c->isFull() && !servMgr->streamFull())
						{
							LOG_NETWORK("push: 0x%x to %s: OK",index,hostName);

							Servent *s = servMgr->allocServent();
							if (s)
								s->initGiv(h,index,pid);
						}else
							LOG_NETWORK("push: 0x%x to %s: FULL",index,hostName);
					}
					ret = R_ACCEPTED;
				}else{
					LOG_NETWORK("push: 0x%x to %s: ROUTE",index,hostName);
					routeID = pid;
					ret = R_ROUTE;
				}
		
			}
			break;
		case GNU_FUNC_HIT: // hit
			{
				int i;
				int num = data.readChar();	// hits
				int port = data.readShort();		// port
				int ip = data.readLong();		// ip
				ip = SWAP4(ip);
				int spd = data.readLong();		// speed/broadcast

				Host h(ip,port);
				char hostName[64];

				h.IPtoStr(hostName);

				bool dataValid=true;

				//if (servMgr->isReplyID(in.id) || (spd==1))
				{
					ChanHit *hits[100];
					int numHits=0;

					for(i=0; i<num; i++)
					{
						int index,bitrate,listeners;


						index = data.readLong();		// index
						bitrate = data.readShort();		// bitrate
						listeners = data.readShort();	// listeners

						char fname[128];
						data.readString(fname,sizeof(fname));
						fname[sizeof(fname)-1] = 0;

						ChanHit ch;
						ch.init();
						ch.firewalled = false;		// default to NO as we dont get the info until the next section.
						ch.host = h;
						ch.index = index;
						ch.numListeners = listeners;
						ch.hops = in.hops;
						ch.packetID = in.id;

						ChanInfo info;


						{
							char xmlData[4000];
							int xlen = data.readString(xmlData,sizeof(xmlData));

							if ((strncmp(xmlData,"<?xml",5)==0) && (xlen < GnuPacket::MAX_DATA))
							{
								//LOG_NETWORK("Hit XML: %s",xmlData);

								MemoryStream xm(xmlData,xlen);
								XML xml;
								xml.read(xm);
								XML::Node *n = xml.findNode("channel");
								if (n)
								{
									info.init(n);
									ch.upTime = n->findAttrInt("uptime");
									ch.numSkips = n->findAttrInt("skips");
	
									XML::Node *sn = xml.findNode("servent");
									if (sn)
									{
										char *ag = sn->findAttr("agent");
										if (ag)
										{
											strncpy(ch.agentStr,ag,16);
											ch.agentStr[15]=0;
										}
									}
								}else
									LOG_NETWORK("Missing Channel node for %s",fname);
							}else
							{
								LOG_NETWORK("Missing XML data for %s",fname);
								//LOG_NETWORK("%s",xmlData);
								dataValid = false;
							}
						}

						if (info.id.isSet())
						{
							ChanHit *chp = chanMgr->addHit(info,ch);		

							if ((chp) && (numHits<100))
								hits[numHits++] = chp;
						}

					}


					int vendor = data.readLong();	// vendor ID

					int pubLen = data.readChar();	// public sec length - should be 2

					int f1 = data.readChar() & 0xff; // flags 1
					int f2 = data.readChar() & 0xff; // flags 2

					pubLen -= 2;
					while (pubLen-->0)
						data.readChar();


					int privLen = data.readChar();	// private sec length

					while (privLen-->0)
						data.readChar();

					// not used anymore
					GnuID queryID;
					data.read(queryID.id,16);

					bool isBroadcastHit=false;
					if (f1 & 32)
						isBroadcastHit = (f2 & 32)!=0;

					for(i=0; i<numHits; i++)
					{
						if (f1 & 1)
							hits[i]->firewalled = (f2 & 1)!=0;

						if (f1 & 4)
							hits[i]->busy = (f2 & 4)!=0;

						if (f1 & 8)
							hits[i]->stable = (f2 & 8)!=0;
					}
					char flstr[64];
					flstr[0]=0;
					if (f2 & 1) strcat(flstr,"Push,");
					if (f2 & 4) strcat(flstr,"Busy,");
					if (f2 & 8) strcat(flstr,"Stable,");
					if (f2 & 32) strcat(flstr,"Broadcast,");



	
					if (dataValid)
					{
						if ((spd == 0) && (!isBroadcastHit))
						{
							if (servMgr->isReplyID(queryID))
							{
								ret = R_ACCEPTED;
								LOG_NETWORK("self-hit: %s 0x%02x %s %d chan",hostName,f2,flstr,num);
							}else
							{
								routeID = queryID;
								ret = R_ROUTE;
								LOG_NETWORK("route-hit: %s 0x%02x %s %d chan",hostName,f2,flstr,num);
							}
						}else
						{
							ret = R_BROADCAST;
							LOG_NETWORK("broadcast-hit: %s 0x%02x %s %d chan",hostName,f2,flstr,num);
						}
					}else{
						ret = R_DISCARD;
					}

				}
			}
			break;
		default:
			LOG_NETWORK("packet: %d",in.func);
			break;
	}

	
	if ((in.ttl > 10) || (in.hops > 10) || (in.ttl==0))
		if ((ret == R_BROADCAST) || (ret == R_ROUTE))
			ret = R_DEAD;

	return ret;
}