ResourcePackUpdater/src/main.py

from aiohttp import web
import os
import hmac
from pack import Pack

class App:
    def __init__(self):
        self.pack = Pack()
        print("Done")
        self.app = web.Application()
        self.app.add_routes([web.post('/update', self.updatePack)])
        self.app.router.add_static('/files', "./files")
        print("Started")
        web.run_app(self.app)

    async def updatePack(self, request):
        try:
            json = await request.json()
            text = await request.read()
        except:
            return web.Response(status=500, text="Invalid json or text")
        
        header_signature = request.headers.get('X-Hub-Signature')
        print("Hook recieved")
        if "ref" in json and json["ref"] == 'refs/heads/release':
            if (status := self.verify_signature(text, header_signature)) == 200: 
                self.pack.pull()
                self.pack.collate()
                self.pack.compress()
                self.pack.hash()
                self.pack.upload()
        else:
            status = 404
        print(status)
        return web.Response(status=status)

    def verify_signature(self, request_data, header_signature):
        # do not store your secret key in your code, pull from environment variable
        secret_key = os.environ.get('WEBHOOK_SECRET')

        if not header_signature:
            return 404

        # separate the signature from the sha1 indication
        sha_name, signature = header_signature.split('=')
        if sha_name != 'sha1':
            return 501

        # create a new hmac with the secret key and the request data
        mac = hmac.new(secret_key.encode(), msg=request_data, digestmod='sha1')

        # verify the digest matches the signature
        if not hmac.compare_digest(mac.hexdigest(), signature):
            return 404
        return 200

if __name__ == '__main__':
    app = App()