buildbot-transifex/buildbot_transifex/webhook.py

"""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