Redact sensitive info from debug logs (#1069)

Redacts sensitive data when debug logging device responses such as mac,
location and usernames

kasa/tests/discovery_fixtures.py

@@ -90,21 +90,26 @@ def create_discovery_mock(ip: str, fixture_data: dict):
         query_data: dict
         device_type: str
         encrypt_type: str
-        _datagram: bytes
         login_version: int | None = None
         port_override: int | None = None
 
+        @property
+        def _datagram(self) -> bytes:
+            if self.default_port == 9999:
+                return XorEncryption.encrypt(json_dumps(self.discovery_data))[4:]
+            else:
+                return (
+                    b"\x02\x00\x00\x01\x01[\x00\x00\x00\x00\x00\x00W\xcev\xf8"
+                    + json_dumps(self.discovery_data).encode()
+                )
+
     if "discovery_result" in fixture_data:
-        discovery_data = {"result": fixture_data["discovery_result"]}
+        discovery_data = {"result": fixture_data["discovery_result"].copy()}
         device_type = fixture_data["discovery_result"]["device_type"]
         encrypt_type = fixture_data["discovery_result"]["mgt_encrypt_schm"][
             "encrypt_type"
         ]
         login_version = fixture_data["discovery_result"]["mgt_encrypt_schm"].get("lv")
-        datagram = (
-            b"\x02\x00\x00\x01\x01[\x00\x00\x00\x00\x00\x00W\xcev\xf8"
-            + json_dumps(discovery_data).encode()
-        )
         dm = _DiscoveryMock(
             ip,
             80,
@@ -113,16 +118,14 @@ def create_discovery_mock(ip: str, fixture_data: dict):
             fixture_data,
             device_type,
             encrypt_type,
-            datagram,
             login_version,
         )
     else:
         sys_info = fixture_data["system"]["get_sysinfo"]
-        discovery_data = {"system": {"get_sysinfo": sys_info}}
+        discovery_data = {"system": {"get_sysinfo": sys_info.copy()}}
         device_type = sys_info.get("mic_type") or sys_info.get("type")
         encrypt_type = "XOR"
         login_version = None
-        datagram = XorEncryption.encrypt(json_dumps(discovery_data))[4:]
         dm = _DiscoveryMock(
             ip,
             9999,
@@ -131,7 +134,6 @@ def create_discovery_mock(ip: str, fixture_data: dict):
             fixture_data,
             device_type,
             encrypt_type,
-            datagram,
             login_version,
         )
 

kasa/tests/test_discovery.py

@@ -2,6 +2,7 @@
 # ruff: noqa: S106
 
 import asyncio
+import logging
 import re
 import socket
 from unittest.mock import MagicMock
@@ -565,3 +566,38 @@ async def test_do_discover_external_cancel(mocker):
     with pytest.raises(asyncio.TimeoutError):
         async with asyncio_timeout(0):
             await dp.wait_for_discovery_to_complete()
+
+
+async def test_discovery_redaction(discovery_mock, caplog: pytest.LogCaptureFixture):
+    """Test query sensitive info redaction."""
+    mac = "12:34:56:78:9A:BC"
+
+    if discovery_mock.default_port == 9999:
+        sysinfo = discovery_mock.discovery_data["system"]["get_sysinfo"]
+        if "mac" in sysinfo:
+            sysinfo["mac"] = mac
+        elif "mic_mac" in sysinfo:
+            sysinfo["mic_mac"] = mac
+    else:
+        discovery_mock.discovery_data["result"]["mac"] = mac
+
+    # Info no message logging
+    caplog.set_level(logging.INFO)
+    await Discover.discover()
+
+    assert mac not in caplog.text
+
+    caplog.set_level(logging.DEBUG)
+
+    # Debug no redaction
+    caplog.clear()
+    Discover._redact_data = False
+    await Discover.discover()
+    assert mac in caplog.text
+
+    # Debug redaction
+    caplog.clear()
+    Discover._redact_data = True
+    await Discover.discover()
+    assert mac not in caplog.text
+    assert "12:34:56:00:00:00" in caplog.text

kasa/tests/test_protocol.py

@@ -8,9 +8,12 @@ import os
 import pkgutil
 import struct
 import sys
+from typing import cast
 
 import pytest
 
+from kasa.iot import IotDevice
+
 from ..aestransport import AesTransport
 from ..credentials import Credentials
 from ..device import Device
@@ -21,8 +24,12 @@ from ..klaptransport import KlapTransport, KlapTransportV2
 from ..protocol import (
     BaseProtocol,
     BaseTransport,
+    mask_mac,
+    redact_data,
 )
 from ..xortransport import XorEncryption, XorTransport
+from .conftest import device_iot
+from .fakeprotocol_iot import FakeIotTransport
 
 
 @pytest.mark.parametrize(
@@ -676,3 +683,63 @@ def test_deprecated_protocol():
         host = "127.0.0.1"
         proto = TPLinkSmartHomeProtocol(host=host)
         assert proto.config.host == host
+
+
+@device_iot
+async def test_iot_queries_redaction(dev: IotDevice, caplog: pytest.LogCaptureFixture):
+    """Test query sensitive info redaction."""
+    device_id = "123456789ABCDEF"
+    cast(FakeIotTransport, dev.protocol._transport).proto["system"]["get_sysinfo"][
+        "deviceId"
+    ] = device_id
+
+    # Info no message logging
+    caplog.set_level(logging.INFO)
+    await dev.update()
+    assert device_id not in caplog.text
+
+    caplog.set_level(logging.DEBUG, logger="kasa")
+    # The fake iot protocol also logs so disable it
+    test_logger = logging.getLogger("kasa.tests.fakeprotocol_iot")
+    test_logger.setLevel(logging.INFO)
+
+    # Debug no redaction
+    caplog.clear()
+    cast(IotProtocol, dev.protocol)._redact_data = False
+    await dev.update()
+    assert device_id in caplog.text
+
+    # Debug redaction
+    caplog.clear()
+    cast(IotProtocol, dev.protocol)._redact_data = True
+    await dev.update()
+    assert device_id not in caplog.text
+    assert "REDACTED_" + device_id[9::] in caplog.text
+
+
+async def test_redact_data():
+    """Test redact data function."""
+    data = {
+        "device_id": "123456789ABCDEF",
+        "owner": "0987654",
+        "mac": "12:34:56:78:90:AB",
+        "ip": "192.168.1",
+        "no_val": None,
+    }
+    excpected_data = {
+        "device_id": "REDACTED_ABCDEF",
+        "owner": "**REDACTED**",
+        "mac": "12:34:56:00:00:00",
+        "ip": "**REDACTEX**",
+        "no_val": None,
+    }
+    REDACTORS = {
+        "device_id": lambda x: "REDACTED_" + x[9::],
+        "owner": None,
+        "mac": mask_mac,
+        "ip": lambda x: "127.0.0." + x.split(".")[3],
+    }
+
+    redacted_data = redact_data(data, REDACTORS)
+
+    assert redacted_data == excpected_data

kasa/tests/test_smartprotocol.py

@@ -1,8 +1,11 @@
 import logging
+from typing import cast
 
 import pytest
 import pytest_mock
 
+from kasa.smart import SmartDevice
+
 from ..exceptions import (
     SMART_RETRYABLE_ERRORS,
     DeviceError,
@@ -10,6 +13,7 @@ from ..exceptions import (
     SmartErrorCode,
 )
 from ..smartprotocol import SmartProtocol, _ChildProtocolWrapper
+from .conftest import device_smart
 from .fakeprotocol_smart import FakeSmartTransport
 
 DUMMY_QUERY = {"foobar": {"foo": "bar", "bar": "foo"}}
@@ -409,3 +413,34 @@ async def test_incomplete_list(mocker, caplog):
         "Device 127.0.0.123 returned empty results list for method get_preset_rules"
         in caplog.text
     )
+
+
+@device_smart
+async def test_smart_queries_redaction(
+    dev: SmartDevice, caplog: pytest.LogCaptureFixture
+):
+    """Test query sensitive info redaction."""
+    device_id = "123456789ABCDEF"
+    cast(FakeSmartTransport, dev.protocol._transport).info["get_device_info"][
+        "device_id"
+    ] = device_id
+
+    # Info no message logging
+    caplog.set_level(logging.INFO)
+    await dev.update()
+    assert device_id not in caplog.text
+
+    caplog.set_level(logging.DEBUG)
+
+    # Debug no redaction
+    caplog.clear()
+    dev.protocol._redact_data = False
+    await dev.update()
+    assert device_id in caplog.text
+
+    # Debug redaction
+    caplog.clear()
+    dev.protocol._redact_data = True
+    await dev.update()
+    assert device_id not in caplog.text
+    assert "REDACTED_" + device_id[9::] in caplog.text