Remove CodeNotary specific integrity checking

The current code is specific to how CodeNotary was doing integrity
checking. A future integrity checking mechanism likely will work
differently (e.g. through EROFS based containers). Remove the current
code to make way for a future implementation.

Author: agners

supervisor/addons/addon.py

@@ -1510,13 +1510,6 @@ def _restore_data():
         _LOGGER.info("Finished restore for add-on %s", self.slug)
         return wait_for_start
 
-    def check_trust(self) -> Awaitable[None]:
-        """Calculate Addon docker content trust.
-
-        Return Coroutine.
-        """
-        return self.instance.check_trust()
-
     @Job(
         name="addon_restart_after_problem",
         throttle_period=WATCHDOG_THROTTLE_PERIOD,

supervisor/api/security.py

@@ -1,19 +1,14 @@
 """Init file for Supervisor Security RESTful API."""
 
-import asyncio
-import logging
 from typing import Any
 
 from aiohttp import web
-import attr
 import voluptuous as vol
 
 from ..const import ATTR_CONTENT_TRUST, ATTR_FORCE_SECURITY, ATTR_PWNED
 from ..coresys import CoreSysAttributes
 from .utils import api_process, api_validate
 
-_LOGGER: logging.Logger = logging.getLogger(__name__)
-
 # pylint: disable=no-value-for-parameter
 SCHEMA_OPTIONS = vol.Schema(
     {
@@ -54,6 +49,9 @@ async def options(self, request: web.Request) -> None:
 
     @api_process
     async def integrity_check(self, request: web.Request) -> dict[str, Any]:
-        """Run backend integrity check."""
-        result = await asyncio.shield(self.sys_security.integrity_check())
-        return attr.asdict(result)
+        """Run backend integrity check.
+
+        CodeNotary integrity checking has been removed. This endpoint now returns
+        an error indicating the feature is currently non-functional.
+        """
+        return {"error": "No integrity checking available"}

supervisor/docker/addon.py

@@ -835,16 +835,6 @@ async def stop(self, remove_container: bool = True) -> None:
         ):
             self.sys_resolution.dismiss_issue(self.addon.device_access_missing_issue)
 
-    async def _validate_trust(self, image_id: str) -> None:
-        """Validate trust of content."""
-        if not self.addon.signed:
-            return
-
-        checksum = image_id.partition(":")[2]
-        return await self.sys_security.verify_content(
-            cast(str, self.addon.codenotary), checksum
-        )
-
     @Job(
         name="docker_addon_hardware_events",
         conditions=[JobCondition.OS_AGENT],

supervisor/docker/homeassistant.py

@@ -5,7 +5,7 @@
 import logging
 import re
 
-from awesomeversion import AwesomeVersion, AwesomeVersionCompareException
+from awesomeversion import AwesomeVersion
 from docker.types import Mount
 
 from ..const import LABEL_MACHINE
@@ -244,13 +244,3 @@ def is_initialize(self) -> Awaitable[bool]:
             self.image,
             self.sys_homeassistant.version,
         )
-
-    async def _validate_trust(self, image_id: str) -> None:
-        """Validate trust of content."""
-        try:
-            if self.version in {None, LANDINGPAGE} or self.version < _VERIFY_TRUST:
-                return
-        except AwesomeVersionCompareException:
-            return
-
-        await super()._validate_trust(image_id)

supervisor/docker/interface.py

@@ -374,17 +374,7 @@ async def process_pull_image_log(reference: PullLogEntry) -> None:
                 platform=MAP_ARCH[image_arch],
             )
 
-            # Validate content
-            try:
-                await self._validate_trust(cast(str, docker_image.id))
-            except CodeNotaryError:
-                with suppress(docker.errors.DockerException):
-                    await self.sys_run_in_executor(
-                        self.sys_docker.images.remove,
-                        image=f"{image}:{version!s}",
-                        force=True,
-                    )
-                raise
+            # CodeNotary content trust validation has been removed
 
             # Tag latest
             if latest:
@@ -755,24 +745,3 @@ def run_inside(self, command: str) -> Awaitable[CommandReturn]:
         return self.sys_run_in_executor(
             self.sys_docker.container_run_inside, self.name, command
         )
-
-    async def _validate_trust(self, image_id: str) -> None:
-        """Validate trust of content."""
-        checksum = image_id.partition(":")[2]
-        return await self.sys_security.verify_own_content(checksum)
-
-    @Job(
-        name="docker_interface_check_trust",
-        on_condition=DockerJobError,
-        concurrency=JobConcurrency.GROUP_REJECT,
-    )
-    async def check_trust(self) -> None:
-        """Check trust of exists Docker image."""
-        try:
-            image = await self.sys_run_in_executor(
-                self.sys_docker.images.get, f"{self.image}:{self.version!s}"
-            )
-        except (docker.errors.DockerException, requests.RequestException):
-            return
-
-        await self._validate_trust(cast(str, image.id))

supervisor/homeassistant/core.py

@@ -420,13 +420,6 @@ def logs(self) -> Awaitable[bytes]:
         """
         return self.instance.logs()
 
-    def check_trust(self) -> Awaitable[None]:
-        """Calculate HomeAssistant docker content trust.
-
-        Return Coroutine.
-        """
-        return self.instance.check_trust()
-
     async def stats(self) -> DockerStats:
         """Return stats of Home Assistant."""
         try:

supervisor/plugins/base.py

@@ -76,13 +76,6 @@ def in_progress(self) -> bool:
         """Return True if a task is in progress."""
         return self.instance.in_progress
 
-    def check_trust(self) -> Awaitable[None]:
-        """Calculate plugin docker content trust.
-
-        Return Coroutine.
-        """
-        return self.instance.check_trust()
-
     def logs(self) -> Awaitable[bytes]:
         """Get docker plugin logs.
 

supervisor/resolution/evaluations/source_mods.py

@@ -1,14 +1,11 @@
 """Evaluation class for Content Trust."""
 
-import errno
 import logging
 from pathlib import Path
 
 from ...const import CoreState
 from ...coresys import CoreSys
-from ...exceptions import CodeNotaryError, CodeNotaryUntrusted
-from ...utils.codenotary import calc_checksum_path_sourcecode
-from ..const import ContextType, IssueType, UnhealthyReason, UnsupportedReason
+from ..const import UnsupportedReason
 from .base import EvaluateBase
 
 _SUPERVISOR_SOURCE = Path("/usr/src/supervisor/supervisor")
@@ -44,29 +41,4 @@ async def evaluate(self) -> bool:
             _LOGGER.warning("Disabled content-trust, skipping evaluation")
             return False
 
-        # Calculate sume of the sourcecode
-        try:
-            checksum = await self.sys_run_in_executor(
-                calc_checksum_path_sourcecode, _SUPERVISOR_SOURCE
-            )
-        except OSError as err:
-            if err.errno == errno.EBADMSG:
-                self.sys_resolution.add_unhealthy_reason(
-                    UnhealthyReason.OSERROR_BAD_MESSAGE
-                )
-
-            self.sys_resolution.create_issue(
-                IssueType.CORRUPT_FILESYSTEM, ContextType.SYSTEM
-            )
-            _LOGGER.error("Can't calculate checksum of source code: %s", err)
-            return False
-
-        # Validate checksum
-        try:
-            await self.sys_security.verify_own_content(checksum)
-        except CodeNotaryUntrusted:
-            return True
-        except CodeNotaryError:
-            pass
-
         return False

supervisor/security/module.py

@@ -11,20 +11,10 @@
     FILE_HASSIO_SECURITY,
 )
 from ..coresys import CoreSys, CoreSysAttributes
