Merge pull request #103 from dknowles2/diagnostics

Add support for returning redacted diagnostics

pyschlage/common.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+from copy import deepcopy
 from dataclasses import dataclass, field, fields
 from threading import Lock as Mutex
+from typing import Any
 
 from .auth import Auth
 
@@ -29,3 +31,31 @@ def _update_with(self, json, *args, **kwargs):
         with self._mu:
             for f in fields(new_obj):
                 setattr(self, f.name, getattr(new_obj, f.name))
+
+
+def redact(json: dict[Any, Any], *, allowed: list[str]) -> dict[Any, Any]:
+    """Returns a copy of the given JSON dict with non-allowed keys redacted."""
+    if len(allowed) == 1 and allowed[0] == "*":
+        return deepcopy(json)
+
+    allowed_here = {}
+    for allow in allowed:
+        k, _, children = allow.partition(".")
+        if k not in allowed_here:
+            allowed_here[k] = []
+        if not children:
+            children = "*"
+        allowed_here[k].append(children)
+
+    ret = {}
+    for k, v in json.items():
+        if isinstance(v, dict):
+            ret[k] = redact(v, allowed=allowed_here.get(k, []))
+        elif k in allowed_here:
+            ret[k] = v
+        else:
+            if isinstance(v, list):
+                ret[k] = ["<REDACTED>"]
+            else:
+                ret[k] = "<REDACTED>"
+    return ret

pyschlage/lock.py

@@ -2,11 +2,12 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from typing import Any
 
 from .auth import Auth
 from .code import AccessCode
-from .common import Mutable
+from .common import Mutable, redact
 from .exceptions import NotAuthenticatedError
 from .log import LockLog
 from .user import User
@@ -95,7 +96,9 @@ class Lock(Mutable):
     access_codes: dict[str, AccessCode] | None = None
     """Access codes for this lock, keyed by their ID."""
 
-    _cat: str = ""
+    _cat: str = field(default="", repr=False)
+
+    _json: dict[Any, Any] = field(default_factory=dict, repr=False)
 
     @staticmethod
     def request_path(device_id: str | None = None) -> str:
@@ -149,6 +152,53 @@ def from_json(cls, auth: Auth, json: dict) -> Lock:
             mac_address=attributes.get("macAddress"),
             users=users,
             _cat=json["CAT"],
+            _json=json,
+        )
+
+    def get_diagnostics(self) -> dict[Any, Any]:
+        """Returns a redacted dict of the raw JSON for diagnostics purposes."""
+        return redact(
+            self._json,
+            allowed=[
+                "attributes.accessCodeLength",
+                "attributes.actAlarmBuzzerEnabled",
+                "attributes.actAlarmState",
+                "attributes.actuationCurrentMax",
+                "attributes.alarmSelection",
+                "attributes.alarmSensitivity",
+                "attributes.alarmState",
+                "attributes.autoLockTime",
+                "attributes.batteryChangeDate",
+                "attributes.batteryLevel",
+                "attributes.batteryLowState",
+                "attributes.batterySaverConfig",
+                "attributes.batterySaverState",
+                "attributes.beeperEnabled",
+                "attributes.bleFirmwareVersion",
+                "attributes.firmwareUpdate",
+                "attributes.homePosCurrentMax",
+                "attributes.keypadFirmwareVersion",
+                "attributes.lockAndLeaveEnabled",
+                "attributes.lockState",
+                "attributes.lockStateMetadata",
+                "attributes.mainFirmwareVersion",
+                "attributes.mode",
+                "attributes.modelName",
+                "attributes.periodicDeepQueryTimeSetting",
+                "attributes.psPollEnabled",
+                "attributes.timezone",
+                "attributes.wifiFirmwareVersion",
+                "attributes.wifiRssi",
+                "connected",
+                "connectivityUpdated",
+                "created",
+                "devicetypeId",
+                "lastUpdated",
+                "modelName",
+                "name",
+                "role",
+                "timezone",
+            ],
         )
 
     def _is_wifi_lock(self) -> bool:

tests/test_common.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from pyschlage import common
+
+
+@pytest.fixture
+def json_dict() -> dict[Any, Any]:
+    return {
+        "a": "foo",
+        "b": 1,
+        "c": {
+            "c0": "foo",
+            "c1": 1,
+            "c2": {
+                "c20": "foo",
+            },
+            "c3": ["foo"],
+        },
+        "d": ["foo"],
+    }
+
+
+def test_redact_allow_asterisk(json_dict: dict[Any, Any]):
+    assert common.redact(json_dict, allowed=["*"]) == json_dict
+
+
+def test_redact_allow_all(json_dict: dict[Any, Any]):
+    assert common.redact(json_dict, allowed=["a", "b", "c.*", "d"]) == json_dict
+    assert (
+        common.redact(
+            json_dict, allowed=["a", "b", "c.c0", "c.c1", "c.c2", "c.c3", "d"]
+        )
+        == json_dict
+    )
+    assert common.redact(json_dict, allowed=["a", "b", "c", "d"]) == json_dict
+
+
+def test_redact_all(json_dict: dict[Any, Any]):
+    want = {
+        "a": "<REDACTED>",
+        "b": "<REDACTED>",
+        "c": {
+            "c0": "<REDACTED>",
+            "c1": "<REDACTED>",
+            "c2": {
+                "c20": "<REDACTED>",
+            },
+            "c3": ["<REDACTED>"],
+        },
+        "d": ["<REDACTED>"],
+    }
+    assert common.redact(json_dict, allowed=[]) == want
+
+
+def test_redact_partial(json_dict: dict[Any, Any]):
+    want = {
+        "a": "foo",
+        "b": 1,
+        "c": {
+            "c0": "foo",
+            "c1": "<REDACTED>",
+            "c2": {
+                "c20": "<REDACTED>",
+            },
+            "c3": ["<REDACTED>"],
+        },
+        "d": ["<REDACTED>"],
+    }
+    assert common.redact(json_dict, allowed=["a", "b", "c.c0"])

tests/test_lock.py

@@ -58,6 +58,72 @@ def test_from_json_wifi_lock_unavailable(
         assert lock.is_locked is None
         assert lock.is_jammed is None
 
+    def test_diagnostics(self, mock_auth: mock.Mock, lock_json: dict) -> None:
+        lock = Lock.from_json(mock_auth, lock_json)
+        want = {
+            "CAT": "<REDACTED>",
+            "SAT": "<REDACTED>",
+            "attributes": {
+                "CAT": "<REDACTED>",
+                "SAT": "<REDACTED>",
+                "accessCodeLength": 4,
+                "actAlarmBuzzerEnabled": 0,
+                "actAlarmState": 0,
+                "actuationCurrentMax": 226,
+                "alarmSelection": 0,
+                "alarmSensitivity": 0,
+                "alarmState": 0,
+                "autoLockTime": 0,
+                "batteryChangeDate": 1669017530,
+                "batteryLevel": 95,
+                "batteryLowState": 0,
+                "batterySaverConfig": {"activePeriod": [], "enabled": 0},
+                "batterySaverState": 0,
+                "beeperEnabled": 1,
+                "bleFirmwareVersion": "0118.000103.015",
+                "diagnostics": {},
+                "firmwareUpdate": {
+                    "status": {"additionalInfo": None, "updateStatus": None}
+                },
+                "homePosCurrentMax": 153,
+                "keypadFirmwareVersion": "03.00.00250052",
+                "lockAndLeaveEnabled": 1,
+                "lockState": 1,
+                "lockStateMetadata": {
+                    "UUID": None,
+                    "actionType": "periodicDeepQuery",
+                    "clientId": None,
+                    "name": None,
+                },
+                "macAddress": "<REDACTED>",
+                "mainFirmwareVersion": "10.00.00264232",
+                "mode": 2,
+                "modelName": "__model_name__",
+                "periodicDeepQueryTimeSetting": 60,
+                "psPollEnabled": 1,
+                "serialNumber": "<REDACTED>",
+                "timezone": -20,
+                "wifiFirmwareVersion": "03.15.00.01",
+                "wifiRssi": -42,
+            },
+            "connected": True,
+            "connectivityUpdated": "2022-12-04T20:58:22.000Z",
+            "created": "2020-04-05T21:53:11.000Z",
+            "deviceId": "<REDACTED>",
+            "devicetypeId": "be489wifi",
+            "lastUpdated": "2022-12-04T20:58:22.000Z",
+            "macAddress": "<REDACTED>",
+            "modelName": "__model_name__",
+            "name": "Door Lock",
+            "physicalId": "<REDACTED>",
+            "relatedDevices": ["<REDACTED>"],
+            "role": "owner",
+            "serialNumber": "<REDACTED>",
+            "timezone": -20,
+            "users": ["<REDACTED>"],
+        }
+        assert lock.get_diagnostics() == want
+
     def test_refresh(
         self, mock_auth: mock.Mock, lock_json: dict, access_code_json: dict
     ) -> None: