Šepot — end-to-end encrypted messenger

Initial public release. Anonymous on-device identity (no phone number,
e-mail, or password), Signal-protocol cryptography (X3DH + PQXDH, Double
Ratchet, sealed sender) via libsignal, a metadata-minimizing relay,
group chats, disappearing messages, encrypted local storage, optional
Tor transport, and a UI localized into Czech, English, German, Russian,
and Italian.

Monorepo: Flutter client (app/), Rust relay + wire types (rust/),
protobuf definitions (proto/), self-host deploy (deploy/).

Licensed under AGPL-3.0 with an app-store distribution exception (§7).
This commit is contained in:
Julius Joska
2026-07-11 17:06:39 +02:00
commit 8081d98c67
309 changed files with 44102 additions and 0 deletions
+4007
View File
File diff suppressed because it is too large Load Diff
+23
View File
@@ -0,0 +1,23 @@
[workspace]
resolver = "2"
members = ["sepot-proto", "sepot-relay"]
[workspace.package]
edition = "2021"
license = "AGPL-3.0"
repository = "https://gitea.ajtak.it/julius.joska/sepot"
[workspace.dependencies]
prost = "0.13"
prost-build = "0.13"
tokio = { version = "1", features = ["full"] }
anyhow = "1"
thiserror = "2"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
# Zrcadlo [patch.crates-io] z signalapp/libsignal v0.96.4 — jejich workspace
# patch se na git závislost nepropaguje, bez něj se libsignal-protocol
# nesestaví proti crates.io verzi curve25519-dalek.
[patch.crates-io]
curve25519-dalek = { git = "https://github.com/signalapp/curve25519-dalek", tag = "signal-curve25519-4.1.3" }
+11
View File
@@ -0,0 +1,11 @@
[package]
name = "sepot-proto"
version = "0.1.0"
edition.workspace = true
license.workspace = true
[dependencies]
prost = { workspace = true }
[build-dependencies]
prost-build = { workspace = true }
+4
View File
@@ -0,0 +1,4 @@
fn main() -> std::io::Result<()> {
println!("cargo:rerun-if-changed=../../proto/sepot.proto");
prost_build::compile_protos(&["../../proto/sepot.proto"], &["../../proto"])
}
+44
View File
@@ -0,0 +1,44 @@
//! Wire typy Šepotu generované z `proto/sepot.proto` (jediný zdroj pravdy).
pub mod v1 {
include!(concat!(env!("OUT_DIR"), "/sepot.v1.rs"));
}
pub use prost::Message;
/// Tvrdý limit velikosti payload obálky (message_queue CHECK).
pub const MAX_ENVELOPE_PAYLOAD: usize = 256 * 1024;
/// Doménový separátor podpisu AUTH výzvy.
pub const AUTH_CONTEXT: &[u8] = b"sepot-auth-v1";
#[cfg(test)]
mod tests {
use super::v1::*;
use prost::Message;
#[test]
fn envelope_roundtrip() {
let env = Envelope {
dst_address: vec![7u8; 32],
msg_uuid: vec![1u8; 16],
content_hint: ContentHint::Default as i32,
payload: vec![0xAB; 128],
};
let bytes = env.encode_to_vec();
let back = Envelope::decode(bytes.as_slice()).unwrap();
assert_eq!(env, back);
}
#[test]
fn client_frame_oneof() {
let frame = ClientFrame {
frame: Some(client_frame::Frame::Ack(Ack {
msg_uuid: vec![9u8; 16],
})),
};
let bytes = frame.encode_to_vec();
let back = ClientFrame::decode(bytes.as_slice()).unwrap();
assert_eq!(frame, back);
}
}
+51
View File
@@ -0,0 +1,51 @@
[package]
name = "sepot-relay"
version = "0.1.0"
edition.workspace = true
license.workspace = true
[dependencies]
sepot-proto = { path = "../sepot-proto" }
prost = { workspace = true }
tokio = { workspace = true }
anyhow = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
axum = { version = "0.8", features = ["ws"] }
sqlx = { version = "0.8", default-features = false, features = [
"postgres",
"runtime-tokio",
"tls-rustls",
"chrono",
"migrate",
"macros",
] }
dashmap = "6"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
base64 = "0.22"
hex = "0.4"
chrono = "0.4"
rand = "0.9"
futures-util = "0.3"
# Streaming download příloh z disku (ReaderStream)
tokio-util = { version = "0.7", features = ["io"] }
# Oficiální Signal krypto — POUZE pro roli certifikační autority sealed
# senderu (ServerCertificate/SenderCertificate) a ověření XEdDSA podpisů
# při AUTH. Relay NIKDY nešifruje/nedešifruje zprávy (hloupý relay).
libsignal-protocol = { git = "https://github.com/signalapp/libsignal", tag = "v0.96.4" }
libsignal-core = { git = "https://github.com/signalapp/libsignal", tag = "v0.96.4" }
tower_governor = "0.8.0"
tower-http = { version = "0.7.0", features = ["timeout"] }
# Push budíčky (UnifiedPush/WebPush = POST na endpoint). rustls, ať build
# zůstane bez OpenSSL (stejně jako sqlx tls-rustls). Bez default featur —
# nechceme native-tls ani zbytečné kodeky.
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] }
# FCM fast mode: RS256 JWT pro OAuth2 service account flow
jsonwebtoken = "9"
[dev-dependencies]
tokio = { workspace = true, features = ["full"] }
+97
View File
@@ -0,0 +1,97 @@
-- Šepot relay — init schéma.
-- Klíče Curve25519 mají 33 B (libsignal serializace s type bytem 0x05),
-- ne 32 B. ŽÁDNÝ sloupec pro IP, user-agent ani jinou PII.
CREATE TABLE identities (
account_id BYTEA PRIMARY KEY,
identity_key BYTEA NOT NULL UNIQUE,
registered_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CHECK (octet_length(identity_key) = 33),
CHECK (octet_length(account_id) = 16)
);
CREATE TABLE devices (
device_uuid BYTEA PRIMARY KEY,
account_id BYTEA NOT NULL REFERENCES identities(account_id) ON DELETE CASCADE,
device_id INTEGER NOT NULL,
registration_id INTEGER NOT NULL,
delivery_address BYTEA NOT NULL UNIQUE,
last_seen_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (account_id, device_id),
CHECK (octet_length(delivery_address) = 32),
CHECK (octet_length(device_uuid) = 16)
);
CREATE INDEX idx_devices_account ON devices(account_id);
CREATE TABLE signed_prekeys (
device_uuid BYTEA NOT NULL REFERENCES devices(device_uuid) ON DELETE CASCADE,
key_id INTEGER NOT NULL,
public_key BYTEA NOT NULL,
signature BYTEA NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (device_uuid, key_id),
CHECK (octet_length(public_key) = 33),
CHECK (octet_length(signature) = 64)
);
CREATE TABLE kyber_prekeys (
device_uuid BYTEA NOT NULL REFERENCES devices(device_uuid) ON DELETE CASCADE,
key_id INTEGER NOT NULL,
public_key BYTEA NOT NULL,
signature BYTEA NOT NULL,
is_last_resort BOOLEAN NOT NULL DEFAULT false,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (device_uuid, key_id),
CHECK (octet_length(signature) = 64),
CHECK (octet_length(public_key) <= 2048)
);
CREATE TABLE one_time_prekeys (
device_uuid BYTEA NOT NULL REFERENCES devices(device_uuid) ON DELETE CASCADE,
key_id INTEGER NOT NULL,
public_key BYTEA NOT NULL,
PRIMARY KEY (device_uuid, key_id),
CHECK (octet_length(public_key) = 33)
);
CREATE INDEX idx_otp_device ON one_time_prekeys(device_uuid);
CREATE TABLE message_queue (
msg_uuid BYTEA PRIMARY KEY,
dst_address BYTEA NOT NULL,
envelope BYTEA NOT NULL,
enqueued_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ NOT NULL,
CHECK (octet_length(envelope) <= 262144),
CHECK (octet_length(msg_uuid) = 16)
);
-- Pozn.: žádný sloupec delivered_at — ACK klienta = okamžitý hard-delete
-- řádku, fronta se po doručení nedrží.
CREATE INDEX idx_queue_dst ON message_queue(dst_address, enqueued_at);
CREATE INDEX idx_queue_expiry ON message_queue(expires_at);
CREATE TABLE push_tokens (
device_uuid BYTEA PRIMARY KEY REFERENCES devices(device_uuid) ON DELETE CASCADE,
platform SMALLINT NOT NULL,
token BYTEA,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- Klíče relaye pro sealed sender (trust root + server certificate).
-- Trust root keypair i server keypair se generují a ukládají sem při prvním
-- startu relaye. Sloupec trust_root_private tedy reálně obsahuje privátní
-- klíč trust rootu. Pro produkci se doporučuje po vygenerování privátní klíč
-- trust rootu z DB odstranit (nastavit na NULL) a jeho zálohu držet offline —
-- k běžnému provozu relaye stačí trust_root_public, server keypair a cert.
CREATE TABLE relay_keys (
id SMALLINT PRIMARY KEY DEFAULT 1 CHECK (id = 1),
trust_root_public BYTEA NOT NULL,
trust_root_private BYTEA,
server_public BYTEA NOT NULL,
server_private BYTEA NOT NULL,
server_certificate BYTEA NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
@@ -0,0 +1,16 @@
-- Přílohy: opaque šifrované bloby mimo message_queue.
-- Relay vidí jen blob_id + velikost + expiraci. ŽÁDNÁ vazba na účet,
-- zařízení ani zprávu — kdo blob nahrál/stáhl se neukládá (privacy).
-- Samotná data leží na disku (SEPOT_ATTACHMENTS_DIR), v DB jen metadata
-- kvůli TTL úklidu.
CREATE TABLE attachments (
blob_id BYTEA PRIMARY KEY,
size_bytes BIGINT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ NOT NULL,
CHECK (octet_length(blob_id) = 16),
CHECK (size_bytes > 0)
);
CREATE INDEX idx_attachments_expiry ON attachments(expires_at);
+134
View File
@@ -0,0 +1,134 @@
//! APNs sender — nativní iOS push, BEZ Googlu.
//!
//! Pro privacy messenger je na iOS správné buzení přímo přes Apple Push
//! Notification service, ne přes Firebase (žádný Google v cestě). Stejné
//! tvrdé pravidlo jako všude: posílá se JEN tichý „background" budíček
//! `{"aps":{"content-available":1}}` — žádný obsah, žádný odesílatel.
//! Klient se po probuzení sám připojí na WS a vyleje frontu.
//!
//! Credentials (env): `SEPOT_APNS_CREDENTIALS` = cesta k APNs auth key
//! (.p8, EC private key z Apple Developer → Keys), `SEPOT_APNS_KEY_ID`,
//! `SEPOT_APNS_TEAM_ID`, `SEPOT_APNS_TOPIC` (= bundle id cz.sepot.sepot).
//! `SEPOT_APNS_PRODUCTION` (default true; TestFlight i App Store =
//! production APNs, jen debug build z Xcode jede sandbox).
//! Bez env proměnných zůstává APNs vypnutý a registrace se tiše ignorují.
//!
//! Auth: ES256 JWT (iss=team, iat=now) podepsaný APNs klíčem, platný ~1 h,
//! sdílený přes všechny odeslání (cache, refresh po 50 min).
use std::sync::Arc;
use chrono::Utc;
use jsonwebtoken::{Algorithm, EncodingKey, Header};
use serde::Serialize;
use tokio::sync::Mutex;
const PROD_HOST: &str = "https://api.push.apple.com";
const SANDBOX_HOST: &str = "https://api.sandbox.push.apple.com";
#[derive(Serialize)]
struct Claims<'a> {
iss: &'a str,
iat: i64,
}
pub struct ApnsSender {
team_id: String,
key_id: String,
topic: String,
host: &'static str,
key: EncodingKey,
/// (jwt, vydáno_unix)
cached: Mutex<Option<(String, i64)>>,
}
impl ApnsSender {
/// Načte APNs credentials z env. None = iOS nativní push vypnutý.
pub fn from_env() -> Option<Arc<ApnsSender>> {
let path = std::env::var("SEPOT_APNS_CREDENTIALS").ok()?;
let key_id = std::env::var("SEPOT_APNS_KEY_ID").ok()?;
let team_id = std::env::var("SEPOT_APNS_TEAM_ID").ok()?;
let topic =
std::env::var("SEPOT_APNS_TOPIC").unwrap_or_else(|_| "cz.sepot.sepot".into());
let production = std::env::var("SEPOT_APNS_PRODUCTION")
.map(|v| v != "false" && v != "0")
.unwrap_or(true);
let pem = match std::fs::read(&path) {
Ok(p) => p,
Err(e) => {
tracing::warn!("APNs klíč nejde přečíst: {e}");
return None;
}
};
let key = match EncodingKey::from_ec_pem(&pem) {
Ok(k) => k,
Err(e) => {
tracing::warn!("APNs klíč (.p8) nejde načíst: {e}");
return None;
}
};
tracing::info!(production, "APNs nativní push aktivní");
Some(Arc::new(ApnsSender {
team_id,
key_id,
topic,
host: if production { PROD_HOST } else { SANDBOX_HOST },
key,
cached: Mutex::new(None),
}))
}
async fn auth_token(&self) -> Option<String> {
let now = Utc::now().timestamp();
let mut cached = self.cached.lock().await;
if let Some((jwt, issued)) = cached.as_ref() {
// APNs token platí max 1 h; refresh po 50 min (a min 20 s starý)
if now - issued < 3000 {
return Some(jwt.clone());
}
}
let mut header = Header::new(Algorithm::ES256);
header.kid = Some(self.key_id.clone());
let jwt = jsonwebtoken::encode(
&header,
&Claims {
iss: &self.team_id,
iat: now,
},
&self.key,
)
.map_err(|e| tracing::warn!("APNs JWT podpis selhal: {e}"))
.ok()?;
*cached = Some((jwt.clone(), now));
Some(jwt)
}
/// Tichý budíček na APNs device token. Best-effort; loguje jen stav
/// (nikdy token). `apns-push-type: background` + `content-available:1`
/// = probuzení bez viditelné notifikace (tu si klient případně ukáže
/// lokálně sám, bez obsahu).
pub async fn send_wake(&self, client: &reqwest::Client, device_token: &str) {
let Some(jwt) = self.auth_token().await else {
return;
};
let url = format!("{}/3/device/{}", self.host, device_token);
let resp = client
.post(&url)
.header("authorization", format!("bearer {jwt}"))
.header("apns-topic", &self.topic)
.header("apns-push-type", "background")
.header("apns-priority", "5")
.header("apns-expiration", "0")
.body(r#"{"aps":{"content-available":1}}"#)
.send()
.await;
match resp {
Ok(r) if !r.status().is_success() => {
tracing::debug!(status = %r.status(), "APNs budíček odmítnut");
}
Err(e) => tracing::debug!("APNs budíček selhal: {e}"),
_ => {}
}
}
}
+372
View File
@@ -0,0 +1,372 @@
//! Přílohy — opaque šifrované bloby mimo message_queue.
//!
//! Model Signal: klient soubor zašifruje vlastním klíčem (AES-256-GCM),
//! sem nahraje UŽ ŠIFROVANÝ blob a klíč+digest pošle uvnitř E2E zprávy.
//! Relay proto vidí jen náhodné bajty a nikdy neví, ke které zprávě ani
//! komu blob patří:
//!
//! - upload vyžaduje X-Sepot-Auth (anti-abuse), ale vazba zařízení→blob
//! se NIKAM neukládá ani neloguje
//! - download je bez auth — blob_id (128 bitů) je bearer capability,
//! zná ho jen odesílatel a příjemce (dorazil E2E kanálem)
//! - TTL 30 dní, pak hodinový úklid smaže řádek i soubor
//!
//! Data leží na disku (`SEPOT_ATTACHMENTS_DIR`), v DB jen metadata pro
//! úklid (blob_id, velikost, expirace).
use std::path::{Path as FsPath, PathBuf};
use std::sync::Arc;
use std::time::Duration as StdDuration;
use axum::body::Body;
use axum::extract::{Path, State};
use axum::http::{header, HeaderMap, StatusCode};
use axum::response::Response;
use axum::Json;
use chrono::{Duration, Utc};
use futures_util::StreamExt;
use rand::RngCore;
use serde::Serialize;
use tokio::io::AsyncWriteExt;
use crate::state::AppState;
/// TTL blobu — stejné okno jako offline fronta zpráv.
pub const ATTACHMENT_TTL_DAYS: i64 = 30;
/// Výchozí strop velikosti (šifrovaný blob). Přepis env `SEPOT_MAX_ATTACHMENT_BYTES`.
const DEFAULT_MAX_BYTES: u64 = 25 * 1024 * 1024;
/// Osiřelé `.part` soubory (pád mezi zápisem a přejmenováním) starší než
/// tohle se při úklidu mažou.
const PART_MAX_AGE: StdDuration = StdDuration::from_secs(24 * 3600);
#[derive(Clone)]
pub struct Config {
pub dir: PathBuf,
pub max_bytes: u64,
}
impl Config {
pub fn from_env() -> Self {
let dir = std::env::var("SEPOT_ATTACHMENTS_DIR")
.unwrap_or_else(|_| "./attachments".into());
let max_bytes = std::env::var("SEPOT_MAX_ATTACHMENT_BYTES")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(DEFAULT_MAX_BYTES);
Self {
dir: PathBuf::from(dir),
max_bytes,
}
}
fn blob_path(&self, blob_hex: &str) -> PathBuf {
self.dir.join(blob_hex)
}
fn tmp_dir(&self) -> PathBuf {
self.dir.join("tmp")
}
}
type ApiResult<T> = Result<T, (StatusCode, String)>;
fn internal<E: std::fmt::Display>(e: E) -> (StatusCode, String) {
tracing::warn!("interní chyba (přílohy): {e}");
(StatusCode::INTERNAL_SERVER_ERROR, "interní chyba".into())
}
/// Neexistence, expirace i špatný formát ID = stejná odpověď (nic neleakovat).
fn not_found() -> (StatusCode, String) {
(StatusCode::NOT_FOUND, "neznámá příloha".into())
}
// ── POST /v1/attachments ───────────────────────────────────────────────
#[derive(Serialize)]
pub struct UploadOut {
pub blob_id: String,
pub expires_at_ms: i64,
pub max_bytes: u64,
}
/// Streamuje tělo na disk (žádné buffrování celého blobu v RAM), strop
/// velikosti se vynucuje průběžně. Tělo = raw šifrované bajty
/// (application/octet-stream). Auth podpis: `"POST /v1/attachments\n<ts>"`.
pub async fn upload(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
body: Body,
) -> ApiResult<(StatusCode, Json<UploadOut>)> {
crate::rest::authenticate_device(&state, &headers, "POST /v1/attachments").await?;
let cfg = &state.attachments;
// Rychlé odmítnutí podle Content-Length, než tečou data
if let Some(len) = headers
.get(header::CONTENT_LENGTH)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.parse::<u64>().ok())
{
if len > cfg.max_bytes {
return Err((
StatusCode::PAYLOAD_TOO_LARGE,
"příloha je příliš velká".into(),
));
}
}
// ThreadRng je !Send — nesmí přežít přes await, proto vlastní scope.
let blob_id = {
let mut rng = rand::rng();
let mut b = vec![0u8; 16];
rng.fill_bytes(&mut b);
b
};
let blob_hex = hex::encode(&blob_id);
tokio::fs::create_dir_all(cfg.tmp_dir())
.await
.map_err(internal)?;
let tmp_path = cfg.tmp_dir().join(format!("{blob_hex}.part"));
let written = match write_stream(body, &tmp_path, cfg.max_bytes).await {
Ok(n) => n,
Err(e) => {
let _ = tokio::fs::remove_file(&tmp_path).await;
return Err(e);
}
};
if written == 0 {
let _ = tokio::fs::remove_file(&tmp_path).await;
return Err((StatusCode::BAD_REQUEST, "prázdné tělo".into()));
}
let expires_at = Utc::now() + Duration::days(ATTACHMENT_TTL_DAYS);
let final_path = cfg.blob_path(&blob_hex);
tokio::fs::rename(&tmp_path, &final_path)
.await
.map_err(internal)?;
let inserted = sqlx::query(
"INSERT INTO attachments (blob_id, size_bytes, expires_at) VALUES ($1, $2, $3)",
)
.bind(&blob_id)
.bind(written as i64)
.bind(expires_at)
.execute(&state.pool)
.await;
if let Err(e) = inserted {
let _ = tokio::fs::remove_file(&final_path).await;
return Err(internal(e));
}
// Log bez blob_id — jen velikost (kapacitní metrika)
tracing::debug!(size = written, "příloha uložena");
Ok((
StatusCode::CREATED,
Json(UploadOut {
blob_id: blob_hex,
expires_at_ms: expires_at.timestamp_millis(),
max_bytes: cfg.max_bytes,
}),
))
}
async fn write_stream(body: Body, path: &FsPath, max_bytes: u64) -> ApiResult<u64> {
let mut file = tokio::fs::File::create(path).await.map_err(internal)?;
let mut stream = body.into_data_stream();
let mut written: u64 = 0;
while let Some(chunk) = stream.next().await {
let chunk =
chunk.map_err(|_| (StatusCode::BAD_REQUEST, "tělo se nepodařilo přečíst".into()))?;
written += chunk.len() as u64;
if written > max_bytes {
return Err((
StatusCode::PAYLOAD_TOO_LARGE,
"příloha je příliš velká".into(),
));
}
file.write_all(&chunk).await.map_err(internal)?;
}
file.sync_all().await.map_err(internal)?;
Ok(written)
}
// ── GET /v1/attachments/{blob_id} ──────────────────────────────────────
/// Bez auth (bearer capability), bez logování blob_id. Streamuje z disku.
pub async fn download(
State(state): State<Arc<AppState>>,
Path(blob_hex): Path<String>,
) -> Result<Response, (StatusCode, String)> {
// 16 B hex + jen lowercase hex znaky, jinak jednotné 404
if blob_hex.len() != 32 || !blob_hex.bytes().all(|b| b.is_ascii_hexdigit()) {
return Err(not_found());
}
let blob_id = hex::decode(&blob_hex).map_err(|_| not_found())?;
let row: Option<(i64,)> = sqlx::query_as(
"SELECT size_bytes FROM attachments WHERE blob_id = $1 AND expires_at > now()",
)
.bind(&blob_id)
.fetch_optional(&state.pool)
.await
.map_err(internal)?;
let Some((size,)) = row else {
return Err(not_found());
};
let file = tokio::fs::File::open(state.attachments.blob_path(&blob_hex))
.await
.map_err(|_| not_found())?;
let stream = tokio_util::io::ReaderStream::new(file);
Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, "application/octet-stream")
.header(header::CONTENT_LENGTH, size)
.header(header::CACHE_CONTROL, "no-store")
.body(Body::from_stream(stream))
.map_err(internal)
}
// ── Úklid ──────────────────────────────────────────────────────────────
/// Hodinový úklid: expirované přílohy (řádek + soubor), expirovaná fronta
/// zpráv (dosud se jen filtrovala při čtení) a osiřelé `.part` soubory.
pub fn spawn_cleanup(state: Arc<AppState>) {
tokio::spawn(async move {
let mut tick = tokio::time::interval(StdDuration::from_secs(3600));
loop {
tick.tick().await;
if let Err(e) = sweep(&state).await {
tracing::warn!("úklid příloh selhal: {e}");
}
}
});
}
pub(crate) async fn sweep(state: &AppState) -> anyhow::Result<()> {
let expired: Vec<(Vec<u8>,)> =
sqlx::query_as("DELETE FROM attachments WHERE expires_at < now() RETURNING blob_id")
.fetch_all(&state.pool)
.await?;
let count = expired.len();
for (blob_id,) in expired {
let path = state.attachments.blob_path(&hex::encode(blob_id));
let _ = tokio::fs::remove_file(path).await;
}
if count > 0 {
tracing::info!(count, "expirované přílohy smazány");
}
sqlx::query("DELETE FROM message_queue WHERE expires_at < now()")
.execute(&state.pool)
.await?;
// Osiřelé .part (pád uprostřed uploadu) starší než PART_MAX_AGE
if let Ok(mut entries) = tokio::fs::read_dir(state.attachments.tmp_dir()).await {
while let Ok(Some(entry)) = entries.next_entry().await {
let stale = entry
.metadata()
.await
.ok()
.and_then(|m| m.modified().ok())
.and_then(|t| t.elapsed().ok())
.is_some_and(|age| age > PART_MAX_AGE);
if stale {
let _ = tokio::fs::remove_file(entry.path()).await;
}
}
}
Ok(())
}
#[cfg(test)]
mod db_tests {
//! Integrace nad reálným Postgresem (dev DB) — spouštět
//! `cargo test -p sepot-relay -- --ignored`. Uklízí po sobě.
use super::*;
async fn test_state() -> AppState {
let url = std::env::var("DATABASE_URL")
.unwrap_or_else(|_| "postgres://sepot:sepot_dev@127.0.0.1/sepot_relay".into());
let pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(2)
.connect(&url)
.await
.expect("dev DB");
let certs = crate::certs::CertAuthority::init(&pool).await.unwrap();
let dir = std::env::temp_dir().join(format!("sepot-att-test-{}", std::process::id()));
std::fs::create_dir_all(dir.join("tmp")).unwrap();
AppState::new(
pool,
certs,
Config {
dir,
max_bytes: 1024,
},
)
}
fn rand_blob_id() -> Vec<u8> {
let mut b = vec![0u8; 16];
rand::rng().fill_bytes(&mut b);
b
}
#[tokio::test]
#[ignore = "vyžaduje běžící Postgres (dev DB)"]
async fn sweep_maze_expirovane_vcetne_souboru() {
let state = test_state().await;
// expirovaná příloha se souborem na disku
let dead = rand_blob_id();
let dead_path = state.attachments.blob_path(&hex::encode(&dead));
std::fs::write(&dead_path, b"ciphertext").unwrap();
sqlx::query(
"INSERT INTO attachments (blob_id, size_bytes, expires_at)
VALUES ($1, 10, now() - interval '1 hour')",
)
.bind(&dead)
.execute(&state.pool)
.await
.unwrap();
// živá příloha zůstává
let live = rand_blob_id();
let live_path = state.attachments.blob_path(&hex::encode(&live));
std::fs::write(&live_path, b"ciphertext").unwrap();
sqlx::query(
"INSERT INTO attachments (blob_id, size_bytes, expires_at)
VALUES ($1, 10, now() + interval '1 day')",
)
.bind(&live)
.execute(&state.pool)
.await
.unwrap();
sweep(&state).await.unwrap();
assert!(!dead_path.exists(), "expirovaný soubor měl zmizet");
assert!(live_path.exists(), "živý soubor musí zůstat");
let remaining: i64 =
sqlx::query_scalar("SELECT count(*) FROM attachments WHERE blob_id = $1")
.bind(&live)
.fetch_one(&state.pool)
.await
.unwrap();
assert_eq!(remaining, 1);
// úklid po testu
sqlx::query("DELETE FROM attachments WHERE blob_id = $1")
.bind(&live)
.execute(&state.pool)
.await
.unwrap();
let _ = std::fs::remove_dir_all(&state.attachments.dir);
}
}
+106
View File
@@ -0,0 +1,106 @@
//! Certifikační autorita sealed senderu (server-signed).
//!
//! Relay drží dlouhodobý server keypair certifikovaný trust rootem a vydává
//! klientům krátkodobé SenderCertificate. Klienti mají trust root pinnutý.
use libsignal_protocol::{
DeviceId, KeyPair, PrivateKey, PublicKey, SenderCertificate, ServerCertificate, Timestamp,
};
use sqlx::PgPool;
/// Platnost vydaného SenderCertificate (~7 dní).
pub const SENDER_CERT_VALIDITY_MS: u64 = 7 * 24 * 3600 * 1000;
pub struct CertAuthority {
pub trust_root_public: Vec<u8>,
server_private: PrivateKey,
server_certificate: Vec<u8>,
}
impl CertAuthority {
/// Načte klíče z DB; při prvním startu je vygeneruje a uloží.
/// Trust root private v DB je dev pohodlí kostry — v produkci se
/// vygeneruje offline a v DB bude jen server keypair + certifikát.
pub async fn init(pool: &PgPool) -> anyhow::Result<Self> {
let row: Option<(Vec<u8>, Vec<u8>, Vec<u8>)> = sqlx::query_as(
"SELECT trust_root_public, server_private, server_certificate FROM relay_keys WHERE id = 1",
)
.fetch_optional(pool)
.await?;
if let Some((trust_root_public, server_private, server_certificate)) = row {
return Ok(Self {
trust_root_public,
server_private: PrivateKey::deserialize(&server_private)
.map_err(|e| anyhow::anyhow!("server_private: {e}"))?,
server_certificate,
});
}
let mut rng = rand::rng();
let trust_root = KeyPair::generate(&mut rng);
let server = KeyPair::generate(&mut rng);
let server_cert = ServerCertificate::new(
1,
server.public_key,
&trust_root.private_key,
&mut rng,
)
.map_err(|e| anyhow::anyhow!("ServerCertificate: {e}"))?;
let server_cert_bytes = server_cert
.serialized()
.map_err(|e| anyhow::anyhow!("serialized: {e}"))?
.to_vec();
sqlx::query(
"INSERT INTO relay_keys
(id, trust_root_public, trust_root_private, server_public, server_private, server_certificate)
VALUES (1, $1, $2, $3, $4, $5)",
)
.bind(trust_root.public_key.serialize().to_vec())
.bind(trust_root.private_key.serialize())
.bind(server.public_key.serialize().to_vec())
.bind(server.private_key.serialize())
.bind(&server_cert_bytes)
.execute(pool)
.await?;
tracing::info!("vygenerován nový trust root + server certificate");
Ok(Self {
trust_root_public: trust_root.public_key.serialize().to_vec(),
server_private: server.private_key,
server_certificate: server_cert_bytes,
})
}
/// Vydá SenderCertificate pro autentizované zařízení.
/// `sender_uuid` = hex(account_id) — interní routing handle, NE fingerprint.
pub fn issue(
&self,
sender_uuid: &str,
device_id: u32,
identity_key: &[u8],
now_ms: u64,
) -> anyhow::Result<(Vec<u8>, u64)> {
let expiration = now_ms + SENDER_CERT_VALIDITY_MS;
let mut rng = rand::rng();
let cert = SenderCertificate::new(
sender_uuid.to_string(),
None,
PublicKey::deserialize(identity_key)
.map_err(|e| anyhow::anyhow!("identity_key: {e}"))?,
DeviceId::try_from(device_id).map_err(|e| anyhow::anyhow!("device_id: {e}"))?,
Timestamp::from_epoch_millis(expiration),
ServerCertificate::deserialize(&self.server_certificate)
.map_err(|e| anyhow::anyhow!("server_certificate: {e}"))?,
&self.server_private,
&mut rng,
)
.map_err(|e| anyhow::anyhow!("SenderCertificate: {e}"))?;
let bytes = cert
.serialized()
.map_err(|e| anyhow::anyhow!("serialized: {e}"))?
.to_vec();
Ok((bytes, expiration))
}
}
+154
View File
@@ -0,0 +1,154 @@
//! FCM HTTP v1 sender — opt-in „fast mode".
//!
//! Stejné tvrdé pravidlo jako u UnifiedPush: posílá se JEN konstantní
//! ťuk `{"v":1}` v data payloadu — žádný obsah, žádný odesílatel.
//! Uživatel si FCM vědomě zapíná (vystavuje svůj push token Googlu
//! výměnou za spolehlivé/rychlé buzení bez další aplikace).
//!
//! Credentials: `SEPOT_FCM_CREDENTIALS` = cesta k service account JSON
//! (Google SA s rolí Firebase Cloud Messaging Admin). Bez env proměnné
//! zůstává fast mode vypnutý a FCM registrace se tiše ignorují.
//!
//! OAuth2: RS256 JWT podepsaný SA klíčem → access token (cache do
//! expirace s 60s rezervou).
use std::sync::Arc;
use chrono::Utc;
use jsonwebtoken::{Algorithm, EncodingKey, Header};
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;
const FCM_SCOPE: &str = "https://www.googleapis.com/auth/firebase.messaging";
#[derive(Deserialize)]
struct ServiceAccount {
project_id: String,
client_email: String,
private_key: String,
token_uri: String,
}
#[derive(Serialize)]
struct Claims<'a> {
iss: &'a str,
scope: &'a str,
aud: &'a str,
iat: i64,
exp: i64,
}
pub struct FcmSender {
project_id: String,
client_email: String,
token_uri: String,
key: EncodingKey,
/// (access_token, expirace unix)
cached: Mutex<Option<(String, i64)>>,
}
impl FcmSender {
/// Načte SA credentials z `SEPOT_FCM_CREDENTIALS`. None = fast mode off.
pub fn from_env() -> Option<Arc<FcmSender>> {
let path = std::env::var("SEPOT_FCM_CREDENTIALS").ok()?;
let raw = match std::fs::read(&path) {
Ok(r) => r,
Err(e) => {
tracing::warn!("FCM credentials nejdou přečíst: {e}");
return None;
}
};
let sa: ServiceAccount = match serde_json::from_slice(&raw) {
Ok(s) => s,
Err(e) => {
tracing::warn!("FCM credentials nejdou parsovat: {e}");
return None;
}
};
let key = match EncodingKey::from_rsa_pem(sa.private_key.as_bytes()) {
Ok(k) => k,
Err(e) => {
tracing::warn!("FCM privátní klíč nejde načíst: {e}");
return None;
}
};
tracing::info!("FCM fast mode aktivní (projekt nakonfigurován)");
Some(Arc::new(FcmSender {
project_id: sa.project_id,
client_email: sa.client_email,
token_uri: sa.token_uri,
key,
cached: Mutex::new(None),
}))
}
async fn access_token(&self, http: &reqwest::Client) -> Option<String> {
let now = Utc::now().timestamp();
let mut cached = self.cached.lock().await;
if let Some((token, exp)) = cached.as_ref() {
if *exp > now + 60 {
return Some(token.clone());
}
}
let claims = Claims {
iss: &self.client_email,
scope: FCM_SCOPE,
aud: &self.token_uri,
iat: now,
exp: now + 3600,
};
let jwt = jsonwebtoken::encode(&Header::new(Algorithm::RS256), &claims, &self.key)
.map_err(|e| tracing::warn!("FCM JWT podpis selhal: {e}"))
.ok()?;
#[derive(Deserialize)]
struct TokenOut {
access_token: String,
expires_in: i64,
}
let resp: TokenOut = http
.post(&self.token_uri)
.form(&[
("grant_type", "urn:ietf:params:oauth:grant-type:jwt-bearer"),
("assertion", &jwt),
])
.send()
.await
.ok()?
.json()
.await
.map_err(|e| tracing::debug!("FCM token endpoint: {e}"))
.ok()?;
let token = resp.access_token.clone();
*cached = Some((resp.access_token, now + resp.expires_in));
Some(token)
}
/// Pošle budíček na FCM registrační token. Best-effort; loguje se jen
/// stavový kód (nikdy token).
pub async fn send_wake(&self, http: &reqwest::Client, device_token: &str) {
let Some(access) = self.access_token(http).await else {
return;
};
let url = format!(
"https://fcm.googleapis.com/v1/projects/{}/messages:send",
self.project_id
);
let body = serde_json::json!({
"message": {
"token": device_token,
"data": { "v": "1" },
"android": { "priority": "HIGH" }
}
});
match http.post(&url).bearer_auth(access).json(&body).send().await {
Ok(resp) if !resp.status().is_success() => {
tracing::debug!(status = %resp.status(), "FCM budíček odmítnut");
}
Err(e) => tracing::debug!("FCM budíček selhal: {e}"),
_ => {}
}
}
}
+163
View File
@@ -0,0 +1,163 @@
//! Šepot relay — hloupý store-and-forward server.
//!
//! Nevidí obsah (E2E), nevidí odesílatele (sealed sender), neloguje IP.
//! Zná jen: rotovatelný routing token příjemce, msg_uuid (dedup) a opaque
//! payload. Jediná krypto role: certifikační autorita sealed senderu
//! (vydává SenderCertificate) + ověření XEdDSA podpisu při AUTH.
mod apns;
mod attachments;
mod certs;
mod fcm;
mod push;
mod rest;
mod state;
mod ws;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use axum::extract::DefaultBodyLimit;
use axum::http::StatusCode;
use axum::routing::{delete, get, post};
use axum::Router;
use sqlx::postgres::PgPoolOptions;
use tower_governor::governor::GovernorConfigBuilder;
use tower_governor::key_extractor::SmartIpKeyExtractor;
use tower_governor::GovernorLayer;
use tower_http::timeout::TimeoutLayer;
use tracing_subscriber::EnvFilter;
/// Max velikost request body (registrace se 100 OTP klíči má ~15 KB).
const MAX_BODY_BYTES: usize = 64 * 1024;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
// Log formát bez IP adres a bez identifikátorů klientů.
tracing_subscriber::fmt()
.with_env_filter(
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
)
.init();
let database_url = std::env::var("DATABASE_URL")
.unwrap_or_else(|_| "postgres://sepot:sepot_dev@127.0.0.1/sepot_relay".into());
let bind = std::env::var("SEPOT_BIND").unwrap_or_else(|_| "127.0.0.1:8470".into());
let pool = PgPoolOptions::new()
.max_connections(8)
.connect(&database_url)
.await?;
sqlx::migrate!("./migrations").run(&pool).await?;
let certs = certs::CertAuthority::init(&pool).await?;
let attachments_cfg = attachments::Config::from_env();
tokio::fs::create_dir_all(attachments_cfg.dir.join("tmp")).await?;
let state = Arc::new(state::AppState::new(pool, certs, attachments_cfg));
attachments::spawn_cleanup(state.clone());
// ── Rate limiting (základní minimum; plný PoW/Privacy Pass později) ─
//
// Klíč = klientská IP: za traefikem X-Forwarded-For/X-Real-Ip (na
// relay dosáhne jen traefik z docker sítě), jinak peer adresa.
// In-memory token bucket — nikdy se nezapisuje na disk (privacy).
//
// Známý limit: Tor onion klienti přicházejí z tor kontejneru bez XFF
// → sdílí jeden bucket. Pro v1 (nízký provoz) přijatelné, řeší až
// Privacy Pass — limity proto raději volnější.
// registrace: zápis do DB bez auth → nejpřísnější (burst 10, +1/5 min)
let gov_register = Arc::new(
GovernorConfigBuilder::default()
.key_extractor(SmartIpKeyExtractor)
.burst_size(10)
.period(Duration::from_secs(300))
.finish()
.expect("governor config"),
);
// čtecí API (bundle žere OTP klíče → drain ochrana): burst 60, +2/s
let gov_general = Arc::new(
GovernorConfigBuilder::default()
.key_extractor(SmartIpKeyExtractor)
.burst_size(60)
.per_second(2)
.finish()
.expect("governor config"),
);
// přílohy: upload je autentizovaný, ale drahý (disk) → burst 30, +1/s
let gov_attachments = Arc::new(
GovernorConfigBuilder::default()
.key_extractor(SmartIpKeyExtractor)
.burst_size(30)
.per_second(1)
.finish()
.expect("governor config"),
);
// Governor drží mapu IP→bucket v paměti — periodicky vyhazovat staré
for limiter in [
gov_register.limiter().clone(),
gov_general.limiter().clone(),
gov_attachments.limiter().clone(),
] {
tokio::spawn(async move {
let mut tick = tokio::time::interval(Duration::from_secs(60));
loop {
tick.tick().await;
limiter.retain_recent();
}
});
}
let register_routes = Router::new()
.route("/v1/register", post(rest::register))
.layer(GovernorLayer::new(gov_register));
let general_routes = Router::new()
.route("/v1/server-info", get(rest::server_info))
.route("/v1/keys/{delivery_address}", get(rest::fetch_bundle))
.route("/v1/certificate/delivery", get(rest::certificate_delivery))
.route("/v1/identity", delete(rest::delete_identity))
.layer(GovernorLayer::new(gov_general));
let rest_routes = register_routes
.merge(general_routes)
.layer(TimeoutLayer::with_status_code(
StatusCode::REQUEST_TIMEOUT,
Duration::from_secs(30),
));
// Přílohy: velká těla tečou streamem na disk (strop hlídá handler,
// DefaultBodyLimit se na raw Body nevztahuje) + delší timeout na
// pomalé linky (25 MiB přes mobilní síť).
let attachment_routes = Router::new()
.route("/v1/attachments", post(attachments::upload))
.route("/v1/attachments/{blob_id}", get(attachments::download))
.layer(GovernorLayer::new(gov_attachments))
.layer(TimeoutLayer::with_status_code(
StatusCode::REQUEST_TIMEOUT,
Duration::from_secs(120),
));
let app = Router::new()
// health bez limitů — interní docker/uptime checky
.route("/v1/health", get(rest::health))
// WS má vlastní limity (conn cap per IP, auth deadline, frame rate)
.route("/v1/ws", get(ws::ws_handler))
.merge(rest_routes)
.merge(attachment_routes)
.layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
.with_state(state);
tracing::info!(%bind, "sepot-relay startuje");
let listener = tokio::net::TcpListener::bind(&bind).await?;
// ConnectInfo: peer adresa pro rate-limit fallback a WS conn cap
axum::serve(
listener,
app.into_make_service_with_connect_info::<SocketAddr>(),
)
.await?;
Ok(())
}
+260
View File
@@ -0,0 +1,260 @@
//! Push budíčky. TVRDÉ PRAVIDLO: přes push se posílá
//! VŽDY jen prázdný „ťuk" `{"v":1}` — nikdy ciphertext, nikdy odesílatel.
//! Klient se po probuzení sám připojí na WS a vyzvedne frontu sealed-sender
//! kanálem. Endpoint se nikde neloguje.
//!
//! Podporované cesty:
//! - UnifiedPush (3) / WebPush (4): HTTP POST na endpoint URL distributora.
//! - FCM (1) / APNs (2): opt-in „fast mode" — vyžaduje externí credentials
//! (Firebase/Apple). Implementováno v `fcm.rs` / `apns.rs`; zapojeno níže
//! v `send_wake`. Bez nakonfigurovaných credentials se cesta tiše přeskočí
//! a klient spadne zpět na výchozí UnifiedPush.
//!
//! SSRF: endpoint zadává klient, takže relay by jinak POSToval na libovolné
//! URL. `endpoint_allowed` proto vynucuje https (http jen pro .onion) a
//! odmítá cíle, které resolvují na interní/loopback/privátní adresy.
use std::net::IpAddr;
use std::time::Duration;
use reqwest::Client;
/// Platform kódy (musí sedět s proto RegisterPush.Platform).
pub const PLATFORM_NONE: i16 = 0;
pub const PLATFORM_FCM: i16 = 1;
pub const PLATFORM_APNS: i16 = 2;
pub const PLATFORM_UNIFIEDPUSH: i16 = 3;
pub const PLATFORM_WEBPUSH: i16 = 4;
pub const PLATFORM_MAX: i16 = 4;
/// Konstantní budíček — žádná informace. Šifrovat (RFC 8291) netřeba,
/// není co chránit.
const WAKE_BODY: &str = "{\"v\":1}";
/// HTTP klient pro budíčky: bez redirectů (anti-SSRF přes 3xx), krátký
/// timeout, generický UA.
pub fn build_client() -> Client {
Client::builder()
.redirect(reqwest::redirect::Policy::none())
.timeout(Duration::from_secs(5))
.connect_timeout(Duration::from_secs(3))
.user_agent("sepot-relay")
.build()
.expect("reqwest klient")
}
/// Pošle budíček. Best-effort: chyby se tiše polykají (offline zařízení,
/// vypnutý distributor…). Nic o endpointu se neloguje.
pub async fn send_wake(
client: &Client,
platform: i16,
token: &[u8],
fcm: Option<&crate::fcm::FcmSender>,
apns: Option<&crate::apns::ApnsSender>,
) {
match platform {
PLATFORM_UNIFIEDPUSH | PLATFORM_WEBPUSH => {
let Ok(url) = std::str::from_utf8(token) else { return };
if !endpoint_allowed(url).await {
tracing::debug!("push endpoint odmítnut (scheme/SSRF guard)");
return;
}
post_wake(client, url).await;
}
PLATFORM_FCM => {
// Opt-in fast mode — jen když je relay nakonfigurován
// (SEPOT_FCM_CREDENTIALS); jinak tichý skip, default
// (UnifiedPush) funguje vždy.
let (Some(sender), Ok(device_token)) = (fcm, std::str::from_utf8(token)) else {
tracing::debug!("FCM budíček přeskočen (bez credentials)");
return;
};
sender.send_wake(client, device_token).await;
}
PLATFORM_APNS => {
let (Some(sender), Ok(device_token)) = (apns, std::str::from_utf8(token)) else {
tracing::debug!("APNs budíček přeskočen (bez credentials)");
return;
};
sender.send_wake(client, device_token).await;
}
_ => {}
}
}
/// Samotný POST budíčku (bez guardu — ten řeší volající). Oddělené kvůli
/// testovatelnosti proti loopback mocku (guard by loopback jinak odmítl).
async fn post_wake(client: &Client, url: &str) {
let _ = client
.post(url)
.header("content-type", "application/json")
.header("ttl", "2419200") // WebPush TTL (28 dní), ntfy ignoruje
.body(WAKE_BODY)
.send()
.await;
}
/// Povolit jen bezpečné cíle. https vždy; http jen pro .onion (samo šifruje).
/// Odmítnout hostitele resolvující na interní adresy (blind-SSRF ochrana).
///
/// ZNÁMÁ LIMITACE (DNS rebinding / TOCTOU): guard resolvuje host přes
/// `lookup_host` zde, ale samotný POST v `post_wake` resolvuje znovu. Mezi
/// kontrolou a requestem může útočníkem řízené DNS vrátit napřed veřejnou a
/// pak interní adresu, čímž guard obejde. Plný fix = resolvovat jednou a
/// připojit se přímo na ověřenou IP (reqwest `resolve()` / vlastní connector),
/// což je mimo rozsah tohoto vydání. Dopad je zúžen tím, že tělo je konstantní
/// `{"v":1}`, redirecty jsou vypnuté a odpověď se nikam nevrací (blind).
async fn endpoint_allowed(url: &str) -> bool {
let Ok(parsed) = reqwest::Url::parse(url) else {
return false;
};
let Some(host) = parsed.host_str() else {
return false;
};
let is_onion = host.ends_with(".onion");
match parsed.scheme() {
"https" => {}
// .onion šifruje sám; relay stejně nemá tor egress, request tiše
// selže — ale schéma povolíme, ať je to připravené.
"http" if is_onion => return true,
_ => return false,
}
// .onion přes https (neobvyklé) — neresolvuje se přes DNS, pustit.
if is_onion {
return true;
}
// Vlastnit host+port, ať `parsed` nepřežívá await na lookup_host.
let host = host.to_string();
let port = parsed.port_or_known_default().unwrap_or(443);
drop(parsed);
let resolved = tokio::net::lookup_host((host.as_str(), port)).await;
let Ok(addrs) = resolved else {
return false; // host nejde resolvovat → nepouštět
};
let mut saw_any = false;
for a in addrs {
saw_any = true;
if is_internal_ip(&a.ip()) {
return false; // aspoň jedna interní adresa → celé odmítnout
}
}
saw_any
}
fn is_internal_v4(v4: &std::net::Ipv4Addr) -> bool {
v4.is_loopback()
|| v4.is_private()
|| v4.is_link_local()
|| v4.is_unspecified()
|| v4.is_broadcast()
|| v4.is_documentation()
// 100.64.0.0/10 CGNAT (Tailscale apod.)
|| (v4.octets()[0] == 100 && (v4.octets()[1] & 0xc0) == 0x40)
}
fn is_internal_ip(ip: &IpAddr) -> bool {
match ip {
IpAddr::V4(v4) => is_internal_v4(v4),
IpAddr::V6(v6) => {
let o = v6.octets();
v6.is_loopback()
|| v6.is_unspecified()
|| (o[0] & 0xfe) == 0xfc // fc00::/7 ULA
|| (o[0] == 0xfe && (o[1] & 0xc0) == 0x80) // fe80::/10 link-local
|| v6.to_ipv4_mapped().map(|v4| is_internal_v4(&v4)).unwrap_or(false)
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::Ipv4Addr;
#[test]
fn interni_adresy_se_odmitnou() {
for ip in [
"127.0.0.1",
"10.0.0.5",
"192.168.1.1",
"172.16.0.5",
"169.254.1.1",
"0.0.0.0",
"100.64.0.1", // Tailscale CGNAT
] {
assert!(is_internal_ip(&ip.parse().unwrap()), "{ip} mělo být interní");
}
}
#[test]
fn verejne_adresy_projdou() {
for ip in ["1.1.1.1", "8.8.8.8", "203.0.113.10", "2606:4700:4700::1111"] {
assert!(!is_internal_ip(&ip.parse().unwrap()), "{ip} nemělo být interní");
}
}
#[test]
fn ipv4_mapped_v6_loopback_je_interni() {
let mapped: IpAddr = "::ffff:127.0.0.1".parse().unwrap();
assert!(is_internal_ip(&mapped));
}
#[tokio::test]
async fn schema_a_ssrf_guard() {
// http (ne-onion) se odmítne
assert!(!endpoint_allowed("http://ntfy.sh/abc").await);
// loopback se odmítne i přes https
assert!(!endpoint_allowed("https://127.0.0.1/abc").await);
assert!(!endpoint_allowed("https://localhost/abc").await);
// interní IP se odmítne
assert!(!endpoint_allowed("https://10.0.0.5:8000/x").await);
// http na .onion se pustí (schéma OK)
assert!(endpoint_allowed("http://cipjsctoeyf4fcrntvdvt22o2ansgr73qvthmewo246vuntd7cqmykyd.onion/x").await);
// nesmysl se odmítne
assert!(!endpoint_allowed("není url").await);
assert!(!endpoint_allowed("ftp://ntfy.sh/x").await);
}
#[test]
fn cgnat_hranice() {
assert!(is_internal_v4(&Ipv4Addr::new(100, 64, 0, 1)));
assert!(is_internal_v4(&Ipv4Addr::new(100, 127, 255, 255)));
// 100.63 a 100.128 jsou mimo CGNAT blok → veřejné
assert!(!is_internal_v4(&Ipv4Addr::new(100, 63, 0, 1)));
assert!(!is_internal_v4(&Ipv4Addr::new(100, 128, 0, 1)));
}
/// post_wake musí poslat POST s tělem přesně `{"v":1}` — žádný obsah.
/// Mock server na loopbacku (guard se tu záměrně obchází — testuje jen
/// tvar POSTu, guard má vlastní testy).
#[tokio::test]
async fn post_wake_posle_prazdny_tuk() {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
let (mut sock, _) = listener.accept().await.unwrap();
let mut buf = vec![0u8; 4096];
let n = sock.read(&mut buf).await.unwrap();
let req = String::from_utf8_lossy(&buf[..n]).to_string();
sock.write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n")
.await
.unwrap();
req
});
let client = build_client();
post_wake(&client, &format!("http://{addr}/up?token=abc")).await;
let req = server.await.unwrap();
assert!(req.starts_with("POST /up?token=abc "), "čekal POST, dostal: {req}");
assert!(req.contains("{\"v\":1}"), "tělo není prázdný ťuk: {req}");
// budíček nesmí nést nic dalšího
assert!(!req.to_lowercase().contains("payload"));
}
}
+438
View File
@@ -0,0 +1,438 @@
//! REST endpointy — registrace, prekey bundle, sender cert.
//! Chybové odpovědi nikdy neleakují interní detail; logy bez identifikátorů.
use std::sync::Arc;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, StatusCode};
use axum::Json;
use base64::prelude::*;
use libsignal_protocol::PublicKey;
use rand::RngCore;
use serde::{Deserialize, Serialize};
use crate::state::AppState;
type ApiResult<T> = Result<T, (StatusCode, String)>;
fn internal<E: std::fmt::Display>(e: E) -> (StatusCode, String) {
tracing::warn!("interní chyba: {e}");
(StatusCode::INTERNAL_SERVER_ERROR, "interní chyba".into())
}
fn bad(msg: &str) -> (StatusCode, String) {
(StatusCode::BAD_REQUEST, msg.into())
}
fn b64(name: &str, value: &str, expect_len: Option<usize>) -> ApiResult<Vec<u8>> {
let bytes = BASE64_STANDARD
.decode(value)
.map_err(|_| bad(&format!("pole {name} není platný base64")))?;
if let Some(len) = expect_len {
if bytes.len() != len {
return Err(bad(&format!("pole {name} má špatnou délku")));
}
}
Ok(bytes)
}
pub async fn health() -> Json<serde_json::Value> {
Json(serde_json::json!({ "status": "ok" }))
}
pub async fn server_info(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
Json(serde_json::json!({
"frame_version": 1,
"trust_root": BASE64_STANDARD.encode(&state.certs.trust_root_public),
"capabilities": ["pqxdh", "sealed-sender"],
}))
}
// ── POST /v1/register ──────────────────────────────────────────────────
#[derive(Deserialize)]
pub struct PreKeyIn {
pub key_id: i32,
pub public_key: String,
#[serde(default)]
pub signature: Option<String>,
#[serde(default)]
pub is_last_resort: bool,
}
#[derive(Deserialize)]
pub struct RegisterIn {
pub identity_key: String,
pub device_id: i32,
pub registration_id: i32,
pub delivery_address: String,
pub signed_prekey: PreKeyIn,
pub kyber_prekey: PreKeyIn,
#[serde(default)]
pub one_time_prekeys: Vec<PreKeyIn>,
}
#[derive(Serialize)]
pub struct RegisterOut {
pub account_id: String,
pub device_uuid: String,
}
/// Strop OTP klíčů v jedné registraci (klient posílá 100).
const MAX_ONE_TIME_PREKEYS: usize = 100;
pub async fn register(
State(state): State<Arc<AppState>>,
Json(body): Json<RegisterIn>,
) -> ApiResult<(StatusCode, Json<RegisterOut>)> {
if body.one_time_prekeys.len() > MAX_ONE_TIME_PREKEYS {
return Err(bad("příliš mnoho one-time prekeys"));
}
let identity_key = b64("identity_key", &body.identity_key, Some(33))?;
let delivery_address = b64("delivery_address", &body.delivery_address, Some(32))?;
let spk_pub = b64("signed_prekey.public_key", &body.signed_prekey.public_key, Some(33))?;
let spk_sig = b64(
"signed_prekey.signature",
body.signed_prekey.signature.as_deref().ok_or_else(|| bad("chybí podpis signed prekey"))?,
Some(64),
)?;
let kyber_pub = b64("kyber_prekey.public_key", &body.kyber_prekey.public_key, None)?;
let kyber_sig = b64(
"kyber_prekey.signature",
body.kyber_prekey.signature.as_deref().ok_or_else(|| bad("chybí podpis kyber prekey"))?,
Some(64),
)?;
// Hygiena: podpisy prekeys musí sedět k identity klíči už při uploadu.
let identity = PublicKey::deserialize(&identity_key).map_err(|_| bad("neplatný identity klíč"))?;
if !identity.verify_signature(&spk_pub, &spk_sig) {
return Err(bad("podpis signed prekey nesedí"));
}
if !identity.verify_signature(&kyber_pub, &kyber_sig) {
return Err(bad("podpis kyber prekey nesedí"));
}
// ThreadRng je !Send — nesmí přežít přes await, proto vlastní scope.
let (account_id, device_uuid) = {
let mut rng = rand::rng();
let mut account_id = vec![0u8; 16];
rng.fill_bytes(&mut account_id);
let mut device_uuid = vec![0u8; 16];
rng.fill_bytes(&mut device_uuid);
(account_id, device_uuid)
};
let mut tx = state.pool.begin().await.map_err(internal)?;
let inserted = sqlx::query(
"INSERT INTO identities (account_id, identity_key) VALUES ($1, $2)
ON CONFLICT (identity_key) DO NOTHING",
)
.bind(&account_id)
.bind(&identity_key)
.execute(&mut *tx)
.await
.map_err(internal)?;
if inserted.rows_affected() == 0 {
return Err((StatusCode::CONFLICT, "identita už existuje".into()));
}
sqlx::query(
"INSERT INTO devices (device_uuid, account_id, device_id, registration_id, delivery_address)
VALUES ($1, $2, $3, $4, $5)",
)
.bind(&device_uuid)
.bind(&account_id)
.bind(body.device_id)
.bind(body.registration_id)
.bind(&delivery_address)
.execute(&mut *tx)
.await
.map_err(internal)?;
sqlx::query(
"INSERT INTO signed_prekeys (device_uuid, key_id, public_key, signature)
VALUES ($1, $2, $3, $4)",
)
.bind(&device_uuid)
.bind(body.signed_prekey.key_id)
.bind(&spk_pub)
.bind(&spk_sig)
.execute(&mut *tx)
.await
.map_err(internal)?;
sqlx::query(
"INSERT INTO kyber_prekeys (device_uuid, key_id, public_key, signature, is_last_resort)
VALUES ($1, $2, $3, $4, $5)",
)
.bind(&device_uuid)
.bind(body.kyber_prekey.key_id)
.bind(&kyber_pub)
.bind(&kyber_sig)
.bind(body.kyber_prekey.is_last_resort)
.execute(&mut *tx)
.await
.map_err(internal)?;
for otp in &body.one_time_prekeys {
let otp_pub = b64("one_time_prekeys.public_key", &otp.public_key, Some(33))?;
sqlx::query(
"INSERT INTO one_time_prekeys (device_uuid, key_id, public_key) VALUES ($1, $2, $3)",
)
.bind(&device_uuid)
.bind(otp.key_id)
.bind(&otp_pub)
.execute(&mut *tx)
.await
.map_err(internal)?;
}
tx.commit().await.map_err(internal)?;
tracing::info!("nová registrace zařízení");
Ok((
StatusCode::CREATED,
Json(RegisterOut {
account_id: hex::encode(account_id),
device_uuid: hex::encode(device_uuid),
}),
))
}
// ── GET /v1/keys/{delivery_address} ───────────────────────────────────
#[derive(Serialize)]
pub struct PreKeyOut {
pub key_id: i32,
pub public_key: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub signature: Option<String>,
}
#[derive(Serialize)]
pub struct DeviceBundleOut {
pub device_id: i32,
pub registration_id: i32,
pub identity_key: String,
pub signed_prekey: PreKeyOut,
pub kyber_prekey: PreKeyOut,
#[serde(skip_serializing_if = "Option::is_none")]
pub one_time_prekey: Option<PreKeyOut>,
}
#[derive(Serialize)]
pub struct BundlesOut {
pub devices: Vec<DeviceBundleOut>,
}
pub async fn fetch_bundle(
State(state): State<Arc<AppState>>,
Path(delivery_address): Path<String>,
) -> ApiResult<Json<BundlesOut>> {
let address = hex::decode(&delivery_address).map_err(|_| bad("neplatná adresa"))?;
let account: Option<(Vec<u8>, Vec<u8>)> = sqlx::query_as(
"SELECT d.account_id, i.identity_key
FROM devices d JOIN identities i USING (account_id)
WHERE d.delivery_address = $1",
)
.bind(&address)
.fetch_optional(&state.pool)
.await
.map_err(internal)?;
let Some((account_id, identity_key)) = account else {
return Err((StatusCode::NOT_FOUND, "adresa neznámá".into()));
};
// Multi-device fanout: bundle pro všechna zařízení identity.
let devices: Vec<(Vec<u8>, i32, i32)> = sqlx::query_as(
"SELECT device_uuid, device_id, registration_id FROM devices WHERE account_id = $1",
)
.bind(&account_id)
.fetch_all(&state.pool)
.await
.map_err(internal)?;
let mut out = Vec::with_capacity(devices.len());
for (device_uuid, device_id, registration_id) in devices {
let spk: (i32, Vec<u8>, Vec<u8>) = sqlx::query_as(
"SELECT key_id, public_key, signature FROM signed_prekeys
WHERE device_uuid = $1 ORDER BY created_at DESC LIMIT 1",
)
.bind(&device_uuid)
.fetch_one(&state.pool)
.await
.map_err(internal)?;
let kyber: (i32, Vec<u8>, Vec<u8>) = sqlx::query_as(
"SELECT key_id, public_key, signature FROM kyber_prekeys
WHERE device_uuid = $1 ORDER BY is_last_resort ASC, created_at DESC LIMIT 1",
)
.bind(&device_uuid)
.fetch_one(&state.pool)
.await
.map_err(internal)?;
// Spotřebuj jeden one-time prekey (DELETE ... RETURNING).
let otp: Option<(i32, Vec<u8>)> = sqlx::query_as(
"DELETE FROM one_time_prekeys
WHERE (device_uuid, key_id) IN (
SELECT device_uuid, key_id FROM one_time_prekeys
WHERE device_uuid = $1 ORDER BY key_id LIMIT 1)
RETURNING key_id, public_key",
)
.bind(&device_uuid)
.fetch_optional(&state.pool)
.await
.map_err(internal)?;
out.push(DeviceBundleOut {
device_id,
registration_id,
identity_key: BASE64_STANDARD.encode(&identity_key),
signed_prekey: PreKeyOut {
key_id: spk.0,
public_key: BASE64_STANDARD.encode(spk.1),
signature: Some(BASE64_STANDARD.encode(spk.2)),
},
kyber_prekey: PreKeyOut {
key_id: kyber.0,
public_key: BASE64_STANDARD.encode(kyber.1),
signature: Some(BASE64_STANDARD.encode(kyber.2)),
},
one_time_prekey: otp.map(|(key_id, public_key)| PreKeyOut {
key_id,
public_key: BASE64_STANDARD.encode(public_key),
signature: None,
}),
});
}
Ok(Json(BundlesOut { devices: out }))
}
// ── Autentizace zařízení (X-Sepot-Auth) ────────────────────────────────
/// Max stáří podepsaného timestampu v X-Sepot-Auth (replay okno).
const AUTH_WINDOW_MS: i64 = 300_000;
pub(crate) struct AuthedDevice {
pub account_id: Vec<u8>,
pub identity_key: Vec<u8>,
pub device_id: i32,
}
/// Ověří hlavičku `X-Sepot-Auth: <device_uuid_hex>.<ts_ms>.<sig_b64>`,
/// kde sig = XEdDSA identity klíčem zařízení nad `"<method_path>\n<ts_ms>"`.
/// Sdílí ji sender certificate a upload příloh; do logů nejde nic.
pub(crate) async fn authenticate_device(
state: &AppState,
headers: &HeaderMap,
method_path: &str,
) -> Result<AuthedDevice, (StatusCode, String)> {
let unauthorized = || (StatusCode::UNAUTHORIZED, "neautorizováno".to_string());
let header = headers
.get("x-sepot-auth")
.and_then(|v| v.to_str().ok())
.ok_or_else(unauthorized)?;
let mut parts = header.splitn(3, '.');
let (device_hex, ts_str, sig_b64) = (
parts.next().ok_or_else(unauthorized)?,
parts.next().ok_or_else(unauthorized)?,
parts.next().ok_or_else(unauthorized)?,
);
let device_uuid = hex::decode(device_hex).map_err(|_| unauthorized())?;
let ts: i64 = ts_str.parse().map_err(|_| unauthorized())?;
let signature = BASE64_STANDARD.decode(sig_b64).map_err(|_| unauthorized())?;
let now_ms = chrono::Utc::now().timestamp_millis();
if (now_ms - ts).abs() > AUTH_WINDOW_MS {
return Err(unauthorized());
}
let row: Option<(Vec<u8>, Vec<u8>, i32)> = sqlx::query_as(
"SELECT d.account_id, i.identity_key, d.device_id
FROM devices d JOIN identities i USING (account_id)
WHERE d.device_uuid = $1",
)
.bind(&device_uuid)
.fetch_optional(&state.pool)
.await
.map_err(internal)?;
let Some((account_id, identity_key, device_id)) = row else {
return Err(unauthorized());
};
let message = format!("{method_path}\n{ts}");
let identity = PublicKey::deserialize(&identity_key).map_err(|_| unauthorized())?;
if !identity.verify_signature(message.as_bytes(), &signature) {
return Err(unauthorized());
}
Ok(AuthedDevice {
account_id,
identity_key,
device_id,
})
}
// ── DELETE /v1/identity ────────────────────────────────────────────────
/// Deregistrace účtu (GDPR čl. 17, store požadavek na smazání účtu).
/// Auth stejným podpisem jako sender certificate — smazat účet smí jen
/// držitel identity klíče. Kaskáda v DB odnese devices → prekeys +
/// push_tokens; fronta zpráv je vázaná na delivery_address bez FK,
/// maže se explicitně v téže transakci. Přílohy vazbu na účet nemají
/// (bearer capability) — dožijí TTL sweep.
pub async fn delete_identity(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
) -> ApiResult<StatusCode> {
let authed = authenticate_device(&state, &headers, "DELETE /v1/identity").await?;
state
.delete_identity(&authed.account_id)
.await
.map_err(internal)?;
tracing::info!("identita smazána na žádost zařízení");
Ok(StatusCode::NO_CONTENT)
}
// ── GET /v1/certificate/delivery ───────────────────────────────────────
#[derive(Serialize)]
pub struct CertificateOut {
pub sender_certificate: String,
pub trust_root: String,
pub sender_uuid: String,
pub expires_at_ms: u64,
}
pub async fn certificate_delivery(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
) -> ApiResult<Json<CertificateOut>> {
let authed =
authenticate_device(&state, &headers, "GET /v1/certificate/delivery").await?;
let now_ms = chrono::Utc::now().timestamp_millis();
let sender_uuid = hex::encode(&authed.account_id);
let (cert, expires_at_ms) = state
.certs
.issue(
&sender_uuid,
authed.device_id as u32,
&authed.identity_key,
now_ms as u64,
)
.map_err(internal)?;
Ok(Json(CertificateOut {
sender_certificate: BASE64_STANDARD.encode(cert),
trust_root: BASE64_STANDARD.encode(&state.certs.trust_root_public),
sender_uuid,
expires_at_ms,
}))
}
+416
View File
@@ -0,0 +1,416 @@
use chrono::{Duration, Utc};
use dashmap::DashMap;
use prost::Message;
use sepot_proto::v1;
use sepot_proto::MAX_ENVELOPE_PAYLOAD;
use sqlx::PgPool;
use tokio::sync::mpsc;
use crate::certs::CertAuthority;
/// TTL offline fronty.
pub const QUEUE_TTL_DAYS: i64 = 30;
pub struct AppState {
pub pool: PgPool,
/// delivery_address → kanál na živé WS spojení zařízení.
pub online: DashMap<Vec<u8>, mpsc::Sender<v1::ServerFrame>>,
/// klientská IP → počet souběžných WS spojení (anti-flood cap).
/// Jen v paměti, nikam se neloguje ani neukládá.
pub ws_conns: DashMap<std::net::IpAddr, u32>,
pub certs: CertAuthority,
/// HTTP klient na push budíčky (bez redirectů, krátký timeout).
pub http: reqwest::Client,
/// Úložiště šifrovaných příloh (adresář + strop velikosti).
pub attachments: crate::attachments::Config,
/// FCM fast mode (None = nenakonfigurováno, budíčky jen UnifiedPush).
pub fcm: Option<std::sync::Arc<crate::fcm::FcmSender>>,
/// APNs nativní iOS push (None = nenakonfigurováno).
pub apns: Option<std::sync::Arc<crate::apns::ApnsSender>>,
}
impl AppState {
pub fn new(pool: PgPool, certs: CertAuthority, attachments: crate::attachments::Config) -> Self {
Self {
pool,
online: DashMap::new(),
ws_conns: DashMap::new(),
certs,
http: crate::push::build_client(),
attachments,
fcm: crate::fcm::FcmSender::from_env(),
apns: crate::apns::ApnsSender::from_env(),
}
}
/// Přijme obálku: validace → perzistentní fronta (dedup) → pokud je
/// příjemce online, doručí hned. At-least-once: záznam z fronty maže
/// až ACK příjemce, ne odeslání.
pub async fn route_envelope(
&self,
env: v1::Envelope,
) -> anyhow::Result<v1::receipt::Status> {
if env.dst_address.len() != 32
|| env.msg_uuid.len() != 16
|| env.payload.is_empty()
|| env.payload.len() > MAX_ENVELOPE_PAYLOAD
{
anyhow::bail!("neplatná obálka");
}
let dst = env.dst_address.clone();
let encoded = env.encode_to_vec();
let expires = Utc::now() + Duration::days(QUEUE_TTL_DAYS);
sqlx::query(
"INSERT INTO message_queue (msg_uuid, dst_address, envelope, expires_at)
VALUES ($1, $2, $3, $4) ON CONFLICT (msg_uuid) DO NOTHING",
)
.bind(&env.msg_uuid)
.bind(&dst)
.bind(&encoded)
.bind(expires)
.execute(&self.pool)
.await?;
if let Some(tx) = self.online.get(&dst) {
let deliver = v1::ServerFrame {
frame: Some(v1::server_frame::Frame::Deliver(env)),
};
if tx.send(deliver).await.is_ok() {
return Ok(v1::receipt::Status::DeliveredOnline);
}
}
// Offline (nebo online kanál selhal) → zkusit push budíček.
self.wake(&dst).await;
Ok(v1::receipt::Status::StoredOffline)
}
/// Budíček pro offline příjemce: dohledá jeho push token a pošle prázdný
/// „ťuk". Best-effort, na pozadí — neblokuje receipt odesílateli.
/// Endpoint se NELOGUJE.
async fn wake(&self, dst_address: &[u8]) {
let Some((platform, token)) = self.push_target(dst_address).await else {
return; // bez tokenu / anonymní režim → žádný push
};
let client = self.http.clone();
let fcm = self.fcm.clone();
let apns = self.apns.clone();
tokio::spawn(async move {
crate::push::send_wake(&client, platform, &token, fcm.as_deref(), apns.as_deref())
.await;
});
}
/// Dohledá push cíl (platform, token) pro delivery_address příjemce.
/// `pub(crate)` kvůli integračnímu testu correctness JOINu.
pub(crate) async fn push_target(&self, dst_address: &[u8]) -> Option<(i16, Vec<u8>)> {
let row: Option<(i16, Option<Vec<u8>>)> = sqlx::query_as(
"SELECT p.platform, p.token FROM push_tokens p
JOIN devices d USING (device_uuid)
WHERE d.delivery_address = $1",
)
.bind(dst_address)
.fetch_optional(&self.pool)
.await
.ok()
.flatten();
match row {
Some((platform, Some(token))) => Some((platform, token)),
_ => None,
}
}
/// Registrace/aktualizace push endpointu zařízení. platform=NONE token
/// smaže (anonymní režim). Volá se z autentizovaného WS.
pub async fn register_push(
&self,
device_uuid: &[u8],
platform: i16,
endpoint: Option<&[u8]>,
) -> anyhow::Result<()> {
if platform == crate::push::PLATFORM_NONE || endpoint.is_none() {
sqlx::query("DELETE FROM push_tokens WHERE device_uuid = $1")
.bind(device_uuid)
.execute(&self.pool)
.await?;
return Ok(());
}
sqlx::query(
"INSERT INTO push_tokens (device_uuid, platform, token, updated_at)
VALUES ($1, $2, $3, now())
ON CONFLICT (device_uuid)
DO UPDATE SET platform = excluded.platform,
token = excluded.token,
updated_at = now()",
)
.bind(device_uuid)
.bind(platform)
.bind(endpoint)
.execute(&self.pool)
.await?;
Ok(())
}
/// Deregistrace účtu (GDPR čl. 17). Kaskáda v DB odnese devices →
/// prekeys + push_tokens; fronta zpráv je vázaná na delivery_address
/// bez FK, maže se explicitně v téže transakci. Přílohy vazbu na účet
/// nemají (bearer capability) — dožijí TTL sweep.
pub async fn delete_identity(&self, account_id: &[u8]) -> anyhow::Result<()> {
let mut tx = self.pool.begin().await?;
sqlx::query(
"DELETE FROM message_queue WHERE dst_address IN (
SELECT delivery_address FROM devices WHERE account_id = $1)",
)
.bind(account_id)
.execute(&mut *tx)
.await?;
sqlx::query("DELETE FROM identities WHERE account_id = $1")
.bind(account_id)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(())
}
/// Po AUTH: vyleje nedoručenou frontu zařízení (DELIVER stream).
pub async fn flush_queue(
&self,
delivery_address: &[u8],
tx: &mpsc::Sender<v1::ServerFrame>,
) -> anyhow::Result<usize> {
let rows: Vec<(Vec<u8>,)> = sqlx::query_as(
"SELECT envelope FROM message_queue
WHERE dst_address = $1 AND expires_at > now()
ORDER BY enqueued_at",
)
.bind(delivery_address)
.fetch_all(&self.pool)
.await?;
let count = rows.len();
for (encoded,) in rows {
let env = v1::Envelope::decode(encoded.as_slice())?;
let frame = v1::ServerFrame {
frame: Some(v1::server_frame::Frame::Deliver(env)),
};
if tx.send(frame).await.is_err() {
break;
}
}
Ok(count)
}
/// ACK příjemce → hard-delete z fronty (žádná retence po doručení).
pub async fn ack(&self, delivery_address: &[u8], msg_uuid: &[u8]) -> anyhow::Result<()> {
sqlx::query("DELETE FROM message_queue WHERE msg_uuid = $1 AND dst_address = $2")
.bind(msg_uuid)
.bind(delivery_address)
.execute(&self.pool)
.await?;
Ok(())
}
}
#[cfg(test)]
mod db_tests {
//! Integrace push registrace nad reálným Postgresem. #[ignore] — spouštět
//! s běžící dev DB: `cargo test -p sepot-relay -- --ignored`. Používá
//! náhodná ID a po sobě uklízí (žádná pollution existujících dat).
use super::*;
use rand::RngCore;
async fn test_state() -> AppState {
let url = std::env::var("DATABASE_URL")
.unwrap_or_else(|_| "postgres://sepot:sepot_dev@127.0.0.1/sepot_relay".into());
let pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(2)
.connect(&url)
.await
.expect("dev DB");
let certs = crate::certs::CertAuthority::init(&pool).await.unwrap();
AppState::new(pool, certs, crate::attachments::Config::from_env())
}
fn rand_bytes(n: usize) -> Vec<u8> {
let mut v = vec![0u8; n];
rand::rng().fill_bytes(&mut v);
v
}
/// Naseje identitu+zařízení. Vrací (account_id, device_uuid, delivery_address).
async fn seed_device(state: &AppState) -> (Vec<u8>, Vec<u8>, Vec<u8>) {
let account_id = rand_bytes(16);
let identity_key = rand_bytes(33); // CHECK octet_length = 33
let device_uuid = rand_bytes(16);
let delivery_address = rand_bytes(32);
sqlx::query("INSERT INTO identities (account_id, identity_key) VALUES ($1, $2)")
.bind(&account_id)
.bind(&identity_key)
.execute(&state.pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO devices
(device_uuid, account_id, device_id, registration_id, delivery_address)
VALUES ($1, $2, 1, 1, $3)",
)
.bind(&device_uuid)
.bind(&account_id)
.bind(&delivery_address)
.execute(&state.pool)
.await
.unwrap();
(account_id, device_uuid, delivery_address)
}
async fn cleanup(state: &AppState, account_id: &[u8], delivery_address: &[u8]) {
sqlx::query("DELETE FROM message_queue WHERE dst_address = $1")
.bind(delivery_address)
.execute(&state.pool)
.await
.unwrap();
// CASCADE smaže devices i push_tokens
sqlx::query("DELETE FROM identities WHERE account_id = $1")
.bind(account_id)
.execute(&state.pool)
.await
.unwrap();
}
#[tokio::test]
#[ignore = "vyžaduje běžící Postgres (dev DB)"]
async fn register_push_upsert_a_lookup() {
let state = test_state().await;
let (account_id, device_uuid, delivery_address) = seed_device(&state).await;
// bez registrace → žádný cíl
assert!(state.push_target(&delivery_address).await.is_none());
// UnifiedPush endpoint
let ep1 = b"https://ntfy.sh/UPabc".to_vec();
state
.register_push(&device_uuid, crate::push::PLATFORM_UNIFIEDPUSH, Some(&ep1))
.await
.unwrap();
assert_eq!(
state.push_target(&delivery_address).await,
Some((crate::push::PLATFORM_UNIFIEDPUSH, ep1))
);
// upsert na WebPush (jiná platforma i token)
let ep2 = b"https://push.example/w".to_vec();
state
.register_push(&device_uuid, crate::push::PLATFORM_WEBPUSH, Some(&ep2))
.await
.unwrap();
assert_eq!(
state.push_target(&delivery_address).await,
Some((crate::push::PLATFORM_WEBPUSH, ep2))
);
// NONE → smaže (anonymní režim)
state
.register_push(&device_uuid, crate::push::PLATFORM_NONE, None)
.await
.unwrap();
assert!(state.push_target(&delivery_address).await.is_none());
cleanup(&state, &account_id, &delivery_address).await;
}
#[tokio::test]
#[ignore = "vyžaduje běžící Postgres (dev DB)"]
async fn route_envelope_offline_ulozi_a_spusti_wake() {
let state = test_state().await;
let (account_id, device_uuid, delivery_address) = seed_device(&state).await;
// Endpoint schválně na loopback → guard ho odmítne, žádný externí POST
// (test zůstává hermetický). Ověřujeme, že offline cesta projde a
// route_envelope se na budíčku nezhroutí.
state
.register_push(
&device_uuid,
crate::push::PLATFORM_UNIFIEDPUSH,
Some(b"https://localhost/up"),
)
.await
.unwrap();
let env = v1::Envelope {
dst_address: delivery_address.clone(),
msg_uuid: rand_bytes(16),
content_hint: 0,
payload: vec![0xAB; 64],
};
let status = state.route_envelope(env).await.unwrap();
assert_eq!(status, v1::receipt::Status::StoredOffline);
// obálka je ve frontě (příjemce byl offline)
let queued: i64 =
sqlx::query_scalar("SELECT count(*) FROM message_queue WHERE dst_address = $1")
.bind(&delivery_address)
.fetch_one(&state.pool)
.await
.unwrap();
assert_eq!(queued, 1);
cleanup(&state, &account_id, &delivery_address).await;
}
#[tokio::test]
#[ignore = "vyžaduje běžící Postgres (dev DB)"]
async fn delete_identity_smaze_ucet_frontu_i_push() {
let state = test_state().await;
let (account_id, device_uuid, delivery_address) = seed_device(&state).await;
// účet má push token i nedoručenou zprávu ve frontě
state
.register_push(
&device_uuid,
crate::push::PLATFORM_UNIFIEDPUSH,
Some(b"https://localhost/up"),
)
.await
.unwrap();
sqlx::query(
"INSERT INTO message_queue (msg_uuid, dst_address, envelope, expires_at)
VALUES ($1, $2, $3, now() + interval '30 days')",
)
.bind(rand_bytes(16))
.bind(&delivery_address)
.bind(vec![0xCDu8; 64])
.execute(&state.pool)
.await
.unwrap();
state.delete_identity(&account_id).await.unwrap();
let identities: i64 =
sqlx::query_scalar("SELECT count(*) FROM identities WHERE account_id = $1")
.bind(&account_id)
.fetch_one(&state.pool)
.await
.unwrap();
let devices: i64 =
sqlx::query_scalar("SELECT count(*) FROM devices WHERE account_id = $1")
.bind(&account_id)
.fetch_one(&state.pool)
.await
.unwrap();
let queued: i64 =
sqlx::query_scalar("SELECT count(*) FROM message_queue WHERE dst_address = $1")
.bind(&delivery_address)
.fetch_one(&state.pool)
.await
.unwrap();
let tokens: i64 =
sqlx::query_scalar("SELECT count(*) FROM push_tokens WHERE device_uuid = $1")
.bind(&device_uuid)
.fetch_one(&state.pool)
.await
.unwrap();
assert_eq!((identities, devices, queued, tokens), (0, 0, 0, 0));
// idempotence: druhé smazání neexistujícího účtu nesmí spadnout
state.delete_identity(&account_id).await.unwrap();
}
}
+335
View File
@@ -0,0 +1,335 @@
//! WS relay kanál: AUTH → CHALLENGE → AUTH_PROOF,
//! pak ENVELOPE/ACK od klienta a RECEIPT/DELIVER od serveru.
//! Logy záměrně bez identifikátorů zařízení a bez IP.
//!
//! Anti-abuse (minimální sada):
//! - cap souběžných spojení per IP (flood ochranu doplňuje CrowdSec na edgi)
//! - deadline na AUTH — neautentizované spojení se po pár sekundách zavře
//! - max velikost WS zprávy těsně nad MAX_ENVELOPE_PAYLOAD
//! - rate limit rámců per spojení (token okno) — přetečení = disconnect
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;
use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade};
use axum::extract::{ConnectInfo, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::IntoResponse;
use futures_util::{SinkExt, StreamExt};
use libsignal_protocol::PublicKey;
use prost::Message;
use rand::RngCore;
use sepot_proto::{v1, AUTH_CONTEXT, MAX_ENVELOPE_PAYLOAD};
use tokio::sync::mpsc;
use tokio::time::Instant;
use crate::state::AppState;
/// Max souběžných WS spojení z jedné IP (za traefikem = reálná klientská
/// IP z X-Real-Ip; tor kontejner sdílí jednu — proto ne moc těsně).
const MAX_CONNS_PER_IP: u32 = 32;
/// Neautentizované spojení zavřít po tomto čase.
const AUTH_DEADLINE: Duration = Duration::from_secs(10);
/// Rate limit rámců od klienta: max FRAME_WINDOW_MAX za FRAME_WINDOW.
const FRAME_WINDOW: Duration = Duration::from_secs(10);
const FRAME_WINDOW_MAX: u32 = 200;
/// Klientská IP: hlavičky od trusted proxy (na relay dosáhne jen traefik
/// z docker sítě), fallback peer adresa (dev/tor).
///
/// TRUST MODEL — POZOR: X-Real-Ip / X-Forwarded-For se přebírají bezpodmínečně
/// a řídí per-IP limity (cap spojení, rate limit). To je bezpečné VÝHRADNĚ
/// proto, že před relayem stojí důvěryhodná reverzní proxy (traefik), která
/// tyhle hlavičky vždy přepisuje reálnou klientskou IP. Klientem poslané
/// hodnoty tak nikdy neprojdou. Kdyby byl relay vystaven přímo do internetu
/// (bez proxy, která hlavičky sanitizuje), útočník by je mohl libovolně
/// podvrhnout, tím rozprostřít provoz přes falešné IP a všechny per-IP limity
/// obejít. Relay proto NIKDY nenasazovat bez takové proxy.
fn client_ip(headers: &HeaderMap, peer: SocketAddr) -> IpAddr {
let from_header = |name: &str| {
headers
.get(name)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.split(',').next())
.and_then(|v| v.trim().parse::<IpAddr>().ok())
};
from_header("x-real-ip")
.or_else(|| from_header("x-forwarded-for"))
.unwrap_or_else(|| peer.ip())
}
/// Drop-guard: úbytek čítače spojení i při panice/cancelu handleru.
struct ConnGuard {
state: Arc<AppState>,
ip: IpAddr,
}
impl Drop for ConnGuard {
fn drop(&mut self) {
self.state.ws_conns.remove_if_mut(&self.ip, |_, n| {
*n -= 1;
*n == 0
});
}
}
pub async fn ws_handler(
State(state): State<Arc<AppState>>,
ConnectInfo(peer): ConnectInfo<SocketAddr>,
headers: HeaderMap,
ws: WebSocketUpgrade,
) -> impl IntoResponse {
let ip = client_ip(&headers, peer);
let over_cap = {
let mut entry = state.ws_conns.entry(ip).or_insert(0);
if *entry >= MAX_CONNS_PER_IP {
true
} else {
*entry += 1;
false
}
};
if over_cap {
// Log bez IP — jen fakt, že cap zafungoval
tracing::debug!("ws spojení odmítnuto (cap per IP)");
return StatusCode::TOO_MANY_REQUESTS.into_response();
}
let guard = ConnGuard {
state: state.clone(),
ip,
};
ws.max_message_size(MAX_ENVELOPE_PAYLOAD + 4096)
.max_frame_size(MAX_ENVELOPE_PAYLOAD + 4096)
.on_upgrade(move |socket| async move {
let _guard = guard;
handle(state, socket).await;
})
.into_response()
}
/// Stav mezi AUTH a AUTH_PROOF.
struct Pending {
nonce_client: Vec<u8>,
nonce_server: Vec<u8>,
delivery_address: Vec<u8>,
identity_key: Vec<u8>,
device_uuid: Vec<u8>,
}
fn server_frame(frame: v1::server_frame::Frame) -> v1::ServerFrame {
v1::ServerFrame { frame: Some(frame) }
}
fn error_frame(code: u32, message: &str) -> v1::ServerFrame {
server_frame(v1::server_frame::Frame::Error(v1::Error {
code,
message: message.into(),
}))
}
async fn handle(state: Arc<AppState>, socket: WebSocket) {
let (mut sink, mut stream) = socket.split();
let (tx, mut rx) = mpsc::channel::<v1::ServerFrame>(64);
let writer = tokio::spawn(async move {
while let Some(frame) = rx.recv().await {
let bytes = frame.encode_to_vec();
if sink.send(WsMessage::Binary(bytes.into())).await.is_err() {
break;
}
}
});
let mut pending: Option<Pending> = None;
let mut authed_address: Option<Vec<u8>> = None;
// device_uuid ověřeného zařízení — klíč pro push registraci.
let mut authed_device: Option<Vec<u8>> = None;
// Anti-abuse okno: počítá VŠECHNY rámce od klienta (envelope/ack/pong)
let mut window_start = Instant::now();
let mut window_frames: u32 = 0;
loop {
// Před AUTH platí deadline — viset neautentizovaný nejde
let next = if authed_address.is_none() {
match tokio::time::timeout(AUTH_DEADLINE, stream.next()).await {
Ok(next) => next,
Err(_) => {
tracing::debug!("ws zavřeno: AUTH deadline");
break;
}
}
} else {
stream.next().await
};
let Some(Ok(msg)) = next else { break };
let now = Instant::now();
if now.duration_since(window_start) > FRAME_WINDOW {
window_start = now;
window_frames = 0;
}
window_frames += 1;
if window_frames > FRAME_WINDOW_MAX {
let _ = tx.send(error_frame(429, "příliš mnoho rámců")).await;
break;
}
let data = match msg {
WsMessage::Binary(b) => b,
WsMessage::Close(_) => break,
_ => continue,
};
let Ok(frame) = v1::ClientFrame::decode(data.as_ref()) else {
let _ = tx.send(error_frame(400, "neplatný rámec")).await;
break;
};
match frame.frame {
Some(v1::client_frame::Frame::Auth(auth)) if authed_address.is_none() => {
if auth.device_uuid.len() != 16 || auth.nonce_client.len() != 16 {
let _ = tx.send(error_frame(400, "neplatný AUTH")).await;
break;
}
let row: Option<(Vec<u8>, Vec<u8>)> = sqlx::query_as(
"SELECT d.delivery_address, i.identity_key
FROM devices d JOIN identities i USING (account_id)
WHERE d.device_uuid = $1",
)
.bind(&auth.device_uuid)
.fetch_optional(&state.pool)
.await
.ok()
.flatten();
let Some((delivery_address, identity_key)) = row else {
let _ = tx.send(error_frame(401, "neautorizováno")).await;
break;
};
let mut nonce_server = vec![0u8; 32];
rand::rng().fill_bytes(&mut nonce_server);
let _ = tx
.send(server_frame(v1::server_frame::Frame::Challenge(
v1::Challenge {
nonce_server: nonce_server.clone(),
},
)))
.await;
pending = Some(Pending {
nonce_client: auth.nonce_client,
nonce_server,
delivery_address,
identity_key,
device_uuid: auth.device_uuid,
});
}
Some(v1::client_frame::Frame::AuthProof(proof)) if pending.is_some() => {
let pd = pending.take().expect("pending ověřen výše");
let mut message =
Vec::with_capacity(pd.nonce_client.len() + pd.nonce_server.len() + AUTH_CONTEXT.len());
message.extend_from_slice(&pd.nonce_client);
message.extend_from_slice(&pd.nonce_server);
message.extend_from_slice(AUTH_CONTEXT);
let verified = PublicKey::deserialize(&pd.identity_key)
.map(|pk| pk.verify_signature(&message, &proof.signature))
.unwrap_or(false);
if !verified {
let _ = tx.send(error_frame(401, "neplatný podpis")).await;
break;
}
state
.online
.insert(pd.delivery_address.clone(), tx.clone());
authed_address = Some(pd.delivery_address.clone());
// Potvrzení AUTH — klient až teď smí hlásit „připojeno"
let _ = tx
.send(server_frame(v1::server_frame::Frame::AuthOk(
v1::AuthOk {},
)))
.await;
// last_seen jen v hrubé granularitě (privacy — zaokrouhleno na hodinu).
let _ = sqlx::query(
"UPDATE devices SET last_seen_at = date_trunc('hour', now())
WHERE device_uuid = $1",
)
.bind(&pd.device_uuid)
.execute(&state.pool)
.await;
match state.flush_queue(&pd.delivery_address, &tx).await {
Ok(count) => {
tracing::info!(count, "klient autentizován, fronta vylita")
}
Err(e) => tracing::warn!("flush fronty selhal: {e}"),
}
authed_device = Some(pd.device_uuid);
}
Some(v1::client_frame::Frame::RegisterPush(reg)) if authed_address.is_some() => {
if let Some(device) = &authed_device {
let platform = reg.platform as i16;
if platform < 0 || platform > crate::push::PLATFORM_MAX {
let _ = tx.send(error_frame(400, "neznámá push platforma")).await;
} else {
let endpoint =
(!reg.endpoint.is_empty()).then_some(reg.endpoint.as_slice());
// Endpoint se ZÁMĚRNĚ neloguje (metadatový bod).
if let Err(e) =
state.register_push(device, platform, endpoint).await
{
tracing::warn!("register_push selhal: {e}");
}
}
}
}
Some(v1::client_frame::Frame::Envelope(envelope)) if authed_address.is_some() => {
let msg_uuid = envelope.msg_uuid.clone();
match state.route_envelope(envelope).await {
Ok(status) => {
let _ = tx
.send(server_frame(v1::server_frame::Frame::Receipt(
v1::Receipt {
msg_uuid,
status: status as i32,
},
)))
.await;
}
Err(_) => {
let _ = tx.send(error_frame(400, "obálka odmítnuta")).await;
}
}
}
Some(v1::client_frame::Frame::Ack(ack)) if authed_address.is_some() => {
if let Some(addr) = &authed_address {
if let Err(e) = state.ack(addr, &ack.msg_uuid).await {
tracing::warn!("ack selhal: {e}");
}
}
}
Some(v1::client_frame::Frame::Pong(_)) => {}
_ => {
let _ = tx.send(error_frame(403, "rámec mimo pořadí")).await;
break;
}
}
}
if let Some(addr) = authed_address {
state.online.remove(&addr);
}
writer.abort();
tracing::debug!("ws spojení uzavřeno");
}