# retoor <retoor@molodetz.nl>
import hashlib
import json
import logging
import string
import time
from typing import Any
import httpx
import jwt
from devplacepy.config import SECONDS_PER_DAY
from devplacepy.database import get_setting
from devplacepy.push.providers.base import (
ACCEPTED,
DEAD,
REJECTED,
Delivery,
PushProvider,
)
from devplacepy.services.base import ConfigField
from devplacepy.utils import DEFAULT_PUSH_URL, PUSH_ICON, generate_uid
logger = logging.getLogger(__name__)
TEAM_ID_KEY = "push_apns_team_id"
KEY_ID_KEY = "push_apns_key_id"
AUTH_KEY_KEY = "push_apns_auth_key"
TOPIC_KEY = "push_apns_topic"
ENVIRONMENT_KEY = "push_apns_environment"
PROVIDER_LABEL = "Apple Push (APNs)"
DEFAULT_ENVIRONMENT = "production"
HOSTS = {
"production": "api.push.apple.com",
"sandbox": "api.sandbox.push.apple.com",
}
ENVIRONMENT_OPTIONS = [
{"value": "production", "label": "Production"},
{"value": "sandbox", "label": "Sandbox"},
]
TOKEN_REFRESH_SECONDS = 45 * 60
TOKEN_MIN_LENGTH = 64
TOKEN_MAX_LENGTH = 200
THREAD_ID = "devplace-notification"
PUSH_TYPE = "alert"
PRIORITY = "10"
DEAD_REASONS = frozenset(
{
"BadDeviceToken",
"DeviceTokenNotForTopic",
"ExpiredToken",
"Unregistered",
"TopicDisallowed",
}
)
_token_state: dict[str, Any] = {}
def _setting(key: str) -> str:
return get_setting(key, "").strip()
def _environment() -> str:
value = _setting(ENVIRONMENT_KEY) or DEFAULT_ENVIRONMENT
return value if value in HOSTS else DEFAULT_ENVIRONMENT
def host() -> str:
return HOSTS[_environment()]
def _fingerprint(team_id: str, key_id: str, auth_key: str) -> str:
return hashlib.sha256(f"{team_id}:{key_id}:{auth_key}".encode("utf-8")).hexdigest()
def provider_token(team_id: str, key_id: str, auth_key: str) -> str:
fingerprint = _fingerprint(team_id, key_id, auth_key)
issued_at = int(time.time())
state = _token_state.get("current")
if (
state
and state["fingerprint"] == fingerprint
and issued_at - state["issued_at"] < TOKEN_REFRESH_SECONDS
):
if state["token"] is None:
raise ValueError(state["error"])
return state["token"]
try:
token = jwt.encode(
{"iss": team_id, "iat": issued_at},
auth_key,
algorithm="ES256",
headers={"kid": key_id},
)
except Exception as exc:
message = f"APNs auth key is not usable: {exc}"
_token_state["current"] = {
"token": None,
"error": message,
"issued_at": issued_at,
"fingerprint": fingerprint,
}
logger.error(message)
raise ValueError(message) from exc
_token_state["current"] = {
"token": token,
"error": "",
"issued_at": issued_at,
"fingerprint": fingerprint,
}
return token
def _reason(response: httpx.Response) -> str:
try:
body = response.json()
except ValueError:
return ""
if isinstance(body, dict):
return str(body.get("reason", ""))
return ""
class ApnsProvider(PushProvider):
name = "apns"
label = PROVIDER_LABEL
config_fields = [
ConfigField(
TEAM_ID_KEY,
"Team ID",
type="str",
default="",
help="Ten character Apple Developer team identifier, used as the token iss claim.",
group=PROVIDER_LABEL,
),
ConfigField(
KEY_ID_KEY,
"Key ID",
type="str",
default="",
help="Ten character identifier of the APNs auth key, sent as the token kid header.",
group=PROVIDER_LABEL,
),
ConfigField(
AUTH_KEY_KEY,
"Auth key (.p8)",
type="text",
default="",
secret=True,
help="Contents of the APNs .p8 signing key, including the BEGIN and END lines. Leave blank to keep the stored key.",
group=PROVIDER_LABEL,
),
ConfigField(
TOPIC_KEY,
"Topic",
type="str",
default="",
help="Bundle identifier of the receiving app, sent as the apns-topic header.",
group=PROVIDER_LABEL,
),
ConfigField(
ENVIRONMENT_KEY,
"Environment",
type="select",
default=DEFAULT_ENVIRONMENT,
options=ENVIRONMENT_OPTIONS,
help="Production delivers to App Store builds, sandbox to development builds.",
group=PROVIDER_LABEL,
),
]
def is_configured(self) -> bool:
return bool(
_setting(TEAM_ID_KEY)
and _setting(KEY_ID_KEY)
and _setting(AUTH_KEY_KEY)
and _setting(TOPIC_KEY)
)
def parse_registration(self, body: dict[str, Any]) -> dict[str, Any] | None:
token = body.get("token")
if not isinstance(token, str):
return None
token = token.strip()
if not TOKEN_MIN_LENGTH <= len(token) <= TOKEN_MAX_LENGTH:
return None
if any(character not in string.hexdigits for character in token):
return None
return {"token": token}
def prepare(self, payload: dict[str, Any]) -> str:
return json.dumps(
{
"aps": {
"alert": {
"title": payload.get("title") or "DevPlace",
"body": payload.get("message") or "",
},
"sound": "default",
"thread-id": THREAD_ID,
},
"url": payload.get("url") or DEFAULT_PUSH_URL,
"icon": payload.get("icon") or PUSH_ICON,
}
)
def headers(self) -> dict[str, str]:
return {
"authorization": f"bearer {provider_token(_setting(TEAM_ID_KEY), _setting(KEY_ID_KEY), _setting(AUTH_KEY_KEY))}",
"apns-topic": _setting(TOPIC_KEY),
"apns-push-type": PUSH_TYPE,
"apns-priority": PRIORITY,
"apns-expiration": str(int(time.time()) + SECONDS_PER_DAY),
"apns-id": generate_uid(),
"content-type": "application/json",
}
async def deliver(
self, client: httpx.AsyncClient, registration: dict[str, Any], prepared: str
) -> Delivery:
token = (registration.get("token") or "").strip()
if not token:
return Delivery(DEAD, "missing device token")
try:
headers = self.headers()
response = await client.post(
f"https://{host()}/3/device/{token}",
headers=headers,
content=prepared.encode("utf-8"),
)
except (httpx.HTTPError, ValueError) as exc:
return Delivery(REJECTED, str(exc))
if response.status_code == 200:
return Delivery(ACCEPTED)
reason = _reason(response)
detail = f"{response.status_code} {reason}".strip()
if response.status_code == 410 or reason in DEAD_REASONS:
return Delivery(DEAD, detail)
return Delivery(REJECTED, detail)
View Git Blame Copy Permalink