secret_overwrite_test_python35.py

import json
import codecs
import ssl
import sys
import urllib.request
import xml.etree.ElementTree as Tree
from collections import namedtuple

_DEFAULT_TEST_URL = 'https://anspedite-3.atlassian.net'


# noinspection PyProtectedMember
def _set_secret(descriptor_url, product_url, secret):
    path = product_url + '/plugins/servlet/oauth/consumer-info'
    ids = Tree.fromstring(urllib.request.urlopen(path, context=ssl._create_unverified_context()).read())
    client_key = ids.find('key').text
    public_key = ids.find('publicKey').text
    description = ids.find('description').text
    json_reader = codecs.getreader('utf-8')
    descriptor_response = urllib.request.urlopen(descriptor_url, context=ssl._create_unverified_context())
    descriptor = json.load(json_reader(descriptor_response))
    install_payload = {
        'baseUrl': product_url,
        'clientKey': client_key,
        'description': description,
        'eventType': 'installed',
        'key': descriptor['key'],
        'pluginsVersion': '1.2.0',
        'productType': client_key.split(':')[0],
        'publicKey': public_key,
        'serverVersion': '73000',
        'sharedSecret': secret
    }
    installed_callback_url = descriptor['baseUrl'] + descriptor['lifecycle']['installed']
    data = json.dumps(install_payload).encode('UTF-8')
    content_length = len(data)
    headers = {'Content-Type': 'application/json', 'Content-length': content_length}
    request = urllib.request.Request(installed_callback_url, data, headers)

    try:
        response = urllib.request.urlopen(request, context=ssl._create_unverified_context())
    except urllib.request.HTTPError as e:
        return e.code, e.read()
    return response.getcode(), response.read()


_TestResult = namedtuple('_TestResult', ['result', 'message', 'status_code', 'body'])


def _product_to_url(product):
    if product.lower() in ('jira', 'both'):
        return _DEFAULT_TEST_URL

    if product.lower() == 'confluence':
        return _DEFAULT_TEST_URL + '/wiki'

    if product.lower().startswith('https://'):
        return product

    raise Exception('Please specify "jira" ,"confluence" or "both" as the supported product')


def test_vulnerability(descriptor_url,
                       product='jira',
                       secret='aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'):
    try:
        product_url = _product_to_url(product)
        status_code, body = _set_secret(descriptor_url, product_url, secret)
        if status_code < 300:
            result = _TestResult('vulnerable',
                                 'Your add-on appears to be VULNERABLE '
                                 '(it returned a status code of {}).'.format(status_code),
                                 status_code, body)
        elif 300 <= status_code < 400:
            result = _TestResult('invalid',
                                 'Your add-on sent a redirect of some kind (status code was {}). '.format(status_code) +
                                 'Could not determine if it is vulnerable',
                                 status_code, body)
        elif 400 <= status_code < 500:
            result = _TestResult('safe',
                                 'Your add-on appears to be SAFE '
                                 '(it blocked the request with status code {}).'.format(status_code),
                                 status_code, body)
        else:
            result = _TestResult('invalid',
                                 'Your add-on returned an error status of {}'.format(status_code) +
                                 '. Could not determine whether it is safe.',
                                 status_code, body)
    except Exception as e:
        result = _TestResult('invalid', 'There was an exception while testing your add-on for a vulnerability. ' +
                             'Could not determine whether it is safe.', -1, e)
        raise e
    return result


if __name__ == "__main__":
    if '-h' in sys.argv or len(sys.argv) < 2 or len(sys.argv) > 4:
        print('')
        print('USAGE: python secret_overwrite_test.py add-on-descriptor-url [jira|confluence|both]')
        print('')
        print('add-on-descriptor-url: The https:// url to your atlassian-connect.json '
              '(the marketplace-hosted one or the one hosted by your service will both work)')
        print('jira|confluence|both: whether your add-on is for jira, confluence, or both (defaults to jira)')
        print('')
        print('EXAMPLE 1: '
              'python secret_overwrite_test.py https://jira-addon.example.com/atlassian-connect.json')
        print('')
        print('EXAMPLE 2: '
              'python secret_overwrite_test.py https://confluence-addon.example.com/atlassian-connect.json confluence')
        print('')
    else:
        test_result = test_vulnerability(*sys.argv[1:])
        print(test_result.message)
        if test_result.result == 'vulnerable':
            print('Your add-on should respond with 401 (unauthorized) to '
                  're-installation requests that don\'t have a valid JWT signed with the old secret')
        elif test_result.status_code == -1:
            print(test_result.body)