315 lines
9.9 KiB
Python
315 lines
9.9 KiB
Python
import logging
|
|
import base64
|
|
import hmac
|
|
import hashlib
|
|
import requests
|
|
import yaml
|
|
import os
|
|
|
|
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')
|
|
if SNAP_USER_DATA is None:
|
|
filename = 'config.yaml'
|
|
log_file = 'webhook.log'
|
|
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)
|
|
|
|
srht_public_key = Ed25519PublicKey.from_public_bytes(
|
|
base64.b64decode(config['srht_payload_sig']))
|
|
|
|
|
|
@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"
|
|
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)
|
|
|
|
@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}ed issue #{oid}: {title}\n"
|
|
body += f"{description}"
|
|
|
|
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"
|
|
body += f"{description}"
|
|
|
|
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
|