Phone alerts: a phone registers its Cloud Messaging token and key (PUT /api/v1/user/push), and each notification its account is told goes to it sealed to that key, sent keyless as the VM's service account; pins OxygenFlow main (refs oxygenhub-android#16) aeed63c00ad7

-- Phones that take alerts (oxygenhub-android#16): one per device sign-in,

-- gone with it when the token is revoked. The Cloud Messaging registration

-- token addresses the phone; alerts are sealed to its X25519 public key

-- (hex), so the push service sees only ciphertext.

CREATE TABLE push_devices (

    token_id   BIGINT PRIMARY KEY REFERENCES tokens(id) ON DELETE CASCADE,

    account_id BIGINT NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,

    fcm_token  TEXT NOT NULL,

    public_key TEXT NOT NULL,

    created_ms BIGINT NOT NULL,

    updated_ms BIGINT NOT NULL

);

CREATE INDEX push_devices_by_account ON push_devices (account_id);

-- Phones that take alerts (oxygenhub-android#16): one per device sign-in,

-- gone with it when the token is revoked. The Cloud Messaging registration

-- token addresses the phone; alerts are sealed to its X25519 public key

-- (hex), so the push service sees only ciphertext.

CREATE TABLE push_devices (

    token_id   INTEGER PRIMARY KEY REFERENCES tokens(id) ON DELETE CASCADE,

    account_id INTEGER NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,

    fcm_token  TEXT NOT NULL,

    public_key TEXT NOT NULL,

    created_ms INTEGER NOT NULL,

    updated_ms INTEGER NOT NULL

);

CREATE INDEX push_devices_by_account ON push_devices (account_id);

//! Phones that take alerts (oxygenhub-android#16).

use super::{now_ms, Pool};

use sqlx::FromRow;

#[derive(Clone, Debug, FromRow)]

pub struct Device {

    pub token_id: i64,

    pub account_id: i64,

    pub fcm_token: String,

    pub public_key: String,

}

/// Register (or refresh) the phone signed in with token `token_id`.

pub async fn register(pool: &Pool, token_id: i64, account_id: i64, fcm_token: &str, public_key: &str) -> sqlx::Result<()> {

    let now = now_ms();

    sqlx::query(

        "INSERT INTO push_devices (token_id, account_id, fcm_token, public_key, created_ms, updated_ms) VALUES ($1, $2, $3, $4, $5, $5) \

         ON CONFLICT (token_id) DO UPDATE SET fcm_token = excluded.fcm_token, public_key = excluded.public_key, updated_ms = excluded.updated_ms",

    )

    .bind(token_id)

    .bind(account_id)

    .bind(fcm_token)

    .bind(public_key)

    .bind(now)

    .execute(pool)

    .await?;

    Ok(())

}

pub async fn forget(pool: &Pool, token_id: i64) -> sqlx::Result<bool> {

    Ok(sqlx::query("DELETE FROM push_devices WHERE token_id = $1").bind(token_id).execute(pool).await?.rows_affected() > 0)

}

/// Forget a registration Cloud Messaging says is gone.

pub async fn forget_fcm(pool: &Pool, fcm_token: &str) -> sqlx::Result<()> {

    sqlx::query("DELETE FROM push_devices WHERE fcm_token = $1").bind(fcm_token).execute(pool).await?;

    Ok(())

}

pub async fn of_account(pool: &Pool, account_id: i64) -> sqlx::Result<Vec<Device>> {

    sqlx::query_as("SELECT token_id, account_id, fcm_token, public_key FROM push_devices WHERE account_id = $1")

        .bind(account_id)

        .fetch_all(pool)

        .await

}

//! Phone alerts (oxygenhub-android#16, oxygenhub-gcp decision 0008): each

//! notification an account is told also goes to its phones through Firebase

//! Cloud Messaging, sealed to each phone's key (`flow_core::alert`), so the

//! push service carries only ciphertext and a wake-up.

//!

//! The hub sends as the service account its VM runs as: the metadata

//! server's token, kept until a minute before it expires. Off while

//! `[push] fcm_project` is empty.

use crate::db::push as pdb;

use crate::Shared;

use std::sync::Mutex;

use std::time::{Duration, Instant};

/// Most phones one account's alert goes to.

const MOST_DEVICES: usize = 20;

static TOKEN: Mutex<Option<(String, Instant)>> = Mutex::new(None);

/// What a phone is told, before it is sealed: small, as a message's data

/// is at most 4 KB.

#[derive(serde::Serialize, serde::Deserialize, Debug, PartialEq)]

pub struct Alert {

    pub kind: String,

    pub title: String,

    pub url: String,

}

/// Whether alerts are on.

pub fn enabled(app: &Shared) -> bool {

    !app.config.push.fcm_project.is_empty()

}

/// Send `alert` to every phone of `account_id`, in the background.

pub fn send(app: &Shared, account_id: i64, alert: Alert) {

    if !enabled(app) {

        return;

    }

    let app = app.clone();

    tokio::spawn(async move {

        if let Err(e) = send_now(&app, account_id, &alert).await {

            tracing::warn!("push: alert for account {account_id}: {e}");

        }

    });

}

async fn access_token(app: &Shared, client: &reqwest::Client) -> Result<String, String> {

    if let Some((t, until)) = TOKEN.lock().unwrap().clone() {

        if Instant::now() < until {

            return Ok(t);

        }

    }

    #[derive(serde::Deserialize)]

    struct Got {

        access_token: String,

        expires_in: u64,

    }

    let got: Got = client

        .get(&app.config.push.token_url)

        .header("Metadata-Flavor", "Google")

        .send()

        .await

        .and_then(|r| r.error_for_status())

        .map_err(|e| format!("the metadata server's token: {e}"))?

        .json()

        .await

        .map_err(|e| format!("the metadata server's token: {e}"))?;

    let until = Instant::now() + Duration::from_secs(got.expires_in.saturating_sub(60));

    *TOKEN.lock().unwrap() = Some((got.access_token.clone(), until));

    Ok(got.access_token)

}

/// Send `alert` to each of the account's phones now; a phone Cloud

/// Messaging no longer knows is forgotten.

pub async fn send_now(app: &Shared, account_id: i64, alert: &Alert) -> Result<usize, String> {

    let devices = pdb::of_account(&app.db, account_id).await.map_err(|e| e.to_string())?;

    if devices.is_empty() {

        return Ok(0);

    }

    let client = reqwest::Client::builder().timeout(Duration::from_secs(20)).build().map_err(|e| e.to_string())?;

    let token = access_token(app, &client).await?;

    let plain = serde_json::to_vec(alert).map_err(|e| e.to_string())?;

    let url = format!("{}/v1/projects/{}/messages:send", app.config.push.fcm_url.trim_end_matches('/'), app.config.push.fcm_project);

    let mut sent = 0;

    for d in devices.iter().take(MOST_DEVICES) {

        let Some(key) = hex::decode(&d.public_key).ok().and_then(|k| <[u8; 32]>::try_from(k).ok()) else { continue };

        let Some(sealed) = flow_core::alert::seal(&key, &plain) else { continue };

        // Data only, high priority: the app wakes, opens it and shows it;

        // the push service has no title or body to show or read.

        let body = serde_json::json!({ "message": {

            "token": d.fcm_token,

            "data": { "sealed": crate::oidc::b64url(&sealed) },

            "android": { "priority": "high" },

        }});

        let r = client.post(&url).bearer_auth(&token).json(&body).send().await.map_err(|e| e.to_string())?;

        let status = r.status();

        if status.is_success() {

            sent += 1;

        } else {

            let text = r.text().await.unwrap_or_default();

            if status == reqwest::StatusCode::NOT_FOUND || text.contains("UNREGISTERED") {

                pdb::forget_fcm(&app.db, &d.fcm_token).await.ok();

            } else {

                tracing::warn!("push: Cloud Messaging said {status}: {}", text.chars().take(200).collect::<String>());

            }

        }

    }

    Ok(sent)

}

//! Phone alerts (oxygenhub-android#16): a phone registers its Cloud

//! Messaging token and its key with the token it signed in with; each

//! notification its account is told goes to it sealed to that key, so the

//! push service sees no title; a registration Cloud Messaging no longer

//! knows is forgotten, and one goes with its token.

use axum::body::Body;

use axum::http::{header, Request, StatusCode};

use axum::routing::{get, post};

use axum::{Json, Router};

use http_body_util::BodyExt;

use oxygenhub::config::Config;

use oxygenhub::db;

use oxygenhub::Shared;

use std::sync::{Arc, Mutex};

use tower::ServiceExt;

type Sent = Arc<Mutex<Vec<serde_json::Value>>>;

/// Google's metadata server and Cloud Messaging, as far as the hub uses them.

async fn fake_google() -> (String, Sent) {

    let sent: Sent = Arc::default();

    let kept = sent.clone();

    let app = Router::new()

        .route(

            "/token",

            get(|h: axum::http::HeaderMap| async move {

                assert_eq!(h.get("metadata-flavor").and_then(|v| v.to_str().ok()), Some("Google"));

                Json(serde_json::json!({ "access_token": "vm-token", "expires_in": 3600 }))

            }),

        )

        .route(

            "/v1/projects/demo/messages:send",

            post(move |h: axum::http::HeaderMap, Json(body): Json<serde_json::Value>| {

                let kept = kept.clone();

                async move {

                    assert_eq!(h.get("authorization").and_then(|v| v.to_str().ok()), Some("Bearer vm-token"));

                    let gone = body["message"]["token"] == "gone";

                    kept.lock().unwrap().push(body);

                    if gone {

                        (

                            StatusCode::NOT_FOUND,

                            Json(serde_json::json!({ "error": { "status": "NOT_FOUND", "details": [{ "errorCode": "UNREGISTERED" }] } })),

                        )

                    } else {

                        (StatusCode::OK, Json(serde_json::json!({ "name": "projects/demo/messages/1" })))

                    }

                }

            }),

        );

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

    let base = format!("http://{}", listener.local_addr().unwrap());

    tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });

    (base, sent)

}

