address copilot comments

Author: Lash-L

src/roborock_local_server/certs.py

@@ -56,7 +56,9 @@ def _needs_refresh(self) -> bool:
         except Exception:
             return True
         deadline = datetime.now(timezone.utc) + timedelta(days=self.config.tls.renew_days_before)
-        return cert.not_valid_after_utc <= deadline
+        if cert.not_valid_after_utc <= deadline:
+            return True
+        return not self._certificate_matches_configuration(cert)
 
     def _read_cloudflare_token(self) -> str:
         token = self.paths.cloudflare_token_file.read_text(encoding="utf-8").strip()
@@ -96,6 +98,36 @@ def _redact_command(command: list[str]) -> str:
                 redact_next = True
         return " ".join(redacted)
 
+    @staticmethod
+    def _redact_text(text: str, sensitive_values: Iterable[str]) -> str:
+        redacted = text
+        for value in sorted({item for item in sensitive_values if item}, key=len, reverse=True):
+            redacted = redacted.replace(value, "<redacted>")
+        return redacted
+
+    @staticmethod
+    def _sensitive_values(command: list[str], env: dict[str, str]) -> list[str]:
+        sensitive_values: list[str] = []
+        for index, part in enumerate(command[:-1]):
+            if part in {"--eab-kid", "--eab-hmac-key"}:
+                sensitive_values.append(command[index + 1])
+        cloudflare_token = env.get("CF_Token", "").strip()
+        if cloudflare_token:
+            sensitive_values.append(cloudflare_token)
+        return sensitive_values
+
+    def _certificate_matches_configuration(self, cert: x509.Certificate) -> bool:
+        try:
+            san_names = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value.get_values_for_type(
+                x509.DNSName
+            )
+        except x509.ExtensionNotFound:
+            return False
+        actual_domains = {name.strip().lower() for name in san_names if name.strip()}
+        _primary_domain, issue_domains = self._certificate_domains()
+        expected_domains = {domain.strip().lower() for domain in issue_domains if domain.strip()}
+        return actual_domains == expected_domains
+
     def _run_acme(self, args: Iterable[str]) -> None:
         self.paths.acme_dir.mkdir(parents=True, exist_ok=True)
         if not ACME_SH_PATH.exists():
@@ -111,6 +143,7 @@ def _run_acme(self, args: Iterable[str]) -> None:
             self.config.tls.acme_server,
         ]
         display_command = self._redact_command(command)
+        sensitive_values = self._sensitive_values(command, env)
         LOG.info("Running ACME command: %s", display_command)
         result = subprocess.run(
             command,
@@ -121,7 +154,7 @@ def _run_acme(self, args: Iterable[str]) -> None:
             check=False,
         )
         if result.stdout.strip():
-            LOG.info("ACME output:\n%s", result.stdout.strip())
+            LOG.info("ACME output:\n%s", self._redact_text(result.stdout.strip(), sensitive_values))
         if result.returncode != 0:
             raise RuntimeError(f"ACME command failed ({result.returncode}): {display_command}")
 

src/roborock_local_server/ha_addon.py

@@ -106,6 +106,13 @@ def _require_pin(value: object, *, field_name: str) -> str:
     return text
 
 
+def _normalize_acme_server(value: object, *, field_name: str) -> str:
+    normalized = str(value or "").strip().lower() or "zerossl"
+    if normalized not in {"zerossl", "actalis"}:
+        raise ValueError(f"{field_name} must be 'zerossl' or 'actalis'")
+    return normalized
+
+
 def _load_options(path: Path) -> dict[str, Any]:
     if not path.exists():
         return {}
