pxyscand/src/PXServer.cc

// Copyright (C) 2003, 2004 Stephane Thiell
//
// This file is part of pxyscand (from pxys)
//
// 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.
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
//
#define RCSID "$Id: PXServer.cc,v 1.5 2004/01/10 14:32:29 mbuna Exp $"

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include "PXServer.h"
#include "PXConfig.h"
#include "PXCore.h"
#include "PXScan.h"
#include "PXScanManager.h"
#include "PXSession.h"
#include <cassert>
#include <cstring>
#include <iostream>

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

using std::clog;
using std::endl;

#define CACHE_AUTOSAVE_DELAY 900

#define NOPROXY_CACHE_FILE "noproxy"
#define PROXY_CACHE_FILE   "proxy"

const uint32_t kProxyCacheSize = 1009; /* a prime number please */

PXServer::PXServer(PXConfig *inConfig, PXScanManager *inScanManager,
                   PXCore *inCore)
  : PXRepeater(CACHE_AUTOSAVE_DELAY),
    mConfig(inConfig), mScanManager(inScanManager), mCore(inCore),
    mSessionCount(0), mSessionRejectedCount(0), mQueryCount(0),
    mScannedCount(0), mProxyCount(0), mNoScanHits(0), mIPCacheHits(0),
    mPXCacheHits(0), mInProgressCount(0)
  {
  mCacheMutex = peak_task_mutex_create(peak_task_self());
  mScanMutex = peak_task_mutex_create(peak_task_self());
  
  // Initialize IPv4 cache (non-proxy)
  mIP4Cache = new PXIPCacheT<in_addr>(
    mConfig->cache.dir,
    htonl(0x50584346), /* 'PXCF' */
    mConfig->cache.maxips,
    mConfig->cache.expire,
    0 // no associated data
    );
  if (mIP4Cache->SwapIn(NOPROXY_CACHE_FILE))
    clog << "PXServer::PXServer: IP4Cache SwapIn failed" << endl;
  
  // Initialize IPv4 proxy cache
  mPX4Cache = new PXIPCacheT<in_addr>(
    mConfig->cache.dir,
    htonl(0x50585046), /* 'PXPF' */
    kProxyCacheSize,
    mConfig->cache.proxy_expire,
    sizeof(PXProxyBinInfo)
    );
  if (mPX4Cache->SwapIn(PROXY_CACHE_FILE))
    clog << "PXServer::PXServer: PX4Cache SwapIn failed" << endl;
  }

PXServer::~PXServer()
  {
  delete mPX4Cache;
  delete mIP4Cache;
  peak_release(mScanMutex);
  peak_release(mCacheMutex);
  }

int
PXServer::Start()
  {
  struct sockaddr_in sin;
  
  memset(&sin, 0, sizeof(sin));
  sin.sin_family = AF_INET;
  sin.sin_addr = mConfig->opas.local_address;
  sin.sin_port = htons(mConfig->opas.port);
  
  mLStream = peak_stream_socket_create((struct sockaddr *)&sin, sizeof(sin),
                                       PEAK_STREAM_OPT_LISTEN
                                       |PEAK_STREAM_OPT_AUTOSCHEDULE,
                                       EventCallback,
                                       this);
  
  if (!mLStream)
    {
    clog << "PXServer::Start: peak_stream_socket_create failed" << endl;
    return -1;
    }
  return 0;
  }

void
PXServer::Rehash()
  {
  mIP4Cache->SwapOut(NOPROXY_CACHE_FILE);
  mPX4Cache->SwapOut(PROXY_CACHE_FILE);
  }

void
PXServer::SessionClosed(PXSession *inSession)
  {
  mSessions.remove(inSession);
  }

