Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 77 additions & 20 deletions src/zeroconf/_dns.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
from ._exceptions import AbstractMethodException
from ._utils.net import _is_v6_address
from ._utils.time import current_time_millis
from .const import _CLASS_MASK, _CLASS_UNIQUE, _TYPE_ANY
from .const import _CLASS_MASK, _CLASS_UNIQUE, _CLASSES, _TYPE_ANY, _TYPES

_LEN_BYTE = 1
_LEN_SHORT = 2
Expand All @@ -28,6 +28,20 @@
_float = float
_int = int


def _type_label(type_: int) -> str:
return _TYPES.get(type_, f"unknown-type-{type_}")


def _class_label(class_: int) -> str:
return _CLASSES.get(class_, f"unknown-class-{class_}")


def _format_display(kind: str, fields: list[tuple[str, str]]) -> str:
body = " ".join([f"{key}={value}" if key else value for key, value in fields])
return f"<{kind} {body}>"


if TYPE_CHECKING:
from ._protocol.incoming import DNSIncoming
from ._protocol.outgoing import DNSOutgoing
Expand Down Expand Up @@ -58,6 +72,39 @@ def __eq__(self, other: Any) -> bool:
"""Equal when key, type and class match."""
return isinstance(other, DNSEntry) and self._dns_entry_matches(other)

@property
def type_label(self) -> str:
"""Human readable label for the record type."""
return _type_label(self.type)

@property
def class_label(self) -> str:
"""Human readable label for the record class."""
return _class_label(self.class_)

def _display_fields(self) -> list[tuple[str, str]]:
fields = [("name", self.name), ("type", self.type_label), ("class", self.class_label)]
if self.unique:
fields.append(("", "unique"))
return fields

@staticmethod
def get_class_(class_: int) -> str:
"""Compatibility alias for the class_label property."""
return _class_label(class_)

@staticmethod
def get_type(t: int) -> str:
"""Compatibility alias for the type_label property."""
return _type_label(t)

def entry_to_string(self, hdr: str, other: bytes | str | None) -> str:
"""Compatibility alias rendering through the display formatter."""
fields = self._display_fields()
if other is not None:
fields.append(("data", str(other)))
return _format_display(hdr, fields)

def _dns_entry_matches(self, other: DNSEntry) -> bool:
return self.key == other.key and self.type == other.type and self.class_ == other.class_

Expand Down Expand Up @@ -86,12 +133,8 @@ def __hash__(self) -> int:
return self._hash

def __repr__(self) -> str:
return "{}[question,{},{},{}]".format(
self.get_type(self.type),
"QU" if self.unicast else "QM",
self.get_class_(self.class_),
self.name,
)
mode = "QU" if self.unicast else "QM"
return _format_display("DNSQuestion", [("mode", mode), *self._display_fields()])

def answered_by(self, rec: DNSRecord) -> bool:
return self.class_ == rec.class_ and self.type in (rec.type, _TYPE_ANY) and self.name == rec.name
Expand Down Expand Up @@ -188,6 +231,17 @@ def _set_created_ttl(self, created: _float, ttl: _int) -> None:
self.created = created
self.ttl = ttl

def _display_fields(self) -> list[tuple[str, str]]:
remaining = int(self.get_remaining_ttl(current_time_millis()))
return [*DNSEntry._display_fields(self), ("ttl", f"{self.ttl} ({remaining} remaining)")]

def _repr_with(self, *details: tuple[str, str]) -> str:
return _format_display(type(self).__name__, [*self._display_fields(), *details])

def to_string(self, other: bytes | str) -> str:
"""Compatibility alias rendering through the display formatter."""
return self._repr_with(("data", str(other)))

def _suppressed_by_answer(self, answer: DNSRecord) -> bool:
"""True when the answer matches this record with at least half its TTL left."""
return self == answer and self.ttl / 2 < answer.ttl
Expand Down Expand Up @@ -220,14 +274,13 @@ def __hash__(self) -> int:

def __repr__(self) -> str:
try:
return self.to_string(
socket.inet_ntop(
socket.AF_INET6 if _is_v6_address(self.address) else socket.AF_INET,
self.address,
)
data = socket.inet_ntop(
socket.AF_INET6 if _is_v6_address(self.address) else socket.AF_INET,
self.address,
)
except (ValueError, OSError):
return self.to_string(str(self.address))
data = str(self.address)
return self._repr_with(("data", data))

def write(self, out: DNSOutgoing) -> None:
out.write_string(self.address)
Expand Down Expand Up @@ -281,6 +334,9 @@ def __hash__(self) -> int:
"""Hash to compare like DNSHinfo."""
return self._hash

def __repr__(self) -> str:
return self._repr_with(("cpu", self.cpu), ("os", self.os))

def write(self, out: DNSOutgoing) -> None:
"""Write the rdata to an outgoing packet."""
out.write_character_string(self.cpu.encode("utf-8"))
Expand Down Expand Up @@ -325,9 +381,8 @@ def __hash__(self) -> int:
return self._hash

def __repr__(self) -> str:
return self.to_string(
self.next_name + "," + "|".join([self.get_type(type_) for type_ in self.rdtypes])
)
covered = "|".join([_type_label(t) for t in self.rdtypes])
return self._repr_with(("next_name", self.next_name), ("covers", covered))

def write(self, out: DNSOutgoing) -> None:
"""Write the rdata to an outgoing packet."""
Expand Down Expand Up @@ -397,7 +452,7 @@ def __hash__(self) -> int:
return self._hash

def __repr__(self) -> str:
return self.to_string(self.alias)
return self._repr_with(("alias", self.alias))

@property
def max_size_compressed(self) -> int:
Expand Down Expand Up @@ -453,6 +508,9 @@ def __hash__(self) -> int:
"""Hash to compare like DNSService."""
return self._hash

def __repr__(self) -> str:
return self._repr_with(("server", self.server), ("port", str(self.port)))

def write(self, out: DNSOutgoing) -> None:
out.write_short(self.priority)
out.write_short(self.weight)
Expand Down Expand Up @@ -514,9 +572,8 @@ def __hash__(self) -> int:
return self._hash

def __repr__(self) -> str:
if len(self.text) > 16:
return self.to_string(f"{len(self.text)} bytes")
return self.to_string(self.text)
data = f"{len(self.text)} bytes" if len(self.text) > 16 else str(self.text)
return self._repr_with(("data", data))

def write(self, out: DNSOutgoing) -> None:
out.write_string(self.text)
Expand Down
2 changes: 1 addition & 1 deletion tests/services/test_info.py
Original file line number Diff line number Diff line change
Expand Up @@ -1090,7 +1090,7 @@ async def test_invalid_a_addresses(caplog):
info = make_service_info(type_, registration_name, properties=desc, server=host, addresses=[])
info.load_from_cache(aiozc.zeroconf)
assert not info.addresses
assert "Encountered invalid address while processing record" in caplog.text
assert "Encountered invalid address while processing" in caplog.text

await aiozc.async_close()

Expand Down
22 changes: 19 additions & 3 deletions tests/test_dns.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,22 @@ def teardown_module():
log.setLevel(original_logging_level)


def test_display_compat_shims():
"""The legacy display helpers stay callable and route through the new formatter."""
record = r.DNSPointer(
"_http._tcp.local.", const._TYPE_PTR, const._CLASS_IN, const._DNS_OTHER_TTL, "demo._http._tcp.local."
)
assert r.DNSEntry.get_type(const._TYPE_PTR) == "ptr"
assert r.DNSEntry.get_type(4242) == "unknown-type-4242"
assert r.DNSEntry.get_class_(const._CLASS_IN) == "in"
assert r.DNSEntry.get_class_(4242) == "unknown-class-4242"
assert record.entry_to_string("hdr", None).startswith("<hdr ")
assert "data=extra" in record.entry_to_string("hdr", "extra")
assert "data=payload" in record.to_string("payload")
assert record.type_label == "ptr"
assert record.class_label == "in"


class TestDunder(unittest.TestCase):
def test_dns_text_repr(self):
# There was an issue on Python 3 that prevented DNSText's repr
Expand All @@ -53,7 +69,7 @@ def test_dns_pointer_repr(self):
@unittest.skipIf(os.environ.get("SKIP_IPV6"), "IPv6 tests disabled")
def test_dns_address_repr(self):
address = r.DNSAddress("irrelevant", const._TYPE_SOA, const._CLASS_IN, 1, b"a")
assert repr(address).endswith("b'a'")
assert "data=b'a'" in repr(address)

address_ipv4 = r.DNSAddress(
"irrelevant",
Expand All @@ -62,7 +78,7 @@ def test_dns_address_repr(self):
1,
socket.inet_pton(socket.AF_INET, "127.0.0.1"),
)
assert repr(address_ipv4).endswith("127.0.0.1")
assert "data=127.0.0.1" in repr(address_ipv4)

address_ipv6 = r.DNSAddress(
"irrelevant",
Expand All @@ -71,7 +87,7 @@ def test_dns_address_repr(self):
1,
socket.inet_pton(socket.AF_INET6, "::1"),
)
assert repr(address_ipv6).endswith("::1")
assert "data=::1" in repr(address_ipv6)

def test_dns_question_repr(self):
question = r.DNSQuestion("irrelevant", const._TYPE_SRV, const._CLASS_IN | const._CLASS_UNIQUE)
Expand Down
6 changes: 3 additions & 3 deletions tests/test_protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -754,7 +754,7 @@ def test_qm_packet_parser():
)
parsed = DNSIncoming(qm_packet)
assert parsed.questions[0].unicast is False
assert ",QM," in str(parsed.questions[0])
assert "mode=QM" in str(parsed.questions[0])


# MDNS 115 Standard query 0x0000 PTR _companion-link._tcp.local, "QU" question OPT
Expand All @@ -766,7 +766,7 @@ def test_qu_packet_parser():
)
parsed = DNSIncoming(qu_packet)
assert parsed.questions[0].unicast is True
assert ",QU," in str(parsed.questions[0])
assert "mode=QU" in str(parsed.questions[0])


def test_parse_packet_with_nsec_record():
Expand All @@ -780,7 +780,7 @@ def test_parse_packet_with_nsec_record():
)
parsed = DNSIncoming(nsec_packet)
nsec_record = cast(r.DNSNsec, parsed.answers()[3])
assert "nsec," in str(nsec_record)
assert "type=nsec" in str(nsec_record)
assert nsec_record.rdtypes == [16, 33]
assert nsec_record.next_name == "MyHome54 (2)._meshcop._udp.local."

Expand Down
Loading