Issue 6291 - Use client certificate for Windows Store uploads

sitescripts/extensions/bin/createNightlies.py

 # This file is part of the Adblock Plus web scripts,
 # Copyright (C) 2006-present eyeo GmbH
 #
 # Adblock Plus is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License version 3 as
 # published by the Free Software Foundation.
 #
 # Adblock Plus is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with Adblock Plus.  If not, see <http://www.gnu.org/licenses/>.
 
 """
 
 Nightly builds generation script
 ================================
 
   This script generates nightly builds of extensions, together
   with changelogs and documentation.
 
 """
 
 import argparse
 import ConfigParser
+import binascii
 import base64
 import hashlib
 import hmac
 import json
 import logging
 import os
 import pipes
-import random
 import shutil
 import struct
 import subprocess
 import sys
 import tempfile
 import time
+import uuid
 from urllib import urlencode
 import urllib2
 import urlparse
 import zipfile
 import contextlib
+from xml.dom.minidom import parse as parseXml
 
-from xml.dom.minidom import parse as parseXml
+from Crypto.PublicKey import RSA
+from Crypto.Signature import PKCS1_v1_5
+import Crypto.Hash.SHA256
 
 from sitescripts.extensions.utils import (
     compareVersions, Configuration,
     writeAndroidUpdateManifest,
 )
 from sitescripts.utils import get_config, get_template
 
 MAX_BUILDS = 50
 
 
@@ skipping 388 matching lines @@
         try:
             for i, entry in enumerate(current[platform]):
                 if entry[filter_key] == filter_value:
                     del current[platform][i]
                 if len(current[platform]) == 0:
                     del current[platform]
         except KeyError:
             pass
         self.write_downloads_lockfile(current)
 
-    def generate_jwt_request(self, issuer, secret, url, method, data=None,
-                             add_headers=[]):
-        header = {
-            'alg': 'HS256',     # HMAC-SHA256
-            'typ': 'JWT',
+    def azure_jwt_signature_fnc(self):
+        return (
+            'RS256',
+            lambda s, m: PKCS1_v1_5.new(s).sign(Crypto.Hash.SHA256.new(m)),
+        )
+
+    def mozilla_jwt_signature_fnc(self):
+        return (
+            'HS256',
+            lambda s, m: hmac.new(s, msg=m, digestmod=hashlib.sha256).digest(),
+        )
+
+    def sign_jwt(self, issuer, secret, url, signature_fnc, jwt_headers={}):
+        alg, fnc = signature_fnc()
+
+        header = {'typ': 'JWT'}
+        header.update(jwt_headers)
+        header.update({'alg': alg})
+
+        issued = int(time.time())
+        expires = issued + 60
+
+        payload = {
+            'aud': url,
+            'iss': issuer,
+            'sub': issuer,
+            'jti': str(uuid.uuid4()),
+            'iat': issued,
+            'nbf': issued,
+            'exp': expires,
         }
 
-        issued = int(time.time())
-        payload = {
-            'iss': issuer,
-            'jti': random.random(),
-            'iat': issued,
-            'exp': issued + 60,
-        }
+        segments = [base64.urlsafe_b64encode(json.dumps(header)),
+                    base64.urlsafe_b64encode(json.dumps(payload))]
 
-        hmac_data = '{}.{}'.format(
-            base64.b64encode(json.dumps(header)),
-            base64.b64encode(json.dumps(payload)),
-        )
+        signature = fnc(secret, b'.'.join(segments))
+        segments.append(base64.urlsafe_b64encode(signature))
+        return b'.'.join(segments)
 
-        signature = hmac.new(secret, msg=hmac_data,
-                             digestmod=hashlib.sha256).digest()
-        token = '{}.{}'.format(hmac_data, base64.b64encode(signature))
+    def generate_mozilla_jwt_request(self, issuer, secret, url, method,
+                                     data=None, add_headers=[]):
+        signed = self.sign_jwt(issuer, secret, url,
+                               self.mozilla_jwt_signature_fnc)
 
         request = urllib2.Request(url, data)
-        request.add_header('Authorization', 'JWT ' + token)
+        request.add_header('Authorization', 'JWT ' + signed)
         for header in add_headers:
             request.add_header(*header)
         request.get_method = lambda: method
 
         return request
 
     def uploadToMozillaAddons(self):
         import urllib3
 
         config = get_config()
 
         upload_url = ('https://addons.mozilla.org/api/v3/addons/{}/'
                       'versions/{}/').format(self.extensionID, self.version)
 
         with open(self.path, 'rb') as file:
             data, content_type = urllib3.filepost.encode_multipart_formdata({
                 'upload': (
                     os.path.basename(self.path),
                     file.read(),
                     'application/x-xpinstall',
                 ),
             })
 
-        request = self.generate_jwt_request(
+        request = self.generate_mozilla_jwt_request(
             config.get('extensions', 'amo_key'),
             config.get('extensions', 'amo_secret'),
             upload_url,
             'PUT',
             data,
             [('Content-Type', content_type)],
         )
 
         try:
             urllib2.urlopen(request).close()
@@ skipping 15 matching lines @@
         os.remove(self.path)
 
     def download_from_mozilla_addons(self, buildtype, version, app_id):
         config = get_config()
         iss = config.get('extensions', 'amo_key')
         secret = config.get('extensions', 'amo_secret')
 
         url = ('https://addons.mozilla.org/api/v3/addons/{}/'
                'versions/{}/').format(app_id, version)
 
-        request = self.generate_jwt_request(iss, secret, url, 'GET')
+        request = self.generate_mozilla_jwt_request(
+            iss, secret, url, 'GET',
+        )
         response = json.load(urllib2.urlopen(request))
 
         filename = '{}-{}.xpi'.format(self.basename, version)
         self.path = os.path.join(
             config.get('extensions', 'nightliesDirectory'),
             self.basename,
             filename,
         )
 
         necessary = ['passed_review', 'reviewed', 'processed', 'valid']
         if all(response[x] for x in necessary):
             download_url = response['files'][0]['download_url']
             checksum = response['files'][0]['hash']
 
-            request = self.generate_jwt_request(iss, secret, download_url,
-                                                'GET')
+            request = self.generate_mozilla_jwt_request(
+                iss, secret, download_url, 'GET',
+            )
             try:
                 response = urllib2.urlopen(request)
             except urllib2.HTTPError as e:
                 logging.error(e.read())
 
             # Verify the extension's integrity
             file_content = response.read()
             sha256 = hashlib.sha256(file_content)
             returned_checksum = '{}:{}'.format(sha256.name, sha256.hexdigest())
 
@@ skipping 70 matching lines @@
         request.get_method = lambda: 'POST'
         request.add_header('Authorization', auth_token)
         request.add_header('x-goog-api-version', '2')
         request.add_header('Content-Length', '0')
 
         response = json.load(opener.open(request))
 
         if any(status not in ('OK', 'ITEM_PENDING_REVIEW') for status in response['status']):
             raise Exception({'status': response['status'], 'statusDetail': response['statusDetail']})
 
+    def generate_certificate_token_request(self, url, private_key):
+        # Construct the token request according to
+        # https://docs.microsoft.com/en-us/azure/active-directory/develop/active-directory-certificate-credentials
+        hex_val = binascii.a2b_hex(self.config.thumbprint)
+        x5t = base64.urlsafe_b64encode(hex_val).decode()
+
+        key = RSA.importKey(private_key)
+
+        signed = self.sign_jwt(self.config.clientID, key, url,
+                               self.azure_jwt_signature_fnc,
+                               jwt_headers={'x5t': x5t})
+
+        # generate oauth parameters for login.microsoft.com
+        oauth_params = {
+            'grant_type': 'client_credentials',
+            'client_id': self.config.clientID,
+            'resource': 'https://graph.windows.net',
+            'client_assertion_type': 'urn:ietf:params:oauth:client-assertion-'
+                                     'type:jwt-bearer',
+            'client_assertion': signed,
+        }
+
+        request = urllib2.Request(url, urlencode(oauth_params))
+        request.get_method = lambda: 'POST'
+
+        return request
+
     def get_windows_store_access_token(self):
-        # use refresh token to obtain a valid access token
-        # https://docs.microsoft.com/en-us/azure/active-directory/active-directory-protocols-oauth-code#refreshing-the-access-tokens
-        server = 'https://login.microsoftonline.com'
-        token_path = '{}/{}/oauth2/token'.format(server, self.config.tenantID)
+        # use client certificate to obtain a valid access token
+        url_template = 'https://login.microsoftonline.com/{}/oauth2/token'
+        url = url_template.format(self.config.tenantID)
+
+        with open(self.config.privateKey, 'r') as fp:
+            private_key = fp.read()
 
         opener = urllib2.build_opener(HTTPErrorBodyHandler)
-        post_data = urlencode([
-            ('refresh_token', self.config.refreshToken),
-            ('client_id', self.config.clientID),
-            ('client_secret', self.config.clientSecret),
-            ('grant_type', 'refresh_token'),
-            ('resource', 'https://graph.windows.net'),
-        ])
-        request = urllib2.Request(token_path, post_data)
+        request = self.generate_certificate_token_request(url, private_key)
+
         with contextlib.closing(opener.open(request)) as response:
             data = json.load(response)
             auth_token = '{0[token_type]} {0[access_token]}'.format(data)
 
         return auth_token
 
     def upload_appx_file_to_windows_store(self, file_upload_url):
         # Add .appx file to a .zip file
         zip_path = os.path.splitext(self.path)[0] + '.zip'
         with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf:
@@ skipping 130 matching lines @@
 
             # update nightlies config
             self.config.latestRevision = self.revision
 
             if (self.config.type == 'gecko' and
                     self.config.galleryID and
                     get_config().has_option('extensions', 'amo_key')):
                 self.uploadToMozillaAddons()
             elif self.config.type == 'chrome' and self.config.clientID and self.config.clientSecret and self.config.refreshToken:
                 self.uploadToChromeWebStore()
-            elif self.config.type == 'edge' and self.config.clientID and self.config.clientSecret and self.config.refreshToken and self.config.tenantID:
+            elif self.config.type == 'edge' and self.config.clientID and self.config.tenantID and self.config.privateKey and self.config.thumbprint:
                 self.upload_to_windows_store()
 
         finally:
             # clean up
             if self.tempdir:
                 shutil.rmtree(self.tempdir, ignore_errors=True)
 
     def download(self):
         download_info = self.read_downloads_lockfile()
         downloads = self.downloadable_repos.intersection(download_info.keys())
@@ skipping 61 matching lines @@
 
     file = open(nightlyConfigFile, 'wb')
     nightlyConfig.write(file)
 
 
 if __name__ == '__main__':
     parser = argparse.ArgumentParser()
     parser.add_argument('--download', action='store_true', default=False)
     args = parser.parse_args()
     main(args.download)