desec-stack/api/desecapi/serializers/domains.py

import dns.name
import dns.zone
from rest_framework import serializers

from api import settings
from desecapi.models import Domain, RR_SET_TYPES_AUTOMATIC
from desecapi.validators import ReadOnlyOnUpdateValidator

from .records import RRsetSerializer


class DomainSerializer(serializers.ModelSerializer):
    default_error_messages = {
        **serializers.Serializer.default_error_messages,
        "name_unavailable": "This domain name conflicts with an existing zone, or is disallowed by policy.",
    }
    zonefile = serializers.CharField(write_only=True, required=False, allow_blank=True)

    class Meta:
        model = Domain
        fields = (
            "created",
            "published",
            "name",
            "keys",
            "minimum_ttl",
            "touched",
            "zonefile",
        )
        read_only_fields = (
            "published",
            "minimum_ttl",
        )
        extra_kwargs = {
            "name": {"trim_whitespace": False},
        }

    def __init__(self, *args, include_keys=False, **kwargs):
        self.include_keys = include_keys
        self.import_zone = None
        super().__init__(*args, **kwargs)

    def get_fields(self):
        fields = super().get_fields()
        if not self.include_keys:
            fields.pop("keys")
        fields["name"].validators.append(ReadOnlyOnUpdateValidator())
        return fields

    def validate_name(self, value):
        if not Domain(name=value, owner=self.context["request"].user).is_registrable():
            raise serializers.ValidationError(
                self.default_error_messages["name_unavailable"], code="name_unavailable"
            )
        return value

    def parse_zonefile(self, domain_name: str, zonefile: str):
        try:
            self.import_zone = dns.zone.from_text(
                zonefile,
                origin=dns.name.from_text(domain_name),
                allow_include=False,
                check_origin=False,
                relativize=False,
            )
        except dns.zonefile.CNAMEAndOtherData:
            raise serializers.ValidationError(
                {
                    "zonefile": [
                        "No other records with the same name are allowed alongside a CNAME record."
                    ]
                }
            )
        except ValueError as e:
            if "has non-origin SOA" in str(e):
                raise serializers.ValidationError(
                    {
                        "zonefile": [
                            f"Zonefile includes an SOA record for a name different from {domain_name}."
                        ]
                    }
                )
            raise e
        except dns.exception.SyntaxError as e:
            try:
                line = str(e).split(":")[1]
                raise serializers.ValidationError(
                    {"zonefile": [f"Zonefile contains syntax error in line {line}."]}
                )
            except IndexError:
                raise serializers.ValidationError(
                    {"zonefile": [f"Could not parse zonefile: {str(e)}"]}
                )

    def validate(self, attrs):
        if attrs.get("zonefile") is not None:
            self.parse_zonefile(attrs.get("name"), attrs.pop("zonefile"))
        return super().validate(attrs)

    def create(self, validated_data):
        # save domain
        if (
            "minimum_ttl" not in validated_data
            and Domain(name=validated_data["name"]).is_locally_registrable
        ):
            validated_data.update(minimum_ttl=60)
        domain: Domain = super().create(validated_data)

        # save RRsets if zonefile was given
        nodes = getattr(self.import_zone, "nodes", None)
        if nodes:
            zone_name = dns.name.from_text(validated_data["name"])
            min_ttl, max_ttl = domain.minimum_ttl, settings.MAXIMUM_TTL
            data = [
                {
                    "type": dns.rdatatype.to_text(rrset.rdtype),
                    "ttl": max(min_ttl, min(max_ttl, rrset.ttl)),
                    "subname": (owner_name - zone_name).to_text()
                    if owner_name - zone_name != dns.name.empty
                    else "",
                    "records": [rr.to_text() for rr in rrset],
                }
                for owner_name, node in nodes.items()
                for rrset in node.rdatasets
                if (
                    dns.rdatatype.to_text(rrset.rdtype)
                    not in (
                        RR_SET_TYPES_AUTOMATIC
                        | {  # do not import automatically managed record types
                            "CDS",
                            "CDNSKEY",
                            "DNSKEY",
                        }  # do not import these, as this would likely be unexpected
                    )
                    and not (
                        owner_name - zone_name == dns.name.empty
                        and rrset.rdtype == dns.rdatatype.NS
                    )  # ignore apex NS
                )
            ]

            rrset_list_serializer = RRsetSerializer(
                data=data, context=dict(domain=domain), many=True
            )
            # The following line raises if data passed validation by dnspython during zone file parsing,
            # but is rejected by validation in RRsetSerializer. See also
            # test_create_domain_zonefile_import_validation
            try:
                rrset_list_serializer.is_valid(raise_exception=True)
            except serializers.ValidationError as e:
                if isinstance(e.detail, serializers.ReturnList):
                    # match the order of error messages with the RRsets provided to the
                    # serializer to make sense to the client
                    def fqdn(idx):
                        return (data[idx]["subname"] + "." + domain.name).lstrip(".")

                    raise serializers.ValidationError(
                        {
                            "zonefile": [
                                f"{fqdn(idx)}/{data[idx]['type']}: {err}"
                                for idx, d in enumerate(e.detail)
                                for _, errs in d.items()
                                for err in errs
                            ]
                        }
                    )

                raise e

            rrset_list_serializer.save()

        return domain