Force / Enforce security if service is not available (#2744)

* Force / Enforce security if service is not available

* add options

* Add tests

* force security on test

* force security add-on validation

* Adjust style like codenotary

* Different exception type for backend error

* Adjust messages

* add comments

* ditch, not needed

* Address comment

* fix build

.github/workflows/builder.yml

@@ -114,28 +114,30 @@ jobs:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_TOKEN }}
 
+      - name: Login to GitHub Container Registry
+        if: needs.init.outputs.publish == 'true'
+        uses: docker/login-action@v1
+        with:
+          registry: ghcr.io
+          username: ${{ secrets.GIT_USER }}
+          password: ${{ secrets.GIT_TOKEN }}
+
       - name: Set build arguments
         if: needs.init.outputs.publish == 'false'
         run: echo "BUILD_ARGS=--test" >> $GITHUB_ENV
 
       - name: Build supervisor
-        uses: home-assistant/builder@2021.02.0
+        uses: home-assistant/builder@2021.03.3
         with:
           args: |
             $BUILD_ARGS \
             --${{ matrix.arch }} \
             --target /data \
+            --with-codenotary "${{ secrets.VCN_USER }}" "${{ secrets.VCN_PASSWORD }}" "${{ secrets.VCN_ORG }}" \
+            --validate-from "${{ secrets.VCN_ORG }}" \
+            --validate-cache "${{ secrets.VCN_ORG }}" \
             --generic ${{ needs.init.outputs.version }}
 
-      - name: Signing image
-        if: needs.init.outputs.publish == 'true'
-        uses: home-assistant/actions/helpers/codenotary@master
-        with:
-          source: docker://homeassistant/${{ matrix.arch }}-hassio-supervisor:${{ needs.init.outputs.version }}
-          user: ${{ secrets.VCN_USER }}
-          password: ${{ secrets.VCN_PASSWORD }}
-          organisation: ${{ secrets.VCN_ORG }}
-
   codenotary:
     name: CodeNotary signature
     needs: init
@@ -196,7 +198,7 @@ jobs:
         uses: actions/checkout@v2
 
       - name: Build the Supervisor
-        uses: home-assistant/builder@2021.02.0
+        uses: home-assistant/builder@2021.03.3
         with:
           args: |
             --test \

build.json

@@ -1,11 +1,12 @@
 {
   "image": "homeassistant/{arch}-hassio-supervisor",
+  "shadow_repository": "ghcr.io/home-assistant",
   "build_from": {
-    "aarch64": "homeassistant/aarch64-base-python:3.8-alpine3.13",
+    "aarch64": "ghcr.io/home-assistant/aarch64-base-python:3.8-alpine3.13",
-    "armhf": "homeassistant/armhf-base-python:3.8-alpine3.13",
+    "armhf": "ghcr.io/home-assistant/armhf-base-python:3.8-alpine3.13",
-    "armv7": "homeassistant/armv7-base-python:3.8-alpine3.13",
+    "armv7": "ghcr.io/home-assistant/armv7-base-python:3.8-alpine3.13",
-    "amd64": "homeassistant/amd64-base-python:3.8-alpine3.13",
+    "amd64": "ghcr.io/home-assistant/amd64-base-python:3.8-alpine3.13",
-    "i386": "homeassistant/i386-base-python:3.8-alpine3.13"
+    "i386": "ghcr.io/home-assistant/i386-base-python:3.8-alpine3.13"
   },
   "args": {
     "VCN_VERSION": "0.9.4"

supervisor/api/addons.py

@@ -103,7 +103,8 @@ from ..const import (
 )
 from ..coresys import CoreSysAttributes
 from ..docker.stats import DockerStats
-from ..exceptions import APIError, APIForbidden
+from ..exceptions import APIError, APIForbidden, PwnedError, PwnedSecret
+from ..utils.pwned import check_pwned_password
 from ..validate import docker_ports
 from .utils import api_process, api_process_raw, api_validate
 
@@ -338,12 +339,30 @@ class APIAddons(CoreSysAttributes):
         """Validate user options for add-on."""
         addon = self._extract_addon_installed(request)
         data = {ATTR_MESSAGE: "", ATTR_VALID: True}
+
+        # Validate config
         try:
             addon.schema(addon.options)
         except vol.Invalid as ex:
             data[ATTR_MESSAGE] = humanize_error(addon.options, ex)
             data[ATTR_VALID] = False
 
+        # Validate security
+        if self.sys_config.force_security:
+            for secret in addon.pwned:
+                try:
+                    await check_pwned_password(self.sys_websession, secret)
+                    continue
+                except PwnedSecret:
+                    data[ATTR_MESSAGE] = "Add-on use pwned secrets!"
+                except PwnedError as err:
+                    data[
+                        ATTR_MESSAGE
+                    ] = f"Error happening on pwned secrets check: {err!s}!"
+
+                data[ATTR_VALID] = False
+                break
+
         return data
 
     @api_process

supervisor/api/supervisor.py

@@ -21,6 +21,7 @@ from ..const import (
     ATTR_DEBUG_BLOCK,
     ATTR_DESCRIPTON,
     ATTR_DIAGNOSTICS,
+    ATTR_FORCE_SECURITY,
     ATTR_HEALTHY,
     ATTR_ICON,
     ATTR_IP_ADDRESS,
@@ -65,6 +66,7 @@ SCHEMA_OPTIONS = vol.Schema(
         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(),
     }
 )
 
@@ -115,6 +117,7 @@ class APISupervisor(CoreSysAttributes):
             ATTR_DEBUG_BLOCK: self.sys_config.debug_block,
             ATTR_DIAGNOSTICS: self.sys_config.diagnostics,
             ATTR_CONTENT_TRUST: self.sys_config.content_trust,
+            ATTR_FORCE_SECURITY: self.sys_config.force_security,
             ATTR_ADDONS: list_addons,
             ATTR_ADDONS_REPOSITORIES: self.sys_config.addons_repositories,
         }
@@ -148,6 +151,9 @@ class APISupervisor(CoreSysAttributes):
         if ATTR_CONTENT_TRUST in body:
             self.sys_config.content_trust = body[ATTR_CONTENT_TRUST]
 
+        if ATTR_FORCE_SECURITY in body:
+            self.sys_config.force_security = body[ATTR_FORCE_SECURITY]
+
         if ATTR_ADDONS_REPOSITORIES in body:
             new = set(body[ATTR_ADDONS_REPOSITORIES])
             await asyncio.shield(self.sys_store.update_repositories(new))

supervisor/config.py

@@ -13,6 +13,7 @@ from .const import (
     ATTR_DEBUG,
     ATTR_DEBUG_BLOCK,
     ATTR_DIAGNOSTICS,
+    ATTR_FORCE_SECURITY,
     ATTR_LAST_BOOT,
     ATTR_LOGGING,
     ATTR_TIMEZONE,
@@ -157,6 +158,16 @@ class CoreConfig(FileConfiguration):
         """Set content trust is enabled/disabled."""
         self._data[ATTR_CONTENT_TRUST] = value
 
+    @property
+    def force_security(self) -> bool:
+        """Return if force security is enabled/disabled."""
+        return self._data[ATTR_FORCE_SECURITY]
+
+    @force_security.setter
+    def force_security(self, value: bool) -> None:
+        """Set force security is enabled/disabled."""
+        self._data[ATTR_FORCE_SECURITY] = value
+
     @property
     def path_supervisor(self) -> Path:
         """Return Supervisor data path."""

supervisor/const.py

@@ -314,6 +314,7 @@ ATTR_WATCHDOG = "watchdog"
 ATTR_WEBUI = "webui"
 ATTR_WIFI = "wifi"
 ATTR_CONTENT_TRUST = "content_trust"
+ATTR_FORCE_SECURITY = "force_security"
 
 PROVIDE_SERVICE = "provide"
 NEED_SERVICE = "need"

supervisor/coresys.py

@@ -13,7 +13,7 @@ import sentry_sdk
 from .config import CoreConfig
 from .const import ENV_SUPERVISOR_DEV
 from .docker import DockerAPI
-from .exceptions import CodeNotaryUntrusted
+from .exceptions import CodeNotaryError, CodeNotaryUntrusted
 from .resolution.const import UnhealthyReason
 from .utils.codenotary import vcn_validate
 
@@ -628,6 +628,11 @@ class CoreSysAttributes:
 
         try:
             await vcn_validate(checksum, path, org="home-assistant.io")
-        except CodeNotaryUntrusted:
+        except CodeNotaryUntrusted as err:
             self.sys_resolution.unhealthy = UnhealthyReason.UNTRUSTED
+            self.sys_capture_exception(err)
             raise
+        except CodeNotaryError:
+            if self.sys_config.force_security:
+                raise
+            return

supervisor/exceptions.py

@@ -321,6 +321,10 @@ class PwnedError(HassioError):
     """Errors while checking pwned passwords."""
 
 
+class PwnedSecret(PwnedError):
+    """Pwned secrets found."""
+
+
 class PwnedConnectivityError(PwnedError):
     """Connectivity errors while checking pwned passwords."""
 
@@ -336,6 +340,10 @@ class CodeNotaryUntrusted(CodeNotaryError):
     """Error on untrusted content."""
 
 
+class CodeNotaryBackendError(CodeNotaryError):
+    """CodeNotary backend error happening."""
+
+
 # docker/api
 
 

supervisor/resolution/checks/addon_pwned.py

@@ -1,11 +1,10 @@
 """Helpers to check core security."""
-from contextlib import suppress
 from datetime import timedelta
 from typing import List, Optional
 
 from ...const import AddonState, CoreState
 from ...coresys import CoreSys
-from ...exceptions import PwnedConnectivityError, PwnedError
+from ...exceptions import PwnedConnectivityError, PwnedError, PwnedSecret
 from ...jobs.const import JobCondition, JobExecutionLimit
 from ...jobs.decorator import Job
 from ...utils.pwned import check_pwned_password
@@ -38,27 +37,26 @@ class CheckAddonPwned(CheckBase):
             # check passwords
             for secret in secrets:
                 try:
-                    if not await check_pwned_password(self.sys_websession, secret):
+                    await check_pwned_password(self.sys_websession, secret)
-                        continue
                 except PwnedConnectivityError:
                     self.sys_supervisor.connectivity = False
                     return
+                except PwnedSecret:
+                    # Check possible suggestion
+                    if addon.state == AddonState.STARTED:
+                        suggestions = [SuggestionType.EXECUTE_STOP]
+                    else:
+                        suggestions = None
+
+                    self.sys_resolution.create_issue(
+                        IssueType.PWNED,
+                        ContextType.ADDON,
+                        reference=addon.slug,
+                        suggestions=suggestions,
+                    )
+                    break
                 except PwnedError:
-                    continue
+                    pass
-
-                # Check possible suggestion
-                if addon.state == AddonState.STARTED:
-                    suggestions = [SuggestionType.EXECUTE_STOP]
-                else:
-                    suggestions = None
-
-                self.sys_resolution.create_issue(
-                    IssueType.PWNED,
-                    ContextType.ADDON,
-                    reference=addon.slug,
-                    suggestions=suggestions,
-                )
-                break
 
     @Job(conditions=[JobCondition.INTERNET_SYSTEM])
     async def approve_check(self, reference: Optional[str] = None) -> bool:
@@ -76,10 +74,12 @@ class CheckAddonPwned(CheckBase):
 
         # Check if still pwned
         for secret in secrets:
-            with suppress(PwnedError):
+            try:
-                if not await check_pwned_password(self.sys_websession, secret):
+                await check_pwned_password(self.sys_websession, secret)
-                    continue
+            except PwnedSecret:
-            return True
+                return True
+            except PwnedError:
+                pass
 
         return False
 

supervisor/utils/codenotary.py

@@ -9,7 +9,7 @@ from typing import Optional, Set, Tuple, Union
 
 import async_timeout
 
-from ..exceptions import CodeNotaryError, CodeNotaryUntrusted
+from ..exceptions import CodeNotaryBackendError, CodeNotaryError, CodeNotaryUntrusted
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
@@ -87,7 +87,7 @@ async def vcn_validate(
         ) from err
 
     if _ATTR_ERROR in data_json:
-        raise CodeNotaryError(data_json[_ATTR_ERROR], _LOGGER.warning)
+        raise CodeNotaryBackendError(data_json[_ATTR_ERROR], _LOGGER.warning)
 
     if data_json[_ATTR_VERIFICATION][_ATTR_STATUS] == 0:
         _CACHE.add((checksum, path, org, signer))

supervisor/utils/pwned.py

@@ -6,7 +6,7 @@ from typing import Set
 
 import aiohttp
 
-from ..exceptions import PwnedConnectivityError, PwnedError
+from ..exceptions import PwnedConnectivityError, PwnedError, PwnedSecret
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 _API_CALL: str = "https://api.pwnedpasswords.com/range/{hash}"
@@ -14,14 +14,14 @@ _API_CALL: str = "https://api.pwnedpasswords.com/range/{hash}"
 _CACHE: Set[str] = set()
 
 
-async def check_pwned_password(websession: aiohttp.ClientSession, sha1_pw: str) -> bool:
+async def check_pwned_password(websession: aiohttp.ClientSession, sha1_pw: str) -> None:
     """Check if password is pwned."""
     sha1_pw = sha1_pw.upper()
 
     # Chech hit cache
     sha1_short = sha1_pw[:5]
     if sha1_short in _CACHE:
-        return True
+        raise PwnedSecret()
 
     try:
         async with websession.get(
@@ -38,11 +38,9 @@ async def check_pwned_password(websession: aiohttp.ClientSession, sha1_pw: str)
             if not sha1_pw.endswith(line.split(":")[0]):
                 continue
             _CACHE.add(sha1_short)
-            return True
+            raise PwnedSecret()
 
     except (aiohttp.ClientError, asyncio.TimeoutError) as err:
         raise PwnedConnectivityError(
             f"Can't fetch HIBP data: {err}", _LOGGER.warning
         ) from err
-
-    return False

supervisor/validate.py

@@ -16,6 +16,7 @@ from .const import (
     ATTR_DEBUG_BLOCK,
     ATTR_DIAGNOSTICS,
     ATTR_DNS,
+    ATTR_FORCE_SECURITY,
     ATTR_HASSOS,
     ATTR_HOMEASSISTANT,
     ATTR_IMAGE,
@@ -149,6 +150,7 @@ SCHEMA_SUPERVISOR_CONFIG = vol.Schema(
         vol.Optional(ATTR_DEBUG_BLOCK, default=False): vol.Boolean(),
         vol.Optional(ATTR_DIAGNOSTICS, default=None): vol.Maybe(vol.Boolean()),
         vol.Optional(ATTR_CONTENT_TRUST, default=True): vol.Boolean(),
+        vol.Optional(ATTR_FORCE_SECURITY, default=False): vol.Boolean(),
     },
     extra=vol.REMOVE_EXTRA,
 )

tests/api/test_supervisor.py

@@ -0,0 +1,25 @@
+"""Test Supervisor API."""
+
+import pytest
+
+from supervisor.coresys import CoreSys
+
+
+@pytest.mark.asyncio
+async def test_api_supervisor_options_force_security(api_client, coresys: CoreSys):
+    """Test supervisor options force security."""
+    assert not coresys.config.force_security
+
+    await api_client.post("/supervisor/options", json={"force_security": True})
+
+    assert coresys.config.force_security
+
+
+@pytest.mark.asyncio
+async def test_api_supervisor_options_content_trust(api_client, coresys: CoreSys):
+    """Test supervisor options content trust."""
+    assert coresys.config.content_trust
+
+    await api_client.post("/supervisor/options", json={"content_trust": False})
+
+    assert not coresys.config.content_trust

tests/resolution/check/test_check_addon_pwned.py

@@ -4,6 +4,7 @@ from unittest.mock import AsyncMock, patch
 
 from supervisor.const import AddonState, CoreState
 from supervisor.coresys import CoreSys
+from supervisor.exceptions import PwnedSecret
 from supervisor.resolution.checks.addon_pwned import CheckAddonPwned
 from supervisor.resolution.const import IssueType, SuggestionType
 
@@ -36,7 +37,7 @@ async def test_check(coresys: CoreSys):
 
     with patch(
         "supervisor.resolution.checks.addon_pwned.check_pwned_password",
-        AsyncMock(return_value=True),
+        AsyncMock(side_effect=PwnedSecret()),
     ) as mock:
         await addon_pwned.run_check.__wrapped__(addon_pwned)
         assert not mock.called
@@ -44,7 +45,7 @@ async def test_check(coresys: CoreSys):
     addon.pwned.add("123456")
     with patch(
         "supervisor.resolution.checks.addon_pwned.check_pwned_password",
-        AsyncMock(return_value=False),
+        AsyncMock(return_value=None),
     ) as mock:
         await addon_pwned.run_check.__wrapped__(addon_pwned)
         assert mock.called
@@ -53,7 +54,7 @@ async def test_check(coresys: CoreSys):
 
     with patch(
         "supervisor.resolution.checks.addon_pwned.check_pwned_password",
-        AsyncMock(return_value=True),
+        AsyncMock(side_effect=PwnedSecret()),
     ) as mock:
         await addon_pwned.run_check.__wrapped__(addon_pwned)
         assert mock.called
@@ -76,20 +77,20 @@ async def test_approve(coresys: CoreSys):
 
     with patch(
         "supervisor.resolution.checks.addon_pwned.check_pwned_password",
-        AsyncMock(return_value=True),
+        AsyncMock(side_effect=PwnedSecret()),
     ):
         assert await addon_pwned.approve_check(reference=addon.slug)
 
     with patch(
         "supervisor.resolution.checks.addon_pwned.check_pwned_password",
-        AsyncMock(return_value=False),
+        AsyncMock(return_value=None),
     ):
         assert not await addon_pwned.approve_check(reference=addon.slug)
 
     addon.is_installed = False
     with patch(
         "supervisor.resolution.checks.addon_pwned.check_pwned_password",
-        AsyncMock(return_value=True),
+        AsyncMock(side_effect=PwnedSecret()),
     ):
         assert not await addon_pwned.approve_check(reference=addon.slug)
 

tests/test_updater.py

@@ -11,6 +11,7 @@ URL_TEST = "https://version.home-assistant.io/stable.json"
 async def test_fetch_versions(coresys: CoreSys) -> None:
     """Test download and sync version."""
 
+    coresys.config.force_security = True
     await coresys.updater.fetch_data()
 
     async with coresys.websession.get(URL_TEST) as request: