mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-07-08 17:56:33 +00:00
Full support of bytes arrays on gvariant (#2024)
* Full support of bytes arrays on gvariant * Fix overlay ssid * cleanup ssid * fix test
This commit is contained in:
parent
4cb5770ee0
commit
2321890dde
@ -68,3 +68,4 @@ class WirelessProperties:
|
|||||||
|
|
||||||
properties: dict = attr.ib()
|
properties: dict = attr.ib()
|
||||||
security: dict = attr.ib()
|
security: dict = attr.ib()
|
||||||
|
ssid: str = attr.ib()
|
||||||
|
@ -1,7 +1,7 @@
|
|||||||
"""Connection object for Network Manager."""
|
"""Connection object for Network Manager."""
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
from ...const import ATTR_ADDRESS, ATTR_IPV4, ATTR_METHOD, ATTR_PREFIX
|
from ...const import ATTR_ADDRESS, ATTR_IPV4, ATTR_METHOD, ATTR_PREFIX, ATTR_SSID
|
||||||
from ...utils.gdbus import DBus
|
from ...utils.gdbus import DBus
|
||||||
from ..const import (
|
from ..const import (
|
||||||
DBUS_ATTR_802_WIRELESS,
|
DBUS_ATTR_802_WIRELESS,
|
||||||
@ -134,6 +134,7 @@ class NetworkConnection(NetworkAttributes):
|
|||||||
self._wireless = WirelessProperties(
|
self._wireless = WirelessProperties(
|
||||||
data.get(DBUS_ATTR_802_WIRELESS, {}),
|
data.get(DBUS_ATTR_802_WIRELESS, {}),
|
||||||
data.get(DBUS_ATTR_802_WIRELESS_SECURITY, {}),
|
data.get(DBUS_ATTR_802_WIRELESS_SECURITY, {}),
|
||||||
|
bytes(data.get(DBUS_ATTR_802_WIRELESS, {}).get(ATTR_SSID, [])).decode(),
|
||||||
)
|
)
|
||||||
|
|
||||||
self._device = NetworkDevice(
|
self._device = NetworkDevice(
|
||||||
|
@ -35,10 +35,7 @@ def interface_update_payload(interface, **kwargs) -> str:
|
|||||||
kwargs[ATTR_SSID] = ", ".join(
|
kwargs[ATTR_SSID] = ", ".join(
|
||||||
[
|
[
|
||||||
f"0x{x}"
|
f"0x{x}"
|
||||||
for x in interface.connection.wireless.properties[ATTR_SSID]
|
for x in interface.connection.wireless.ssid.encode().hex(",").split(",")
|
||||||
.encode()
|
|
||||||
.hex(",")
|
|
||||||
.split(",")
|
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -10,6 +10,8 @@ from signal import SIGINT
|
|||||||
from typing import Any, Dict, List, Optional, Set
|
from typing import Any, Dict, List, Optional, Set
|
||||||
import xml.etree.ElementTree as ET
|
import xml.etree.ElementTree as ET
|
||||||
|
|
||||||
|
import sentry_sdk
|
||||||
|
|
||||||
from ..exceptions import (
|
from ..exceptions import (
|
||||||
DBusFatalError,
|
DBusFatalError,
|
||||||
DBusInterfaceError,
|
DBusInterfaceError,
|
||||||
@ -31,7 +33,7 @@ RE_GVARIANT_STRING_ESC: re.Pattern[Any] = re.compile(
|
|||||||
RE_GVARIANT_STRING: re.Pattern[Any] = re.compile(
|
RE_GVARIANT_STRING: re.Pattern[Any] = re.compile(
|
||||||
r"(?<=(?: |{|\[|\(|<))'(.*?)'(?=(?:|]|}|,|\)|>))"
|
r"(?<=(?: |{|\[|\(|<))'(.*?)'(?=(?:|]|}|,|\)|>))"
|
||||||
)
|
)
|
||||||
RE_GVARIANT_BINARY: re.Pattern[Any] = re.compile(r"<\[byte (.*?)\]>")
|
RE_GVARIANT_BINARY: re.Pattern[Any] = re.compile(r"\[byte (.*?)\]")
|
||||||
RE_GVARIANT_TUPLE_O: re.Pattern[Any] = re.compile(r"\"[^\"\\]*(?:\\.[^\"\\]*)*\"|(\()")
|
RE_GVARIANT_TUPLE_O: re.Pattern[Any] = re.compile(r"\"[^\"\\]*(?:\\.[^\"\\]*)*\"|(\()")
|
||||||
RE_GVARIANT_TUPLE_C: re.Pattern[Any] = re.compile(
|
RE_GVARIANT_TUPLE_C: re.Pattern[Any] = re.compile(
|
||||||
r"\"[^\"\\]*(?:\\.[^\"\\]*)*\"|(,?\))"
|
r"\"[^\"\\]*(?:\\.[^\"\\]*)*\"|(,?\))"
|
||||||
@ -61,12 +63,7 @@ DBUS_METHOD_GETALL: str = "org.freedesktop.DBus.Properties.GetAll"
|
|||||||
def _convert_bytes(value: str) -> str:
|
def _convert_bytes(value: str) -> str:
|
||||||
"""Convert bytes to string or byte-array."""
|
"""Convert bytes to string or byte-array."""
|
||||||
data: bytes = bytes(int(char, 0) for char in value.split(", "))
|
data: bytes = bytes(int(char, 0) for char in value.split(", "))
|
||||||
try:
|
return f"[{', '.join(str(char) for char in data)}]"
|
||||||
text = data.decode()
|
|
||||||
except UnicodeDecodeError:
|
|
||||||
return f"<[{', '.join(str(char) for char in data)}]>"
|
|
||||||
|
|
||||||
return f"<'{text}'>"
|
|
||||||
|
|
||||||
|
|
||||||
class DBus:
|
class DBus:
|
||||||
@ -153,7 +150,8 @@ class DBus:
|
|||||||
try:
|
try:
|
||||||
return json.loads(json_raw)
|
return json.loads(json_raw)
|
||||||
except json.JSONDecodeError as err:
|
except json.JSONDecodeError as err:
|
||||||
_LOGGER.critical("Can't parse '%s': '%s' - %s", json_raw, raw, err)
|
_LOGGER.error("Can't parse '%s': '%s' - %s", json_raw, raw, err)
|
||||||
|
sentry_sdk.capture_exception(err)
|
||||||
raise DBusParseError() from err
|
raise DBusParseError() from err
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
@ -32,4 +32,4 @@ async def test_interface_update_payload_wireless(network_interface):
|
|||||||
)
|
)
|
||||||
assert DBus.parse_gvariant(data)["ipv4"]["method"] == "manual"
|
assert DBus.parse_gvariant(data)["ipv4"]["method"] == "manual"
|
||||||
assert DBus.parse_gvariant(data)["ipv4"]["address-data"][0]["address"] == "1.1.1.1"
|
assert DBus.parse_gvariant(data)["ipv4"]["address-data"][0]["address"] == "1.1.1.1"
|
||||||
assert DBus.parse_gvariant(data)["802-11-wireless"]["ssid"] == "NETT"
|
assert DBus.parse_gvariant(data)["802-11-wireless"]["ssid"] == [78, 69, 84, 84]
|
||||||
|
@ -325,7 +325,7 @@ def test_networkmanager_binary_data():
|
|||||||
"mode": "infrastructure",
|
"mode": "infrastructure",
|
||||||
"security": "802-11-wireless-security",
|
"security": "802-11-wireless-security",
|
||||||
"seen-bssids": ["7C:2E:BD:98:1B:06"],
|
"seen-bssids": ["7C:2E:BD:98:1B:06"],
|
||||||
"ssid": "NETT",
|
"ssid": [78, 69, 84, 84],
|
||||||
},
|
},
|
||||||
"connection": {
|
"connection": {
|
||||||
"id": "NETT",
|
"id": "NETT",
|
||||||
@ -370,7 +370,7 @@ def test_networkmanager_binary_data():
|
|||||||
"mode": "infrastructure",
|
"mode": "infrastructure",
|
||||||
"security": "802-11-wireless-security",
|
"security": "802-11-wireless-security",
|
||||||
"seen-bssids": ["7C:2E:BD:98:1B:06"],
|
"seen-bssids": ["7C:2E:BD:98:1B:06"],
|
||||||
"ssid": "NETT",
|
"ssid": [78, 69, 84, 84],
|
||||||
},
|
},
|
||||||
"802-11-wireless-security": {"auth-alg": "open", "key-mgmt": "wpa-psk"},
|
"802-11-wireless-security": {"auth-alg": "open", "key-mgmt": "wpa-psk"},
|
||||||
"connection": {
|
"connection": {
|
||||||
@ -402,3 +402,23 @@ def test_networkmanager_binary_data():
|
|||||||
"proxy": {},
|
"proxy": {},
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def test_v6():
|
||||||
|
"""Test IPv6 Property."""
|
||||||
|
raw = "({'addresses': <[([byte 0x20, 0x01, 0x04, 0x70, 0x79, 0x2d, 0x00, 0x01, 0x00, 0x12, 0x00, 0x00, 0x00, 0x00, 0x00, 0x10], uint32 64, [byte 0x20, 0x01, 0x04, 0x70, 0x79, 0x2d, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01])]>, 'dns': <[[byte 0x20, 0x01, 0x04, 0x70, 0x79, 0x2d, 0x00, 0x01, 0x00, 0x12, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05]]>})"
|
||||||
|
|
||||||
|
data = DBus.parse_gvariant(raw)
|
||||||
|
|
||||||
|
assert data == [
|
||||||
|
{
|
||||||
|
"addresses": [
|
||||||
|
[
|
||||||
|
[32, 1, 4, 112, 121, 45, 0, 1, 0, 18, 0, 0, 0, 0, 0, 16],
|
||||||
|
64,
|
||||||
|
[32, 1, 4, 112, 121, 45, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1],
|
||||||
|
]
|
||||||
|
],
|
||||||
|
"dns": [[32, 1, 4, 112, 121, 45, 0, 1, 0, 18, 0, 0, 0, 0, 0, 5]],
|
||||||
|
}
|
||||||
|
]
|
||||||
|
Loading…
x
Reference in New Issue
Block a user