Move xmlrpc credentials to header (#308)

changelog.txt

@@ -1,3 +1,6 @@
+Version 0.35.3 (2022-02-23)
+- Move xmlrpc credentials to header
+
 Version 0.35.2 (2022-02-22)
 - Remove password from Exceptions
 

hahomematic/client.py

@@ -41,7 +41,12 @@
 )
 from hahomematic.device import HmDevice
 from hahomematic.exceptions import BaseHomematicException, HaHomematicException
-from hahomematic.helpers import build_api_url, get_channel_no, parse_ccu_sys_var
+from hahomematic.helpers import (
+    build_xml_rpc_uri,
+    build_headers,
+    get_channel_no,
+    parse_ccu_sys_var,
+)
 from hahomematic.json_rpc_client import JsonRpcAioHttpClient
 from hahomematic.xml_rpc_proxy import XmlRpcProxy
 
@@ -921,42 +926,46 @@ def __init__(
             if self._central_config.callback_port
             else self.central.local_port
         )
+        self.has_credentials: bool = (
+            self._central_config.username is not None
+            and self._central_config.password is not None
+        )
         self.init_url: str = f"http://{self._callback_host}:{self._callback_port}"
-        self.api_url = build_api_url(
+        self.xml_rpc_uri = build_xml_rpc_uri(
             host=self._central_config.host,
             port=interface_config.port,
             path=interface_config.path,
-            username=self._central_config.username,
-            password=self._central_config.password,
             tls=self._central_config.tls,
         )
-        self.has_credentials: bool = (
-            self._central_config.username is not None
-            and self._central_config.password is not None
+        self.xml_rpc_headers = build_headers(
+            username=self._central_config.username,
+            password=self._central_config.password,
         )
-        self.version: str | None = None
         self.xml_rpc_proxy: XmlRpcProxy = XmlRpcProxy(
             self.central.loop,
             max_workers=1,
-            uri=self.api_url,
+            uri=self.xml_rpc_uri,
+            headers=self.xml_rpc_headers,
             tls=self._central_config.tls,
             verify_tls=self._central_config.verify_tls,
         )
         self.xml_rpc_proxy_read: XmlRpcProxy = XmlRpcProxy(
             self.central.loop,
             max_workers=1,
-            uri=self.api_url,
+            uri=self.xml_rpc_uri,
+            headers=self.xml_rpc_headers,
             tls=self._central_config.tls,
             verify_tls=self._central_config.verify_tls,
         )
+        self.version: str | None = None
 
     async def get_client(self) -> Client:
         """Identify the used client."""
         try:
             self.version = await self.xml_rpc_proxy.getVersion()
         except BaseHomematicException as hhe:
             raise HaHomematicException(
-                f"Failed to get backend version. Not creating client: {self.api_url}"
+                f"Failed to get backend version. Not creating client: {self.xml_rpc_uri}"
             ) from hhe
         if self.version:
             if "Homegear" in self.version or "pydevccu" in self.version:

hahomematic/exceptions.py

@@ -1,20 +1,15 @@
 """Module for HaHomematicExceptions."""
 from __future__ import annotations
 
-import re
 from typing import Any
 
-USERNAME_PASSWORD_DIVIDER = ":"
-CREDENTIALS_URL_DIVIDER = "@"
-REPLACE_TEXT = f"{USERNAME_PASSWORD_DIVIDER}<PASSWORD>{CREDENTIALS_URL_DIVIDER}"
-
 
 class BaseHomematicException(Exception):
     """hahomematic base exception."""
 
     def __init__(self, name: str, *args: Any) -> None:
         """Init the HaHomematicException."""
-        super().__init__(_replace_password_in_args(args))
+        super().__init__(*args)
         self.name = name
 
 
@@ -40,52 +35,3 @@ class HaHomematicException(BaseHomematicException):
     def __init__(self, *args: Any) -> None:
         """Init the HaHomematicException."""
         super().__init__("HaHomematicException", *args)
-
-
-def _replace_password_in_args(args: Any) -> Any:
-    """Replace in args."""
-    if isinstance(args, str):
-        return _replace_password_in_text(input_text=args)
-    if isinstance(args, Exception):
-        return _replace_password_in_exception(parent_exception=args)
-    if isinstance(args, dict):
-        return _replace_password_in_dict(input_dict=args)
-    if isinstance(args, tuple):
-        new_args: list[Any] = []
-        for arg in args:
-            if isinstance(arg, str):
-                new_args.append(_replace_password_in_text(input_text=arg))
-            elif isinstance(arg, Exception):
-                new_args.append(_replace_password_in_exception(parent_exception=arg))
-            else:
-                new_args.append(arg)
-        return tuple(new_args)
-    return args
-
-
-def _replace_password_in_exception(parent_exception: Exception) -> Exception:
-    """Try replace password by special string."""
-    parent_exception.__dict__.update(
-        _replace_password_in_dict(parent_exception.__dict__)
-    )
-    return parent_exception
-
-
-def _replace_password_in_dict(input_dict: dict[str, Any]) -> dict[str, Any]:
-    """Try replace password in dict by special string."""
-
-    for k, value in input_dict.items():
-        if isinstance(value, str):
-            input_dict[k] = _replace_password_in_text(input_text=value)
-        if isinstance(value, dict):
-            input_dict[k] = _replace_password_in_dict(input_dict=value)
-    return input_dict
-
-
-def _replace_password_in_text(input_text: str) -> str:
-    """Try replace password by special string."""
-
-    regex = f"{USERNAME_PASSWORD_DIVIDER}(.*){CREDENTIALS_URL_DIVIDER}"
-    if replaced_text := re.sub(regex, REPLACE_TEXT, input_text):
-        return replaced_text
-    return input_text

hahomematic/helpers.py

@@ -3,6 +3,7 @@
 """
 from __future__ import annotations
 
+import base64
 from datetime import datetime
 import logging
 import os
@@ -54,40 +55,31 @@ def generate_unique_id(
     return f"{domain}_{unique_id}".lower()
 
 
-def make_http_credentials(
-    username: str | None = None, password: str | None = None
-) -> str:
-    """Build auth part for api_url."""
-    credentials = ""
-    if username is None:
-        return credentials
-    if username is not None:
-        if ":" in username:
-            return credentials
-        credentials += username
-    if credentials and password is not None:
-        credentials += f":{password}"
-    return f"{credentials}@"
-
-
-def build_api_url(
+def build_xml_rpc_uri(
     host: str,
     port: int,
     path: str | None,
-    username: str | None = None,
-    password: str | None = None,
     tls: bool = False,
 ) -> str:
     """Build XML-RPC API URL from components."""
-    credentials = make_http_credentials(username, password)
     scheme = "http"
     if not path:
         path = ""
     if path and not path.startswith("/"):
         path = f"/{path}"
     if tls:
         scheme += "s"
-    return f"{scheme}://{credentials}{host}:{port}{path}"
+    return f"{scheme}://{host}:{port}{path}"
+
+
+def build_headers(
+    username: str | None = None,
+    password: str | None = None,
+) -> list[tuple[str, str]]:
+    """Build XML-RPC API header."""
+    cred_bytes = f"{username}:{password}".encode("utf-8")
+    base64_message = base64.b64encode(cred_bytes).decode("utf-8")
+    return [("Authorization", f"Basic {base64_message}")]
 
 
 def check_or_create_directory(directory: str) -> bool:

hahomematic/hmcli.py

@@ -9,7 +9,7 @@
 from xmlrpc.client import ServerProxy
 
 from hahomematic.const import PARAMSET_KEY_MASTER, PARAMSET_KEY_VALUES
-from hahomematic.helpers import build_api_url, get_tls_context
+from hahomematic.helpers import build_xml_rpc_uri, build_headers, get_tls_context
 
 
 def main() -> None:
@@ -67,18 +67,17 @@ def main() -> None:
     )
     args = parser.parse_args()
 
-    url = build_api_url(
+    url = build_xml_rpc_uri(
         host=args.host,
         port=args.port,
         path=args.path,
-        username=args.username,
-        password=args.password,
         tls=args.tls,
     )
+    headers = build_headers(username=args.username, password=args.password)
     context = None
     if args.tls:
         context = get_tls_context(verify_tls=args.verify)
-    proxy = ServerProxy(url, context=context)
+    proxy = ServerProxy(url, context=context, headers=headers)
 
     try:
         if args.paramset_key == PARAMSET_KEY_VALUES and args.value is None:

setup.py

@@ -15,7 +15,7 @@ def readme():
     },
 PACKAGE_NAME = "hahomematic"
 HERE = os.path.abspath(os.path.dirname(__file__))
-VERSION = "0.35.2"
+VERSION = "0.35.3"
 
 PACKAGES = find_packages(exclude=["tests", "tests.*", "dist", "build"])
 

tests/test_central.py

@@ -83,12 +83,12 @@ async def test_central(central, loop) -> None:
     assert usage_types[HmEntityUsage.CE_SECONDARY] == 126
 
     assert len(central.hm_devices) == 362
-    assert len(central.hm_entities) == 4651
+    assert len(central.hm_entities) == 4655
     assert len(data) == 362
     assert len(custom_entities) == 289
     assert len(ce_channels) == 101
     assert len(entity_types) == 6
-    assert len(parameters) == 178
+    assert len(parameters) == 179
 
 
 @pytest.mark.asyncio