Add automatic censoring to logging and refactor utils to folder (#96)

* Add automatic censoring to logging and refactor utils to folder

* Use asterisks instead of CENSORED

* Add extra key to check that it only modifies what is intended

* Remove censor_location and associated tests

mytoyota/api.py

@@ -1,12 +1,9 @@
 """Toyota Connected Services API"""
-import logging
 from typing import Optional
 
 from .const import BASE_URL, BASE_URL_CARS
 from .controller import Controller
 
-_LOGGER: logging.Logger = logging.getLogger(__package__)
-
 
 class Api:
     """Controller class."""

mytoyota/client.py

@@ -36,7 +36,8 @@
     ToyotaRegionNotSupported,
 )
 from .statistics import Statistics
-from .utils import is_valid_locale
+from .utils.locale import is_valid_locale
+from .utils.logs import censor, censor_vin
 from .vehicle import Vehicle
 
 _LOGGER: logging.Logger = logging.getLogger(__package__)
@@ -140,7 +141,7 @@ async def set_alias(self, vehicle_id: int, new_alias: str) -> dict:
             ToyotaApiError: Toyota's API returned an error.
         """
         _LOGGER.debug(
-            f"Setting new alias: {new_alias} for vehicle with id: {vehicle_id}"
+            f"Setting new alias: {new_alias} for vehicle with id: {censor(str(vehicle_id))}"
         )
         result = await self.api.set_vehicle_alias_endpoint(
             vehicle_id=vehicle_id, new_alias=new_alias
@@ -239,7 +240,7 @@ async def get_vehicle_status(self, vehicle: dict) -> Vehicle:
             ToyotaInternalError: An error occurred when making a request.
             ToyotaApiError: Toyota's API returned an error.
         """
