jeans/supervisor
jeans/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-07-25 18:16:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

gvariant: parse byte string (#2041)

Browse Source 
* Parse byte string

* fix lint
This commit is contained in:
Pascal Vizeli 2020-09-11 18:36:19 +02:00 committed by GitHub
parent 8b4a137252
commit 7be508214a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 77 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
supervisor/utils/gdbus.py
View File

@ -33,12 +33,20 @@ RE_GVARIANT_STRING_ESC: re.Pattern[Any] = re.compile(
RE_GVARIANT_STRING: re.Pattern[Any] = re.compile(
r"(?<=(?: |{|\[|\(|<))'(.*?)'(?=(?:|]|}|,|\)|>))"
)
RE_GVARIANT_BINARY: re.Pattern[Any] = re.compile(r"\[byte (.*?)\]")
RE_GVARIANT_BINARY: re.Pattern[Any] = re.compile(
r"\"[^\"\\]*(?:\\.[^\"\\]*)*\"|\[byte (.*?)\]"
)
RE_GVARIANT_BINARY_STRING: re.Pattern[Any] = re.compile(
r"\"[^\"\\]*(?:\\.[^\"\\]*)*\"|<?b\'(.*?)\'>?"
)
RE_GVARIANT_TUPLE_O: re.Pattern[Any] = re.compile(r"\"[^\"\\]*(?:\\.[^\"\\]*)*\"|(\()")
RE_GVARIANT_TUPLE_C: re.Pattern[Any] = re.compile(
r"\"[^\"\\]*(?:\\.[^\"\\]*)*\"|(,?\))"
)
RE_BIN_STRING_OCT: re.Pattern[Any] = re.compile(r"\\\\(\d{3})")
RE_BIN_STRING_HEX: re.Pattern[Any] = re.compile(r"\\\\x(\d{2})")
RE_MONITOR_OUTPUT: re.Pattern[Any] = re.compile(r".+?: (?P<signal>[^ ].+) (?P<data>.*)")
# Map GDBus to errors
@ -66,6 +74,13 @@ def _convert_bytes(value: str) -> str:
return f"[{', '.join(str(char) for char in data)}]"
def _convert_bytes_string(value: str) -> str:
"""Convert bytes to string or byte-array."""
data = RE_BIN_STRING_OCT.sub(lambda x: chr(int(x.group(1), 8)), value)
data = RE_BIN_STRING_HEX.sub(lambda x: chr(int(f"0x{x.group(1)}", 0)), data)
return f"[{', '.join(str(char) for char in list(char for char in data.encode()))}]"
class DBus:
"""DBus handler."""
@ -120,15 +135,23 @@ class DBus:
def parse_gvariant(raw: str) -> Any:
"""Parse GVariant input to python."""
# Process first string
json_raw = RE_GVARIANT_BINARY.sub(
lambda x: _convert_bytes(x.group(1)),
raw,
)
json_raw = RE_GVARIANT_STRING_ESC.sub(
lambda x: x.group(0).replace('"', '\\"'), json_raw
lambda x: x.group(0).replace('"', '\\"'), raw
)
json_raw = RE_GVARIANT_STRING.sub(r'"\1"', json_raw)
# Handle Bytes
json_raw = RE_GVARIANT_BINARY.sub(
lambda x: x.group(0) if not x.group(1) else _convert_bytes(x.group(1)),
json_raw,
)
json_raw = RE_GVARIANT_BINARY_STRING.sub(
lambda x: x.group(0)
if not x.group(1)
else _convert_bytes_string(x.group(1)),
json_raw,
)
# Remove complex type handling
json_raw: str = RE_GVARIANT_TYPE.sub(
lambda x: x.group(0) if not x.group(1) else "", json_raw

48
tests/utils/test_gvariant_parser.py
View File

@ -404,6 +404,54 @@ def test_networkmanager_binary_data():
]
def test_networkmanager_binary_string_data():
"""Test NetworkManager Binary string datastrings."""
raw = "({'802-11-wireless': {'mac-address-blacklist': <@as []>, 'mac-address': <b'*~_\\\\035\\\\311'>, 'mode': <'infrastructure'>, 'security': <'802-11-wireless-security'>, 'seen-bssids': <['7C:2E:BD:98:1B:06']>, 'ssid': <[byte 0x4e, 0x45, 0x54, 0x54]>}, 'connection': {'id': <'NETT'>, 'interface-name': <'wlan0'>, 'permissions': <@as []>, 'timestamp': <uint64 1598526799>, 'type': <'802-11-wireless'>, 'uuid': <'13f9af79-a6e9-4e07-9353-165ad57bf1a8'>}, 'ipv6': {'address-data': <@aa{sv} []>, 'addresses': <@a(ayuay) []>, 'dns': <@aay []>, 'dns-search': <@as []>, 'method': <'auto'>, 'route-data': <@aa{sv} []>, 'routes': <@a(ayuayu) []>}, '802-11-wireless-security': {'auth-alg': <'open'>, 'key-mgmt': <'wpa-psk'>}, 'ipv4': {'address-data': <@aa{sv} []>, 'addresses': <@aau []>, 'dns': <@au []>, 'dns-search': <@as []>, 'method': <'auto'>, 'route-data': <@aa{sv} []>, 'routes': <@aau []>}, 'proxy': {}},)"
data = DBus.parse_gvariant(raw)
assert data == [
{
"802-11-wireless": {
"mac-address": [42, 126, 95, 29, 195, 137],
"mac-address-blacklist": [],
"mode": "infrastructure",
"security": "802-11-wireless-security",
"seen-bssids": ["7C:2E:BD:98:1B:06"],
"ssid": [78, 69, 84, 84],
},
"802-11-wireless-security": {"auth-alg": "open", "key-mgmt": "wpa-psk"},
"connection": {
"id": "NETT",
"interface-name": "wlan0",
"permissions": [],
"timestamp": 1598526799,
"type": "802-11-wireless",
"uuid": "13f9af79-a6e9-4e07-9353-165ad57bf1a8",
},
"ipv4": {
"address-data": [],
"addresses": [],
"dns": [],
"dns-search": [],
"method": "auto",
"route-data": [],
"routes": [],
},
"ipv6": {
"address-data": [],
"addresses": [],
"dns": [],
"dns-search": [],
"method": "auto",
"route-data": [],
"routes": [],
},
"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]]>})"