Github mirror handler for LXQt

github.py

import hmac
import hashlib
import json
import logging
import os
import subprocess
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer

try:
        from config import GIT_DIRECTORY, PORT, SECRET
except ImportError:
        logging.error("A config.py file must exist and contain the GIT_DIRECTORY, PORT and SECRET settings")
        exit(1)


def pshb_verify_signature(secret, body, signature):
        digest = hmac.new(secret, body, digestmod=hashlib.sha1).hexdigest()
        return digest == signature


class GithubMirrorHandler(BaseHTTPRequestHandler):
        def error(self):
                self.send_response(500)
                self.end_headers()
                return

        def call(self, args):
                proc = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                stdout, stderr = proc.communicate()
                self.wfile.write(b"Command: " + bytes(" ".join(args), "utf-8") + b"\n")
                self.wfile.write(b"STDOUT: " + stdout + b"\n")
                self.wfile.write(b"STDERR: " + stderr + b"\n")

        def do_POST(self):
                sig = self.headers.get("X-Hub-Signature", "")
                if not sig.startswith("sha1="):
                        self.error()
                length = self.headers.get("Content-Length", "0")
                if not length.isdigit():
                        self.error()
                sig = sig.split("=")[1]
                body = self.rfile.read(int(length))
                ok = pshb_verify_signature(bytes(SECRET, "utf-8"), body, sig)
                if not ok:
                        self.error()
                self.send_response(200)
                self.send_header("Content-Type", "text/plain")
                self.end_headers()
                data = json.loads(body.decode("utf-8"))
                repo = os.path.join(GIT_DIRECTORY, data["repository"]["full_name"] + ".git")
                self.call(["git", "--git-dir=%s" % (repo), "fetch", data["repository"]["ssh_url"]])
                self.call(["git", "--git-dir=%s" % (repo), "reset", "--soft", "FETCH_HEAD"])


def main():
        server_address = ("", PORT)
        httpd = HTTPServer(server_address, GithubMirrorHandler)
        try:
                logging.info("Serving on %s:%r" % server_address)
                httpd.serve_forever()
        except KeyboardInterrupt:
                logging.info("Process terminated")
                sys.exit(0)


if __name__ == "__main__":
        main()