Handle UnicodeDecodeError (#4110)

supervisor/store/git.py

@@ -61,7 +61,8 @@ class GitRepo(CoreSysAttributes):
             except (
                 git.InvalidGitRepositoryError,
                 git.NoSuchPathError,
-                git.GitCommandError,
+                git.CommandError,
+                UnicodeDecodeError,
             ) as err:
                 _LOGGER.error("Can't load %s", self.path)
                 raise StoreGitError() from err
@@ -71,7 +72,7 @@ class GitRepo(CoreSysAttributes):
             try:
                 _LOGGER.debug("Integrity check add-on %s repository", self.path)
                 await self.sys_run_in_executor(self.repo.git.execute, ["git", "fsck"])
-            except git.GitCommandError as err:
+            except git.CommandError as err:
                 _LOGGER.error("Integrity check on %s failed: %s.", self.path, err)
                 raise StoreGitError() from err
 
@@ -104,7 +105,8 @@ class GitRepo(CoreSysAttributes):
             except (
                 git.InvalidGitRepositoryError,
                 git.NoSuchPathError,
-                git.GitCommandError,
+                git.CommandError,
+                UnicodeDecodeError,
             ) as err:
                 _LOGGER.error("Can't clone %s repository: %s.", self.url, err)
                 raise StoreGitCloneError() from err
@@ -159,9 +161,10 @@ class GitRepo(CoreSysAttributes):
             except (
                 git.InvalidGitRepositoryError,
                 git.NoSuchPathError,
-                git.GitCommandError,
+                git.CommandError,
                 ValueError,
                 AssertionError,
+                UnicodeDecodeError,
             ) as err:
                 _LOGGER.error("Can't update %s repo: %s.", self.url, err)
                 self.sys_resolution.create_issue(

supervisor/utils/codenotary.py

@@ -25,6 +25,7 @@ _CACHE: set[tuple[str, str]] = set()
 
 _ATTR_ERROR: Final = "error"
 _ATTR_STATUS: Final = "status"
+_FALLBACK_ERROR: Final = "Unknown CodeNotary backend issue"
 
 
 def calc_checksum(data: str | bytes) -> str:
@@ -75,11 +76,14 @@ async def cas_validate(
     # Check if Notarized
     if proc.returncode != 0 and not data:
         if error:
+            try:
                 error = error.decode("utf-8")
+            except UnicodeDecodeError as err:
+                raise CodeNotaryBackendError(_FALLBACK_ERROR, _LOGGER.warning) from err
             if "not notarized" in error:
                 raise CodeNotaryUntrusted()
         else:
-            error = "Unknown CodeNotary backend issue"
+            error = _FALLBACK_ERROR
         raise CodeNotaryBackendError(error, _LOGGER.warning)
 
     # Parse data

tests/store/test_repository_git.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 from pathlib import Path
 from unittest.mock import AsyncMock, patch
 
-from git import GitCommandError, GitError, InvalidGitRepositoryError, NoSuchPathError
+from git import GitCommandError, InvalidGitRepositoryError, NoSuchPathError
 import pytest
 
 from supervisor.coresys import CoreSys
@@ -44,10 +44,15 @@ async def test_git_clone(
 
 @pytest.mark.parametrize(
     "git_error",
-    [InvalidGitRepositoryError(), NoSuchPathError(), GitCommandError("clone")],
+    [
+        InvalidGitRepositoryError(),
+        NoSuchPathError(),
+        GitCommandError("clone"),
+        UnicodeDecodeError("decode", b"", 0, 0, ""),
+    ],
 )
 async def test_git_clone_error(
-    coresys: CoreSys, tmp_path: Path, clone_from: AsyncMock, git_error: GitError
+    coresys: CoreSys, tmp_path: Path, clone_from: AsyncMock, git_error: Exception
 ):
     """Test git clone error."""
     repo = GitRepo(coresys, tmp_path, REPO_URL)
@@ -77,12 +82,11 @@ async def test_git_load(coresys: CoreSys, tmp_path: Path):
         InvalidGitRepositoryError(),
         NoSuchPathError(),
         GitCommandError("init"),
+        UnicodeDecodeError("decode", b"", 0, 0, ""),
         [AsyncMock(), GitCommandError("fsck")],
     ],
 )
-async def test_git_load_error(
+async def test_git_load_error(coresys: CoreSys, tmp_path: Path, git_errors: Exception):
-    coresys: CoreSys, tmp_path: Path, git_errors: GitError | list[GitError | None]
-):
     """Test git load error."""
     repo = GitRepo(coresys, tmp_path, REPO_URL)
 

tests/utils/test_codenotary.py

@@ -31,11 +31,7 @@ def fixture_subprocess_exec(request):
     if response.exception:
         communicate_return = AsyncMock(side_effect=response.exception)
     else:
-        err_return = None
+        communicate_return = AsyncMock(return_value=(response.data, response.error))
-        if response.error:
-            err_return = Mock(decode=Mock(return_value=response.error))
-
-        communicate_return = AsyncMock(return_value=(response.data, err_return))
 
     exec_return = Mock(returncode=response.returncode, communicate=communicate_return)
 
@@ -72,11 +68,25 @@ async def test_invalid_checksum():
         )
 
 
+@pytest.mark.parametrize(
+    "subprocess_exec",
+    [SubprocessResponse(returncode=1, error=b"x is not notarized")],
+)
+async def test_not_notarized_error(subprocess_exec):
+    """Test received a not notarized error response from command."""
+    with pytest.raises(CodeNotaryUntrusted):
+        await cas_validate(
+            "notary@home-assistant.io",
+            "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
+        )
+
+
 @pytest.mark.parametrize(
     "subprocess_exec",
     [
-        SubprocessResponse(returncode=1, error="test"),
+        SubprocessResponse(returncode=1, error=b"test"),
         SubprocessResponse(returncode=0, data='{"error":"asn1: structure error"}'),
+        SubprocessResponse(returncode=1, error="test".encode("utf-16")),
     ],
     indirect=True,
 )