Using CAS for content-trust

.github/workflows/builder.yml

@@ -33,6 +33,7 @@ on:
       - setup.py
 
 env:
+  DEFAULT_PYTHON: 3.9
   BUILD_NAME: supervisor
   BUILD_TYPE: supervisor
   WHEELS_TAG: 3.9-alpine3.14
@@ -138,7 +139,7 @@ jobs:
           CAS_API_KEY: ${{ secrets.CAS_TOKEN }}
 
   codenotary:
-    name: CodeNotary signature
+    name: CAS signature
     needs: init
     runs-on: ubuntu-latest
     steps:
@@ -148,6 +149,20 @@ jobs:
         with:
           fetch-depth: 0
 
+      - name: Set up Python ${{ env.DEFAULT_PYTHON }}
+        if: needs.init.outputs.publish == 'true'
+        uses: actions/[email protected]
+        with:
+          python-version: ${{ env.DEFAULT_PYTHON }}
+
+      - name: Install dirhash and calc hash
+        if: needs.init.outputs.publish == 'true'
+        id: dirhash
+        run: |
+          pip3 install dirhash
+          dir_hash="$(dirhash "${{ github.workspace }}" -a sha256 --match "*.py")"
+          echo "::set-output name=dirhash::${dir_hash}"
+
       - name: Set version
         if: needs.init.outputs.publish == 'true'
         uses: home-assistant/actions/helpers/version@master
@@ -158,10 +173,8 @@ jobs:
         if: needs.init.outputs.publish == 'true'
         uses: home-assistant/actions/helpers/codenotary@master
         with:
-          source: dir://${{ github.workspace }}
-          user: ${{ secrets.VCN_USER }}
-          password: ${{ secrets.VCN_PASSWORD }}
-          organisation: ${{ secrets.VCN_ORG }}
+          source: hash://${{ steps.dirhash.outputs.dirhash }}
+          token: ${{ secrets.CAS_TOKEN }}
 
   version:
     name: Update version

.github/workflows/ci.yaml

@@ -10,7 +10,7 @@ on:
 env:
   DEFAULT_PYTHON: 3.9
   PRE_COMMIT_HOME: ~/.cache/pre-commit
-  DEFAULT_VCN: v0.9.8
+  DEFAULT_CAS: v1.0.1
 
 jobs:
   # Separate job to pre-populate the base dependency cache
@@ -351,10 +351,10 @@ jobs:
         id: python
         with:
           python-version: ${{ matrix.python-version }}
-      - name: Install VCN tools
-        uses: home-assistant/actions/helpers/vcn@master
+      - name: Install CAS tools
+        uses: home-assistant/actions/helpers/cas@master
         with:
-          vcn_version: ${{ env.DEFAULT_VCN }}
+          version: ${{ env.DEFAULT_CAS }}
       - name: Restore Python virtual environment
         id: cache-venv
         uses: actions/[email protected]

Dockerfile

@@ -5,10 +5,12 @@ ENV \
     S6_SERVICES_GRACETIME=10000 \
     SUPERVISOR_API=http://localhost
 
-ARG BUILD_ARCH
-WORKDIR /usr/src
+ARG \
+    BUILD_ARCH \
+    CAS_VERSION
 
 # Install base
+WORKDIR /usr/src
 RUN \
     set -x \
     && apk add --no-cache \
@@ -18,7 +20,20 @@ RUN \
         libffi \
         libpulse \
         musl \
-        openssl
+        openssl \
+    && apk add --no-cache --virtual .build-dependencies \
+        build-base \
+        go \
+    \
+    && git clone -b "v${CAS_VERSION}" --depth 1 \
+        https://github.com/codenotary/cas \
+    && cd cas \
+    && make cas \
+    && mv cas /usr/bin/cas \
+    \
+    && apk del .build-dependencies \
+    && rm -rf /root/go /root/.cache \
+    && rm -rf /usr/src/cas
 
 # Install requirements
 COPY requirements.txt .

build.yaml

@@ -9,6 +9,8 @@ build_from:
 codenotary:
   signer: [email protected]
   base_image: [email protected]
+args:
+  CAS_VERSION: 1.0.1
 labels:
   io.hass.type: supervisor
   org.opencontainers.image.title: Home Assistant Supervisor

requirements.txt

@@ -11,6 +11,7 @@ cpe==1.2.1
 cryptography==36.0.1
 debugpy==1.5.1
 deepmerge==1.0.1
+dirhash==0.2.1
 docker==5.0.3
 gitpython==3.1.26
 jinja2==3.0.3

rootfs/root/.cas-trusted-signing-pub-key

