feat(api): Allow overwriting IP address for specific (sub)domain in multi-update

Author: molikuner

api/desecapi/tests/test_dyndns12update.py

@@ -317,6 +317,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"
@@ -390,6 +420,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"
@@ -433,6 +493,18 @@ 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.assertStatus(response, 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,22 +62,35 @@ 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
 
-    def _find_action(self, param_keys, separator) -> UpdateAction:
+    def _find_action(
+        self, param_keys, separator, use_remote_ip_fallback=False
+    ) -> 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
-        as a fallback.
+        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").
+
+        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).
+            use_remote_ip_fallback (bool): If True, uses the client's remote IP as a
+                fallback if no other parameters are found.
 
         Returns:
-            UpdateAction: A dataclass instance (`SetIPs`, `UpdateWithSubnet`, or `PreserveIPs`)
-            representing the action to be taken.
+            UpdateAction or None: A dataclass instance (`SetIPs`, `UpdateWithSubnet`, or
+            `PreserveIPs`) representing the action to be taken, or None if no relevant
+            parameter was found and the fallback to client IP is disabled.
         """
         # Check URL parameters
         for param_key in param_keys:
@@ -119,12 +132,13 @@ def _find_action(self, param_keys, separator) -> UpdateAction:
                     return SetIPs(ips=params)
 
         # Check remote IP address
-        client_ip = self.request.META.get("REMOTE_ADDR")
-        if separator in client_ip:
-            return SetIPs(ips=[client_ip])
+        if use_remote_ip_fallback:
+            client_ip = self.request.META.get("REMOTE_ADDR")
+            if separator in client_ip:
+                return SetIPs(ips=[client_ip])
 
         # give up
-        return SetIPs(ips=[])
+        return None
 
     @staticmethod
     def _sanitize_qnames(qnames_str) -> set[str]:
@@ -245,12 +259,26 @@ def get(self, request, *args, **kwargs) -> Response:
             grouped_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=":"),
+            "A": self._find_action(
+                self.IPV4_PARAMS, separator=".", use_remote_ip_fallback=True
+            )
+            or SetIPs(ips=[]),
+            "AAAA": self._find_action(
+                self.IPV6_PARAMS, separator=":", use_remote_ip_fallback=True
+            )
+            or SetIPs(ips=[]),
         }
 
         data = []
-        for subname in self.subnames:
+        for qname, subname in zip(self.qnames, self.subnames):
+            scoped_ipv4_params = [f"{qname}.{p}" for p in self.IPV4_PARAMS]
+            scoped_ipv6_params = [f"{qname}.{p}" for p in self.IPV6_PARAMS]
+            domain_actions = {
+                "A": self._find_action(scoped_ipv4_params, separator=".")
+                or actions["A"],
+                "AAAA": self._find_action(scoped_ipv6_params, separator=":")
+                or actions["AAAA"],
+            }
             subname_records = grouped_records.get(subname, [])
 
             data += [
@@ -260,7 +288,7 @@ def get(self, request, *args, **kwargs) -> Response:
                     "ttl": 60,
                     "records": records,
                 }
-                for type_, action in actions.items()
+                for type_, action in domain_actions.items()
                 if (records := self._get_records(subname_records, 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