Allow overwriting IP address for specific (sub)domain in dyndns request

api/desecapi/tests/test_dyndns12update.py

@@ -331,6 +331,36 @@ def test_update_multiple_v4(self):
         self.assertIP(ipv4=new_ip)
         self.assertIP(subname="sub", ipv4=new_ip)
 
+    def test_update_multiple_with_overwrite(self):
+        # /nic/update?hostname=sub1.a.io,sub2.a.io,sub3.a.io&myip=1.2.3.4&ipv6=::1&b.io.ipv6=::2
+        new_ip4 = "1.2.3.4"
+        new_ip6 = "::1"
+        new_ip6_overwrite = "::2"
+        domain1 = "sub1." + self.my_domain.name
+        domain2 = "sub2." + self.my_domain.name
+        domain3 = "sub3." + self.my_domain.name
+
+        with self.assertRequests(
+            self.request_pdns_zone_update(self.my_domain.name),
+            self.request_pdns_zone_axfr(self.my_domain.name),
+        ):
+            response = self.client.get(
+                self.reverse("v1:dyndns12update"),
+                {
+                    "hostname": f"{domain1},{domain2},{domain3}",
+                    "myip": new_ip4,
+                    "ipv6": new_ip6,
+                    f"{domain2.lower()}.ipv6": new_ip6_overwrite,
+                },
+            )
+
+        self.assertStatus(response, status.HTTP_200_OK)
+        self.assertEqual(response.data, "good")
+
+        self.assertIP(subname="sub1", ipv4=new_ip4, ipv6=new_ip6)
+        self.assertIP(subname="sub2", ipv4=new_ip4, ipv6=new_ip6_overwrite)
+        self.assertIP(subname="sub3", ipv4=new_ip4, ipv6=new_ip6)
+
     def test_update_multiple_username_param(self):
         # /nic/update?username=a.io,sub.a.io&myip=1.2.3.4
         new_ip = "1.2.3.4"
@@ -404,6 +434,36 @@ def test_update_multiple_with_subnet(self):
         self.assertIP(subname="sub1", ipv4="10.1.0.1")
         self.assertIP(subname="sub2", ipv4="10.1.0.2")
 
+    def test_update_multiple_with_subnet_and_ip_override(self):
+        # /nic/update?hostname=a.io,b.io&myip=10.1.0.0/16&a.io=192.168.1.1
+        domain1 = "sub1." + self.my_domain.name
+        domain2 = "sub2." + self.my_domain.name
+        self.create_rr_set(
+            self.my_domain, ["10.0.0.1"], subname="sub1", type="A", ttl=60
+        )
+        self.create_rr_set(
+            self.my_domain, ["10.0.0.2"], subname="sub2", type="A", ttl=60
+        )
+
+        with self.assertRequests(
+            self.request_pdns_zone_update(self.my_domain.name),
+            self.request_pdns_zone_axfr(self.my_domain.name),
+        ):
+            response = self.client.get(
+                self.reverse("v1:dyndns12update"),
+                {
+                    "hostname": f"{domain1},{domain2}",
+                    "myip": "10.1.0.0/16",
+                    f"{domain1.lower()}.myip": "192.168.1.1",
+                },
+            )
+
+        self.assertStatus(response, status.HTTP_200_OK)
+        self.assertEqual(response.data, "good")
+
+        self.assertIP(subname="sub1", ipv4="192.168.1.1")
+        self.assertIP(subname="sub2", ipv4="10.1.0.2")
+
     def test_update_multiple_with_one_being_already_up_to_date(self):
         # /nic/update?hostname=a.io,sub.a.io&myip=1.2.3.4
         new_ip = "1.2.3.4"
@@ -447,6 +507,20 @@ def test_update_same_domain_twice(self):
 
         self.assertIP(ipv4=new_ip)
 
+    def test_update_overwrite_with_invalid_subnet(self):
+        # /nic/update?hostname=a.io&a.io.myip=1.2.3.4/64
+        domain1 = self.create_domain(owner=self.owner).name
+
+        with self.assertRequests():
+            response = self.client.get(
+                self.reverse("v1:dyndns12update"),
+                {"hostname": f"{domain1}", f"{domain1.lower()}.myip": "1.2.3.4/64"},
+            )
+
+        self.assertContains(
+            response, "invalid subnet", status_code=status.HTTP_400_BAD_REQUEST
+        )
+
     def test_update_multiple_with_invalid_subnet(self):
         # /nic/update?hostname=sub1.a.io,sub2.a.io&myip=1.2.3.4/64
         domain1 = "sub1." + self.my_domain.name

