68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
import json
|
|
import os
|
|
import urllib.parse
|
|
import urllib.request
|
|
from flask import Flask, request, jsonify
|
|
|
|
app = Flask(__name__)
|
|
|
|
KUMA_BASE_URL = os.getenv("KUMA_BASE_URL", "http://localhost:3001/api/push")
|
|
MAPPING_FILE = os.getenv("MAPPING_FILE", "mapping.json")
|
|
# Shared secret for authentication
|
|
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "super-secret-key-change-me")
|
|
|
|
def load_mapping():
|
|
if not os.path.exists(MAPPING_FILE):
|
|
return {}
|
|
try:
|
|
with open(MAPPING_FILE, "r") as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
print(f"[ERROR] Failed to parse {MAPPING_FILE}: {e}")
|
|
return {}
|
|
|
|
@app.route("/webhook/ppp", methods=["POST"])
|
|
def ppp_webhook():
|
|
# 1. Validate Secret Token in Header
|
|
auth_header = request.headers.get("X-API-Key")
|
|
if not auth_header or auth_header != WEBHOOK_SECRET:
|
|
print(f"[WARN] Unauthorized webhook attempt from IP: {request.remote_addr}")
|
|
return jsonify({"status": "error", "message": "Unauthorized"}), 401
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
|
|
user = data.get("user")
|
|
status = data.get("status")
|
|
ip = data.get("ip", "0.0.0.0")
|
|
msg = data.get("msg", "")
|
|
ping = data.get("ping", 0)
|
|
|
|
if not user or not status:
|
|
return jsonify({"status": "error", "message": "Missing required fields"}), 400
|
|
|
|
mapping = load_mapping()
|
|
token = mapping.get(user)
|
|
|
|
if not token:
|
|
print(f"[WARN] No token found for user: '{user}'")
|
|
return jsonify({"status": "ignored", "message": f"User '{user}' not mapped."}), 404
|
|
|
|
kuma_status = "up" if str(status).lower() == "up" else "down"
|
|
if not msg:
|
|
msg = f"PPP status is {kuma_status.upper()} for user '{user}' ({ip})"
|
|
|
|
encoded_msg = urllib.parse.quote(msg)
|
|
kuma_url = f"{KUMA_BASE_URL}/{token}?status={kuma_status}&msg={encoded_msg}&ping={ping}"
|
|
|
|
try:
|
|
req = urllib.request.Request(kuma_url, headers={"User-Agent": "MikroTik-Bridge/1.0"})
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
print(f"[INFO] Forwarded push for '{user}' -> status: {kuma_status}")
|
|
return jsonify({"status": "success"}), 200
|
|
except Exception as e:
|
|
print(f"[ERROR] Failed to contact Uptime Kuma: {e}")
|
|
return jsonify({"status": "error", "message": str(e)}), 500
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=3000)
|