Compare commits

..
152 changed files with 2542 additions and 5136 deletions
-1
View File
@@ -1 +0,0 @@
.containerignore
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as \"total!\"\n FROM (SELECT DISTINCT cid, size_bytes FROM blobs) t",
"query": "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as \"total!\" FROM blobs",
"describe": {
"columns": [
{
@@ -16,5 +16,5 @@
null
]
},
"hash": "155efbae4cd55f73ec0709dda7b18a76e92065e6ae4a6081bd38a19821fbfcc3"
"hash": "0890b2c7c921005f58ed0e57b6e062b2085ce804a4cccb27b4ae2ba6711f24c4"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT storage_key FROM blobs WHERE cid = $1 LIMIT 1",
"query": "SELECT storage_key FROM blobs WHERE cid = $1",
"describe": {
"columns": [
{
@@ -18,5 +18,5 @@
false
]
},
"hash": "9fb9e128076b20ff067d01955221488ce7e5b886dba0529fb073c3e0461fe030"
"hash": "6131bb5b39ca81bdbb193c0a9867bead8d9f3d793ad4eca97a79d166467a5052"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT cid, takedown_ref FROM blobs WHERE cid = $1 ORDER BY takedown_ref NULLS LAST LIMIT 1",
"query": "SELECT cid, takedown_ref FROM blobs WHERE cid = $1",
"describe": {
"columns": [
{
@@ -24,5 +24,5 @@
true
]
},
"hash": "5996484ff0f8dbc3b278cfd01b8375dbf7bf6da8d903145b12871dda6e1fd5d9"
"hash": "62942bd21d545eb15bfea4f46378b6c2ebfe12b8bc9e27c63a6c0f77a9105303"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT storage_key as \"storage_key!\" FROM blobs b\n WHERE created_by_user = $1\n AND NOT EXISTS (\n SELECT 1 FROM blobs o\n WHERE o.cid = b.cid AND o.created_by_user <> $1\n )",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "storage_key!",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false
]
},
"hash": "8844d942ef2810afc386e5a9838624ee07a43c380d2df31efdba5cf299aab571"
}
@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT u.id as user_id, u.did\n FROM users u\n JOIN repos r ON r.user_id = u.id\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "user_id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "did",
"type_info": "Text"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false
]
},
"hash": "8951136879711bca5b562c34f88e691a8ee16f370f6ef9b88ddb3873ddf2b45f"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key)\n VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (cid, created_by_user) DO NOTHING RETURNING cid",
"query": "INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key)\n VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (cid) DO NOTHING RETURNING cid",
"describe": {
"columns": [
{
@@ -22,5 +22,5 @@
false
]
},
"hash": "996e5513fb55670fe3304a6046381e377da6a187dfa3347bd285078a7b4410f2"
"hash": "8afea2b745385348f4c78b51f74145d6718bfcf9a3a0c218109ec691aeb930ba"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT rb.blob_cid, rb.record_uri\n FROM record_blobs rb\n LEFT JOIN blobs b ON rb.blob_cid = b.cid AND b.created_by_user = $1\n WHERE rb.repo_id = $1 AND b.cid IS NULL AND rb.blob_cid > $2\n ORDER BY rb.blob_cid\n LIMIT $3",
"query": "SELECT rb.blob_cid, rb.record_uri\n FROM record_blobs rb\n LEFT JOIN blobs b ON rb.blob_cid = b.cid\n WHERE rb.repo_id = $1 AND b.cid IS NULL AND rb.blob_cid > $2\n ORDER BY rb.blob_cid\n LIMIT $3",
"describe": {
"columns": [
{
@@ -26,5 +26,5 @@
false
]
},
"hash": "f8bb421e07e47f7b0a3b2789e368abfa6ad64152e8660e3efc7117b2d9320f22"
"hash": "8e88ec169f0ef14c7207944cd4c2c0970e302c0457f9e317ec752dc13a5b1393"
}
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM blobs WHERE cid = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text"
]
},
"nullable": []
},
"hash": "d2990ce7f233d2489bb36a63920571c9f454a0605cc463829693d581bc0dce12"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1 LIMIT 1",
"query": "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1",
"describe": {
"columns": [
{
@@ -30,5 +30,5 @@
false
]
},
"hash": "03f129e4984e1bed9e87294adc9caf1730906d889101b9039113ec8aa234618d"
"hash": "dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key)\n SELECT DISTINCT b.cid, b.mime_type, b.size_bytes, $1::uuid, b.storage_key\n FROM blobs b WHERE b.cid = $2\n ON CONFLICT (cid, created_by_user) DO NOTHING",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": []
},
"hash": "e248d71f595abf0207b01bc2f4e1f312d0c96b0f2f5131dfc13bfbb42a79d886"
}
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT storage_key as \"storage_key!\" FROM blobs WHERE created_by_user = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "storage_key!",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false
]
},
"hash": "f59010ecdd7f782489e0e03288a06dacd72b33d04c1e2b98475018ad25485852"
}
Generated
+22 -26
View File
@@ -7665,7 +7665,7 @@ dependencies = [
[[package]]
name = "tranquil-api"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"axum",
"backon",
@@ -7712,7 +7712,7 @@ dependencies = [
[[package]]
name = "tranquil-auth"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"anyhow",
"base32",
@@ -7737,7 +7737,7 @@ dependencies = [
[[package]]
name = "tranquil-cache"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"async-trait",
"base64 0.22.1",
@@ -7752,7 +7752,7 @@ dependencies = [
[[package]]
name = "tranquil-comms"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"async-trait",
"base64 0.22.1",
@@ -7772,22 +7772,19 @@ dependencies = [
"tranquil-config",
"tranquil-db-traits",
"tranquil-signal",
"tranquil-types",
"uuid",
]
[[package]]
name = "tranquil-config"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"confique",
"serde",
"tranquil-types",
]
[[package]]
name = "tranquil-crypto"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"aes-gcm",
"base64 0.22.1",
@@ -7803,7 +7800,7 @@ dependencies = [
[[package]]
name = "tranquil-db"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"async-trait",
"chrono",
@@ -7820,7 +7817,7 @@ dependencies = [
[[package]]
name = "tranquil-db-traits"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"async-trait",
"base64 0.22.1",
@@ -7836,7 +7833,7 @@ dependencies = [
[[package]]
name = "tranquil-infra"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"async-trait",
"bytes",
@@ -7849,7 +7846,7 @@ dependencies = [
[[package]]
name = "tranquil-lexicon"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"chrono",
"futures",
@@ -7869,7 +7866,7 @@ dependencies = [
[[package]]
name = "tranquil-oauth"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"anyhow",
"axum",
@@ -7893,7 +7890,7 @@ dependencies = [
[[package]]
name = "tranquil-oauth-server"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"async-trait",
"axum",
@@ -7929,7 +7926,7 @@ dependencies = [
[[package]]
name = "tranquil-pds"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"aes-gcm",
"anyhow",
@@ -8020,7 +8017,7 @@ dependencies = [
[[package]]
name = "tranquil-repo"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"bytes",
"cid",
@@ -8032,7 +8029,7 @@ dependencies = [
[[package]]
name = "tranquil-ripple"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"async-trait",
"backon",
@@ -8061,7 +8058,7 @@ dependencies = [
[[package]]
name = "tranquil-scopes"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"axum",
"futures",
@@ -8078,7 +8075,7 @@ dependencies = [
[[package]]
name = "tranquil-server"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"arc-swap",
"axum",
@@ -8115,7 +8112,7 @@ dependencies = [
[[package]]
name = "tranquil-signal"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"async-trait",
"chrono",
@@ -8130,14 +8127,13 @@ dependencies = [
"tokio",
"tokio-util",
"tracing",
"tranquil-types",
"url",
"uuid",
]
[[package]]
name = "tranquil-storage"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"async-trait",
"aws-config",
@@ -8154,7 +8150,7 @@ dependencies = [
[[package]]
name = "tranquil-store"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"async-trait",
"bytes",
@@ -8202,7 +8198,7 @@ dependencies = [
[[package]]
name = "tranquil-sync"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"anyhow",
"axum",
@@ -8224,7 +8220,7 @@ dependencies = [
[[package]]
name = "tranquil-types"
version = "0.6.7"
version = "0.6.6"
dependencies = [
"base64 0.22.1",
"chrono",
+1 -1
View File
@@ -26,7 +26,7 @@ members = [
]
[workspace.package]
version = "0.6.7"
version = "0.6.6"
edition = "2024"
license = "AGPL-3.0-or-later"
+1 -1
View File
@@ -1,4 +1,4 @@
ARG DISTROLESS_IMAGE=gcr.io/distroless/cc-debian13:latest@sha256:a017e74bd2a12d98342dbecd33d121d2b160415ed777573dc1808969e989d94d
ARG DISTROLESS_IMAGE=gcr.io/distroless/cc-debian13:latest@sha256:1e3c6d9c255be500eb680cdea0ad07554f52ae92dfcbdf07043a2a435b4c1fe3
FROM node:24-trixie-slim AS frontend
RUN corepack enable && corepack prepare pnpm@latest --activate
-2
View File
@@ -67,7 +67,6 @@ podman-compose -f docker-compose.prod.yaml up -d
### Installation Guides
- [Nix](docs/2_INSTALL_NIX.md)
- [Alpine](docs/2_INSTALL_ALPINE.md)
- [Containers](docs/2_INSTALL_CONTAINERS.md)
## Community
@@ -80,7 +79,6 @@ We currently don't have a shared space to chat and organize Tranquil things, but
- [@oyster.cafe](https://tangled.org/did:plc:3fwecdnvtcscjnrx2p4n7alz)
- [@nel.pet](https://tangled.org/did:plc:h5wsnqetncv6lu2weom35lg2)
- [@jola.dev](https://tangled.org/did:plc:bvraa6gajy4tfr3eh2sisdkr)
### Amazing contributors
@@ -38,8 +38,7 @@ pub async fn send_email(
.log_db_err("in send_email")?
.ok_or(ApiError::AccountNotFound)?;
let email = tranquil_types::EmailAddress::new(&user.email.ok_or(ApiError::NoEmail)?)
.map_err(|e| ApiError::InvalidRequest(e.to_string()))?;
let email = user.email.ok_or(ApiError::NoEmail)?;
let (user_id, handle) = (user.id, user.handle);
let hostname = &tranquil_config::get().server.hostname;
let subject = input
@@ -51,8 +50,9 @@ pub async fn send_email(
.infra
.enqueue_comms(
Some(user_id),
&tranquil_types::Recipient::Email(email),
tranquil_db_traits::CommsChannel::Email,
tranquil_db_traits::CommsType::AdminEmail,
&email,
Some(&subject),
content,
None,
@@ -19,10 +19,10 @@ pub async fn update_account_email(
Json(input): Json<UpdateAccountEmailInput>,
) -> Result<Json<EmptyResponse>, ApiError> {
let account = input.account.trim();
let email = tranquil_types::EmailAddress::new(&input.email)?;
if account.is_empty() {
let email = input.email.trim();
if account.is_empty() || email.is_empty() {
return Err(ApiError::InvalidRequest(
"Account is required, silly!".into(),
"account and email are required".into(),
));
}
let account_did: Did = account
@@ -32,7 +32,7 @@ pub async fn update_account_email(
match state
.repos
.user
.admin_update_email(&account_did, &email)
.admin_update_email(&account_did, email)
.await
{
Ok(0) => Err(ApiError::AccountNotFound),
@@ -66,11 +66,11 @@ pub async fn update_account_handle(
{
return Err(ApiError::InvalidHandle(None));
}
let primary = tranquil_pds::handle::ServiceDomains::for_user_handles().primary();
let handle = if input_handle.contains('.') {
input_handle.to_string()
let available_domains = tranquil_config::get().server.available_user_domain_list();
let handle = if !input_handle.contains('.') {
format!("{}.{}", input_handle, &available_domains[0])
} else {
format!("{}.{}", input_handle, primary)
input_handle.to_string()
};
let old_handle = state.repos.user.get_handle_by_did(did).await.ok().flatten();
let user_id = state
+42
View File
@@ -1,8 +1,10 @@
use axum::{Json, extract::State};
use serde::{Deserialize, Serialize};
use tracing::{error, warn};
use tranquil_pds::api::error::{ApiError, DbResultExt};
use tranquil_pds::auth::{Admin, Auth};
use tranquil_pds::state::AppState;
use tranquil_types::CidLink;
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
@@ -181,6 +183,46 @@ pub async fn update_server_config(
}
if let Some(ref logo_cid) = req.logo_cid {
let old_logo_cid = state
.repos
.infra
.get_server_config("logo_cid")
.await
.ok()
.flatten();
let should_delete_old = match (&old_logo_cid, logo_cid.is_empty()) {
(Some(old), true) => Some(old.clone()),
(Some(old), false) if old != logo_cid => Some(old.clone()),
_ => None,
};
if let Some(old_cid_str) = should_delete_old {
match CidLink::new(old_cid_str) {
Ok(old_cid) => {
if let Ok(Some(storage_key)) = state
.repos
.infra
.get_blob_storage_key_by_cid(&old_cid)
.await
{
if let Err(e) = state.blob_store.delete(&storage_key).await {
error!("Failed to delete old logo blob from storage: {:?}", e);
}
if let Err(e) = state.repos.infra.delete_blob_by_cid(&old_cid).await {
error!("Failed to delete old logo blob record: {:?}", e);
}
}
}
Err(e) => {
warn!(
"Old logo CID in database is invalid, skipping cleanup: {:?}",
e
);
}
}
}
if logo_cid.is_empty() {
state
.repos
+36 -65
View File
@@ -1,12 +1,11 @@
use bcrypt::{DEFAULT_COST, hash};
use chrono::{DateTime, Utc};
use std::collections::HashMap;
use tracing::{error, warn};
use tracing::error;
use tranquil_db_traits::{CommsChannel, DidWebOverrides, SessionRepository, UserRepository};
use tranquil_pds::api::error::ApiError;
use tranquil_pds::api::error::DbResultExt;
use tranquil_pds::types::{AtIdentifier, Did, Handle, PasswordHash};
use tranquil_types::{DiscordUsername, SignalUsername, TelegramUsername};
pub struct ResolvedRepo {
pub user_id: uuid::Uuid,
@@ -157,70 +156,42 @@ pub struct ChannelInput<'a> {
pub fn extract_verification_recipient(
channel: CommsChannel,
input: &ChannelInput<'_>,
) -> Result<tranquil_pds::comms::VerificationTarget, ApiError> {
) -> Result<String, ApiError> {
match channel {
CommsChannel::Email => {
let raw = trimmed(input.email).ok_or(ApiError::MissingEmail)?;
let address = tranquil_types::EmailAddress::new(raw)?;
Ok(tranquil_pds::comms::VerificationTarget::direct(
tranquil_db_traits::Recipient::Email(address),
))
}
CommsChannel::Signal => {
let raw = trimmed(input.signal_username).ok_or(ApiError::MissingSignalNumber)?;
let username = SignalUsername::new(raw)?;
Ok(tranquil_pds::comms::VerificationTarget::direct(
tranquil_db_traits::Recipient::Signal(username),
))
}
CommsChannel::Telegram => {
let raw = trimmed(input.telegram_username).ok_or(ApiError::MissingTelegramUsername)?;
let username = TelegramUsername::new(raw)?;
tranquil_pds::comms::VerificationTarget::resolve(
channel,
username.as_str(),
input.email,
)
}
CommsChannel::Discord => {
let raw = trimmed(input.discord_username).ok_or(ApiError::MissingDiscordId)?;
let username = DiscordUsername::new(raw)?;
tranquil_pds::comms::VerificationTarget::resolve(
channel,
username.as_str(),
input.email,
)
}
}
}
fn trimmed(raw: Option<&str>) -> Option<&str> {
raw.map(str::trim).filter(|value| !value.is_empty())
}
pub async fn implicitly_verify_channel(
user_repo: &dyn UserRepository,
did: &Did,
user_id: uuid::Uuid,
preferred_channel: CommsChannel,
context: &'static str,
) {
let Ok(Some(prefs)) = user_repo.get_comms_prefs(user_id).await else {
return;
};
let Some(recipient) = tranquil_pds::comms::recipient_for(&prefs, preferred_channel) else {
warn!(
did = %did,
preferred = ?preferred_channel,
"We skipped implicit verification on {context} because the account doesn't have a valid recipient"
);
return;
};
if let Err(e) = user_repo
.set_channel_verified(did, recipient.channel())
.await
{
warn!("Implicit verification on {context} failed: {:?}", e);
CommsChannel::Email => match input.email {
Some(e) if !e.trim().is_empty() => Ok(e.trim().to_string()),
_ => Err(ApiError::MissingEmail),
},
CommsChannel::Discord => match input.discord_username {
Some(username) if !username.trim().is_empty() => {
let clean = username.trim().to_lowercase();
if !tranquil_pds::api::validation::is_valid_discord_username(&clean) {
return Err(ApiError::InvalidRequest(
"Invalid Discord username. Must be 2-32 lowercase characters (letters, numbers, underscores, periods)".into(),
));
}
Ok(clean)
}
_ => Err(ApiError::MissingDiscordId),
},
CommsChannel::Telegram => match input.telegram_username {
Some(username) if !username.trim().is_empty() => {
let clean = username.trim().trim_start_matches('@');
if !tranquil_pds::api::validation::is_valid_telegram_username(clean) {
return Err(ApiError::InvalidRequest(
"Invalid Telegram username. Must be 5-32 characters, alphanumeric or underscore".into(),
));
}
Ok(clean.to_string())
}
_ => Err(ApiError::MissingTelegramUsername),
},
CommsChannel::Signal => match input.signal_username {
Some(username) if !username.trim().is_empty() => {
Ok(username.trim().trim_start_matches('@').to_lowercase())
}
_ => Err(ApiError::MissingSignalNumber),
},
}
}
+2 -2
View File
@@ -368,8 +368,8 @@ pub async fn create_delegated_account(
.as_ref()
.map(|e| e.trim().to_string())
.filter(|e| !e.is_empty());
if let Some(email) = &email
&& tranquil_types::EmailAddress::new(email).is_err()
if let Some(ref email) = email
&& !tranquil_pds::api::validation::is_valid_email(email)
{
return Err(ApiError::InvalidEmail);
}
+10 -19
View File
@@ -143,20 +143,7 @@ async fn handle_command(state: AppState, interaction: Interaction) -> Response {
None => {
return Json(json!({
"type": 4,
"data": {"content": "Couldn't identify user", "flags": 64}
}))
.into_response();
}
};
let (discord_user_id, discord_username) = match (
tranquil_types::DiscordUserId::new(&discord_user_id),
tranquil_types::DiscordUsername::new(&discord_username),
) {
(Ok(discord_user_id), Ok(discord_username)) => (discord_user_id, discord_username),
_ => {
return Json(json!({
"type": 4,
"data": {"content": "Couldn't verify your Discord account", "flags": 64}
"data": {"content": "Could not identify user", "flags": 64}
}))
.into_response();
}
@@ -197,14 +184,18 @@ async fn handle_command(state: AppState, interaction: Interaction) -> Response {
discord_user_id = %discord_user_id,
"Verified Discord user and stored user ID"
);
comms_repo::try_channel_verified_notice(
if let Err(e) = comms_repo::enqueue_channel_verified(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user_id,
&tranquil_types::Recipient::Discord(discord_user_id),
tranquil_db_traits::CommsChannel::Discord,
&discord_user_id,
&tranquil_config::get().server.hostname,
)
.await;
.await
{
warn!(error = %e, "Failed to enqueue channel verified notification");
}
Json(json!({
"type": 4,
"data": {"content": "Verified", "flags": 64}
@@ -308,11 +299,11 @@ mod tests {
fn parse_handle_whitespace_trimmed() {
let options = vec![InteractionOption {
name: "handle".to_string(),
value: serde_json::json!(" oystercafe.jola.dev "),
value: serde_json::json!(" alice.example.com "),
}];
assert_eq!(
parse_start_handle(Some(&options)),
Some("oystercafe.jola.dev".to_string()),
Some("alice.example.com".to_string()),
);
}
}
+4 -4
View File
@@ -51,7 +51,7 @@ async fn try_reactivate_migration(
handle: &Handle,
email: &Option<String>,
verification_channel: tranquil_db_traits::CommsChannel,
verification_recipient: Option<&tranquil_pds::comms::VerificationTarget>,
verification_recipient: Option<&str>,
) -> Option<Response> {
let reactivate_input = tranquil_db_traits::MigrationReactivationInput {
did: did.clone(),
@@ -271,8 +271,8 @@ pub async fn create_account(
.as_ref()
.map(|e| e.trim().to_string())
.filter(|e| !e.is_empty());
if let Some(email) = &email
&& tranquil_types::EmailAddress::new(email).is_err()
if let Some(ref email) = email
&& !tranquil_pds::api::validation::is_valid_email(email)
{
return ApiError::InvalidEmail.into_response();
}
@@ -393,7 +393,7 @@ pub async fn create_account(
&handle,
&email,
verification_channel,
verification_recipient.as_ref(),
verification_recipient.as_deref(),
)
.await
{
+26 -13
View File
@@ -132,9 +132,12 @@ pub async fn well_known_did(State(state): State<AppState>, headers: HeaderMap) -
let host_header = get_header_str(&headers, http::header::HOST).unwrap_or(hostname);
let host_without_port = host_header.split(':').next().unwrap_or(host_header);
if host_without_port != hostname_without_port {
let is_subdomain = tranquil_pds::handle::ServiceDomains::served()
.split_handle(host_without_port)
.is_some();
let is_subdomain = cfg
.server
.available_user_domain_list()
.into_iter()
.chain(std::iter::once(hostname_without_port.to_string()))
.any(|d| host_without_port.ends_with(&format!(".{}", d)));
if is_subdomain {
return serve_handle_did_doc(&state, host_without_port, hostname).await;
}
@@ -579,16 +582,26 @@ pub async fn update_handle(
"Inappropriate language in handle".into(),
)));
}
let handle_domains = tranquil_pds::handle::ServiceDomains::for_user_handles();
let split = handle_domains.split_handle(&new_handle);
let is_domain_itself = handle_domains.contains(&new_handle);
let handle: Handle = if (!new_handle.contains('.') || split.is_some()) && !is_domain_itself {
let (short_part, full_handle) = match split {
Some((_domain, short)) => (short.to_string(), new_handle.clone()),
None => (
new_handle.clone(),
format!("{}.{}", new_handle, handle_domains.primary()),
),
let handle_domains = tranquil_config::get().server.user_handle_domain_list();
let matched_handle_domain = handle_domains
.iter()
.filter(|d| new_handle.ends_with(&format!(".{}", d)))
.max_by_key(|d| d.len())
.cloned();
let is_domain_itself = handle_domains.iter().any(|d| d == &new_handle);
let handle: Handle = if (!new_handle.contains('.') || matched_handle_domain.is_some())
&& !is_domain_itself
{
let (short_part, full_handle) = match &matched_handle_domain {
Some(domain) => {
let suffix = format!(".{}", domain);
let short = new_handle.strip_suffix(&suffix).unwrap_or(&new_handle);
(short.to_string(), new_handle.clone())
}
None => {
let primary = &handle_domains[0];
(new_handle.clone(), format!("{}.{}", new_handle, primary))
}
};
if full_handle == current_handle {
let handle: Handle = match full_handle.parse() {
@@ -35,27 +35,16 @@ pub async fn request_plc_operation_signature(
.log_db_err("creating PLC token")?;
let hostname = &tranquil_config::get().server.hostname;
match tranquil_pds::comms::comms_repo::enqueue_notice(
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_plc_operation(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user_id,
tranquil_pds::comms::Notice::PlcOperation {
token: &display_token,
},
&display_token,
hostname,
)
.await
{
Ok(Some(_)) => {}
Ok(None) => {
return Err(ApiError::InvalidRequest(
"We couldn't deliver the PLC operation code to your notification channels. Please contact the PDS owner."
.into(),
));
}
Err(e) => {
warn!("Failed to enqueue PLC operation notification: {:?}", e);
}
warn!("Failed to enqueue PLC operation notification: {:?}", e);
}
info!("PLC operation signature requested for user {}", auth.did);
Ok(Json(EmptyResponse {}))
+4 -1
View File
@@ -9,7 +9,10 @@ use tranquil_pds::api::ApiError;
use tranquil_pds::api::error::DbResultExt;
use tranquil_pds::auth::{Auth, Permissive};
use tranquil_pds::circuit_breaker::with_circuit_breaker;
use tranquil_pds::plc::{PlcError, PlcService, create_update_op, sign_operation};
use tranquil_pds::plc::{
PlcError, PlcService, create_update_op, missing_required_rotation_key, sign_operation,
signing_key_to_did_key,
};
use tranquil_pds::state::AppState;
#[derive(Debug, Deserialize)]
@@ -315,17 +315,18 @@ pub async fn enqueue_signup_verification(
user_id: uuid::Uuid,
did: &Did,
channel: CommsChannel,
target: &tranquil_pds::comms::VerificationTarget,
recipient: &str,
) {
let token =
tranquil_pds::auth::verification_token::generate_signup_token(did, channel, &target.id);
tranquil_pds::auth::verification_token::generate_signup_token(did, channel, recipient);
let formatted = tranquil_pds::auth::verification_token::format_token_for_display(&token);
let hostname = &tranquil_config::get().server.hostname;
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_signup_verification(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user_id,
target,
channel,
recipient,
&formatted,
hostname,
)
@@ -340,17 +341,18 @@ pub async fn enqueue_migration_verification(
user_id: uuid::Uuid,
did: &Did,
channel: CommsChannel,
target: &tranquil_pds::comms::VerificationTarget,
recipient: &str,
) {
let token =
tranquil_pds::auth::verification_token::generate_migration_token(did, channel, &target.id);
tranquil_pds::auth::verification_token::generate_migration_token(did, channel, recipient);
let formatted = tranquil_pds::auth::verification_token::format_token_for_display(&token);
let hostname = &tranquil_config::get().server.hostname;
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_migration_verification(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user_id,
target,
channel,
recipient,
&formatted,
hostname,
)
+2 -8
View File
@@ -467,15 +467,9 @@ pub fn api_routes() -> axum::Router<AppState> {
pub fn well_known_api_routes() -> axum::Router<AppState> {
use axum::routing::get;
let routes = axum::Router::new()
axum::Router::new()
.route("/did.json", get(identity::well_known_did))
.route("/atproto-did", get(identity::well_known_atproto_did));
if tranquil_config::get().server.enable_caddy_on_demand_tls {
routes.route("/caddy/ask", get(server::caddy_ask))
} else {
routes
}
.route("/atproto-did", get(identity::well_known_atproto_did))
}
pub fn webhook_routes() -> axum::Router<AppState> {
+59 -65
View File
@@ -19,7 +19,6 @@ pub struct NotificationPrefsOutput {
pub telegram_verified: bool,
pub signal_username: Option<String>,
pub signal_verified: bool,
pub legacy_login_alerts: bool,
}
pub async fn get_notification_prefs(
@@ -33,26 +32,6 @@ pub async fn get_notification_prefs(
.await
.log_db_err("get notification prefs")?
.ok_or(ApiError::AccountNotFound)?;
let user_id = state
.repos
.user
.get_id_by_did(&auth.did)
.await
.log_db_err("get user by did")?
.ok_or(ApiError::AccountNotFound)?;
let legacy_login_alerts = state
.repos
.infra
.get_account_preferences(user_id)
.await
.log_db_err("get legacy login alert prefs")?
.iter()
.find(|(name, _)| name == "legacy_login_alerts")
.and_then(|(_, value)| value.as_bool())
.unwrap_or(true);
Ok(Json(NotificationPrefsOutput {
preferred_channel: prefs.preferred_channel,
email: prefs.email,
@@ -62,7 +41,6 @@ pub async fn get_notification_prefs(
telegram_verified: prefs.telegram_verified,
signal_username: prefs.signal_username,
signal_verified: prefs.signal_verified,
legacy_login_alerts,
}))
}
@@ -143,7 +121,6 @@ pub struct UpdateNotificationPrefsInput {
pub discord_username: Option<String>,
pub telegram_username: Option<String>,
pub signal_username: Option<String>,
pub legacy_login_alerts: Option<bool>,
}
#[derive(Serialize)]
@@ -159,16 +136,12 @@ pub async fn request_channel_verification(
user_id: uuid::Uuid,
did: &Did,
channel: CommsChannel,
id: &str,
identifier: &str,
handle: Option<&Handle>,
) -> Result<String, ApiError> {
if channel.verifies_via_bot() {
return Err(ApiError::InvalidRequest(
"This channel verifies through a bot. Please message the bot first so that it's able to reply with useful info".into(),
));
}
let token =
tranquil_pds::auth::verification_token::generate_channel_update_token(did, channel, id);
let token = tranquil_pds::auth::verification_token::generate_channel_update_token(
did, channel, identifier,
);
let formatted_token = tranquil_pds::auth::verification_token::format_token_for_display(&token);
match channel {
@@ -177,11 +150,10 @@ pub async fn request_channel_verification(
let handle = handle.ok_or_else(|| {
ApiError::InternalError(Some("Email verification requires a handle".into()))
})?;
let new_email = tranquil_types::EmailAddress::new(id)?;
tranquil_pds::comms::comms_repo::enqueue_email_update(
state.repos.infra.as_ref(),
user_id,
&new_email,
identifier,
handle,
&formatted_token,
hostname,
@@ -192,10 +164,10 @@ pub async fn request_channel_verification(
_ => {
let hostname = &tranquil_config::get().server.hostname;
let encoded_token = urlencoding::encode(&formatted_token);
let encoded_id = urlencoding::encode(id);
let encoded_identifier = urlencoding::encode(identifier);
let verify_link = format!(
"https://{}/app/verify?token={}&identifier={}",
hostname, encoded_token, encoded_id
hostname, encoded_token, encoded_identifier
);
let prefs = state
.repos
@@ -217,14 +189,26 @@ pub async fn request_channel_verification(
strings.channel_verification_subject,
&[("hostname", hostname)],
);
let recipient = tranquil_db_traits::Recipient::new(channel, id)?;
let recipient = match channel {
CommsChannel::Telegram => state
.repos
.user
.get_telegram_chat_id(user_id)
.await
.ok()
.flatten()
.map(|id| id.to_string())
.unwrap_or_else(|| identifier.to_string()),
_ => identifier.to_string(),
};
state
.repos
.infra
.enqueue_comms(
Some(user_id),
&recipient,
channel,
tranquil_db_traits::CommsType::ChannelVerification,
&recipient,
Some(&subject),
&body,
Some(json!({"code": formatted_token})),
@@ -246,7 +230,14 @@ async fn process_messaging_channel_update(
effective_channel: CommsChannel,
verification_required: &mut Vec<CommsChannel>,
) -> Result<(), ApiError> {
if raw_value.trim().is_empty() {
let clean = match channel {
CommsChannel::Discord => raw_value.trim().to_lowercase(),
CommsChannel::Telegram => raw_value.trim_start_matches('@').to_string(),
CommsChannel::Signal => raw_value.trim().trim_start_matches('@').to_lowercase(),
CommsChannel::Email => raw_value.trim().to_lowercase(),
};
if clean.is_empty() {
if effective_channel == channel {
return Err(ApiError::InvalidRequest(format!(
"Cannot remove {:?} while it is the preferred notification channel",
@@ -278,12 +269,26 @@ async fn process_messaging_channel_update(
return Ok(());
}
let clean = match channel {
CommsChannel::Discord => tranquil_types::DiscordUsername::new(raw_value)?.to_string(),
CommsChannel::Telegram => tranquil_types::TelegramUsername::new(raw_value)?.to_string(),
CommsChannel::Signal => tranquil_types::SignalUsername::new(raw_value)?.to_string(),
CommsChannel::Email => tranquil_types::EmailAddress::new(raw_value)?.to_string(),
let valid = match channel {
CommsChannel::Discord => tranquil_pds::api::validation::is_valid_discord_username(&clean),
CommsChannel::Telegram => tranquil_pds::api::validation::is_valid_telegram_username(&clean),
CommsChannel::Signal => tranquil_pds::comms::is_valid_signal_username(&clean),
CommsChannel::Email => tranquil_pds::api::validation::is_valid_email(&clean),
};
if !valid {
return Err(match channel {
CommsChannel::Discord => ApiError::InvalidRequest(
"Invalid Discord username. Must be 2-32 lowercase characters (letters, numbers, underscores, periods)".into(),
),
CommsChannel::Telegram => ApiError::InvalidRequest(
"Invalid Telegram username. Must be 5-32 characters, alphanumeric or underscore".into(),
),
CommsChannel::Signal => ApiError::InvalidRequest(
"Invalid Signal username. Must be a 3-32 character nickname, a dot, then a 2-20 digit discriminator".into(),
),
CommsChannel::Email => ApiError::InvalidEmail,
});
}
match channel {
CommsChannel::Discord => state
@@ -366,25 +371,23 @@ pub async fn update_notification_prefs(
info!(did = %auth.did, channel = ?effective_channel, "Updated preferred notification channel");
}
if let Some(new_email) = &input.email {
let email = tranquil_types::EmailAddress::new(new_email).map_err(|_| {
if new_email.trim().is_empty() {
ApiError::InvalidRequest("Email can't be empty".into())
} else {
ApiError::InvalidEmail
}
})?;
if let Some(ref new_email) = input.email {
let email_clean = new_email.trim().to_lowercase();
if email_clean.is_empty() {
return Err(ApiError::InvalidRequest("Email cannot be empty".into()));
}
if !current_email
.as_deref()
.is_some_and(|e| e.eq_ignore_ascii_case(email.as_str()))
{
if !tranquil_pds::api::validation::is_valid_email(&email_clean) {
return Err(ApiError::InvalidEmail);
}
if current_email.as_ref().map(|e| e.to_lowercase()) != Some(email_clean.clone()) {
request_channel_verification(
&state,
user_id,
&auth.did,
CommsChannel::Email,
email.as_str(),
&email_clean,
Some(&handle),
)
.await?;
@@ -432,15 +435,6 @@ pub async fn update_notification_prefs(
.await?;
}
if let Some(alerts) = input.legacy_login_alerts {
state
.repos
.infra
.upsert_account_preference(user_id, "legacy_login_alerts", json!(alerts))
.await
.log_db_err("update legacy login alert prefs")?;
}
Ok(Json(UpdateNotificationPrefsOutput {
success: true,
verification_required,
+18 -9
View File
@@ -148,13 +148,7 @@ pub async fn upload_blob(
size, cid_str
);
if let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await {
let _ = state.blob_store.delete(&temp_key).await;
error!("Failed to copy blob to final location: {:?}", e);
return Err(ApiError::InternalError(Some("Failed to store blob".into())));
}
if let Err(e) = state
match state
.repos
.blob
.insert_blob(
@@ -166,9 +160,24 @@ pub async fn upload_blob(
)
.await
{
Ok(_) => {}
Err(e) => {
let _ = state.blob_store.delete(&temp_key).await;
error!("Failed to insert blob record: {:?}", e);
return Err(ApiError::InternalError(None));
}
};
if let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await {
let _ = state.blob_store.delete(&temp_key).await;
error!("Failed to insert blob record: {:?}", e);
return Err(ApiError::InternalError(None));
if let Err(db_err) = state.repos.blob.delete_blob_by_cid(&cid_link).await {
error!(
"Failed to clean up orphaned blob record after copy failure: {:?}",
db_err
);
}
error!("Failed to copy blob to final location: {:?}", e);
return Err(ApiError::InternalError(Some("Failed to store blob".into())));
}
let _ = state.blob_store.delete(&temp_key).await;
@@ -586,27 +586,16 @@ pub async fn request_account_delete(
.await
.log_db_err("creating deletion token")?;
let hostname = &tranquil_config::get().server.hostname;
match tranquil_pds::comms::comms_repo::enqueue_notice(
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_account_deletion(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user_id,
tranquil_pds::comms::Notice::AccountDeletion {
code: &confirmation_token,
},
&confirmation_token,
hostname,
)
.await
{
Ok(Some(_)) => {}
Ok(None) => {
return Err(ApiError::InvalidRequest(
"We couldn't deliver the deletion code to your notification channels. Please contact the PDS owner."
.into(),
));
}
Err(e) => {
warn!("Failed to enqueue account deletion notification: {:?}", e);
}
warn!("Failed to enqueue account deletion notification: {:?}", e);
}
info!("Account deletion requested for user {}", session_mfa.did());
Ok(Json(EmptyResponse {}))
-43
View File
@@ -1,43 +0,0 @@
use axum::extract::{Query, State};
use axum::http::StatusCode;
use serde::de::Error as _;
use serde::{Deserialize, Deserializer};
use tracing::error;
use tranquil_pds::handle::ServiceDomains;
use tranquil_pds::state::AppState;
use tranquil_pds::types::Handle;
pub struct AskedDomain(Handle);
impl<'de> Deserialize<'de> for AskedDomain {
fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
let raw = String::deserialize(deserializer)?;
let without_root_dot = raw.strip_suffix('.').unwrap_or(&raw);
Handle::new(without_root_dot)
.map(Self)
.map_err(D::Error::custom)
}
}
#[derive(Deserialize)]
pub struct CaddyAskQuery {
pub domain: AskedDomain,
}
pub async fn caddy_ask(
State(state): State<AppState>,
Query(ask): Query<CaddyAskQuery>,
) -> StatusCode {
let AskedDomain(handle) = ask.domain;
if ServiceDomains::served().contains(handle.as_str()) {
return StatusCode::OK;
}
match state.repos.user.get_by_handle(&handle).await {
Ok(Some(_)) => StatusCode::OK,
Ok(None) => StatusCode::NOT_FOUND,
Err(e) => {
error!("caddy ask couldn't look up handle {handle}: {e:?}");
StatusCode::INTERNAL_SERVER_ERROR
}
}
}
+43 -54
View File
@@ -71,7 +71,7 @@ pub async fn request_email_update(
let Some(_current_email) = user.email else {
return Err(ApiError::InvalidRequest(
"Account doesn't have an email address".into(),
"account does not have an email address".into(),
));
};
@@ -89,43 +89,36 @@ pub async fn request_email_update(
ApiError::InternalError(Some("Failed to generate verification code".into()))
})?;
if let Some(Json(inp)) = &input
&& let Some(new_email) = inp.new_email.as_deref()
&& let Ok(address) = tranquil_types::EmailAddress::new(new_email)
if let Some(Json(ref inp)) = input
&& let Some(ref new_email) = inp.new_email
{
let pending = PendingEmailUpdate {
new_email: address.as_str().to_string(),
token_hash: hash_token(&token),
authorized: false,
};
let cache_key = tranquil_pds::cache_keys::email_update_key(&auth.did);
if let Ok(json) = serde_json::to_string(&pending)
&& let Err(e) = state.cache.set(&cache_key, &json, EMAIL_UPDATE_TTL).await
{
warn!("Failed to cache pending email update: {:?}", e);
let new_email = new_email.trim().to_lowercase();
if !new_email.is_empty() && tranquil_pds::api::validation::is_valid_email(&new_email) {
let pending = PendingEmailUpdate {
new_email,
token_hash: hash_token(&token),
authorized: false,
};
if let Ok(json) = serde_json::to_string(&pending) {
let cache_key = tranquil_pds::cache_keys::email_update_key(&auth.did);
if let Err(e) = state.cache.set(&cache_key, &json, EMAIL_UPDATE_TTL).await {
warn!("Failed to cache pending email update: {:?}", e);
}
}
}
}
let hostname = &tranquil_config::get().server.hostname;
match tranquil_pds::comms::comms_repo::enqueue_notice(
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_short_token_email(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user.id,
tranquil_pds::comms::Notice::ShortTokenEmail { token: &token },
&token,
hostname,
)
.await
{
Ok(Some(_)) => {}
Ok(None) => {
return Err(ApiError::InvalidRequest(
"We couldn't deliver the confirmation code to your notification channels. Please contact the PDS owner."
.into(),
));
}
Err(e) => {
warn!("Failed to enqueue email update notification: {:?}", e);
}
warn!("Failed to enqueue email update notification: {:?}", e);
}
}
@@ -157,11 +150,13 @@ pub async fn confirm_email(
.log_db_err("getting email info")?
.ok_or(ApiError::AccountNotFound)?;
let Some(email) = &user.email else {
let Some(ref email) = user.email else {
return Err(ApiError::InvalidEmail);
};
let provided_email = tranquil_types::EmailAddress::new(input.email.trim())?;
if provided_email.as_str() != email.to_lowercase() {
let current_email = email.to_lowercase();
let provided_email = input.email.trim().to_lowercase();
if provided_email != current_email {
return Err(ApiError::InvalidEmail);
}
@@ -175,7 +170,7 @@ pub async fn confirm_email(
let verified = tranquil_pds::auth::verification_token::verify_signup_token(
&confirmation_code,
CommsChannel::Email,
provided_email.as_str(),
&provided_email,
);
match verified {
@@ -231,14 +226,17 @@ pub async fn update_email(
let user_id = user.id;
let current_email = user.email.clone();
let email_verified = user.email_verified;
let new_email = tranquil_types::EmailAddress::new(input.email.trim()).map_err(|_| {
ApiError::InvalidRequest(
let new_email = input.email.trim().to_lowercase();
if !tranquil_pds::api::validation::is_valid_email(&new_email) {
return Err(ApiError::InvalidRequest(
"This email address is not supported, please use a different email.".into(),
)
})?;
));
}
let email_unchanged = current_email
.as_ref()
.map(|c| new_email.as_str() == c.to_lowercase())
.map(|c| new_email == c.to_lowercase())
.unwrap_or(false);
if email_unchanged {
@@ -285,7 +283,7 @@ pub async fn update_email(
if let Some(pending_json) = state.cache.get(&cache_key).await
&& let Ok(pending) = serde_json::from_str::<PendingEmailUpdate>(&pending_json)
&& pending.authorized
&& pending.new_email == new_email.as_str()
&& pending.new_email == new_email
{
authorized_via_link = true;
let _ = state.cache.delete(&cache_key).await;
@@ -352,26 +350,24 @@ pub async fn update_email(
state
.repos
.user
.update_email(user_id, new_email.as_str())
.update_email(user_id, &new_email)
.await
.log_db_err("updating email")?;
let verification_token = tranquil_pds::auth::verification_token::generate_signup_token(
did,
CommsChannel::Email,
new_email.as_str(),
&new_email,
);
let formatted_token =
tranquil_pds::auth::verification_token::format_token_for_display(&verification_token);
let hostname = &tranquil_config::get().server.hostname;
let target = tranquil_pds::comms::VerificationTarget::direct(
tranquil_db_traits::Recipient::Email(new_email.clone()),
);
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_signup_verification(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user_id,
&target,
tranquil_db_traits::CommsChannel::Email,
&new_email,
&formatted_token,
hostname,
)
@@ -569,22 +565,15 @@ pub async fn check_email_in_use(
_rate_limit: RateLimited<VerificationCheckLimit>,
Json(input): Json<CheckEmailInUseInput>,
) -> Result<Json<InUseOutput>, ApiError> {
let raw = input.email.trim();
let email = tranquil_types::EmailAddress::new(raw).map_err(|_| {
ApiError::InvalidRequest(
if raw.is_empty() {
"Email is required"
} else {
"Invalid email address"
}
.into(),
)
})?;
let email = input.email.trim().to_lowercase();
if email.is_empty() {
return Err(ApiError::InvalidRequest("email is required".into()));
}
let count = state
.repos
.user
.count_accounts_by_email(email.as_str())
.count_accounts_by_email(&email)
.await
.map_err(|e| {
error!("DB error checking email usage: {:?}", e);
+1 -6
View File
@@ -77,12 +77,7 @@ pub async fn describe_server(State(state): State<AppState>) -> Json<DescribeServ
let pds_hostname = &cfg.server.hostname;
Json(DescribeServerOutput {
available_user_domains: match cfg.server.user_handle_domains.as_deref() {
Some(domains) if !domains.is_empty() => {
domains.iter().map(|d| d.as_str().to_owned()).collect()
}
_ => vec![cfg.server.hostname_without_port().to_owned()],
},
available_user_domains: cfg.server.user_handle_domain_list(),
invite_code_required: cfg.server.invite_code_required,
did: format!("did:web:{}", pds_hostname),
links: DescribeServerLinks {
-2
View File
@@ -1,6 +1,5 @@
pub mod account_status;
pub mod app_password;
pub mod caddy;
pub mod email;
pub mod invite;
pub mod logo;
@@ -23,7 +22,6 @@ pub use account_status::{
request_account_delete,
};
pub use app_password::{create_app_password, list_app_passwords, revoke_app_password};
pub use caddy::caddy_ask;
pub use email::{
authorize_email_update, check_channel_verified, check_email_in_use, check_email_update_status,
check_email_verified, confirm_email, request_email_update, update_email,
@@ -116,8 +116,8 @@ pub async fn create_passkey_account(
.as_ref()
.map(|e| e.trim().to_string())
.filter(|e| !e.is_empty());
if let Some(email) = &email
&& tranquil_types::EmailAddress::new(email).is_err()
if let Some(ref email) = email
&& !tranquil_pds::api::validation::is_valid_email(email)
{
return Err(ApiError::InvalidEmail);
}
@@ -703,11 +703,11 @@ pub async fn request_passkey_recovery(
urlencoding::encode(&recovery_token)
);
let _ = tranquil_pds::comms::comms_repo::enqueue_notice(
let _ = tranquil_pds::comms::comms_repo::enqueue_passkey_recovery(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user.id,
tranquil_pds::comms::Notice::PasskeyRecovery { url: &recovery_url },
&recovery_url,
hostname,
)
.await;
@@ -776,14 +776,21 @@ pub async fn recover_passkey_account(
if result.passkeys_deleted > 0 {
info!(did = %input.did, count = result.passkeys_deleted, "Deleted lost passkeys during account recovery");
}
crate::common::implicitly_verify_channel(
state.repos.user.as_ref(),
&input.did,
user.id,
user.preferred_comms_channel,
"passkey recovery",
)
.await;
if let Ok(Some(prefs)) = state.repos.user.get_comms_prefs(user.id).await {
let actual_channel =
tranquil_pds::comms::resolve_delivery_channel(&prefs, user.preferred_comms_channel);
if let Err(e) = state
.repos
.user
.set_channel_verified(&input.did, actual_channel)
.await
{
warn!(
"Failed to implicitly verify channel on passkey recovery: {:?}",
e
);
}
}
info!(did = %input.did, "Passkey-only account recovered with temporary password");
Ok(Json(SuccessResponse { success: true }))
}
+17 -12
View File
@@ -90,13 +90,11 @@ pub async fn request_password_reset(
return Err(ApiError::InternalError(None));
}
let hostname = &tranquil_config::get().server.hostname;
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_notice(
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_password_reset(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user_id,
tranquil_pds::comms::Notice::PasswordReset {
code: &display_code,
},
&display_code,
hostname,
)
.await
@@ -195,14 +193,21 @@ pub async fn reset_password(
}
}))
.await;
crate::common::implicitly_verify_channel(
state.repos.user.as_ref(),
&user.did,
user_id,
user.preferred_comms_channel,
"password reset",
)
.await;
if let Ok(Some(prefs)) = state.repos.user.get_comms_prefs(user_id).await {
let actual_channel =
tranquil_pds::comms::resolve_delivery_channel(&prefs, user.preferred_comms_channel);
if let Err(e) = state
.repos
.user
.set_channel_verified(&user.did, actual_channel)
.await
{
warn!(
"Failed to implicitly verify channel on password reset: {:?}",
e
);
}
}
info!("Password reset completed for user {}", user_id);
Ok(Json(EmptyResponse {}))
}
+86 -101
View File
@@ -220,34 +220,21 @@ pub async fn create_session(
}
Ok(tranquil_pds::auth::legacy_2fa::Legacy2faOutcome::ChallengeSent(code)) => {
let hostname = &tranquil_config::get().server.hostname;
match tranquil_pds::comms::comms_repo::enqueue_notice(
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_2fa_code(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
row.id,
tranquil_pds::comms::Notice::TwoFactorCode {
code: code.as_str(),
},
code.as_str(),
hostname,
)
.await
{
Ok(Some(_)) => {}
Ok(None) => {
tranquil_pds::auth::legacy_2fa::clear_challenge(state.cache.as_ref(), &row.did)
.await;
return Err(ApiError::InvalidRequest(
"We couldn't deliver the verification code to your notification channels. Please contact the PDS owner."
.into(),
));
}
Err(e) => {
error!("Failed to send 2FA code: {:?}", e);
tranquil_pds::auth::legacy_2fa::clear_challenge(state.cache.as_ref(), &row.did)
.await;
return Err(ApiError::InternalError(Some(
"Failed to send verification code. Please try again.".into(),
)));
}
error!("Failed to send 2FA code: {:?}", e);
tranquil_pds::auth::legacy_2fa::clear_challenge(state.cache.as_ref(), &row.did)
.await;
return Err(ApiError::InternalError(Some(
"Failed to send verification code. Please try again.".into(),
)));
}
return Err(ApiError::AuthFactorTokenRequired);
}
@@ -330,39 +317,23 @@ pub async fn create_session(
return Err(ApiError::InternalError(None));
}
if is_legacy_login && !used_totp_factor {
let alerts_enabled = match state.repos.infra.get_account_preferences(row.id).await {
Ok(prefs) => prefs
.iter()
.find(|(name, _)| name == "legacy_login_alerts")
.and_then(|(_, value)| value.as_bool())
.unwrap_or(true),
Err(e) => {
warn!("Failed to fetch legacy login alert preference: {:?}", e);
true
}
};
if alerts_enabled {
warn!(
did = %row.did,
ip = %client_ip,
"Legacy login on TOTP-enabled account - sending notification"
);
let hostname = &tranquil_config::get().server.hostname;
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_notice(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
row.id,
tranquil_pds::comms::Notice::LegacyLoginAlert {
channel: row.preferred_comms_channel,
ip: client_ip,
},
hostname,
)
.await
{
error!("Failed to queue legacy login notification: {:?}", e);
}
warn!(
did = %row.did,
ip = %client_ip,
"Legacy login on TOTP-enabled account - sending notification"
);
let hostname = &tranquil_config::get().server.hostname;
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_legacy_login(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
row.id,
hostname,
client_ip,
row.preferred_comms_channel,
)
.await
{
error!("Failed to queue legacy login notification: {:?}", e);
}
}
let handle = row.handle.clone();
@@ -883,13 +854,15 @@ pub async fn confirm_signup(
}
};
let Some(id) = row.channel_identifier() else {
warn!(
did = %input.did,
channel = ?row.channel,
"We can't confirm signup because the account doesn't have an identifier on file"
);
return Err(ApiError::InvalidRequest("Invalid verification code".into()));
let identifier = match row.channel {
tranquil_db_traits::CommsChannel::Email => row.email.clone().unwrap_or_default(),
tranquil_db_traits::CommsChannel::Discord => {
row.discord_username.clone().unwrap_or_default()
}
tranquil_db_traits::CommsChannel::Telegram => {
row.telegram_username.clone().unwrap_or_default()
}
tranquil_db_traits::CommsChannel::Signal => row.signal_username.clone().unwrap_or_default(),
};
let normalized_token =
@@ -897,7 +870,7 @@ pub async fn confirm_signup(
match tranquil_pds::auth::verification_token::verify_signup_token(
&normalized_token,
row.channel,
id,
&identifier,
) {
Ok(token_data) => {
if token_data.did != input.did {
@@ -953,11 +926,10 @@ pub async fn confirm_signup(
};
let hostname = &tranquil_config::get().server.hostname;
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_notice(
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_welcome(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
row.id,
tranquil_pds::comms::Notice::Welcome,
hostname,
)
.await
@@ -976,35 +948,6 @@ pub async fn confirm_signup(
}))
}
async fn resend_signup_verification(
state: &AppState,
row: &tranquil_db_traits::UserResendVerification,
did: &Did,
context: &'static str,
) -> bool {
let Some(id) = row.channel_identifier() else {
warn!(did = %did, channel = ?row.channel, "We skipped {context} because the account doesn't have a recipient on file");
return false;
};
match tranquil_pds::comms::VerificationTarget::resolve(row.channel, id, row.email.as_deref()) {
Ok(target) => {
crate::identity::provision::enqueue_signup_verification(
state,
row.id,
did,
row.channel,
&target,
)
.await;
true
}
Err(_) => {
warn!(did = %did, channel = ?row.channel, "We skipped {context} because the account doesn't have a valid recipient");
false
}
}
}
const AUTO_VERIFY_DEBOUNCE: std::time::Duration = std::time::Duration::from_secs(120);
pub struct AutoResendResult {
@@ -1033,15 +976,37 @@ pub async fn auto_resend_verification(state: &AppState, did: &Did) -> Option<Aut
handle: row.handle.clone(),
channel: row.channel,
};
if row.channel.verifies_via_bot() || debounced {
let is_bot_channel = matches!(
row.channel,
tranquil_db_traits::CommsChannel::Telegram | tranquil_db_traits::CommsChannel::Discord
);
if is_bot_channel || debounced {
return Some(result);
}
if resend_signup_verification(state, &row, did, "auto-resend verification").await {
let _ = state
.cache
.set(&debounce_key, "1", AUTO_VERIFY_DEBOUNCE)
.await;
let recipient = match row.channel {
tranquil_db_traits::CommsChannel::Email => row.email.clone().unwrap_or_default(),
tranquil_db_traits::CommsChannel::Signal => row.signal_username.clone().unwrap_or_default(),
_ => return Some(result),
};
if recipient.is_empty() {
warn!(
"No recipient configured for auto-resend verification: {}",
did
);
return Some(result);
}
crate::identity::provision::enqueue_signup_verification(
state,
row.id,
did,
row.channel,
&recipient,
)
.await;
let _ = state
.cache
.set(&debounce_key, "1", AUTO_VERIFY_DEBOUNCE)
.await;
Some(result)
}
@@ -1071,12 +1036,32 @@ pub async fn resend_verification(
return Err(ApiError::InternalError(None));
}
};
if row.channel_verification.has_any_verified() {
let is_verified = row.channel_verification.has_any_verified();
if is_verified {
return Err(ApiError::InvalidRequest(
"Account is already verified".into(),
));
}
resend_signup_verification(&state, &row, &input.did, "resend verification").await;
let recipient = match row.channel {
tranquil_db_traits::CommsChannel::Email => row.email.clone().unwrap_or_default(),
tranquil_db_traits::CommsChannel::Discord => {
row.discord_username.clone().unwrap_or_default()
}
tranquil_db_traits::CommsChannel::Telegram => {
row.telegram_username.clone().unwrap_or_default()
}
tranquil_db_traits::CommsChannel::Signal => row.signal_username.clone().unwrap_or_default(),
};
crate::identity::provision::enqueue_signup_verification(
&state,
row.id,
&input.did,
row.channel,
&recipient,
)
.await;
Ok(Json(SuccessResponse { success: true }))
}
+6 -4
View File
@@ -6,8 +6,8 @@ use tranquil_pds::api::error::{ApiError, DbResultExt};
use tranquil_pds::auth::{
Active, Auth, decrypt_totp_secret, encrypt_totp_secret, generate_backup_codes,
generate_qr_png_base64, generate_totp_secret, generate_totp_uri, hash_backup_code,
is_backup_code_format, verify_backup_code, verify_password_mfa, verify_totp_code,
verify_totp_mfa,
is_backup_code_format, require_legacy_session_mfa, verify_backup_code, verify_password_mfa,
verify_totp_code, verify_totp_mfa,
};
use tranquil_pds::rate_limit::{TotpVerifyLimit, check_user_rate_limit_with_message};
use tranquil_pds::state::AppState;
@@ -163,9 +163,11 @@ pub async fn disable_totp(
auth: Auth<Active>,
Json(input): Json<DisableTotpInput>,
) -> Result<Json<EmptyResponse>, ApiError> {
let session_mfa = require_legacy_session_mfa(&state, &auth).await?;
let _rate_limit = check_user_rate_limit_with_message::<TotpVerifyLimit>(
&state,
auth.did.as_str(),
session_mfa.did(),
"Too many verification attempts. Please try again in a few minutes.",
)
.await?;
@@ -182,7 +184,7 @@ pub async fn disable_totp(
tranquil_pds::auth::legacy_2fa::clear_challenge(state.cache.as_ref(), &auth.did).await;
info!(did = %password_mfa.did(), "TOTP disabled (verified via {} and {})", password_mfa.method(), totp_mfa.method());
info!(did = %session_mfa.did(), "TOTP disabled (verified via {} and {})", password_mfa.method(), totp_mfa.method());
Ok(Json(EmptyResponse {}))
}
+11 -14
View File
@@ -57,9 +57,9 @@ pub async fn resend_migration_verification(
let channel = input
.channel
.unwrap_or(tranquil_db_traits::CommsChannel::Email);
let id = input.identifier.trim().to_lowercase();
let identifier = input.identifier.trim().to_lowercase();
let user = match state.repos.user.get_by_email(&id).await {
let user = match state.repos.user.get_by_email(&identifier).await {
Ok(Some(u)) => u,
Ok(None) => {
return Ok(Json(ResendMigrationVerificationOutput { sent: true }));
@@ -73,18 +73,15 @@ pub async fn resend_migration_verification(
if user.email_verified {
return Ok(Json(ResendMigrationVerificationOutput { sent: true }));
}
let target = tranquil_pds::comms::VerificationTarget::resolve(channel, &id, Some(&id)).ok();
if let Some(target) = target {
crate::identity::provision::enqueue_migration_verification(
&state, user.id, &user.did, channel, &target,
)
.await;
} else {
warn!(
channel = ?channel,
"We skipped migration verification because unfortunately the account doesn't have a valid recipient"
);
}
crate::identity::provision::enqueue_migration_verification(
&state,
user.id,
&user.did,
channel,
&identifier,
)
.await;
info!(did = %user.did, channel = ?channel, "Resent migration verification");
+28 -33
View File
@@ -66,7 +66,8 @@ pub async fn verify_token_internal(
handle_channel_update(state, &token_data.did, token_data.channel, &identifier).await
}
VerificationPurpose::Signup => {
handle_signup_verification(state, &token_data.did, token_data.channel).await
handle_signup_verification(state, &token_data.did, token_data.channel, &identifier)
.await
}
}
}
@@ -166,7 +167,7 @@ async fn handle_channel_update(
info!(did = %did, channel = ?channel, "Channel verified successfully");
notify_channel_verified(state, user_id, channel).await;
notify_channel_verified(state, user_id, channel, identifier).await;
Ok(Json(VerifyTokenOutput {
success: true,
@@ -176,49 +177,43 @@ async fn handle_channel_update(
}))
}
async fn notify_channel_verified(state: &AppState, user_id: uuid::Uuid, channel: CommsChannel) {
let prefs = match state.repos.user.get_comms_prefs(user_id).await {
Ok(Some(prefs)) => prefs,
Ok(None) => {
warn!(
user_id = %user_id,
channel = ?channel,
"We skipped channel-verified notice because the account doesn't have comms preferences"
);
return;
}
Err(e) => {
warn!(
user_id = %user_id,
channel = ?channel,
error = ?e,
"We skipped channel-verified notice because we couldn't load the account's comms preferences"
);
return;
}
async fn notify_channel_verified(
state: &AppState,
user_id: uuid::Uuid,
channel: CommsChannel,
identifier: &str,
) {
let recipient = match channel {
CommsChannel::Telegram => state
.repos
.user
.get_telegram_chat_id(user_id)
.await
.ok()
.flatten()
.map(|id| id.to_string())
.unwrap_or_else(|| identifier.to_string()),
_ => identifier.to_string(),
};
let Some(recipient) = tranquil_pds::comms::recipient_for(&prefs, channel) else {
warn!(
user_id = %user_id,
channel = ?channel,
"We skipped channel-verified notice because the account doesn't have a valid recipient"
);
return;
};
comms_repo::try_channel_verified_notice(
if let Err(e) = comms_repo::enqueue_channel_verified(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user_id,
channel,
&recipient,
&tranquil_config::get().server.hostname,
)
.await;
.await
{
warn!(error = %e, "Failed to enqueue channel verified notification");
}
}
async fn handle_signup_verification(
state: &AppState,
did: &Did,
channel: CommsChannel,
identifier: &str,
) -> Result<Json<VerifyTokenOutput>, ApiError> {
let user = state
.repos
@@ -243,7 +238,7 @@ async fn handle_signup_verification(
info!(did = %did, channel = ?channel, "Signup verified successfully");
notify_channel_verified(state, user.id, channel).await;
notify_channel_verified(state, user.id, channel, identifier).await;
Ok(Json(VerifyTokenOutput {
success: true,
+13 -24
View File
@@ -76,13 +76,6 @@ pub async fn handle_telegram_webhook(
return StatusCode::OK;
}
};
let username = match tranquil_types::TelegramUsername::new(username) {
Ok(username) => username,
Err(e) => {
warn!(error = %e, "We ignored a /start with an invalid Telegram username");
return StatusCode::OK;
}
};
debug!(
telegram_username = %username,
@@ -102,21 +95,17 @@ pub async fn handle_telegram_webhook(
chat_id = from.id,
"Verified Telegram user and stored chat_id"
);
match tranquil_types::TelegramChatId::from_i64(from.id) {
Some(chat_id) => {
comms_repo::try_channel_verified_notice(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user_id,
&tranquil_types::Recipient::Telegram(chat_id),
&tranquil_config::get().server.hostname,
)
.await;
}
None => warn!(
chat_id = from.id,
"We skipped verified notice because the Telegram chat ID can't be 0"
),
if let Err(e) = comms_repo::enqueue_channel_verified(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user_id,
tranquil_db_traits::CommsChannel::Telegram,
&from.id.to_string(),
&tranquil_config::get().server.hostname,
)
.await
{
warn!(error = %e, "Failed to enqueue channel verified notification");
}
}
Ok(None) => {
@@ -189,8 +178,8 @@ mod tests {
#[test]
fn payload_with_extra_whitespace_trimmed() {
assert_eq!(
parse_start_handle(Some("/start oys_nel_pet ")),
Some("oys.nel.pet".to_string()),
parse_start_handle(Some("/start alice_example_com ")),
Some("alice.example.com".to_string()),
);
}
}
-1
View File
@@ -7,7 +7,6 @@ license.workspace = true
[dependencies]
tranquil-config = { workspace = true }
tranquil-signal = { workspace = true }
tranquil-types = { workspace = true }
async-trait = { workspace = true }
base64 = { workspace = true }
+15 -24
View File
@@ -1,21 +1,20 @@
use lettre::Message;
use lettre::message::Mailbox;
use lettre::message::header::{ContentType, MIME_VERSION_1_0};
use lettre::message::header::ContentType;
use lettre::message::header::{Header, HeaderName, HeaderValue};
use uuid::Uuid;
use super::types::EmailDomain;
use crate::sender::SendError;
use crate::{CommsType, QueuedComms};
use crate::types::{CommsType, QueuedComms};
pub(super) fn build(
from: &Mailbox,
qc: &QueuedComms,
to: &tranquil_types::EmailAddress,
apply_atmos_categories: bool,
) -> Result<Message, SendError> {
let to: Mailbox = to
.as_str()
let to: Mailbox = qc
.recipient
.parse()
.map_err(|e: lettre::address::AddressError| SendError::InvalidRecipient(e.to_string()))?;
let subject = qc.subject.as_deref().unwrap_or("Notification");
@@ -25,7 +24,6 @@ pub(super) fn build(
.to(to)
.subject(subject)
.message_id(Some(message_id))
.header(MIME_VERSION_1_0)
.header(ContentType::TEXT_PLAIN);
let category = apply_atmos_categories
@@ -102,7 +100,7 @@ fn atmos_category(comms_type: CommsType) -> Option<AtmosCategory> {
#[cfg(test)]
mod tests {
use super::*;
use crate::{CommsChannel, CommsStatus};
use crate::types::{CommsChannel, CommsStatus};
use chrono::Utc;
use uuid::Uuid;
@@ -110,10 +108,6 @@ mod tests {
"Test Sender <noreply@nel.pet>".parse().unwrap()
}
fn to(recipient: &str) -> tranquil_types::EmailAddress {
tranquil_types::EmailAddress::new(recipient).unwrap()
}
fn fixture(recipient: &str, subject: Option<&str>, body: &str) -> QueuedComms {
QueuedComms {
id: Uuid::new_v4(),
@@ -140,7 +134,6 @@ mod tests {
let msg = build(
&from_mailbox(),
&fixture("user@nel.pet", Some("Welcome"), "Hello world."),
&to("user@nel.pet"),
false,
)
.unwrap();
@@ -149,7 +142,6 @@ mod tests {
assert!(raw.contains("From: \"Test Sender\" <noreply@nel.pet>"));
assert!(raw.contains("To: user@nel.pet"));
assert!(raw.contains("Subject: Welcome"));
assert!(raw.contains("MIME-Version: 1.0"));
assert!(lower.contains("content-type: text/plain"));
assert!(raw.contains("Hello world."));
}
@@ -159,7 +151,6 @@ mod tests {
let msg = build(
&from_mailbox(),
&fixture("user@nel.pet", Some("héllo wörld"), "Body"),
&to("user@jola.dev"),
false,
)
.unwrap();
@@ -170,8 +161,12 @@ mod tests {
#[test]
fn header_injection_rejected() {
let result = tranquil_types::EmailAddress::new("x@jola.dev\r\nBcc: evil@x");
assert!(result.is_err());
let result = build(
&from_mailbox(),
&fixture("x@nel.pet\r\nBcc: evil@x", Some("s"), "b"),
false,
);
assert!(matches!(result, Err(SendError::InvalidRecipient(_))));
}
#[test]
@@ -179,14 +174,13 @@ mod tests {
let msg = build(
&from_mailbox(),
&fixture("user@nel.pet", Some("hi\r\nBcc: evil@nel.pet"), "body"),
&to("user@jola.dev"),
false,
)
.expect("subject CRLF should be encoded, not rejected");
let raw = String::from_utf8(msg.formatted()).unwrap();
assert!(
!raw.contains("Bcc:"),
"CRLF in subject mustn't produce a Bcc header: {raw}"
"CRLF in subject must not produce a Bcc header: {raw}"
);
assert!(
raw.contains("Subject: ="),
@@ -199,7 +193,6 @@ mod tests {
let msg = build(
&from_mailbox(),
&fixture("user@nel.pet", Some("s"), "b"),
&to("user@jola.dev"),
false,
)
.unwrap();
@@ -219,7 +212,6 @@ mod tests {
let msg = build(
&from_mailbox(),
&fixture("user@nel.pet", None, "Body"),
&to("user@nel.pet"),
false,
)
.unwrap();
@@ -232,7 +224,6 @@ mod tests {
let msg = build(
&from_mailbox(),
&fixture("user@Nel.PET", Some("s"), "b"),
&to("user@nel.pet"),
false,
)
.unwrap();
@@ -246,7 +237,7 @@ mod tests {
comms_type: CommsType::PasswordReset,
..fixture("user@nel.pet", Some("s"), "b")
};
let msg = build(&from_mailbox(), &qc, &to("user@jola.dev"), true).unwrap();
let msg = build(&from_mailbox(), &qc, true).unwrap();
let raw = String::from_utf8(msg.formatted()).unwrap();
assert!(raw.contains("X-Atmos-Category: password-reset"));
}
@@ -257,7 +248,7 @@ mod tests {
comms_type: CommsType::PasswordReset,
..fixture("user@nel.pet", Some("s"), "b")
};
let msg = build(&from_mailbox(), &qc, &to("user@nel.pet"), false).unwrap();
let msg = build(&from_mailbox(), &qc, false).unwrap();
let raw = String::from_utf8(msg.formatted()).unwrap();
assert!(!raw.contains("X-Atmos-Category"));
}
@@ -268,7 +259,7 @@ mod tests {
comms_type: CommsType::AdminEmail,
..fixture("user@nel.pet", Some("s"), "b")
};
let msg = build(&from_mailbox(), &qc, &to("user@nel.pet"), true).unwrap();
let msg = build(&from_mailbox(), &qc, true).unwrap();
let raw = String::from_utf8(msg.formatted()).unwrap();
assert!(!raw.contains("X-Atmos-Category"));
}
+4 -17
View File
@@ -25,7 +25,7 @@ use self::types::{
SmtpUsername, TlsMode,
};
use crate::sender::{CommsSender, SendError};
use crate::{CommsChannel, QueuedComms};
use crate::types::{CommsChannel, QueuedComms};
pub struct EmailSender {
from: Mailbox,
@@ -193,22 +193,9 @@ impl CommsSender for EmailSender {
CommsChannel::Email
}
async fn send(
&self,
notification: &QueuedComms,
recipient: &tranquil_types::Recipient,
) -> Result<(), SendError> {
let tranquil_types::Recipient::Email(address) = recipient else {
return Err(SendError::InvalidRecipient(
"Recipient isn't an email address".into(),
));
};
let mut message = message::build(
&self.from,
notification,
address,
wants_atmos_categories(&self.mode),
)?;
async fn send(&self, notification: &QueuedComms) -> Result<(), SendError> {
let mut message =
message::build(&self.from, notification, wants_atmos_categories(&self.mode))?;
if let Some(signer) = &self.dkim {
signer.sign(&mut message);
}
+3 -1
View File
@@ -1,6 +1,7 @@
pub mod email;
mod locale;
mod sender;
mod types;
pub use email::EmailSender;
pub use locale::{
@@ -9,5 +10,6 @@ pub use locale::{
};
pub use sender::{
CommsSender, DiscordSender, SendError, SignalSender, TelegramSender, is_valid_phone_number,
is_valid_signal_username,
};
pub use tranquil_db_traits::{CommsChannel, CommsStatus, CommsType, QueuedComms};
pub use types::{CommsChannel, CommsStatus, CommsType, NewComms, QueuedComms};
+15 -39
View File
@@ -3,7 +3,7 @@ use reqwest::Client;
use serde_json::json;
use std::time::Duration;
use tranquil_db_traits::{CommsChannel, QueuedComms};
use super::types::{CommsChannel, QueuedComms};
const HTTP_TIMEOUT_SECS: u64 = 30;
const MAX_RETRIES: u32 = 3;
@@ -12,11 +12,7 @@ const INITIAL_RETRY_DELAY_MS: u64 = 500;
#[async_trait]
pub trait CommsSender: Send + Sync {
fn channel(&self) -> CommsChannel;
async fn send(
&self,
notification: &QueuedComms,
recipient: &tranquil_types::Recipient,
) -> Result<(), SendError>;
async fn send(&self, notification: &QueuedComms) -> Result<(), SendError>;
}
#[derive(Debug, thiserror::Error)]
@@ -144,6 +140,10 @@ pub fn is_valid_phone_number(number: &str) -> bool {
!remaining.is_empty() && remaining.chars().all(|c| c.is_ascii_digit())
}
pub fn is_valid_signal_username(username: &str) -> bool {
tranquil_signal::SignalUsername::parse(username).is_ok()
}
const DISCORD_API_BASE: &str = "https://discord.com/api/v10";
#[derive(Clone)]
@@ -355,17 +355,8 @@ impl CommsSender for DiscordSender {
CommsChannel::Discord
}
async fn send(
&self,
notification: &QueuedComms,
recipient: &tranquil_types::Recipient,
) -> Result<(), SendError> {
let tranquil_types::Recipient::Discord(user_id) = recipient else {
return Err(SendError::InvalidRecipient(
"Recipient isn't a Discord user ID".into(),
));
};
let channel_id = self.open_dm_channel(user_id.as_str()).await?;
async fn send(&self, notification: &QueuedComms) -> Result<(), SendError> {
let channel_id = self.open_dm_channel(&notification.recipient).await?;
let subject = notification.subject.as_deref().unwrap_or("Notification");
let content = format!("**{}**\n\n{}", subject, notification.body);
@@ -462,22 +453,14 @@ impl CommsSender for TelegramSender {
CommsChannel::Telegram
}
async fn send(
&self,
notification: &QueuedComms,
recipient: &tranquil_types::Recipient,
) -> Result<(), SendError> {
let tranquil_types::Recipient::Telegram(chat_id) = recipient else {
return Err(SendError::InvalidRecipient(
"Recipient isn't a Telegram chat ID".into(),
));
};
async fn send(&self, notification: &QueuedComms) -> Result<(), SendError> {
let chat_id = &notification.recipient;
let subject = escape_html(notification.subject.as_deref().unwrap_or("Notification"));
let body = escape_html(&notification.body);
let text = format!("<b>{}</b>\n\n{}", subject, body);
let url = format!("https://api.telegram.org/bot{}/sendMessage", self.bot_token);
let payload = json!({
"chat_id": chat_id.as_str(),
"chat_id": chat_id,
"text": text,
"parse_mode": "HTML"
});
@@ -505,16 +488,9 @@ impl CommsSender for SignalSender {
CommsChannel::Signal
}
async fn send(
&self,
notification: &QueuedComms,
recipient: &tranquil_types::Recipient,
) -> Result<(), SendError> {
let tranquil_types::Recipient::Signal(username) = recipient else {
return Err(SendError::InvalidRecipient(
"Recipient isn't a Signal username".into(),
));
};
async fn send(&self, notification: &QueuedComms) -> Result<(), SendError> {
let username = tranquil_signal::SignalUsername::parse(&notification.recipient)
.map_err(|e| SendError::InvalidRecipient(e.to_string()))?;
let client = self
.slot
@@ -529,7 +505,7 @@ impl CommsSender for SignalSender {
let mut last_error = None;
for attempt in 0..MAX_RETRIES {
match client.send(username, message.clone()).await {
match client.send(&username, message.clone()).await {
Ok(()) => return Ok(()),
Err(e) => {
let err_str = e.to_string();
+51
View File
@@ -0,0 +1,51 @@
use uuid::Uuid;
pub use tranquil_db_traits::{CommsChannel, CommsStatus, CommsType, QueuedComms};
pub struct NewComms {
pub user_id: Uuid,
pub channel: CommsChannel,
pub comms_type: CommsType,
pub recipient: String,
pub subject: Option<String>,
pub body: String,
pub metadata: Option<serde_json::Value>,
}
impl NewComms {
pub fn new(
user_id: Uuid,
channel: CommsChannel,
comms_type: CommsType,
recipient: String,
subject: Option<String>,
body: String,
) -> Self {
Self {
user_id,
channel,
comms_type,
recipient,
subject,
body,
metadata: None,
}
}
pub fn email(
user_id: Uuid,
comms_type: CommsType,
recipient: String,
subject: String,
body: String,
) -> Self {
Self::new(
user_id,
CommsChannel::Email,
comms_type,
recipient,
Some(subject),
body,
)
}
}
+5 -18
View File
@@ -32,10 +32,6 @@ fn fixture(recipient: &str, subject: &str, body: &str) -> QueuedComms {
}
}
fn to(recipient: &str) -> tranquil_types::Recipient {
tranquil_types::Recipient::new(CommsChannel::Email, recipient).unwrap()
}
fn build_smarthost_sender(host: &str, port: u16) -> EmailSender {
build_smarthost_sender_with_total_timeout(host, port, Duration::from_secs(10))
}
@@ -105,9 +101,7 @@ async fn spawn_stub(rcpt_response: &'static [u8]) -> u16 {
async fn rcpt_550_classifies_as_smtp_permanent() {
let port = spawn_stub(b"550 5.1.1 user unknown\r\n").await;
let sender = build_smarthost_sender("127.0.0.1", port);
let result = sender
.send(&fixture("oys@nel.pet", "x", "x"), &to("oys@nel.pet"))
.await;
let result = sender.send(&fixture("nel@nel.pet", "x", "x")).await;
match result {
Err(SendError::SmtpPermanent(_)) => {}
other => panic!("expected SmtpPermanent, got {other:?}"),
@@ -118,9 +112,7 @@ async fn rcpt_550_classifies_as_smtp_permanent() {
async fn rcpt_421_classifies_as_smtp_transient() {
let port = spawn_stub(b"421 4.7.0 try again later\r\n").await;
let sender = build_smarthost_sender("127.0.0.1", port);
let result = sender
.send(&fixture("oys@nel.pet", "x", "x"), &to("oys@nel.pet"))
.await;
let result = sender.send(&fixture("nel@nel.pet", "x", "x")).await;
match result {
Err(SendError::SmtpTransient(_)) => {}
other => panic!("expected SmtpTransient, got {other:?}"),
@@ -128,13 +120,10 @@ async fn rcpt_421_classifies_as_smtp_transient() {
}
#[tokio::test]
async fn send_rejects_mismatched_recipient_variant() {
async fn invalid_recipient_classifies_as_invalid_recipient() {
let port = spawn_stub(b"250 OK\r\n").await;
let sender = build_smarthost_sender("127.0.0.1", port);
let recipient = tranquil_types::Recipient::new(CommsChannel::Signal, "oys.01").unwrap();
let result = sender
.send(&fixture("oys@nel.pet", "x", "x"), &recipient)
.await;
let result = sender.send(&fixture("not-an-address", "x", "x")).await;
match result {
Err(SendError::InvalidRecipient(_)) => {}
other => panic!("expected InvalidRecipient, got {other:?}"),
@@ -157,9 +146,7 @@ async fn smarthost_silent_relay_hits_total_timeout() {
let sender =
build_smarthost_sender_with_total_timeout("127.0.0.1", port, Duration::from_millis(500));
let start = std::time::Instant::now();
let result = sender
.send(&fixture("oys@nel.pet", "x", "x"), &to("oys@nel.pet"))
.await;
let result = sender.send(&fixture("nel@nel.pet", "x", "x")).await;
let elapsed = start.elapsed();
match result {
Err(SendError::Timeout) => {}
-2
View File
@@ -5,6 +5,4 @@ edition.workspace = true
license.workspace = true
[dependencies]
serde = { workspace = true }
tranquil-types = { workspace = true }
confique = { workspace = true }
+22 -27
View File
@@ -2,7 +2,6 @@ use confique::Config;
use std::fmt;
use std::path::PathBuf;
use std::sync::OnceLock;
use tranquil_types::Domain;
static CONFIG: OnceLock<TranquilConfig> = OnceLock::new();
@@ -31,6 +30,7 @@ impl fmt::Display for ConfigError {
}
impl std::error::Error for ConfigError {}
/// Initialize the global configuration. Must be called once at startup before
/// any other code accesses the configuration. Panics if called more than once.
pub fn init(config: TranquilConfig) {
@@ -224,12 +224,6 @@ impl TranquilConfig {
}
}
if let Err(e) = Domain::new(self.server.hostname_without_port()) {
errors.push(format!(
"server.hostname (PDS_HOSTNAME) must be a plain domain, {e}"
));
}
// -- email -----------------------------------------------------------
self.email
.validate(self.server.hostname_without_port(), &mut errors);
@@ -434,7 +428,7 @@ pub struct ServerConfig {
pub hostname: String,
/// Address to bind the HTTP server to.
#[config(env = "SERVER_HOST", default = "[::1]")]
#[config(env = "SERVER_HOST", default = "127.0.0.1")]
pub host: String,
/// Port to bind the HTTP server to.
@@ -444,21 +438,13 @@ pub struct ServerConfig {
/// List of domains for user handles.
/// Defaults to the PDS hostname when not set.
#[config(env = "PDS_USER_HANDLE_DOMAINS", parse_env = split_comma_list)]
pub user_handle_domains: Option<Vec<Domain>>,
pub user_handle_domains: Option<Vec<String>>,
/// Enable PDS-hosted did:web identities. Hosting did:web requires a
/// long-term commitment to serve DID documents; opt-in only.
#[config(env = "ENABLE_PDS_HOSTED_DID_WEB", default = false)]
pub enable_pds_hosted_did_web: bool,
/// The caddy on-demand TLS requires we serve
/// the endpoint `/.well-known/caddy/ask`.
/// It will be used so that caddy can create TLS
/// certs for us on the fly
/// and we don't have to do annoying wildcard certs.
#[config(env = "ENABLE_CADDY_ON_DEMAND_TLS", default = true)]
pub enable_caddy_on_demand_tls: bool,
/// iykyk!
#[config(env = "RFC_MOO_COMPLIANCE", default = false)]
pub rfc_moo_compliance: bool,
@@ -587,6 +573,20 @@ impl ServerConfig {
pub fn banned_word_list(&self) -> Vec<String> {
self.banned_words.clone().unwrap_or_default()
}
/// Returns the user handle domains, falling back to `[hostname_without_port]`.
pub fn user_handle_domain_list(&self) -> Vec<String> {
self.user_handle_domains
.as_deref()
.filter(|v| !v.is_empty())
.map(|v| v.to_vec())
.unwrap_or_else(|| vec![self.hostname_without_port().to_string()])
}
/// Alias for `user_handle_domain_list` (for callers that were using the now-removed `available_user_domains` field).
pub fn available_user_domain_list(&self) -> Vec<String> {
self.user_handle_domain_list()
}
}
#[derive(Debug, Config)]
@@ -770,12 +770,8 @@ pub struct StorageConfig {
#[config(env = "S3_ENDPOINT")]
pub s3_endpoint: Option<String>,
/// Path on the storage for the S3 blob backend.
#[config(env = "S3_PATH", default = "")]
pub s3_path: String,
/// Repository backend: `postgres` by default, or `tranquil-store`, our embedded db.
/// `tranquil-store` is our own solution, not a tried-and-tested out-of-box database. If you're not feeling brave, don't choose this.
/// tranquil-store is EXPERIMENTAL!!!! RISK OF TOTAL DATA LOSS.
#[config(env = "REPO_BACKEND", default = "postgres")]
pub repo_backend: String,
}
@@ -1488,13 +1484,12 @@ pub struct ImportConfig {
/// trimming whitespace and dropping empty entries.
///
/// Signature matches confique's `parse_env` expectation: `fn(&str) -> Result<T, E>`.
fn split_comma_list<T: std::str::FromStr>(value: &str) -> Result<Vec<T>, T::Err> {
value
fn split_comma_list(value: &str) -> Result<Vec<String>, std::convert::Infallible> {
Ok(value
.split(',')
.map(str::trim)
.map(|item| item.trim().to_string())
.filter(|item| !item.is_empty())
.map(T::from_str)
.collect()
.collect())
}
#[derive(Debug, Config)]
+2 -2
View File
@@ -70,12 +70,12 @@ pub trait BlobRepository: Send + Sync {
takedown_ref: Option<&str>,
) -> Result<bool, DbError>;
async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<bool, DbError>;
async fn delete_blobs_by_user(&self, user_id: Uuid) -> Result<u64, DbError>;
async fn get_blob_storage_keys_by_user(&self, user_id: Uuid) -> Result<Vec<String>, DbError>;
async fn ensure_blob_ownership(&self, user_id: Uuid, cid: &CidLink) -> Result<bool, DbError>;
async fn insert_record_blobs(
&self,
repo_id: Uuid,
+61 -3
View File
@@ -1,7 +1,7 @@
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use tranquil_types::{Did, Handle, InviteCode};
use tranquil_types::{CidLink, Did, Handle, InviteCode};
use uuid::Uuid;
use crate::DbError;
@@ -44,7 +44,60 @@ impl InviteCodeState {
}
}
pub use tranquil_types::{CommsChannel, Recipient};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, sqlx::Type)]
#[serde(rename_all = "lowercase")]
#[sqlx(type_name = "comms_channel", rename_all = "snake_case")]
pub enum CommsChannel {
Email,
Discord,
Telegram,
Signal,
}
impl CommsChannel {
pub fn as_str(self) -> &'static str {
match self {
Self::Email => "email",
Self::Discord => "discord",
Self::Telegram => "telegram",
Self::Signal => "signal",
}
}
pub fn display_name(self) -> &'static str {
match self {
Self::Email => "email",
Self::Discord => "Discord",
Self::Telegram => "Telegram",
Self::Signal => "Signal",
}
}
}
impl std::str::FromStr for CommsChannel {
type Err = InvalidCommsChannel;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"email" => Ok(Self::Email),
"discord" => Ok(Self::Discord),
"telegram" => Ok(Self::Telegram),
"signal" => Ok(Self::Signal),
_ => Err(InvalidCommsChannel),
}
}
}
#[derive(Debug, Clone)]
pub struct InvalidCommsChannel;
impl std::fmt::Display for InvalidCommsChannel {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("invalid comms channel")
}
}
impl std::error::Error for InvalidCommsChannel {}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, sqlx::Type)]
#[sqlx(type_name = "comms_type", rename_all = "snake_case")]
@@ -173,8 +226,9 @@ pub trait InfraRepository: Send + Sync {
async fn enqueue_comms(
&self,
user_id: Option<Uuid>,
recipient: &Recipient,
channel: CommsChannel,
comms_type: CommsType,
recipient: &str,
subject: Option<&str>,
body: &str,
metadata: Option<serde_json::Value>,
@@ -363,6 +417,10 @@ pub trait InfraRepository: Send + Sync {
async fn delete_server_config(&self, key: &str) -> Result<(), DbError>;
async fn get_blob_storage_key_by_cid(&self, cid: &CidLink) -> Result<Option<String>, DbError>;
async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<(), DbError>;
async fn get_admin_account_info_by_did(
&self,
did: &Did,
+3 -3
View File
@@ -25,7 +25,7 @@ pub use infra::{
AdminAccountInfo, CommsChannel, CommsStatus, CommsType, DeletionRequest,
DeletionRequestWithToken, InfraRepository, InviteCodeInfo, InviteCodeRow, InviteCodeSortOrder,
InviteCodeState, InviteCodeUse, NotificationHistoryRow, PasswordResetInfo, PlcTokenInfo,
QueuedComms, Recipient, ReservedSigningKey, ReservedSigningKeyFull,
QueuedComms, ReservedSigningKey, ReservedSigningKeyFull,
};
pub use invite_code::{InviteCodeError, ValidatedInviteCode};
pub use oauth::{
@@ -36,8 +36,8 @@ pub use repo::{
AccountStatus, ApplyCommitError, ApplyCommitInput, ApplyCommitResult, CommitEventData,
EventBlockInline, EventBlocks, FullRecordInfo, ImportBlock, ImportRecord, ImportRepoError,
PruneCount, RecordDelete, RecordInfo, RecordUpsert, RecordWithTakedown, RepoAccountInfo,
RepoEventNotifier, RepoEventReceiver, RepoEventType, RepoIdentity, RepoInfo, RepoListItem,
RepoRepository, RepoSeqEvent, RepoWithoutRev, SequencedEvent, UserNeedingRecordBlobsBackfill,
RepoEventNotifier, RepoEventReceiver, RepoEventType, RepoInfo, RepoListItem, RepoRepository,
RepoSeqEvent, RepoWithoutRev, SequencedEvent, UserNeedingRecordBlobsBackfill,
UserWithoutBlocks,
};
pub use scope::{DbScope, InvalidScopeError};
-8
View File
@@ -171,12 +171,6 @@ pub struct UserNeedingRecordBlobsBackfill {
pub did: Did,
}
#[derive(Debug, Clone)]
pub struct RepoIdentity {
pub user_id: Uuid,
pub did: Did,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RepoSeqEvent {
pub seq: SequenceNumber,
@@ -551,8 +545,6 @@ pub trait RepoRepository: Send + Sync {
limit: i64,
) -> Result<Vec<UserNeedingRecordBlobsBackfill>, DbError>;
async fn get_all_repo_identities(&self) -> Result<Vec<RepoIdentity>, DbError>;
async fn insert_record_blobs(
&self,
repo_id: Uuid,
+6 -27
View File
@@ -220,11 +220,7 @@ pub trait UserRepository: Send + Sync {
channel: CommsChannel,
) -> Result<Option<bool>, DbError>;
async fn admin_update_email(
&self,
did: &Did,
email: &tranquil_types::EmailAddress,
) -> Result<u64, DbError>;
async fn admin_update_email(&self, did: &Did, email: &str) -> Result<u64, DbError>;
async fn admin_update_handle(&self, did: &Did, handle: &Handle) -> Result<u64, DbError>;
@@ -270,11 +266,13 @@ pub trait UserRepository: Send + Sync {
async fn store_telegram_chat_id(
&self,
telegram_username: &tranquil_types::TelegramUsername,
telegram_username: &str,
chat_id: i64,
handle: Option<&Handle>,
) -> Result<Option<Uuid>, DbError>;
async fn get_telegram_chat_id(&self, user_id: Uuid) -> Result<Option<i64>, DbError>;
async fn set_unverified_discord(
&self,
user_id: Uuid,
@@ -283,8 +281,8 @@ pub trait UserRepository: Send + Sync {
async fn store_discord_user_id(
&self,
discord_username: &tranquil_types::DiscordUsername,
discord_id: &tranquil_types::DiscordUserId,
discord_username: &str,
discord_id: &str,
handle: Option<&Handle>,
) -> Result<Option<Uuid>, DbError>;
@@ -913,25 +911,6 @@ pub struct UserResendVerification {
pub channel_verification: ChannelVerificationStatus,
}
macro_rules! channel_identifier {
($name:ty) => {
impl $name {
pub fn channel_identifier(&self) -> Option<&str> {
match self.channel {
CommsChannel::Email => self.email.as_deref(),
CommsChannel::Discord => self.discord_username.as_deref(),
CommsChannel::Telegram => self.telegram_username.as_deref(),
CommsChannel::Signal => self.signal_username.as_deref(),
}
.filter(|identifier| !identifier.is_empty())
}
}
};
}
channel_identifier!(UserConfirmSignup);
channel_identifier!(UserResendVerification);
#[derive(Debug, Clone)]
pub struct UserResetCodeInfo {
pub id: Uuid,
+20 -35
View File
@@ -33,7 +33,7 @@ impl BlobRepository for PostgresBlobRepository {
let result = sqlx::query_scalar!(
r#"INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (cid, created_by_user) DO NOTHING RETURNING cid"#,
ON CONFLICT (cid) DO NOTHING RETURNING cid"#,
cid.as_str(),
mime_type,
size_bytes,
@@ -49,7 +49,7 @@ impl BlobRepository for PostgresBlobRepository {
async fn get_blob_metadata(&self, cid: &CidLink) -> Result<Option<BlobMetadata>, DbError> {
let result = sqlx::query!(
"SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1 LIMIT 1",
"SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1",
cid.as_str()
)
.fetch_optional(&self.pool)
@@ -68,7 +68,7 @@ impl BlobRepository for PostgresBlobRepository {
cid: &CidLink,
) -> Result<Option<BlobWithTakedown>, DbError> {
let result = sqlx::query!(
"SELECT cid, takedown_ref FROM blobs WHERE cid = $1 ORDER BY takedown_ref NULLS LAST LIMIT 1",
"SELECT cid, takedown_ref FROM blobs WHERE cid = $1",
cid.as_str()
)
.fetch_optional(&self.pool)
@@ -86,13 +86,11 @@ impl BlobRepository for PostgresBlobRepository {
}
async fn get_blob_storage_key(&self, cid: &CidLink) -> Result<Option<String>, DbError> {
let result = sqlx::query_scalar!(
"SELECT storage_key FROM blobs WHERE cid = $1 LIMIT 1",
cid.as_str()
)
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
let result =
sqlx::query_scalar!("SELECT storage_key FROM blobs WHERE cid = $1", cid.as_str())
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(result)
}
@@ -149,8 +147,7 @@ impl BlobRepository for PostgresBlobRepository {
async fn sum_blob_storage(&self) -> Result<i64, DbError> {
let result = sqlx::query_scalar!(
r#"SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as "total!"
FROM (SELECT DISTINCT cid, size_bytes FROM blobs) t"#
r#"SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as "total!" FROM blobs"#
)
.fetch_one(&self.pool)
.await
@@ -176,6 +173,15 @@ impl BlobRepository for PostgresBlobRepository {
Ok(result.rows_affected() > 0)
}
async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<bool, DbError> {
let result = sqlx::query!("DELETE FROM blobs WHERE cid = $1", cid.as_str())
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(result.rows_affected() > 0)
}
async fn delete_blobs_by_user(&self, user_id: Uuid) -> Result<u64, DbError> {
let result = sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id)
.execute(&self.pool)
@@ -187,12 +193,7 @@ impl BlobRepository for PostgresBlobRepository {
async fn get_blob_storage_keys_by_user(&self, user_id: Uuid) -> Result<Vec<String>, DbError> {
let results = sqlx::query_scalar!(
r#"SELECT storage_key as "storage_key!" FROM blobs b
WHERE created_by_user = $1
AND NOT EXISTS (
SELECT 1 FROM blobs o
WHERE o.cid = b.cid AND o.created_by_user <> $1
)"#,
r#"SELECT storage_key as "storage_key!" FROM blobs WHERE created_by_user = $1"#,
user_id
)
.fetch_all(&self.pool)
@@ -202,22 +203,6 @@ impl BlobRepository for PostgresBlobRepository {
Ok(results)
}
async fn ensure_blob_ownership(&self, user_id: Uuid, cid: &CidLink) -> Result<bool, DbError> {
let result = sqlx::query!(
r#"INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key)
SELECT DISTINCT b.cid, b.mime_type, b.size_bytes, $1::uuid, b.storage_key
FROM blobs b WHERE b.cid = $2
ON CONFLICT (cid, created_by_user) DO NOTHING"#,
user_id,
cid.as_str()
)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(result.rows_affected() > 0)
}
async fn insert_record_blobs(
&self,
repo_id: Uuid,
@@ -253,7 +238,7 @@ impl BlobRepository for PostgresBlobRepository {
let results = sqlx::query!(
r#"SELECT rb.blob_cid, rb.record_uri
FROM record_blobs rb
LEFT JOIN blobs b ON rb.blob_cid = b.cid AND b.created_by_user = $1
LEFT JOIN blobs b ON rb.blob_cid = b.cid
WHERE rb.repo_id = $1 AND b.cid IS NULL AND rb.blob_cid > $2
ORDER BY rb.blob_cid
LIMIT $3"#,
+25 -5
View File
@@ -5,9 +5,9 @@ use tranquil_db_traits::{
AdminAccountInfo, CommsChannel, CommsStatus, CommsType, DbError, DeletionRequest,
DeletionRequestWithToken, InfraRepository, InviteCodeError, InviteCodeInfo, InviteCodeRow,
InviteCodeSortOrder, InviteCodeState, InviteCodeUse, NotificationHistoryRow, PlcTokenInfo,
QueuedComms, Recipient, ReservedSigningKey, ReservedSigningKeyFull, ValidatedInviteCode,
QueuedComms, ReservedSigningKey, ReservedSigningKeyFull, ValidatedInviteCode,
};
use tranquil_types::{Did, InviteCode};
use tranquil_types::{CidLink, Did, InviteCode};
use uuid::Uuid;
use super::col;
@@ -29,8 +29,9 @@ impl InfraRepository for PostgresInfraRepository {
async fn enqueue_comms(
&self,
user_id: Option<Uuid>,
recipient: &Recipient,
channel: CommsChannel,
comms_type: CommsType,
recipient: &str,
subject: Option<&str>,
body: &str,
metadata: Option<serde_json::Value>,
@@ -41,9 +42,9 @@ impl InfraRepository for PostgresInfraRepository {
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id"#,
user_id,
recipient.channel() as CommsChannel,
channel as CommsChannel,
comms_type as CommsType,
recipient.as_str(),
recipient,
subject,
body,
metadata
@@ -1009,6 +1010,25 @@ impl InfraRepository for PostgresInfraRepository {
Ok(())
}
async fn get_blob_storage_key_by_cid(&self, cid: &CidLink) -> Result<Option<String>, DbError> {
let result =
sqlx::query_scalar!("SELECT storage_key FROM blobs WHERE cid = $1", cid.as_str())
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(result)
}
async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<(), DbError> {
sqlx::query!("DELETE FROM blobs WHERE cid = $1", cid.as_str())
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
async fn get_admin_account_info_by_did(
&self,
did: &Did,
+2 -25
View File
@@ -4,9 +4,8 @@ use sqlx::PgPool;
use tranquil_db_traits::{
AccountStatus, CommitEventData, DbError, EventBlockInline, EventBlocks, FullRecordInfo,
ImportBlock, ImportRecord, ImportRepoError, PruneCount, RecordInfo, RecordWithTakedown,
RepoAccountInfo, RepoEventType, RepoIdentity, RepoInfo, RepoListItem, RepoRepository,
RepoWithoutRev, SequenceNumber, SequencedEvent, UserNeedingRecordBlobsBackfill,
UserWithoutBlocks,
RepoAccountInfo, RepoEventType, RepoInfo, RepoListItem, RepoRepository, RepoWithoutRev,
SequenceNumber, SequencedEvent, UserNeedingRecordBlobsBackfill, UserWithoutBlocks,
};
use tranquil_types::{AtUri, CidLink, Did, Handle, Nsid, Rkey, Tid};
use uuid::Uuid;
@@ -1651,28 +1650,6 @@ impl RepoRepository for PostgresRepoRepository {
.collect()
}
async fn get_all_repo_identities(&self) -> Result<Vec<RepoIdentity>, DbError> {
let rows = sqlx::query!(
r#"
SELECT u.id as user_id, u.did
FROM users u
JOIN repos r ON r.user_id = u.id
"#
)
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
rows.into_iter()
.map(|r| {
Ok(RepoIdentity {
user_id: r.user_id,
did: column(r.did, col::USERS_DID)?,
})
})
.collect()
}
async fn insert_record_blobs(
&self,
repo_id: Uuid,
+13 -11
View File
@@ -660,14 +660,10 @@ impl UserRepository for PostgresUserRepository {
}))
}
async fn admin_update_email(
&self,
did: &Did,
email: &tranquil_types::EmailAddress,
) -> Result<u64, DbError> {
async fn admin_update_email(&self, did: &Did, email: &str) -> Result<u64, DbError> {
let result = sqlx::query!(
"UPDATE users SET email = $1 WHERE did = $2",
email.as_str(),
email,
did.as_str()
)
.execute(&self.pool)
@@ -3304,11 +3300,10 @@ impl UserRepository for PostgresUserRepository {
async fn store_discord_user_id(
&self,
discord_username: &tranquil_types::DiscordUsername,
discord_id: &tranquil_types::DiscordUserId,
discord_username: &str,
discord_id: &str,
handle: Option<&Handle>,
) -> Result<Option<Uuid>, DbError> {
let (discord_username, discord_id) = (discord_username.as_str(), discord_id.as_str());
let result = match handle {
Some(h) => sqlx::query_scalar!(
"UPDATE users SET discord_id = $2, discord_verified = TRUE, updated_at = NOW() WHERE LOWER(discord_username) = LOWER($1) AND discord_username IS NOT NULL AND handle = $3 RETURNING id",
@@ -3367,11 +3362,10 @@ impl UserRepository for PostgresUserRepository {
async fn store_telegram_chat_id(
&self,
telegram_username: &tranquil_types::TelegramUsername,
telegram_username: &str,
chat_id: i64,
handle: Option<&Handle>,
) -> Result<Option<Uuid>, DbError> {
let telegram_username = telegram_username.as_str();
let result = match handle {
Some(h) => sqlx::query_scalar!(
"UPDATE users SET telegram_chat_id = $2, telegram_verified = TRUE, updated_at = NOW() WHERE LOWER(telegram_username) = LOWER($1) AND telegram_username IS NOT NULL AND handle = $3 RETURNING id",
@@ -3399,6 +3393,14 @@ impl UserRepository for PostgresUserRepository {
Ok(result)
}
async fn get_telegram_chat_id(&self, user_id: Uuid) -> Result<Option<i64>, DbError> {
let row = sqlx::query_scalar!("SELECT telegram_chat_id FROM users WHERE id = $1", user_id)
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(row.flatten())
}
async fn get_password_reset_info(
&self,
email: &str,
@@ -1,5 +1,4 @@
use super::*;
use tranquil_scopes::{ParsedScope, parse_scope};
use tranquil_types::Nsid;
#[derive(Debug, Serialize)]
@@ -11,7 +10,6 @@ pub struct ScopeInfo {
pub display_name: String,
pub granted: Option<bool>,
pub restricted: bool,
pub superseded: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub effective_scope: Option<String>,
}
@@ -29,7 +27,6 @@ pub struct PermissionSetInfo {
pub expanded: Vec<ScopeInfo>,
pub granted: Option<bool>,
pub restricted: bool,
pub superseded: bool,
}
#[derive(Debug, Serialize)]
@@ -43,13 +40,6 @@ pub struct FailedSetInfo {
pub reason: tranquil_scopes::ResolveFailure,
}
#[derive(Debug, Serialize)]
pub struct RejectedScopeInfo {
// The scope exactly as the client requested it, which may be invalid or malformed.
pub scope: String,
pub reason: tranquil_scopes::ScopeRejection,
}
#[derive(Debug, Serialize)]
pub struct ConsentResponse {
pub request_uri: String,
@@ -59,9 +49,7 @@ pub struct ConsentResponse {
pub logo_uri: Option<String>,
pub scopes: Vec<ScopeInfo>,
pub permission_sets: Vec<PermissionSetInfo>,
pub transition_supersedes: bool,
pub failed_sets: Vec<FailedSetInfo>,
pub rejected_scopes: Vec<RejectedScopeInfo>,
pub show_consent: bool,
pub did: Did,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -164,13 +152,9 @@ pub async fn consent_get(
Some(grant) => scope_resolution::Authority::Delegated(&grant.granted_scopes),
None => scope_resolution::Authority::FullSelf,
};
let effective = scope_resolution::resolve_effective_scopes(
&*state.cache,
requested_scope_str,
authority,
client_metadata.as_ref().and_then(|m| m.scope.as_deref()),
)
.await;
let effective =
scope_resolution::resolve_effective_scopes(&*state.cache, requested_scope_str, authority)
.await;
let requested_scopes: Vec<&str> = effective.permitted.split_whitespace().collect();
let preferences = state
.repos
@@ -182,7 +166,16 @@ pub async fn consent_get(
.iter()
.map(|p| (p.scope.as_str(), p.granted))
.collect();
let presented_item_strings = effective.outcome.unexpanded_scopes();
let presented_item_strings: Vec<String> = effective
.outcome
.passthrough
.iter()
.cloned()
.chain(effective.outcome.sets.iter().map(|g| match &g.aud {
Some(a) => format!("include:{}?aud={}", g.nsid, a),
None => format!("include:{}", g.nsid),
}))
.collect();
let show_consent = should_show_consent(
state.repos.oauth.as_ref(),
&did,
@@ -192,9 +185,6 @@ pub async fn consent_get(
.await
.unwrap_or(true);
let has_granular_scopes = requested_scopes.iter().any(|s| is_granular_scope(s));
let has_transition_generic = requested_scopes
.iter()
.any(|s| matches!(parse_scope(s), ParsedScope::TransitionGeneric));
let grant_scope_str: Option<&str> =
delegation_grant.as_ref().map(|g| g.granted_scopes.as_str());
@@ -247,8 +237,6 @@ pub async fn consent_get(
)
};
let granted = pref_map.get(scope).copied();
let superseded = has_transition_generic
&& tranquil_scopes::superseded_by_transition_generic(&parse_scope(scope));
ScopeInfo {
scope: scope.to_string(),
category,
@@ -257,7 +245,6 @@ pub async fn consent_get(
display_name,
granted,
restricted,
superseded,
effective_scope,
}
};
@@ -274,10 +261,12 @@ pub async fn consent_get(
.sets
.iter()
.map(|g| {
let include_scope = g.include_token();
let include_scope = match &g.aud {
Some(a) => format!("include:{}?aud={}", g.nsid, a),
None => format!("include:{}", g.nsid),
};
let expanded: Vec<ScopeInfo> = g.expanded.iter().map(|s| make_scope_info(s)).collect();
let restricted = !expanded.is_empty() && expanded.iter().all(|s| s.restricted);
let superseded = !expanded.is_empty() && expanded.iter().all(|s| s.superseded);
PermissionSetInfo {
nsid: g.nsid.clone(),
aud: g.aud.clone(),
@@ -287,7 +276,6 @@ pub async fn consent_get(
include_scope,
expanded,
restricted,
superseded,
}
})
.collect();
@@ -303,16 +291,6 @@ pub async fn consent_get(
})
.collect();
let rejected_scopes: Vec<RejectedScopeInfo> = effective
.outcome
.rejected
.iter()
.map(|r| RejectedScopeInfo {
scope: r.scope.clone(),
reason: r.reason,
})
.collect();
let account_handle = state
.repos
.user
@@ -354,9 +332,6 @@ pub async fn consent_get(
(None, None, None, None)
};
let transition_supersedes =
scopes.iter().any(|s| s.superseded) || permission_sets.iter().any(|s| s.superseded);
Json(ConsentResponse {
request_uri: query.request_uri.clone(),
client_id: request_data.parameters.client_id.clone(),
@@ -365,9 +340,7 @@ pub async fn consent_get(
logo_uri: client_metadata.as_ref().and_then(|m| m.logo_uri.clone()),
scopes,
permission_sets,
transition_supersedes,
failed_sets,
rejected_scopes,
show_consent,
did: did.clone(),
handle: account_handle,
@@ -459,19 +432,9 @@ pub async fn consent_post(
Some(grant) => scope_resolution::Authority::Delegated(&grant.granted_scopes),
None => scope_resolution::Authority::FullSelf,
};
let client_scope = state
.client_metadata_cache
.get(&request_data.parameters.client_id)
.await
.ok()
.and_then(|m| m.scope);
let effective = scope_resolution::resolve_effective_scopes(
&*state.cache,
original_scope_str,
authority,
client_scope.as_deref(),
)
.await;
let effective =
scope_resolution::resolve_effective_scopes(&*state.cache, original_scope_str, authority)
.await;
let include_token = |nsid: &str, aud: &Option<String>| -> String {
match aud {
Some(a) => format!("include:{}?aud={}", nsid, a),
@@ -498,7 +461,19 @@ pub async fn consent_post(
),
);
}
let presented_items = effective.outcome.unexpanded_scopes();
let presented_items: Vec<String> = effective
.outcome
.passthrough
.iter()
.cloned()
.chain(
effective
.outcome
.sets
.iter()
.map(|g| include_token(&g.nsid, &g.aud)),
)
.collect();
let atproto_was_requested = presented_items.iter().any(|s| s == "atproto");
if atproto_was_requested && !form.approved_scopes.contains(&"atproto".to_string()) {
return json_error(
@@ -517,6 +492,14 @@ pub async fn consent_post(
);
}
let approved_scope_str = final_approved.join(" ");
let has_valid_scope = final_approved.iter().all(|s| is_valid_scope(s));
if !has_valid_scope {
return json_error(
StatusCode::BAD_REQUEST,
"invalid_request",
"Invalid scope format",
);
}
if form.remember {
let preferences: Vec<ScopePreference> = presented_items
.iter()
@@ -572,31 +572,20 @@ pub async fn authorize_post(
{
Ok(challenge) => {
let hostname = &tranquil_config::get().server.hostname;
match enqueue_notice(
if let Err(e) = enqueue_2fa_code(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user.id,
Notice::TwoFactorCode {
code: &challenge.code,
},
&challenge.code,
hostname,
)
.await
{
Ok(Some(_)) => {}
Ok(None) => {
return show_login_error(
"We couldn't deliver this verification code to your notification channels. Please contact the PDS owner.",
json_response,
);
}
Err(e) => {
tracing::warn!(
did = %user.did,
error = %e,
"Failed to enqueue 2FA notification"
);
}
tracing::warn!(
did = %user.did,
error = %e,
"Failed to enqueue 2FA notification"
);
}
let channel_name = user.preferred_comms_channel.display_name();
if json_response {
@@ -918,32 +907,20 @@ pub async fn authorize_select(
{
Ok(challenge) => {
let hostname = &tranquil_config::get().server.hostname;
match enqueue_notice(
if let Err(e) = enqueue_2fa_code(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user.id,
Notice::TwoFactorCode {
code: &challenge.code,
},
&challenge.code,
hostname,
)
.await
{
Ok(Some(_)) => {}
Ok(None) => {
return json_error(
StatusCode::BAD_REQUEST,
"invalid_request",
"We couldn't deliver this verification code to your notification chanels. Please contact the PDS owner.",
);
}
Err(e) => {
tracing::warn!(
did = %form.did,
error = %e,
"Failed to enqueue 2FA notification"
);
}
tracing::warn!(
did = %form.did,
error = %e,
"Failed to enqueue 2FA notification"
);
}
let channel_name = user.preferred_comms_channel.display_name();
return Json(serde_json::json!({
@@ -12,8 +12,7 @@ use serde::{Deserialize, Serialize};
use subtle::ConstantTimeEq;
use tranquil_db_traits::{ScopePreference, WebauthnChallengeType};
use tranquil_pds::auth::{BareLoginIdentifier, NormalizedLoginIdentifier};
use tranquil_pds::comms::Notice;
use tranquil_pds::comms::comms_repo::enqueue_notice;
use tranquil_pds::comms::comms_repo::enqueue_2fa_code;
use tranquil_pds::oauth::{
AuthFlow, DeviceData, DeviceId, OAuthError, Prompt, SessionId, db::should_show_consent,
};
@@ -78,6 +77,15 @@ fn is_granular_scope(s: &str) -> bool {
|| s.starts_with("identity:")
}
fn is_valid_scope(s: &str) -> bool {
s == "atproto"
|| s == "transition:generic"
|| s == "transition:chat.bsky"
|| s == "transition:email"
|| is_granular_scope(s)
|| s.starts_with("include:")
}
fn extract_device_cookie(headers: &HeaderMap) -> Option<tranquil_types::DeviceId> {
headers
.get("cookie")
@@ -1273,31 +1273,16 @@ pub async fn authorize_passkey_finish(
.await
{
Ok(challenge) => {
match enqueue_notice(
if let Err(e) = enqueue_2fa_code(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user.id,
Notice::TwoFactorCode {
code: &challenge.code,
},
&challenge.code,
pds_hostname,
)
.await
{
Ok(Some(_)) => {}
Ok(None) => {
return (
StatusCode::BAD_REQUEST,
Json(serde_json::json!({
"error": "invalid_request",
"error_description": "We couldn't deliver the verification code to your notification channels. Please contact the PDS owner! <3"
})),
)
.into_response();
}
Err(e) => {
tracing::warn!(did = %did, error = %e, "Failed to enqueue 2FA notification");
}
tracing::warn!(did = %did, error = %e, "Failed to enqueue 2FA notification");
}
let channel_name = user.preferred_comms_channel.display_name();
let redirect_url = format!(
@@ -1,10 +1,8 @@
use tranquil_db_traits::DbScope;
use tranquil_pds::cache::Cache;
use tranquil_pds::delegation::{GrantCoverage, grant_coverage, intersect_scopes};
use tranquil_pds::delegation::intersect_scopes;
use tranquil_pds::oauth::permission_set_resolver::expand_scopes;
use tranquil_scopes::{
ExpansionOutcome, ParsedScope, RejectedScope, RepoScope, ScopeRejection, parse_scope,
};
use tranquil_scopes::ExpansionOutcome;
pub enum Authority<'a> {
FullSelf,
@@ -22,12 +20,8 @@ pub async fn resolve_effective_scopes(
cache: &dyn Cache,
requested: &str,
authority: Authority<'_>,
client_scope: Option<&str>,
) -> EffectiveScopes {
let mut outcome = expand_scopes(cache, requested).await;
if let Some(registered) = client_scope.map(str::trim).filter(|s| !s.is_empty()) {
reject_unregistered(&mut outcome, registered);
}
let outcome = expand_scopes(cache, requested).await;
let expanded = outcome.to_scope_string();
let permitted = match authority {
Authority::FullSelf => expanded,
@@ -36,60 +30,6 @@ pub async fn resolve_effective_scopes(
EffectiveScopes { permitted, outcome }
}
fn reject_unregistered(outcome: &mut ExpansionOutcome, registered: &str) {
let mut rejected = Vec::new();
let mut keep = |scope: String| match grant_coverage(registered, &scope) {
GrantCoverage::Full => Some(scope),
GrantCoverage::Narrowed(narrowed) => {
rejected.extend(narrowed_out(&scope, &narrowed).map(|scope| RejectedScope {
scope,
reason: ScopeRejection::NotRegistered,
}));
Some(narrowed)
}
GrantCoverage::Withheld => {
rejected.push(RejectedScope {
scope,
reason: ScopeRejection::NotRegistered,
});
None
}
};
outcome.passthrough = std::mem::take(&mut outcome.passthrough)
.into_iter()
.filter_map(&mut keep)
.collect();
outcome.sets = std::mem::take(&mut outcome.sets)
.into_iter()
.filter(|group| keep(group.include_token()).is_some())
.collect();
outcome.rejected.extend(rejected);
}
/// The repo actions dropped when `requested` was narrowed to `narrowed`, as a scope of their own.
/// Only repo scopes are ever narrowed; anything else yields `None`.
fn narrowed_out(requested: &str, narrowed: &str) -> Option<String> {
let (ParsedScope::Repo(requested), ParsedScope::Repo(narrowed)) =
(parse_scope(requested), parse_scope(narrowed))
else {
return None;
};
let actions: std::collections::HashSet<_> = requested
.actions
.difference(&narrowed.actions)
.copied()
.collect();
(!actions.is_empty()).then(|| {
RepoScope {
collection: requested.collection,
actions,
}
.to_scope_string()
})
}
#[cfg(test)]
mod tests {
use super::*;
@@ -125,7 +65,6 @@ mod tests {
&c,
"atproto include:io.atcr.authFullApp",
Authority::FullSelf,
None,
)
.await;
assert!(eff.permitted.contains("atproto"));
@@ -149,7 +88,6 @@ mod tests {
&c,
"atproto include:io.atcr.authFullApp",
Authority::Delegated(&granted),
None,
)
.await;
assert!(eff.permitted.contains("atproto"));
@@ -159,106 +97,4 @@ mod tests {
);
assert!(!eff.permitted.contains("identity"));
}
#[tokio::test]
async fn unrecognized_scopes_never_reach_permitted() {
let c = MemoryCache::new();
let eff = resolve_effective_scopes(&c, "atproto chat", Authority::FullSelf, None).await;
assert!(eff.permitted.split_whitespace().any(|s| s == "atproto"));
assert!(
!eff.permitted.split_whitespace().any(|s| s == "chat"),
"permitted was {:?}",
eff.permitted
);
assert_eq!(eff.outcome.rejected.len(), 1);
assert_eq!(eff.outcome.rejected[0].reason, ScopeRejection::Unrecognized);
}
#[tokio::test]
async fn scopes_absent_from_client_metadata_are_rejected() {
let c = MemoryCache::new();
let eff = resolve_effective_scopes(
&c,
"atproto identity:*",
Authority::FullSelf,
Some("atproto"),
)
.await;
assert!(!eff.permitted.split_whitespace().any(|s| s == "identity:*"));
assert_eq!(eff.outcome.rejected.len(), 1);
assert_eq!(eff.outcome.rejected[0].scope, "identity:*");
assert_eq!(
eff.outcome.rejected[0].reason,
ScopeRejection::NotRegistered
);
}
#[tokio::test]
async fn wildcard_client_registration_covers_narrower_request() {
let c = MemoryCache::new();
let eff = resolve_effective_scopes(
&c,
"atproto repo:app.bsky.feed.post?action=create",
Authority::FullSelf,
Some("atproto repo:*"),
)
.await;
assert!(eff.outcome.rejected.is_empty());
assert!(
eff.permitted
.contains("repo:app.bsky.feed.post?action=create")
);
}
#[tokio::test]
async fn absent_client_metadata_scope_constrains_nothing() {
let c = MemoryCache::new();
let eff =
resolve_effective_scopes(&c, "atproto identity:*", Authority::FullSelf, None).await;
assert!(eff.outcome.rejected.is_empty());
assert!(eff.permitted.contains("identity:*"));
}
#[tokio::test]
async fn set_expanded_scopes_bypass_the_client_registration_check() {
let c = cache_with("io.atcr.authFullApp", "identity:*").await;
let eff = resolve_effective_scopes(
&c,
"atproto include:io.atcr.authFullApp",
Authority::FullSelf,
Some("atproto include:io.atcr.authFullApp"),
)
.await;
assert!(
eff.outcome.rejected.is_empty(),
"a permission set legitimately expands to scopes the client never registered"
);
assert!(eff.permitted.contains("identity:*"));
}
#[tokio::test]
async fn actions_narrowed_out_by_client_metadata_are_reported_as_rejected() {
let c = MemoryCache::new();
let eff = resolve_effective_scopes(
&c,
"atproto repo:app.bsky.feed.post?action=create&action=delete",
Authority::FullSelf,
Some("atproto repo:*?action=create"),
)
.await;
assert!(
eff.permitted
.split_whitespace()
.any(|s| s == "repo:app.bsky.feed.post?action=create"),
"permitted was {:?}",
eff.permitted
);
assert_eq!(
eff.outcome.rejected,
vec![RejectedScope {
scope: "repo:app.bsky.feed.post?action=delete".to_string(),
reason: ScopeRejection::NotRegistered,
}]
);
}
}
@@ -5,6 +5,7 @@ use serde::{Deserialize, Serialize};
use tranquil_pds::oauth::{
AuthorizationRequestParameters, ClientAuth, CodeChallengeMethod, OAuthError, Prompt,
RequestData, RequestId, ResponseMode, ResponseType,
scopes::{ParsedScope, parse_scope},
};
use tranquil_pds::rate_limit::{OAuthParLimit, OAuthRateLimited};
use tranquil_pds::state::AppState;
@@ -83,7 +84,7 @@ pub async fn pushed_authorization_request(
let client_metadata = client_cache.get(&request.client_id).await?;
client_cache.validate_redirect_uri(&client_metadata, &request.redirect_uri)?;
let client_auth = determine_client_auth(&request)?;
let validated_scope = normalize_scope(&request.scope)?;
let validated_scope = validate_scope(&request.scope, &client_metadata)?;
let request_id = RequestId::generate();
let expires_at = Utc::now() + Duration::seconds(PAR_EXPIRY_SECONDS);
let response_mode = parse_response_mode(request.response_mode.as_deref())?;
@@ -164,7 +165,10 @@ fn determine_client_auth(request: &ParRequest) -> Result<ClientAuth, OAuthError>
Ok(ClientAuth::None)
}
fn normalize_scope(requested_scope: &Option<String>) -> Result<Option<String>, OAuthError> {
fn validate_scope(
requested_scope: &Option<String>,
client_metadata: &tranquil_pds::oauth::ClientMetadata,
) -> Result<Option<String>, OAuthError> {
let scope_str = match requested_scope {
Some(s) if !s.is_empty() => s,
_ => return Ok(Some("atproto".to_string())),
@@ -173,14 +177,80 @@ fn normalize_scope(requested_scope: &Option<String>) -> Result<Option<String>, O
if requested_scopes.is_empty() {
return Ok(Some("atproto".to_string()));
}
if !requested_scopes.contains(&"atproto") {
if let Some(unknown) = requested_scopes
.iter()
.find(|s| matches!(parse_scope(s), ParsedScope::Unknown(_)))
{
return Err(OAuthError::InvalidScope(format!(
"Unsupported scope: {}",
unknown
)));
}
let has_transition = requested_scopes.iter().any(|s| {
matches!(
parse_scope(s),
ParsedScope::TransitionGeneric
| ParsedScope::TransitionChat
| ParsedScope::TransitionEmail
)
});
let has_granular = requested_scopes.iter().any(|s| {
matches!(
parse_scope(s),
ParsedScope::Repo(_)
| ParsedScope::Blob(_)
| ParsedScope::Rpc(_)
| ParsedScope::Account(_)
| ParsedScope::Identity(_)
| ParsedScope::Include(_)
)
});
if has_transition && has_granular {
return Err(OAuthError::InvalidScope(
"The atproto scope is required".to_string(),
"Cannot mix transition scopes with granular scopes. Use either transition:* scopes OR granular scopes (repo:*, blob:*, rpc:*, account:*, include:*), not both.".to_string()
));
}
if let Some(client_scope) = &client_metadata.scope {
let client_scopes: Vec<&str> = client_scope.split_whitespace().collect();
if let Some(unregistered) = requested_scopes
.iter()
.find(|scope| !client_scopes.iter().any(|cs| scope_matches(cs, scope)))
{
return Err(OAuthError::InvalidScope(format!(
"Scope '{}' not registered for this client",
unregistered
)));
}
}
Ok(Some(requested_scopes.join(" ")))
}
fn scope_matches(client_scope: &str, requested_scope: &str) -> bool {
if client_scope == requested_scope {
return true;
}
fn get_resource_type(scope: &str) -> &str {
let base = scope.split('?').next().unwrap_or(scope);
base.split(':').next().unwrap_or(base)
}
let client_type = get_resource_type(client_scope);
let requested_type = get_resource_type(requested_scope);
if client_type == requested_type {
let client_base = client_scope.split('?').next().unwrap_or(client_scope);
if client_base.contains('*') {
return true;
}
}
false
}
fn parse_response_type(value: &str) -> Result<ResponseType, OAuthError> {
match value {
"code" => Ok(ResponseType::Code),
@@ -230,45 +300,3 @@ fn parse_prompt(value: Option<&str>) -> Result<Option<Prompt>, OAuthError> {
))),
}
}
#[cfg(test)]
mod tests {
use super::*;
fn normalized(scope: Option<&str>) -> Result<Option<String>, OAuthError> {
normalize_scope(&scope.map(str::to_string))
}
#[test]
fn absent_or_blank_scope_defaults_to_atproto() {
assert_eq!(normalized(None).unwrap().as_deref(), Some("atproto"));
assert_eq!(normalized(Some("")).unwrap().as_deref(), Some("atproto"));
assert_eq!(normalized(Some(" ")).unwrap().as_deref(), Some("atproto"));
}
#[test]
fn scope_without_atproto_is_invalid() {
assert!(matches!(
normalized(Some("repo:*?action=create blob:*/*")),
Err(OAuthError::InvalidScope(_))
));
}
#[test]
fn atproto_need_not_come_first() {
assert_eq!(
normalized(Some("repo:*?action=create atproto"))
.unwrap()
.as_deref(),
Some("repo:*?action=create atproto")
);
}
#[test]
fn unrecognized_scopes_still_pass_par() {
assert_eq!(
normalized(Some("atproto chat")).unwrap().as_deref(),
Some("atproto chat")
);
}
}
@@ -157,7 +157,6 @@ pub async fn handle_authorization_code_grant(
&*state.cache,
requested_for_resolve,
authority,
client_metadata.scope.as_deref(),
)
.await;
if !effective.outcome.failures.is_empty() {
@@ -202,10 +201,7 @@ pub async fn handle_authorization_code_grant(
details: None,
code: None,
current_refresh_token: Some(refresh_token.clone()),
// Filtered but unexpanded: a remembered consent skips the consent screen, so the raw
// request can still hold scopes the client no longer registers. Sets stay as `include:`
// tokens so refresh re-resolves them.
scope: Some(effective.outcome.unexpanded_scopes().join(" ")),
scope: requested_scope.clone(),
controller_did: controller_did.clone(),
};
state
@@ -278,13 +274,10 @@ async fn recompute_resolved_scope(
Some(g) => crate::endpoints::authorize::scope_resolution::Authority::Delegated(g),
None => crate::endpoints::authorize::scope_resolution::Authority::FullSelf,
};
// No client metadata check here: `token_data.scope` was already filtered against it when
// the token was issued, so there is nothing for a re-check to remove.
let effective = crate::endpoints::authorize::scope_resolution::resolve_effective_scopes(
&*state.cache,
requested,
authority,
None,
)
.await;
if !effective.outcome.failures.is_empty() {
@@ -789,16 +789,13 @@ pub async fn check_handle_available(
}
};
let available_domains = tranquil_pds::handle::ServiceDomains::for_user_handles();
if let Some(d) = &query.domain
&& !available_domains.contains(d.as_str())
let available_domains = tranquil_config::get().server.available_user_domain_list();
if let Some(ref d) = query.domain
&& !available_domains.iter().any(|ad| ad == d)
{
return Err(ApiError::InvalidRequest("Unknown user domain".into()));
}
let domain = query
.domain
.as_deref()
.unwrap_or_else(|| available_domains.primary().as_str());
let domain = query.domain.as_deref().unwrap_or(&available_domains[0]);
let full_handle = format!("{}.{}", validated, domain);
let handle: tranquil_pds::types::Handle = match full_handle.parse() {
Ok(h) => h,
@@ -885,57 +882,87 @@ pub async fn complete_registration(
let cfg = tranquil_config::get();
let hostname = &cfg.server.hostname;
let available_domains = tranquil_pds::handle::ServiceDomains::for_user_handles();
let available_domains = cfg.server.available_user_domain_list();
let split = available_domains.split_handle(&input.handle);
let matched_domain = available_domains
.iter()
.filter(|d| input.handle.ends_with(&format!(".{}", d)))
.max_by_key(|d| d.len());
let handle: tranquil_pds::types::Handle = if !input.handle.contains('.') || split.is_some() {
let handle_to_validate = match split {
Some((_domain, short)) => short,
None => input.handle.as_str(),
let handle: tranquil_pds::types::Handle =
if !input.handle.contains('.') || matched_domain.is_some() {
let handle_to_validate = match matched_domain {
Some(domain) => input
.handle
.strip_suffix(&format!(".{}", domain))
.unwrap_or(&input.handle),
None => &input.handle,
};
match tranquil_pds::api::validation::validate_short_handle(handle_to_validate) {
Ok(h) => format!("{}.{}", h, matched_domain.unwrap_or(&available_domains[0]))
.parse()
.map_err(|_| ApiError::InvalidHandle(None))?,
Err(_) => return Err(ApiError::InvalidHandle(None)),
}
} else {
match tranquil_pds::api::validation::validate_full_domain_handle(&input.handle) {
Ok(h) => h,
Err(_) => return Err(ApiError::InvalidHandle(None)),
}
};
match tranquil_pds::api::validation::validate_short_handle(handle_to_validate) {
Ok(h) => format!(
"{}.{}",
h,
split
.map(|(d, _)| d)
.unwrap_or_else(|| available_domains.primary())
)
.parse()
.map_err(|_| ApiError::InvalidHandle(None))?,
Err(_) => return Err(ApiError::InvalidHandle(None)),
}
} else {
match tranquil_pds::api::validation::validate_full_domain_handle(&input.handle) {
Ok(h) => h,
Err(_) => return Err(ApiError::InvalidHandle(None)),
}
};
let verification_channel = input
.verification_channel
.unwrap_or(tranquil_db_traits::CommsChannel::Email);
let effective_email = input
.email
.clone()
.or_else(|| {
pending_preview
.provider_email
let verification_recipient = match verification_channel {
tranquil_db_traits::CommsChannel::Email => {
let email = input
.email
.clone()
.map(|e| e.into_inner())
})
.map(|e| e.trim().to_string())
.filter(|e| !e.is_empty());
let target = tranquil_api::common::extract_verification_recipient(
verification_channel,
&tranquil_api::common::ChannelInput {
email: effective_email.as_deref(),
discord_username: input.discord_username.as_deref(),
telegram_username: input.telegram_username.as_deref(),
signal_username: input.signal_username.as_deref(),
.or_else(|| {
pending_preview
.provider_email
.clone()
.map(|e| e.into_inner())
})
.map(|e| e.trim().to_string())
.filter(|e| !e.is_empty());
match email {
Some(e) if !e.is_empty() => e,
_ => return Err(ApiError::MissingEmail),
}
}
tranquil_db_traits::CommsChannel::Discord => match &input.discord_username {
Some(username) if !username.trim().is_empty() => {
let clean = username.trim().to_lowercase();
if !tranquil_pds::api::validation::is_valid_discord_username(&clean) {
return Err(ApiError::InvalidRequest(
"Invalid Discord username. Must be 2-32 lowercase characters (letters, numbers, underscores, periods)".into(),
));
}
clean
}
_ => return Err(ApiError::MissingDiscordId),
},
)?;
tranquil_db_traits::CommsChannel::Telegram => match &input.telegram_username {
Some(username) if !username.trim().is_empty() => {
let clean = username.trim().trim_start_matches('@');
if !tranquil_pds::api::validation::is_valid_telegram_username(clean) {
return Err(ApiError::InvalidRequest(
"Invalid Telegram username. Must be 5-32 characters, alphanumeric or underscore".into(),
));
}
clean.to_string()
}
_ => return Err(ApiError::MissingTelegramUsername),
},
tranquil_db_traits::CommsChannel::Signal => match &input.signal_username {
Some(username) if !username.trim().is_empty() => {
username.trim().trim_start_matches('@').to_lowercase()
}
_ => return Err(ApiError::MissingSignalNumber),
},
};
let email = input
.email
@@ -949,11 +976,18 @@ pub async fn complete_registration(
.map(|e| e.trim().to_string())
.filter(|e| !e.is_empty());
if let Some(e) = &email
&& tranquil_types::EmailAddress::new(e).is_err()
{
return Err(ApiError::InvalidEmail);
}
let email = match &email {
Some(e) => {
if e.len() > 254 {
return Err(ApiError::InvalidEmail);
}
if !tranquil_pds::api::validation::is_valid_email(e) {
return Err(ApiError::InvalidEmail);
}
Some(e.clone())
}
None => None,
};
let invite_registration =
check_registration_invite(&state, input.invite_code.as_deref()).await?;
@@ -1300,11 +1334,10 @@ pub async fn complete_registration(
}
let hostname = &tranquil_config::get().server.hostname;
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_notice(
if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_welcome(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
user_id.unwrap_or(uuid::Uuid::nil()),
tranquil_pds::comms::Notice::Welcome,
hostname,
)
.await
@@ -1341,7 +1374,7 @@ pub async fn complete_registration(
let verification_token = tranquil_pds::auth::verification_token::generate_signup_token(
&did,
verification_channel,
&target.id,
&verification_recipient,
);
let formatted_token =
tranquil_pds::auth::verification_token::format_token_for_display(&verification_token);
@@ -1349,7 +1382,8 @@ pub async fn complete_registration(
state.repos.user.as_ref(),
state.repos.infra.as_ref(),
uid,
&target,
verification_channel,
&verification_recipient,
&formatted_token,
hostname,
)
+2 -26
View File
@@ -763,7 +763,8 @@ impl From<crate::api::validation::HandleValidationError> for ApiError {
HandleValidationError::BannedWord => {
Self::InvalidHandle(Some("Inappropriate language in handle".to_string()))
}
HandleValidationError::UnusableHandleDomain => Self::InternalError(Some(e.to_string())),
HandleValidationError::UnusableHandleDomain
| HandleValidationError::NoHandleDomains => Self::InternalError(Some(e.to_string())),
_ => Self::InvalidHandle(Some(e.to_string())),
}
}
@@ -775,31 +776,6 @@ impl From<jacquard_common::types::string::AtStrError> for ApiError {
}
}
impl From<tranquil_types::InvalidEmailAddress> for ApiError {
fn from(_: tranquil_types::InvalidEmailAddress) -> Self {
Self::InvalidEmail
}
}
macro_rules! invalid_request_from {
($($err:ty),* $(,)?) => {
$(
impl From<$err> for ApiError {
fn from(e: $err) -> Self {
Self::InvalidRequest(e.to_string())
}
}
)*
};
}
invalid_request_from!(
tranquil_types::InvalidSignalUsername,
tranquil_types::InvalidTelegramUsername,
tranquil_types::InvalidDiscordUsername,
tranquil_types::InvalidRecipient,
);
impl From<crate::plc::PlcError> for ApiError {
fn from(e: crate::plc::PlcError) -> Self {
use crate::plc::PlcError;
+1 -2
View File
@@ -293,11 +293,10 @@ async fn proxy_handler(
return ApiError::InvalidRequest(format!("Invalid XRPC method: {}", method))
.into_response();
};
let scope_aud = format!("{}#{}", resolved.did, service_id);
if let Err(e) = crate::auth::scope_check::check_rpc_scope(
&auth_user.auth_source,
auth_user.scope.as_deref(),
&scope_aud,
&resolved.did,
&method_nsid,
) {
return e.into_response();
+188 -9
View File
@@ -1,11 +1,102 @@
use crate::types::Handle;
use std::fmt;
pub const MAX_EMAIL_LENGTH: usize = 254;
pub const MAX_LOCAL_PART_LENGTH: usize = 64;
pub const MAX_DOMAIN_LENGTH: usize = 253;
pub const MAX_DOMAIN_LABEL_LENGTH: usize = 63;
const EMAIL_LOCAL_SPECIAL_CHARS: &str = ".!#$%&'*+/=?^_`{|}~-";
pub const MIN_HANDLE_LENGTH: usize = 3;
pub const MAX_HANDLE_LENGTH: usize = 253;
pub const MAX_SERVICE_HANDLE_LOCAL_PART: usize = 18;
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EmailValidationError {
Empty,
TooLong,
MissingAtSign,
EmptyLocalPart,
LocalPartTooLong,
InvalidLocalPart,
EmptyDomain,
DomainTooLong,
MissingDomainDot,
InvalidDomainLabel,
}
impl fmt::Display for EmailValidationError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Empty => write!(f, "Email cannot be empty"),
Self::TooLong => write!(
f,
"Email exceeds maximum length of {} characters",
MAX_EMAIL_LENGTH
),
Self::MissingAtSign => write!(f, "Email must contain @"),
Self::EmptyLocalPart => write!(f, "Email local part cannot be empty"),
Self::LocalPartTooLong => write!(f, "Email local part exceeds maximum length"),
Self::InvalidLocalPart => write!(f, "Email local part contains invalid characters"),
Self::EmptyDomain => write!(f, "Email domain cannot be empty"),
Self::DomainTooLong => write!(f, "Email domain exceeds maximum length"),
Self::MissingDomainDot => write!(f, "Email domain must contain a dot"),
Self::InvalidDomainLabel => write!(f, "Email domain contains invalid label"),
}
}
}
impl std::error::Error for EmailValidationError {}
fn validate_email_detailed(email: &str) -> Result<(), EmailValidationError> {
if email.is_empty() {
return Err(EmailValidationError::Empty);
}
if email.len() > MAX_EMAIL_LENGTH {
return Err(EmailValidationError::TooLong);
}
let parts: Vec<&str> = email.rsplitn(2, '@').collect();
if parts.len() != 2 {
return Err(EmailValidationError::MissingAtSign);
}
let domain = parts[0];
let local = parts[1];
if local.is_empty() {
return Err(EmailValidationError::EmptyLocalPart);
}
if local.len() > MAX_LOCAL_PART_LENGTH {
return Err(EmailValidationError::LocalPartTooLong);
}
if local.starts_with('.') || local.ends_with('.') || local.contains("..") {
return Err(EmailValidationError::InvalidLocalPart);
}
if !local
.chars()
.all(|c| c.is_ascii_alphanumeric() || EMAIL_LOCAL_SPECIAL_CHARS.contains(c))
{
return Err(EmailValidationError::InvalidLocalPart);
}
if domain.is_empty() {
return Err(EmailValidationError::EmptyDomain);
}
if domain.len() > MAX_DOMAIN_LENGTH {
return Err(EmailValidationError::DomainTooLong);
}
if !domain.contains('.') {
return Err(EmailValidationError::MissingDomainDot);
}
if !domain.split('.').all(|label| {
!label.is_empty()
&& label.len() <= MAX_DOMAIN_LABEL_LENGTH
&& !label.starts_with('-')
&& !label.ends_with('-')
&& label.chars().all(|c| c.is_ascii_alphanumeric() || c == '-')
}) {
return Err(EmailValidationError::InvalidDomainLabel);
}
Ok(())
}
#[derive(Debug, PartialEq)]
pub enum HandleValidationError {
Empty,
@@ -20,6 +111,7 @@ pub enum HandleValidationError {
InvalidSyntax,
DisallowedTld,
UnusableHandleDomain,
NoHandleDomains,
}
impl std::fmt::Display for HandleValidationError {
@@ -51,6 +143,9 @@ impl std::fmt::Display for HandleValidationError {
f,
"This server's handle domain has a reserved TLD, so no handle under it is a valid atproto handle"
),
Self::NoHandleDomains => {
write!(f, "No handle domains are configured on this server")
}
}
}
}
@@ -120,14 +215,21 @@ pub fn validate_short_handle(handle: &str) -> Result<String, HandleValidationErr
}
pub fn resolve_handle_input(input: &str) -> Result<Handle, HandleValidationError> {
let domains = crate::handle::ServiceDomains::for_user_handles();
let split = domains.split_handle(input);
let available_domains = tranquil_config::get().server.available_user_domain_list();
let matched_domain = available_domains
.iter()
.filter(|d| input.ends_with(&format!(".{}", d)))
.max_by_key(|d| d.len());
if !input.contains('.') || split.is_some() {
let (short, domain) = split
.map(|(domain, short)| (short, domain))
.unwrap_or((input, domains.primary()));
let validated = validate_short_handle(short)?;
if !input.contains('.') || matched_domain.is_some() {
let handle_to_validate = match matched_domain {
Some(domain) => input.strip_suffix(&format!(".{}", domain)).unwrap_or(input),
None => input,
};
let validated = validate_short_handle(handle_to_validate)?;
let domain = matched_domain
.or_else(|| available_domains.first())
.ok_or(HandleValidationError::NoHandleDomains)?;
let handle = Handle::new(format!("{}.{}", validated, domain))
.map_err(|_| HandleValidationError::InvalidSyntax)?;
match handle.has_disallowed_tld() {
@@ -144,9 +246,11 @@ pub fn domain_forms_valid_handles(domain: &str) -> bool {
}
pub fn warn_unusable_handle_domains() {
crate::handle::ServiceDomains::for_user_handles()
tranquil_config::get()
.server
.user_handle_domain_list()
.iter()
.filter(|domain| !domain_forms_valid_handles(domain.as_str()))
.filter(|domain| !domain_forms_valid_handles(domain))
.for_each(|domain| {
tracing::error!(
domain = %domain,
@@ -213,6 +317,23 @@ pub fn validate_service_handle(
Ok(handle.to_lowercase())
}
pub fn is_valid_email(email: &str) -> bool {
validate_email_detailed(email.trim()).is_ok()
}
pub fn is_valid_telegram_username(username: &str) -> bool {
let clean = username.strip_prefix('@').unwrap_or(username);
(5..=32).contains(&clean.len()) && clean.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
}
pub fn is_valid_discord_username(username: &str) -> bool {
(2..=32).contains(&username.len())
&& username
.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_' || c == '.')
&& !username.contains("..")
}
#[cfg(test)]
mod tests {
use super::*;
@@ -379,4 +500,62 @@ mod tests {
Err(HandleValidationError::Reserved)
);
}
#[test]
fn test_valid_emails() {
assert!(is_valid_email("user@example.com"));
assert!(is_valid_email("user.name@example.com"));
assert!(is_valid_email("user+tag@example.com"));
assert!(is_valid_email("user@sub.example.com"));
assert!(is_valid_email("USER@EXAMPLE.COM"));
assert!(is_valid_email("user123@example123.com"));
assert!(is_valid_email("a@b.co"));
}
#[test]
fn test_invalid_emails() {
assert!(!is_valid_email(""));
assert!(!is_valid_email("user"));
assert!(!is_valid_email("user@"));
assert!(!is_valid_email("@example.com"));
assert!(!is_valid_email("user@example"));
assert!(!is_valid_email("user@@example.com"));
assert!(!is_valid_email("user@.example.com"));
assert!(!is_valid_email("user@example..com"));
assert!(!is_valid_email(".user@example.com"));
assert!(!is_valid_email("user.@example.com"));
assert!(!is_valid_email("user..name@example.com"));
assert!(!is_valid_email("user@-example.com"));
assert!(!is_valid_email("user@example-.com"));
}
#[test]
fn test_trimmed_whitespace() {
assert!(is_valid_email(" user@example.com "));
}
#[test]
fn test_valid_discord_usernames() {
assert!(is_valid_discord_username("ab"));
assert!(is_valid_discord_username("alice"));
assert!(is_valid_discord_username("user_name"));
assert!(is_valid_discord_username("user.name"));
assert!(is_valid_discord_username("user123"));
assert!(is_valid_discord_username("a_b.c_d"));
assert!(is_valid_discord_username(
"12345678901234567890123456789012"
));
}
#[test]
fn test_invalid_discord_usernames() {
assert!(!is_valid_discord_username(""));
assert!(!is_valid_discord_username("a"));
assert!(!is_valid_discord_username("Alice"));
assert!(!is_valid_discord_username("ALICE"));
assert!(!is_valid_discord_username("user-name"));
assert!(!is_valid_discord_username("user..name"));
assert!(!is_valid_discord_username("user name"));
assert!(!is_valid_discord_username(
"123456789012345678901234567890123"
));
}
}
+5 -4
View File
@@ -1,9 +1,10 @@
mod service;
pub use service::repo::Notice;
pub use service::{CommsService, VerificationTarget, recipient_for, repo as comms_repo};
pub use tranquil_comms::{
CommsChannel, CommsSender, CommsStatus, CommsType, DEFAULT_LOCALE, DiscordSender, EmailSender,
NotificationStrings, QueuedComms, SendError, SignalSender, TelegramSender, VALID_LOCALES,
format_message, get_strings, is_valid_phone_number, validate_locale,
NewComms, NotificationStrings, QueuedComms, SendError, SignalSender, TelegramSender,
VALID_LOCALES, format_message, get_strings, is_valid_phone_number, is_valid_signal_username,
validate_locale,
};
pub use service::{CommsService, repo as comms_repo, resolve_delivery_channel};
+377 -332
View File
@@ -7,13 +7,9 @@ use chrono::Utc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
use tranquil_comms::{
CommsChannel, CommsSender, CommsType, NotificationStrings, SendError, format_message,
get_strings,
CommsChannel, CommsSender, CommsType, NewComms, SendError, format_message, get_strings,
};
use tranquil_db_traits::{
DbError, InfraRepository, QueuedComms, Recipient, UserCommsPrefs, UserRepository,
};
use tranquil_types::{DiscordUserId, EmailAddress, SignalUsername, TelegramChatId};
use tranquil_db_traits::{InfraRepository, QueuedComms, UserCommsPrefs, UserRepository};
use uuid::Uuid;
pub struct CommsService {
@@ -51,6 +47,23 @@ impl CommsService {
self
}
pub async fn enqueue(&self, item: NewComms) -> Result<Uuid, tranquil_db_traits::DbError> {
let id = self
.infra_repo
.enqueue_comms(
Some(item.user_id),
item.channel,
item.comms_type,
&item.recipient,
item.subject.as_deref(),
&item.body,
item.metadata,
)
.await?;
debug!(comms_id = %id, "Comms enqueued");
Ok(id)
}
pub fn has_senders(&self) -> bool {
!self.senders.is_empty()
}
@@ -113,28 +126,8 @@ impl CommsService {
async fn process_item(&self, item: QueuedComms) {
let comms_id = item.id;
// Re-checking because there's been a trip into the DB and back, can't trust type -> string -> *maybe* type
let recipient = match tranquil_db_traits::Recipient::new(item.channel, &item.recipient) {
Ok(recipient) => recipient,
Err(e) => {
warn!(
comms_id = %comms_id,
error = %e,
"We marked comms item as permanently failed because its recipient is invalid"
);
if let Err(db_err) = self.mark_failed_permanent(comms_id, &e.to_string()).await {
error!(
comms_id = %comms_id,
error = %db_err,
"Failed to mark comms as failed"
);
}
return;
}
};
let result = match self.senders.get(&item.channel) {
Some(sender) => sender.send(&item, &recipient).await,
Some(sender) => sender.send(&item).await,
None => {
warn!(
comms_id = %comms_id,
@@ -196,234 +189,131 @@ impl CommsService {
}
}
// Think about the situation on Telegram and Discord where the user must message a given bot *first* in order to hydrate a chat ID into our system so that we can in fact send things.
// If we can think of a better way to simply error-out later, instead of falling back to email when say Telegram is in an aborted state, let's do that.
pub struct VerificationTarget {
pub id: String,
pub recipient: Recipient,
struct ResolvedRecipient {
channel: tranquil_db_traits::CommsChannel,
recipient: String,
}
impl VerificationTarget {
pub fn direct(recipient: Recipient) -> Self {
Self {
id: recipient.as_str().to_string(),
recipient,
}
}
pub fn resolve(
channel: CommsChannel,
id: &str,
fallback: Option<&str>,
) -> Result<Self, crate::api::error::ApiError> {
let direct = (!channel.verifies_via_bot())
.then(|| Recipient::new(channel, id))
.and_then(Result::ok);
let recipient = direct.map_or_else(|| fallback_recipient(fallback), Ok)?;
Ok(Self {
id: id.to_string(),
recipient,
})
}
pub fn resolve_delivery_channel(
prefs: &UserCommsPrefs,
channel: tranquil_db_traits::CommsChannel,
) -> tranquil_db_traits::CommsChannel {
resolve_recipient(prefs, channel).channel
}
fn fallback_recipient(fallback: Option<&str>) -> Result<Recipient, crate::api::error::ApiError> {
let raw = fallback
.map(str::trim)
.filter(|email| !email.is_empty())
.ok_or(crate::api::error::ApiError::InvalidRequest(
"Verification over this channel needs an email address. Message the bot first".into(),
))?;
EmailAddress::new(raw)
.map(Recipient::Email)
.map_err(|_| crate::api::error::ApiError::InvalidEmail)
}
pub fn recipient_for(prefs: &UserCommsPrefs, channel: CommsChannel) -> Option<Recipient> {
let fallback = || email_recipient(prefs);
fn resolve_recipient(
prefs: &UserCommsPrefs,
channel: tranquil_db_traits::CommsChannel,
) -> ResolvedRecipient {
let email_fallback = || ResolvedRecipient {
channel: tranquil_db_traits::CommsChannel::Email,
recipient: prefs.email.clone().unwrap_or_default(),
};
match channel {
CommsChannel::Email => fallback(),
CommsChannel::Telegram => prefs
tranquil_db_traits::CommsChannel::Email => email_fallback(),
tranquil_db_traits::CommsChannel::Telegram => prefs
.telegram_chat_id
.and_then(TelegramChatId::from_i64)
.map(Recipient::Telegram)
.or_else(fallback),
CommsChannel::Discord => prefs
.map(|id| ResolvedRecipient {
channel,
recipient: id.to_string(),
})
.unwrap_or_else(email_fallback),
tranquil_db_traits::CommsChannel::Discord => prefs
.discord_id
.as_deref()
.and_then(|id| DiscordUserId::new(id).ok())
.map(Recipient::Discord)
.or_else(fallback),
CommsChannel::Signal => prefs
.as_ref()
.filter(|id| !id.is_empty())
.map(|id| ResolvedRecipient {
channel,
recipient: id.clone(),
})
.unwrap_or_else(email_fallback),
tranquil_db_traits::CommsChannel::Signal => prefs
.signal_username
.as_deref()
.and_then(|name| SignalUsername::new(name).ok())
.map(Recipient::Signal)
.or_else(fallback),
.as_ref()
.filter(|n| !n.is_empty())
.map(|n| ResolvedRecipient {
channel,
recipient: n.clone(),
})
.unwrap_or_else(email_fallback),
}
}
fn email_recipient(prefs: &UserCommsPrefs) -> Option<Recipient> {
prefs
.email
.as_deref()
.and_then(|email| EmailAddress::new(email).ok())
.map(Recipient::Email)
}
pub mod repo {
use super::*;
use tranquil_db_traits::DbError;
pub enum Notice<'a> {
Welcome,
PasswordReset { code: &'a str },
TwoFactorCode { code: &'a str },
AccountDeletion { code: &'a str },
PlcOperation { token: &'a str },
PasskeyRecovery { url: &'a str },
ShortTokenEmail { token: &'a str },
LegacyLoginAlert { channel: CommsChannel, ip: &'a str },
}
impl Notice<'_> {
fn comms_type(&self) -> CommsType {
match self {
Self::Welcome => CommsType::Welcome,
Self::PasswordReset { .. } => CommsType::PasswordReset,
Self::TwoFactorCode { .. } => CommsType::TwoFactorCode,
Self::AccountDeletion { .. } => CommsType::AccountDeletion,
Self::PlcOperation { .. } => CommsType::PlcOperation,
Self::PasskeyRecovery { .. } => CommsType::PasskeyRecovery,
Self::ShortTokenEmail { .. } => CommsType::EmailUpdate,
Self::LegacyLoginAlert { .. } => CommsType::LegacyLoginAlert,
}
}
// Yes yes I know, hardcoded, non-email-based accounts will have already bailed by now, don't worry. Emails are not special.
fn channel(&self) -> Option<CommsChannel> {
match self {
Self::ShortTokenEmail { .. } => Some(CommsChannel::Email),
Self::LegacyLoginAlert { channel, .. } => Some(*channel),
_ => None,
}
}
fn subject(&self, strings: &NotificationStrings) -> &'static str {
match self {
Self::Welcome => strings.welcome_subject,
Self::PasswordReset { .. } => strings.password_reset_subject,
Self::TwoFactorCode { .. } => strings.two_factor_code_subject,
Self::AccountDeletion { .. } => strings.account_deletion_subject,
Self::PlcOperation { .. } => strings.plc_operation_subject,
Self::PasskeyRecovery { .. } => strings.passkey_recovery_subject,
Self::ShortTokenEmail { .. } => strings.email_update_subject,
Self::LegacyLoginAlert { .. } => strings.legacy_login_subject,
}
}
fn body(&self, strings: &NotificationStrings, handle: &str, hostname: &str) -> String {
match self {
Self::Welcome => format_message(
strings.welcome_body,
&[("hostname", hostname), ("handle", handle)],
),
Self::PasswordReset { code } => format_message(
strings.password_reset_body,
&[("handle", handle), ("code", code)],
),
Self::TwoFactorCode { code } => format_message(
strings.two_factor_code_body,
&[("handle", handle), ("code", code)],
),
Self::AccountDeletion { code } => format_message(
strings.account_deletion_body,
&[("handle", handle), ("code", code)],
),
Self::PlcOperation { token } => format_message(
strings.plc_operation_body,
&[("handle", handle), ("token", token)],
),
Self::PasskeyRecovery { url } => format_message(
strings.passkey_recovery_body,
&[("handle", handle), ("url", url)],
),
Self::ShortTokenEmail { token } => {
let verify_page = format!("https://{hostname}/app/settings");
format_message(
strings.short_token_body,
&[
("handle", handle),
("code", token),
("verify_page", &verify_page),
],
)
}
Self::LegacyLoginAlert { ip, .. } => {
let timestamp = Utc::now().format("%Y-%m-%d %H:%M:%S UTC").to_string();
format_message(
strings.legacy_login_body,
&[
("handle", handle),
("timestamp", &timestamp),
("ip", ip),
("hostname", hostname),
],
)
}
}
}
}
pub async fn enqueue_notice(
pub async fn enqueue_welcome(
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
notice: Notice<'_>,
hostname: &str,
) -> Result<Option<Uuid>, DbError> {
) -> Result<Uuid, DbError> {
let prefs = user_repo
.get_comms_prefs(user_id)
.await?
.ok_or(DbError::NotFound)?;
let channel = notice.channel().unwrap_or(prefs.preferred_channel);
let Some(recipient) = recipient_for(&prefs, channel) else {
warn!(
user_id = %user_id,
channel = ?channel,
"We skipped queuing this notice because the account doesn't have a valid recipient"
);
return Ok(None);
};
let strings = get_strings(locale_of(&prefs));
let subject = format_message(notice.subject(strings), &[("hostname", hostname)]);
let body = notice.body(strings, prefs.handle.as_str(), hostname);
let strings = get_strings(prefs.preferred_locale.as_deref().unwrap_or("en"));
let body = format_message(
strings.welcome_body,
&[("hostname", hostname), ("handle", &prefs.handle)],
);
let subject = format_message(strings.welcome_subject, &[("hostname", hostname)]);
let resolved = resolve_recipient(&prefs, prefs.preferred_channel);
infra_repo
.enqueue_comms(
Some(user_id),
&recipient,
notice.comms_type(),
resolved.channel,
CommsType::Welcome,
&resolved.recipient,
Some(&subject),
&body,
None,
)
.await
.map(Some)
}
fn locale_of(prefs: &UserCommsPrefs) -> &str {
prefs.preferred_locale.as_deref().unwrap_or("en")
pub async fn enqueue_password_reset(
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
code: &str,
hostname: &str,
) -> Result<Uuid, DbError> {
let prefs = user_repo
.get_comms_prefs(user_id)
.await?
.ok_or(DbError::NotFound)?;
let strings = get_strings(prefs.preferred_locale.as_deref().unwrap_or("en"));
let body = format_message(
strings.password_reset_body,
&[("handle", &prefs.handle), ("code", code)],
);
let subject = format_message(strings.password_reset_subject, &[("hostname", hostname)]);
let resolved = resolve_recipient(&prefs, prefs.preferred_channel);
infra_repo
.enqueue_comms(
Some(user_id),
resolved.channel,
CommsType::PasswordReset,
&resolved.recipient,
Some(&subject),
&body,
None,
)
.await
}
pub async fn enqueue_email_update(
infra_repo: &dyn InfraRepository,
user_id: Uuid,
new_email: &EmailAddress,
new_email: &str,
handle: &crate::types::Handle,
code: &str,
hostname: &str,
) -> Result<Uuid, DbError> {
let strings = get_strings("en");
let encoded_email = urlencoding::encode(new_email.as_str());
let encoded_email = urlencoding::encode(new_email);
let encoded_token = urlencoding::encode(code);
let verify_page = format!("https://{}/app/verify", hostname);
let verify_link = format!(
@@ -443,8 +333,186 @@ pub mod repo {
infra_repo
.enqueue_comms(
Some(user_id),
&Recipient::Email(new_email.clone()),
tranquil_db_traits::CommsChannel::Email,
CommsType::EmailUpdate,
new_email,
Some(&subject),
&body,
None,
)
.await
}
pub async fn enqueue_email_update_token(
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
raw_token: &str,
display_code: &str,
hostname: &str,
) -> Result<Uuid, DbError> {
let prefs = user_repo
.get_comms_prefs(user_id)
.await?
.ok_or(DbError::NotFound)?;
let strings = get_strings(prefs.preferred_locale.as_deref().unwrap_or("en"));
let current_email = prefs.email.unwrap_or_default();
let verify_page = format!("https://{}/app/settings", hostname);
let verify_link = format!(
"https://{}/xrpc/_account.authorizeEmailUpdate?token={}",
hostname,
urlencoding::encode(raw_token)
);
let body = format_message(
strings.email_update_body,
&[
("handle", &prefs.handle),
("code", display_code),
("verify_page", &verify_page),
("verify_link", &verify_link),
],
);
let subject = format_message(strings.email_update_subject, &[("hostname", hostname)]);
infra_repo
.enqueue_comms(
Some(user_id),
tranquil_db_traits::CommsChannel::Email,
CommsType::EmailUpdate,
&current_email,
Some(&subject),
&body,
None,
)
.await
}
pub async fn enqueue_short_token_email(
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
token: &str,
hostname: &str,
) -> Result<Uuid, DbError> {
let prefs = user_repo
.get_comms_prefs(user_id)
.await?
.ok_or(DbError::NotFound)?;
let strings = get_strings(prefs.preferred_locale.as_deref().unwrap_or("en"));
let current_email = prefs.email.clone().unwrap_or_default();
let subject_template = strings.email_update_subject;
let body_template = strings.short_token_body;
let comms_type = CommsType::EmailUpdate;
let verify_page = format!("https://{}/app/settings", hostname);
let body = format_message(
body_template,
&[
("handle", &prefs.handle),
("code", token),
("verify_page", &verify_page),
],
);
let subject = format_message(subject_template, &[("hostname", hostname)]);
infra_repo
.enqueue_comms(
Some(user_id),
tranquil_db_traits::CommsChannel::Email,
comms_type,
&current_email,
Some(&subject),
&body,
None,
)
.await
}
pub async fn enqueue_account_deletion(
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
code: &str,
hostname: &str,
) -> Result<Uuid, DbError> {
let prefs = user_repo
.get_comms_prefs(user_id)
.await?
.ok_or(DbError::NotFound)?;
let strings = get_strings(prefs.preferred_locale.as_deref().unwrap_or("en"));
let body = format_message(
strings.account_deletion_body,
&[("handle", &prefs.handle), ("code", code)],
);
let subject = format_message(strings.account_deletion_subject, &[("hostname", hostname)]);
let resolved = resolve_recipient(&prefs, prefs.preferred_channel);
infra_repo
.enqueue_comms(
Some(user_id),
resolved.channel,
CommsType::AccountDeletion,
&resolved.recipient,
Some(&subject),
&body,
None,
)
.await
}
pub async fn enqueue_plc_operation(
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
token: &str,
hostname: &str,
) -> Result<Uuid, DbError> {
let prefs = user_repo
.get_comms_prefs(user_id)
.await?
.ok_or(DbError::NotFound)?;
let strings = get_strings(prefs.preferred_locale.as_deref().unwrap_or("en"));
let body = format_message(
strings.plc_operation_body,
&[("handle", &prefs.handle), ("token", token)],
);
let subject = format_message(strings.plc_operation_subject, &[("hostname", hostname)]);
let resolved = resolve_recipient(&prefs, prefs.preferred_channel);
infra_repo
.enqueue_comms(
Some(user_id),
resolved.channel,
CommsType::PlcOperation,
&resolved.recipient,
Some(&subject),
&body,
None,
)
.await
}
pub async fn enqueue_passkey_recovery(
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
recovery_url: &str,
hostname: &str,
) -> Result<Uuid, DbError> {
let prefs = user_repo
.get_comms_prefs(user_id)
.await?
.ok_or(DbError::NotFound)?;
let strings = get_strings(prefs.preferred_locale.as_deref().unwrap_or("en"));
let body = format_message(
strings.passkey_recovery_body,
&[("handle", &prefs.handle), ("url", recovery_url)],
);
let subject = format_message(strings.passkey_recovery_subject, &[("hostname", hostname)]);
let resolved = resolve_recipient(&prefs, prefs.preferred_channel);
infra_repo
.enqueue_comms(
Some(user_id),
resolved.channel,
CommsType::PasskeyRecovery,
&resolved.recipient,
Some(&subject),
&body,
None,
@@ -456,7 +524,8 @@ pub mod repo {
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
target: &VerificationTarget,
channel: tranquil_db_traits::CommsChannel,
recipient: &str,
token: &str,
hostname: &str,
) -> Result<Uuid, DbError> {
@@ -464,13 +533,13 @@ pub mod repo {
.get_comms_prefs(user_id)
.await?
.ok_or(DbError::NotFound)?;
let strings = get_strings(locale_of(&prefs));
let encoded_id = urlencoding::encode(&target.id);
let strings = get_strings(prefs.preferred_locale.as_deref().unwrap_or("en"));
let encoded_recipient = urlencoding::encode(recipient);
let encoded_token = urlencoding::encode(token);
let verify_page = format!("https://{}/app/verify", hostname);
let verify_link = format!(
"https://{}/app/verify?token={}&identifier={}",
hostname, encoded_token, encoded_id
hostname, encoded_token, encoded_recipient
);
let body = format_message(
strings.migration_verification_body,
@@ -488,8 +557,9 @@ pub mod repo {
infra_repo
.enqueue_comms(
Some(user_id),
&target.recipient,
channel,
CommsType::MigrationVerification,
recipient,
Some(&subject),
&body,
None,
@@ -501,10 +571,12 @@ pub mod repo {
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
target: &VerificationTarget,
channel: tranquil_db_traits::CommsChannel,
recipient: &str,
code: &str,
hostname: &str,
) -> Result<Uuid, DbError> {
let comms_channel = channel;
let prefs = match user_repo.get_comms_prefs(user_id).await {
Ok(p) => p,
Err(e) => {
@@ -512,14 +584,17 @@ pub mod repo {
None
}
};
let locale = prefs.as_ref().map(locale_of).unwrap_or("en");
let locale = prefs
.as_ref()
.and_then(|p| p.preferred_locale.as_deref())
.unwrap_or("en");
let strings = get_strings(locale);
let encoded_token = urlencoding::encode(code);
let encoded_id = urlencoding::encode(&target.id);
let encoded_recipient = urlencoding::encode(recipient);
let verify_page = format!("https://{}/app/verify", hostname);
let verify_link = format!(
"https://{}/app/verify?token={}&identifier={}",
hostname, encoded_token, encoded_id
hostname, encoded_token, encoded_recipient
);
let body = format_message(
strings.signup_verification_body,
@@ -537,8 +612,80 @@ pub mod repo {
infra_repo
.enqueue_comms(
Some(user_id),
&target.recipient,
comms_channel,
CommsType::EmailVerification,
recipient,
Some(&subject),
&body,
None,
)
.await
}
pub async fn enqueue_2fa_code(
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
code: &str,
hostname: &str,
) -> Result<Uuid, DbError> {
let prefs = user_repo
.get_comms_prefs(user_id)
.await?
.ok_or(DbError::NotFound)?;
let strings = get_strings(prefs.preferred_locale.as_deref().unwrap_or("en"));
let body = format_message(
strings.two_factor_code_body,
&[("handle", &prefs.handle), ("code", code)],
);
let subject = format_message(strings.two_factor_code_subject, &[("hostname", hostname)]);
let resolved = resolve_recipient(&prefs, prefs.preferred_channel);
infra_repo
.enqueue_comms(
Some(user_id),
resolved.channel,
CommsType::TwoFactorCode,
&resolved.recipient,
Some(&subject),
&body,
None,
)
.await
}
pub async fn enqueue_legacy_login(
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
hostname: &str,
client_ip: &str,
channel: tranquil_db_traits::CommsChannel,
) -> Result<Uuid, DbError> {
let prefs = user_repo
.get_comms_prefs(user_id)
.await?
.ok_or(DbError::NotFound)?;
let strings = get_strings(prefs.preferred_locale.as_deref().unwrap_or("en"));
let timestamp = chrono::Utc::now()
.format("%Y-%m-%d %H:%M:%S UTC")
.to_string();
let body = format_message(
strings.legacy_login_body,
&[
("handle", &prefs.handle),
("timestamp", &timestamp),
("ip", client_ip),
("hostname", hostname),
],
);
let subject = format_message(strings.legacy_login_subject, &[("hostname", hostname)]);
let resolved = resolve_recipient(&prefs, channel);
infra_repo
.enqueue_comms(
Some(user_id),
resolved.channel,
CommsType::LegacyLoginAlert,
&resolved.recipient,
Some(&subject),
&body,
None,
@@ -550,19 +697,20 @@ pub mod repo {
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
recipient: &Recipient,
channel: tranquil_db_traits::CommsChannel,
recipient: &str,
hostname: &str,
) -> Result<Uuid, DbError> {
let prefs = user_repo
.get_comms_prefs(user_id)
.await?
.ok_or(DbError::NotFound)?;
let strings = get_strings(locale_of(&prefs));
let strings = get_strings(prefs.preferred_locale.as_deref().unwrap_or("en"));
let body = format_message(
strings.channel_verified_body,
&[
("handle", &prefs.handle),
("channel", recipient.channel().display_name()),
("channel", channel.display_name()),
("hostname", hostname),
],
);
@@ -570,116 +718,13 @@ pub mod repo {
infra_repo
.enqueue_comms(
Some(user_id),
recipient,
channel,
CommsType::ChannelVerified,
recipient,
Some(&subject),
&body,
None,
)
.await
}
pub async fn try_channel_verified_notice(
user_repo: &dyn UserRepository,
infra_repo: &dyn InfraRepository,
user_id: Uuid,
recipient: &Recipient,
hostname: &str,
) {
if let Err(e) =
enqueue_channel_verified(user_repo, infra_repo, user_id, recipient, hostname).await
{
warn!(error = %e, "Failed to enqueue channel verified notification");
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn bot_channel_recipients_fall_back_to_email() {
let telegram =
VerificationTarget::resolve(CommsChannel::Telegram, "123456789", Some("user@jola.dev"))
.unwrap();
assert_eq!(telegram.recipient.channel(), CommsChannel::Email);
let discord = VerificationTarget::resolve(
CommsChannel::Discord,
"274656283714826240",
Some("user@jola.dev"),
)
.unwrap();
assert_eq!(discord.recipient.channel(), CommsChannel::Email);
}
#[test]
fn resolution_keeps_id_for_bot_channels() {
let target =
VerificationTarget::resolve(CommsChannel::Telegram, "oys_01", Some("user@jola.dev"))
.unwrap();
assert_eq!(target.id, "oys_01");
assert_eq!(target.recipient.channel(), CommsChannel::Email);
}
#[test]
fn direct_channels_parse_id() {
let email = VerificationTarget::resolve(CommsChannel::Email, "user@nel.pet", None).unwrap();
assert_eq!(email.recipient.as_str(), "user@nel.pet");
let signal = VerificationTarget::resolve(CommsChannel::Signal, "oys.01", None).unwrap();
assert_eq!(signal.recipient.channel(), CommsChannel::Signal);
}
#[test]
fn signal_falls_back_when_id_isnt_username() {
let target =
VerificationTarget::resolve(CommsChannel::Signal, "oys", Some("user@jola.dev"))
.unwrap();
assert_eq!(target.recipient.channel(), CommsChannel::Email);
}
#[test]
fn resolve_fails_without_fallback() {
assert!(VerificationTarget::resolve(CommsChannel::Telegram, "oys_01", None).is_err());
assert!(VerificationTarget::resolve(CommsChannel::Signal, "oys", None).is_err());
}
}
#[cfg(test)]
mod recipient_for_tests {
use super::*;
fn undeliverable_prefs() -> UserCommsPrefs {
UserCommsPrefs {
email: None,
handle: "oys.nel.pet".parse().unwrap(),
preferred_channel: CommsChannel::Telegram,
preferred_locale: None,
telegram_chat_id: None,
discord_id: None,
signal_username: None,
}
}
#[test]
fn undeliverable_prefs_resolve_to_none_on_every_channel() {
let prefs = undeliverable_prefs();
assert_eq!(recipient_for(&prefs, CommsChannel::Telegram), None);
assert_eq!(recipient_for(&prefs, CommsChannel::Email), None);
}
#[test]
fn zero_chat_id_falls_back_to_email() {
let prefs = UserCommsPrefs {
telegram_chat_id: Some(0),
email: Some("oys@jola.dev".into()),
..undeliverable_prefs()
};
assert_eq!(
recipient_for(&prefs, CommsChannel::Telegram),
Some(Recipient::Email(EmailAddress::new("oys@jola.dev").unwrap()))
);
}
}
+16 -133
View File
@@ -3,16 +3,8 @@ pub mod reserved;
use crate::types::{Did, Handle};
use hickory_resolver::TokioAsyncResolver;
use hickory_resolver::config::{ResolverConfig, ResolverOpts};
use std::sync::LazyLock;
use thiserror::Error;
pub use tranquil_types::Domain;
static HOSTNAME_DOMAIN: LazyLock<Domain> = LazyLock::new(|| {
Domain::new(tranquil_config::get().server.hostname_without_port())
.expect("server.hostname is validated at config load")
});
#[derive(Error, Debug)]
pub enum HandleResolutionError {
#[error("DNS lookup failed: {0}")]
@@ -93,137 +85,28 @@ pub async fn verify_handle_ownership(
}
}
#[derive(Clone, Copy)]
pub struct ServiceDomains<'a> {
user_domains: &'a [Domain],
hostname: &'a Domain,
serve_hostname: bool,
}
impl ServiceDomains<'static> {
pub fn for_user_handles() -> Self {
Self::from_config(false)
}
pub fn served() -> Self {
Self::from_config(true)
}
fn from_config(serve_hostname: bool) -> Self {
let server = &tranquil_config::get().server;
Self {
user_domains: server.user_handle_domains.as_deref().unwrap_or_default(),
hostname: &HOSTNAME_DOMAIN,
serve_hostname,
}
}
}
impl<'a> ServiceDomains<'a> {
pub fn iter(&self) -> impl Iterator<Item = &'a Domain> {
let hostname = (self.serve_hostname || self.user_domains.is_empty())
.then_some(self.hostname)
.filter(|h| !self.user_domains.contains(h));
self.user_domains.iter().chain(hostname)
}
pub fn primary(&self) -> &'a Domain {
self.user_domains.first().unwrap_or(self.hostname)
}
pub fn contains(&self, name: &str) -> bool {
self.iter().any(|d| d.eq_name(name))
}
pub fn split_handle<'h>(&self, handle: &'h str) -> Option<(&'a Domain, &'h str)> {
self.iter()
.filter_map(|d| d.strip_from(handle).map(|short| (d, short)))
.max_by_key(|(d, _)| d.as_str().len())
pub fn is_service_domain_handle(handle: &str, hostname: &str) -> bool {
if !handle.contains('.') {
return true;
}
let service_domains = tranquil_config::try_get()
.map(|c| c.server.user_handle_domain_list())
.unwrap_or_else(|| vec![hostname.to_string()]);
service_domains
.iter()
.any(|domain| handle.ends_with(&format!(".{}", domain)) || handle == domain)
}
#[cfg(test)]
mod tests {
use super::{Domain, ServiceDomains};
use std::sync::LazyLock;
static HOST: LazyLock<Domain> = LazyLock::new(|| "pds.oyster.cafe".parse().unwrap());
fn domains(user_domains: &[Domain], serve_hostname: bool) -> ServiceDomains<'_> {
ServiceDomains {
user_domains,
hostname: &HOST,
serve_hostname,
}
}
fn owned(list: &[&str]) -> Vec<Domain> {
list.iter().map(|d| d.parse().unwrap()).collect()
}
use super::*;
#[test]
fn thostname_until_domains_are_configured() {
assert!(domains(&[], false).contains("pds.oyster.cafe"));
assert_eq!(domains(&[], false).primary(), "pds.oyster.cafe");
let configured = owned(&["oyster.cafe"]);
assert!(!domains(&configured, false).contains("pds.oyster.cafe"));
assert!(domains(&configured, false).contains("oyster.cafe"));
}
#[test]
fn served_set_covers_hostname_and_handle_domains() {
let configured = owned(&["oyster.cafe"]);
assert!(domains(&configured, true).contains("pds.oyster.cafe"));
assert!(domains(&configured, true).contains("oyster.cafe"));
}
#[test]
fn hostname_in_list_is_yielded_once() {
let configured = owned(&["pds.oyster.cafe", "oyster.cafe"]);
let served: Vec<&str> = domains(&configured, true)
.iter()
.map(Domain::as_str)
.collect();
assert_eq!(served, ["pds.oyster.cafe", "oyster.cafe"]);
let configured = owned(&["PDS.Oyster.Cafe"]);
let served: Vec<&str> = domains(&configured, true)
.iter()
.map(Domain::as_str)
.collect();
assert_eq!(served, ["pds.oyster.cafe"]);
}
#[test]
fn matching_case_insensitive() {
let configured = owned(&["oyster.cafe"]);
assert!(domains(&configured, false).contains("Oyster.Cafe"));
let (domain, short) = domains(&configured, false)
.split_handle("NEL.OYSTER.CAFE")
.unwrap();
assert_eq!(domain, "oyster.cafe");
assert_eq!(short, "NEL");
}
#[test]
fn longest_matching_domain_wins() {
let configured = owned(&["oyster.cafe", "pets.oyster.cafe"]);
let (domain, short) = domains(&configured, false)
.split_handle("nel.pets.oyster.cafe")
.unwrap();
assert_eq!(domain, "pets.oyster.cafe");
assert_eq!(short, "nel");
}
#[test]
fn split_handle_requires_a_dot() {
let configured = owned(&["oyster.cafe"]);
assert_eq!(
domains(&configured, false).split_handle("oyster.cafe"),
None
);
assert_eq!(
domains(&configured, false).split_handle("notoyster.cafe"),
None
);
fn test_is_service_domain_handle() {
assert!(is_service_domain_handle("nel.oyster.cafe", "oyster.cafe"));
assert!(is_service_domain_handle("oyster.cafe", "oyster.cafe"));
assert!(is_service_domain_handle("myhandle", "oyster.cafe"));
assert!(!is_service_domain_handle("lyna.nel.pet", "oyster.cafe"));
assert!(!is_service_domain_handle("myhandle.xyz", "oyster.cafe"));
}
}
@@ -3,8 +3,8 @@ use crate::cache_keys::permission_set_key;
use serde::{Deserialize, Serialize};
use std::time::Duration;
use tranquil_scopes::{
ExpansionOutcome, FailedSet, ParsedScope, RejectedScope, ResolveFailure, ResolvedSetGroup,
ScopeExpansionError, ScopeRejection, fetch_and_expand, parse_include_scope, parse_scope,
ExpansionOutcome, FailedSet, ResolveFailure, ResolvedSetGroup, ScopeExpansionError,
fetch_and_expand, parse_include_scope,
};
use tranquil_types::Nsid;
@@ -32,12 +32,6 @@ pub async fn expand_scopes(cache: &dyn Cache, scope_string: &str) -> ExpansionOu
let mut outcome = ExpansionOutcome::default();
for tok in scope_string.split_whitespace() {
match tok.strip_prefix("include:") {
None if matches!(parse_scope(tok), ParsedScope::Unknown(_)) => {
outcome.rejected.push(RejectedScope {
scope: tok.to_string(),
reason: ScopeRejection::Unrecognized,
})
}
None => outcome.passthrough.push(tok.to_string()),
Some(rest) => {
let (nsid, aud) = parse_include_scope(rest);
@@ -242,28 +236,4 @@ mod tests {
assert_eq!(out.failures.len(), 1);
assert_eq!(out.failures[0].given_nsid, "nonexistent.fake.permissionSet");
}
#[tokio::test]
async fn unrecognized_scopes_are_rejected_not_passed_through() {
let cache = MemoryCache::new();
let out = expand_scopes(&cache, "atproto chat").await;
assert_eq!(out.passthrough, vec!["atproto".to_string()]);
assert!(
!out.flat_scopes().iter().any(|s| s == "chat"),
"an unrecognized scope must never reach the effective scope set"
);
assert_eq!(out.rejected.len(), 1);
assert_eq!(out.rejected[0].scope, "chat");
assert_eq!(out.rejected[0].reason, ScopeRejection::Unrecognized);
}
#[tokio::test]
async fn structurally_invalid_granular_scopes_are_rejected() {
let cache = MemoryCache::new();
let out = expand_scopes(&cache, "atproto rpc:*?aud=*").await;
assert_eq!(out.passthrough, vec!["atproto".to_string()]);
assert_eq!(out.rejected.len(), 1);
assert_eq!(out.rejected[0].scope, "rpc:*?aud=*");
assert_eq!(out.rejected[0].reason, ScopeRejection::Unrecognized);
}
}
+1 -1
View File
@@ -188,7 +188,7 @@ impl PlcClient {
let timeout_secs = cfg.map_or(10, |c| c.plc.timeout_secs);
let connect_timeout_secs = cfg.map_or(5, |c| c.plc.connect_timeout_secs);
let fetch_policy = tranquil_types::ReachPolicy::from_private_fetch(
cfg.is_some_and(|c| c.server.allow_private_fetch),
cfg.map_or(false, |c| c.server.allow_private_fetch),
);
let client = Client::builder()
.timeout(Duration::from_secs(timeout_secs))
+1 -165
View File
@@ -3,16 +3,13 @@ use cid::Cid;
use ipld_core::ipld::Ipld;
use jacquard_repo::commit::Commit;
use jacquard_repo::storage::BlockStore;
use std::collections::BTreeSet;
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::interval;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
use tranquil_db_traits::{
BlobRepository, InfraRepository, RepoRepository, SsoRepository, UserRepository,
};
use tranquil_db_traits::{BlobRepository, RepoRepository, SsoRepository, UserRepository};
use tranquil_store::blockstore::CidBytes;
use tranquil_store::bloom::BloomFilter;
use tranquil_types::{AtUri, CidLink, Did};
@@ -310,83 +307,6 @@ async fn process_record_blobs(
Ok((user_id, did, blob_refs_found))
}
const OWNERSHIP_CHUNK_SIZE: usize = 500;
async fn process_blob_ownership(
repo_repo: &dyn RepoRepository,
blob_repo: &dyn BlobRepository,
block_store: &AnyBlockStore,
user_id: uuid::Uuid,
did: Did,
) -> Result<(uuid::Uuid, Did, u64), (uuid::Uuid, &'static str)> {
let records = repo_repo
.get_all_records(user_id)
.await
.map_err(|_| (user_id, "failed to fetch records"))?;
let mut cids: BTreeSet<CidLink> = BTreeSet::new();
for chunk in records.chunks(OWNERSHIP_CHUNK_SIZE) {
futures::future::join_all(chunk.iter().map(|record| async move {
let uri = format!("{}/{}", record.collection.as_str(), record.rkey.as_str());
let cid = match Cid::from_str(record.record_cid.as_str()) {
Ok(c) => c,
Err(e) => {
warn!(user_id = %user_id, record = %uri, error = %e, "skipping record with unparseable CID");
return None;
}
};
let block_bytes = match block_store.get(&cid).await {
Ok(Some(b)) => b,
Ok(None) => {
warn!(user_id = %user_id, record = %uri, "skipping record where block is missing in the block store");
return None;
}
Err(e) => {
warn!(user_id = %user_id, record = %uri, error = %e, "skipping record because block couldn't be read");
return None;
}
};
let record_ipld: Ipld = match serde_ipld_dagcbor::from_slice(&block_bytes) {
Ok(v) => v,
Err(e) => {
warn!(user_id = %user_id, record = %uri, error = %e, "skipping record because block couldn't be decoded");
return None;
}
};
Some(
crate::sync::import::find_blob_refs_ipld(&record_ipld, 0)
.into_iter()
.map(|blob_ref| blob_ref.cid)
.collect::<Vec<_>>(),
)
}))
.await
.into_iter()
.flatten()
.flatten()
.for_each(|cid| {
cids.insert(cid);
});
}
let mut granted = 0u64;
for cid in &cids {
if blob_repo
.ensure_blob_ownership(user_id, cid)
.await
.map_err(|_| (user_id, "failed to grant ownership"))?
{
granted += 1;
}
}
Ok((user_id, did, granted))
}
pub async fn backfill_record_blobs(repo_repo: Arc<dyn RepoRepository>, block_store: AnyBlockStore) {
let users_needing_backfill = match repo_repo.get_users_needing_record_blobs_backfill(100).await
{
@@ -432,90 +352,6 @@ pub async fn backfill_record_blobs(repo_repo: Arc<dyn RepoRepository>, block_sto
info!(success, failed, "Completed record_blobs backfill");
}
const BLOB_OWNERSHIP_BACKFILL_KEY: &str = "blob_ownership_backfilled";
pub async fn backfill_blob_ownership(
infra_repo: Arc<dyn InfraRepository>,
repo_repo: Arc<dyn RepoRepository>,
blob_repo: Arc<dyn BlobRepository>,
block_store: AnyBlockStore,
) {
match infra_repo
.get_server_config(BLOB_OWNERSHIP_BACKFILL_KEY)
.await
{
Ok(Some(_)) => return,
Ok(None) => {}
Err(e) => {
error!("Failed to read blob ownership backfill marker: {:?}", e);
return;
}
}
let repos = match repo_repo.get_all_repo_identities().await {
Ok(rows) => rows,
Err(e) => {
error!("Failed to query repos for blob ownership backfill: {:?}", e);
return;
}
};
if repos.is_empty() {
debug!("No repos need blob ownership backfill",);
return;
}
info!(
count = repos.len(),
"Backfilling blob ownership for existing repos"
);
let mut success = 0;
let mut failed = 0;
for chunk in repos.chunks(OWNERSHIP_CHUNK_SIZE) {
let results = futures::future::join_all(chunk.iter().map(|repo| {
let repo_repo = repo_repo.clone();
let blob_repo = blob_repo.clone();
let block_store = block_store.clone();
async move {
process_blob_ownership(
repo_repo.as_ref(),
blob_repo.as_ref(),
&block_store,
repo.user_id,
repo.did.clone(),
)
.await
}
}))
.await;
results.iter().for_each(|r| match r {
Ok((user_id, did, granted)) => {
if *granted > 0 {
info!(user_id = %user_id, did = %did, granted = granted, "Granted blob ownership");
}
success += 1;
}
Err((user_id, reason)) => {
warn!(user_id = %user_id, reason = reason, "Failed to backfill blob ownership");
failed += 1;
}
});
}
if let Err(e) = infra_repo
.upsert_server_config(BLOB_OWNERSHIP_BACKFILL_KEY, "1")
.await
{
error!("Failed to set blob ownership backfill marker: {:?}", e);
}
info!(success, failed, "Completed blob ownership backfill");
}
#[allow(clippy::too_many_arguments)]
pub async fn start_scheduled_tasks(
user_repo: Arc<dyn UserRepository>,
+1 -1
View File
@@ -251,7 +251,7 @@ impl AppState {
let mut state = match cfg.storage.repo_backend() {
tranquil_config::RepoBackend::TranquilStore => {
tracing::info!("tranquil-store repo backend active. Thank you so much for believing in us!");
tracing::info!("tranquil-store repo backend active. EXPERIMENTAL!");
Self::from_store(shutdown).await
}
tranquil_config::RepoBackend::Postgres => {
@@ -1,54 +1,8 @@
mod common;
use common::{base_url, client, create_account_and_login, get_test_repos, user_id_of};
use common::{base_url, client, create_account_and_login, get_test_repos};
use serde_json::{Value, json};
use tranquil_db_traits::{CommsChannel, CommsType};
use tranquil_types::{Did, Recipient};
type Repos = tranquil_db::PostgresRepositories;
async fn set_prefs(
client: &reqwest::Client,
base: &str,
token: &str,
prefs: serde_json::Value,
) -> reqwest::Response {
client
.post(format!("{}/xrpc/_account.updateNotificationPrefs", base))
.header("Authorization", format!("Bearer {}", token))
.json(&prefs)
.send()
.await
.unwrap()
}
async fn confirm_channel(
client: &reqwest::Client,
base: &str,
token: &str,
channel: &str,
id: &str,
code: &str,
) -> reqwest::Response {
client
.post(format!("{}/xrpc/_account.confirmChannelVerification", base))
.header("Authorization", format!("Bearer {}", token))
.json(&json!({"channel": channel, "identifier": id, "code": code}))
.send()
.await
.unwrap()
}
async fn latest_notices(
repos: &Repos,
user_id: uuid::Uuid,
n: i64,
) -> Vec<tranquil_db_traits::QueuedComms> {
repos
.infra
.get_latest_comms_for_user(user_id, CommsType::ChannelVerified, n)
.await
.expect("DB error")
}
use tranquil_types::Did;
#[tokio::test]
async fn test_get_notification_history() {
@@ -57,15 +11,21 @@ async fn test_get_notification_history() {
let repos = get_test_repos().await;
let (token, did) = create_account_and_login(&client).await;
let user_id = user_id_of(repos, &Did::new(did).unwrap()).await;
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
for i in 0..3 {
repos
.infra
.enqueue_comms(
Some(user_id),
&Recipient::new(CommsChannel::Email, "test@nel.pet").unwrap(),
CommsChannel::Email,
CommsType::Welcome,
"test@example.com",
Some(&format!("Subject {}", i)),
&format!("Body {}", i),
None,
@@ -97,13 +57,16 @@ async fn test_verify_channel_discord() {
let base = base_url().await;
let (token, _did) = create_account_and_login(&client).await;
let resp = set_prefs(
&client,
base,
&token,
json!({ "discordUsername": "testuser123" }),
)
.await;
let prefs = json!({
"discordUsername": "testuser123"
});
let resp = client
.post(format!("{}/xrpc/_account.updateNotificationPrefs", base))
.header("Authorization", format!("Bearer {}", token))
.json(&prefs)
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body: Value = resp.json().await.unwrap();
assert!(
@@ -130,95 +93,31 @@ async fn test_verify_channel_invalid_code() {
let base = base_url().await;
let (token, _did) = create_account_and_login(&client).await;
let resp = set_prefs(
&client,
base,
&token,
json!({ "telegramUsername": "testuser" }),
)
.await;
assert_eq!(resp.status(), 200);
let resp = confirm_channel(
&client,
base,
&token,
"telegram",
"testuser",
"XXXX-XXXX-XXXX-XXXX",
)
.await;
assert_eq!(resp.status(), 400);
}
#[tokio::test]
async fn test_channel_verified_notice_delivers_over_email_until_chat_id_is_stored() {
let client = client();
let base = base_url().await;
let repos = get_test_repos().await;
let (token, did) = create_account_and_login(&client).await;
let did = Did::new(did).unwrap();
let user_id = user_id_of(repos, &did).await;
let id = "10987654321";
let resp = set_prefs(&client, base, &token, json!({ "telegramUsername": id })).await;
assert_eq!(resp.status(), 200);
let code = |did: &Did| {
tranquil_pds::auth::verification_token::generate_channel_update_token(
did,
CommsChannel::Telegram,
id,
)
};
let resp = confirm_channel(&client, base, &token, "telegram", id, &code(&did)).await;
assert_eq!(resp.status(), 200);
let snapshot = |notices: &[tranquil_db_traits::QueuedComms]| {
notices
.iter()
.map(|notice| (notice.channel, notice.recipient.clone()))
.collect::<Vec<_>>()
};
let notices = latest_notices(repos, user_id, 5).await;
assert!(
notices
.iter()
.all(|notice| notice.channel != CommsChannel::Telegram),
"Telegram identifier entered the queue as a chat ID: {:?}",
snapshot(&notices)
);
assert!(
notices
.iter()
.any(|notice| notice.channel == CommsChannel::Email),
"The notice should fall back to email: {:?}",
snapshot(&notices)
);
repos
.user
.store_telegram_chat_id(
&tranquil_types::TelegramUsername::new(id).unwrap(),
10987654321,
None,
)
let prefs = json!({
"telegramUsername": "testuser"
});
let resp = client
.post(format!("{}/xrpc/_account.updateNotificationPrefs", base))
.header("Authorization", format!("Bearer {}", token))
.json(&prefs)
.send()
.await
.expect("DB error")
.expect("The Telegram username didn't match a user");
let resp = confirm_channel(&client, base, &token, "telegram", id, &code(&did)).await;
.unwrap();
assert_eq!(resp.status(), 200);
let notices = latest_notices(repos, user_id, 10).await;
assert!(
notices
.iter()
.any(|notice| notice.channel == CommsChannel::Telegram
&& notice.recipient == "10987654321"),
"A stored chat ID should receive the notice: {:?}",
snapshot(&notices)
);
let input = json!({
"channel": "telegram",
"identifier": "testuser",
"code": "XXXX-XXXX-XXXX-XXXX"
});
let resp = client
.post(format!("{}/xrpc/_account.confirmChannelVerification", base))
.header("Authorization", format!("Bearer {}", token))
.json(&input)
.send()
.await
.unwrap();
assert_eq!(resp.status(), 400);
}
#[tokio::test]
@@ -227,15 +126,18 @@ async fn test_verify_channel_not_set() {
let base = base_url().await;
let (token, _did) = create_account_and_login(&client).await;
let resp = confirm_channel(
&client,
base,
&token,
"signal",
"123456",
"XXXX-XXXX-XXXX-XXXX",
)
.await;
let input = json!({
"channel": "signal",
"identifier": "123456",
"code": "XXXX-XXXX-XXXX-XXXX"
});
let resp = client
.post(format!("{}/xrpc/_account.confirmChannelVerification", base))
.header("Authorization", format!("Bearer {}", token))
.json(&input)
.send()
.await
.unwrap();
assert_eq!(resp.status(), 400);
}
@@ -246,8 +148,17 @@ async fn test_update_email_via_notification_prefs() {
let repos = get_test_repos().await;
let (token, did) = create_account_and_login(&client).await;
let unique_email = format!("newemail_{}@jola.dev", uuid::Uuid::new_v4());
let resp = set_prefs(&client, base, &token, json!({ "email": unique_email })).await;
let unique_email = format!("newemail_{}@example.com", uuid::Uuid::new_v4());
let prefs = json!({
"email": unique_email
});
let resp = client
.post(format!("{}/xrpc/_account.updateNotificationPrefs", base))
.header("Authorization", format!("Bearer {}", token))
.json(&prefs)
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body: Value = resp.json().await.unwrap();
assert!(
@@ -257,7 +168,12 @@ async fn test_update_email_via_notification_prefs() {
.contains(&json!("email"))
);
let user_id = user_id_of(repos, &Did::new(did).unwrap()).await;
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
let comms = repos
.infra
@@ -286,7 +202,18 @@ async fn test_update_email_via_notification_prefs() {
.unwrap_or_default()
});
let resp = confirm_channel(&client, base, &token, "email", &unique_email, &code).await;
let input = json!({
"channel": "email",
"identifier": unique_email,
"code": code
});
let resp = client
.post(format!("{}/xrpc/_account.confirmChannelVerification", base))
.header("Authorization", format!("Bearer {}", token))
.json(&input)
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let resp = client
+18 -44
View File
@@ -16,7 +16,7 @@ async fn test_send_email_success() {
.bearer_auth(&access_jwt)
.json(&json!({
"recipientDid": did,
"senderDid": "did:plc:oystercafe",
"senderDid": "did:plc:admin",
"content": "Hello, this is a test email from the admin.",
"subject": "Test Admin Email"
}))
@@ -26,7 +26,12 @@ async fn test_send_email_success() {
assert_eq!(res.status(), StatusCode::OK);
let body: Value = res.json().await.expect("Invalid JSON");
assert_eq!(body["sent"], true);
let user_id = common::user_id_of(repos, &Did::new(did).unwrap()).await;
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
let comms = repos
.infra
.get_latest_comms_for_user(user_id, CommsType::AdminEmail, 1)
@@ -52,7 +57,7 @@ async fn test_send_email_default_subject() {
.bearer_auth(&access_jwt)
.json(&json!({
"recipientDid": did,
"senderDid": "did:plc:oystercafe",
"senderDid": "did:plc:admin",
"content": "Email without subject"
}))
.send()
@@ -61,7 +66,12 @@ async fn test_send_email_default_subject() {
assert_eq!(res.status(), StatusCode::OK);
let body: Value = res.json().await.expect("Invalid JSON");
assert_eq!(body["sent"], true);
let user_id = common::user_id_of(repos, &Did::new(did).unwrap()).await;
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
let comms = repos
.infra
.get_latest_comms_for_user(user_id, CommsType::AdminEmail, 10)
@@ -91,7 +101,7 @@ async fn test_send_email_recipient_not_found() {
.bearer_auth(&access_jwt)
.json(&json!({
"recipientDid": "did:plc:nonexistent",
"senderDid": "did:plc:oystercafe",
"senderDid": "did:plc:admin",
"content": "Test content"
}))
.send()
@@ -112,7 +122,7 @@ async fn test_send_email_missing_content() {
.bearer_auth(&access_jwt)
.json(&json!({
"recipientDid": did,
"senderDid": "did:plc:oystercafe",
"senderDid": "did:plc:admin",
"content": ""
}))
.send()
@@ -133,7 +143,7 @@ async fn test_send_email_missing_recipient() {
.bearer_auth(&access_jwt)
.json(&json!({
"recipientDid": "",
"senderDid": "did:plc:oystercafe",
"senderDid": "did:plc:admin",
"content": "Test content"
}))
.send()
@@ -150,7 +160,7 @@ async fn test_send_email_requires_auth() {
.post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
.json(&json!({
"recipientDid": "did:plc:test",
"senderDid": "did:plc:oystercafe",
"senderDid": "did:plc:admin",
"content": "Test content"
}))
.send()
@@ -158,39 +168,3 @@ async fn test_send_email_requires_auth() {
.expect("Failed to send email");
assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
}
#[tokio::test]
async fn test_send_email_rejects_garbage_stored_email() {
let client = common::client();
let base_url = common::base_url().await;
let repos = common::get_test_repos().await;
let (access_jwt, did) = common::create_admin_account_and_login(&client).await;
let user_id = common::user_id_of(repos, &Did::new(did.clone()).unwrap()).await;
repos
.user
.update_email(user_id, "not-an-email")
.await
.expect("DB error");
let res = client
.post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
.bearer_auth(&access_jwt)
.json(&json!({
"recipientDid": did,
"content": "This email should never go out"
}))
.send()
.await
.expect("Failed to send email");
assert_eq!(res.status(), StatusCode::BAD_REQUEST);
let comms = repos
.infra
.get_latest_comms_for_user(user_id, CommsType::AdminEmail, 1)
.await
.expect("DB error");
assert!(
comms.is_empty(),
"A garbage stored email doesn't reach the queue"
);
}
-113
View File
@@ -1,113 +0,0 @@
mod common;
use common::*;
use futures::StreamExt;
use reqwest::StatusCode;
use serde_json::{Value, json};
#[ctor::ctor]
fn enable_on_demand_tls() {
unsafe {
std::env::set_var("ENABLE_CADDY_ON_DEMAND_TLS", "true");
std::env::set_var("PDS_USER_HANDLE_DOMAINS", "handles.pds.test");
}
}
async fn create_hosted_account() -> String {
let client = client();
let short_handle = format!("caddy{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let payload = json!({
"handle": short_handle,
"email": format!("{}@oyster.cafe", short_handle),
"password": "Testpass123!"
});
let res = client
.post(format!(
"{}/xrpc/com.atproto.server.createAccount",
base_url().await
))
.json(&payload)
.send()
.await
.expect("failed to create account");
assert_eq!(res.status(), StatusCode::OK);
let body: Value = res
.json()
.await
.expect("createAccount response wasn't JSON");
body["handle"]
.as_str()
.expect("createAccount didn't return a handle")
.to_string()
}
async fn ask(client: &reqwest::Client, domain: &str) -> StatusCode {
client
.get(format!("{}/.well-known/caddy/ask", base_url().await))
.query(&[("domain", domain)])
.send()
.await
.expect("failed to query ask endpoint")
.status()
}
#[tokio::test]
async fn test_caddy_ask_allows_hosted_handle() {
let client = client();
let handle = create_hosted_account().await;
assert_eq!(ask(&client, &handle).await, StatusCode::OK);
assert_eq!(ask(&client, &handle.to_uppercase()).await, StatusCode::OK);
assert_eq!(ask(&client, &format!("{handle}.")).await, StatusCode::OK);
}
#[tokio::test]
async fn test_caddy_ask_denies_unhosted_and_invalid_domains() {
let client = client();
let unknown = format!("ghost-{}.handles.pds.test", uuid::Uuid::new_v4().simple());
assert_eq!(ask(&client, &unknown).await, StatusCode::NOT_FOUND);
assert_eq!(ask(&client, "nel.pet").await, StatusCode::NOT_FOUND);
assert_eq!(
ask(&client, "!!not-a-handle").await,
StatusCode::BAD_REQUEST
);
assert_eq!(ask(&client, "").await, StatusCode::BAD_REQUEST);
let res = client
.get(format!("{}/.well-known/caddy/ask", base_url().await))
.send()
.await
.expect("failed to query ask endpoint");
assert_eq!(res.status(), StatusCode::BAD_REQUEST);
}
#[tokio::test]
async fn test_caddy_ask_allows_handle_domain_apexes() {
let client = client();
base_url().await;
futures::stream::iter(
tranquil_config::get()
.server
.user_handle_domains
.iter()
.flatten(),
)
.for_each(|domain| async {
assert_eq!(ask(&client, domain.as_str()).await, StatusCode::OK);
})
.await;
}
#[tokio::test]
async fn test_caddy_ask_allows_the_pds_hostname_beside_handle_domains() {
let client = client();
base_url().await;
let cfg = tranquil_config::get();
let hostname = cfg.server.hostname_without_port();
assert!(
!cfg.server
.user_handle_domains
.iter()
.flatten()
.any(|d| d == hostname),
"this test only means something if hostname is outside the handle domains"
);
assert_eq!(ask(&client, hostname).await, StatusCode::OK);
}
-15
View File
@@ -586,8 +586,6 @@ async fn spawn_server(config: ServerConfig) -> ServerInstance {
TEST_BLOCK_STORE.set(state.block_store.clone()).ok();
if let Some((cache, distributed_rate_limiter)) = config.cache {
state = state.with_cache(cache, distributed_rate_limiter);
} else {
tranquil_pds::state::set_rate_limiting_disabled(true);
}
TEST_APP_STATE.set(state.clone()).ok();
tranquil_sync::listener::start_sequencer_listener(state.clone()).await;
@@ -926,19 +924,6 @@ pub async fn get_test_repos() -> &'static Arc<tranquil_db::PostgresRepositories>
TEST_REPOS.get().expect("TEST_REPOS not initialized")
}
#[allow(dead_code)]
pub async fn user_id_of(
repos: &tranquil_db::PostgresRepositories,
did: &tranquil_types::Did,
) -> uuid::Uuid {
repos
.user
.get_id_by_did(did)
.await
.expect("DB error")
.expect("User not found")
}
#[allow(dead_code)]
pub async fn get_test_block_store() -> &'static tranquil_pds::repo::AnyBlockStore {
base_url().await;
+19 -45
View File
@@ -69,7 +69,7 @@ async fn test_request_email_update_returns_token_required() {
let client = common::client();
let base_url = common::base_url().await;
let handle = format!("er{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@nel.pet", handle);
let email = format!("{}@example.com", handle);
let (access_jwt, _) = create_verified_account(&client, base_url, &handle, &email).await;
let res = client
@@ -92,9 +92,9 @@ async fn test_update_email_flow_success() {
let base_url = common::base_url().await;
let repos = common::get_test_repos().await;
let handle = format!("eu{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@nel.pet", handle);
let email = format!("{}@example.com", handle);
let (access_jwt, did) = create_verified_account(&client, base_url, &handle, &email).await;
let new_email = format!("new_{}@jola.dev", handle);
let new_email = format!("new_{}@example.com", handle);
let res = client
.post(format!(
@@ -139,9 +139,9 @@ async fn test_update_email_requires_token_when_verified() {
let client = common::client();
let base_url = common::base_url().await;
let handle = format!("ed{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@nel.pet", handle);
let email = format!("{}@example.com", handle);
let (access_jwt, _) = create_verified_account(&client, base_url, &handle, &email).await;
let new_email = format!("direct_{}@jola.dev", handle);
let new_email = format!("direct_{}@example.com", handle);
let res = client
.post(format!("{}/xrpc/com.atproto.server.updateEmail", base_url))
@@ -160,7 +160,7 @@ async fn test_update_email_same_email_noop() {
let client = common::client();
let base_url = common::base_url().await;
let handle = format!("es{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@nel.pet", handle);
let email = format!("{}@example.com", handle);
let (access_jwt, _) = create_verified_account(&client, base_url, &handle, &email).await;
let res = client
@@ -182,9 +182,9 @@ async fn test_update_email_invalid_token() {
let client = common::client();
let base_url = common::base_url().await;
let handle = format!("eb{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@nel.pet", handle);
let email = format!("{}@example.com", handle);
let (access_jwt, _) = create_verified_account(&client, base_url, &handle, &email).await;
let new_email = format!("badtok_{}@jola.dev", handle);
let new_email = format!("badtok_{}@example.com", handle);
let res = client
.post(format!(
@@ -219,7 +219,7 @@ async fn test_update_email_no_auth() {
let res = client
.post(format!("{}/xrpc/com.atproto.server.updateEmail", base_url))
.json(&json!({ "email": "test@jola.dev" }))
.json(&json!({ "email": "test@example.com" }))
.send()
.await
.expect("Failed to send request");
@@ -233,7 +233,7 @@ async fn test_update_email_invalid_format() {
let client = common::client();
let base_url = common::base_url().await;
let handle = format!("ef{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@nel.pet", handle);
let email = format!("{}@example.com", handle);
let (access_jwt, _) = create_verified_account(&client, base_url, &handle, &email).await;
let res = client
@@ -252,7 +252,7 @@ async fn test_confirm_email_confirms_existing_email() {
let base_url = common::base_url().await;
let repos = common::get_test_repos().await;
let handle = format!("ec{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@nel.pet", handle);
let email = format!("{}@example.com", handle);
let res = client
.post(format!(
@@ -327,7 +327,7 @@ async fn test_confirm_email_rejects_wrong_email() {
let base_url = common::base_url().await;
let repos = common::get_test_repos().await;
let handle = format!("ew{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@jola.dev", handle);
let email = format!("{}@example.com", handle);
let res = client
.post(format!(
@@ -378,7 +378,7 @@ async fn test_confirm_email_rejects_wrong_email() {
.post(format!("{}/xrpc/com.atproto.server.confirmEmail", base_url))
.bearer_auth(&access_jwt)
.json(&json!({
"email": "different@jola.dev",
"email": "different@example.com",
"token": code
}))
.send()
@@ -394,7 +394,7 @@ async fn test_confirm_email_invalid_token() {
let client = common::client();
let base_url = common::base_url().await;
let handle = format!("ei{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@jola.dev", handle);
let email = format!("{}@example.com", handle);
let res = client
.post(format!(
@@ -437,7 +437,7 @@ async fn test_unverified_account_can_update_email_without_token() {
let base_url = common::base_url().await;
let repos = common::get_test_repos().await;
let handle = format!("ev{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@nel.pet", handle);
let email = format!("{}@example.com", handle);
let res = client
.post(format!(
@@ -473,10 +473,10 @@ async fn test_unverified_account_can_update_email_without_token() {
let body: Value = res.json().await.expect("Invalid JSON");
assert_eq!(
body["tokenRequired"], false,
"An unverified account shouldn't require a token"
"Unverified account should not require token"
);
let new_email = format!("new_{}@jola.dev", handle);
let new_email = format!("new_{}@example.com", handle);
let res = client
.post(format!("{}/xrpc/com.atproto.server.updateEmail", base_url))
.bearer_auth(&access_jwt)
@@ -508,11 +508,11 @@ async fn test_update_email_to_same_as_another_user_allowed() {
let repos = common::get_test_repos().await;
let handle1 = format!("d1{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email1 = format!("{}@jola.dev", handle1);
let email1 = format!("{}@example.com", handle1);
let (_, _) = create_verified_account(&client, base_url, &handle1, &email1).await;
let handle2 = format!("d2{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email2 = format!("{}@jola.dev", handle2);
let email2 = format!("{}@example.com", handle2);
let (access_jwt2, did2) = create_verified_account(&client, base_url, &handle2, &email2).await;
let res = client
@@ -554,29 +554,3 @@ async fn test_update_email_to_same_as_another_user_allowed() {
.email;
assert_eq!(user_email, Some(email1.clone()));
}
#[tokio::test]
async fn test_check_email_in_use_distinguishes_empty_from_invalid() {
let client = common::client();
let base = common::base_url().await;
let resp = client
.post(format!("{}/xrpc/_account.checkEmailInUse", base))
.json(&json!({ "email": "not-an-email" }))
.send()
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
let body: Value = resp.json().await.unwrap();
assert_eq!(body["message"], "Invalid email address");
let resp = client
.post(format!("{}/xrpc/_account.checkEmailInUse", base))
.json(&json!({ "email": " " }))
.send()
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
let body: Value = resp.json().await.unwrap();
assert_eq!(body["message"], "Email is required");
}
+12 -2
View File
@@ -15,7 +15,12 @@ async fn test_delete_record_marks_blocks_obsolete() {
let (did, jwt) = setup_new_user("gc-after-delete").await;
let did = Did::new(did).expect("setup_new_user returned a valid DID");
let user_id = user_id_of(repos, &did).await;
let user_id = repos
.user
.get_id_by_did(&did)
.await
.expect("DB error")
.expect("User not found");
let collection = Nsid::new("app.bsky.feed.post".to_string()).expect("valid NSID");
let rkey = Rkey::new(format!("gc_test_{}", Utc::now().timestamp_millis())).expect("valid rkey");
@@ -105,7 +110,12 @@ async fn test_update_record_marks_old_record_block_obsolete() {
let (did, jwt) = setup_new_user("gc-after-update").await;
let did = Did::new(did).expect("setup_new_user returned a valid DID");
let user_id = user_id_of(repos, &did).await;
let user_id = repos
.user
.get_id_by_did(&did)
.await
.expect("DB error")
.expect("User not found");
let collection = Nsid::new("app.bsky.feed.post".to_string()).expect("valid NSID");
let rkey =
+23 -54
View File
@@ -1,9 +1,9 @@
mod common;
use common::{base_url, client, create_account_and_login, get_test_repos, user_id_of};
use common::{base_url, client, create_account_and_login, get_test_repos};
use reqwest::StatusCode;
use serde_json::{Value, json};
use tranquil_db_traits::{CommsChannel, CommsType};
use tranquil_db_traits::CommsType;
use tranquil_types::Did;
async fn enable_totp_for_user(did: &str) {
@@ -26,7 +26,13 @@ async fn set_allow_legacy_login(did: &str, allow: bool) {
async fn get_2fa_code_from_queue(did: &str) -> Option<String> {
let repos = get_test_repos().await;
let user_id = user_id_of(repos, &Did::new(did).unwrap()).await;
let parsed_did = Did::new(did.to_string()).unwrap();
let user_id = repos
.user
.get_id_by_did(&parsed_did)
.await
.expect("DB error")
.expect("User not found");
let comms = repos
.infra
@@ -50,7 +56,13 @@ async fn get_2fa_code_from_queue(did: &str) -> Option<String> {
async fn clear_2fa_challenges_for_user(did: &str) {
let repos = get_test_repos().await;
let user_id = user_id_of(repos, &Did::new(did).unwrap()).await;
let parsed_did = Did::new(did.to_string()).unwrap();
let user_id = repos
.user
.get_id_by_did(&parsed_did)
.await
.expect("DB error")
.expect("User not found");
let _ = repos
.infra
@@ -60,7 +72,13 @@ async fn clear_2fa_challenges_for_user(did: &str) {
async fn set_email_auth_factor(did: &str, enabled: bool) {
let repos = get_test_repos().await;
let user_id = user_id_of(repos, &Did::new(did).unwrap()).await;
let parsed_did = Did::new(did.to_string()).unwrap();
let user_id = repos
.user
.get_id_by_did(&parsed_did)
.await
.expect("DB error")
.expect("User not found");
repos
.infra
@@ -113,55 +131,6 @@ async fn test_legacy_2fa_auth_factor_required() {
);
}
#[tokio::test]
async fn test_legacy_2fa_undeliverable_channel_fails_login() {
let client = client();
let base = base_url().await;
let repos = get_test_repos().await;
let (_token, did) = create_account_and_login(&client).await;
enable_totp_for_user(&did).await;
set_allow_legacy_login(&did, true).await;
let parsed_did = Did::new(did.clone()).unwrap();
repos
.user
.set_channel_verified(&parsed_did, CommsChannel::Discord)
.await
.expect("DB error");
let user_id = user_id_of(repos, &parsed_did).await;
repos
.user
.update_email(user_id, &format!("undeliverable-{}", uuid::Uuid::new_v4()))
.await
.expect("DB error");
let handle = get_handle(&did).await;
let resp = client
.post(format!("{}/xrpc/com.atproto.server.createSession", base))
.json(&json!({
"identifier": handle,
"password": "Testpass123!"
}))
.send()
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
let body: Value = resp.json().await.unwrap();
assert_eq!(body["error"], "InvalidRequest");
assert!(
body["message"]
.as_str()
.unwrap_or("")
.contains("couldn't deliver the verification code"),
"the response should say the code couldn't be delivered: {body}"
);
assert!(
get_2fa_code_from_queue(&did).await.is_none(),
"the comms queue should stay empty for this user"
);
}
#[tokio::test]
async fn test_legacy_2fa_valid_code_succeeds() {
let client = client();
+19 -7
View File
@@ -1,18 +1,24 @@
mod common;
use tranquil_db_traits::{CommsChannel, CommsStatus, CommsType};
use tranquil_types::{Did, Recipient};
use tranquil_types::Did;
#[tokio::test]
async fn test_enqueue_comms() {
let repos = common::get_test_repos().await;
let (_, did) = common::create_account_and_login(&common::client()).await;
let user_id = common::user_id_of(repos, &Did::new(did).unwrap()).await;
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
repos
.infra
.enqueue_comms(
Some(user_id),
&Recipient::new(CommsChannel::Email, "test@nel.pet").unwrap(),
CommsChannel::Email,
CommsType::Welcome,
"test@example.com",
Some("Test Subject"),
"Test body",
None,
@@ -26,7 +32,7 @@ async fn test_enqueue_comms() {
.expect("DB error");
let row = comms.first().expect("Comms not found");
assert_eq!(row.user_id, Some(user_id));
assert_eq!(row.recipient, "test@nel.pet");
assert_eq!(row.recipient, "test@example.com");
assert_eq!(row.subject.as_deref(), Some("Test Subject"));
assert_eq!(row.body, "Test body");
assert_eq!(row.channel, CommsChannel::Email);
@@ -38,20 +44,26 @@ async fn test_enqueue_comms() {
async fn test_comms_queue_status_index() {
let repos = common::get_test_repos().await;
let (_, did) = common::create_account_and_login(&common::client()).await;
let user_id = common::user_id_of(repos, &Did::new(did).unwrap()).await;
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
let initial_count = repos
.infra
.count_comms_by_type(user_id, CommsType::PasswordReset)
.await
.expect("Failed to count");
for i in 0..5 {
let recipient = format!("test{}@jola.dev", i);
let recipient = format!("test{}@example.com", i);
repos
.infra
.enqueue_comms(
Some(user_id),
&Recipient::new(CommsChannel::Email, &recipient).unwrap(),
CommsChannel::Email,
CommsType::PasswordReset,
&recipient,
Some("Test"),
"Body",
None,
+6 -7
View File
@@ -1057,7 +1057,7 @@ async fn test_granular_scope_repo_create_only() {
let url = base_url().await;
let http_client = client();
let (token, did, _) =
get_oauth_token_with_scope("atproto repo:app.bsky.feed.post?action=create blob:*/*").await;
get_oauth_token_with_scope("repo:app.bsky.feed.post?action=create blob:*/*").await;
let now = chrono::Utc::now().to_rfc3339();
let create_res = http_client
.post(format!("{}/xrpc/com.atproto.repo.createRecord", url))
@@ -1111,7 +1111,7 @@ async fn test_granular_scope_wildcard_collection() {
let url = base_url().await;
let http_client = client();
let (token, did, _) = get_oauth_token_with_scope(
"atproto repo:app.bsky.*?action=create&action=update&action=delete blob:*/*",
"repo:app.bsky.*?action=create&action=update&action=delete blob:*/*",
)
.await;
let now = chrono::Utc::now().to_rfc3339();
@@ -1168,7 +1168,7 @@ async fn test_granular_scope_wildcard_collection() {
async fn test_granular_scope_email_read() {
let url = base_url().await;
let http_client = client();
let (token, did, _) = get_oauth_token_with_scope("atproto account:email?action=read").await;
let (token, did, _) = get_oauth_token_with_scope("account:email?action=read").await;
let session_res = http_client
.get(format!("{}/xrpc/com.atproto.server.getSession", url))
.bearer_auth(&token)
@@ -1189,7 +1189,7 @@ async fn test_granular_scope_email_read() {
async fn test_granular_scope_no_email_access() {
let url = base_url().await;
let http_client = client();
let (token, did, _) = get_oauth_token_with_scope("atproto repo:*?action=create blob:*/*").await;
let (token, did, _) = get_oauth_token_with_scope("repo:*?action=create blob:*/*").await;
let session_res = http_client
.get(format!("{}/xrpc/com.atproto.server.getSession", url))
.bearer_auth(&token)
@@ -1210,8 +1210,7 @@ async fn test_granular_scope_no_email_access() {
async fn test_granular_scope_rpc_specific_method() {
let url = base_url().await;
let http_client = client();
let (token, _, _) =
get_oauth_token_with_scope("atproto rpc:app.bsky.feed.getTimeline?aud=*").await;
let (token, _, _) = get_oauth_token_with_scope("rpc:app.bsky.feed.getTimeline?aud=*").await;
let allowed_res = http_client
.get(format!("{}/xrpc/com.atproto.server.getServiceAuth", url))
.bearer_auth(&token)
@@ -1276,7 +1275,7 @@ async fn test_granular_scope_rpc_aud_with_service_id() {
let url = base_url().await;
let http_client = client();
let (token, _, _) = get_oauth_token_with_scope(
"atproto rpc:app.bsky.feed.getTimeline?aud=did:web:api.bsky.app#bsky_appview",
"rpc:app.bsky.feed.getTimeline?aud=did:web:api.bsky.app#bsky_appview",
)
.await;
let allowed_res = http_client
+2 -510
View File
@@ -3,7 +3,7 @@ mod helpers;
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use chrono::Utc;
use common::{base_url, client, get_test_repos};
use common::{base_url, client};
use helpers::verify_new_account;
use reqwest::StatusCode;
use serde_json::{Value, json};
@@ -22,16 +22,9 @@ fn generate_pkce() -> (String, String) {
}
async fn setup_mock_client_metadata(redirect_uri: &str) -> MockServer {
setup_mock_client_metadata_with_scope(redirect_uri, None).await
}
async fn setup_mock_client_metadata_with_scope(
redirect_uri: &str,
scope: Option<&str>,
) -> MockServer {
let mock_server = MockServer::start().await;
let client_id = mock_server.uri();
let mut metadata = json!({
let metadata = json!({
"client_id": client_id,
"client_name": "Test OAuth Scope Client",
"redirect_uris": [redirect_uri],
@@ -40,9 +33,6 @@ async fn setup_mock_client_metadata_with_scope(
"token_endpoint_auth_method": "none",
"dpop_bound_access_tokens": false
});
if let Some(scope) = scope {
metadata["scope"] = json!(scope);
}
Mock::given(method("GET"))
.and(path("/"))
.respond_with(ResponseTemplate::new(200).set_body_json(metadata))
@@ -703,501 +693,3 @@ async fn test_dereference_scope_requires_auth() {
"Should require authentication"
);
}
#[tokio::test]
async fn test_unrecognized_scope_reaches_consent_and_is_never_granted() {
let url = base_url().await;
let http_client = client();
let redirect_uri = "https://example.com/callback";
let suffix = &uuid::Uuid::new_v4().simple().to_string()[..4];
let handle = format!("badscope{}", suffix);
let email = format!("badscope{}@example.com", suffix);
let password = "BadscopePass123!";
let create_res = http_client
.post(format!("{}/xrpc/com.atproto.server.createAccount", url))
.json(&json!({ "handle": handle, "email": email, "password": password }))
.send()
.await
.expect("Account creation failed");
assert_eq!(create_res.status(), StatusCode::OK);
let account: Value = create_res.json().await.unwrap();
let user_did = account["did"].as_str().unwrap().to_string();
let _ = verify_new_account(&http_client, &user_did).await;
let mock_client = setup_mock_client_metadata(redirect_uri).await;
let client_id = mock_client.uri();
let (code_verifier, code_challenge) = generate_pkce();
let par_res = http_client
.post(format!("{}/oauth/par", url))
.form(&[
("response_type", "code"),
("client_id", &client_id),
("redirect_uri", redirect_uri),
("code_challenge", &code_challenge),
("code_challenge_method", "S256"),
("scope", "atproto chat"),
])
.send()
.await
.expect("PAR failed");
assert!(
par_res.status() == StatusCode::OK || par_res.status() == StatusCode::CREATED,
"PAR must not reject an unrecognized scope, got {}",
par_res.status()
);
let par_body: Value = par_res.json().await.unwrap();
let request_uri = par_body["request_uri"].as_str().unwrap().to_string();
let auth_res = http_client
.post(format!("{}/oauth/authorize", url))
.header("Content-Type", "application/json")
.header("Accept", "application/json")
.json(&json!({
"request_uri": request_uri,
"username": &handle,
"password": password,
"remember_device": false
}))
.send()
.await
.expect("Authorize failed");
assert_eq!(auth_res.status(), StatusCode::OK);
let auth_body: Value = auth_res.json().await.unwrap();
let location = auth_body["redirect_uri"].as_str().unwrap().to_string();
assert!(
location.contains("/oauth/consent"),
"should land on the consent screen, got {}",
location
);
let consent_get: Value = http_client
.get(format!(
"{}/oauth/authorize/consent?request_uri={}",
url, request_uri
))
.send()
.await
.expect("Consent GET failed")
.json()
.await
.unwrap();
let rejected = consent_get["rejected_scopes"].as_array().unwrap();
assert_eq!(rejected.len(), 1, "got {:?}", rejected);
assert_eq!(rejected[0]["scope"].as_str(), Some("chat"));
assert_eq!(rejected[0]["reason"].as_str(), Some("unrecognized"));
assert!(
!consent_get["scopes"]
.as_array()
.unwrap()
.iter()
.any(|s| s["scope"] == "chat"),
"an unrecognized scope must never be offered as grantable"
);
let consent_res = http_client
.post(format!("{}/oauth/authorize/consent", url))
.header("Content-Type", "application/json")
.json(&json!({
"request_uri": request_uri,
"approved_scopes": ["atproto", "chat"],
"remember": false
}))
.send()
.await
.expect("Consent POST failed");
assert_eq!(consent_res.status(), StatusCode::OK);
let consent_body: Value = consent_res.json().await.unwrap();
let location = consent_body["redirect_uri"].as_str().unwrap().to_string();
let code = location
.split("code=")
.nth(1)
.unwrap()
.split('&')
.next()
.unwrap();
let token_res = http_client
.post(format!("{}/oauth/token", url))
.form(&[
("grant_type", "authorization_code"),
("code", code),
("redirect_uri", redirect_uri),
("code_verifier", &code_verifier),
("client_id", &client_id),
])
.send()
.await
.expect("Token request failed");
assert_eq!(token_res.status(), StatusCode::OK);
let token_body: Value = token_res.json().await.unwrap();
let granted = token_body["scope"].as_str().unwrap();
assert!(
granted.split_whitespace().any(|s| s == "atproto"),
"granted scope was {:?}",
granted
);
assert!(
!granted.split_whitespace().any(|s| s == "chat"),
"an unrecognized scope leaked into the issued token: {:?}",
granted
);
}
struct PendingAuthorization {
client_id: String,
request_uri: String,
code_verifier: String,
// Where authorize sent us: the consent screen, or straight to the client with a code.
location: String,
_mock: MockServer,
}
const REDIRECT_URI: &str = "https://example.com/callback";
async fn par_and_login(
handle_prefix: &str,
requested_scope: &str,
client_scope: Option<&str>,
before_login: impl AsyncFnOnce(&str, &str),
) -> PendingAuthorization {
let url = base_url().await;
let http_client = client();
let suffix = &uuid::Uuid::new_v4().simple().to_string()[..4];
let handle = format!("{}{}", handle_prefix, suffix);
let password = format!("{}Pass123!", handle_prefix);
let create_res = http_client
.post(format!("{}/xrpc/com.atproto.server.createAccount", url))
.json(&json!({
"handle": handle,
"email": format!("{}{}@example.com", handle_prefix, suffix),
"password": password
}))
.send()
.await
.expect("Account creation failed");
assert_eq!(create_res.status(), StatusCode::OK);
let account: Value = create_res.json().await.unwrap();
let did = account["did"].as_str().unwrap().to_string();
let _ = verify_new_account(&http_client, &did).await;
let mock = setup_mock_client_metadata_with_scope(REDIRECT_URI, client_scope).await;
let client_id = mock.uri();
let (code_verifier, code_challenge) = generate_pkce();
let par_res = http_client
.post(format!("{}/oauth/par", url))
.form(&[
("response_type", "code"),
("client_id", &client_id),
("redirect_uri", REDIRECT_URI),
("code_challenge", &code_challenge),
("code_challenge_method", "S256"),
("scope", requested_scope),
])
.send()
.await
.expect("PAR failed");
assert_eq!(par_res.status(), StatusCode::CREATED, "PAR should succeed");
let par_body: Value = par_res.json().await.unwrap();
let request_uri = par_body["request_uri"].as_str().unwrap().to_string();
before_login(&did, &client_id).await;
let auth_res = http_client
.post(format!("{}/oauth/authorize", url))
.header("Content-Type", "application/json")
.header("Accept", "application/json")
.json(&json!({
"request_uri": request_uri,
"username": &handle,
"password": &password,
"remember_device": false
}))
.send()
.await
.expect("Authorize failed");
assert_eq!(auth_res.status(), StatusCode::OK);
let auth_body: Value = auth_res.json().await.unwrap();
let location = auth_body["redirect_uri"].as_str().unwrap().to_string();
PendingAuthorization {
client_id,
request_uri,
code_verifier,
location,
_mock: mock,
}
}
async fn exchange_code(pending: &PendingAuthorization, location: &str) -> Value {
let code = location
.split("code=")
.nth(1)
.expect("redirect should carry a code")
.split('&')
.next()
.unwrap();
let token_res = client()
.post(format!("{}/oauth/token", base_url().await))
.form(&[
("grant_type", "authorization_code"),
("code", code),
("redirect_uri", REDIRECT_URI),
("code_verifier", &pending.code_verifier),
("client_id", &pending.client_id),
])
.send()
.await
.expect("Token request failed");
assert_eq!(token_res.status(), StatusCode::OK);
token_res.json().await.unwrap()
}
fn has_scope(scope_str: &str, scope: &str) -> bool {
scope_str.split_whitespace().any(|s| s == scope)
}
#[tokio::test]
async fn test_par_rejects_scope_without_atproto() {
let url = base_url().await;
let mock = setup_mock_client_metadata(REDIRECT_URI).await;
let client_id = mock.uri();
let (_, code_challenge) = generate_pkce();
let par_res = client()
.post(format!("{}/oauth/par", url))
.form(&[
("response_type", "code"),
("client_id", &client_id),
("redirect_uri", REDIRECT_URI),
("code_challenge", &code_challenge),
("code_challenge_method", "S256"),
("scope", "repo:*?action=create"),
])
.send()
.await
.expect("PAR failed");
assert_eq!(par_res.status(), StatusCode::BAD_REQUEST);
let body: Value = par_res.json().await.unwrap();
assert_eq!(
body["error"].as_str(),
Some("invalid_scope"),
"got {:?}",
body
);
}
#[tokio::test]
async fn test_scope_missing_from_client_metadata_is_not_registered_on_consent() {
let pending = par_and_login(
"unreg",
"atproto identity:*",
Some("atproto"),
async |_, _| {},
)
.await;
assert!(pending.location.contains("/oauth/consent"));
let url = base_url().await;
let consent_get: Value = client()
.get(format!(
"{}/oauth/authorize/consent?request_uri={}",
url, pending.request_uri
))
.send()
.await
.expect("Consent GET failed")
.json()
.await
.unwrap();
let rejected = consent_get["rejected_scopes"].as_array().unwrap();
assert_eq!(rejected.len(), 1, "got {:?}", rejected);
assert_eq!(rejected[0]["scope"].as_str(), Some("identity:*"));
assert_eq!(rejected[0]["reason"].as_str(), Some("not_registered"));
let consent_res = client()
.post(format!("{}/oauth/authorize/consent", url))
.header("Content-Type", "application/json")
.json(&json!({
"request_uri": pending.request_uri,
"approved_scopes": ["atproto", "identity:*"],
"remember": false
}))
.send()
.await
.expect("Consent POST failed");
assert_eq!(consent_res.status(), StatusCode::OK);
let consent_body: Value = consent_res.json().await.unwrap();
let token = exchange_code(&pending, consent_body["redirect_uri"].as_str().unwrap()).await;
let granted = token["scope"].as_str().unwrap();
assert!(!has_scope(granted, "identity:*"), "granted {:?}", granted);
}
/// A remembered consent skips the consent screen, so the scope stored on the token must be
/// filtered at issuance rather than copied from the raw request. Otherwise a scope the client
/// has since dropped from its metadata survives in storage and comes back on refresh.
#[tokio::test]
async fn test_remembered_scope_later_unregistered_never_reaches_a_token() {
let pending = par_and_login(
"remember",
"atproto identity:*",
Some("atproto"),
async |did, client_id| {
let prefs =
["atproto", "identity:*"].map(|scope| tranquil_pds::oauth::db::ScopePreference {
scope: scope.to_string(),
granted: true,
});
get_test_repos()
.await
.oauth
.upsert_scope_preferences(
&did.parse().unwrap(),
&tranquil_types::ClientId::new(client_id.to_string()),
&prefs,
)
.await
.expect("seeding scope preferences failed");
},
)
.await;
assert!(
!pending.location.contains("/oauth/consent"),
"remembered consent should skip the consent screen, got {}",
pending.location
);
let token = exchange_code(&pending, &pending.location).await;
assert!(!has_scope(token["scope"].as_str().unwrap(), "identity:*"));
let token_id = {
let payload = token["access_token"]
.as_str()
.unwrap()
.split('.')
.nth(1)
.unwrap();
let claims: Value =
serde_json::from_slice(&URL_SAFE_NO_PAD.decode(payload).unwrap()).unwrap();
tranquil_types::TokenId::new(claims["sid"].as_str().expect("sid claim"))
};
let row = get_test_repos()
.await
.oauth
.get_token_by_id(&token_id)
.await
.expect("get_token_by_id query failed")
.expect("token row should exist");
let row_scope = row.scope.expect("token row should have a scope");
assert!(
!has_scope(&row_scope, "identity:*"),
"stored {:?}",
row_scope
);
let refresh_res = client()
.post(format!("{}/oauth/token", base_url().await))
.form(&[
("grant_type", "refresh_token"),
("refresh_token", token["refresh_token"].as_str().unwrap()),
("client_id", &pending.client_id),
])
.send()
.await
.expect("Refresh request failed");
assert_eq!(refresh_res.status(), StatusCode::OK);
let refreshed: Value = refresh_res.json().await.unwrap();
assert!(
!has_scope(refreshed["scope"].as_str().unwrap(), "identity:*"),
"refresh granted {:?}",
refreshed["scope"]
);
}
const PROXY_LXM: &str = "io.atcr.getManifest";
const PROXY_SERVICE_ID: &str = "test_service";
async fn setup_mock_proxy_service() -> (MockServer, String) {
let server = MockServer::start().await;
let uri = server.uri();
let host = uri
.strip_prefix("http://")
.expect("mock uri should be http");
let did = format!("did:web:{host}");
Mock::given(method("GET"))
.and(path("/.well-known/did.json"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"id": did,
"service": [{
"id": format!("#{PROXY_SERVICE_ID}"),
"type": "TestService",
"serviceEndpoint": uri,
}]
})))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path(format!("/xrpc/{PROXY_LXM}")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({ "manifest": [] })))
.mount(&server)
.await;
(server, did)
}
async fn proxied_call(access_token: &str, service_did: &str) -> reqwest::Response {
client()
.get(format!("{}/xrpc/{}", base_url().await, PROXY_LXM))
.bearer_auth(access_token)
.header("atproto-proxy", format!("{service_did}#{PROXY_SERVICE_ID}"))
.send()
.await
.expect("Proxied request failed")
}
#[tokio::test]
async fn test_rpc_scope_with_service_fragment_allows_proxied_call() {
let (_service, service_did) = setup_mock_proxy_service().await;
let scope = format!("atproto rpc:{PROXY_LXM}?aud={service_did}#{PROXY_SERVICE_ID}");
let (session, _mock) = create_user_and_oauth_session_with_scope(
"rpcfrag",
"https://example.com/rpc-fragment-callback",
&scope,
)
.await;
let res = proxied_call(&session.access_token, &service_did).await;
let status = res.status();
assert_eq!(
status,
StatusCode::OK,
"a scope audienced to did#serviceId must cover a call proxied to that service, got {} {:?}",
status,
res.text().await
);
}
#[tokio::test]
async fn test_rpc_scope_with_bare_did_is_refused_for_proxied_call() {
let (_service, service_did) = setup_mock_proxy_service().await;
let scope = format!("atproto rpc:{PROXY_LXM}?aud={service_did}");
let (session, _mock) = create_user_and_oauth_session_with_scope(
"rpcbare",
"https://example.com/rpc-bare-callback",
&scope,
)
.await;
let res = proxied_call(&session.access_token, &service_did).await;
let status = res.status();
assert_eq!(
status,
StatusCode::FORBIDDEN,
"a bare DID audience must not cover a call proxied to a service on that DID, got {} {:?}",
status,
res.text().await
);
}
@@ -57,7 +57,12 @@ async fn repair_fails_loud_on_missing_leaf_block() {
res.text().await
);
let user_id = user_id_of(repos, &Did::new(did.clone()).unwrap()).await;
let user_id = repos
.user
.get_id_by_did(&Did::new(did.clone()).unwrap())
.await
.expect("DB error")
.expect("user not found");
let root_str = repos
.repo
+8 -1
View File
@@ -498,7 +498,14 @@ async fn test_apply_writes_create_then_delete_same_rkey() {
}
async fn repo_id_for_did(did: &str) -> uuid::Uuid {
user_id_of(get_test_repos().await, &Did::new(did).expect("valid DID")).await
let repos = get_test_repos().await;
let parsed = Did::new(did).expect("valid did");
repos
.user
.get_id_by_did(&parsed)
.await
.expect("lookup user_id")
.expect("user exists")
}
async fn follow_uris_pointing_to(repo_id: uuid::Uuid, target_did: &str) -> Vec<String> {
+43 -1
View File
@@ -1,5 +1,5 @@
mod common;
use tranquil_pds::comms::{SendError, is_valid_phone_number};
use tranquil_pds::comms::{SendError, is_valid_phone_number, is_valid_signal_username};
use tranquil_pds::image::{ImageError, ImageProcessor};
#[test]
@@ -46,6 +46,48 @@ fn test_phone_number_validation() {
}
}
#[test]
fn test_signal_username_validation() {
assert!(is_valid_signal_username("alice.01"));
assert!(is_valid_signal_username("bob_smith.99"));
assert!(is_valid_signal_username("user123.42"));
assert!(is_valid_signal_username("lu1.01"));
assert!(is_valid_signal_username("a_very_long_username_here.55"));
assert!(is_valid_signal_username("alice.123"));
assert!(is_valid_signal_username("alice.999999999"));
assert!(is_valid_signal_username("alice.18446744073709551615"));
assert!(!is_valid_signal_username("alice"));
assert!(!is_valid_signal_username("alice.1"));
assert!(!is_valid_signal_username("alice.001"));
assert!(!is_valid_signal_username("abc.00"));
assert!(!is_valid_signal_username("alice.0"));
assert!(!is_valid_signal_username("alice.999999999999999999999"));
assert!(!is_valid_signal_username(".01"));
assert!(!is_valid_signal_username("ab.01"));
assert!(!is_valid_signal_username(""));
assert!(!is_valid_signal_username("1alice.01"));
assert!(!is_valid_signal_username("alice!.01"));
assert!(!is_valid_signal_username("alice .01"));
assert!(!is_valid_signal_username("a".repeat(33).as_str()));
[
"alice.01; rm -rf /",
"bob.01 && cat /etc/passwd",
"user.01`id`",
"test.01$(whoami)",
]
.iter()
.for_each(|malicious| {
assert!(
!is_valid_signal_username(malicious),
"Command injection '{}' should be rejected",
malicious
);
});
}
#[test]
fn test_image_file_size_limits() {
let processor = ImageProcessor::new();
+1 -10
View File
@@ -3,7 +3,7 @@ mod common;
use common::{base_url, client, create_account_and_login, get_test_repos};
use reqwest::StatusCode;
use serde_json::{Value, json};
use tranquil_db_traits::{CommsChannel, CommsType, SsoAction, SsoProviderType};
use tranquil_db_traits::{CommsChannel, SsoAction, SsoProviderType};
use tranquil_oauth::{
AuthorizationRequestParameters, CodeChallengeMethod, RequestData, ResponseType,
};
@@ -781,7 +781,6 @@ async fn test_sso_complete_registration_multichannel_discord() {
.json(&json!({
"token": token,
"handle": handle_prefix,
"email": "sso_discord_reg@jola.dev",
"verification_channel": "discord",
"discord_username": discord_id
}))
@@ -811,14 +810,6 @@ async fn test_sso_complete_registration_multichannel_discord() {
let user = user.unwrap();
assert_eq!(user.channel, CommsChannel::Discord);
assert_eq!(user.discord_username.as_deref(), Some(discord_id));
let quered = repos
.infra
.get_latest_comms_for_user(user.id, CommsType::EmailVerification, 1)
.await
.unwrap();
let comms = quered.first().expect("We queued up a verification email");
assert_eq!(comms.channel, CommsChannel::Email);
assert_eq!(comms.recipient, "sso_discord_reg@jola.dev");
}
#[tokio::test]
+14 -160
View File
@@ -4,7 +4,7 @@ mod helpers;
use std::sync::Arc;
use tranquil_db::PostgresRepositories;
use tranquil_db_traits::{Backlink, BacklinkPath, CommsChannel, CommsType};
use tranquil_types::{AtUri, CidLink, Did, Handle, Nsid, Recipient, Rkey, Tid};
use tranquil_types::{AtUri, CidLink, Did, Handle, Nsid, Rkey, Tid};
use uuid::Uuid;
async fn create_store_repos() -> Arc<PostgresRepositories> {
@@ -986,162 +986,6 @@ async fn parity_blob_duplicate_insert() {
assert_eq!(pg_dup, store_dup);
}
#[tokio::test(flavor = "multi_thread")]
async fn parity_blob_shared_between_repos() {
let f = ParityFixture::new().await;
let did_a = test_did("shareda");
let did_b = test_did("sharedb");
let (pg_a, store_a) = seed_repos(&f, &did_a, &test_handle("shareda")).await;
let (pg_b, store_b) = seed_repos(&f, &did_b, &test_handle("sharedb")).await;
let cid = test_cid(210);
let pg_first =
f.pg.blob
.insert_blob(&cid, "image/png", 100, pg_a, "blobs/shared.png")
.await
.unwrap();
let store_first = f
.store
.blob
.insert_blob(&cid, "image/png", 100, store_a, "blobs/shared.png")
.await
.unwrap();
assert_eq!(pg_first, store_first);
let pg_second =
f.pg.blob
.insert_blob(&cid, "image/png", 100, pg_b, "blobs/shared.png")
.await
.unwrap();
let store_second = f
.store
.blob
.insert_blob(&cid, "image/png", 100, store_b, "blobs/shared.png")
.await
.unwrap();
assert_eq!(pg_second, store_second);
assert!(pg_second.is_some());
for (pg_uid, store_uid) in [(pg_a, store_a), (pg_b, store_b)] {
assert_eq!(f.pg.blob.count_blobs_by_user(pg_uid).await.unwrap(), 1);
assert_eq!(
f.store.blob.count_blobs_by_user(store_uid).await.unwrap(),
1
);
assert_eq!(
f.pg.blob
.list_blobs_by_user(pg_uid, None, 100)
.await
.unwrap(),
vec![cid.clone()]
);
assert_eq!(
f.store
.blob
.list_blobs_by_user(store_uid, None, 100)
.await
.unwrap(),
vec![cid.clone()]
);
assert!(
f.pg.blob
.get_blob_storage_keys_by_user(pg_uid)
.await
.unwrap()
.is_empty()
);
assert!(
f.store
.blob
.get_blob_storage_keys_by_user(store_uid)
.await
.unwrap()
.is_empty()
);
}
}
#[tokio::test(flavor = "multi_thread")]
async fn parity_ensure_blob_ownership() {
let f = ParityFixture::new().await;
let did_a = test_did("ensurea");
let did_b = test_did("ensureb");
let (pg_a, store_a) = seed_repos(&f, &did_a, &test_handle("ensurea")).await;
let (pg_b, store_b) = seed_repos(&f, &did_b, &test_handle("ensureb")).await;
let cid = test_cid(211);
f.pg.blob
.insert_blob(&cid, "image/png", 100, pg_a, "blobs/ensure.png")
.await
.unwrap();
f.store
.blob
.insert_blob(&cid, "image/png", 100, store_a, "blobs/ensure.png")
.await
.unwrap();
assert!(f.pg.blob.ensure_blob_ownership(pg_b, &cid).await.unwrap());
assert!(
f.store
.blob
.ensure_blob_ownership(store_b, &cid)
.await
.unwrap()
);
assert!(!f.pg.blob.ensure_blob_ownership(pg_b, &cid).await.unwrap());
assert!(
!f.store
.blob
.ensure_blob_ownership(store_b, &cid)
.await
.unwrap()
);
let absent = test_cid(212);
assert!(
!f.pg
.blob
.ensure_blob_ownership(pg_b, &absent)
.await
.unwrap()
);
assert!(
!f.store
.blob
.ensure_blob_ownership(store_b, &absent)
.await
.unwrap()
);
for (pg_uid, store_uid) in [(pg_a, store_a), (pg_b, store_b)] {
assert_eq!(f.pg.blob.count_blobs_by_user(pg_uid).await.unwrap(), 1);
assert_eq!(
f.store.blob.count_blobs_by_user(store_uid).await.unwrap(),
1
);
assert!(
f.pg.blob
.get_blob_storage_keys_by_user(pg_uid)
.await
.unwrap()
.is_empty()
);
assert!(
f.store
.blob
.get_blob_storage_keys_by_user(store_uid)
.await
.unwrap()
.is_empty()
);
}
}
#[tokio::test]
async fn parity_get_all_records() {
let f = ParityFixture::new().await;
@@ -1199,8 +1043,9 @@ async fn parity_comms_queue() {
f.pg.infra
.enqueue_comms(
Some(pg_uid),
&Recipient::new(CommsChannel::Email, "test@jola.dev").unwrap(),
CommsChannel::Email,
CommsType::Welcome,
"test@example.com",
Some("Welcome"),
"Welcome body",
None,
@@ -1213,8 +1058,9 @@ async fn parity_comms_queue() {
.infra
.enqueue_comms(
Some(store_uid),
&Recipient::new(CommsChannel::Email, "test@jola.dev").unwrap(),
CommsChannel::Email,
CommsType::Welcome,
"test@example.com",
Some("Welcome"),
"Welcome body",
None,
@@ -1721,7 +1567,7 @@ async fn parity_plc_tokens() {
}
#[tokio::test]
async fn parity_blob_takedown() {
async fn parity_blob_delete_and_takedown() {
let f = ParityFixture::new().await;
let did = test_did("blobdel");
let handle = test_handle("blobdel");
@@ -1758,6 +1604,14 @@ async fn parity_blob_takedown() {
pg_with_td.as_ref().map(|b| b.takedown_ref.as_deref()),
store_with_td.as_ref().map(|b| b.takedown_ref.as_deref())
);
f.pg.blob.delete_blob_by_cid(&cid).await.unwrap();
f.store.blob.delete_blob_by_cid(&cid).await.unwrap();
let pg_meta = f.pg.blob.get_blob_metadata(&cid).await.unwrap();
let store_meta = f.store.blob.get_blob_metadata(&cid).await.unwrap();
assert!(pg_meta.is_none());
assert!(store_meta.is_none());
}
#[tokio::test]
@@ -109,7 +109,13 @@ async fn assert_record_gone(did: &Did, rkey: &Rkey) {
}
async fn user_id_for(did: &Did) -> uuid::Uuid {
user_id_of(get_test_repos().await, did).await
get_test_repos()
.await
.user
.get_id_by_did(did)
.await
.expect("DB error looking up the user id")
.expect("User not found")
}
#[tokio::test]
@@ -1,6 +1,7 @@
use tranquil_lexicon::is_valid_did;
use tranquil_pds::api::validation::{
HandleValidationError, MAX_SERVICE_HANDLE_LOCAL_PART, validate_short_handle,
HandleValidationError, MAX_DOMAIN_LABEL_LENGTH, MAX_EMAIL_LENGTH, MAX_LOCAL_PART_LENGTH,
MAX_SERVICE_HANDLE_LOCAL_PART, is_valid_email, validate_short_handle,
};
use tranquil_pds::validation::{validate_collection_nsid, validate_password, validate_record_key};
@@ -260,3 +261,83 @@ fn test_handle_whitespace_handling() {
Err(HandleValidationError::ContainsSpaces)
));
}
#[test]
fn test_email_length_boundaries() {
let long_local = format!("{}@example.com", "a".repeat(MAX_LOCAL_PART_LENGTH));
assert!(is_valid_email(&long_local));
let too_long_local = format!("{}@example.com", "a".repeat(MAX_LOCAL_PART_LENGTH + 1));
assert!(!is_valid_email(&too_long_local));
let very_long_email = format!("a@{}.com", "a".repeat(240));
if very_long_email.len() <= MAX_EMAIL_LENGTH {
assert!(is_valid_email(&very_long_email) || !is_valid_email(&very_long_email));
}
}
#[test]
fn test_email_local_part_special_chars() {
assert!(is_valid_email("user.name@example.com"));
assert!(is_valid_email("user+tag@example.com"));
assert!(is_valid_email("user!def@example.com"));
assert!(is_valid_email("user#abc@example.com"));
assert!(is_valid_email("user$def@example.com"));
assert!(is_valid_email("user%abc@example.com"));
assert!(is_valid_email("user&def@example.com"));
assert!(is_valid_email("user'abc@example.com"));
assert!(is_valid_email("user*def@example.com"));
assert!(is_valid_email("user=abc@example.com"));
assert!(is_valid_email("user?def@example.com"));
assert!(is_valid_email("user^abc@example.com"));
assert!(is_valid_email("user_def@example.com"));
assert!(is_valid_email("user`abc@example.com"));
assert!(is_valid_email("user{def@example.com"));
assert!(is_valid_email("user|abc@example.com"));
assert!(is_valid_email("user}def@example.com"));
assert!(is_valid_email("user~abc@example.com"));
assert!(is_valid_email("user-def@example.com"));
}
#[test]
fn test_email_local_part_dots() {
assert!(!is_valid_email(".user@example.com"));
assert!(!is_valid_email("user.@example.com"));
assert!(!is_valid_email("user..name@example.com"));
assert!(is_valid_email("user.name@example.com"));
assert!(is_valid_email("u.s.e.r@example.com"));
}
#[test]
fn test_email_domain_labels() {
let long_label = "a".repeat(MAX_DOMAIN_LABEL_LENGTH);
let valid_domain = format!("user@{}.com", long_label);
assert!(is_valid_email(&valid_domain));
let too_long_label = "a".repeat(MAX_DOMAIN_LABEL_LENGTH + 1);
let invalid_domain = format!("user@{}.com", too_long_label);
assert!(!is_valid_email(&invalid_domain));
}
#[test]
fn test_email_domain_hyphens() {
assert!(!is_valid_email("user@-example.com"));
assert!(!is_valid_email("user@example-.com"));
assert!(is_valid_email("user@ex-ample.com"));
assert!(is_valid_email("user@ex--ample.com"));
}
#[test]
fn test_email_domain_must_have_dot() {
assert!(!is_valid_email("user@localhost"));
assert!(!is_valid_email("user@example"));
assert!(is_valid_email("user@a.b"));
}
#[test]
fn test_email_invalid_chars() {
assert!(!is_valid_email("user name@example.com"));
assert!(!is_valid_email("user\t@example.com"));
assert!(!is_valid_email("user\n@example.com"));
assert!(!is_valid_email("user@exam ple.com"));
}
+1 -1
View File
@@ -48,7 +48,7 @@ pub static SCOPE_DEFINITIONS: LazyLock<HashMap<&'static str, ScopeDefinition>> =
category: ScopeCategory::Transition,
required: false,
description: "Generic transition scope for compatibility",
display_name: "Generic Access",
display_name: "Transition Access",
},
ScopeDefinition {
scope: "transition:chat.bsky",
+3 -3
View File
@@ -16,7 +16,7 @@ pub use parser::{
ParsedScope, RepoAction, RepoScope, RpcScope, parse_scope, parse_scope_string,
};
pub use permission_set::{
ExpansionOutcome, FailedSet, FetchedSet, RejectedScope, ResolveFailure, ResolvedSetGroup,
ScopeExpansionError, ScopeRejection, fetch_and_expand, parse_include_scope,
ExpansionOutcome, FailedSet, FetchedSet, ResolveFailure, ResolvedSetGroup, ScopeExpansionError,
fetch_and_expand, parse_include_scope,
};
pub use permissions::{ScopePermissions, superseded_by_transition_generic};
pub use permissions::ScopePermissions;
@@ -44,19 +44,6 @@ pub enum ResolveFailure {
EmptyPermissions,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ScopeRejection {
Unrecognized,
NotRegistered,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RejectedScope {
pub scope: String,
pub reason: ScopeRejection,
}
#[derive(Debug, Clone)]
pub struct FailedSet {
// NSID and aud are left as strings to avoid issues from malformed requests.
@@ -74,21 +61,11 @@ pub struct ResolvedSetGroup {
pub expanded: Vec<String>,
}
impl ResolvedSetGroup {
pub fn include_token(&self) -> String {
match &self.aud {
Some(aud) => format!("include:{}?aud={}", self.nsid, aud),
None => format!("include:{}", self.nsid),
}
}
}
#[derive(Debug, Clone, Default)]
pub struct ExpansionOutcome {
pub passthrough: Vec<String>,
pub sets: Vec<ResolvedSetGroup>,
pub failures: Vec<FailedSet>,
pub rejected: Vec<RejectedScope>,
}
impl ExpansionOutcome {
@@ -110,16 +87,6 @@ impl ExpansionOutcome {
pub fn to_scope_string(&self) -> String {
self.flat_scopes().join(" ")
}
/// The scopes that survived filtering, as requested: passthrough scopes plus the `include:`
/// token of each resolved set, without expanding the sets.
pub fn unexpanded_scopes(&self) -> Vec<String> {
self.passthrough
.iter()
.cloned()
.chain(self.sets.iter().map(ResolvedSetGroup::include_token))
.collect()
}
}
#[derive(Debug, Deserialize)]
@@ -844,7 +811,6 @@ mod tests {
given_aud: None,
reason: ResolveFailure::NotFound,
}],
rejected: vec![],
};
let flat = out.flat_scopes();
assert_eq!(
@@ -873,7 +839,6 @@ mod tests {
expanded: vec!["repo:x".into(), "rpc:io.atcr.getManifest".into()],
}],
failures: vec![],
rejected: vec![],
};
let flat = out.flat_scopes();
assert_eq!(flat, vec!["repo:x", "rpc:io.atcr.getManifest"]);
+32 -205
View File
@@ -43,26 +43,7 @@ impl ScopePermissions {
has_transition_email,
}
}
}
/// Whether holding `transition:generic` makes `scope` redundant.
pub fn superseded_by_transition_generic(scope: &ParsedScope) -> bool {
match scope {
ParsedScope::Repo(_) | ParsedScope::Blob(_) => true,
ParsedScope::Rpc(rpc) => !rpc
.lxm
.as_deref()
.is_some_and(|lxm| lxm == "*" || lxm.starts_with("chat.bsky.")),
ParsedScope::Account(_)
| ParsedScope::Identity(_)
| ParsedScope::TransitionEmail
| ParsedScope::TransitionChat => false,
ParsedScope::Include(_) => false,
ParsedScope::TransitionGeneric | ParsedScope::Atproto | ParsedScope::Unknown(_) => false,
}
}
impl ScopePermissions {
pub fn has_scope(&self, scope: &str) -> bool {
self.scopes.contains(scope)
}
@@ -177,17 +158,22 @@ impl ScopePermissions {
}
pub fn assert_rpc(&self, aud: &str, lxm: &Nsid) -> Result<(), ScopeError> {
let is_chat = lxm.starts_with("chat.bsky.");
if is_chat && self.has_transition_chat {
return Ok(());
if lxm.starts_with("chat.bsky.") {
if self.has_transition_chat {
return Ok(());
}
if self.has_transition_generic && !self.has_transition_chat {
return Err(ScopeError::InsufficientScope {
required: "transition:chat.bsky".to_string(),
message: format!(
"Chat access requires transition:chat.bsky scope to call {}",
lxm
),
});
}
}
// `transition:generic` covers every lexicon except chat. Note it does not *block* chat:
// holding it must never remove access a granular `rpc:chat.bsky.*` scope would grant on
// its own, so chat requests fall through to the granular check below rather than
// failing here.
if self.has_transition_generic && !is_chat {
if self.has_transition_generic {
return Ok(());
}
@@ -212,24 +198,13 @@ impl ScopePermissions {
});
if has_permission {
return Ok(());
}
// Point a caller holding only `transition:generic` at the scope it actually needs,
// rather than at a granular rpc scope it probably did not mean to request.
Err(match is_chat && self.has_transition_generic {
true => ScopeError::InsufficientScope {
required: "transition:chat.bsky".to_string(),
message: format!(
"Chat access requires transition:chat.bsky scope to call {}",
lxm
),
},
false => ScopeError::InsufficientScope {
Ok(())
} else {
Err(ScopeError::InsufficientScope {
required: format!("rpc:{}?aud={}", lxm, aud),
message: format!("Insufficient scope to call {} on {}", lxm, aud),
},
})
})
}
}
pub fn assert_account(
@@ -237,6 +212,10 @@ impl ScopePermissions {
attr: AccountAttr,
action: AccountAction,
) -> Result<(), ScopeError> {
if self.has_transition_generic {
return Ok(());
}
if attr == AccountAttr::Email && action == AccountAction::Read && self.has_transition_email
{
return Ok(());
@@ -266,7 +245,8 @@ impl ScopePermissions {
}
pub fn allows_email_read(&self) -> bool {
self.has_transition_email
self.has_transition_generic
|| self.has_transition_email
|| self
.find_account_scopes()
.any(|a| a.attr == AccountAttr::Email || a.attr == AccountAttr::Wildcard)
@@ -289,6 +269,10 @@ impl ScopePermissions {
}
pub fn assert_identity(&self, attr: IdentityAttr) -> Result<(), ScopeError> {
if self.has_transition_generic {
return Ok(());
}
let has_permission = self.find_identity_scopes().any(|identity_scope| {
identity_scope.attr == IdentityAttr::Wildcard || identity_scope.attr == attr
});
@@ -352,7 +336,6 @@ impl Default for ScopePermissions {
#[cfg(test)]
mod tests {
use super::*;
use crate::parser::parse_scope;
fn c(s: &str) -> Nsid {
s.parse().unwrap()
@@ -529,10 +512,10 @@ mod tests {
}
#[test]
fn test_transition_generic_does_not_grant_identity() {
fn test_transition_generic_grants_identity() {
let perms = ScopePermissions::from_scope_string(Some("transition:generic"));
assert!(!perms.allows_identity(IdentityAttr::Handle));
assert!(!perms.allows_identity(IdentityAttr::Wildcard));
assert!(perms.allows_identity(IdentityAttr::Handle));
assert!(perms.allows_identity(IdentityAttr::Wildcard));
}
#[test]
@@ -614,160 +597,4 @@ mod tests {
&c("app.bsky.feed.getAuthorFeed")
));
}
#[test]
fn transition_generic_supersedes_granular_scopes() {
for scope in [
"repo:app.bsky.feed.post?action=create",
"blob:image/png",
"rpc:app.bsky.actor.getProfile?aud=*",
] {
assert!(
superseded_by_transition_generic(&parse_scope(scope)),
"{scope} should be superseded by transition:generic"
);
}
}
#[test]
fn transition_generic_does_not_supersede_chat() {
// assert_rpc rejects chat.bsky.* when transition:generic is held without
// transition:chat.bsky, so neither the transition scope nor an rpc scope that
// could reach a chat lexicon is covered by it.
for scope in [
"transition:chat.bsky",
"rpc:chat.bsky.convo.sendMessage?aud=*",
"rpc:*?aud=did:web:api.bsky.app",
"account:email?action=manage",
"account:email?action=read",
"account:status?action=read",
"identity:handle",
"identity:*",
"transition:email",
] {
assert!(
!superseded_by_transition_generic(&parse_scope(scope)),
"{scope} must not be treated as superseded"
);
}
}
#[test]
fn transition_generic_does_not_supersede_itself_or_baseline() {
assert!(!superseded_by_transition_generic(&parse_scope(
"transition:generic"
)));
assert!(!superseded_by_transition_generic(&parse_scope("atproto")));
}
#[test]
fn superseded_matches_enforcement_for_chat_and_feed() {
// Cross-check against ScopePermissions so the two cannot drift apart.
let perms = ScopePermissions::from_scope_string(Some("atproto transition:generic"));
let feed = Nsid::new("app.bsky.feed.getTimeline").unwrap();
let chat = Nsid::new("chat.bsky.convo.sendMessage").unwrap();
assert!(perms.allows_rpc("did:web:api.bsky.app", &feed));
assert!(!perms.allows_rpc("did:web:api.bsky.app", &chat));
}
#[test]
fn granular_chat_rpc_works_without_transition_generic() {
// Baseline for the test below: on its own, a granular chat rpc scope grants chat.
let perms = ScopePermissions::from_scope_string(Some(
"atproto rpc:chat.bsky.convo.sendMessage?aud=*",
));
assert!(perms.allows_rpc("did:web:api.bsky.chat", &c("chat.bsky.convo.sendMessage")));
}
#[test]
fn transition_generic_does_not_revoke_granular_chat_rpc() {
// Adding a broader scope must never remove access. transition:generic does not cover
// chat lexicons, but it must not stop a granular chat rpc scope from doing so either.
let perms = ScopePermissions::from_scope_string(Some(
"atproto transition:generic rpc:chat.bsky.convo.sendMessage?aud=*",
));
assert!(perms.allows_rpc("did:web:api.bsky.chat", &c("chat.bsky.convo.sendMessage")));
// ...and still grants everything else it covers.
assert!(perms.allows_rpc("did:web:api.bsky.app", &c("app.bsky.feed.getTimeline")));
}
#[test]
fn transition_generic_does_not_widen_granular_chat_rpc() {
// The granular scope grants exactly one chat lexicon; transition:generic must not be
// read as covering the rest of chat.
let perms = ScopePermissions::from_scope_string(Some(
"atproto transition:generic rpc:chat.bsky.convo.sendMessage?aud=*",
));
assert!(!perms.allows_rpc("did:web:api.bsky.chat", &c("chat.bsky.convo.deleteMessage")));
}
#[test]
fn chat_denial_still_names_the_scope_the_caller_needs() {
// transition:generic alone: the useful advice is "ask for transition:chat.bsky",
// not "ask for rpc:chat.bsky.convo.listConvos".
let generic = ScopePermissions::from_scope_string(Some("atproto transition:generic"));
let err = generic
.assert_rpc("did:web:api.bsky.chat", &c("chat.bsky.convo.listConvos"))
.expect_err("chat must be denied without transition:chat.bsky");
match err {
ScopeError::InsufficientScope { required, .. } => {
assert_eq!(required, "transition:chat.bsky");
}
other => panic!("unexpected error: {other:?}"),
}
// Without transition:generic the granular scope is the right thing to name.
let bare = ScopePermissions::from_scope_string(Some("atproto"));
let err = bare
.assert_rpc("did:web:api.bsky.chat", &c("chat.bsky.convo.listConvos"))
.expect_err("chat must be denied with no rpc scope at all");
match err {
ScopeError::InsufficientScope { required, .. } => {
assert!(required.starts_with("rpc:chat.bsky.convo.listConvos"));
}
other => panic!("unexpected error: {other:?}"),
}
}
#[test]
fn transition_generic_does_not_grant_account_management() {
// "no account management actions: change handle, change email, delete or deactivate
// account, migrate account" -- atproto OAuth spec.
let perms = ScopePermissions::from_scope_string(Some("atproto transition:generic"));
assert!(!perms.allows_account(AccountAttr::Email, AccountAction::Manage));
assert!(!perms.allows_account(AccountAttr::Repo, AccountAction::Manage));
assert!(!perms.allows_account(AccountAttr::Status, AccountAction::Manage));
}
#[test]
fn transition_generic_does_not_grant_email_read() {
// Reading the account email is what transition:email is for.
let perms = ScopePermissions::from_scope_string(Some("atproto transition:generic"));
assert!(!perms.allows_email_read());
assert!(!perms.allows_account(AccountAttr::Email, AccountAction::Read));
}
#[test]
fn granular_scopes_still_grant_alongside_transition_generic() {
// Removing the short-circuit must not stop an explicitly granted scope from working.
let perms = ScopePermissions::from_scope_string(Some(
"atproto transition:generic account:email?action=manage identity:handle",
));
assert!(perms.allows_account(AccountAttr::Email, AccountAction::Manage));
assert!(perms.allows_identity(IdentityAttr::Handle));
let with_email = ScopePermissions::from_scope_string(Some(
"atproto transition:generic transition:email",
));
assert!(with_email.allows_email_read());
}
#[test]
fn transition_generic_still_grants_what_the_spec_says_it_does() {
let perms = ScopePermissions::from_scope_string(Some("atproto transition:generic"));
assert!(perms.allows_repo(RepoAction::Create, &c("app.bsky.feed.post")));
assert!(perms.allows_repo(RepoAction::Delete, &c("app.bsky.feed.post")));
assert!(perms.allows_blob("image/png"));
assert!(perms.allows_rpc("did:web:api.bsky.app", &c("app.bsky.feed.getTimeline")));
}
}
+3 -17
View File
@@ -10,8 +10,7 @@ use tranquil_pds::comms::{CommsService, DiscordSender, EmailSender, SignalSender
use tranquil_pds::crawlers::{Crawlers, start_crawlers_service};
use tranquil_pds::scheduled::{
backfill_blob_ownership, backfill_record_blobs, backfill_repo_rev, backfill_user_blocks,
start_scheduled_tasks,
backfill_record_blobs, backfill_repo_rev, backfill_user_blocks, start_scheduled_tasks,
};
use tranquil_pds::state::AppState;
@@ -78,12 +77,9 @@ async fn main() -> ExitCode {
}
config
.server
.user_handle_domains
.user_handle_domain_list()
.iter()
.flatten()
.filter(|d| {
!tranquil_pds::api::validation::domain_forms_valid_handles(d.as_str())
})
.filter(|d| !tranquil_pds::api::validation::domain_forms_valid_handles(d))
.for_each(|d| {
eprintln!(
"account creation under handle domain {d} will be rejected because its TLD is reserved"
@@ -196,21 +192,11 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
let backfill_repo_repo = state.repos.repo.clone();
let backfill_block_store = state.block_store.clone();
let ownership_repo_repo = state.repos.repo.clone();
let ownership_infra_repo = state.repos.infra.clone();
let ownership_blob_repo = state.repos.blob.clone();
let ownership_block_store = state.block_store.clone();
tokio::spawn(async move {
tokio::join!(
backfill_repo_rev(backfill_repo_repo.clone(), backfill_block_store.clone()),
backfill_user_blocks(backfill_repo_repo.clone(), backfill_block_store.clone()),
backfill_record_blobs(backfill_repo_repo, backfill_block_store),
backfill_blob_ownership(
ownership_infra_repo,
ownership_repo_repo,
ownership_blob_repo,
ownership_block_store
)
);
});

Some files were not shown because too many files have changed in this diff Show More