api/desecapi/views/dyndns.py

@@ -62,6 +62,9 @@ class DynDNS12UpdateView(generics.GenericAPIView):
     serializer_class = RRsetSerializer
     throttle_scope = "dyndns"
 
+    IPV4_PARAMS = ["myip", "myipv4", "ip"]
+    IPV6_PARAMS = ["myipv6", "ipv6", "myip", "ip"]
+
     @property
     def throttle_scope_bucket(self):
         return self.domain.name
@@ -70,11 +73,17 @@ def _find_action(self, param_keys, separator) -> UpdateAction:
         """
         Parses the request for IP parameters and determines the appropriate update action.
 
-        This method checks a given list of parameter keys in the request URL. It handles
-        plain IP addresses, comma-separated lists of IPs, the "preserve" keyword, and
-        subnet notation (e.g., "10.0.0.0/24"). It also uses the client's remote IP
+        This method checks a given list of parameter keys in the request URL. The keys can
+        be global (e.g. ['myip']) or scoped to a specific hostname (e.g. ['example.com.myip']).
+
+        It handles plain IP addresses, comma-separated lists of IPs, the "preserve" keyword,
+        and subnet notation (e.g., "10.0.0.0/24"). It also uses the client's remote IP
         as a fallback.
 
+        Args:
+            param_keys (list): A list of parameter keys to check for in the request.
+            separator (str): The IP address separator ("." for IPv4, ":" for IPv6).
+
         Returns:
             UpdateAction: A dataclass instance (`SetIPs`, `UpdateWithSubnet`, or `PreserveIPs`)
             representing the action to be taken.
@@ -257,10 +266,17 @@ def get(self, request, *args, **kwargs) -> Response:
         for rrset in instances:
             subname_records[rrset.subname].extend(rrset.records.all())
 
-        actions = {
-            "A": self._find_action(["myip", "myipv4", "ip"], separator="."),
-            "AAAA": self._find_action(["myipv6", "ipv6", "myip", "ip"], separator=":"),
-        }
+        actions = {}
+        for subname in self.subnames:
+            fqdn = ".".join(filter(None, (subname, self.domain.name)))
+            actions[("A", subname)] = self._find_action(
+                [f"{fqdn}.{p}" for p in self.IPV4_PARAMS] + self.IPV4_PARAMS,
+                separator=".",
+            )
+            actions[("AAAA", subname)] = self._find_action(
+                [f"{fqdn}.{p}" for p in self.IPV6_PARAMS] + self.IPV6_PARAMS,
+                separator=":",
+            )
 
         data = [
             {
@@ -269,8 +285,7 @@ def get(self, request, *args, **kwargs) -> Response:
                 "ttl": 60,
                 "records": records,
             }
-            for subname in self.subnames
-            for type_, action in actions.items()
+            for (type_, subname), action in actions.items()
             if (records := self._get_records(subname_records[subname], action))
             is not None
         ]

docs/dyndns/update-api.rst

@@ -159,6 +159,31 @@ query parameter, such as ``ipv6=2a01:a:b:c::1/64``.
     Note that using an encrypted connection (TLS) does *not* protect against
     this attack, as TLS does not protect the IP address.
 
+Per-Hostname IP Addresses
+-------------------------
+When updating multiple hostnames at once, it is possible to specify different
+IP information for each hostname. This is done by prefixing the IP parameter
+with the hostname it applies to.
+
+For example, to set a global IPv4 address for all hostnames but a specific
+IPv6 address for ``host2.example.com``, you would send a request like this::
+
+  ?hostname=host1.example.com,host2.example.com
+  &myip=1.2.3.4
+  &host2.example.com.myipv6=2001:db8::1
+
+This would set the IPv4 address of ``host1.example.com`` and ``host2.example.com``
+to ``1.2.3.4``. Additionally, there would be an IPv6 address with the value
+``2001:db8::1`` set for ``host2.example.com``.
+
+For each hostname, the server will look for IP parameters in the following order:
+
+1. A parameter prefixed with that specific hostname (e.g., ``host2.example.com.myipv6``).
+2. A global, non-prefixed parameter (e.g., ``myip``).
+3. The remote IP address of the client making the request (if applicable).
+
+This allows for flexible and powerful combinations of updates in a single API call.
+
 Update Response
 ```````````````
 If successful, the server will return a response with status ``200 OK`` and