@@ -0,0 +1,4 @@
+-----BEGIN PUBLIC KEY-----
+MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE03LvYuz79GTJx4uKp3w6NrSe5JZI
+iBtgzzYi0YQYtZO/r+xFpgDJEa0gLHkXtl94fpqrFiN89In83lzaszbZtA==
+-----END PUBLIC KEY-----

supervisor/docker/interface.py

@@ -633,7 +633,7 @@ def _validate_trust(
         """Validate trust of content."""
         checksum = image_id.partition(":")[2]
         job = asyncio.run_coroutine_threadsafe(
-            self.sys_security.verify_own_content(checksum=checksum), self.sys_loop
+            self.sys_security.verify_own_content(checksum), self.sys_loop
         )
         job.result(timeout=20)
 

supervisor/resolution/evaluations/source_mods.py

@@ -5,6 +5,7 @@
 from ...const import CoreState
 from ...coresys import CoreSys
 from ...exceptions import CodeNotaryError, CodeNotaryUntrusted
+from ...utils.codenotary import calc_checksum_path_sourcecode
 from ..const import UnsupportedReason
 from .base import EvaluateBase
 
@@ -41,8 +42,13 @@ async def evaluate(self) -> None:
             _LOGGER.warning("Disabled content-trust, skipping evaluation")
             return
 
+        # Calculate sume of the sourcecode
+        checksum = await self.sys_run_in_executor(
+            calc_checksum_path_sourcecode, _SUPERVISOR_SOURCE
+        )
+
         try:
-            await self.sys_security.verify_own_content(path=_SUPERVISOR_SOURCE)
+            await self.sys_security.verify_own_content(checksum)
         except CodeNotaryUntrusted:
             return True
         except CodeNotaryError:

supervisor/security.py

@@ -1,7 +1,6 @@
 """Fetch last versions from webserver."""
 import logging
-from pathlib import Path
-from typing import Awaitable, Optional
+from typing import Awaitable
 
 from .const import (
     ATTR_CONTENT_TRUST,
@@ -11,7 +10,7 @@
 )
 from .coresys import CoreSys, CoreSysAttributes
 from .exceptions import CodeNotaryError, CodeNotaryUntrusted, PwnedError
-from .utils.codenotary import vcn_validate
+from .utils.codenotary import cas_validate
 from .utils.common import FileConfiguration
 from .utils.pwned import check_pwned_password
 from .validate import SCHEMA_SECURITY_CONFIG
@@ -57,16 +56,14 @@ def pwned(self, value: bool) -> None:
         """Set pwned is enabled/disabled."""
         self._data[ATTR_PWNED] = value
 
-    async def verify_own_content(
-        self, checksum: Optional[str] = None, path: Optional[Path] = None
-    ) -> Awaitable[None]:
+    async def verify_own_content(self, checksum: str) -> Awaitable[None]:
         """Verify content from HA org."""
         if not self.content_trust:
             _LOGGER.warning("Disabled content-trust, skip validation")
             return
 
         try:
-            await vcn_validate(checksum, path, org="home-assistant.io")
+            await cas_validate(checksum=checksum, signer="notary@home-assistant.io")
         except CodeNotaryUntrusted:
             raise
         except CodeNotaryError:

supervisor/supervisor.py

@@ -127,7 +127,7 @@ async def update_apparmor(self) -> None:
 
         # Validate
         try:
-            await self.sys_security.verify_own_content(checksum=calc_checksum(data))
+            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!",

supervisor/updater.py

@@ -207,7 +207,7 @@ async def fetch_data(self):
 
         # Validate
         try:
-            await self.sys_security.verify_own_content(checksum=calc_checksum(data))
+            await self.sys_security.verify_own_content(calc_checksum(data))
         except CodeNotaryUntrusted as err:
             raise UpdaterError(
                 "Content-Trust is broken for the version file fetch!", _LOGGER.critical

supervisor/utils/codenotary.py

@@ -1,27 +1,28 @@
 """Small wrapper for CodeNotary."""
-# pylint:  disable=unreachable
 import asyncio
 import hashlib
 import json
 import logging
 from pathlib import Path
 import shlex
-from typing import Optional, Union
+from typing import Final, Union
 
 import async_timeout
+from dirhash import dirhash
 
 from . import clean_env
 from ..exceptions import CodeNotaryBackendError, CodeNotaryError, CodeNotaryUntrusted
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
-_VCN_CMD: str = "vcn authenticate --silent --output json"
-_CACHE: set[tuple[str, Path, str, str]] = set()
+_CAS_CMD: str = (
+    "cas authenticate --signerID {signer} --silent --output json --hash {sum}"
+)
+_CACHE: set[tuple[str, str]] = set()
 
 
-_ATTR_ERROR = "error"
-_ATTR_VERIFICATION = "verification"
-_ATTR_STATUS = "status"
+_ATTR_ERROR: Final = "error"
+_ATTR_STATUS: Final = "status"
 
 
 def calc_checksum(data: Union[str, bytes]) -> str:
@@ -31,36 +32,24 @@ def calc_checksum(data: Union[str, bytes]) -> str:
     return hashlib.sha256(data).hexdigest()
 
 
-async def vcn_validate(
-    checksum: Optional[str] = None,
-    path: Optional[Path] = None,
-    org: Optional[str] = None,
-    signer: Optional[str] = None,
+def calc_checksum_path_sourcecode(folder: Path) -> str:
+    """Calculate checksum for a path source code."""
+    return dirhash(folder.as_posix(), "sha256", match=["*.py"])
+
+
+async def cas_validate(
+    signer: str,
+    checksum: str,
 ) -> None:
     """Validate data against CodeNotary."""
-    return None
-    if (checksum, path, org, signer) in _CACHE:
+    if (checksum, signer) in _CACHE:
         return
-    command = shlex.split(_VCN_CMD)
 
     # Generate command for request
-    if org:
-        command.extend(["--org", org])
-    elif signer:
-        command.extend(["--signerID", signer])
-
-    if checksum:
-        command.extend(["--hash", checksum])
-    elif path:
-        if path.is_dir:
-            command.append(f"dir://{path.as_posix()}")
-        else:
-            command.append(path.as_posix())
-    else:
-        RuntimeError("At least path or checksum need to be set!")
+    command = shlex.split(_CAS_CMD.format(signer=signer, sum=checksum))
 
     # Request notary authorization
-    _LOGGER.debug("Send vcn command: %s", command)
+    _LOGGER.debug("Send cas command: %s", command)
     try:
         proc = await asyncio.create_subprocess_exec(
             *command,
@@ -93,7 +82,7 @@ async def vcn_validate(
     if _ATTR_ERROR in data_json:
         raise CodeNotaryBackendError(data_json[_ATTR_ERROR], _LOGGER.warning)
 
-    if data_json[_ATTR_VERIFICATION][_ATTR_STATUS] == 0:
-        _CACHE.add((checksum, path, org, signer))
+    if data_json[_ATTR_STATUS] == 0:
+        _CACHE.add((checksum, signer))
     else:
         raise CodeNotaryUntrusted()

tests/resolution/evaluation/test_evaluate_source_mods.py

@@ -1,5 +1,7 @@
 """Test evaluation base."""
 # pylint: disable=import-error,protected-access
+import os
+from pathlib import Path
 from unittest.mock import AsyncMock, patch
 
 from supervisor.const import CoreState
@@ -10,21 +12,25 @@
 
 async def test_evaluation(coresys: CoreSys):
     """Test evaluation."""
-    sourcemods = EvaluateSourceMods(coresys)
-    coresys.core.state = CoreState.RUNNING
-
-    assert sourcemods.reason not in coresys.resolution.unsupported
-    coresys.security.verify_own_content = AsyncMock(side_effect=CodeNotaryUntrusted)
-    await sourcemods()
-    assert sourcemods.reason in coresys.resolution.unsupported
-
-    coresys.security.verify_own_content = AsyncMock(side_effect=CodeNotaryError)
-    await sourcemods()
-    assert sourcemods.reason not in coresys.resolution.unsupported
-
-    coresys.security.verify_own_content = AsyncMock()
-    await sourcemods()
-    assert sourcemods.reason not in coresys.resolution.unsupported
+    with patch(
+        "supervisor.resolution.evaluations.source_mods._SUPERVISOR_SOURCE",
+        Path(os.getcwd()),
+    ):
+        sourcemods = EvaluateSourceMods(coresys)
+        coresys.core.state = CoreState.RUNNING
+
+        assert sourcemods.reason not in coresys.resolution.unsupported
+        coresys.security.verify_own_content = AsyncMock(side_effect=CodeNotaryUntrusted)
+        await sourcemods()
+        assert sourcemods.reason in coresys.resolution.unsupported
+
+        coresys.security.verify_own_content = AsyncMock(side_effect=CodeNotaryError)
+        await sourcemods()
+        assert sourcemods.reason not in coresys.resolution.unsupported
+
+        coresys.security.verify_own_content = AsyncMock()
+        await sourcemods()
+        assert sourcemods.reason not in coresys.resolution.unsupported
 
 
 async def test_did_run(coresys: CoreSys):