Remove Codenotary integrity check (#6236)

* Formally deprecate CodeNotary build config

* 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.

* Drop CodeNotary integrity fixups

* Drop unused tests

* Fix pytest

* Fix pytest

* Remove CodeNotary related exceptions and handling

Remove CodeNotary related exceptions and handling from the Docker
interface.

* Drop unnecessary comment

* Remove Codenotary specific IssueType/SuggestionType

* Drop Codenotary specific environment and secret reference

* Remove unused constants

* Introduce APIGone exception for removed APIs

Introduce a new exception class APIGone to indicate that certain API
features have been removed and are no longer available. Update the
security integrity check endpoint to raise this new exception instead
of a generic APIError, providing clearer communication to clients that
the feature has been intentionally removed.

* Drop content trust

A cosign based signature verification will likely be named differently
to avoid confusion with existing implementations. For now, remove the
content trust option entirely.

* Drop code sign test

* Remove source_mods/content_trust evaluations

* Remove content_trust reference in bootstrap.py

* Fix security tests

* Drop unused tests

* Drop codenotary from schema

Since we have "remove extra" in voluptuous, we can remove the
codenotary field from the addon schema.

* Remove content_trust from tests

* Remove content_trust unsupported reason

* Remove unnecessary comment

* Remove unrelated pytest

* Remove unrelated fixtures

Author: agners

.github/workflows/builder.yml

@@ -170,8 +170,6 @@ jobs:
             --target /data \
             --cosign \
             --generic ${{ needs.init.outputs.version }}
-        env:
-          CAS_API_KEY: ${{ secrets.CAS_TOKEN }}
 
   version:
     name: Update version
@@ -293,33 +291,6 @@ jobs:
             exit 1
           fi
 
-      - name: Check the Supervisor code sign
-        if: needs.init.outputs.publish == 'true'
-        run: |
-          echo "Enable Content-Trust"
-          test=$(docker exec hassio_cli ha security options --content-trust=true --no-progress --raw-json | jq -r '.result')
-          if [ "$test" != "ok" ]; then
-            exit 1
-          fi
-
-          echo "Run supervisor health check"
-          test=$(docker exec hassio_cli ha resolution healthcheck --no-progress --raw-json | jq -r '.result')
-          if [ "$test" != "ok" ]; then
-            exit 1
-          fi
-
-          echo "Check supervisor unhealthy"
-          test=$(docker exec hassio_cli ha resolution info --no-progress --raw-json | jq -r '.data.unhealthy[]')
-          if [ "$test" != "" ]; then
-            exit 1
-          fi
-
-          echo "Check supervisor supported"
-          test=$(docker exec hassio_cli ha resolution info --no-progress --raw-json | jq -r '.data.unsupported[]')
-          if [[ "$test" =~ source_mods ]]; then
-            exit 1
-          fi
-
       - name: Create full backup
         id: backup
         run: |

supervisor/addons/addon.py

@@ -1513,13 +1513,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/addons/model.py

@@ -103,7 +103,6 @@
 from .const import (
     ATTR_BACKUP,
     ATTR_BREAKING_VERSIONS,
-    ATTR_CODENOTARY,
     ATTR_PATH,
     ATTR_READ_ONLY,
     AddonBackupMode,
@@ -632,13 +631,8 @@ def with_journald(self) -> bool:
 
     @property
     def signed(self) -> bool:
-        """Return True if the image is signed."""
-        return ATTR_CODENOTARY in self.data
-
-    @property
-    def codenotary(self) -> str | None:
-        """Return Signer email address for CAS."""
-        return self.data.get(ATTR_CODENOTARY)
+        """Currently no signing support."""
+        return False
 
     @property
     def breaking_versions(self) -> list[AwesomeVersion]:

supervisor/addons/validate.py

@@ -207,6 +207,12 @@ def _warn_addon_config(config: dict[str, Any]):
             name,
         )
 
+    if ATTR_CODENOTARY in config:
+        _LOGGER.warning(
+            "Add-on '%s' uses deprecated 'codenotary' field in config. This field is no longer used and will be ignored. Please report this to the maintainer.",
+            name,
+        )
+
     return config
 
 
@@ -417,7 +423,6 @@ def _migrate(config: dict[str, Any]):
         vol.Optional(ATTR_BACKUP, default=AddonBackupMode.HOT): vol.Coerce(
             AddonBackupMode
         ),
-        vol.Optional(ATTR_CODENOTARY): vol.Email(),
         vol.Optional(ATTR_OPTIONS, default={}): dict,
         vol.Optional(ATTR_SCHEMA, default={}): vol.Any(
             vol.Schema({str: SCHEMA_ELEMENT}),

supervisor/api/security.py

@@ -1,24 +1,20 @@
 """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 supervisor.exceptions import APIGone
+
+from ..const import 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(
     {
         vol.Optional(ATTR_PWNED): vol.Boolean(),
-        vol.Optional(ATTR_CONTENT_TRUST): vol.Boolean(),
         vol.Optional(ATTR_FORCE_SECURITY): vol.Boolean(),
     }
 )
@@ -31,7 +27,6 @@ class APISecurity(CoreSysAttributes):
     async def info(self, request: web.Request) -> dict[str, Any]:
         """Return Security information."""
         return {
-            ATTR_CONTENT_TRUST: self.sys_security.content_trust,
             ATTR_PWNED: self.sys_security.pwned,
             ATTR_FORCE_SECURITY: self.sys_security.force,
         }
@@ -43,8 +38,6 @@ async def options(self, request: web.Request) -> None:
 
         if ATTR_PWNED in body:
             self.sys_security.pwned = body[ATTR_PWNED]
-        if ATTR_CONTENT_TRUST in body:
-            self.sys_security.content_trust = body[ATTR_CONTENT_TRUST]
         if ATTR_FORCE_SECURITY in body:
             self.sys_security.force = body[ATTR_FORCE_SECURITY]
 
@@ -54,6 +47,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 gone.
+        """
+        raise APIGone("Integrity check feature has been removed.")

supervisor/api/supervisor.py

@@ -16,14 +16,12 @@
     ATTR_BLK_READ,
     ATTR_BLK_WRITE,
     ATTR_CHANNEL,
-    ATTR_CONTENT_TRUST,
     ATTR_COUNTRY,
     ATTR_CPU_PERCENT,
     ATTR_DEBUG,
     ATTR_DEBUG_BLOCK,
     ATTR_DETECT_BLOCKING_IO,
     ATTR_DIAGNOSTICS,
-    ATTR_FORCE_SECURITY,
     ATTR_HEALTHY,
     ATTR_ICON,
     ATTR_IP_ADDRESS,
@@ -69,8 +67,6 @@
         vol.Optional(ATTR_DEBUG): vol.Boolean(),
         vol.Optional(ATTR_DEBUG_BLOCK): vol.Boolean(),
         vol.Optional(ATTR_DIAGNOSTICS): vol.Boolean(),
-        vol.Optional(ATTR_CONTENT_TRUST): vol.Boolean(),
-        vol.Optional(ATTR_FORCE_SECURITY): vol.Boolean(),
         vol.Optional(ATTR_AUTO_UPDATE): vol.Boolean(),
         vol.Optional(ATTR_DETECT_BLOCKING_IO): vol.Coerce(DetectBlockingIO),
         vol.Optional(ATTR_COUNTRY): str,

supervisor/bootstrap.py

@@ -105,7 +105,6 @@ async def initialize_coresys() -> CoreSys:
 
     if coresys.dev:
         coresys.updater.channel = UpdateChannel.DEV
-        coresys.security.content_trust = False
 
     # Convert datetime
     logging.Formatter.converter = lambda *args: coresys.now().timetuple()

supervisor/docker/addon.py

@@ -846,16 +846,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

@@ -31,15 +31,12 @@
 )
 from ..coresys import CoreSys
 from ..exceptions import (
-    CodeNotaryError,
-    CodeNotaryUntrusted,
     DockerAPIError,
     DockerError,
     DockerJobError,
     DockerLogOutOfOrder,
     DockerNotFound,
     DockerRequestError,
-    DockerTrustError,
 )
 from ..jobs import SupervisorJob
 from ..jobs.const import JOB_GROUP_DOCKER_INTERFACE, JobConcurrency
@@ -425,18 +422,6 @@ 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
-
             # Tag latest
             if latest:
                 _LOGGER.info(
@@ -462,16 +447,6 @@ async def process_pull_image_log(reference: PullLogEntry) -> None:
             raise DockerError(
                 f"Unknown error with {image}:{version!s} -> {err!s}", _LOGGER.error
             ) from err
-        except CodeNotaryUntrusted as err:
-            raise DockerTrustError(
-                f"Pulled image {image}:{version!s} failed on content-trust verification!",
-                _LOGGER.critical,
-            ) from err
-        except CodeNotaryError as err:
-            raise DockerTrustError(
-                f"Error happened on Content-Trust check for {image}:{version!s}: {err!s}",
-                _LOGGER.error,
-            ) from err
         finally:
             if listener:
                 self.sys_bus.remove_listener(listener)
@@ -809,24 +784,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))