New: Threema Integration

.strict-typing

@@ -550,6 +550,7 @@ homeassistant.components.telegram_bot.*
 homeassistant.components.teslemetry.*
 homeassistant.components.text.*
 homeassistant.components.thethingsnetwork.*
+homeassistant.components.threema.*
 homeassistant.components.threshold.*
 homeassistant.components.tibber.*
 homeassistant.components.tile.*

homeassistant/components/threema/__init__.py

@@ -0,0 +1,60 @@
+"""The Threema Gateway integration."""
+
+from __future__ import annotations
+
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+
+from .client import ThreemaAPIClient, ThreemaAuthError, ThreemaConnectionError
+from .const import CONF_API_SECRET, CONF_GATEWAY_ID, CONF_PRIVATE_KEY, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+PLATFORMS: list[Platform] = [Platform.NOTIFY]
+
+type ThreemaConfigEntry = ConfigEntry[ThreemaAPIClient]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ThreemaConfigEntry) -> bool:
+    """Set up Threema Gateway from a config entry."""
+    client = ThreemaAPIClient(
+        hass,
+        gateway_id=entry.data[CONF_GATEWAY_ID],
+        api_secret=entry.data[CONF_API_SECRET],
+        private_key=entry.data.get(CONF_PRIVATE_KEY),
+    )
+
+    try:
+        await client.validate_credentials()
+    except ThreemaAuthError as err:
+        raise ConfigEntryAuthFailed(
+            translation_domain=DOMAIN,
+            translation_key="invalid_auth",
+        ) from err
+    except ThreemaConnectionError as err:
+        raise ConfigEntryNotReady(
+            translation_domain=DOMAIN,
+            translation_key="cannot_connect",
+        ) from err
+
+    entry.runtime_data = client
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    entry.async_on_unload(entry.add_update_listener(_async_update_listener))
+    return True
+
+
+async def _async_update_listener(
+    hass: HomeAssistant, entry: ThreemaConfigEntry
+) -> None:
+    """Reload entry when config is updated (e.g. subentry added/removed)."""
+    await hass.config_entries.async_reload(entry.entry_id)
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ThreemaConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/threema/client.py

@@ -0,0 +1,133 @@
+"""Threema Gateway API client."""
+
+from __future__ import annotations
+
+import logging
+
+from threema.gateway import Connection, GatewayError, key
+from threema.gateway.e2e import TextMessage
+from threema.gateway.exception import GatewayServerError
+from threema.gateway.simple import TextMessage as SimpleTextMessage
+
+from homeassistant.core import HomeAssistant
+
+_LOGGER = logging.getLogger(__name__)
+
+# HTTP 401 from Threema Gateway means invalid credentials
+_HTTP_UNAUTHORIZED = 401
+
+
+class ThreemaConnectionError(Exception):
+    """Error to indicate a connection issue with the Threema Gateway."""
+
+
+class ThreemaAuthError(Exception):
+    """Error to indicate invalid credentials for the Threema Gateway."""
+
+
+class ThreemaSendError(Exception):
+    """Error to indicate a message send failure."""
+
+
+class ThreemaAPIClient:
+    """Threema Gateway API client."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        gateway_id: str,
+        api_secret: str,
+        private_key: str | None = None,
+    ) -> None:
+        """Initialize the Threema API client."""
+        self.hass = hass
+        self.gateway_id = gateway_id
+        self.api_secret = api_secret
+        self.private_key = private_key
+
+    def _get_connection(self) -> Connection:
+        """Get a Threema Gateway connection.
+
+        Note: Connection manages its own aiohttp session lifecycle.
+        Do not pass HA's shared session as Connection will close it.
+        """
+        return Connection(
+            identity=self.gateway_id,
+            secret=self.api_secret,
+            key=self.private_key,
+        )
+
+    async def validate_credentials(self) -> None:
+        """Validate the Gateway credentials by checking credits.
+
+        Raises ThreemaAuthError for invalid credentials.
+        Raises ThreemaConnectionError for other failures.
+        """
+        try:
+            async with self._get_connection() as conn:
+                remaining_credits = await conn.get_credits()
+                _LOGGER.debug(
+                    "Gateway credentials validated, credits: %s",
+                    remaining_credits,
+                )
+        except GatewayServerError as err:
+            if err.status == _HTTP_UNAUTHORIZED:
+                raise ThreemaAuthError("Invalid Threema Gateway credentials") from err
+            raise ThreemaConnectionError(
+                f"Gateway server error validating credentials: {err}"
+            ) from err
+        except GatewayError as err:
+            raise ThreemaConnectionError(
+                f"Gateway error validating credentials: {err}"
+            ) from err
+
+    async def send_text_message(self, recipient_id: str, text: str) -> str:
+        """Send a text message to a Threema ID.
+
+        Returns the message ID on success.
+        Raises ThreemaSendError on failure.
+        """
+        async with self._get_connection() as conn:
+            if self.private_key:
+                _LOGGER.debug("Sending E2E encrypted message to %s", recipient_id)
+                message = TextMessage(
+                    connection=conn,
+                    to_id=recipient_id,
+                    text=text,
+                )
+            else:
+                _LOGGER.debug("Sending simple message to %s", recipient_id)
+                message = SimpleTextMessage(
+                    connection=conn,
+                    to_id=recipient_id,
+                    text=text,
+                )
+
+            try:
+                message_id: str = await message.send()
+            except GatewayServerError as err:
+                if err.status == _HTTP_UNAUTHORIZED:
+                    raise ThreemaAuthError(
+                        "Invalid Threema Gateway credentials"
+                    ) from err
+                raise ThreemaSendError(
+                    f"Gateway server error sending message to {recipient_id}: {err}"
+                ) from err
+            except GatewayError as err:
+                raise ThreemaSendError(
+                    f"Gateway error sending message to {recipient_id}: {err}"
+                ) from err
+
+        _LOGGER.debug("Message sent to %s (ID: %s)", recipient_id, message_id)
+        return message_id
+
+
+def generate_key_pair() -> tuple[str, str]:
+    """Generate a new key pair for E2E encryption using official SDK.
+
+    Returns tuple of (private_key, public_key) as encoded strings.
+    """
+    private_key_obj, public_key_obj = key.Key.generate_pair()
+    private_key_str = key.Key.encode(private_key_obj)
+    public_key_str = key.Key.encode(public_key_obj)
+    return private_key_str, public_key_str