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
9 changes: 8 additions & 1 deletion src/zeroconf/_services/info.pxd
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,14 @@ cdef class ServiceInfo(RecordUpdateListener):
@cython.locals(cache=DNSCache)
cpdef bint _load_from_cache(self, object zc, double now)

@cython.locals(length="unsigned char", index="unsigned int", key_value=bytes, key_sep_value=tuple)
@cython.locals(
length="unsigned char",
index="unsigned int",
key_value=bytes,
key=bytes,
sep=bytes,
value=bytes,
)
cdef void _unpack_text_into_properties(self)

@cython.locals(k=bytes, v=bytes)
Expand Down
8 changes: 5 additions & 3 deletions src/zeroconf/_services/info.py
Original file line number Diff line number Diff line change
Expand Up @@ -464,10 +464,12 @@ def _unpack_text_into_properties(self) -> None:
length = text[index]
index += 1
key_value = text[index : index + length]
key_sep_value = key_value.partition(b"=")
key = key_sep_value[0]
key, sep, value = key_value.partition(b"=")
if key not in properties:
properties[key] = key_sep_value[2] or None
# RFC 6763 section 6.4 distinguishes a key with no '=' (a
# boolean attribute: present, no value) from `key=` (present
# with an empty value), so test the separator, not the value.
properties[key] = value if sep else None
index += length

self._properties = properties
Expand Down
58 changes: 58 additions & 0 deletions tests/services/test_info.py
Original file line number Diff line number Diff line change
Expand Up @@ -630,6 +630,64 @@ def test_service_info_duplicate_properties_txt_records(self):
assert info.properties[b"ci"] == b"2"
zc.close()

def test_service_info_empty_value_txt_record(self):
"""Verify `key=` decodes to an empty value, not to a valueless `key`."""
zc = r.Zeroconf(interfaces=["127.0.0.1"])
service_name = "name._type._tcp.local."
service_type = "_type._tcp.local."
service_server = "ash-1.local."
text = b"\x03rm=\x02rs\x05ve=05"
info = ServiceInfo(
service_type,
service_name,
22,
0,
0,
{"path": "/~paulsm/"},
service_server,
addresses=[socket.inet_aton("10.0.1.2")],
)
info.async_update_records(
zc,
r.current_time_millis(),
[
r.RecordUpdate(
r.DNSText(
service_name,
const._TYPE_TXT,
const._CLASS_IN | const._CLASS_UNIQUE,
120,
text,
),
None,
)
],
)
assert info.properties[b"rm"] == b""
assert info.properties[b"rs"] is None
assert info.properties[b"ve"] == b"05"

# The string-facing API must preserve the distinction too.
assert info.decoded_properties["rm"] == ""
assert info.decoded_properties["rs"] is None
assert info.decoded_properties["ve"] == "05"

# Re-encoding either view of the properties must reproduce the received rdata.
for properties in (info.properties, info.decoded_properties):
assert (
ServiceInfo(
service_type,
service_name,
22,
0,
0,
properties,
service_server,
).text
== text
)
zc.close()


def test_multiple_addresses():
type_ = "_http._tcp.local."
Expand Down
Loading