-        _LOGGER.debug(f"Getting status for vehicle - {vehicle}...")
+        _LOGGER.debug(f"Getting status for vehicle - {censor_vin(vehicle['vin'])}...")
 
         vin = vehicle["vin"]
         data = await asyncio.gather(
@@ -345,7 +346,7 @@ async def get_driving_statistics(  # pylint: disable=too-many-branches
             ToyotaApiError: Toyota's API returned an error.
         """
 
-        _LOGGER.debug(f"Getting statistics for {vin}...")
+        _LOGGER.debug(f"Getting statistics for {censor_vin(vin)}...")
         _LOGGER.debug(f"Interval: {interval} - from_date: {from_date} - unit: {unit}")
 
         if interval not in INTERVAL_SUPPORTED:
@@ -354,7 +355,7 @@ async def get_driving_statistics(  # pylint: disable=too-many-branches
         stats_interval = interval
 
         if from_date is not None and arrow.get(from_date) > arrow.now():
-            return [{"error_mesg": "This is not a timemachine!", "error_code": 5}]
+            return [{"error_mesg": "This is not a time machine!", "error_code": 5}]
 
         if from_date is None:
             if interval is DAY:

mytoyota/controller.py

@@ -24,7 +24,8 @@
     UUID,
 )
 from mytoyota.exceptions import ToyotaApiError, ToyotaInternalError, ToyotaLoginError
-from mytoyota.utils import is_valid_token
+from mytoyota.utils.logs import censor_dict
+from mytoyota.utils.token import is_valid_token
 
 _LOGGER: logging.Logger = logging.getLogger(__package__)
 
@@ -174,15 +175,15 @@ async def request(
                 }
             )
 
-        _LOGGER.debug(f"Additional headers: {headers}")
+        _LOGGER.debug(f"Additional headers: {censor_dict(headers.copy())}")
 
         # Cannot authenticate with aiohttp (returns 415),
         # but it works with httpx.
         _LOGGER.debug("Creating client...")
         _LOGGER.debug(f"Base headers: {BASE_HEADERS} - Timeout: {TIMEOUT}")
         async with httpx.AsyncClient(headers=BASE_HEADERS, timeout=TIMEOUT) as client:
             _LOGGER.debug(
-                f"Requesting {url} - Method: {method} - Body: {body} - Parameters: {params}"
+                f"Body: {censor_dict(body) if body else body} - Parameters: {params}"
             )
             response = await client.request(
                 method, url, headers=headers, json=body, params=params
@@ -211,6 +212,4 @@ async def request(
                     "HTTP: " + str(response.status_code) + " - " + response.text
                 )
 
-        _LOGGER.debug(f"Raw result: {result}")
-
         return result

mytoyota/location.py

@@ -12,8 +12,6 @@ class ParkingLocation:
     timestamp: int = 0
 
     def __init__(self, parking: dict) -> None:
-        _LOGGER.debug("Raw parking location data: %s", str(parking))
-
         self.latitude = float(parking.get("lat", 0.0))
         self.longitude = float(parking.get("lon", 0.0))
         self.timestamp = int(parking.get("timestamp", 0))

mytoyota/sensors.py

@@ -1,11 +1,8 @@
 """Sensor representation for mytoyota"""
-import logging
 from typing import Optional
 
 from mytoyota.const import CLOSED, INCAR, LOCKED, OFF, STATE, WARNING
 
-_LOGGER: logging.Logger = logging.getLogger(__package__)
-
 
 class Hood:
     """Representation of the hood of the car"""

mytoyota/statistics.py

@@ -24,7 +24,7 @@
     WEEK,
     YEAR,
 )
-from mytoyota.utils import (
+from mytoyota.utils.conversions import (
     convert_to_liter_per_100_miles,
     convert_to_miles,
     convert_to_mpg,

mytoyota/status.py

@@ -4,7 +4,7 @@
 
 from mytoyota.const import DOORS, HOOD, KEY, LIGHTS, WINDOWS
 from mytoyota.sensors import Doors, Hood, Key, Lights, Windows
-from mytoyota.utils import convert_to_miles
+from mytoyota.utils.conversions import convert_to_miles
 
 _LOGGER: logging.Logger = logging.getLogger(__package__)
 

mytoyota/utils/conversions.py

@@ -0,0 +1,24 @@
+"""Conversion utilities used"""
+import logging
+
+_LOGGER: logging.Logger = logging.getLogger(__package__)
+
+
+def convert_to_miles(kilometers: float) -> float:
+    """Convert kilometers to miles"""
+    _LOGGER.debug(f"Converting {kilometers} to miles...")
+    return round(kilometers * 0.621371192, 4)
+
+
+def convert_to_liter_per_100_miles(liters: float) -> float:
+    """Convert liters per 100 km to liters per 100 miles"""
+    _LOGGER.debug("Converting to L/100miles...")
+    return round(liters * 1.609344, 4)
+
+
+def convert_to_mpg(liters_per_100_km: float) -> float:
+    """Convert to miles per UK gallon (MPG)"""
+    _LOGGER.debug("Converting to MPG...")
+    if liters_per_100_km > 0.0:
+        return round(282.5 / liters_per_100_km, 4)
+    return 0.0

mytoyota/utils/formatters.py

@@ -0,0 +1,12 @@
+"""Formatters used"""
+
+
+def format_odometer(raw: list) -> dict:
+    """Formats odometer information from a list to a dict."""
+    instruments: dict = {}
+    for instrument in raw:
+        instruments[instrument["type"]] = instrument["value"]
+        if "unit" in instrument:
+            instruments[instrument["type"] + "_unit"] = instrument["unit"]
+
+    return instruments

mytoyota/utils/locale.py

@@ -0,0 +1,14 @@
+"""Locale validation utilities"""
+from langcodes import Language
+from langcodes.tag_parser import LanguageTagError
+
+
+def is_valid_locale(locale: str) -> bool:
+    """Is locale string valid."""
+    valid = False
+    if locale:
+        try:
+            valid = Language.get(locale).is_valid()
+        except LanguageTagError:
+            pass
+    return valid

mytoyota/utils/logs.py

@@ -0,0 +1,53 @@
+"""Log utilities"""
+
+
+def censor(text) -> str:
+    """
+    Replaces characters in a str with the asterisks
+    text: The text to censure.
+    """
+    char = "*"
+    text = text if text else ""
+    return text[0] + (len(text) - 1) * char if text else text
+
+
+def censor_cookie(cookie) -> str:
+    """
+    Replaces characters in a str with the asterisks
+    text: The text to censure.
+    """
+    string = cookie.split("=")
+    return string[0] + "=" + censor(string[1])
+
+
+def censor_vin(vin) -> str:
+    """
+    Replaces parts of the vin number with asterisks.
+    """
+    vin = vin if vin else ""
+    return vin[:-8] + "********" if vin else vin
+
+
+def censor_dict(dictionary) -> dict:
+    """
+    Censors token, vin and other private info in a dict.
+    """
+    if "vin" in dictionary:
+        dictionary["vin"] = censor_vin(dictionary["vin"])
+
+    if "VIN" in dictionary:
+        dictionary["VIN"] = censor_vin(dictionary["VIN"])
+
+    if "X-TME-TOKEN" in dictionary:
+        dictionary["X-TME-TOKEN"] = censor(dictionary["X-TME-TOKEN"])
+
+    if "uuid" in dictionary:
+        dictionary["uuid"] = censor(dictionary["uuid"])
+
+    if "id" in dictionary:
+        dictionary["id"] = censor(str(dictionary["id"]))
+
+    if "Cookie" in dictionary:
+        dictionary["Cookie"] = censor_cookie(dictionary["Cookie"])
+
+    return dictionary

mytoyota/utils/token.py

@@ -0,0 +1,13 @@
+"""Token validation utilities"""
+from mytoyota.const import TOKEN_LENGTH
+from mytoyota.exceptions import ToyotaInvalidToken
+
+
+def is_valid_token(token: str) -> bool:
+    """Checks if token is the correct length"""
+    if token and len(token) == TOKEN_LENGTH and token.endswith("..*"):
+        return True
+
+    raise ToyotaInvalidToken(
+        f"Token must end with '..*' and be {TOKEN_LENGTH} characters long."
+    )

mytoyota/vehicle.py

@@ -5,7 +5,8 @@
 from mytoyota.hvac import Hvac
 from mytoyota.location import ParkingLocation
 from mytoyota.status import Energy, Odometer, Sensors
-from mytoyota.utils import format_odometer
+from mytoyota.utils.formatters import format_odometer
+from mytoyota.utils.logs import censor_vin
 
 _LOGGER: logging.Logger = logging.getLogger(__package__)
 
@@ -55,13 +56,9 @@ def __init__(
             _LOGGER.error("No vehicle information provided!")
             return
 
-        _LOGGER.debug("Raw connected services data: %s", str(connected_services))
-
         if connected_services is not None:
             self.is_connected = self._has_connected_services_enabled(connected_services)
 
-        _LOGGER.debug("Raw vehicle info: %s", str(vehicle_info))
-
         # Vehicle information
         self.id = vehicle_info.get("id", None)  # pylint: disable=invalid-name
         self.vin = vehicle_info.get("vin", None)
@@ -72,8 +69,6 @@ def __init__(
 
         if self.is_connected:
 
-            _LOGGER.debug("Raw status data: %s", str(status))
-
             remote_control_info = remote_control.get("VehicleInfo", {})
 
             # Extract mileage and if the car reports in km or mi
@@ -151,13 +146,13 @@ def _has_connected_services_enabled(self, json_dict: dict) -> bool:
 
             _LOGGER.error(
                 "Please setup Connected Services if you want live data from the car. (%s)",
-                self.vin,
+                censor_vin(self.vin),
             )
             return False
         _LOGGER.error(
             "Your vehicle does not support Connected services (%s). You can find out if your "
             "vehicle is compatible by checking the manual that comes with your car.",
-            self.vin,
+            censor_vin(self.vin),
         )
         return False
 

tests/test_utils.py

@@ -3,16 +3,16 @@
 import pytest  # pylint: disable=import-error
 
 from mytoyota.exceptions import ToyotaInvalidToken
-from mytoyota.utils import (
+
+# pylint: disable=no-self-use
+from mytoyota.utils.conversions import (
     convert_to_liter_per_100_miles,
     convert_to_miles,
     convert_to_mpg,
-    format_odometer,
-    is_valid_locale,
-    is_valid_token,
 )
-
-# pylint: disable=no-self-use
+from mytoyota.utils.formatters import format_odometer
+from mytoyota.utils.locale import is_valid_locale
+from mytoyota.utils.token import is_valid_token
 
 
 class TestUtils: