d40d3a62b8
Track all webhook responses data, and allow using this between alertgroup-related webhooks (e.g. use firing webhook response data when templating the acknowledge webhook request data). NOTE: dropping the table is not backwards compatible but the feature is not enabled (and in any case it would drop log entries only used for status display)
118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
from collections import defaultdict
|
|
|
|
from rest_framework import serializers
|
|
from rest_framework.validators import UniqueTogetherValidator
|
|
|
|
from apps.webhooks.models import Webhook, WebhookResponse
|
|
from common.api_helpers.custom_fields import TeamPrimaryKeyRelatedField
|
|
from common.api_helpers.utils import CurrentOrganizationDefault, CurrentTeamDefault, CurrentUserDefault
|
|
from common.jinja_templater import apply_jinja_template
|
|
from common.jinja_templater.apply_jinja_template import JinjaTemplateError, JinjaTemplateWarning
|
|
|
|
|
|
class WebhookResponseSerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = WebhookResponse
|
|
fields = [
|
|
"timestamp",
|
|
"url",
|
|
"request_trigger",
|
|
"request_headers",
|
|
"request_data",
|
|
"status_code",
|
|
"content",
|
|
]
|
|
|
|
|
|
class WebhookSerializer(serializers.ModelSerializer):
|
|
id = serializers.CharField(read_only=True, source="public_primary_key")
|
|
organization = serializers.HiddenField(default=CurrentOrganizationDefault())
|
|
team = TeamPrimaryKeyRelatedField(allow_null=True, default=CurrentTeamDefault())
|
|
user = serializers.HiddenField(default=CurrentUserDefault())
|
|
last_run = serializers.SerializerMethodField()
|
|
trigger_type = serializers.CharField(required=True)
|
|
forward_all = serializers.BooleanField(allow_null=True, required=False)
|
|
last_response_log = serializers.SerializerMethodField()
|
|
trigger_type_name = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = Webhook
|
|
fields = [
|
|
"id",
|
|
"name",
|
|
"team",
|
|
"data",
|
|
"user",
|
|
"username",
|
|
"password",
|
|
"authorization_header",
|
|
"organization",
|
|
"trigger_template",
|
|
"headers",
|
|
"url",
|
|
"data",
|
|
"forward_all",
|
|
"http_method",
|
|
"trigger_type",
|
|
"trigger_type_name",
|
|
"last_run",
|
|
"last_response_log",
|
|
]
|
|
extra_kwargs = {
|
|
"authorization_header": {"write_only": True},
|
|
"name": {"required": True, "allow_null": False, "allow_blank": False},
|
|
"password": {"write_only": True},
|
|
"url": {"required": True, "allow_null": False, "allow_blank": False},
|
|
}
|
|
|
|
validators = [UniqueTogetherValidator(queryset=Webhook.objects.all(), fields=["name", "organization"])]
|
|
|
|
def _validate_template_field(self, template):
|
|
try:
|
|
apply_jinja_template(template, alert_payload=defaultdict(str), alert_group_id="alert_group_1")
|
|
except JinjaTemplateError as e:
|
|
raise serializers.ValidationError(e.fallback_message)
|
|
except JinjaTemplateWarning:
|
|
# Suppress render exceptions since we do not have a representative payload to test with
|
|
pass
|
|
return template
|
|
|
|
def validate_trigger_template(self, trigger_template):
|
|
if not trigger_template:
|
|
return None
|
|
return self._validate_template_field(trigger_template)
|
|
|
|
def validate_headers(self, headers):
|
|
if not headers:
|
|
return None
|
|
return self._validate_template_field(headers)
|
|
|
|
def validate_url(self, url):
|
|
if not url:
|
|
return None
|
|
return self._validate_template_field(url)
|
|
|
|
def validate_data(self, data):
|
|
if not data:
|
|
return None
|
|
return self._validate_template_field(data)
|
|
|
|
def validate_forward_all(self, data):
|
|
if data is None:
|
|
return False
|
|
return data
|
|
|
|
def get_last_run(self, obj):
|
|
last_run = ""
|
|
if last_log := obj.responses.all().last():
|
|
last_run = last_log.timestamp.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
return last_run
|
|
|
|
def get_last_response_log(self, obj):
|
|
return WebhookResponseSerializer(obj.responses.all().last()).data
|
|
|
|
def get_trigger_type_name(self, obj):
|
|
trigger_type_name = ""
|
|
if obj.trigger_type:
|
|
trigger_type_name = Webhook.TRIGGER_TYPES[int(obj.trigger_type)][1]
|
|
return trigger_type_name
|