ppf/proxywatchd.py

#!/usr/bin/env python

import threading
import time, random, string, re, copy
import requests
#from geoip import geolite2

import config

import mysqlite
from misc import _log
import rocksock

class WorkerJob():
	def __init__(self, proxy, proto, failcount):
		self.proxy = proxy
		self.proto = proto
		self.failcount = failcount
		self.nextcheck = None
		self.duration = None

	def is_drone_bl(self, proxy):
		p = proxy.split(':')[0]
		proxies = {'http':'socks4://%s:%s@%s' % (p,p,random.choice(config.torhosts))}
		resp = requests.get('http://dronebl.org/lookup?ip=%s' % p, proxies=proxies)
		if 'No incidents regarding' in resp.text: return 0
		else: return 1

	def connect_socket(self):
		protos = ['http', 'socks5', 'socks4'] if self.proto is None else self.proto

		for proto in protos:
			torhost = random.choice(config.torhosts)
			duration = time.time()
			proxies = [
				rocksock.RocksockProxyFromURL('socks4://%s' % torhost),
				rocksock.RocksockProxyFromURL('%s://%s' % (proto, self.proxy)),
			]

			srv = random.choice(config.servers).strip()
			try:
				sock = rocksock.Rocksock(host=srv, port=6697, ssl=True, proxies=proxies, timeout=config.timeout)
				sock.connect()
				sock.send('%s\n' % random.choice(['NICK', 'USER', 'JOIN', 'MODE', 'PART', 'INVITE', 'KNOCK', 'WHOIS', 'WHO', 'NOTICE', 'PRIVMSG', 'PING', 'QUIT']))
				return sock, proto, duration, torhost, srv
			except KeyboardInterrupt as e:
				raise(e)
			except: sock.disconnect()

		return None, None, None, None, None

	def run(self):
		self.nextcheck = (time.time() + 1800 + ((1+int(self.failcount)) * 3600))

		sock, proto, duration, tor, srv = self.connect_socket()
		if not sock:
			self.failcount += 1
			return
		try:
			recv = sock.recv(6)

			# good data
			if re.match('^(:|ERROR|PING|PONG|NOTICE|\*\*\*)', recv, re.IGNORECASE):
				duration = (time.time() - duration)
				self.nextcheck = (time.time() + 1800)

				#match = geolite2.lookup(proxy[0].split(':')[0])
				match = None
				if match is not None: match = match.country
				else: match = 'unknown'

				#dronebl = self.is_drone_bl(proxy[0])
				self.proto = proto
				self.duation = duration
				self.failcount = 0
				_log('%s://%s; c: %s; d: %d sec(s); tor: %s; srv: %s; recv: %s' % (proto, self.proxy, match, duration, tor, srv, recv), 'xxxxx')
		except KeyboardInterrupt as e:
			raise e
		except:
			self.failcount += 1
		finally:
			sock.disconnect()


class WorkerThread():
	def __init__ (self, id):
		self.id = id
		self.done = threading.Event()
		self.thread = None
		self.workqueue = []
		self.workdone = []
	def stop(self):
		self.done.set()
	def term(self):
		if self.thread: self.thread.join()
	def add_jobs(self, jobs):
		self.workqueue.extend(jobs)
	def jobcount(self):
		return len(self.workqueue)
	def collect(self):
		wd = copy.copy(self.workdone)
		self.workdone = []
		return wd
	def start_thread(self):
		self.thread = threading.Thread(target=self.workloop)
		self.thread.daemon = True
		self.thread.start()
	def workloop(self):
		while True:
			if len(self.workqueue):
				job = self.workqueue.pop()
				job.run()
				self.workdone.append(job)
			elif not self.thread:
				break
			if self.done.is_set(): break
			time.sleep(0.01)
		if self.thread:
			_log("thread %s terminated", self.id)

class Proxywatchd():

	def stop(self):
		_log('Requesting proxywatchd to halt (%d thread(s))' % len([item for item in self.threads if True]))
		self.stopping.set()

	def __init__(self):
		config.load()
		self.threads = []
		self.stopping = threading.Event()

		# create table if needed
		self.mysqlite = mysqlite.mysqlite(config.database, str)
		self.mysqlite.execute('CREATE TABLE IF NOT EXISTS proxylist (proxy BLOB, country BLOB, added INT, failed INT, tested INT, source BLOB, dronebl INT, proto TEXT, duration INT)')
		self.mysqlite.commit()
		self.mysqlite.close()
		self.mysqlite = None

		self.echoise = time.time() - 3600;
		self.ticks = time.time() - 3600;
		self.jobs = []
		self.collected = []

	def prepare_jobs(self):
		q = 'SELECT proxy,proto,failed FROM proxylist WHERE failed<? and tested<? ORDER BY RANDOM()' # ' LIMIT ?'
		rows = self.mysqlite.execute(q, (config.maxfail, time.time())).fetchall()
		for row in rows:
			job = WorkerJob(row[0], row[1], row[2])
			self.jobs.append(job)

	def collect_work(self):
		for wt in self.threads:
			self.collected.extend(wt.collect())

	def submit_collected(self):
		query = 'UPDATE proxylist SET failed=?,tested=?,dronebl=?,country=?,proto=?,duration=? WHERE proxy=?'
		for job in self.collected:
			self.mysqlite.execute(query, (job.failcount, job.nextcheck, 1, "unknown", job.proto, job.duration, job.proxy))
		self.mysqlite.commit()

	def run_background(self):
		t = threading.Thread(target=self.run)
                t.daemon = True
                t.start()

	def run(self):
		_log('Starting proxywatchd..', 'notice')
		self.mysqlite = mysqlite.mysqlite(config.database, str)

		for i in range(config.watchd_threads):
			threadid = ''.join( [ random.choice(string.letters) for x in range(5) ] )
			wt = WorkerThread(threadid)
			if config.watchd_threads > 1:
				wt.start_thread()
			self.threads.append(wt)

		while True:

			if self.stopping.is_set():
				for wt in self.threads:
					wt.stop()
				for wt in self.threads:
					wt.term()
				self.collect_work()
				self.submit_collected()
				break

			if len(self.jobs) == 0:
				self.prepare_jobs()
				if len(self.jobs):
					jpt = len(self.jobs)/config.watchd_threads
					if len(self.jobs)/float(config.watchd_threads) - jpt > 0.0: jpt += 1
					for tid in range(config.watchd_threads):
						self.threads[tid].add_jobs(self.jobs[tid*jpt:tid*jpt+jpt])
					self.jobs = []

			if config.watchd_threads == 1: # single_thread scenario
				self.threads[0].workloop()

			self.collect_work()

			if len(self.collected) > 50:
				self.submit_collected()

			time.sleep(1)

		self.mysqlite.close()

if __name__ == '__main__':
	w = Proxywatchd()
	try:
		w.run()
	except KeyboardInterrupt as e:
		w.stop()