webhooks/webhooks.py

import logging
import base64
import hmac
import hashlib
import requests
import yaml
import os
import redis

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from flask import Flask, jsonify, request, abort

app = Flask(__name__)

# setup the environment
SNAP_USER_DATA = os.environ.get('SNAP_USER_DATA')
CONFIG_FILE = os.environ.get('CONFIG_FILE')
LOG_FILE = os.environ.get('LOG_FILE')
REDIS_HOST = os.environ.get('REDIS_HOST', 'localhost')
REDIS_PORT = os.environ.get('REDIS_PORT', 6379)
if SNAP_USER_DATA is None:
    if CONFIG_FILE is None:
        filename = 'config.yaml'
    else:
        filename = CONFIG_FILE

    if LOG_FILE is None:
        log_file = 'webhook.log'
    else:
        log_file = LOG_FILE
else:
    filename = SNAP_USER_DATA + '/config.yaml'
    log_file = SNAP_USER_DATA + '/webhook.log'

logging.basicConfig(level=logging.DEBUG, filename=log_file)

with open(filename, "rb") as config_file:
    config = yaml.load(config_file, Loader=yaml.SafeLoader)

rs = redis.Redis(host=REDIS_HOST, port=REDIS_PORT)

srht_public_key = Ed25519PublicKey.from_public_bytes(
        base64.b64decode(config['srht_payload_sig']))
# GitHub webhooks
# https://docs.github.com/en/developers/webhooks-and-events/webhooks/webhook-events-and-payloads

# Gittea webhooks
# https://docs.gitea.io/en-us/webhooks/

@app.route('/gitea/<gateway>', methods=['POST'])
def gitea_event(gateway=None):
    secret = bytes(config["gitea_secret"], "utf-8")
    signature = request.headers["X-Gitea-Signature"]
    event = request.headers["X-Gitea-Event-Type"]

    sigcheck = hmac.new(secret, msg=request.data, digestmod=hashlib.sha256).hexdigest()

    if gateway is None:
        logging.info("MISSING GATEWAY")
        abort(400)

    if sigcheck != signature:
        logging.info("INVALID GITEA SIGNATURE")
        abort(401)

    if event == "push":
        logging.debug(f"Gitea - {event} >> {gateway} ")
        gt_push_event(gateway, request.json)

    elif event == "issues":
        logging.debug(f"Gitea - {event} >> {gateway} ")
        gt_issue_event(gateway, request.json)

    elif event == "issue_comment":
        logging.debug(f"Gitea - {event} >> {gateway} ")
        gt_comment_event(gateway, request.json)

    else:
        logging.info("UNKNOWN GITEA EVENT")
        logging.debug(f"Gitea - {event} >> {gateway} ")
        logging.debug(request.json)

    return "", 200

def gt_push_event(gateway, data):
    logging.debug(f">> {gateway}")
    # logging.debug(data)
    project = data["repository"]["full_name"]
    username = data["pusher"]["username"]
    branch = data['ref'][len('refs/heads/'):]
    count = len(data["commits"])
    s = "s" if count > 1 else ""

    body = f"[{project}] {username} pushed {count} commit{s} to {branch}:\n"
    for commit in data["commits"]:
        cid = commit["id"][:8]
        title = commit["message"]
        url = commit["url"]
        body += f"- [{cid}]({url}) {title}\n"

    logging.debug(body)
    mb_url = config["codeberg_gateway"]
    payload = { 'gateway': gateway, 'username': "codeberg", 'text': body.rstrip() }
    r = requests.post(mb_url, json=payload)

def gt_issue_event(gateway, data):
    logging.debug(f">> {gateway}")
    # logging.debug(data)
    project = data["repository"]["full_name"]
    issue = data["issue"]
    username = issue["user"]["username"]
    number = issue["number"]
    title = issue["title"]
    description = issue["body"]
    action = data["action"]

    body = f"[{project}] {username} {action} issue #{number}: {title}\n"
    if action in ["opened", "reopened"]:
        body += f"{description}"

    logging.debug(body)
    mb_url = config["codeberg_gateway"]
    payload = { 'gateway': gateway, 'username': "codeberg", 'text': body.rstrip() }
    r = requests.post(mb_url, json=payload)

def gt_comment_event(gateway, data):
    logging.debug(f">> {gateway}")
    # logging.debug(data)
    project = data["repository"]["full_name"]
    issue = data["issue"]
    comment = data["comment"]
    username = comment["user"]["username"]
    number = issue["number"]
    title = issue["title"]
    action = data["action"]
    note = data["comment"]["body"]

    body = f"[{project}] {username} commented on issue #{number}: {title}\n"
    body += f"> {note}"

    logging.debug(body)
    mb_url = config["codeberg_gateway"]
    payload = { 'gateway': gateway, 'username': "codeberg", 'text': body.rstrip() }
    r = requests.post(mb_url, json=payload)

# GitLab webhooks
# https://docs.gitlab.com/ee/user/project/integrations/webhook_events.html

@app.route('/gitlab/<gateway>', methods=['POST'])
def gitlab_event(gateway=None):
    token = request.headers['X-Gitlab-Token']
    event = request.headers['X-Gitlab-Event']

    if gateway is None:
        logging.info("MISSING GATEWAY")
        abort(400)

    if token != config['gitlab_token']:
        logging.info("INVALID GITLAB TOKEN")
        abort(401)

    if event == "Push Hook":
        logging.debug(f"GitLab - {event} >> {gateway} ")
        gl_push_event(gateway, request.json)

    elif event == "Issue Hook":
        logging.debug(f"GitLab - {event} >> {gateway} ")
        gl_issue_event(gateway, request.json)

    elif event == "Note Hook":
        logging.debug(f"GitLab - {event} >> {gateway} ")
        gl_comment_event(gateway, request.json)

    elif event == "Merge Request Hook":
        logging.debug(f"GitLab - {event} >> {gateway} ")
        gl_mr_event(gateway, request.json)

    elif event == "Release Hook":
        logging.debug(f"GitLab - {event} >> {gateway} ")
        gl_release_event(gateway, request.json)

    else:
        logging.info("UNKNOWN GITLAB EVENT")
        logging.debug(f"GitLab - {event} >> {gateway} ")
        logging.debug(request.json)

    return "", 200

def gl_push_event(gateway, data):
    logging.debug(f">> {gateway}")
    # logging.debug(data)
    project = data["project"]["path_with_namespace"]
    username = data["user_username"]
    branch = data['ref'][len('refs/heads/'):]
    count = data["total_commits_count"]
    homepage = data["repository"]["homepage"]
    s = "s" if count > 1 else ""

    body = f"[{project}] {username} pushed {count} commit{s} to {branch}:\n"
    for commit in data["commits"]:
        cid = commit["id"][:8]
        title = commit["title"]
        url = commit["url"]
        body += f"- [{cid}]({url}) {title}\n"

    logging.debug(body)
    mb_url = config["gitlab_gateway"]
    payload = { 'gateway': gateway, 'username': "gitlab", 'text': body.rstrip() }
    r = requests.post(mb_url, json=payload)

def gl_issue_event(gateway, data):
    logging.debug(f">> {gateway}")
    # logging.debug(data)
    project = data["project"]["path_with_namespace"]
    username = data["user"]["username"]
    oid = data["object_attributes"]["iid"]
    title = data["object_attributes"]["title"]
    description = data['object_attributes']['description']

    if "action" in data["object_attributes"]:
        action = data["object_attributes"]["action"]
    else:
        action = "?"
    body = f"[{project}] {username} {action} issue #{oid}: {title}\n"
    if action in ["open", "reopen"]:
        body += f"{description}"
    elif action == "update":
        return

    logging.debug(body)
    mb_url = config["gitlab_gateway"]
    payload = { 'gateway': gateway, 'username': "gitlab", 'text': body }
    r = requests.post(mb_url, json=payload)

def gl_mr_event(gateway, data):
    logging.debug(f">> {gateway}")
    # logging.debug(data)
    project = data["project"]["path_with_namespace"]
    username = data["user"]["username"]
    oid = data["object_attributes"]["iid"]
    title = data["object_attributes"]["title"]
    description = data['object_attributes']['description']

    if "action" in data["object_attributes"]:
        action = data["object_attributes"]["action"]
    else:
        action = "?"
    body = f"[{project}] {username} {action} merge request !{oid}: {title}\n"
    if action in ["open", "reopen"]:
        body += f"{description}"
    elif action == "update":
        return

    logging.debug(body)
    mb_url = config["gitlab_gateway"]
    payload = { 'gateway': gateway, 'username': "gitlab", 'text': body }
    r = requests.post(mb_url, json=payload)