void
PXServer::CreateStatus(PXSStatus **outStatus, size_t *outSize) const
  {
  size_t n = mScanManager->GetModuleInfoCount();
  size_t size = sizeof(PXSStatus) + n * sizeof(PXSMInfo);
  
  PXSStatus *statusPtr;
  
  *outStatus = statusPtr = (PXSStatus *)peak_allocate(size);
  *outSize = size;
  memset(statusPtr, 0, size);
  
  statusPtr->head.sig = htonl(PXYSCAND_SIG);
  statusPtr->head.ver = htonl(PX_VERSION);
  statusPtr->head.cmd = htonl(PX_CMD_STATUS);
  
  statusPtr->curIPScan = htonl(mScanManager->GetIPScanCount());
  statusPtr->maxIPScan = htonl(mScanManager->GetIPScanMax());
  statusPtr->curRunScan = htonl(mScanManager->GetRunScanCount());
  statusPtr->maxRunScan = htonl(mScanManager->GetRunScanMax());
  statusPtr->scanQSize = htonl(mScanManager->GetScanQSize());
  
  statusPtr->targetAddr = mScanManager->GetTargetAddress();
  statusPtr->targetPort = htonl((uint32_t)mScanManager->GetTargetPort());
  
  statusPtr->targetLastCheck =
    htonl(peak_time() - mScanManager->GetLastTargetCheck());
  
  statusPtr->servScannedCount = htonl(mScannedCount);
  statusPtr->servProxyCount = htonl(mProxyCount);
  
  statusPtr->uptime = htonl(peak_time() - mCore->GetBirthTime());
  statusPtr->timeout = htonl(mConfig->scanner.timeout);
  
  statusPtr->numMInfo = htonl(n);
  
  for (size_t i = 0; i < n; i++)
    {
    PXSMInfo *info = (reinterpret_cast<PXSMInfo *>(statusPtr + 1) + i);
    const PXModuleInfo *minfo = mScanManager->GetMInfo(i);
    info->port = htonl(minfo->port);
    info->connCount = htonl(*minfo->connCountPtr);
    info->proxyCount = htonl(*minfo->proxyCountPtr);
    snprintf(info->shortName, sizeof(info->shortName), "%s", minfo->shortName);
    }
  }

void
PXServer::GetStats(PXSStats &oStats)
  {
  oStats.servQueryCount = htonl(mQueryCount);
  oStats.servScannedCount = htonl(mScannedCount);
  oStats.servProxyCount = htonl(mProxyCount);
  oStats.servNoScanHits = htonl(mNoScanHits);
  oStats.servIPCacheHits = htonl(mIPCacheHits);
  oStats.servPXCacheHits = htonl(mPXCacheHits);
  oStats.servInProgressCount = htonl(mInProgressCount);
  
  oStats.servIP4CacheCount = htonl(mIP4Cache->GetCount());
  oStats.servIP4CacheSize = htonl(mConfig->cache.maxips);
  oStats.servIP4CacheExpire = htonl(mConfig->cache.expire);
  oStats.servPX4CacheCount = htonl(mPX4Cache->GetCount());
  oStats.servPX4CacheSize = htonl(kProxyCacheSize);
  oStats.servPX4CacheExpire = htonl(mConfig->cache.proxy_expire);
  
  oStats.sessCurrent = htonl(mSessions.size());
  oStats.sessCount = htonl(mSessionCount);
  oStats.sessRejectedCount = htonl(mSessionRejectedCount);
  
  snprintf(oStats.version, sizeof(oStats.version), "pxyscand %s "
           "(service msg v%d)", VERSION, PX_VERSION);
  }

//  Query           Complete
// 
// Solution with 1 mutex
// 
// lock A         lock A
//  CHK CACHES     PUT IN CACHE
//  CHK SCANS      REMOVE SCAN
// unlock A       unlock A
// 
// Solution with 2 mutex [choosen]
// 
// lock A         lock A
//  CHK CACHES     PUT IN CACHE
// lock B         unlock A
// unlock A       lock B
//  CHK SCANS      REMOVE SCAN
// unlock B       unlock B


