117 lines
4.1 KiB
Python
117 lines
4.1 KiB
Python
from rest_framework import serializers
|
|
|
|
from common.modules import available_modules, module_off_message
|
|
from webhooks import events, ssrf
|
|
from webhooks.models import WebhookDelivery, WebhookEndpoint
|
|
|
|
|
|
class WebhookEndpointSerializer(serializers.ModelSerializer):
|
|
"""An endpoint as an admin sees it. The secret is only ever a hint here;
|
|
the create and rotate views add the full value to their own response."""
|
|
|
|
secret_hint = serializers.CharField(read_only=True)
|
|
# Who answers for the endpoint: whoever added it, or the admin who last
|
|
# turned it back on after a pause. Never taken from a request body.
|
|
created_by = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = WebhookEndpoint
|
|
fields = (
|
|
"id",
|
|
"url",
|
|
"description",
|
|
"events",
|
|
"format",
|
|
"is_active",
|
|
"disabled_reason",
|
|
"secret_hint",
|
|
"created_by",
|
|
"created_at",
|
|
"updated_at",
|
|
)
|
|
read_only_fields = ("id", "disabled_reason", "created_at", "updated_at")
|
|
|
|
def get_created_by(self, obj):
|
|
user = obj.created_by
|
|
if user is None:
|
|
return None
|
|
return {"id": str(user.id), "name": user.name, "email": user.email}
|
|
|
|
def validate(self, attrs):
|
|
# The secret is minted by the server. Refused loudly rather than
|
|
# dropped, so a caller who thinks they chose one learns they did not.
|
|
if "secret" in self.initial_data:
|
|
raise serializers.ValidationError(
|
|
{"secret": "The signing secret is generated by the server."}
|
|
)
|
|
return attrs
|
|
|
|
def validate_url(self, value):
|
|
value = value.strip()
|
|
try:
|
|
ssrf.validate_url(value)
|
|
except ssrf.UnsafeDestination as exc:
|
|
raise serializers.ValidationError(str(exc)) from None
|
|
return value
|
|
|
|
def validate_description(self, value):
|
|
return value.strip()
|
|
|
|
def validate_events(self, value):
|
|
org = self.context.get("org")
|
|
off = {}
|
|
kept = []
|
|
if org is not None:
|
|
available = available_modules(org)
|
|
off = {
|
|
name: events.event_module(name)
|
|
for name in events.EVENT_NAMES
|
|
if events.event_module(name) not in (None, *available)
|
|
}
|
|
stored = self.instance.events if self.instance is not None else []
|
|
kept = [e for e in stored if e in off]
|
|
if isinstance(value, list):
|
|
named = [v for v in value if isinstance(v, str) and v in off]
|
|
for name in dict.fromkeys(named):
|
|
if name not in kept:
|
|
raise serializers.ValidationError(
|
|
module_off_message(org, off[name])
|
|
)
|
|
value = value + [e for e in kept if e not in value]
|
|
if not isinstance(value, list) or not value:
|
|
raise serializers.ValidationError("Choose at least one event.")
|
|
unknown = [
|
|
v for v in value if not isinstance(v, str) or v not in events.EVENT_NAMES
|
|
]
|
|
if unknown:
|
|
raise serializers.ValidationError(
|
|
f"Unknown event: {', '.join(str(v) for v in unknown[:5])}."
|
|
)
|
|
# Catalogue order, each once.
|
|
return [name for name in events.EVENT_NAMES if name in value]
|
|
|
|
def update(self, instance, validated_data):
|
|
# Turning an endpoint back on clears the reason it was turned off.
|
|
if validated_data.get("is_active") and not instance.is_active:
|
|
instance.disabled_reason = ""
|
|
return super().update(instance, validated_data)
|
|
|
|
|
|
class WebhookDeliverySerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = WebhookDelivery
|
|
fields = (
|
|
"id",
|
|
"endpoint",
|
|
"event",
|
|
"event_id",
|
|
"status",
|
|
"attempts",
|
|
"response_status",
|
|
"error",
|
|
"next_attempt_at",
|
|
"last_attempt_at",
|
|
"created_at",
|
|
"payload",
|
|
)
|
|
read_only_fields = fields
|