Fix: lookup more specific DNS names directly (localstack#9815)

localstack/dns/server.py

@@ -93,6 +93,15 @@
 psutil_cache = TTLCache(maxsize=100, ttl=10)
 
 
+# TODO: update route53 provider to use this util
+def normalise_dns_name(name: DNSLabel | str) -> str:
+    name = str(name)
+    if not name.endswith("."):
+        return f"{name}."
+
+    return name
+
+
 @cached(cache=psutil_cache)
 def list_network_interface_details() -> dict[str, list[snicaddr]]:
     return psutil.net_if_addrs()
@@ -451,7 +460,7 @@ def _resolve_name_from_zones(
         converter = RecordConverter(request, client_address)
 
         # check for direct (not regex based) response
-        zone = self.zones.get(request.q.qname)
+        zone = self.zones.get(normalise_dns_name(request.q.qname))
         if zone is not None:
             for zone_records in zone:
                 rr = converter.to_record(zone_records).try_rr(request.q)
@@ -522,12 +531,14 @@ def flag(f):
 
     def add_host(self, name: str, record: NameRecord):
         LOG.debug("Adding host %s with record %s", name, record)
+        name = normalise_dns_name(name)
         with self.lock:
             self.zones.setdefault(name, [])
             self.zones[name].append(record)
 
     def delete_host(self, name: str, record: NameRecord):
         LOG.debug("Deleting host %s with record %s", name, record)
+        name = normalise_dns_name(name)
         with self.lock:
             if not self.zones.get(name):
                 raise ValueError("Could not find entry %s for name %s in zones", record, name)

tests/unit/test_dns_server.py

@@ -177,6 +177,16 @@ def test_dns_server_wildcard_matching_with_skip(self, dns_server, query_dns):
         assert answer.answer
         assert "122.122.122.122" in answer.to_text()
 
+    def test_dns_server_specific_name_overrides_wildcard(self, dns_server, query_dns):
+        dns_server.add_host("*.example.org", TargetRecord("1.2.3.4", RecordType.A))
+        dns_server.add_host("foo.example.org", TargetRecord("5.6.7.8", RecordType.A))
+
+        answer = query_dns("foo.example.org", "A")
+
+        assert answer.answer
+        assert "5.6.7.8" in answer.to_text()
+        assert "1.2.3.4" not in answer.to_text()
+
     def test_redirect_to_localstack_lifecycle(self, dns_server, query_dns):
         """Test adding records pointing to LS at all times"""
         dns_server.add_host_pointing_to_localstack("*.example.org")