void
PXServer::NewScanIP4(const opas_msg_query &inQuery, PXSession *inSession)
  {
  PXProxyBinInfo *infop;
  time_t t;
  
  mQueryCount++;
  
  if (mConfig->CheckNoScan(inQuery.addr))
    {
    mNoScanHits++;
    inSession->ScanResultNoProxy(inQuery, true);
    return;
    }
  
  peak_task_mutex_lock(mCacheMutex);   // lock the cache
  
  if ((t = mPX4Cache->Get(inQuery.addr, (void**)&infop)) > 0)
    {
    // OMG!
    mPXCacheHits++;
    
    // Update cache hits count, just for the fun.
    infop->cache_hits = htonl(ntohl(infop->cache_hits) + 1);
    
    peak_task_mutex_unlock(mCacheMutex);
    
    inSession->ScanResultProxy(inQuery, true, t,
                               ntohs(infop->type), ntohs(infop->port),
                               infop->descr);
    return;
    }
  
  if ((t = mIP4Cache->Get(inQuery.addr, NULL)) > 0)
    {
    // In cache? If so it's quick!
    mIPCacheHits++;
    
    peak_task_mutex_unlock(mCacheMutex);
    
    inSession->ScanResultNoProxy(inQuery, true);
    return;
    }
  
  peak_task_mutex_lock(mScanMutex);    // lock first (important)
  peak_task_mutex_unlock(mCacheMutex); // ok, others can consult the cache
  
  // Search for existing scan session for this IP...
  PXScan *scan;
  bool newScan;
  scanMapType::const_iterator i = mScans.find(inQuery.addr);
  
  if (i == mScans.end())
    {
    // Normal case, I suppose. Create new "scan" instance and remember it.
    scan = new PXScan(inQuery, inSession, mScanManager);
    mScans.insert(scanMapPairType(inQuery.addr, scan));
    newScan = true;
    }
  else
    {
    // Found... do not scan twice.
    mInProgressCount++;
    scan = i->second;
    if (!scan->IsSessionReferenced(inSession))
      scan->AppendQuery(inQuery, inSession);
    newScan = false;
    }
  
  peak_task_mutex_unlock(mScanMutex);  // "secure!"
  
  if (newScan)
    scan->Schedule();
  }

void
PXServer::DeleteScanIP4(const opas_msg_query &inQuery)
  {
  peak_task_mutex_lock(mScanMutex);
  
  mScannedCount++;
  
  scanMapType::iterator i = mScans.find(inQuery.addr);
  assert(i != mScans.end());
  mScans.erase(i);
  
  peak_task_mutex_unlock(mScanMutex);
  }

void
PXServer::ScanIP4ResultNoProxy(const opas_msg_query &inQuery)
  {
  peak_task_mutex_lock(mCacheMutex);
  mIP4Cache->Put(inQuery.addr, peak_time(), NULL);
  peak_task_mutex_unlock(mCacheMutex); // safe
  
  this->DeleteScanIP4(inQuery);
  }

void
PXServer::ScanIP4ResultProxy(const opas_msg_query &inQuery,
                             uint16_t inTypeOfProxy, uint16_t inPort,
                             const char *inDescr)
  {
  PXProxyBinInfo info;
  memset(&info, 0, sizeof(info));
  info.cache_hits = htonl(0);
  info.type = htons(inTypeOfProxy);
  info.port = htons(inPort);
  strncpy(info.descr, inDescr, sizeof(info.descr) - 1);
  
  peak_task_mutex_lock(mCacheMutex);
  mPX4Cache->Put(inQuery.addr, peak_time(), &info);
  peak_task_mutex_unlock(mCacheMutex);
  
  mProxyCount++;
  
  this->DeleteScanIP4(inQuery);
  }

bool
PXServer::RemoveAddressFromCache(const in_addr &inAddress)
  {
  peak_task_mutex_lock(mCacheMutex);
  bool res = mPX4Cache->Invalidate(inAddress);
  mIP4Cache->Invalidate(inAddress); // Better safe than sorry.
  peak_task_mutex_unlock(mCacheMutex);
  return res;
  }

bool
PXServer::RemoveAddressFromCache(const in6_addr &inAddress)
  {
  bool res = false;
#if 0
  peak_task_mutex_lock(mCacheMutex);
  res = mPX6Cache->Invalidate(inAddress);
  mIP6Cache->Invalidate(inAddress);
  peak_task_mutex_unlock(mCacheMutex);
#endif
  return res;
  }

void
PXServer::ProcessEvent(peak_stream s, int type)
  {
  if (type == PEAK_STREAM_EVT_ACCEPT)
    {
    this->NewConnection(s);
    }
  else
    {
    assert(0);
    }
  }

void
PXServer::NewConnection(peak_stream s)
  {
  peak_stream ns;
  struct sockaddr_in sin;
  int len = sizeof(sin);
  
  if (!(ns = peak_stream_accept_create(s, PEAK_STREAM_OPT_DEFAULT,
                                       PXSession::EventCallback, this)))
    return;

  if (mSessions.size() >= MAX_SESSIONS)
    {
    peak_release(ns);
    mSessionRejectedCount++;
    return;
    }
  
  // Limit DoS attacks... check the peer ASAP (it's already too late!)
  peak_stream_get_address(ns, (struct sockaddr *)&sin, &len);
  
  if (!mConfig->OPASIsAllowed(sin.sin_addr))
    {
    peak_release(ns); // Reject
    mSessionRejectedCount++;
    return;
    }
  
  mSessions.push_back(new PXSession(this, ns));
  mSessionCount++;
  }

void
PXServer::RepeaterFire()
  {
  peak_task_mutex_lock(mCacheMutex);
  this->Rehash();
  peak_task_mutex_unlock(mCacheMutex);
  }

void
PXServer::EventCallback(peak_stream s, int type, void *context)
  {
  ((PXServer *)context)->ProcessEvent(s, type);
  }
Revision:	3252
Committed:	Wed Apr 2 20:41:43 2014 UTC (10 years ago) by michael
Content type:	text/x-c++src
File size:	11402 byte(s)
Log Message:	- Imported pxys2-2.0.0
#	Content
1	// Copyright (C) 2003, 2004 Stephane Thiell
2	//
3	// This file is part of pxyscand (from pxys)
4	//
5	// This program is free software; you can redistribute it and/or
6	// modify it under the terms of the GNU General Public License
7	// as published by the Free Software Foundation; either version 2
8	// of the License, or (at your option) any later version.
9	//
10	// This program is distributed in the hope that it will be useful,
11	// but WITHOUT ANY WARRANTY; without even the implied warranty of
12	// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13	// GNU General Public License for more details.
14	//
15	// You should have received a copy of the GNU General Public License
16	// along with this program; if not, write to the Free Software
17	// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
18	//
19	#define RCSID "$Id: PXServer.cc,v 1.5 2004/01/10 14:32:29 mbuna Exp $"
20
21	#ifdef HAVE_CONFIG_H
22	#include "config.h"
23	#endif
24
25	#include "PXServer.h"
26	#include "PXConfig.h"
27	#include "PXCore.h"
28	#include "PXScan.h"
29	#include "PXScanManager.h"
30	#include "PXSession.h"
31	#include <cassert>
32	#include <cstring>
33	#include <iostream>
34
35	#include <sys/types.h>
36	#include <sys/socket.h>
37	#include <netinet/in.h>
38	#include <arpa/inet.h>
39
40	using std::clog;
41	using std::endl;
42
43	#define CACHE_AUTOSAVE_DELAY 900
44
45	#define NOPROXY_CACHE_FILE "noproxy"
46	#define PROXY_CACHE_FILE "proxy"
47
48	const uint32_t kProxyCacheSize = 1009; /* a prime number please */
49
50	PXServer::PXServer(PXConfig inConfig, PXScanManager inScanManager,
51	PXCore *inCore)
52	: PXRepeater(CACHE_AUTOSAVE_DELAY),
53	mConfig(inConfig), mScanManager(inScanManager), mCore(inCore),
54	mSessionCount(0), mSessionRejectedCount(0), mQueryCount(0),
55	mScannedCount(0), mProxyCount(0), mNoScanHits(0), mIPCacheHits(0),
56	mPXCacheHits(0), mInProgressCount(0)
57	{
58	mCacheMutex = peak_task_mutex_create(peak_task_self());
59	mScanMutex = peak_task_mutex_create(peak_task_self());
60
61	// Initialize IPv4 cache (non-proxy)
62	mIP4Cache = new PXIPCacheT<in_addr>(
63	mConfig->cache.dir,
64	htonl(0x50584346), /* 'PXCF' */
65	mConfig->cache.maxips,
66	mConfig->cache.expire,
67	0 // no associated data
68	);
69	if (mIP4Cache->SwapIn(NOPROXY_CACHE_FILE))
70	clog << "PXServer::PXServer: IP4Cache SwapIn failed" << endl;
71
72	// Initialize IPv4 proxy cache
73	mPX4Cache = new PXIPCacheT<in_addr>(
74	mConfig->cache.dir,
75	htonl(0x50585046), /* 'PXPF' */
76	kProxyCacheSize,
77	mConfig->cache.proxy_expire,
78	sizeof(PXProxyBinInfo)
79	);
80	if (mPX4Cache->SwapIn(PROXY_CACHE_FILE))
81	clog << "PXServer::PXServer: PX4Cache SwapIn failed" << endl;
82	}
83
84	PXServer::~PXServer()
85	{
86	delete mPX4Cache;
87	delete mIP4Cache;
88	peak_release(mScanMutex);
89	peak_release(mCacheMutex);
90	}
91
92	int
93	PXServer::Start()
94	{
95	struct sockaddr_in sin;
96
97	memset(&sin, 0, sizeof(sin));
98	sin.sin_family = AF_INET;
99	sin.sin_addr = mConfig->opas.local_address;
100	sin.sin_port = htons(mConfig->opas.port);
101
102	mLStream = peak_stream_socket_create((struct sockaddr *)&sin, sizeof(sin),
103	PEAK_STREAM_OPT_LISTEN
104	\|PEAK_STREAM_OPT_AUTOSCHEDULE,
105	EventCallback,
106	this);
107
108	if (!mLStream)
109	{
110	clog << "PXServer::Start: peak_stream_socket_create failed" << endl;
111	return -1;
112	}
113	return 0;
114	}
115
116	void
117	PXServer::Rehash()
118	{
119	mIP4Cache->SwapOut(NOPROXY_CACHE_FILE);
120	mPX4Cache->SwapOut(PROXY_CACHE_FILE);
121	}
122
123	void
124	PXServer::SessionClosed(PXSession *inSession)
125	{
126	mSessions.remove(inSession);
127	}
128
129	void
130	PXServer::CreateStatus(PXSStatus *outStatus, size_t outSize) const
131	{
132	size_t n = mScanManager->GetModuleInfoCount();
133	size_t size = sizeof(PXSStatus) + n * sizeof(PXSMInfo);
134
135	PXSStatus *statusPtr;
136
137	outStatus = statusPtr = (PXSStatus )peak_allocate(size);
138	*outSize = size;
139	memset(statusPtr, 0, size);
140
141	statusPtr->head.sig = htonl(PXYSCAND_SIG);
142	statusPtr->head.ver = htonl(PX_VERSION);
143	statusPtr->head.cmd = htonl(PX_CMD_STATUS);
144
145	statusPtr->curIPScan = htonl(mScanManager->GetIPScanCount());
146	statusPtr->maxIPScan = htonl(mScanManager->GetIPScanMax());
147	statusPtr->curRunScan = htonl(mScanManager->GetRunScanCount());
148	statusPtr->maxRunScan = htonl(mScanManager->GetRunScanMax());
149	statusPtr->scanQSize = htonl(mScanManager->GetScanQSize());
150
151	statusPtr->targetAddr = mScanManager->GetTargetAddress();
152	statusPtr->targetPort = htonl((uint32_t)mScanManager->GetTargetPort());
153
154	statusPtr->targetLastCheck =
155	htonl(peak_time() - mScanManager->GetLastTargetCheck());
156
157	statusPtr->servScannedCount = htonl(mScannedCount);
158	statusPtr->servProxyCount = htonl(mProxyCount);
159
160	statusPtr->uptime = htonl(peak_time() - mCore->GetBirthTime());
161	statusPtr->timeout = htonl(mConfig->scanner.timeout);
162
163	statusPtr->numMInfo = htonl(n);
164
165	for (size_t i = 0; i < n; i++)
166	{
167	PXSMInfo info = (reinterpret_cast<PXSMInfo >(statusPtr + 1) + i);
168	const PXModuleInfo *minfo = mScanManager->GetMInfo(i);
169	info->port = htonl(minfo->port);
170	info->connCount = htonl(*minfo->connCountPtr);
171	info->proxyCount = htonl(*minfo->proxyCountPtr);
172	snprintf(info->shortName, sizeof(info->shortName), "%s", minfo->shortName);
173	}
174	}
175
176	void
177	PXServer::GetStats(PXSStats &oStats)
178	{
179	oStats.servQueryCount = htonl(mQueryCount);
180	oStats.servScannedCount = htonl(mScannedCount);
181	oStats.servProxyCount = htonl(mProxyCount);
182	oStats.servNoScanHits = htonl(mNoScanHits);
183	oStats.servIPCacheHits = htonl(mIPCacheHits);
184	oStats.servPXCacheHits = htonl(mPXCacheHits);
185	oStats.servInProgressCount = htonl(mInProgressCount);
186
187	oStats.servIP4CacheCount = htonl(mIP4Cache->GetCount());
188	oStats.servIP4CacheSize = htonl(mConfig->cache.maxips);
189	oStats.servIP4CacheExpire = htonl(mConfig->cache.expire);
190	oStats.servPX4CacheCount = htonl(mPX4Cache->GetCount());
191	oStats.servPX4CacheSize = htonl(kProxyCacheSize);
192	oStats.servPX4CacheExpire = htonl(mConfig->cache.proxy_expire);
193
194	oStats.sessCurrent = htonl(mSessions.size());
195	oStats.sessCount = htonl(mSessionCount);
196	oStats.sessRejectedCount = htonl(mSessionRejectedCount);
197
198	snprintf(oStats.version, sizeof(oStats.version), "pxyscand %s "
199	"(service msg v%d)", VERSION, PX_VERSION);
200	}
201
202	// Query Complete
203	//
204	// Solution with 1 mutex
205	//
206	// lock A lock A
207	// CHK CACHES PUT IN CACHE
208	// CHK SCANS REMOVE SCAN
209	// unlock A unlock A
210	//
211	// Solution with 2 mutex [choosen]
212	//
213	// lock A lock A
214	// CHK CACHES PUT IN CACHE
215	// lock B unlock A
216	// unlock A lock B
217	// CHK SCANS REMOVE SCAN
218	// unlock B unlock B
219
220
221	void
222	PXServer::NewScanIP4(const opas_msg_query &inQuery, PXSession *inSession)
223	{
224	PXProxyBinInfo *infop;
225	time_t t;
226
227	mQueryCount++;
228
229	if (mConfig->CheckNoScan(inQuery.addr))
230	{
231	mNoScanHits++;
232	inSession->ScanResultNoProxy(inQuery, true);
233	return;
234	}
235
236	peak_task_mutex_lock(mCacheMutex); // lock the cache
237
238	if ((t = mPX4Cache->Get(inQuery.addr, (void**)&infop)) > 0)
239	{
240	// OMG!
241	mPXCacheHits++;
242
243	// Update cache hits count, just for the fun.
244	infop->cache_hits = htonl(ntohl(infop->cache_hits) + 1);
245
246	peak_task_mutex_unlock(mCacheMutex);
247
248	inSession->ScanResultProxy(inQuery, true, t,
249	ntohs(infop->type), ntohs(infop->port),
250	infop->descr);
251	return;
252	}
253
254	if ((t = mIP4Cache->Get(inQuery.addr, NULL)) > 0)
255	{
256	// In cache? If so it's quick!
257	mIPCacheHits++;
258
259	peak_task_mutex_unlock(mCacheMutex);
260
261	inSession->ScanResultNoProxy(inQuery, true);
262	return;
263	}
264
265	peak_task_mutex_lock(mScanMutex); // lock first (important)
266	peak_task_mutex_unlock(mCacheMutex); // ok, others can consult the cache
267
268	// Search for existing scan session for this IP...
269	PXScan *scan;
270	bool newScan;
271	scanMapType::const_iterator i = mScans.find(inQuery.addr);
272
273	if (i == mScans.end())
274	{
275	// Normal case, I suppose. Create new "scan" instance and remember it.
276	scan = new PXScan(inQuery, inSession, mScanManager);
277	mScans.insert(scanMapPairType(inQuery.addr, scan));
278	newScan = true;
279	}
280	else
281	{
282	// Found... do not scan twice.
283	mInProgressCount++;
284	scan = i->second;
285	if (!scan->IsSessionReferenced(inSession))
286	scan->AppendQuery(inQuery, inSession);
287	newScan = false;
288	}
289
290	peak_task_mutex_unlock(mScanMutex); // "secure!"
291
292	if (newScan)
293	scan->Schedule();
294	}
295
296	void
297	PXServer::DeleteScanIP4(const opas_msg_query &inQuery)
298	{
299	peak_task_mutex_lock(mScanMutex);
300
301	mScannedCount++;
302
303	scanMapType::iterator i = mScans.find(inQuery.addr);
304	assert(i != mScans.end());
305	mScans.erase(i);
306
307	peak_task_mutex_unlock(mScanMutex);
308	}
309
310	void
311	PXServer::ScanIP4ResultNoProxy(const opas_msg_query &inQuery)
312	{
313	peak_task_mutex_lock(mCacheMutex);
314	mIP4Cache->Put(inQuery.addr, peak_time(), NULL);
315	peak_task_mutex_unlock(mCacheMutex); // safe
316
317	this->DeleteScanIP4(inQuery);
318	}
319
320	void
321	PXServer::ScanIP4ResultProxy(const opas_msg_query &inQuery,
322	uint16_t inTypeOfProxy, uint16_t inPort,
323	const char *inDescr)
324	{
325	PXProxyBinInfo info;
326	memset(&info, 0, sizeof(info));
327	info.cache_hits = htonl(0);
328	info.type = htons(inTypeOfProxy);
329	info.port = htons(inPort);
330	strncpy(info.descr, inDescr, sizeof(info.descr) - 1);
331
332	peak_task_mutex_lock(mCacheMutex);
333	mPX4Cache->Put(inQuery.addr, peak_time(), &info);
334	peak_task_mutex_unlock(mCacheMutex);
335
336	mProxyCount++;
337
338	this->DeleteScanIP4(inQuery);
339	}
340
341	bool
342	PXServer::RemoveAddressFromCache(const in_addr &inAddress)
343	{
344	peak_task_mutex_lock(mCacheMutex);
345	bool res = mPX4Cache->Invalidate(inAddress);
346	mIP4Cache->Invalidate(inAddress); // Better safe than sorry.
347	peak_task_mutex_unlock(mCacheMutex);
348	return res;
349	}
350
351	bool
352	PXServer::RemoveAddressFromCache(const in6_addr &inAddress)
353	{
354	bool res = false;
355	#if 0
356	peak_task_mutex_lock(mCacheMutex);
357	res = mPX6Cache->Invalidate(inAddress);
358	mIP6Cache->Invalidate(inAddress);
359	peak_task_mutex_unlock(mCacheMutex);
360	#endif
361	return res;
362	}
363
364	void
365	PXServer::ProcessEvent(peak_stream s, int type)
366	{
367	if (type == PEAK_STREAM_EVT_ACCEPT)
368	{
369	this->NewConnection(s);
370	}
371	else
372	{
373	assert(0);
374	}
375	}
376
377	void
378	PXServer::NewConnection(peak_stream s)
379	{
380	peak_stream ns;
381	struct sockaddr_in sin;
382	int len = sizeof(sin);
383
384	if (!(ns = peak_stream_accept_create(s, PEAK_STREAM_OPT_DEFAULT,
385	PXSession::EventCallback, this)))
386	return;
387
388	if (mSessions.size() >= MAX_SESSIONS)
389	{
390	peak_release(ns);
391	mSessionRejectedCount++;
392	return;
393	}
394
395	// Limit DoS attacks... check the peer ASAP (it's already too late!)
396	peak_stream_get_address(ns, (struct sockaddr *)&sin, &len);
397
398	if (!mConfig->OPASIsAllowed(sin.sin_addr))
399	{
400	peak_release(ns); // Reject
401	mSessionRejectedCount++;
402	return;
403	}
404
405	mSessions.push_back(new PXSession(this, ns));
406	mSessionCount++;
407	}
408
409	void
410	PXServer::RepeaterFire()
411	{
412	peak_task_mutex_lock(mCacheMutex);
413	this->Rehash();
414	peak_task_mutex_unlock(mCacheMutex);
415	}
416
417	void
418	PXServer::EventCallback(peak_stream s, int type, void *context)
419	{
420	((PXServer *)context)->ProcessEvent(s, type);
421	}