async fn call(app: &Shared, method: &str, path: &str, token: &str, body: Option<serde_json::Value>) -> (StatusCode, String) {

    let mut req = Request::builder().method(method).uri(path).header(header::AUTHORIZATION, format!("Bearer {token}"));

    let body = match body {

        Some(b) => {

            req = req.header(header::CONTENT_TYPE, "application/json");

            Body::from(b.to_string())

        }

        None => Body::empty(),

    };

    let res = oxygenhub::router(app.clone()).oneshot(req.body(body).unwrap()).await.unwrap();

    let status = res.status();

    (status, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())

}

fn b64url_decode(s: &str) -> Vec<u8> {

    let val = |c: u8| match c {

        b'A'..=b'Z' => c - b'A',

        b'a'..=b'z' => c - b'a' + 26,

        b'0'..=b'9' => c - b'0' + 52,

        b'-' => 62,

        _ => 63,

    } as u32;

    let mut out = Vec::new();

    for chunk in s.as_bytes().chunks(4) {

        let n = chunk.iter().enumerate().fold(0u32, |acc, (i, c)| acc | (val(*c) << (18 - 6 * i)));

        out.extend_from_slice(&n.to_be_bytes()[1..chunk.len()]);

    }

    out

}

async fn sent_after(sent: &Sent, n: usize) -> Vec<serde_json::Value> {

    for _ in 0..100 {

        if sent.lock().unwrap().len() >= n {

            break;

        }

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    }

    sent.lock().unwrap().clone()

}

