__main__.py
· 2.4 KiB · Python
Orginalformat
import hashlib
import hmac
import json
import os
from ipaddress import ip_address, ip_network
import requests
from dotenv import load_dotenv
from flask import Flask, abort, request
def create_app():
app = Flask(__name__)
load_dotenv()
app.config["GITHUB_SECRET"] = os.getenv("GITHUB_SECRET")
app.config["DISCORD_WEBHOOK"] = os.getenv("DISCORD_WEBHOOK")
return app
app = create_app()
def verify_src_ip(src_ip):
allowed_ips = requests.get("https://api.github.com/meta").json()["hooks"]
return any(src_ip in ip_network(valid_ip) for valid_ip in allowed_ips)
def verify_hmac_hash(data, signature):
github_secret = bytes(app.config["GITHUB_SECRET"], "ascii")
mac = hmac.new(github_secret, msg=data, digestmod=hashlib.sha1)
return hmac.compare_digest("sha1=" + mac.hexdigest(), signature)
@app.route("/github-payload", methods=["POST"])
def github_payload():
src_ip = ip_address(request.headers.get("X-Real-IP", request.access_route[0]))
if not verify_src_ip(src_ip):
app.logger.debug(f"invalid source ip: {src_ip}... aborting request")
abort(403)
signature = request.headers.get("X-Hub-Signature")
data = request.data
if not verify_hmac_hash(data, signature):
app.logger.debug("failed to verify signature... aborting request")
abort(403)
payload = request.get_json()
match event_type := request.headers.get("X-GitHub-Event"):
case "ping":
return json.dumps({"msg": "pong"})
case "issues":
issues_handler(payload)
case _:
app.logger.debug(f"no handler defined for event {event_type}... continuing")
return ("", 200, None)
def issues_handler(payload):
embed = {
"author": {
"name": payload["sender"]["login"],
"icon_url": payload["sender"]["avatar_url"],
},
"title": f"[{payload['repository']['full_name']}] Issue {payload['action'].capitalize()}: #{payload['issue']['number']} {payload['issue']['title']}",
"url": payload["issue"]["html_url"],
"description": payload["issue"].get("body"),
}
data = {
"username": "Github-OpenGist",
"embeds": [embed],
}
send_discord_webhook(app.config["DISCORD_WEBHOOK"], data)
def send_discord_webhook(webhook_url, data):
requests.post(webhook_url, json=data)
if __name__ == "__main__":
app.run(debug=True, port=5000)
| 1 | import hashlib |
| 2 | import hmac |
| 3 | import json |
| 4 | import os |
| 5 | from ipaddress import ip_address, ip_network |
| 6 | |
| 7 | import requests |
| 8 | from dotenv import load_dotenv |
| 9 | from flask import Flask, abort, request |
| 10 | |
| 11 | |
| 12 | def create_app(): |
| 13 | app = Flask(__name__) |
| 14 | load_dotenv() |
| 15 | app.config["GITHUB_SECRET"] = os.getenv("GITHUB_SECRET") |
| 16 | app.config["DISCORD_WEBHOOK"] = os.getenv("DISCORD_WEBHOOK") |
| 17 | |
| 18 | return app |
| 19 | |
| 20 | |
| 21 | app = create_app() |
| 22 | |
| 23 | |
| 24 | def verify_src_ip(src_ip): |
| 25 | allowed_ips = requests.get("https://api.github.com/meta").json()["hooks"] |
| 26 | return any(src_ip in ip_network(valid_ip) for valid_ip in allowed_ips) |
| 27 | |
| 28 | |
| 29 | def verify_hmac_hash(data, signature): |
| 30 | github_secret = bytes(app.config["GITHUB_SECRET"], "ascii") |
| 31 | mac = hmac.new(github_secret, msg=data, digestmod=hashlib.sha1) |
| 32 | return hmac.compare_digest("sha1=" + mac.hexdigest(), signature) |
| 33 | |
| 34 | |
| 35 | @app.route("/github-payload", methods=["POST"]) |
| 36 | def github_payload(): |
| 37 | src_ip = ip_address(request.headers.get("X-Real-IP", request.access_route[0])) |
| 38 | if not verify_src_ip(src_ip): |
| 39 | app.logger.debug(f"invalid source ip: {src_ip}... aborting request") |
| 40 | abort(403) |
| 41 | |
| 42 | signature = request.headers.get("X-Hub-Signature") |
| 43 | data = request.data |
| 44 | if not verify_hmac_hash(data, signature): |
| 45 | app.logger.debug("failed to verify signature... aborting request") |
| 46 | abort(403) |
| 47 | |
| 48 | payload = request.get_json() |
| 49 | match event_type := request.headers.get("X-GitHub-Event"): |
| 50 | case "ping": |
| 51 | return json.dumps({"msg": "pong"}) |
| 52 | case "issues": |
| 53 | issues_handler(payload) |
| 54 | case _: |
| 55 | app.logger.debug(f"no handler defined for event {event_type}... continuing") |
| 56 | return ("", 200, None) |
| 57 | |
| 58 | |
| 59 | def issues_handler(payload): |
| 60 | embed = { |
| 61 | "author": { |
| 62 | "name": payload["sender"]["login"], |
| 63 | "icon_url": payload["sender"]["avatar_url"], |
| 64 | }, |
| 65 | "title": f"[{payload['repository']['full_name']}] Issue {payload['action'].capitalize()}: #{payload['issue']['number']} {payload['issue']['title']}", |
| 66 | "url": payload["issue"]["html_url"], |
| 67 | "description": payload["issue"].get("body"), |
| 68 | } |
| 69 | data = { |
| 70 | "username": "Github-OpenGist", |
| 71 | "embeds": [embed], |
| 72 | } |
| 73 | send_discord_webhook(app.config["DISCORD_WEBHOOK"], data) |
| 74 | |
| 75 | |
| 76 | def send_discord_webhook(webhook_url, data): |
| 77 | requests.post(webhook_url, json=data) |
| 78 | |
| 79 | |
| 80 | if __name__ == "__main__": |
| 81 | app.run(debug=True, port=5000) |
| 82 |