-from ..exceptions import (
-    CodeNotaryError,
-    CodeNotaryUntrusted,
-    PwnedError,
-    SecurityJobError,
-)
-from ..jobs.const import JobConcurrency
-from ..jobs.decorator import Job, JobCondition
-from ..resolution.const import ContextType, IssueType, SuggestionType
-from ..utils.codenotary import cas_validate
+from ..exceptions import PwnedError
 from ..utils.common import FileConfiguration
 from ..utils.pwned import check_pwned_password
 from ..validate import SCHEMA_SECURITY_CONFIG
-from .const import ContentTrustResult, IntegrityResult
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
@@ -67,30 +57,6 @@ def pwned(self, value: bool) -> None:
         """Set pwned is enabled/disabled."""
         self._data[ATTR_PWNED] = value
 
-    async def verify_content(self, signer: str, checksum: str) -> None:
-        """Verify content on CAS."""
-        if not self.content_trust:
-            _LOGGER.warning("Disabled content-trust, skip validation")
-            return
-
-        try:
-            await cas_validate(signer, checksum)
-        except CodeNotaryUntrusted:
-            raise
-        except CodeNotaryError:
-            if self.force:
-                raise
-            self.sys_resolution.create_issue(
-                IssueType.TRUST,
-                ContextType.SYSTEM,
-                suggestions=[SuggestionType.EXECUTE_INTEGRITY],
-            )
-            return
-
-    async def verify_own_content(self, checksum: str) -> None:
-        """Verify content from HA org."""
-        return await self.verify_content("[email protected]", checksum)
-
     async def verify_secret(self, pwned_hash: str) -> None:
         """Verify pwned state of a secret."""
         if not self.pwned:
@@ -103,73 +69,3 @@ async def verify_secret(self, pwned_hash: str) -> None:
             if self.force:
                 raise
             return
-
-    @Job(
-        name="security_manager_integrity_check",
-        conditions=[JobCondition.INTERNET_SYSTEM],
-        on_condition=SecurityJobError,
-        concurrency=JobConcurrency.REJECT,
-    )
-    async def integrity_check(self) -> IntegrityResult:
-        """Run a full system integrity check of the platform.
-
-        We only allow to install trusted content.
-        This is a out of the band manual check.
-        """
-        result: IntegrityResult = IntegrityResult()
-        if not self.content_trust:
-            _LOGGER.warning(
-                "Skipping integrity check, content_trust is globally disabled"
-            )
-            return result
-
-        # Supervisor
-        try:
-            await self.sys_supervisor.check_trust()
-            result.supervisor = ContentTrustResult.PASS
-        except CodeNotaryUntrusted:
-            result.supervisor = ContentTrustResult.ERROR
-            self.sys_resolution.create_issue(IssueType.TRUST, ContextType.SUPERVISOR)
-        except CodeNotaryError:
-            result.supervisor = ContentTrustResult.FAILED
-
-        # Core
-        try:
-            await self.sys_homeassistant.core.check_trust()
-            result.core = ContentTrustResult.PASS
-        except CodeNotaryUntrusted:
-            result.core = ContentTrustResult.ERROR
-            self.sys_resolution.create_issue(IssueType.TRUST, ContextType.CORE)
-        except CodeNotaryError:
-            result.core = ContentTrustResult.FAILED
-
-        # Plugins
-        for plugin in self.sys_plugins.all_plugins:
-            try:
-                await plugin.check_trust()
-                result.plugins[plugin.slug] = ContentTrustResult.PASS
-            except CodeNotaryUntrusted:
-                result.plugins[plugin.slug] = ContentTrustResult.ERROR
-                self.sys_resolution.create_issue(
-                    IssueType.TRUST, ContextType.PLUGIN, reference=plugin.slug
-                )
-            except CodeNotaryError:
-                result.plugins[plugin.slug] = ContentTrustResult.FAILED
-
-        # Add-ons
-        for addon in self.sys_addons.installed:
-            if not addon.signed:
-                result.addons[addon.slug] = ContentTrustResult.UNTESTED
-                continue
-            try:
-                await addon.check_trust()
-                result.addons[addon.slug] = ContentTrustResult.PASS
-            except CodeNotaryUntrusted:
-                result.addons[addon.slug] = ContentTrustResult.ERROR
-                self.sys_resolution.create_issue(
-                    IssueType.TRUST, ContextType.ADDON, reference=addon.slug
-                )
-            except CodeNotaryError:
-                result.addons[addon.slug] = ContentTrustResult.FAILED
-
-        return result

supervisor/supervisor.py

@@ -23,8 +23,6 @@
 from .docker.stats import DockerStats
 from .docker.supervisor import DockerSupervisor
 from .exceptions import (
-    CodeNotaryError,
-    CodeNotaryUntrusted,
     DockerError,
     HostAppArmorError,
     SupervisorAppArmorError,
@@ -35,7 +33,6 @@
 from .jobs.const import JobCondition, JobThrottle
 from .jobs.decorator import Job
 from .resolution.const import ContextType, IssueType, UnhealthyReason
-from .utils.codenotary import calc_checksum
 from .utils.sentry import async_capture_exception
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -148,20 +145,6 @@ async def update_apparmor(self) -> None:
                 _LOGGER.error,
             ) from err
 
-        # Validate
-        try:
-            await self.sys_security.verify_own_content(calc_checksum(data))
-        except CodeNotaryUntrusted as err:
-            raise SupervisorAppArmorError(
-                "Content-Trust is broken for the AppArmor profile fetch!",
-                _LOGGER.critical,
-            ) from err
-        except CodeNotaryError as err:
-            raise SupervisorAppArmorError(
-                f"CodeNotary error while processing AppArmor fetch: {err!s}",
-                _LOGGER.error,
-            ) from err
-
         # Load
         temp_dir: TemporaryDirectory | None = None
 
@@ -261,13 +244,6 @@ def logs(self) -> Awaitable[bytes]:
         """
         return self.instance.logs()
 
-    def check_trust(self) -> Awaitable[None]:
-        """Calculate Supervisor docker content trust.
-
-        Return Coroutine.
-        """
-        return self.instance.check_trust()
-
     async def stats(self) -> DockerStats:
         """Return stats of Supervisor."""
         try: