servent.cpp

这是和p2p相关的一份源码

// ------------------------------------------------
// File : servent.cpp
// Date: 4-apr-2002
// Author: giles
// Desc: 
//		Servents are the actual connections between clients. They do the handshaking,
//		transfering of data and processing of GnuPackets. Each servent has one socket allocated
//		to it on connect, it uses this to transfer all of its data.
//
// (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 <stdlib.h>
#include "servent.h"
#include "sys.h"
#include "gnutella.h"
#include "xml.h"
#include "html.h"
#include "http.h"
#include "stats.h"
#include "servmgr.h"
#include "peercast.h"



// -----------------------------------
char *Servent::statusMsgs[]=
{
        "NONE",
		"CONNECTING",
        "PROTOCOL",
        "HANDSHAKE",
        "CONNECTED",
        "CLOSING",
		"LISTENING",
		"TIMEOUT",
		"REFUSED",
		"VERIFIED",
		"ERROR",
		"WAIT"
};

// -----------------------------------
char *Servent::typeMsgs[]=
{
		"NONE",
		"ALLOCATED",
        "OUT",
        "IN",
        "SERVER",
		"CHECK",
		"STREAM",
        "LOOKUP"
};

// -----------------------------------
bool	Servent::isPrivate() 
{
	Host h = getHost();
	return servMgr->isFiltered(ServFilter::F_PRIVATE,h);
}
// -----------------------------------
bool	Servent::isAllowed(int a) 
{
	Host h = getHost();

	if (!h.isValid())
		return false;

	if (servMgr->isFiltered(ServFilter::F_BAN,h))
		return false;

	if (!servMgr->isFiltered(ServFilter::F_NETWORK,h))
		return false;

	return (allow&a)!=0;
}

// -----------------------------------
void Servent::init()
{
	outPacketsPri.init(MAX_OUTPACKETS);
	outPacketsNorm.init(MAX_OUTPACKETS);

	seenIDs.init(MAX_HASH);
	permAlloc = false;
	sock = NULL;
	reset();
}
// -----------------------------------
void Servent::reset(bool doClose)
{

	if (doClose)
		close();


	networkID.clear();

	agent.clear();
	sock = NULL;
	outputBitrate = 0;
	chanID.clear();
	chanIndex = -1;
	allow = ALLOW_ALL;
	currPos = 0;
	addMetadata = false;
	pack.func = 255;
	lastConnect = lastPing = lastPacket = 0;
	loginPassword[0] = 0;
	loginMount[0] = 0;
	bytesPerSecond = 0;
	priorityConnect = false;

	outPacketsNorm.reset();
	outPacketsPri.reset();

	seenIDs.clearAll();

	status = S_NONE;
	if (!permAlloc)
		type = T_NONE;
}
// -----------------------------------
Host Servent::getHost()
{
	Host h(0,0);

	if (sock)
		h = sock->host;

	return h;
}

// -----------------------------------
bool Servent::outputPacket(GnuPacket &p, bool pri)
{
	lock.on();

	bool r;
	if (pri)
		r = outPacketsPri.write(p);
	else
		r = outPacketsNorm.write(p);

	lock.off();
	return r;
}

// -----------------------------------
bool Servent::initServer(Host &h)
{
	try
	{
		checkFree();

		status = S_WAIT;

		createSocket();

		sock->bind(h);

		thread.data = this;

		thread.func = serverProc;

		type = T_SERVER;

		if (!sys->startThread(&thread))
			throw StreamException("Can`t start thread");

	}catch(StreamException &e)
	{
		LOG_ERROR("Bad server: %s",e.msg);
		reset();
		return false;
	}

	return true;
}
// -----------------------------------
void Servent::checkFree()
{
	if (sock)
		throw StreamException("Socket already set");
	if (thread.active)
		throw StreamException("Thread already active");
}
// -----------------------------------
void Servent::initIncoming(ClientSocket *s, unsigned int a)
{

	try{

		checkFree();

		type = T_INCOMING;
		sock = s;
		allow = a;
		thread.data = this;
		thread.func = incomingProc;

		setStatus(S_PROTOCOL);

		if (!sys->startThread(&thread))
		{
			reset();
			LOG_ERROR("Unable to start incoming thread");
		}
	}catch(StreamException &e)
	{
		LOG_ERROR("Incoming error: %s",e.msg);
		reset();
	}
}

// -----------------------------------
void Servent::initOutgoing(Host &rh, TYPE ty)
{
	char ipStr[64];
	rh.toStr(ipStr);
	try 
	{
		checkFree();

	    createSocket();

		type = ty;

		sock->timeout = 10000;	// 10 seconds to connect 
		sock->open(rh);

		if (!isAllowed(ALLOW_SERVENT))
			throw StreamException("Servent not allowed");

		thread.data = this;
		thread.func = outgoingProc;

		LOG_NETWORK("Starting outgoing to %s",ipStr);

		if (!sys->startThread(&thread))
			throw StreamException("Can`t start thread");

	}catch(StreamException &e)
	{
		LOG_ERROR("Unable to open connection to %s - %s",ipStr,e.msg);
		reset();
	}
}
// -----------------------------------
void Servent::initGiv(Host &h, int idx, GnuID &id)
{
	char ipStr[64];
	h.toStr(ipStr);
	try 
	{
		checkFree();


		Channel *ch = chanMgr->findChannelByIndex(idx);
		if (!ch)
			throw StreamException("No channel");

		chanID = ch->info.id;
		chanIndex = idx;
		pushID	= id;

	    createSocket();

		sock->open(h);

		if (!isAllowed(ALLOW_DATA))
			throw StreamException("Servent not allowed");
		
		sock->connect();

		thread.data = this;
		thread.func = givProc;

		type = T_STREAM;

		if (!sys->startThread(&thread))
			throw StreamException("Can`t start thread");

	}catch(StreamException &e)
	{
		LOG_ERROR("Giv error: 0x%x to %s: %s",idx,ipStr,e.msg);
		reset();
	}
}
// -----------------------------------
void Servent::createSocket()
{
	if (sock)
		LOG_ERROR("Servent::createSocket attempt made while active");

	sock = sys->createSocket();
}
// -----------------------------------
void Servent::endThread()
{	
	reset(true);
	thread.unlock();
}
// -----------------------------------
void Servent::abort()
{
	thread.active = false;
	if (sock)
		sock->close();
}
// -----------------------------------
void Servent::close()
{	
	thread.active = false;
		
	setStatus(S_CLOSING);
	if (sock)
	{
		sock->close();
		delete sock;
		sock = NULL;
	}
}

// -----------------------------------
void Servent::setStatus(STATUS s)
{
	if (s != status)
	{
		status = s;

		if ((s == S_HANDSHAKE) || (s == S_CONNECTED) || (s == S_LISTENING))
			lastConnect = sys->getTime();
	}

}
// -----------------------------------
void Servent::handshakeOut()
{
	sock->timeout = 10000;

	if (servMgr->allowGnutella)
	    sock->writeLine(GNU_CONNECT);
	else
	    sock->writeLine(GNU_PEERCONN);

	char str[64];
    
	sock->writeLine("%s %s",HTTP_HS_AGENT,PCX_AGENT);
    sock->writeLine("%s %s",PCX_HS_SUBNET,servMgr->network.cstr());
	if (priorityConnect)
	    sock->writeLine("%s %d",PCX_HS_PRIORITY,1);
	
	if (networkID.isSet())
	{
		networkID.toStr(str);
		sock->writeLine("%s %s",PCX_HS_NETWORKID,str);
	}

	servMgr->sessionID.toStr(str);
	sock->writeLine("%s %s",PCX_HS_ID,str);

	
    sock->writeLine("%s %s",PCX_HS_OS,peercastApp->getClientTypeOS());
	
	sock->writeLine("");

	HTTP http(*sock);

	http.checkResponse(200);

	bool subnetValid = false;
	bool versionValid = false;

    while (http.nextHeader())
    {
		char *arg = http.getArgStr();
		if (!arg)
			continue;

		if (http.isHeader(PCX_HS_SUBNET))
			subnetValid = stricmp(arg,servMgr->network.cstr())==0;
		else if (http.isHeader(HTTP_HS_AGENT))
		{
			agent.set(arg);

			if (strnicmp(arg,"PeerCast/",9)==0)
				versionValid = (stricmp(arg+9,MIN_CONNECTVER)>=0) && (stricmp(arg+9,MAX_CONNECTVER)<=0);
		}


		if (type == T_LOOKUP)
		{
			if (http.isHeader("remote-ip"))
			{
				if (!servMgr->forceIP[0])
				{
					Host nh;
					nh.fromStr(arg,DEFAULT_PORT);
					if (nh.ip != servMgr->serverHost.ip)
					{
						LOG_DEBUG("Got new IP: %s",arg);
						servMgr->serverHost.ip = nh.ip;
					}
				}
			}else if (http.isHeader(PCX_HS_MSG))
			{
				if (strcmp(servMgr->rootMsg.cstr(),arg))
				{
					servMgr->rootMsg.set(arg);
					peercastApp->notifyMessage(ServMgr::NT_PEERCAST,arg);
				}

			}else if (http.isHeader(PCX_HS_DL))
			{
				if (servMgr->downloadURL[0]==0)
				{
					strcpy(servMgr->downloadURL,arg);
					peercastApp->notifyMessage(ServMgr::NT_UPGRADE,"There is a new version available, please click here to upgrade your client.");
				}
			}
			else if (http.isHeader(PCX_HS_BCTTL))
				chanMgr->broadcastTTL = atoi(arg);
		}

    }

	if ((stricmp(servMgr->network.cstr(),"peercast")!=0) && (!subnetValid))
		throw HTTPException(HTTP_SC_NOTFOUND,404);

	if (!versionValid)
		throw HTTPException(HTTP_SC_UNAUTHORIZED,401);


    sock->writeLine(GNU_OK);
    sock->writeLine("");
}
// -----------------------------------
void Servent::handshakeIn()
{
	sock->timeout = 10000;


	int osType=0;

	HTTP http(*sock);


	bool subnetValid = false;
	bool versionValid = false;
	bool diffRootVer = false;

    while (http.nextHeader())
    {
		//LOG_DEBUG(http.cmdLine);

		char *arg = http.getArgStr();
		if (!arg)
			continue;

		if (http.isHeader(PCX_HS_SUBNET))
			subnetValid = stricmp(arg,servMgr->network)==0;
		else if (http.isHeader(HTTP_HS_AGENT))
		{
			agent.set(arg);

			if (strnicmp(arg,"PeerCast/",9)==0)
			{
				versionValid = (stricmp(arg+9,MIN_CONNECTVER)>=0) && (stricmp(arg+9,MAX_CONNECTVER)<=0);
				diffRootVer = stricmp(arg+9,MIN_ROOTVER)<0;
			}
		}else if (http.isHeader(PCX_HS_NETWORKID))
		{
			networkID.fromStr(arg);

		}else if (http.isHeader(PCX_HS_PRIORITY))
		{
			priorityConnect = atoi(arg)!=0;

		}else if (http.isHeader(PCX_HS_ID))
		{
			GnuID id;
			id.fromStr(arg);
			if (id.isSame(servMgr->sessionID))
				throw StreamException("Servent loopback");

		}else if (http.isHeader(PCX_HS_OS))
		{
			if (stricmp(arg,PCX_OS_LINUXDYN)==0)
				osType = 1;
			else if (stricmp(arg,PCX_OS_LINUXSTA)==0)
				osType = 2;
			else if (stricmp(arg,PCX_OS_WIN32)==0)
				osType = 3;
			else if (stricmp(arg,PCX_OS_MACOSX)==0)
				osType = 4;
			else if (stricmp(arg,PCX_OS_WINAMP2)==0)
				osType = 5;