607 lines
24 KiB
Python
607 lines
24 KiB
Python
"""Serializers for the authenticated web form management API.
|
|
|
|
Two rules run through all of this:
|
|
|
|
* anything the server derives is `read_only` or popped. `org`, `created_by` and
|
|
`is_published` are never taken from a request body. Publishing in particular
|
|
has its own endpoint because it validates the source state and the form's
|
|
shape, and a writable `is_published` would be a second door onto the same
|
|
transition with none of those checks.
|
|
* the field list is replaced wholesale, with `order` assigned from list
|
|
position. A client that sends its own `order` cannot reorder a form by lying
|
|
about it, and a drag-reorder is one atomic request rather than N.
|
|
"""
|
|
|
|
from urllib.parse import urlparse
|
|
|
|
from django.db import transaction
|
|
from rest_framework import serializers
|
|
|
|
from common.links import api_url
|
|
from common.models import CustomFieldDefinition, Profile, Tags
|
|
from common.modules import (
|
|
available_modules,
|
|
module_is_available,
|
|
module_off_message,
|
|
org_of,
|
|
)
|
|
from webforms.constants import LEAD_FIELD_VALUES
|
|
from webforms.dynamic_serializer import CUSTOM_FIELD_TARGET
|
|
from webforms.models import WebForm, WebFormField, WebFormSubmission
|
|
|
|
# The built-in row source each form target accepts, besides "custom".
|
|
BUILTIN_SOURCE = {
|
|
WebForm.TARGET_LEAD: WebFormField.SOURCE_LEAD,
|
|
WebForm.TARGET_TICKET: WebFormField.SOURCE_TICKET,
|
|
}
|
|
|
|
ALLOWED_URL_SCHEMES = ("http", "https")
|
|
|
|
|
|
NO_SUCH_CUSTOM_FIELD = "No such custom field for this form."
|
|
|
|
|
|
class WebFormFieldSerializer(serializers.ModelSerializer):
|
|
custom_field = serializers.PrimaryKeyRelatedField(
|
|
queryset=CustomFieldDefinition.objects.all(),
|
|
required=False,
|
|
allow_null=True,
|
|
)
|
|
|
|
class Meta:
|
|
model = WebFormField
|
|
fields = (
|
|
"id",
|
|
"source",
|
|
"lead_field",
|
|
"ticket_field",
|
|
"custom_field",
|
|
"label",
|
|
"placeholder",
|
|
"is_required",
|
|
)
|
|
# `order` is absent from `fields` entirely rather than declared
|
|
# read-only: the parent assigns it from list position, so a client
|
|
# sending one is ignored and cannot reorder a form by lying about it.
|
|
read_only_fields = ("id",)
|
|
|
|
def validate_lead_field(self, value):
|
|
if value and value not in LEAD_FIELD_VALUES:
|
|
raise serializers.ValidationError(
|
|
f"'{value}' is not a field a web form may collect."
|
|
)
|
|
return value
|
|
|
|
def validate(self, attrs):
|
|
"""Exactly one target, and the one `source` names.
|
|
|
|
Whether the row suits the FORM (a lead row on a ticket form, a custom
|
|
field on the wrong model) is checked by the parent, which is the only
|
|
place that knows the form's target.
|
|
"""
|
|
named = {
|
|
WebFormField.SOURCE_LEAD: attrs.get("lead_field") or "",
|
|
WebFormField.SOURCE_TICKET: attrs.get("ticket_field") or "",
|
|
WebFormField.SOURCE_CUSTOM: attrs.get("custom_field"),
|
|
}
|
|
if sum(1 for value in named.values() if value) != 1:
|
|
raise serializers.ValidationError(
|
|
"A field must name exactly one of a lead field, a ticket field "
|
|
"or a custom field."
|
|
)
|
|
if not attrs.get("label"):
|
|
# Checked here rather than left to the model: under the view's
|
|
# partial update DRF skips a missing required key, and the write
|
|
# below would then fail on it with a 500.
|
|
raise serializers.ValidationError({"label": "Give the field a label."})
|
|
if not named.get(attrs.get("source")):
|
|
raise serializers.ValidationError(
|
|
{"source": "This does not match the field the row names."}
|
|
)
|
|
|
|
custom_field = named[WebFormField.SOURCE_CUSTOM]
|
|
if custom_field is not None and custom_field.org_id != self.context["org"].id:
|
|
# Deliberately the same message as the wrong-model case in the
|
|
# parent: a caller must not be able to probe which custom field ids
|
|
# exist in another org by comparing error text.
|
|
raise serializers.ValidationError({"custom_field": NO_SUCH_CUSTOM_FIELD})
|
|
return attrs
|
|
|
|
|
|
class WebFormListSerializer(serializers.ModelSerializer):
|
|
submission_count = serializers.IntegerField(read_only=True)
|
|
field_count = serializers.IntegerField(read_only=True)
|
|
|
|
class Meta:
|
|
model = WebForm
|
|
fields = (
|
|
"id",
|
|
"name",
|
|
"target",
|
|
"is_published",
|
|
"created_at",
|
|
"submission_count",
|
|
"field_count",
|
|
)
|
|
|
|
|
|
class WebFormDetailSerializer(serializers.ModelSerializer):
|
|
fields = WebFormFieldSerializer(many=True, required=False)
|
|
embed_html = serializers.SerializerMethodField()
|
|
embed_js = serializers.SerializerMethodField()
|
|
# Whether a secret is stored, never which one. Turnstile fails closed, so a
|
|
# form with a provider and no secret refuses every submission; without this
|
|
# flag a client cannot tell that state from a working one, and cannot tell
|
|
# an empty secret box ("none stored") from a hidden one ("stored, and
|
|
# sending blank would wipe it").
|
|
has_captcha_secret = serializers.SerializerMethodField()
|
|
# Who `assign_to` names and whether they are still active. The picker is
|
|
# built from active members only, so without this a client cannot offer a
|
|
# deactivated assignee as an option, and a select with no matching option
|
|
# clears the field on the next save.
|
|
assign_to_details = serializers.SerializerMethodField()
|
|
# The same shape for every rotation member, active or not, for the same
|
|
# reason: a stored member deactivated since has no option in the picker.
|
|
rotation_members_details = serializers.SerializerMethodField()
|
|
rotation_last_assigned_details = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = WebForm
|
|
fields = (
|
|
"id",
|
|
"name",
|
|
"target",
|
|
"is_published",
|
|
"allowed_origins",
|
|
"submit_button_label",
|
|
"success_mode",
|
|
"success_message",
|
|
"redirect_url",
|
|
"assignment_mode",
|
|
"assign_to",
|
|
"rotation_members",
|
|
"rotation_cap",
|
|
"rotation_last_assigned",
|
|
"notify_profiles",
|
|
"lead_source",
|
|
"tags",
|
|
"ticket_priority",
|
|
"ticket_type",
|
|
"captcha_provider",
|
|
"captcha_site_key",
|
|
"captcha_secret",
|
|
"reject_disposable_email",
|
|
"created_at",
|
|
"fields",
|
|
"embed_html",
|
|
"embed_js",
|
|
"has_captcha_secret",
|
|
"assign_to_details",
|
|
"rotation_members_details",
|
|
"rotation_last_assigned_details",
|
|
)
|
|
# `rotation_last_assigned` is the rotation's cursor. Only the
|
|
# submission path moves it, under a row lock; a writable one would let
|
|
# an admin's stale form overwrite whose turn it is.
|
|
read_only_fields = (
|
|
"id",
|
|
"created_at",
|
|
"is_published",
|
|
"rotation_last_assigned",
|
|
)
|
|
extra_kwargs = {
|
|
# The org pastes this into its own Cloudflare dashboard; we only
|
|
# ever send it to Cloudflare. Never read back.
|
|
"captcha_secret": {"write_only": True},
|
|
}
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
"""Narrow every relation to the caller's own org.
|
|
|
|
`assign_to`, `rotation_members`, `notify_profiles` and `tags` are plain
|
|
model relations, so DRF builds each with a queryset of every row in the
|
|
table. Left alone,
|
|
an admin could point a form at another tenant's profile, and that is
|
|
not confined to a wrong column: `webforms/tasks.py` mails each
|
|
submission to `notify_profiles -> user.email`, so the form would send
|
|
this org's inbound leads to a stranger in a different one.
|
|
|
|
RLS is not a fallback here. `profile` is deliberately outside
|
|
ORG_SCOPED_TABLES because it is read before any tenant context exists,
|
|
so there is no policy behind this and the filter IS the control.
|
|
|
|
With no org in context (schema generation, and nothing else) every
|
|
queryset stays empty, so the failure direction is refusal rather than
|
|
an unscoped lookup.
|
|
"""
|
|
super().__init__(*args, **kwargs)
|
|
org = self.context.get("org")
|
|
if org is None:
|
|
request = self.context.get("request")
|
|
org = getattr(getattr(request, "profile", None), "org", None)
|
|
self._scope("assign_to", Profile.objects.filter(org=org) if org else None)
|
|
self._scope(
|
|
"rotation_members", Profile.objects.filter(org=org) if org else None
|
|
)
|
|
self._scope("notify_profiles", Profile.objects.filter(org=org) if org else None)
|
|
self._scope("tags", Tags.objects.filter(org=org) if org else None)
|
|
|
|
def _scope(self, name, queryset):
|
|
"""Point one relation's lookup at `queryset`, or at nothing.
|
|
|
|
A `many=True` relation validates through its `child_relation`, so
|
|
assigning to the wrapper would leave the real lookup untouched and the
|
|
guard would read as present while doing nothing.
|
|
"""
|
|
field = self.fields[name]
|
|
target = getattr(field, "child_relation", field)
|
|
target.queryset = queryset if queryset is not None else target.queryset.none()
|
|
|
|
def validate_assign_to(self, value):
|
|
"""A deactivated member cannot be made the form's assignee.
|
|
|
|
Submission ignores one anyway (`webforms.service.active_assignee`), so
|
|
accepting it would save a setting that silently does nothing.
|
|
|
|
Keeping the one already stored is allowed. Both clients resend it on
|
|
every save, so refusing it would leave a form whose assignee was
|
|
deactivated later unsaveable until the admin picked somebody else, and
|
|
the stored value is already ignored.
|
|
"""
|
|
stored = getattr(self.instance, "assign_to_id", None)
|
|
if value is not None and not value.is_active and value.id != stored:
|
|
raise serializers.ValidationError(
|
|
"This user is deactivated. Choose an active member."
|
|
)
|
|
return value
|
|
|
|
def validate_rotation_members(self, value):
|
|
"""No deactivated member may be ADDED to a rotation.
|
|
|
|
The rotation skips one anyway (`webforms.service.rotation_assignee`),
|
|
so adding them would save a turn that silently never comes. Members
|
|
already stored are kept, as `validate_assign_to` keeps a stored
|
|
assignee: both clients resend the whole list on every save.
|
|
"""
|
|
stored = (
|
|
set(self.instance.rotation_members.values_list("id", flat=True))
|
|
if self.instance is not None
|
|
else set()
|
|
)
|
|
for profile in value:
|
|
if not profile.is_active and profile.id not in stored:
|
|
raise serializers.ValidationError(
|
|
"This user is deactivated. Choose an active member."
|
|
)
|
|
return value
|
|
|
|
def _validate_assignment(self, attrs):
|
|
"""Rotation needs a lead form and somebody to rotate between.
|
|
|
|
Both read the value the save will leave behind, so a request that
|
|
changes only the target, or only the members, cannot slip a form into
|
|
a state its own two fields contradict.
|
|
"""
|
|
mode = attrs.get(
|
|
"assignment_mode",
|
|
getattr(self.instance, "assignment_mode", WebForm.ASSIGN_PERSON),
|
|
)
|
|
if mode != WebForm.ASSIGN_ROTATION:
|
|
return
|
|
target = attrs.get(
|
|
"target", getattr(self.instance, "target", WebForm.TARGET_LEAD)
|
|
)
|
|
if target != WebForm.TARGET_LEAD:
|
|
raise serializers.ValidationError(
|
|
{
|
|
"assignment_mode": (
|
|
"Rotation is for lead forms only. Tickets from a "
|
|
"ticket form are assigned by your routing rules."
|
|
)
|
|
}
|
|
)
|
|
if "rotation_members" in attrs:
|
|
has_members = bool(attrs["rotation_members"])
|
|
else:
|
|
has_members = (
|
|
self.instance is not None and self.instance.rotation_members.exists()
|
|
)
|
|
if not has_members:
|
|
raise serializers.ValidationError(
|
|
{"rotation_members": "Choose at least one member to rotate between."}
|
|
)
|
|
|
|
# ---- embed snippets -------------------------------------------------
|
|
#
|
|
# Built here rather than in a client because they need the API's own base
|
|
# URL. A browser knows the frontend's origin, not this one, and a relative
|
|
# URL pasted onto a customer's site would point at the customer's server.
|
|
# From `DOMAIN_NAME`, not the request: the web app asks from its own
|
|
# server, whose view of this host is internal (`common.links.api_url`).
|
|
|
|
def _public_base(self, obj):
|
|
return api_url(f"/api/public/forms/{obj.org_id}/{obj.id}/")
|
|
|
|
def get_embed_html(self, obj):
|
|
return (
|
|
f'<iframe src="{self._public_base(obj)}embed/" '
|
|
f'style="width:100%;border:0" height="500" '
|
|
f'title="{obj.name}"></iframe>'
|
|
)
|
|
|
|
def get_has_captcha_secret(self, obj):
|
|
return bool(obj.captcha_secret)
|
|
|
|
@staticmethod
|
|
def _profile_details(profile):
|
|
if profile is None:
|
|
return None
|
|
return {
|
|
"id": str(profile.id),
|
|
"email": profile.user.email,
|
|
"name": profile.user.name,
|
|
"is_active": profile.is_active,
|
|
}
|
|
|
|
def get_assign_to_details(self, obj):
|
|
return self._profile_details(obj.assign_to)
|
|
|
|
def get_rotation_members_details(self, obj):
|
|
members = obj.rotation_members.select_related("user").order_by("id")
|
|
return [self._profile_details(profile) for profile in members]
|
|
|
|
def get_rotation_last_assigned_details(self, obj):
|
|
return self._profile_details(obj.rotation_last_assigned)
|
|
|
|
def get_embed_js(self, obj):
|
|
return (
|
|
f'<div id="bottlecrm-webform-{obj.id}"></div>\n'
|
|
f'<script src="{self._public_base(obj)}embed.js" async></script>'
|
|
)
|
|
|
|
# ---- validation -----------------------------------------------------
|
|
|
|
def validate_redirect_url(self, value):
|
|
"""Only http and https.
|
|
|
|
This value is handed to the embed, which navigates to it. A
|
|
`javascript:` or `data:` payload here would be stored XSS on the
|
|
customer's own site, authored through our admin API.
|
|
"""
|
|
if not value:
|
|
return value
|
|
parsed = urlparse(value)
|
|
if parsed.scheme.lower() not in ALLOWED_URL_SCHEMES:
|
|
raise serializers.ValidationError(
|
|
"Enter an http or https URL. Other schemes are not allowed."
|
|
)
|
|
return value
|
|
|
|
def validate_allowed_origins(self, value):
|
|
if not isinstance(value, list):
|
|
raise serializers.ValidationError(
|
|
'Provide a list of origins, for example ["https://example.com"].'
|
|
)
|
|
cleaned = []
|
|
for entry in value:
|
|
if not isinstance(entry, str) or not entry.strip():
|
|
raise serializers.ValidationError(
|
|
"Each origin must be a non-empty string."
|
|
)
|
|
entry = entry.strip()
|
|
parsed = urlparse(entry)
|
|
if parsed.scheme.lower() not in ALLOWED_URL_SCHEMES or not parsed.netloc:
|
|
raise serializers.ValidationError(
|
|
f"'{entry}' is not a valid origin. Use a scheme and a host, "
|
|
f"for example https://example.com."
|
|
)
|
|
if parsed.path not in ("", "/") or parsed.query or parsed.fragment:
|
|
# An Origin header never carries a path, so an entry with one
|
|
# can never match. It would look configured while doing
|
|
# nothing, which is worse than being rejected.
|
|
raise serializers.ValidationError(
|
|
f"'{entry}' has a path. An origin is only a scheme, host "
|
|
f"and optional port."
|
|
)
|
|
cleaned.append(f"{parsed.scheme}://{parsed.netloc}")
|
|
return cleaned
|
|
|
|
def _validate_target(self, attrs):
|
|
"""The target, and every field row, agree with each other.
|
|
|
|
A target change is refused once the form has submissions: its history
|
|
would then mix leads and tickets, and every stored payload was keyed
|
|
for the old target. Before that it is allowed, but the field list has
|
|
to move with it, because rows written for one target mean nothing to
|
|
the other and the embed would render inputs that write nowhere.
|
|
|
|
A target newly chosen, on create or by a change, is refused while its
|
|
module is off. A form already on that target keeps taking other edits.
|
|
"""
|
|
stored = getattr(self.instance, "target", None)
|
|
target = attrs.get("target", stored or WebForm.TARGET_LEAD)
|
|
changed = stored is not None and target != stored
|
|
|
|
if changed and self.instance.legacy_api_setting_id:
|
|
# The legacy web-to-lead endpoint writes through this form and
|
|
# promises its callers a lead.
|
|
raise serializers.ValidationError(
|
|
{
|
|
"target": (
|
|
"This form backs a legacy web-to-lead API key, so it "
|
|
"can only create leads."
|
|
)
|
|
}
|
|
)
|
|
if changed and self.instance.submissions.exists():
|
|
raise serializers.ValidationError(
|
|
{
|
|
"target": (
|
|
"This form already has submissions, so what it creates "
|
|
"cannot change. Create a new form instead."
|
|
)
|
|
}
|
|
)
|
|
org = self.context["org"]
|
|
module = WebForm.TARGET_MODULES[target]
|
|
if (stored is None or changed) and not module_is_available(org, module):
|
|
raise serializers.ValidationError(
|
|
{"target": module_off_message(org, module)}
|
|
)
|
|
|
|
rows = attrs.get("fields")
|
|
if rows is None:
|
|
if changed and self.instance.fields.exists():
|
|
raise serializers.ValidationError(
|
|
{
|
|
"fields": (
|
|
"Send the new field list with the target change. "
|
|
"The current fields were chosen for the old target."
|
|
)
|
|
}
|
|
)
|
|
return
|
|
|
|
allowed = {BUILTIN_SOURCE[target], WebFormField.SOURCE_CUSTOM}
|
|
custom_model = CUSTOM_FIELD_TARGET[target]
|
|
for row in rows:
|
|
if row["source"] not in allowed:
|
|
raise serializers.ValidationError(
|
|
{
|
|
"fields": (
|
|
f"A {target} form cannot collect a {row['source']} field."
|
|
)
|
|
}
|
|
)
|
|
custom_field = row.get("custom_field")
|
|
if custom_field is not None and custom_field.target_model != custom_model:
|
|
raise serializers.ValidationError(
|
|
{"fields": {"custom_field": NO_SUCH_CUSTOM_FIELD}}
|
|
)
|
|
|
|
def validate(self, attrs):
|
|
self._validate_target(attrs)
|
|
self._validate_assignment(attrs)
|
|
provider = attrs.get(
|
|
"captcha_provider", getattr(self.instance, "captcha_provider", "")
|
|
)
|
|
if provider == WebForm.CAPTCHA_TURNSTILE:
|
|
site_key = attrs.get(
|
|
"captcha_site_key", getattr(self.instance, "captcha_site_key", "")
|
|
)
|
|
secret = attrs.get(
|
|
"captcha_secret", getattr(self.instance, "captcha_secret", "")
|
|
)
|
|
if not site_key or not secret:
|
|
raise serializers.ValidationError(
|
|
"Turnstile needs both a site key and a secret. Verification "
|
|
"fails closed, so a half-configured captcha rejects every "
|
|
"submission."
|
|
)
|
|
return attrs
|
|
|
|
# ---- writes ---------------------------------------------------------
|
|
|
|
def _write_fields(self, instance, rows):
|
|
"""Replace the form's field list with `rows`, in order.
|
|
|
|
Delete-then-recreate rather than a diff. The list is short, the write
|
|
is inside the caller's transaction, and a diff would have to reconcile
|
|
reordering against the two unique constraints for no practical gain.
|
|
"""
|
|
instance.fields.all().delete()
|
|
for index, row in enumerate(rows):
|
|
WebFormField.objects.create(
|
|
form=instance,
|
|
org=instance.org,
|
|
order=index,
|
|
source=row["source"],
|
|
lead_field=row.get("lead_field") or "",
|
|
ticket_field=row.get("ticket_field") or "",
|
|
custom_field=row.get("custom_field"),
|
|
label=row["label"],
|
|
placeholder=row.get("placeholder", ""),
|
|
is_required=row.get("is_required", False),
|
|
)
|
|
|
|
@transaction.atomic
|
|
def create(self, validated_data):
|
|
rows = validated_data.pop("fields", None)
|
|
tags = validated_data.pop("tags", None)
|
|
notify = validated_data.pop("notify_profiles", None)
|
|
members = validated_data.pop("rotation_members", None)
|
|
instance = WebForm.objects.create(
|
|
org=self.context["org"],
|
|
created_by=self.context["request"].profile.user,
|
|
**validated_data,
|
|
)
|
|
if tags is not None:
|
|
instance.tags.set(tags)
|
|
if notify is not None:
|
|
instance.notify_profiles.set(notify)
|
|
if members is not None:
|
|
instance.rotation_members.set(members)
|
|
if rows is not None:
|
|
self._write_fields(instance, rows)
|
|
return instance
|
|
|
|
@transaction.atomic
|
|
def update(self, instance, validated_data):
|
|
rows = validated_data.pop("fields", None)
|
|
tags = validated_data.pop("tags", None)
|
|
notify = validated_data.pop("notify_profiles", None)
|
|
members = validated_data.pop("rotation_members", None)
|
|
for attr, value in validated_data.items():
|
|
setattr(instance, attr, value)
|
|
instance.save()
|
|
if tags is not None:
|
|
instance.tags.set(tags)
|
|
if notify is not None:
|
|
instance.notify_profiles.set(notify)
|
|
if members is not None:
|
|
instance.rotation_members.set(members)
|
|
if rows is not None:
|
|
self._write_fields(instance, rows)
|
|
return instance
|
|
|
|
|
|
class WebFormSubmissionSerializer(serializers.ModelSerializer):
|
|
lead_name = serializers.SerializerMethodField()
|
|
case_name = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = WebFormSubmission
|
|
fields = (
|
|
"id",
|
|
"created_at",
|
|
"status",
|
|
"payload",
|
|
"lead",
|
|
"lead_name",
|
|
"case",
|
|
"case_name",
|
|
"submitted_ip",
|
|
"referer",
|
|
)
|
|
# `reject_reason` is absent on purpose. It is internal triage detail
|
|
# and naming the control that caught a bot is how the next bot gets
|
|
# past it. Agents see the status, which is what they act on.
|
|
|
|
def to_representation(self, instance):
|
|
"""The lead or ticket a submission made is null while its module is off."""
|
|
data = super().to_representation(instance)
|
|
available = available_modules(org_of(instance))
|
|
for field, key in (("lead", "leads"), ("case", "tickets")):
|
|
if key not in available:
|
|
data[field] = data[f"{field}_name"] = None
|
|
return data
|
|
|
|
def get_lead_name(self, obj):
|
|
if obj.lead is None:
|
|
return None
|
|
return str(obj.lead)
|
|
|
|
def get_case_name(self, obj):
|
|
if obj.case is None:
|
|
return None
|
|
return obj.case.name
|