233 lines
9 KiB
Python
233 lines
9 KiB
Python
from collections import defaultdict
|
|
|
|
from rest_framework import serializers
|
|
from rest_framework.validators import UniqueTogetherValidator
|
|
|
|
from apps.api.serializers.labels import LabelsSerializerMixin
|
|
from apps.webhooks.models import Webhook, WebhookResponse
|
|
from apps.webhooks.models.webhook import PUBLIC_WEBHOOK_HTTP_METHODS, WEBHOOK_FIELD_PLACEHOLDER
|
|
from apps.webhooks.presets.preset_options import WebhookPresetOptions
|
|
from common.api_helpers.custom_fields import IntegrationFilteredByOrganizationField, TeamPrimaryKeyRelatedField
|
|
from common.api_helpers.utils import CurrentOrganizationDefault, 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",
|
|
"event_data",
|
|
]
|
|
|
|
|
|
class WebhookSerializer(LabelsSerializerMixin, serializers.ModelSerializer):
|
|
id = serializers.CharField(read_only=True, source="public_primary_key")
|
|
organization = serializers.HiddenField(default=CurrentOrganizationDefault())
|
|
team = TeamPrimaryKeyRelatedField(allow_null=True, required=False)
|
|
user = serializers.HiddenField(default=CurrentUserDefault())
|
|
forward_all = serializers.BooleanField(allow_null=True, required=False)
|
|
last_response_log = serializers.SerializerMethodField()
|
|
trigger_type = serializers.CharField(allow_null=True)
|
|
trigger_type_name = serializers.SerializerMethodField()
|
|
integration_filter = IntegrationFilteredByOrganizationField(
|
|
source="filtered_integrations", many=True, required=False
|
|
)
|
|
|
|
PREFETCH_RELATED = ["labels", "labels__key", "labels__value"]
|
|
|
|
class Meta:
|
|
model = Webhook
|
|
fields = [
|
|
"id",
|
|
"name",
|
|
"is_webhook_enabled",
|
|
"is_legacy",
|
|
"team",
|
|
"user",
|
|
"username",
|
|
"password",
|
|
"authorization_header",
|
|
"organization",
|
|
"trigger_template",
|
|
"headers",
|
|
"url",
|
|
"data",
|
|
"forward_all",
|
|
"http_method",
|
|
"trigger_type",
|
|
"trigger_type_name",
|
|
"last_response_log",
|
|
"integration_filter",
|
|
"preset",
|
|
"labels",
|
|
]
|
|
|
|
validators = [UniqueTogetherValidator(queryset=Webhook.objects.all(), fields=["name", "organization"])]
|
|
|
|
def create(self, validated_data):
|
|
organization = self.context["request"].auth.organization
|
|
labels = validated_data.pop("labels", None)
|
|
|
|
instance = super().create(validated_data)
|
|
self.update_labels_association_if_needed(labels, instance, organization)
|
|
return instance
|
|
|
|
def update(self, instance, validated_data):
|
|
labels = validated_data.pop("labels", None)
|
|
organization = self.context["request"].auth.organization
|
|
self.update_labels_association_if_needed(labels, instance, organization)
|
|
return super().update(instance, validated_data)
|
|
|
|
def to_representation(self, instance):
|
|
result = super().to_representation(instance)
|
|
if instance.password:
|
|
result["password"] = WEBHOOK_FIELD_PLACEHOLDER
|
|
if instance.authorization_header:
|
|
result["authorization_header"] = WEBHOOK_FIELD_PLACEHOLDER
|
|
return result
|
|
|
|
def to_internal_value(self, data):
|
|
webhook = self.instance
|
|
|
|
# Some fields are conditionally required, add none values for missing required fields
|
|
if webhook and webhook.preset and "preset" not in data:
|
|
data["preset"] = webhook.preset
|
|
for key in ["url", "http_method", "trigger_type"]:
|
|
if key not in data:
|
|
if self.instance:
|
|
data[key] = getattr(self.instance, key)
|
|
else:
|
|
data[key] = None
|
|
|
|
# If webhook is being copied instance won't exist to copy values from
|
|
if not webhook and "id" in data:
|
|
webhook = Webhook.objects.get(
|
|
public_primary_key=data["id"], organization=self.context["request"].auth.organization
|
|
)
|
|
|
|
if data.get("password") == WEBHOOK_FIELD_PLACEHOLDER:
|
|
data["password"] = webhook.password
|
|
if data.get("authorization_header") == WEBHOOK_FIELD_PLACEHOLDER:
|
|
data["authorization_header"] = webhook.authorization_header
|
|
return super().to_internal_value(data)
|
|
|
|
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 self.is_field_controlled("url"):
|
|
return url
|
|
|
|
if not url:
|
|
raise serializers.ValidationError(detail="This field is required.")
|
|
return self._validate_template_field(url)
|
|
|
|
def validate_http_method(self, http_method):
|
|
if self.is_field_controlled("http_method"):
|
|
return http_method
|
|
|
|
if http_method not in PUBLIC_WEBHOOK_HTTP_METHODS:
|
|
raise serializers.ValidationError(detail=f"This field must be one of {PUBLIC_WEBHOOK_HTTP_METHODS}.")
|
|
return http_method
|
|
|
|
def validate_trigger_type(self, trigger_type):
|
|
if self.is_field_controlled("trigger_type"):
|
|
return trigger_type
|
|
|
|
if not trigger_type or int(trigger_type) not in Webhook.ALL_TRIGGER_TYPES:
|
|
raise serializers.ValidationError(detail="This field is required.")
|
|
return trigger_type
|
|
|
|
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 validate_preset(self, preset):
|
|
if self.instance and self.instance.preset != preset:
|
|
raise serializers.ValidationError(detail="This field once set cannot be modified.")
|
|
|
|
if preset:
|
|
if preset not in WebhookPresetOptions.WEBHOOK_PRESETS:
|
|
raise serializers.ValidationError(detail=f"{preset} is not a valid preset id.")
|
|
|
|
preset_metadata = WebhookPresetOptions.WEBHOOK_PRESETS[preset].metadata
|
|
for controlled_field in preset_metadata.controlled_fields:
|
|
if controlled_field in self.initial_data:
|
|
if self.instance:
|
|
if self.initial_data[controlled_field] is not None and self.initial_data[
|
|
controlled_field
|
|
] != getattr(self.instance, controlled_field):
|
|
raise serializers.ValidationError(
|
|
detail=f"{controlled_field} is controlled by preset, cannot update"
|
|
)
|
|
elif self.initial_data[controlled_field] is not None:
|
|
raise serializers.ValidationError(
|
|
detail=f"{controlled_field} is controlled by preset, cannot create"
|
|
)
|
|
|
|
return preset
|
|
|
|
def get_last_response_log(self, obj):
|
|
return WebhookResponseSerializer(obj.responses.last()).data
|
|
|
|
def get_trigger_type_name(self, obj):
|
|
trigger_type_name = ""
|
|
if obj.trigger_type is not None:
|
|
trigger_type_name = Webhook.TRIGGER_TYPES[int(obj.trigger_type)][1]
|
|
return trigger_type_name
|
|
|
|
def is_field_controlled(self, field_name):
|
|
if self.instance:
|
|
if not self.instance.preset:
|
|
return False
|
|
elif "preset" not in self.initial_data:
|
|
return False
|
|
|
|
preset_id = self.instance.preset if self.instance else self.initial_data["preset"]
|
|
if preset_id:
|
|
if preset_id not in WebhookPresetOptions.WEBHOOK_PRESETS:
|
|
raise serializers.ValidationError(detail=f"unknown preset {preset_id} referenced")
|
|
|
|
preset = WebhookPresetOptions.WEBHOOK_PRESETS[preset_id]
|
|
if field_name not in preset.metadata.controlled_fields:
|
|
return False
|
|
return True
|
|
|
|
|
|
class WebhookFastSerializer(serializers.ModelSerializer):
|
|
id = serializers.CharField(read_only=True, source="public_primary_key")
|
|
|
|
class Meta:
|
|
model = Webhook
|
|
fields = ["id", "name"]
|