diff --git a/README.md b/README.md index 477d2a8..73cdfc4 100644 --- a/README.md +++ b/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, diff --git a/examples/voip.py b/examples/voip.py index 3a75e94..ae31d62 100644 --- a/examples/voip.py +++ b/examples/voip.py @@ -1,7 +1,7 @@ import asyncio from pathlib import Path -from kalyke import ApnsConfig, ApnsPushType, VoIPClient +from kalyke import VoIPApnsConfig, VoIPClient client = VoIPClient( use_sandbox=True, @@ -9,7 +9,7 @@ ) 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 diff --git a/kalyke/clients/__init__.py b/kalyke/clients/__init__.py index aa6bb26..f17a938 100644 --- a/kalyke/clients/__init__.py +++ b/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) diff --git a/kalyke/clients/apns.py b/kalyke/clients/apns.py index 560de3d..73947fd 100644 --- a/kalyke/clients/apns.py +++ b/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, diff --git a/kalyke/clients/live_activity.py b/kalyke/clients/live_activity.py index 288a18a..cf406d7 100644 --- a/kalyke/clients/live_activity.py +++ b/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, diff --git a/kalyke/clients/voip.py b/kalyke/clients/voip.py index aaa4a56..c892932 100644 --- a/kalyke/clients/voip.py +++ b/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 diff --git a/poetry.lock b/poetry.lock index e3a7344..e33e885 100644 --- a/poetry.lock +++ b/poetry.lock @@ -666,6 +666,23 @@ files = [ {file = "pathspec-0.12.1.tar.gz", hash = "sha256:a482d51503a1ab33b1c67a6c3813a26953dbdc71c31dacaef9a838c4e29f5712"}, ] +[[package]] +name = "pem" +version = "23.1.0" +description = "PEM file parsing in Python." +optional = false +python-versions = ">=3.8" +files = [ + {file = "pem-23.1.0-py3-none-any.whl", hash = "sha256:78bbb1e75b737891350cb9499cbba31da5d59545f360f44163c0bc751cad55d3"}, + {file = "pem-23.1.0.tar.gz", hash = "sha256:06503ff2441a111f853ce4e8b9eb9d5fedb488ebdbf560115d3dd53a1b4afc73"}, +] + +[package.extras] +dev = ["pem[tests,types]", "twisted[tls]"] +docs = ["furo", "myst-parser", "sphinx", "sphinx-notfound-page", "twisted[tls]"] +tests = ["certifi", "coverage[toml] (>=5.0.2)", "pretend", "pyopenssl", "pytest"] +types = ["mypy", "twisted", "types-pyopenssl"] + [[package]] name = "platformdirs" version = "4.1.0" @@ -895,4 +912,4 @@ zstd = ["zstandard (>=0.18.0)"] [metadata] lock-version = "2.0" python-versions = "^3.9" -content-hash = "99a037374423c55da87d91dda0c710b971ec28d4577cc448185ad94aa28ca681" +content-hash = "d380fdabb162afc955917a2148e41445fd926117adb6a2fd86659c909bd868ed" diff --git a/pyproject.toml b/pyproject.toml index 1a9db5f..4209ee0 100644 --- a/pyproject.toml +++ b/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 diff --git a/tests/clients/client/test_get_auth_key_filepath.py b/tests/clients/client/test_get_auth_key_filepath.py new file mode 100644 index 0000000..5912128 --- /dev/null +++ b/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) diff --git a/tests/clients/voip/test_init.py b/tests/clients/voip/test_init.py index 6970ffd..25ec68a 100644 --- a/tests/clients/voip/test_init.py +++ b/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)