this repo has no description
support the challenges from zulip
+43
-1
+1
netdata_zulip_bot/config.py
+1
netdata_zulip_bot/config.py
+3
netdata_zulip_bot/main.py
+3
netdata_zulip_bot/main.py
···
40
40
SERVER_HOST=0.0.0.0
41
41
SERVER_PORT=8080
42
42
43
+
# Netdata webhook challenge secret (required for webhook verification)
44
+
SERVER_CHALLENGE_SECRET=your-challenge-secret-here
45
+
43
46
# Optional: Override Zulip stream (default: netdata-alerts)
44
47
# ZULIP_STREAM=custom-alerts-stream
45
48
"""
+1
netdata_zulip_bot/models.py
+1
netdata_zulip_bot/models.py
+38
-1
netdata_zulip_bot/server.py
+38
-1
netdata_zulip_bot/server.py
···
1
1
"""FastAPI webhook server for receiving Netdata notifications."""
2
2
3
+
import base64
4
+
import hashlib
5
+
import hmac
3
6
from typing import Dict, Any
4
7
5
8
import structlog
6
9
import uvicorn
7
-
from fastapi import FastAPI, HTTPException, Request, status
10
+
from fastapi import FastAPI, HTTPException, Request, status, Query
8
11
from fastapi.responses import JSONResponse
9
12
10
13
from .formatter import ZulipMessageFormatter
···
45
48
async def health_check():
46
49
"""Health check endpoint."""
47
50
return {"status": "healthy", "service": "netdata-zulip-bot"}
51
+
52
+
@self.app.get("/webhook/netdata")
53
+
async def netdata_webhook_challenge(crc_token: str = Query(...)):
54
+
"""Handle Netdata Cloud webhook challenge for verification."""
55
+
if not self.server_config.challenge_secret:
56
+
logger.error("Challenge secret not configured")
57
+
raise HTTPException(
58
+
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
59
+
detail="Challenge secret not configured"
60
+
)
61
+
62
+
try:
63
+
# Create HMAC SHA-256 hash from crc_token and challenge secret
64
+
token_bytes = crc_token.encode('ascii')
65
+
secret_bytes = self.server_config.challenge_secret.encode('utf-8')
66
+
67
+
sha256_hash = hmac.new(
68
+
secret_bytes,
69
+
msg=token_bytes,
70
+
digestmod=hashlib.sha256
71
+
).digest()
72
+
73
+
# Create response with base64 encoded hash
74
+
response_token = 'sha256=' + base64.b64encode(sha256_hash).decode('ascii')
75
+
76
+
logger.info("Responding to Netdata challenge", crc_token=crc_token[:16] + "...")
77
+
return {"response_token": response_token}
78
+
79
+
except Exception as e:
80
+
logger.error("Failed to process challenge", error=str(e))
81
+
raise HTTPException(
82
+
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
83
+
detail="Failed to process challenge"
84
+
)
48
85
49
86
@self.app.post("/webhook/netdata")
50
87
async def netdata_webhook(request: Request):