def gl_release_event(gateway, data):
    logging.debug(f">> {gateway}")
    # logging.debug(data)
    project = data["project"]["path_with_namespace"]
    name = data["name"]
    url = data["url"]

    if data["action"] != "create":
        return

    body = f"[{project}] release {name}\n"
    body += f"{url}"

    logging.debug(body)
    mb_url = config["gitlab_gateway"]
    payload = { 'gateway': gateway, 'username': "gitlab", 'text': body }
    r = requests.post(mb_url, json=payload)

def gl_comment_event(gateway, data):
    logging.debug(f">> {gateway}")
    logging.debug(data)
    project = data["project"]["path_with_namespace"]
    username = data["user"]["username"]
    oid = data["object_attributes"]["id"]
    note = data["object_attributes"]["note"]

    ntype = data["object_attributes"]["noteable_type"]
    if ntype == "Issue":
        iid = data["issue"]["iid"]
        title = data["issue"]["title"]
        item = f"issue #{iid}: {title}"

    elif ntype == "MergeRequest":
        mid = data["merge_request"]["id"]
        title = data["merge_request"]["title"]
        item = f"merge request !{mid}: {title}"

    elif ntype == "Commit":
        cid = data["commit"]["id"][:8]
        item = f"commit {cid}"

    elif ntype == "Snippet":
        sid = data["snippet"]["sid"]
        title = data["snippet"]["title"]
        item = f"snippet {sid}: {title}"

    else:
        item = "?"

    body = f"[{project}] {username} commented on {item}\n"
    body += f"> {note}"

    logging.debug(body)
    mb_url = config["gitlab_gateway"]
    payload = { 'gateway': gateway, 'username': "gitlab", 'text': body }
    r = requests.post(mb_url, json=payload)

@app.route('/srht/<user>/<repo>/<gateway>', methods=['POST'])
def srht_event(user, repo, gateway):
    logging.debug(request.url)
    logging.debug(request.headers)
    payload = request.data
    signature = request.headers["X-Payload-Signature"]
    signature = base64.b64decode(signature)
    nonce = request.headers["X-Payload-Nonce"].encode()
    try:
        srht_public_key.verify(signature, payload + nonce)
        event = request.headers["X-Webhook-Event"]
        if event == "repo:post-update":
            logging.debug(f"SRHT - {event} >> {gateway}")
            # __post_update(user, repo, gateway, request.json)

        elif event == "ticket:create":
            logging.debug(f"SRHT - {event} >> {gateway}")
            # __ticket_create(user, repo, gateway, request.json)

        elif event == "event:create":
            logging.debug(f"SRHT - {event} >> {gateway}")
            # __event_create(user, repo, gateway, request.json)

    except Exception:
        logging.exception('SRHT')

    return '', 200

@app.route('/tars/', methods=['POST'])
def feedbot_cmd():
    token = request.headers['Authorization']
    user_id = request.form['user_id']
    reply = {"username": "tars"}
    cmd_text = request.form['text']
    cmd_help = '''
    The following commands are available:

    xpost-on:   turn on cross posts from mastodon to pnut
    xpost-off:  turn off cross posts from mastodon to pnut
    '''

    if token != "Token " + config['mm_tars_token']:
        logging.debug("INVALID TOKEN")
        abort(401)

    if user_id != config['mm_user_id']:
        logging.debug("UNAUTHORIZED USER")
        abort(403)

    if cmd_text == "xpost-on":
        logging.debug("enable crossposts")
        rs.rpush('TARS', "feedbot_enable")
        reply['text'] = "crossposts will be enabled"

    elif cmd_text == "xpost-off":
        logging.debug("disable crossposts")
        rs.rpush('TARS', "feedbot_disable")
        reply['text'] = "crossposts will be disabled"

    else:
        logging.debug("SHOW HELP?")
        reply['text'] = cmd_help

    return reply, 200