bitbucket_clone_all.py

#!/usr/bin/env python3
import argparse
import requests
import json
import urllib3
from urllib.parse import urlparse
import os
import re
from getpass import getpass


# SUPPRESS WARNINGS ############################################################
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


# CLONE ########################################################################
def do_clone_ssh(repo_list):
    for repo in repo_list:
        parsed_git_url = urlparse(repo)
        directory = OUTPUT_DIR + re.sub('.git', '', parsed_git_url.path)
        os.system('git clone \'{}\' \'{}\''.format(repo, directory))

def do_clone_http(repo_list):
    for repo in repo_list:
        parsed_http_url = urlparse(repo)
        directory = OUTPUT_DIR + re.sub('.git', '', repo.split('/scm/')[1])

        if args.user and args.password:
            # for some unknown reason, urlparse cannot replace username/password directly ...
            parsed_http_url = parsed_http_url._replace(netloc=args.user + ':' + args.password + '@' + parsed_http_url.netloc)

        os.system('git clone \'{}\' \'{}\''.format(parsed_http_url.geturl(), directory))

def do_clone(repo_list, method):
    if method == 'HTTP':
        do_clone_http(repo_list)
    if method == 'SSH':
        do_clone_ssh(repo_list)


# API ##########################################################################
def api_get_repo_list(session, url, method):
    repo_list = []

    r = session.get(url + '/rest/api/1.0/projects?limit=10000')
    projects = json.loads(r.text)

    for project in projects['values']:
        r = session.get(url + '/rest/api/1.0/projects/' + project['key'] + '/repos?limit=10000')
        repos = json.loads(r.text)

        for repo in repos['values']:
            clone_options = repo['links']['clone']
            for clone_option in clone_options:
                if clone_option['name'] == method.lower():
                    repo_list.append(clone_option['href'])

    return repo_list


# MAIN #########################################################################
parser = argparse.ArgumentParser()
parser.add_argument('url', type=str)
parser.add_argument('-u', '--user', type=str)
parser.add_argument('-p', '--password', type=str)
parser.add_argument('-o', '--output-dir', type=str, required=True)
parser.add_argument('-m', '--method', type=str, default='HTTP', help='Cloning method: HTTP or SSH (default: HTTP)')

args = parser.parse_args()

s = requests.Session()
s.verify = False

if args.user:
    if not args.password:
        args.password = getpass("password: ")
    s.auth = (args.user, args.password)
    args.password = args.password.replace('@', '%40') # escape for HTTP later on

OUTPUT_DIR = args.output_dir + '/'

repo_list = api_get_repo_list(s, args.url.rstrip('/'), args.method)
print('[+] Cloning {} repositories...'.format(len(repo_list)))
do_clone(repo_list, args.method)