Handle bad DNS configuration while generating certificate (#537)

* Handle bad DNS configuration while generating certificate * adjust * Apply suggestions from code review Co-authored-by: Martin Hjelmare <[email protected]> --------- Co-authored-by: Martin Hjelmare <[email protected]>
NabuCasa · Dec 14, 2023 · 5cf13f7 · 5cf13f7
1 parent 98deb38
commit 5cf13f7
Show file tree

Hide file tree

Showing 5 changed files with 371 additions and 23 deletions.
diff --git a/hass_nabucasa/acme.py b/hass_nabucasa/acme.py
@@ -75,6 +75,16 @@ def __init__(self, cloud: Cloud[_ClientT], domains: list[str], email: str) -> No
         self._domains = domains
         self._email = email
 
+    @property
+    def email(self) -> str:
+        """Return the email."""
+        return self._email
+
+    @property
+    def domains(self) -> list[str]:
+        """Return the domains."""
+        return self._domains
+
     @property
     def path_account_key(self) -> Path:
         """Return path of account key."""

diff --git a/hass_nabucasa/cloud_api.py b/hass_nabucasa/cloud_api.py
@@ -182,3 +182,17 @@ async def async_migrate_paypal_agreement(cloud: Cloud[_ClientT]) -> dict[str, An
     resp.raise_for_status()
     data: dict[str, Any] = await resp.json()
     return data
+
+
+@_check_token
+async def async_resolve_cname(cloud: Cloud[_ClientT], hostname: str) -> list[str]:
+    """Resolve DNS CNAME."""
+    resp = await cloud.websession.post(
+        f"https://{cloud.accounts_server}/instance/resolve_dns_cname",
+        headers={"authorization": cloud.id_token, USER_AGENT: cloud.client.client_name},
+        json={"hostname": hostname},
+    )
+    _do_log_response(resp)
+    resp.raise_for_status()
+    data: list[str] = await resp.json()
+    return data
diff --git a/hass_nabucasa/remote.py b/hass_nabucasa/remote.py
@@ -190,6 +190,12 @@ async def _recreate_backend(self) -> None:
         await asyncio.sleep(5)
         await self.load_backend()
 
+    async def _recreate_acme(self, domains: list[str], email: str) -> None:
+        """Recreate the acme client."""
+        if self._acme and self._acme.certificate_available:
+            await self._acme.reset_acme()
+        self._acme = AcmeHandler(self.cloud, domains, email)
+
     async def load_backend(self) -> bool:
         """Load backend details."""
         try:
@@ -232,20 +238,27 @@ async def load_backend(self) -> bool:
         if self._acme.common_name:
             ca_domains.add(self._acme.common_name)
 
-        if ca_domains and ca_domains != set(domains):
-            _LOGGER.warning("Invalid certificate found for: (%s)", ",".join(ca_domains))
-            await self._acme.reset_acme()
+        if not self._acme.certificate_available or (
+            ca_domains and ca_domains != set(domains)
+        ):
+            for alias in self.alias or []:
+                if not await self._custom_domain_dns_configuration_is_valid(
+                    instance_domain, alias
+                ):
+                    domains.remove(alias)
+
+        if ca_domains != set(domains):
+            if ca_domains:
+                _LOGGER.warning(
+                    "Invalid certificate found for: (%s)", ",".join(ca_domains)
+                )
+            await self._recreate_acme(domains, email)
 
         self._info_loaded.set()
 
-        should_create_cert = not self._acme.certificate_available
+        should_create_cert = await self._should_renew_certificates()
 
-        if (
-            should_create_cert
-            or self._acme.expire_date is None
-            or self._acme.expire_date
-            < utils.utcnow() + timedelta(days=RENEW_IF_EXPIRES_DAYS)
-        ):
+        if should_create_cert:
             try:
                 self._certificate_status = CertificateStatus.GENERATING
                 await self._acme.issue_certificate()
@@ -255,14 +268,14 @@ async def load_backend(self) -> bool:
                     "Home Assistant Cloud",
                     const.MESSAGE_REMOTE_SETUP,
                 )
+                self._certificate_status = CertificateStatus.ERROR
                 return False
 
-            if should_create_cert:
-                self.cloud.client.user_message(
-                    "cloud_remote_acme",
-                    "Home Assistant Cloud",
-                    const.MESSAGE_REMOTE_READY,
-                )
+            self.cloud.client.user_message(
+                "cloud_remote_acme",
+                "Home Assistant Cloud",
+                const.MESSAGE_REMOTE_READY,
+            )
 
         self._certificate_status = CertificateStatus.LOADED
         await self._acme.hardening_files()
@@ -483,14 +496,11 @@ async def _certificate_handler(self) -> None:
                         await asyncio.sleep(10)
                     continue
 
-                # We can not get here without this being set, but mypy does not know that.
-                assert self._acme is not None
+                if TYPE_CHECKING:
+                    assert self._acme is not None
 
                 # Renew certificate?
-                if self._acme.expire_date is not None and (
-                    self._acme.expire_date
-                    > (utils.utcnow() + timedelta(days=RENEW_IF_EXPIRES_DAYS))
-                ):
+                if not await self._should_renew_certificates():
                     continue
 
                 # Renew certificate
@@ -512,6 +522,7 @@ async def _certificate_handler(self) -> None:
                         self._certificate_status = CertificateStatus.ERROR
                     else:
                         meth = _LOGGER.debug
+                        self._certificate_status = CertificateStatus.READY
 
                     meth("Renewal of ACME certificate failed. Trying again later")
 
@@ -524,3 +535,90 @@ async def _certificate_handler(self) -> None:
 
         _LOGGER.debug("Stopping Remote UI loop")
         await self.close_backend()
+
+    async def _check_cname(self, hostname: str) -> list[str]:
+        """Get CNAME records for hostname."""
+        try:
+            return await cloud_api.async_resolve_cname(self.cloud, hostname)
+        except (asyncio.TimeoutError, aiohttp.ClientError):
+            _LOGGER.error("Can't resolve CNAME for %s", hostname)
+        return []
+
+    async def _custom_domain_dns_configuration_is_valid(
+        self, instance_domain: str, custom_domain: str
+    ) -> bool:
+        """Validate custom domain."""
+        # Check primary entry
+        if instance_domain not in await self._check_cname(custom_domain):
+            return False
+
+        # Check LE entry
+        if f"_acme-challenge.{instance_domain}" not in await self._check_cname(
+            f"_acme-challenge.{custom_domain}"
+        ):
+            return False
+
+        return True
+
+    async def _should_renew_certificates(self) -> bool:
+        """Check if certificates should be renewed."""
+        bad_alias = []
+
+        if TYPE_CHECKING:
+            assert self._acme is not None
+            assert self.instance_domain is not None
+
+        if not self._acme.certificate_available:
+            return True
+
+        if self._acme.expire_date is None:
+            return True
+
+        if self._acme.expire_date > (
+            utils.utcnow() + timedelta(days=RENEW_IF_EXPIRES_DAYS)
+        ):
+            return False
+
+        check_alias = [
+            domain for domain in self._acme.domains if domain != self.instance_domain
+        ]
+
+        if not check_alias:
+            return True
+
+        # Check if defined alias is still valid:
+        for alias in check_alias:
+            # Check primary entry
+            if not await self._custom_domain_dns_configuration_is_valid(
+                self.instance_domain, alias
+            ):
+                bad_alias.append(alias)
+
+        if not bad_alias:
+            # No bad configuration detected
+            return True
+
+        if self._acme.expire_date > (
+            utils.utcnow() + timedelta(days=WARN_RENEW_FAILED_DAYS)
+        ):
+            await self.cloud.client.async_create_repair_issue(
+                identifier=f"warn_bad_custom_domain_configuration_{self._acme.expire_date.timestamp()}",
+                translation_key="warn_bad_custom_domain_configuration",
+                placeholders={"custom_domains": ",".join(bad_alias)},
+                severity="warning",
+            )
+            return False
+
+        # Recreate the acme client with working domains
+        await self.cloud.client.async_create_repair_issue(
+            identifier=f"reset_bad_custom_domain_configuration_{self._acme.expire_date.timestamp()}",
+            translation_key="reset_bad_custom_domain_configuration",
+            placeholders={"custom_domains": ",".join(bad_alias)},
+            severity="error",
+        )
+
+        await self._recreate_acme(
+            [domain for domain in self._acme.domains if domain not in bad_alias],
+            self._acme.email,
+        )
+        return True
diff --git a/tests/common.py b/tests/common.py
@@ -163,6 +163,13 @@ def __init__(self):
         self.expire_date = None
         self.fingerprint = None
 
+        self.email = "[email protected]"
+
+    @property
+    def domains(self):
+        """Return all domains."""
+        return self.alternative_names
+
     def set_false(self):
         """Set certificate as not valid."""
         self.is_valid = False