@@ -163,7 +170,7 @@ def _render_config_toml(
         raise ValueError("tls_mode must be 'provided' or 'cloudflare_acme'")
     tls_base_domain = str(merged.get("tls_base_domain", "") or "").strip()
     tls_email = str(merged.get("tls_email", "") or "").strip()
-    acme_server = str(merged.get("acme_server", "zerossl") or "zerossl").strip().lower() or "zerossl"
+    acme_server = _normalize_acme_server(merged.get("acme_server", "zerossl"), field_name="acme_server")
     acme_eab_kid = str(merged.get("acme_eab_kid", "") or "").strip()
     acme_eab_hmac_key = str(merged.get("acme_eab_hmac_key", "") or "").strip()
     cloudflare_token = str(merged.get("cloudflare_token", "") or "").strip()

tests/test_certs.py

@@ -1,12 +1,39 @@
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 import logging
 import subprocess
 
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.x509.oid import NameOID
 import pytest
 from roborock_local_server.certs import CertificateManager
 from roborock_local_server.config import load_config, resolve_paths
 
 
+def _write_certificate(
+    cert_path: Path,
+    *,
+    common_name: str,
+    san_names: list[str],
+) -> None:
+    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    now = datetime.now(timezone.utc)
+    certificate = (
+        x509.CertificateBuilder()
+        .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)]))
+        .issuer_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)]))
+        .public_key(private_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now - timedelta(days=1))
+        .not_valid_after(now + timedelta(days=30))
+        .add_extension(x509.SubjectAlternativeName([x509.DNSName(name) for name in san_names]), critical=False)
+        .sign(private_key=private_key, algorithm=hashes.SHA256())
+    )
+    cert_path.write_bytes(certificate.public_bytes(encoding=serialization.Encoding.PEM))
+
+
 def test_certificate_manager_passes_actalis_eab_to_register_account(tmp_path: Path) -> None:
     config_file = tmp_path / "config.toml"
     config_file.write_text(
@@ -222,7 +249,11 @@ def test_run_acme_redacts_eab_credentials_in_logs_and_errors(
     manager = CertificateManager(config=config, paths=paths)
 
     def fake_run(*args: object, **kwargs: object) -> subprocess.CompletedProcess[str]:
-        return subprocess.CompletedProcess(args=args[0], returncode=1, stdout="failed")
+        return subprocess.CompletedProcess(
+            args=args[0],
+            returncode=1,
+            stdout="failed with kid-123 hmac-456 and cloudflare-token",
+        )
 
     monkeypatch.setattr("roborock_local_server.certs.ACME_SH_PATH", tmp_path / "acme.sh")
     (tmp_path / "acme.sh").write_text("#!/bin/sh\n", encoding="utf-8")
@@ -245,6 +276,55 @@ def fake_run(*args: object, **kwargs: object) -> subprocess.CompletedProcess[str
     combined_logs = "\n".join(caplog.messages)
     assert "kid-123" not in combined_logs
     assert "hmac-456" not in combined_logs
+    assert "cloudflare-token" not in combined_logs
     assert "<redacted>" in combined_logs
     assert "kid-123" not in str(excinfo.value)
     assert "hmac-456" not in str(excinfo.value)
+
+
+def test_certificate_manager_refreshes_when_cert_domains_do_not_match_config(tmp_path: Path) -> None:
+    config_file = tmp_path / "config.toml"
+    config_file.write_text(
+        """
+[network]
+stack_fqdn = "api-roborock.example.com"
+
+[broker]
+mode = "embedded"
+
+[storage]
+data_dir = "data"
+
+[tls]
+mode = "cloudflare_acme"
+base_domain = "example.com"
+email = "acme@example.com"
+cloudflare_token_file = "secrets/cloudflare_token"
+acme_server = "actalis"
+acme_eab_kid = "kid-123"
+acme_eab_hmac_key = "hmac-456"
+
+[admin]
+password_hash = "pbkdf2_sha256$600000$abc$def"
+session_secret = "abcdefghijklmnopqrstuvwxyz123456"
+protocol_login_email = "user@example.com"
+protocol_login_pin_hash = "pbkdf2_sha256$600000$ghi$jkl"
+        """.strip(),
+        encoding="utf-8",
+    )
+    config = load_config(config_file)
+    paths = resolve_paths(config_file, config)
+    paths.cert_file.parent.mkdir(parents=True, exist_ok=True)
+    paths.key_file.parent.mkdir(parents=True, exist_ok=True)
+    _write_certificate(paths.cert_file, common_name="example.com", san_names=["example.com", "*.example.com"])
+    paths.key_file.write_text("key", encoding="utf-8")
+    manager = CertificateManager(config=config, paths=paths)
+    called = {"value": False}
+
+    def fake_provision() -> None:
+        called["value"] = True
+
+    manager._provision_or_renew = fake_provision  # type: ignore[method-assign]
+
+    assert manager.ensure_certificate() is True
+    assert called["value"] is True

tests/test_ha_addon.py

@@ -315,6 +315,31 @@ def test_write_config_from_home_assistant_options_actalis_writes_eab(tmp_path: P
     assert hmac_path.read_text(encoding="utf-8") == "hmac-456"
 
 
+def test_write_config_from_home_assistant_options_rejects_invalid_acme_server(tmp_path: Path) -> None:
+    options_path = tmp_path / "options.json"
+    config_path = tmp_path / "config.toml"
+
+    _write_options(
+        options_path,
+        {
+            "stack_fqdn": "api-roborock.example.com",
+            "tls_mode": "provided",
+            "acme_server": "bad-ca",
+            "cert_file": "/ssl/fullchain.pem",
+            "key_file": "/ssl/privkey.pem",
+            "admin_password": "secret",
+            "protocol_login_email": "user@example.com",
+            "protocol_login_pin": "654321",
+        },
+    )
+
+    with pytest.raises(ValueError, match="acme_server must be 'zerossl' or 'actalis'"):
+        write_config_from_home_assistant_options(
+            options_path=options_path,
+            config_path=config_path,
+        )
+
+
 def test_write_config_from_home_assistant_options_rejects_external_tls(tmp_path: Path) -> None:
     options_path = tmp_path / "options.json"
     config_path = tmp_path / "config.toml"

tests/test_onboarding_cli.py

@@ -416,9 +416,9 @@ def test_onboarding_server_normalization_defaults_to_port_555() -> None:
 
 
 def test_onboarding_server_normalization_enforces_32_char_limit() -> None:
-    assert sanitize_stack_server("roborockss.luke-lashley.com:555") == "roborockss.luke-lashley.com:555/"
+    assert sanitize_stack_server("abcdefghijklmno.example.com:555") == "abcdefghijklmno.example.com:555/"
     with pytest.raises(ValueError, match="token.r must be at most 32 characters, got 33"):
-        sanitize_stack_server("roborocksss.luke-lashley.com:555")
+        sanitize_stack_server("abcdefghijklmnop.example.com:555")
 
 
 def test_remote_onboarding_api_uses_custom_port_base_url() -> None:

tests/test_onboarding_gui.py

@@ -40,6 +40,6 @@ def test_gui_server_normalization_rejects_non_numeric_port() -> None:
 
 
 def test_gui_server_normalization_enforces_32_char_limit() -> None:
-    assert sanitize_stack_server("roborockss.luke-lashley.com:555") == "roborockss.luke-lashley.com:555/"
+    assert sanitize_stack_server("abcdefghijklmno.example.com:555") == "abcdefghijklmno.example.com:555/"
     with pytest.raises(ValueError, match="token.r must be at most 32 characters, got 33"):
-        sanitize_stack_server("roborocksss.luke-lashley.com:555")
+        sanitize_stack_server("abcdefghijklmnop.example.com:555")