141 lines
5.0 KiB
Python
141 lines
5.0 KiB
Python
"""Transifex webhook handler """
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import requests
|
|
import re
|
|
import time
|
|
|
|
from buildbot.process.properties import Properties
|
|
from buildbot.util import bytes2unicode, unicode2bytes
|
|
from buildbot.www.hooks.base import BaseHookHandler
|
|
|
|
from twisted.internet import defer
|
|
from twisted.python import log
|
|
|
|
|
|
_HEADER_USER_AGENT = 'User-Agent'
|
|
_HEADER_SIGNATURE = 'X-TX-Signature'
|
|
_HEADER_URL_PATH = 'X-TX-Url'
|
|
_HTTP_DATE = 'date'
|
|
_EVENT_KEY = 'event'
|
|
author = 'buildbot-transifex'
|
|
|
|
|
|
class TransifexHandler(BaseHookHandler):
|
|
|
|
def __init__(self, master, secret, transifex_to_github_map, options=None):
|
|
if not options:
|
|
options = {}
|
|
self.secret = secret
|
|
self.master = master
|
|
self.options = options
|
|
self.transifex_to_github_map = transifex_to_github_map
|
|
|
|
|
|
def returnMessage(self, status = False, message = "Unimplemented"):
|
|
output = json.dumps({"status": "OK" if status else "FAIL", "message": message})
|
|
return [output, [('Content-type', 'application/json')]]
|
|
|
|
def _verifyTransifexSignature(
|
|
self, request, content, signature, header_signature
|
|
):
|
|
http_verb = 'POST'
|
|
http_url_path = request.getHeader(_HEADER_URL_PATH)
|
|
http_gmt_date = request.getHeader(_HTTP_DATE)
|
|
content_md5 = hashlib.md5(content).hexdigest()
|
|
msg = b'\n'.join([
|
|
http_verb, http_url_path, http_gmt_date, content_md5
|
|
])
|
|
tx_signature = base64.b64encode(
|
|
hmac.new(
|
|
key=self.rendered_secret,
|
|
msg=msg,
|
|
digestmod=hashlib.sha256
|
|
).digest()
|
|
)
|
|
if tx_signature != header_signature:
|
|
raise ValueError("Tx Signature mismatch")
|
|
|
|
if signature != request.getHeader(_HEADER_SIGNATURE):
|
|
raise ValueError("Signature mismatch")
|
|
return True
|
|
|
|
def process_translation_completed(self, payload, codebase):
|
|
changes = []
|
|
translated_request = self._transform_variables(payload['project'], payload['resource'])
|
|
ts = int(time.time())
|
|
change = {
|
|
'author': author,
|
|
'branch': translated_request["branch"],
|
|
'branch': translated_request["repository"],
|
|
'project': translated_request["project"],
|
|
'properties': {
|
|
"transifex_language": payload.get("language", "None"),
|
|
"transifex_event": payload.get("event", "None"),
|
|
"transifex_project": payload.get("project", "None"),
|
|
"transifex_resource": payload.get("resource", "None"),
|
|
"transifex_branch": "translate_" + payload['language'] + "_" + str(ts)
|
|
}
|
|
}
|
|
if codebase is not None:
|
|
change['codebase'] = codebase
|
|
changes.insert(0, change)
|
|
return changes
|
|
|
|
def _transform_variables(self, transifex_project):
|
|
if transifex_project is None:
|
|
raise ValueError("Unknown project %s from transifex".format(transifex_project))
|
|
key = transifex_project
|
|
_map = self.map[key]
|
|
repository = _map["repository"]
|
|
project = re.sub(r'^.*/(.*?)(\.git)?$', r'\1', repository)
|
|
return{
|
|
'project': project,
|
|
'repository': repository,
|
|
'branch': _map["branch"],
|
|
}
|
|
|
|
@defer.inlineCallbacks
|
|
def getChanges(self, request):
|
|
change = {}
|
|
self.secret = None
|
|
if isinstance(self.options, dict):
|
|
self.secret = self.options.get('secret')
|
|
try:
|
|
content = request.content.read()
|
|
content_text = bytes2unicode(content)
|
|
payload = json.loads(content_text)
|
|
except Exception as exception:
|
|
raise ValueError('Error loading JSON: ' + str(exception))
|
|
|
|
if self.secret is not None:
|
|
p = Properties()
|
|
p.master = self.master
|
|
option = self.options
|
|
rendered_secret = yield p.render(self.secret)
|
|
signature = hmac.new(
|
|
unicode2bytes(rendered_secret),
|
|
unicode2bytes(content_text.strip()),
|
|
digestmod=hashlib.sha256)
|
|
header_signature = bytes2unicode(
|
|
request.getHeader(_HEADER_SIGNATURE))
|
|
self._verifyTransifexSignature(request, content, rendered_secret, signature, header_signature)
|
|
event_type = payload.get("event", "None")
|
|
log.msg("Received event '{}' from transifex".format(event_type))
|
|
|
|
codebase = ""
|
|
changes = []
|
|
|
|
handler_function = getattr(self, 'process_{}'.format(event_type), None)
|
|
if not handler_function:
|
|
log.msg("Ignoring transifex event '{}'".format(event_type))
|
|
else:
|
|
changes = handler_function(payload, event_type, codebase)
|
|
|
|
return (changes, 'transifex')
|
|
|
|
transifex = TransifexHandler
|