Add webhook_listener.py
This commit is contained in:
commit
04aa35c41d
|
@ -0,0 +1,47 @@
|
|||
from flask import Flask, request, abort
|
||||
import hmac
|
||||
import hashlib
|
||||
import os
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
# Nastav si svůj tajný token – musí odpovídat tomu z Bitbucket webhooku
|
||||
WEBHOOK_SECRET = os.environ.get("BITBUCKET_WEBHOOK_SECRET", "hash")
|
||||
|
||||
def verify_signature(payload, header_signature):
|
||||
"""Porovná HMAC podpis z Bitbucketu s vlastním výpočtem"""
|
||||
if not header_signature:
|
||||
return False
|
||||
try:
|
||||
algo, received_sig = header_signature.split("=")
|
||||
if algo != "sha256":
|
||||
return False
|
||||
computed_sig = hmac.new(
|
||||
key=WEBHOOK_SECRET.encode(),
|
||||
msg=payload,
|
||||
digestmod=hashlib.sha256
|
||||
).hexdigest()
|
||||
# Časově bezpečné porovnání
|
||||
return hmac.compare_digest(computed_sig, received_sig)
|
||||
except Exception as e:
|
||||
print(f"Chyba při ověřování podpisu: {e}")
|
||||
return False
|
||||
|
||||
@app.route("/webhook", methods=["POST"])
|
||||
def webhook():
|
||||
payload = request.data
|
||||
header_signature = request.headers.get("X-Hub-Signature")
|
||||
|
||||
if not verify_signature(payload, header_signature):
|
||||
print("⚠️ Neplatný podpis, požadavek odmítnut.")
|
||||
abort(403)
|
||||
|
||||
event_key = request.headers.get("X-Event-Key")
|
||||
print(f"✅ Validní webhook: {event_key}")
|
||||
print(request.json)
|
||||
|
||||
return "OK", 200
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🚀 Webhook listener běží na http://localhost:5000/webhook")
|
||||
app.run(host="0.0.0.0", port=5000, ssl_context=("/etc/letsencrypt/live/cactus.chalupsky.org/cert.pem", "/etc/letsencrypt/live/cactus.chalupsky.org/privkey.pem"))
|
Loading…
Reference in New Issue