#[tokio::test(flavor = "multi_thread")]

async fn a_notification_reaches_the_phone_sealed() {

    let (google, sent) = fake_google().await;

    let dir = tempfile::tempdir().unwrap();

    let mut config = Config::load(dir.path()).unwrap();

    config.push.fcm_project = "demo".into();

    config.push.fcm_url = google.clone();

    config.push.token_url = format!("{google}/token");

    let app = oxygenhub::build(config).await.unwrap();

    let bob = db::accounts::create_account(&app.db, "bob", "[email protected]", "x", "salt", false).await.unwrap();

    let (phone_token, hash) = oxygenhub::auth::new_token();

    db::accounts::create_token(&app.db, bob.id, "pixel", &hash, "repo:read", "", None).await.unwrap();

    // The phone registers its Cloud Messaging token and its key.

    let secret = flow_core::alert::new_secret();

    let key = hex::encode(flow_core::alert::public_key(&secret));

    let bad =

        call(&app, "PUT", "/api/v1/user/push", &phone_token, Some(serde_json::json!({ "fcm_token": "f1", "public_key": "abc" }))).await;

    assert_eq!(bad.0, StatusCode::BAD_REQUEST, "{}", bad.1);

    let (status, body) =

        call(&app, "PUT", "/api/v1/user/push", &phone_token, Some(serde_json::json!({ "fcm_token": "f1", "public_key": key }))).await;

    assert_eq!(status, StatusCode::OK, "{body}");

    assert!(body.contains("\"alerts\":true"), "{body}");

    // A notification for bob: his phone is sent it, sealed.

    oxygenhub::notify::tell(

        &app,

        "issue.assigned",

        "andy/demo#1: Fix it",

        "/andy/demo/issue/1",

        "andy",

        None,

        vec![(bob.id, db::notify::Reason::Participating)],

    )

    .await;

    let got = sent_after(&sent, 1).await;

    assert_eq!(got.len(), 1, "{got:?}");

    let m = &got[0]["message"];

    assert_eq!(m["token"], "f1");

    assert_eq!(m["android"]["priority"], "high");

    assert!(m.get("notification").is_none() && !got[0].to_string().contains("Fix it"), "nothing readable on the way: {got:?}");

    let opened = flow_core::alert::open(&secret, &b64url_decode(m["data"]["sealed"].as_str().unwrap())).expect("sealed to the phone");

    let alert: oxygenhub::push::Alert = serde_json::from_slice(&opened).unwrap();

    assert_eq!(alert.kind, "issue.assigned");

    assert_eq!(alert.title, "📌 Issue Assigned: andy/demo#1: Fix it");

    assert!(alert.url.ends_with("/andy/demo/issue/1") && alert.url.starts_with("http"), "{}", alert.url);

    // A registration Cloud Messaging no longer knows is forgotten.

    let (gone_token, hash) = oxygenhub::auth::new_token();

    db::accounts::create_token(&app.db, bob.id, "old phone", &hash, "repo:read", "", None).await.unwrap();

    call(&app, "PUT", "/api/v1/user/push", &gone_token, Some(serde_json::json!({ "fcm_token": "gone", "public_key": key }))).await;

    assert_eq!(oxygenhub::push::send_now(&app, bob.id, &alert).await, Ok(1));

    let left: Vec<String> = db::push::of_account(&app.db, bob.id).await.unwrap().into_iter().map(|d| d.fcm_token).collect();

    assert_eq!(left, ["f1"]);

    // Unregistering, and signing out (the token revoked), end it.

    assert_eq!(call(&app, "DELETE", "/api/v1/user/push", &phone_token, None).await.0, StatusCode::NO_CONTENT);

    assert!(db::push::of_account(&app.db, bob.id).await.unwrap().is_empty());

    call(&app, "PUT", "/api/v1/user/push", &phone_token, Some(serde_json::json!({ "fcm_token": "f1", "public_key": key }))).await;

    assert_eq!(call(&app, "DELETE", "/api/v1/user/token", &phone_token, None).await.0, StatusCode::NO_CONTENT);

    assert!(db::push::of_account(&app.db, bob.id).await.unwrap().is_empty(), "gone with its token");

}

