~ knotless Search/
↑↓ move↵ open⇧↵ all results

grain.social/grain
Switch repository
Loading…

photography-focused photo sharing platform

grain.social
1 collaborator
0
2

grain-serve pushes by APNs and FCM

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:push/transports

Only what this pull adds. It is built on grain runs on a Rust server modelled on knotless, without hatk.

crates/grain-serve/src/push/apns.rsnew+183−0
View file
@@ -0,0 +1,183 @@
//! Apple's push service, APNs, for iOS devices: HTTP/2, authenticated by
//! a JWT the team's key signs (ES256), naming the key and the team, and
//! good for up to an hour, so one is made every fifty minutes and used for
//! every push between.
//!
//! A push is an alert: `aps` holds the title and body, the default sound
//! and the badge, and the notification's own fields sit beside it, where
//! the app reads where a tap goes ([`Message::data`]).
//!
//! APNs answers 410 for a token no device answers to any more, which is
//! [`Outcome::Gone`]. A 403 for the JWT (`ExpiredProviderToken`,
//! `InvalidProviderToken`) makes a new one for the next try, and 429s,
//! 5xx and failures to reach it are tried again; any other answer is
//! not.
use super::{Message, Outcome, der_of_pem};
use anyhow::Context;
use grain::crypto::P256Key;
use serde_json::{Value, json};
use std::path::PathBuf;
use std::sync::Mutex;
use std::time::{Duration, Instant};
/// APNs for apps signed for the App Store and TestFlight.
pub const PRODUCTION: &str = "https://api.push.apple.com";
/// APNs for apps built for development.
pub const SANDBOX: &str = "https://api.sandbox.push.apple.com";
/// How long a JWT is used for: APNs refuses one older than an hour, and
/// one made more than once in twenty minutes.
pub const JWT_FOR: Duration = Duration::from_secs(50 * 60);
/// How long a push may take.
pub const TIMEOUT: Duration = Duration::from_secs(15);
/// How grain-serve reaches APNs.
#[derive(Debug, Clone, PartialEq)]
pub struct Config {
/// `GRAIN_APNS_KEY_FILE`: the team's key, a `.p8` file.
pub key_file: PathBuf,
/// `GRAIN_APNS_KEY_ID`: the key's ID, as Apple names it.
pub key_id: String,
/// `GRAIN_APNS_TEAM_ID`: the team's ID.
pub team_id: String,
/// `GRAIN_APNS_TOPIC`: the app's bundle ID.
pub topic: String,
/// [`PRODUCTION`], or [`SANDBOX`] where `GRAIN_APNS_SANDBOX` is `1`.
pub url: String,
}
impl Config {
/// APNs as the environment names it: none without a key file. What it
/// leaves unset is empty, which [`Apns::new`] refuses.
pub fn from_vars(get: impl Fn(&str) -> Option<String>) -> Option<Config> {
let set = |name: &str| get(name).filter(|v| !v.is_empty());
let named = |name: &str| set(name).unwrap_or_default();
Some(Config {
key_file: set("GRAIN_APNS_KEY_FILE")?.into(),
key_id: named("GRAIN_APNS_KEY_ID"),
team_id: named("GRAIN_APNS_TEAM_ID"),
topic: named("GRAIN_APNS_TOPIC"),
url: match set("GRAIN_APNS_SANDBOX").as_deref() {
Some("1") => SANDBOX.into(),
_ => PRODUCTION.into(),
},
})
}
}
/// APNs, ready to push.
pub struct Apns {
http: reqwest::Client,
config: Config,
key: P256Key,
/// The JWT in use, and when it was made.
pub(super) jwt: Mutex<Option<(String, Instant)>>,
}
impl Apns {
/// APNs as `config` names it, its key read.
pub fn new(config: Config) -> anyhow::Result<Apns> {
use p256::pkcs8::DecodePrivateKey;
for (name, value) in [
("GRAIN_APNS_KEY_ID", &config.key_id),
("GRAIN_APNS_TEAM_ID", &config.team_id),
("GRAIN_APNS_TOPIC", &config.topic),
] {
if value.is_empty() {
anyhow::bail!("GRAIN_APNS_KEY_FILE needs {name} too");
}
}
let path = config.key_file.display();
let pem = std::fs::read_to_string(&config.key_file)
.with_context(|| format!("reading the APNs key {path}"))?;
let key = der_of_pem(&pem)
.and_then(|der| p256::ecdsa::SigningKey::from_pkcs8_der(&der).ok())
.with_context(|| format!("{path} is not a P-256 key in PKCS #8 PEM"))?;
let key = P256Key::from_bytes(&key.to_bytes())?;
let http = reqwest::Client::builder()
.http2_prior_knowledge()
.timeout(TIMEOUT)
.build()?;
Ok(Apns {
http,
config,
key,
jwt: Mutex::new(None),
})
}
/// The JWT to push with: the one in use, unless it is [`JWT_FOR`] old.
fn jwt(&self) -> String {
let mut jwt = self.jwt.lock().unwrap();
if let Some((token, made)) = &*jwt
&& made.elapsed() < JWT_FOR
{
return token.clone();
}
let header = json!({ "kid": self.config.key_id });
let claims = json!({ "iss": self.config.team_id, "iat": grain::keys::now_secs() });
let token = self.key.sign_jwt(&header, &claims);
*jwt = Some((token.clone(), Instant::now()));
token
}
/// What `message` is to APNs.
pub fn payload(message: &Message) -> Value {
let mut aps = json!({
"alert": { "title": message.title, "body": message.body },
"sound": "default",
});
if let Some(badge) = message.badge {
aps["badge"] = badge.into();
}
// The data first, so that no field of it stands in for `aps`.
let mut payload: serde_json::Map<String, Value> = message
.data
.iter()
.map(|(k, v)| (k.clone(), v.as_str().into()))
.collect();
payload.insert("aps".into(), aps);
payload.into()
}
/// Pushes `message` to the device `token`.
pub async fn send(&self, token: &str, message: &Message) -> Outcome {
// A device token is hex, and names a path under APNs's.
if token.is_empty() || !token.bytes().all(|b| b.is_ascii_hexdigit()) {
return Outcome::Failed("not an APNs device token".into());
}
let url = format!("{}/3/device/{token}", self.config.url);
let sent = self
.http
.post(url)
.bearer_auth(self.jwt())
.header("apns-topic", &self.config.topic)
.header("apns-push-type", "alert")
.json(&Self::payload(message))
.send()
.await;
let res = match sent {
Ok(res) => res,
Err(e) => return Outcome::Retry(format!("APNs: {}", e.without_url())),
};
let status = res.status().as_u16();
let body = res.text().await.unwrap_or_default();
let reason = serde_json::from_str::<Value>(&body)
.ok()
.and_then(|v| v["reason"].as_str().map(str::to_string))
.unwrap_or_default();
let why = format!("APNs {status} {reason}");
match status {
200 => Outcome::Sent,
410 => Outcome::Gone,
403 if reason.ends_with("ProviderToken") => {
*self.jwt.lock().unwrap() = None;
Outcome::Retry(why)
}
429 | 500.. => Outcome::Retry(why),
_ => Outcome::Failed(why),
}
}
}
crates/grain-serve/src/push/fcm.rsnew+223−0
View file
@@ -0,0 +1,223 @@
//! Firebase Cloud Messaging, FCM, for Android devices: its HTTP v1 API,
//! with an OAuth access token a Google service account's key earns. The
//! account signs a JWT asserting the messaging scope (RS256), and Google's
//! token endpoint trades it for a token good for an hour, used, less a
//! minute, for every push in it.
//!
//! A push is data only, every field a string: `title` and `body` beside
//! the notification's own fields ([`Message::data`]). The app draws the
//! notification itself from them, as it must to know where a tap goes; a
//! `notification` block would have Android draw it, and skip the app, while
//! the app is in the background. It goes at high priority, so Doze does not
//! hold it. FCM has no badge: a launcher counts what the app has posted.
//!
//! FCM's `UNREGISTERED` names a token no device answers to any more,
//! which is [`Outcome::Gone`]. A 401 or 403 drops the access token for a
//! new one on the next try, and 429s, 5xx and failures to reach it are
//! tried again; any other answer is not.
use super::{Message, Outcome, der_of_pem};
use anyhow::{Context, bail};
use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use ring::signature::{RSA_PKCS1_SHA256, RsaKeyPair};
use serde::Deserialize;
use serde_json::{Value, json};
use std::path::PathBuf;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;
/// FCM's API.
pub const URL: &str = "https://fcm.googleapis.com";
/// Where a service account's JWT is traded, unless its key file says.
pub const TOKEN_URL: &str = "https://oauth2.googleapis.com/token";
/// The one scope sending needs.
pub const SCOPE: &str = "https://www.googleapis.com/auth/firebase.messaging";
/// How long a push, or a trade for a token, may take.
pub const TIMEOUT: Duration = Duration::from_secs(15);
/// How grain-serve reaches FCM.
#[derive(Debug, Clone, PartialEq)]
pub struct Config {
/// `GRAIN_FCM_KEY_FILE`: the service account's key, the JSON Google
/// gives for it, of an account with the Firebase Messaging role.
pub key_file: PathBuf,
/// `GRAIN_FCM_PROJECT_ID`: the Firebase project, if not the one the key
/// file names.
pub project_id: Option<String>,
/// [`URL`].
pub url: String,
}
impl Config {
/// FCM as the environment names it: none without a key file.
pub fn from_vars(get: impl Fn(&str) -> Option<String>) -> Option<Config> {
let set = |name: &str| get(name).filter(|v| !v.is_empty());
Some(Config {
key_file: set("GRAIN_FCM_KEY_FILE")?.into(),
project_id: set("GRAIN_FCM_PROJECT_ID"),
url: URL.into(),
})
}
}
/// A service account's key file, as far as it is read.
#[derive(Deserialize)]
struct Account {
client_email: String,
private_key: String,
project_id: Option<String>,
token_uri: Option<String>,
}
/// FCM, ready to push.
pub struct Fcm {
http: reqwest::Client,
/// Where pushes are sent.
send_url: String,
client_email: String,
token_url: String,
key: RsaKeyPair,
/// The access token in use, and until when.
access: Mutex<Option<(String, Instant)>>,
}
/// What Google's token endpoint grants.
#[derive(Deserialize)]
struct Granted {
access_token: String,
expires_in: Option<u64>,
}
impl Fcm {
/// FCM as `config` names it, its key file read.
pub fn new(config: Config) -> anyhow::Result<Fcm> {
let path = config.key_file.display();
let json = std::fs::read_to_string(&config.key_file)
.with_context(|| format!("reading the FCM key file {path}"))?;
let account: Account = serde_json::from_str(&json)
.with_context(|| format!("{path} is not a service account's key file"))?;
let key = der_of_pem(&account.private_key)
.and_then(|der| RsaKeyPair::from_pkcs8(&der).ok())
.with_context(|| format!("{path}'s private_key is not an RSA key in PKCS #8 PEM"))?;
let Some(project) = config.project_id.or(account.project_id) else {
bail!("{path} names no project_id, and GRAIN_FCM_PROJECT_ID is not set");
};
let http = reqwest::Client::builder().timeout(TIMEOUT).build()?;
Ok(Fcm {
http,
send_url: format!("{}/v1/projects/{project}/messages:send", config.url),
client_email: account.client_email,
token_url: account.token_uri.unwrap_or_else(|| TOKEN_URL.into()),
key,
access: Mutex::new(None),
})
}
/// The JWT the service account trades for an access token.
fn assertion(&self) -> String {
let now = grain::keys::now_secs();
let header = json!({ "alg": "RS256", "typ": "JWT" });
let claims = json!({
"iss": self.client_email,
"scope": SCOPE,
"aud": self.token_url,
"iat": now,
"exp": now + 3600,
});
let body = format!(
"{}.{}",
URL_SAFE_NO_PAD.encode(header.to_string()),
URL_SAFE_NO_PAD.encode(claims.to_string())
);
let mut sig = vec![0; self.key.public().modulus_len()];
let rng = ring::rand::SystemRandom::new();
self.key
.sign(&RSA_PKCS1_SHA256, &rng, body.as_bytes(), &mut sig)
.expect("an RSA key signs");
format!("{body}.{}", URL_SAFE_NO_PAD.encode(sig))
}
/// The access token to push with: the one in use, unless it is a
/// minute from expiring; else one traded for now.
async fn access_token(&self) -> Result<String, String> {
let mut access = self.access.lock().await;
if let Some((token, until)) = &*access
&& Instant::now() < *until
{
return Ok(token.clone());
}
let form = [
("grant_type", "urn:ietf:params:oauth:grant-type:jwt-bearer"),
("assertion", &self.assertion()),
];
let res = self.http.post(&self.token_url).form(&form).send().await;
let res = res.map_err(|e| format!("FCM's token endpoint: {}", e.without_url()))?;
let status = res.status();
if !status.is_success() {
return Err(format!("FCM's token endpoint answered {status}"));
}
let granted: Granted = res
.json()
.await
.map_err(|e| format!("FCM's token endpoint granted no access_token: {e}"))?;
let lasts = granted.expires_in.unwrap_or(3600).saturating_sub(60);
let until = Instant::now() + Duration::from_secs(lasts);
*access = Some((granted.access_token.clone(), until));
Ok(granted.access_token)
}
/// What `message` is to FCM, for the device `token`.
pub fn payload(token: &str, message: &Message) -> Value {
// The data first, so that no field of it stands in for the text.
let mut data: serde_json::Map<String, Value> = message
.data
.iter()
.map(|(k, v)| (k.clone(), v.as_str().into()))
.collect();
data.insert("title".into(), message.title.as_str().into());
data.insert("body".into(), message.body.as_str().into());
json!({
"message": {
"token": token,
"data": data,
"android": { "priority": "HIGH" },
}
})
}
/// Pushes `message` to the device `token`.
pub async fn send(&self, token: &str, message: &Message) -> Outcome {
let access = match self.access_token().await {
Ok(access) => access,
Err(why) => return Outcome::Retry(why),
};
let sent = self
.http
.post(&self.send_url)
.bearer_auth(access)
.json(&Self::payload(token, message))
.send()
.await;
let res = match sent {
Ok(res) => res,
Err(e) => return Outcome::Retry(format!("FCM: {}", e.without_url())),
};
let status = res.status().as_u16();
let body = res.text().await.unwrap_or_default();
let why = format!("FCM {status}");
match status {
200..=299 => Outcome::Sent,
_ if body.contains("UNREGISTERED") => Outcome::Gone,
401 | 403 => {
*self.access.lock().await = None;
Outcome::Retry(why)
}
429 | 500.. => Outcome::Retry(why),
_ => Outcome::Failed(why),
}
}
}
crates/grain-serve/src/push/mod.rs+94−0
View file
@@ -7,6 +7,14 @@
//! `serviceDid` names too: a device registered for another service is
//! refused rather than kept for pushes this one would send. `platform` is
//! `ios`, `android` or `web`, and `appId` the app's bundle or package id.
//!
//! A push goes by the platform's own service: [`apns`] for `ios`, [`fcm`]
//! for `android`, each set up on its own, so a server with only one sends
//! to its devices and leaves the others' tokens be. `web` devices are
//! kept, but nothing is pushed to them.
pub mod apns;
pub mod fcm;
use crate::App;
use crate::index::Index;
@@ -17,6 +25,7 @@ use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::BTreeMap;
use std::sync::Arc;
/// The methods, as a service token for each names it in `lxm`.
@@ -103,3 +112,88 @@ pub async fn unregister(
index.unregister_push(&did, &device.token)?;
Ok(Json(json!({})))
}
/// A notification as it is pushed.
#[derive(Debug, Clone, PartialEq, Default)]
pub struct Message {
pub title: String,
pub body: String,
/// What the app reads to know where a tap goes: `type`, and the `uri`
/// or `did` it is about.
pub data: BTreeMap<String, String>,
/// The number the app's icon shows, where the platform has one.
pub badge: Option<i64>,
}
/// What became of a push.
#[derive(Debug, Clone, PartialEq)]
pub enum Outcome {
Sent,
/// The service says no device answers to the token any more: it is
/// forgotten.
Gone,
/// It may go if tried again, and why it did not.
Retry(String),
/// It will not go, and why.
Failed(String),
}
/// How grain-serve pushes: by APNs, by FCM, both or neither.
#[derive(Debug, Clone, PartialEq, Default)]
pub struct Config {
pub apns: Option<apns::Config>,
pub fcm: Option<fcm::Config>,
}
impl Config {
/// Push as the environment names it ([`apns::Config::from_vars`],
/// [`fcm::Config::from_vars`]).
pub fn from_vars(get: impl Fn(&str) -> Option<String>) -> Config {
Config {
apns: apns::Config::from_vars(&get),
fcm: fcm::Config::from_vars(&get),
}
}
}
/// The services a server pushes by, each ready.
#[derive(Default)]
pub struct Transports {
pub apns: Option<apns::Apns>,
pub fcm: Option<fcm::Fcm>,
}
impl Transports {
/// The services `config` names, their keys read: a failure where one
/// is named and its key cannot be.
pub fn new(config: &Config) -> anyhow::Result<Transports> {
Ok(Transports {
apns: config.apns.clone().map(apns::Apns::new).transpose()?,
fcm: config.fcm.clone().map(fcm::Fcm::new).transpose()?,
})
}
/// Pushes `message` to the device `token` on `platform`; none for a
/// platform no service here pushes to.
pub async fn send(&self, platform: &str, token: &str, message: &Message) -> Option<Outcome> {
match platform {
"ios" => Some(self.apns.as_ref()?.send(token, message).await),
"android" => Some(self.fcm.as_ref()?.send(token, message).await),
_ => None,
}
}
}
/// The DER a PEM block holds.
fn der_of_pem(pem: &str) -> Option<Vec<u8>> {
use base64::Engine;
let body: String = pem
.lines()
.map(str::trim)
.filter(|l| !l.is_empty() && !l.starts_with("-----"))
.collect();
base64::engine::general_purpose::STANDARD.decode(body).ok()
}
#[cfg(test)]
mod tests;
crates/grain-serve/src/push/test-rsa-key.pemnew+28−0
View file
@@ -0,0 +1,28 @@
-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDa1jowfCVPCbgm
CBuF8mxV6NuBkhIaHxvzEbYvT1hDqE4b5mNB2B/DpmY82WqcdEHHf8ulL56yoUwh
uRjtJB4wCHz8hhRXvPr4i3jJiwQh8Nx936OD02jdtspbRty4A08FiLF5G8tTrx5M
P5ht9o0NOwGy+7nqulmdU9yFGKupl+PapnUPvo/RIudUlpL2+/Z1+tVHav/b8IL5
5dUoI07lThIStoZWBavfhGVld+ZKJzsVL8b5iqSmdhECG7Cy1Q003ixGZLXQnJBU
OVIHLFiH99rJFEOezyJjdZSRBj8epTMcCdFgnH8ND7IN8gXqZmOXdr7ZT4oRYtxT
D9IVcVYpAgMBAAECggEAAXerjvXH2ClrJ6fRa6Q1rtJt3tEWyLk3axR9HWy7u7yf
0rpjaKM0YaQEHxLdUl1eDhTzTc3ZUCpkhauvjv/2DmNZcl83l0i2gX7oCjzlJaxG
3WXaaolBwgd9Imv7HyLgpiWWg4AGNtC1flsV+doqWQegyvkZKfV58ng9z3tOjelP
WN47yAhDY8NYO3fbaK5Z3LEWfDCBXD6ppxa4gNCXq+e3sFTUh25skX0xRN+ZZzHs
qXdf0FDSYuFC9yyty0eF+QuEOxCT/7fBzlfFzICyiStxurMkPRF6SvOL0qcDOlql
6Ia48N5MyVZhENnadF5pcSe9ZocKg3PgmDoJFEw+KQKBgQDuBZZc90UYTltXceHU
U7cvJWQGMa9BTluVpVQS715CGs2Sia78nXNhjqCSkAfeEV0wqt9oO9PLie7OjbIh
2XIfFfHsjhCntipgwync39Rqky6InWF7GbR3Iy+gBNzzaWhbmEexi69Yvyzrf2DK
qcV7jXS5yfnn1fvHLkp2AHNLYwKBgQDrXa1NKgvsPmmBAKNfGTL8mvcwU3RzAJmG
N3isHInm6eJpzOl7V88tuzsgi7gKpB5QDw12h2J+50PHIwGNKBYbGnADOqWpfjqa
YnWXVLxgJS1rScnYCJx7azpYMfejFhGVOJVtg5I7q4NQmf0gFQSI4aWonxydh5OS
SwipRTb8AwKBgQDG7piLPobbF7Iz1ZR/FzDFEKUWoDPviwP0gGdSJAlNioVB8hz0
OMwUP1Zo6EQLD8wpPsbVoRsUxEV5EOhObiR5Yqf9o4a8BP5KPRMh7Vjlcfm0aITu
k79Gy3ukbCwWrIRx+UYCV9LHg5CRirZk65uH9eQW234oi6mRAVluYlX6LQKBgB+F
bzU7YVdlUGnndPnCfTKZamt9ccYJGYaBHVcnh0x2nE0BFW/gZa3BOnz0WVSLgEfG
jE4Iead3P3uG86k86iD9HG1CSHd3oU+VyhvXTFcwA3Iv8qlNFe/tf2XaY0jjJr2u
U0zUZxw4ryB3TxAh4HaYicAYdxOd2aSpCeLpkyCTAoGAdW6G4kxcC6q1JKJkVuhG
wIhNgSvectXYviELDKRgFrAIM7zAcNXQKsCO7FO6dBFN1u2PqDlxp55uOt5e71nU
hOfC8u1tapSrdMimWXkjW0vsa4DFYLm80l2qEXviIOrCAvOgKC+caYCYu4aMzLlQ
eYRRFzOcP+OrsS8R59hj7bY=
-----END PRIVATE KEY-----
crates/grain-serve/src/push/tests.rsnew+558−0
View file
558 lines changed: not loaded until asked for.
crates/grain-serve/Cargo.toml+3−0
View file
@@ -14,7 +14,9 @@ axum.workspace = true
base64.workspace = true
bytes.workspace = true
futures.workspace = true
p256.workspace = true
reqwest.workspace = true
ring.workspace = true
rusqlite.workspace = true
serde.workspace = true
serde_json.workspace = true
@@ -25,5 +27,6 @@ tracing.workspace = true
tracing-subscriber.workspace = true
[dev-dependencies]
axum = { workspace = true, features = ["http2"] }
tokio-tungstenite.workspace = true
zstd.workspace = true
Cargo.lock+2−0
View file
@@ -705,7 +705,9 @@ dependencies = [
"bytes",
"futures",
"grain",
"p256",
"reqwest",
"ring",
"rusqlite",
"serde",
"serde_json",
Cargo.toml+1−0
View file
@@ -23,6 +23,7 @@ hkdf = "0.12"
k256 = { version = "0.13", features = ["ecdsa"] }
p256 = { version = "0.13", features = ["ecdsa"] }
rand_core = { version = "0.6", features = ["getrandom"] }
ring = "0.17"
reqwest = { version = "0.12", default-features = false, features = [
"json",
"rustls-tls",