Merge pull request #195 from nnsnodnb/voip-cert-key

Support cert & key separated for VoIP client

README.md

@@ -101,7 +101,7 @@ asyncio.run(
 import asyncio
 from pathlib import Path
 
-from kalyke import VoIPApnsConfig, ApnsPushType, VoIPClient
+from kalyke import VoIPApnsConfig, VoIPClient
 
 client = VoIPClient(
     use_sandbox=True,

examples/voip.py

@@ -1,15 +1,15 @@
 import asyncio
 from pathlib import Path
 
-from kalyke import ApnsConfig, ApnsPushType, VoIPClient
+from kalyke import VoIPApnsConfig, VoIPClient
 
 client = VoIPClient(
     use_sandbox=True,
     auth_key_filepath=Path("/") / "path" / "to" / "YOUR_VOIP_CERTIFICATE.pem",
 )
 
 payload = {"key": "value"}
-config = ApnsConfig(topic="com.example.App.voip", push_type=ApnsPushType.VOIP)
+config = VoIPApnsConfig(topic="com.example.App.voip")
 
 # Send single VoIP notification
 

kalyke/clients/__init__.py

@@ -1,5 +1,6 @@
 import importlib
 import urllib.parse
+from pathlib import Path
 from typing import Any, Dict, Union
 
 from httpx import AsyncClient, Response
@@ -10,6 +11,7 @@
 
 class __Client(object):
     use_sandbox: bool
+    auth_key_filepath: Union[str, Path]
 
     async def send_message(
         self,
@@ -55,3 +57,9 @@ def _handle_error(self, error_json: Dict[str, Any]) -> ApnsProviderException:
         exception_class = getattr(exceptions_module, reason)
 
         return exception_class(error=error_json)
+
+    def _get_auth_key_filepath(self) -> Path:
+        if isinstance(self.auth_key_filepath, Path):
+            return self.auth_key_filepath
+        else:
+            return Path(self.auth_key_filepath)

kalyke/clients/apns.py

@@ -10,19 +10,12 @@
 from . import __Client as BaseClient
 
 
-@dataclass
+@dataclass(frozen=True)
 class ApnsClient(BaseClient):
     use_sandbox: bool
     team_id: str
     auth_key_id: str
     auth_key_filepath: Union[str, Path]
-    _auth_key_filepath: Path = field(init=False)
-
-    def __post_init__(self):
-        if isinstance(self.auth_key_filepath, Path):
-            self._auth_key_filepath = self.auth_key_filepath
-        else:
-            self._auth_key_filepath = Path(self.auth_key_filepath)
 
     def _init_client(self, apns_config: ApnsConfig) -> AsyncClient:
         headers = apns_config.make_headers()
@@ -31,7 +24,7 @@ def _init_client(self, apns_config: ApnsConfig) -> AsyncClient:
         return client
 
     def _make_authorization(self) -> str:
-        auth_key = self._auth_key_filepath.read_text()
+        auth_key = self._get_auth_key_filepath().read_text()
         token = jwt.encode(
             payload={
                 "iss": self.team_id,

kalyke/clients/live_activity.py

@@ -7,7 +7,7 @@
 from .apns import ApnsClient
 
 
-@dataclass
+@dataclass(frozen=True)
 class LiveActivityClient(ApnsClient):
     async def send_message(
         self,

kalyke/clients/voip.py

@@ -1,6 +1,7 @@
+import warnings
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Dict, Union
+from typing import Any, Dict, Optional, Union
 
 import httpx
 from httpx import AsyncClient
@@ -9,17 +10,16 @@
 from . import __Client as BaseClient
 
 
-@dataclass
+@dataclass(frozen=True)
 class VoIPClient(BaseClient):
     use_sandbox: bool
     auth_key_filepath: Union[str, Path]
-    _auth_key_filepath: Path = field(init=False)
+    key_filepath: Optional[Union[str, Path]] = field(default=None)
+    password: Optional[str] = field(default=None)
 
-    def __post_init__(self):
-        if isinstance(self.auth_key_filepath, Path):
-            self._auth_key_filepath = self.auth_key_filepath
-        else:
-            self._auth_key_filepath = Path(self.auth_key_filepath)
+    def __post_init__(self) -> None:
+        if self.key_filepath is None and self.password is not None:
+            warnings.warn(UserWarning("password is ignored because key_filepath is None."))
 
     async def send_message(
         self,
@@ -36,6 +36,10 @@ async def send_message(
     def _init_client(self, apns_config: VoIPApnsConfig) -> AsyncClient:
         headers = apns_config.make_headers()
         context = httpx.create_ssl_context()
-        context.load_cert_chain(self._auth_key_filepath)
+        context.load_cert_chain(
+            certfile=self._get_auth_key_filepath(),
+            keyfile=self.key_filepath,
+            password=self.password,
+        )
         client = AsyncClient(headers=headers, verify=context, http2=True)
         return client

pyproject.toml

@@ -40,6 +40,7 @@ pytest-cov = "^4.0.0"
 coveralls = "^3.3.1"
 pytest-httpx = ">=0.21.3,<0.29.0"
 pytest-asyncio = ">=0.20.3,<0.24.0"
+pem = "^23.1.0"
 
 [tool.poetry-dynamic-versioning]
 enable = true

tests/clients/client/test_get_auth_key_filepath.py

@@ -0,0 +1,22 @@
+from pathlib import Path
+
+import pytest
+
+from kalyke.clients import __Client
+
+
+@pytest.mark.parametrize(
+    "auth_key_filepath",
+    [
+        Path(__file__).parent.parent / "dummy.p8",
+        "../dummy.p8",
+    ],
+    ids=["Path object", "str"],
+)
+def test_get_auth_key_filepath(auth_key_filepath):
+    client = __Client()
+    client.auth_key_filepath = auth_key_filepath
+
+    actual = client._get_auth_key_filepath()
+
+    assert isinstance(actual, Path)

tests/clients/voip/test_init.py

@@ -1,3 +1,9 @@
+import tempfile
+from pathlib import Path
+
+import pem
+import pytest
+
 from kalyke import VoIPClient
 
 
@@ -17,3 +23,51 @@ def test_initialize_with_str(auth_key_filepath):
     )
 
     assert isinstance(client, VoIPClient)
+
+
+def test_initialize_with_key_filepath(auth_key_filepath):
+    cer, key = pem.parse_file(auth_key_filepath)
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = Path(tmpdir)
+        cer_path = tmpdir_path / "cer.pem"
+        cer_path.write_bytes(cer.as_bytes())
+        key_path = tmpdir_path / "key.pem"
+        key_path.write_bytes(key.as_bytes())
+
+        client = VoIPClient(
+            use_sandbox=True,
+            auth_key_filepath=cer_path,
+            key_filepath=key_path,
+        )
+
+        assert isinstance(client, VoIPClient)
+
+
+def test_initialize_with_key_filepath_and_password(auth_key_filepath):
+    cer, key = pem.parse_file(auth_key_filepath)
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = Path(tmpdir)
+        cer_path = tmpdir_path / "cer.pem"
+        cer_path.write_bytes(cer.as_bytes())
+        key_path = tmpdir_path / "key.pem"
+        key_path.write_bytes(key.as_bytes())
+
+        client = VoIPClient(
+            use_sandbox=True,
+            auth_key_filepath=cer_path,
+            key_filepath=key_path,
+            password="password",
+        )
+
+    assert isinstance(client, VoIPClient)
+
+
+def test_initialize_with_password(auth_key_filepath):
+    with pytest.warns(UserWarning, match="password is ignored because key_filepath is None."):
+        client = VoIPClient(
+            use_sandbox=True,
+            auth_key_filepath=auth_key_filepath,
+            password="password",
+        )
+
+        assert isinstance(client, VoIPClient)