- **Phone alerts (oxygenhub-android#16, oxygenhub-gcp decision 0008).** A phone registers its Cloud Messaging token and an X25519 public key with `PUT /api/v1/user/push`, using the token it signed in with (`DELETE` unregisters; revoking the token ends it too, migration 0101). Each notification an account is told then also goes to its phones, its title and link sealed to each phone's key (`flow_core::alert`), as a data-only, high-priority message, so Google carries only ciphertext. The hub sends as the service account its VM runs as, with the metadata server's token: no key. Off until `[push] fcm_project` (`OXYHUB_PUSH_FCM_PROJECT`) is set; a registration Cloud Messaging no longer knows is forgotten. Pins OxygenFlow main (sealed alerts).

 "curve25519-dalek",

 "ring",

        .route("/user/push", axum::routing::put(register_push).delete(forget_push))

#[derive(Deserialize)]

struct PushRegistration {

    /// The phone's Cloud Messaging registration token.

    fcm_token: String,

    /// The phone's X25519 public key, hex: alerts are sealed to it.

    public_key: String,

}

/// `PUT /api/v1/user/push` (oxygenhub-android#16): the phone signed in

/// with this token takes alerts, sealed to its key; registering again

/// replaces its registration. It ends with the token.

async fn register_push(

    State(app): State<Shared>,

    headers: HeaderMap,

    Json(r): Json<PushRegistration>,

) -> Result<Json<serde_json::Value>, ApiError> {

    let c = caller(&app, &headers).await?;

    let (Some(account), Some(id)) = (c.account.as_ref(), c.token_id) else {

        return Err(ApiError(StatusCode::UNAUTHORIZED, "a phone registers with the token it signed in with".into()));

    };

    let key = r.public_key.trim().to_ascii_lowercase();

    if key.len() != 64 || hex::decode(&key).is_err() {

        return Err(ApiError(StatusCode::BAD_REQUEST, "public_key is 32 bytes of hex".into()));

    }

    let fcm = r.fcm_token.trim();

    if fcm.is_empty() || fcm.len() > 4096 {

        return Err(ApiError(StatusCode::BAD_REQUEST, "fcm_token is the phone's registration token".into()));

    }

    db::push::register(&app.db, id, account.id, fcm, &key).await?;

    Ok(Json(serde_json::json!({ "alerts": crate::push::enabled(&app) })))

}

/// `DELETE /api/v1/user/push`: this phone takes no more alerts.

async fn forget_push(State(app): State<Shared>, headers: HeaderMap) -> Result<StatusCode, ApiError> {

    let c = caller(&app, &headers).await?;

    let Some(id) = c.token_id.filter(|_| c.account.is_some()) else {

        return Err(ApiError(StatusCode::UNAUTHORIZED, "a phone unregisters with the token it signed in with".into()));

    };

    db::push::forget(&app.db, id).await?;

    Ok(StatusCode::NO_CONTENT)

}

    pub push: Push,

/// Phone alerts (oxygenhub-android#16, oxygenhub-gcp decision 0008): a

/// notification is also sent, sealed to each phone's key, through Firebase

/// Cloud Messaging. Off while `fcm_project` is empty. The hub sends as the

/// service account its VM runs as, whose token the metadata server gives:

/// no key.

#[derive(Clone, Debug, Deserialize)]

#[serde(default, deny_unknown_fields)]

pub struct Push {

    /// The Firebase (GCP) project, `OXYHUB_PUSH_FCM_PROJECT`.

    pub fcm_project: String,

    /// Cloud Messaging and the token source; set only to point them

    /// elsewhere (tests).

    pub fcm_url: String,

    pub token_url: String,

}

impl Default for Push {

    fn default() -> Self {

        Push {

            fcm_project: String::new(),

            fcm_url: "https://fcm.googleapis.com".into(),

            token_url: "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token".into(),

        }

    }

}

            push: Push::default(),

        }

        if let Some(v) = get("OXYHUB_PUSH_FCM_PROJECT") {

            self.push.fcm_project = v;

pub mod push;

pub mod push;

        // And to the account's phones, sealed (oxygenhub-android#16).

        let link = format!("{}{url}", app.config.base_url().trim_end_matches('/'));

        crate::push::send(app, id, crate::push::Alert { kind: kind.to_string(), title: subject(kind, title), url: link });

pub(crate) fn b64url(bytes: &[u8]) -> String {