commit 3ce4de773233fad0b30fa5f8cba4efd0bf8dff8f
parent 63a69222fd57e29ab967e8d52793f741ed47bc49
Author: Antoine A <>
Date: Tue, 27 Oct 2026 14:41:09 +0100
apns-relay: fix many remaining issues issues
Diffstat:
12 files changed, 137 insertions(+), 61 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -2661,6 +2661,7 @@ dependencies = [
"anyhow",
"aws-lc-rs",
"axum",
+ "base64 0.23.1",
"clap",
"compact_str",
"http",
diff --git a/Cargo.toml b/Cargo.toml
@@ -20,6 +20,7 @@ lto = "fat"
codegen-units = 1
[workspace.dependencies]
+base64 = "0.23"
thiserror = "2.0"
serde_json = "1.0"
serde = { version = "1.0", features = ["derive"] }
diff --git a/common/taler-common/Cargo.toml b/common/taler-common/Cargo.toml
@@ -14,7 +14,7 @@ doctest = false
[dependencies]
glob = "0.3"
indexmap = "2.7"
-base64 = "0.23"
+base64.workspace = true
tempfile.workspace = true
jiff.workspace = true
serde.workspace = true
diff --git a/debian/etc/taler-apns-relay/conf.d/apns-relay-worker.conf b/debian/etc/taler-apns-relay/conf.d/apns-relay-worker.conf
@@ -2,6 +2,9 @@
[apns-relay-worker]
+# APNs environment: production or sandbox
+# SERVER = production
+
# How often should worker wakeup all registered devices
FREQUENCY = 4h
diff --git a/taler-apns-relay/Cargo.toml b/taler-apns-relay/Cargo.toml
@@ -12,6 +12,7 @@ repository.workspace = true
doctest = false
[dependencies]
+base64.workspace = true
sqlx.workspace = true
serde.workspace = true
serde_json.workspace = true
diff --git a/taler-apns-relay/apns-relay.conf b/taler-apns-relay/apns-relay.conf
@@ -1,4 +1,7 @@
[apns-relay-worker]
+# APNs environment: production or sandbox
+SERVER = production
+
# Path to the Apple Push Notification private key
KEY_FILE = ${APNS_RELAY_HOME}/key.p8
@@ -35,4 +38,4 @@ BIND_TO = 0.0.0.0
CONFIG = postgres:///taler-apns-relay
# Where are the SQL files to setup our tables?
-SQL_DIR = ${DATADIR}/sql/
-\ No newline at end of file
+SQL_DIR = ${DATADIR}/sql/
diff --git a/taler-apns-relay/src/api.rs b/taler-apns-relay/src/api.rs
@@ -26,10 +26,12 @@ use jiff::Timestamp;
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use taler_api::{
- error::ApiResult,
+ error::{ApiResult, failure},
extract::{Path, Req},
};
+use taler_common::error_code::ErrorCode;
+
use crate::db;
pub struct RelayApi {
@@ -65,8 +67,17 @@ pub fn router(state: Arc<RelayApi>) -> Router {
"/devices",
post(
async |State(state): State<Arc<RelayApi>>,
- Req(req): Req<DeviceRegistrationRequest>| {
- db::register(&state.db, &req.token, &Timestamp::now()).await?;
+ Req(DeviceRegistrationRequest { token}): Req<DeviceRegistrationRequest>| {
+ if token.is_empty()
+ || !token.len().is_multiple_of(2)
+ || !token.bytes().all(|b| b.is_ascii_hexdigit()) {
+ return Err(failure(
+ ErrorCode::GENERIC_PARAMETER_MALFORMED,
+ "token must be a nonempty, even-length ASCII hexadecimal string",
+ )
+ .with_path("token"));
+ }
+ db::register(&state.db, &token, &Timestamp::now()).await?;
ApiResult::Ok(NoContent)
},
),
@@ -123,32 +134,29 @@ mod test {
server
.post("/devices")
.json(json!({
- "token": "device1"
+ "token": "aa01"
}))
.await
.assert_no_content();
server
.post("/devices")
.json(json!({
- "token": "device1"
+ "token": "aa01"
}))
.await
.assert_no_content();
server
.post("/devices")
.json(json!({
- "token": "device2"
+ "token": "aa02"
}))
.await
.assert_no_content();
- assert_eq!(
- all_registrations(&pool).await.unwrap(),
- &["device1", "device2"]
- );
+ assert_eq!(all_registrations(&pool).await.unwrap(), &["aa01", "aa02"]);
- server.delete("/devices/device1").await.assert_no_content();
- server.delete("/devices/device3").await.assert_no_content();
- assert_eq!(all_registrations(&pool).await.unwrap(), &["device2"]);
+ server.delete("/devices/aa01").await.assert_no_content();
+ server.delete("/devices/aa03").await.assert_no_content();
+ assert_eq!(all_registrations(&pool).await.unwrap(), &["aa02"]);
}
}
diff --git a/taler-apns-relay/src/apns.rs b/taler-apns-relay/src/apns.rs
@@ -20,6 +20,7 @@ use aws_lc_rs::{
rand::SystemRandom,
signature::{ECDSA_P256_SHA256_FIXED_SIGNING, EcdsaKeyPair},
};
+use base64::{display::Base64Display, engine::general_purpose::URL_SAFE_NO_PAD};
use compact_str::CompactString;
use http::{StatusCode, header::CONTENT_TYPE};
use http_body_util::{BodyExt, Full};
@@ -29,10 +30,10 @@ use hyper_util::rt::{TokioExecutor, TokioTimer};
use jiff::{SignedDuration, Timestamp};
use rustls_pki_types::{PrivateKeyDer, pem::PemObject};
use serde::Deserialize;
-use taler_common::{encoding::base64, error::FmtSource};
+use taler_common::error::FmtSource;
use taler_macros::EnumMeta;
-use crate::config::ApnsConfig;
+use crate::config::{ApnsConfig, ApnsServer};
/// Raw JSON body returned by APNs when a push is rejected.
#[derive(Debug, Deserialize)]
@@ -172,6 +173,7 @@ pub struct Client {
Full<Bytes>,
>,
key_pair: EcdsaKeyPair,
+ server: ApnsServer,
key_id: CompactString,
team_id: CompactString,
bundle_id: CompactString,
@@ -182,6 +184,7 @@ pub struct Client {
impl Client {
pub fn new(cfg: &ApnsConfig) -> anyhow::Result<Self> {
let ApnsConfig {
+ server,
key_path,
key_id,
team_id,
@@ -219,7 +222,7 @@ impl Client {
// Make a signature
let now = Timestamp::now();
- let token = Self::create_token(&key_pair, key_id, team_id, &now)?;
+ let token = Self::create_token(&key_pair, key_id, team_id, &now);
// Prepare the TLS client config
let tls = rustls::ClientConfig::builder()
@@ -247,6 +250,7 @@ impl Client {
Ok(Self {
http,
key_pair,
+ server: *server,
key_id: key_id.clone(),
team_id: team_id.clone(),
bundle_id: bundle_id.clone(),
@@ -259,19 +263,36 @@ impl Client {
let now = Timestamp::now();
// Token expire after an hour
if now.duration_since(self.issued_at) > SignedDuration::from_mins(55) {
- self.token =
- Self::create_token(&self.key_pair, &self.key_id, &self.team_id, &now).unwrap();
- self.issued_at = now;
+ self.refresh_token();
}
- let path = format!(
- "https://{}/3/device/{device_token}",
- "api.sandbox.push.apple.com"
- );
+ let result = self.send_once(device_token).await;
+ if matches!(
+ &result,
+ Err(ApnsError::Err {
+ reason: Reason::ExpiredProviderToken,
+ ..
+ })
+ ) {
+ self.refresh_token();
+ return self.send_once(device_token).await;
+ }
+ result
+ }
+
+ fn refresh_token(&mut self) {
+ let now = Timestamp::now();
+ self.token = Self::create_token(&self.key_pair, &self.key_id, &self.team_id, &now);
+ self.issued_at = now;
+ }
+ async fn send_once(&self, device_token: &str) -> Result<(), ApnsError> {
let req = hyper::Request::builder()
.method(Method::POST)
- .uri(&path)
+ .uri(format!(
+ "https://{}/3/device/{device_token}",
+ self.server.host()
+ ))
.header(CONTENT_TYPE, "application/json")
.header("apns-push-type", "background")
.header("apns-priority", "5")
@@ -352,16 +373,23 @@ impl Client {
key_id: &str,
team_id: &str,
issued_at: &Timestamp,
- ) -> Result<Box<str>, anyhow::Error> {
+ ) -> Box<str> {
let headers = format!(r#"{{"alg":"ES256","kid":"{key_id}"}}"#);
let payload = format!(r#"{{"iss":"{team_id}","iat":{}}}"#, issued_at.as_second());
let token = format!(
"{}.{}",
- base64::fmt(headers.as_bytes()),
- base64::fmt(payload.as_bytes())
+ Base64Display::new(headers.as_bytes(), &URL_SAFE_NO_PAD),
+ Base64Display::new(payload.as_bytes(), &URL_SAFE_NO_PAD)
);
- let signature = key_pair.sign(&SystemRandom::new(), token.as_bytes())?;
+ let signature = key_pair
+ .sign(&SystemRandom::new(), token.as_bytes())
+ .unwrap();
- Ok(format!("Bearer {}.{}", token, base64::fmt(signature.as_ref())).into_boxed_str())
+ format!(
+ "Bearer {}.{}",
+ token,
+ Base64Display::new(signature.as_ref(), &URL_SAFE_NO_PAD)
+ )
+ .into_boxed_str()
}
}
diff --git a/taler-apns-relay/src/config.rs b/taler-apns-relay/src/config.rs
@@ -18,7 +18,10 @@ use std::time::Duration;
use compact_str::CompactString;
use taler_api::{Serve, config::DbCfg};
-use taler_common::config::{CfgErr, Config};
+use taler_common::{
+ config::{CfgErr, Config},
+ map_config,
+};
pub fn parse_db_cfg(cfg: &Config) -> Result<DbCfg, CfgErr> {
DbCfg::parse(cfg.section("apns-relaydb-postgres"))
@@ -39,7 +42,23 @@ impl ServeCfg {
}
}
+#[derive(Debug, Clone, Copy)]
+pub enum ApnsServer {
+ Production,
+ Sandbox,
+}
+
+impl ApnsServer {
+ pub fn host(self) -> &'static str {
+ match self {
+ Self::Production => "api.push.apple.com",
+ Self::Sandbox => "api.sandbox.push.apple.com",
+ }
+ }
+}
+
pub struct ApnsConfig {
+ pub server: ApnsServer,
pub key_path: String,
pub key_id: CompactString,
pub team_id: CompactString,
@@ -50,6 +69,11 @@ impl ApnsConfig {
pub fn parse(cfg: &Config) -> Result<Self, CfgErr> {
let sect = cfg.section("apns-relay-worker");
Ok(Self {
+ server: map_config!(§, "APNs server", "server",
+ "production" => { ApnsServer::Production },
+ "sandbox" => { ApnsServer::Sandbox },
+ )
+ .default(ApnsServer::Production)?,
key_path: sect.path("key_file").require()?,
key_id: sect.cstr("key_id").require()?,
team_id: sect.cstr("team_id").require()?,
diff --git a/taler-apns-relay/src/db.rs b/taler-apns-relay/src/db.rs
@@ -16,7 +16,7 @@
use jiff::Timestamp;
use sqlx::PgPool;
-use taler_api::db::BindHelper;
+use taler_api::{db::BindHelper, serialized};
use taler_common::config::Config;
use crate::config::parse_db_cfg;
@@ -39,32 +39,35 @@ pub async fn dbinit(cfg: &Config, reset: bool) -> anyhow::Result<PgPool> {
/// Register a new device
pub async fn register(db: &PgPool, token: &str, now: &Timestamp) -> sqlx::Result<()> {
- sqlx::query(
- "
+ serialized!(
+ sqlx::query(
+ "
INSERT INTO devices (token, registered_at)
VALUES ($1, $2)
- ON CONFLICT (token) DO NOTHING
+ ON CONFLICT (token) DO UPDATE
+ SET registered_at = GREATEST(devices.registered_at, EXCLUDED.registered_at)
",
- )
- .bind(token)
- .bind_timestamp(now)
- .execute(db)
- .await?;
+ )
+ .bind(token)
+ .bind_timestamp(now)
+ .execute(db)
+ )?;
Ok(())
}
/// Unregister
pub async fn unregister(db: &PgPool, token: &str, before: &Timestamp) -> sqlx::Result<()> {
- sqlx::query(
- "
+ serialized!(
+ sqlx::query(
+ "
DELETE FROM devices
WHERE token = $1 AND registered_at <= $2
",
- )
- .bind(token)
- .bind_timestamp(before)
- .execute(db)
- .await?;
+ )
+ .bind(token)
+ .bind_timestamp(before)
+ .execute(db)
+ )?;
Ok(())
}
diff --git a/taler-apns-relay/src/lib.rs b/taler-apns-relay/src/lib.rs
@@ -37,10 +37,7 @@ pub async fn setup(cfg: &Config, pool: &PgPool, reset: bool) -> anyhow::Result<(
info!(target: "setup", "Check API access and configuration");
let mut client = Client::new(&apns_cfg)?;
- let res = client
- .send("test_device_token")
- .await
- .expect_err("Should fail");
+ let res = client.send("00").await.expect_err("Should fail");
if !matches!(
res,
diff --git a/taler-apns-relay/src/worker.rs b/taler-apns-relay/src/worker.rs
@@ -52,7 +52,7 @@ async fn wakeup(pool: &PgPool, client: &mut Client) -> Result<(), WorkerError> {
// TODO paginate tokens
// TODO send in batches
- for token in tokens {
+ 'devices: for token in tokens {
trace!(target: "worker", "send background notification to {token}");
let mut attempt = 0;
while let Err(e) = client.send(&token).await {
@@ -61,7 +61,6 @@ async fn wakeup(pool: &PgPool, client: &mut Client) -> Result<(), WorkerError> {
match reason {
// Fatal error
Reason::BadCollapseId
- | Reason::BadDeviceToken
| Reason::BadExpirationDate
| Reason::BadMessageId
| Reason::BadPriority
@@ -80,7 +79,6 @@ async fn wakeup(pool: &PgPool, client: &mut Client) -> Result<(), WorkerError> {
Reason::TopicDisallowed
| Reason::BadCertificate
| Reason::BadCertificateEnvironment
- | Reason::ExpiredProviderToken
| Reason::Forbidden
| Reason::InvalidProviderToken
| Reason::MissingProviderToken
@@ -90,24 +88,32 @@ async fn wakeup(pool: &PgPool, client: &mut Client) -> Result<(), WorkerError> {
error!(target: "worker", "config error, check the configuration: {e}");
std::process::exit(6);
}
+ // The client already retried once with a fresh provider token.
+ Reason::ExpiredProviderToken => return Err(e.into()),
// Unregister
- Reason::DeviceTokenNotForTopic
+ Reason::BadDeviceToken
+ | Reason::DeviceTokenNotForTopic
| Reason::ExpiredToken
| Reason::Unregistered => {
db::unregister(
pool,
&token,
×tamp
- .and_then(|s| Timestamp::from_second(s as i64).ok())
+ .and_then(|s| i64::try_from(s).ok())
+ .and_then(|s| Timestamp::from_millisecond(s).ok())
.unwrap_or_else(Timestamp::now),
)
.await?;
- break;
+ continue 'devices;
+ }
+ // Retry this device on the next scheduled pass.
+ Reason::TooManyRequests => {
+ warn!(target: "worker", "notification to {token} rate-limited, deferring to the next pass");
+ continue 'devices;
}
// Wait before retry
Reason::IdleTimeout
| Reason::TooManyProviderTokenUpdates
- | Reason::TooManyRequests
| Reason::InternalServerError
| Reason::ServiceUnavailable
| Reason::Shutdown => {}
@@ -115,8 +121,10 @@ async fn wakeup(pool: &PgPool, client: &mut Client) -> Result<(), WorkerError> {
}
attempt += 1;
+
if attempt >= NB_RETRY {
- error!(target: "worker", "notification to {token} failed more than {NB_RETRY} times, stopping worker");
+ error!(target: "worker", "notification to {token} failed more than {NB_RETRY} times, stopping worker after back-off");
+ tokio::time::sleep(jitter.backoff()).await;
return Err(WorkerError::Apns(e));
} else {
warn!(target: "worker", "notification to {token} failed {e}, retrying after back-off");
