Compare commits

..
Author SHA1 Message Date
nelind bfadc936c0 fix(api): dont verify signature or DID during importRepo 2026-06-11 23:23:37 +02:00
350 changed files with 9274 additions and 22822 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,25 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT DISTINCT r.record_cid AS \"record_cid!\"\n FROM records r\n WHERE r.repo_id = $1\n AND r.record_cid = ANY($2)\n AND NOT EXISTS (\n SELECT 1 FROM UNNEST($3::text[], $4::text[]) AS k(collection, rkey)\n WHERE k.collection = r.collection AND k.rkey = r.rkey\n )\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "record_cid!",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Uuid",
"TextArray",
"TextArray",
"TextArray"
]
},
"nullable": [
false
]
},
"hash": "105807a41c7337e7aa46bace29ab613030fd4fbf6845baddab9c0b2009972c02"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE invite_codes SET available_uses = available_uses - 1 WHERE code = $1 AND available_uses > 0 AND COALESCE(disabled, false) = false",
"query": "UPDATE invite_codes SET available_uses = available_uses - 1 WHERE code = $1",
"describe": {
"columns": [],
"parameters": {
@@ -10,5 +10,5 @@
},
"nullable": []
},
"hash": "0bb2cb6af37bff735b6b380697fc8e1fa2034ca0600e8c16e1e362b722192327"
"hash": "1ee6eda3e44660e7f14fcfe56adc2d41c72901b9c701fc7b992314e5370b32dc"
}
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT block_cid AS \"block_cid!\" FROM user_blocks\n WHERE user_id = $1 AND repo_rev > $2\n ORDER BY repo_rev ASC\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "block_cid!",
"type_info": "Bytea"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
false
]
},
"hash": "30570ed3866840d1258c8768a5c8a23ade40700c05ddbbf7fc4f64bfa95b1ed4"
}
@@ -0,0 +1,18 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE session_tokens SET access_jti = $1, refresh_jti = $2, access_expires_at = $3, refresh_expires_at = $4, updated_at = NOW() WHERE id = $5",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Timestamptz",
"Timestamptz",
"Int4"
]
},
"nullable": []
},
"hash": "4dcee809896ead3de8ca0433856ed424211d79df201d08bbea0e4c576931a234"
}
@@ -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,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM session_tokens WHERE id = $1 AND did = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int4",
"Text"
]
},
"nullable": []
},
"hash": "8003624cedbac8b094c83933578517abfb2eaf8e59d1d52c7ea59bf5d11cfcfe"
}
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM session_tokens WHERE access_jti = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text"
]
},
"nullable": []
},
"hash": "847ce3c34985d0957526c87e0a20c6b4e5daae08a338f7635def682ac0689cf6"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT block_cid AS \"block_cid!\" FROM user_blocks\n WHERE user_id = $1\n ORDER BY repo_rev ASC\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "block_cid!",
"type_info": "Bytea"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false
]
},
"hash": "85cc0cd1e62a30fa67d415b7a01164f962a422513e8f0737553321fd9987a56c"
}
@@ -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"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM session_tokens WHERE access_jti = $1 AND did = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "a27e93bc594babbada10afe5c3e33a65909ec69c579329916833e4b0fe2332d3"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, did, email, email_verified\n FROM users\n WHERE LOWER(email) = $1",
"query": "SELECT id, did, email, email_verified, handle\n FROM users\n WHERE LOWER(email) = $1",
"describe": {
"columns": [
{
@@ -22,6 +22,11 @@
"ordinal": 3,
"name": "email_verified",
"type_info": "Bool"
},
{
"ordinal": 4,
"name": "handle",
"type_info": "Text"
}
],
"parameters": {
@@ -33,8 +38,9 @@
false,
false,
true,
false,
false
]
},
"hash": "55b5e5cb13e09c4261e0452ef7e21240b504096666692c8fbd5c7dcd58607bc7"
"hash": "b230a27fce54d4f79de4ffcc754bf6c7e5a889623e64cdd7aaec791b3553ed83"
}
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT session_id FROM used_refresh_tokens WHERE refresh_jti = $1 FOR UPDATE",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "session_id",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "b2e1736dbe2ab9114e373353bcc299176417f3c9220025f9521591ba62928bd7"
}
@@ -1,82 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT u.used_at, st.id AS session_id, st.did, st.scope, st.controller_did,\n st.access_jti, st.refresh_jti, st.access_expires_at, st.refresh_expires_at,\n k.key_bytes, k.encryption_version\n FROM used_refresh_tokens u\n JOIN session_tokens st ON st.id = u.session_id\n JOIN users us ON st.did = us.did\n JOIN user_keys k ON us.id = k.user_id\n WHERE u.refresh_jti = $1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "used_at",
"type_info": "Timestamptz"
},
{
"ordinal": 1,
"name": "session_id",
"type_info": "Int4"
},
{
"ordinal": 2,
"name": "did",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "scope",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "controller_did",
"type_info": "Text"
},
{
"ordinal": 5,
"name": "access_jti",
"type_info": "Text"
},
{
"ordinal": 6,
"name": "refresh_jti",
"type_info": "Text"
},
{
"ordinal": 7,
"name": "access_expires_at",
"type_info": "Timestamptz"
},
{
"ordinal": 8,
"name": "refresh_expires_at",
"type_info": "Timestamptz"
},
{
"ordinal": 9,
"name": "key_bytes",
"type_info": "Bytea"
},
{
"ordinal": 10,
"name": "encryption_version",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false,
false,
true,
true,
false,
false,
false,
false,
false,
true
]
},
"hash": "c08e3760be75071ac5e281cb3b05b3a02dccbf35b96c3d50c652878a38c9f9e6"
}
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO used_refresh_tokens (refresh_jti, session_id)\n VALUES ($1, $2)\n ON CONFLICT (refresh_jti) DO NOTHING\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Int4"
]
},
"nullable": []
},
"hash": "c0a7b73070e6c5b9baa9caddca3c5d24427b48ed02b5cac913a0c0af1e5c9379"
}
@@ -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"
}
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT session_id FROM used_refresh_tokens WHERE refresh_jti = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "session_id",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "fcd868a192d27fd4eccae92a884e881b8d6f09bf7ae08a9b431a44acbf2f91f3"
}
-55
View File
@@ -4,7 +4,6 @@
In order of importance:
- If your change involves how Tranquil implements atproto make sure its correct! See more below.
- **You must run your change! Every contribution that says "here's xyz. untested." does not help the project.**
- Relevant tests to your PR must pass. The whole suite doesn't have to be proven to have run, because there are a *ton* of tests and they're quite heavy, but hopefully there are existing tests for whatever you're PRing, and if there aren't, please add those too.
- Run cargo fmt :P
@@ -17,60 +16,6 @@ Things that would also be nice but aren't like, a pain in our side:
- Big changes should be stacked PRs that are broken up into digestible pieces. Those stacked PRs should hopefully be able to be merged individually if necessary.
### How we define a "correct" PDS implementation
The atproto specs are notoriously imprecise, ambiguious,
lacks specifications for large parts of the protocol and network (even including what implementing a PDS entails!)
and is generally none specific.
This is bad.
We won't waste time here describing all the ways in which that is problematic,
the important thing for Tranquil is that this means that "follows spec" is not sufficient to describe a "correct" PDS implementation.
Thus we need to come up with a description of "correct".
In order of importance the following rules describe what "correct" means for Tranquil:
- The specs take precedence.
If the spec *is* specific enough then follow it.
Even if the reference implementation doesn't.
- If the specs aren't sufficiently specific
rely on the reference implementation, potential supporting documents or discussions,
and/or community sentiment or common sense.
If the matter is still debated and/or PBCs opinion differs from community sentiment we generally side with the community.
- Examples here include what features and APIs to implement,
here we look at what the reference implementation implements
as well as https://github.com/bluesky-social/atproto/discussions/2350 as a supporting document.
Another example is whether `include` scopes are allowed to use a `*` `aud` parameter.
Discussion here has happened in https://github.com/bluesky-social/atproto/issues/4490.
PBC has voiced an opinion that this should be disallowed,
community sentiment seems to strongly lean to allowing it. Tranquil allows it.
- Please mark locations like this with a `// SPECAMB: ...` comment explaining the ambiguity
and what parts of the reference implementation and/or supporting documents have been used as reference.
- If the reference implementation has behaviour that is only ever relevant for the Bluesky application.
Implementions of such behaviour **must** be gated behind a `bsky-support` cargo feature of the implementing crate.
- Examples here include bluesky feedgen specific service proxying behaviour,
the `app.bsky.actor.getPreferences` and `app.bsky.actor.putPreferences` APIs,
and special handling of the `X-BSKY-TOPICS` HTTP header during service proxying.
- Please add a comment next to these implementations with an explanation of the behaviour.
- Most of these behaviours are required for proper functioning of the official Bluesky client, though not all.
If the behaviour isn't required for the official client consider not implementing it.
- One such behaviour that we have a *hard rule* to never implement is default proxying to a configured Bluesky appview
for `app.bsky.*` APIs and as fallback for `com.atproto.repo.getRecord`.
Many third-party Bluesky clients rely on this behaviour, the official client used to do the same but does not anymore.
Third-party clients breaking because they don't specify an `atproto-proxy` header is thus *not* a Tranquil bug but a bug in said clients.
- Bluesky is the only application that will ever recieve application specific behaviour like this.
It does so only because such a big section of atproto usage is Bluesky
and because Bluesky is the only application that can practically rely on application specific behaviour.
Application specific behaviour for other applications may still be added to Tranquil if such behaviour is a Tranquil feature,
for example for Tranquils rudimentary banned content moderation feature,
and not something said application relies on for proper functioning.
There is bound to be edge cases that these rules don't fully cover.
Here common sense, community sentiment, furthering the goals of atproto itself, and ultimately maintainer opinion take precedence over support for any individual application.
Even Bluesky.
The rules above are meant to capture Tranquils goals of being correct while being community oriented and avoiding as much "Bluesky-defaultism" as possible.
Tranquil is a community atproto PDS, *not* a company-led Bluesky (or other atproto app) PDS.
See also "Tranquil & the world" in docs/1_WELCOME_TO_TRANQUIL_PDS.md.
## Local Development
### Prerequisites
Generated
+77 -89
View File
@@ -105,21 +105,6 @@ version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "250f629c0161ad8107cf89319e990051fae62832fd343083bea452d93e2205fd"
[[package]]
name = "alloc-no-stdlib"
version = "2.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cc7bb162ec39d46ab1ca8c77bf72e890535becd1751bb45f64c597edb4c8c6b3"
[[package]]
name = "alloc-stdlib"
version = "0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0e76a019e91224d279006ff972f1e984179a6e9feb050adba6ce8274aef23195"
dependencies = [
"alloc-no-stdlib",
]
[[package]]
name = "allocator-api2"
version = "0.2.21"
@@ -1265,27 +1250,6 @@ dependencies = [
"cfg_aliases",
]
[[package]]
name = "brotli"
version = "8.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5cc91aac060a7a1e25823bdccbfb6af1875b88f17c6daac97894eed8207166b3"
dependencies = [
"alloc-no-stdlib",
"alloc-stdlib",
"brotli-decompressor",
]
[[package]]
name = "brotli-decompressor"
version = "5.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3a32acac15fe1967bc3986b2a6347dffc965602354ea6f450ad07e8bfd253583"
dependencies = [
"alloc-no-stdlib",
"alloc-stdlib",
]
[[package]]
name = "bs58"
version = "0.5.1"
@@ -2522,6 +2486,7 @@ checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c"
dependencies = [
"crc32fast",
"miniz_oxide",
"zlib-rs",
]
[[package]]
@@ -7665,8 +7630,9 @@ dependencies = [
[[package]]
name = "tranquil-api"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"anyhow",
"axum",
"backon",
"base32",
@@ -7700,25 +7666,27 @@ dependencies = [
"tokio",
"tracing",
"tranquil-config",
"tranquil-db",
"tranquil-db-traits",
"tranquil-lexicon",
"tranquil-pds",
"tranquil-scopes",
"tranquil-signal",
"tranquil-types",
"urlencoding",
"uuid",
"webauthn-rs",
"zip",
]
[[package]]
name = "tranquil-auth"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"anyhow",
"base32",
"base64 0.22.1",
"bcrypt",
"brotli",
"chrono",
"hmac",
"k256",
@@ -7730,19 +7698,17 @@ dependencies = [
"totp-rs",
"tranquil-config",
"tranquil-crypto",
"tranquil-types",
"urlencoding",
"uuid",
]
[[package]]
name = "tranquil-cache"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"async-trait",
"base64 0.22.1",
"redis",
"thiserror 2.0.18",
"tokio-util",
"tracing",
"tranquil-config",
@@ -7752,7 +7718,7 @@ dependencies = [
[[package]]
name = "tranquil-comms"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"async-trait",
"base64 0.22.1",
@@ -7766,28 +7732,27 @@ dependencies = [
"rsa",
"secrecy",
"serde_json",
"sqlx",
"thiserror 2.0.18",
"tokio",
"tracing",
"tranquil-config",
"tranquil-db-traits",
"tranquil-signal",
"tranquil-types",
"uuid",
]
[[package]]
name = "tranquil-config"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"confique",
"serde",
"tranquil-types",
]
[[package]]
name = "tranquil-crypto"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"aes-gcm",
"base64 0.22.1",
@@ -7803,7 +7768,7 @@ dependencies = [
[[package]]
name = "tranquil-db"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"async-trait",
"chrono",
@@ -7820,7 +7785,7 @@ dependencies = [
[[package]]
name = "tranquil-db-traits"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"async-trait",
"base64 0.22.1",
@@ -7836,20 +7801,18 @@ dependencies = [
[[package]]
name = "tranquil-infra"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"async-trait",
"bytes",
"futures",
"serde",
"serde_json",
"thiserror 2.0.18",
"tranquil-types",
"tranquil-config",
]
[[package]]
name = "tranquil-lexicon"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"chrono",
"futures",
@@ -7861,15 +7824,14 @@ dependencies = [
"thiserror 2.0.18",
"tokio",
"tracing",
"tranquil-infra",
"tranquil-types",
"unicode-segmentation",
"urlencoding",
"wiremock",
]
[[package]]
name = "tranquil-oauth"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"anyhow",
"axum",
@@ -7886,16 +7848,14 @@ dependencies = [
"sqlx",
"tokio",
"tracing",
"tranquil-infra",
"tranquil-types",
"uuid",
]
[[package]]
name = "tranquil-oauth-server"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"async-trait",
"axum",
"base64 0.22.1",
"bcrypt",
@@ -7918,9 +7878,7 @@ dependencies = [
"tranquil-config",
"tranquil-crypto",
"tranquil-db-traits",
"tranquil-infra",
"tranquil-pds",
"tranquil-scopes",
"tranquil-types",
"urlencoding",
"uuid",
@@ -7929,7 +7887,7 @@ dependencies = [
[[package]]
name = "tranquil-pds"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"aes-gcm",
"anyhow",
@@ -7941,6 +7899,7 @@ dependencies = [
"base32",
"base64 0.22.1",
"bcrypt",
"bs58",
"bytes",
"chrono",
"ciborium",
@@ -7956,6 +7915,7 @@ dependencies = [
"hmac",
"http 1.4.0",
"image",
"infer",
"ipld-core",
"iroh-car",
"jacquard-common",
@@ -7997,9 +7957,9 @@ dependencies = [
"tranquil-cache",
"tranquil-comms",
"tranquil-config",
"tranquil-crypto",
"tranquil-db",
"tranquil-db-traits",
"tranquil-infra",
"tranquil-lexicon",
"tranquil-oauth",
"tranquil-oauth-server",
@@ -8016,11 +7976,12 @@ dependencies = [
"webauthn-rs",
"webauthn-rs-proto",
"wiremock",
"zip",
]
[[package]]
name = "tranquil-repo"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"bytes",
"cid",
@@ -8032,22 +7993,19 @@ dependencies = [
[[package]]
name = "tranquil-ripple"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"async-trait",
"backon",
"bincode 2.0.1",
"bytes",
"foca",
"futures",
"hkdf",
"metrics",
"parking_lot",
"quinn",
"rand 0.9.2",
"rcgen",
"rustls 0.23.37",
"serde",
"sha2",
"socket2 0.6.3",
"thiserror 2.0.18",
"tokio",
"tokio-util",
@@ -8056,12 +8014,11 @@ dependencies = [
"tranquil-config",
"tranquil-infra",
"uuid",
"x509-parser 0.18.1",
]
[[package]]
name = "tranquil-scopes"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"axum",
"futures",
@@ -8072,13 +8029,12 @@ dependencies = [
"thiserror 2.0.18",
"tokio",
"tracing",
"tranquil-types",
"urlencoding",
]
[[package]]
name = "tranquil-server"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"arc-swap",
"axum",
@@ -8095,7 +8051,6 @@ dependencies = [
"hyper-util",
"quinn",
"rcgen",
"reqwest",
"rustls 0.23.37",
"rustls-pemfile",
"thiserror 2.0.18",
@@ -8115,7 +8070,7 @@ dependencies = [
[[package]]
name = "tranquil-signal"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"async-trait",
"chrono",
@@ -8123,6 +8078,7 @@ dependencies = [
"futures",
"presage",
"rand 0.9.2",
"serde",
"serde_json",
"sqlx",
"tempfile",
@@ -8130,14 +8086,14 @@ dependencies = [
"tokio",
"tokio-util",
"tracing",
"tranquil-types",
"tranquil-signal",
"url",
"uuid",
]
[[package]]
name = "tranquil-storage"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"async-trait",
"aws-config",
@@ -8154,7 +8110,7 @@ dependencies = [
[[package]]
name = "tranquil-store"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"async-trait",
"bytes",
@@ -8194,6 +8150,7 @@ dependencies = [
"tranquil-db",
"tranquil-db-traits",
"tranquil-oauth",
"tranquil-repo",
"tranquil-store",
"tranquil-types",
"uuid",
@@ -8202,7 +8159,7 @@ dependencies = [
[[package]]
name = "tranquil-sync"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"anyhow",
"axum",
@@ -8224,22 +8181,15 @@ dependencies = [
[[package]]
name = "tranquil-types"
version = "0.6.7"
version = "0.6.5"
dependencies = [
"base64 0.22.1",
"chrono",
"cid",
"jacquard-common",
"rand 0.8.5",
"reqwest",
"serde",
"serde_json",
"sqlx",
"thiserror 2.0.18",
"tokio",
"tracing",
"url",
"uuid",
]
[[package]]
@@ -8291,6 +8241,12 @@ version = "2.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9ea3136b675547379c4bd395ca6b938e5ad3c3d20fad76e7fe85f9e0d011419c"
[[package]]
name = "typed-path"
version = "0.12.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8e28f89b80c87b8fb0cf04ab448d5dd0dd0ade2f8891bae878de66a75a28600e"
[[package]]
name = "typenum"
version = "1.19.0"
@@ -9492,6 +9448,20 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "zip"
version = "7.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c42e33efc22a0650c311c2ef19115ce232583abbe80850bc8b66509ebef02de0"
dependencies = [
"crc32fast",
"flate2",
"indexmap 2.13.0",
"memchr",
"typed-path",
"zopfli",
]
[[package]]
name = "zkcredential"
version = "0.1.0"
@@ -9540,12 +9510,30 @@ dependencies = [
"zkcredential",
]
[[package]]
name = "zlib-rs"
version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3be3d40e40a133f9c916ee3f9f4fa2d9d63435b5fbe1bfc6d9dae0aa0ada1513"
[[package]]
name = "zmij"
version = "1.0.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa"
[[package]]
name = "zopfli"
version = "0.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f05cd8797d63865425ff89b5c4a48804f35ba0ce8d125800027ad6017d2b5249"
dependencies = [
"bumpalo",
"crc32fast",
"log",
"simd-adler32",
]
[[package]]
name = "zune-core"
version = "0.5.1"
+5 -14
View File
@@ -26,7 +26,7 @@ members = [
]
[workspace.package]
version = "0.6.7"
version = "0.6.5"
edition = "2024"
license = "AGPL-3.0-or-later"
@@ -47,6 +47,7 @@ tranquil-db = { path = "crates/tranquil-db" }
tranquil-ripple = { path = "crates/tranquil-ripple" }
tranquil-lexicon = { path = "crates/tranquil-lexicon" }
tranquil-pds = { path = "crates/tranquil-pds" }
tranquil-server = { path = "crates/tranquil-server" }
tranquil-sync = { path = "crates/tranquil-sync" }
tranquil-oauth-server = { path = "crates/tranquil-oauth-server" }
tranquil-api = { path = "crates/tranquil-api" }
@@ -137,11 +138,10 @@ tower-layer = "0.3"
tracing = "0.1"
tracing-subscriber = "0.3"
urlencoding = "2.1"
url = "2.5"
uuid = { version = "1.19", features = ["v4", "v5", "v7", "fast-rng", "serde"] }
webauthn-rs = { version = "0.5", features = ["danger-allow-state-serialisation", "danger-user-presence-only-security-keys", "conditional-ui"] }
webauthn-rs-proto = "0.5"
x509-parser = "0.18"
zip = { version = "7.0", default-features = false, features = ["deflate"] }
ciborium = "0.2"
ctor = "0.6"
@@ -153,20 +153,11 @@ wiremock = "0.6"
curve25519-dalek = { git = "https://github.com/signalapp/curve25519-dalek", tag = "signal-curve25519-4.1.3" }
[profile.release]
opt-level = "s"
strip = true
panic = "abort"
codegen-units = 256
[profile.lewis]
inherits = "release"
opt-level = 3
lto = "fat"
strip = true
codegen-units = 1
panic = "abort"
[profile.bench]
opt-level = 3
lto = "fat"
codegen-units = 1
debug = 1
strip = false
+11 -35
View File
@@ -1,6 +1,4 @@
ARG DISTROLESS_IMAGE=gcr.io/distroless/cc-debian13:latest@sha256:a017e74bd2a12d98342dbecd33d121d2b160415ed777573dc1808969e989d94d
FROM node:24-trixie-slim AS frontend
FROM node:24-alpine AS frontend
RUN corepack enable && corepack prepare pnpm@latest --activate
WORKDIR /app
COPY frontend/package.json frontend/pnpm-lock.yaml frontend/pnpm-workspace.yaml ./
@@ -8,28 +6,8 @@ RUN pnpm install --frozen-lockfile
COPY frontend/ ./
RUN pnpm build
FROM rust:1.96-slim-trixie AS builder
RUN apt-get update && apt-get install -y --no-install-recommends \
ca-certificates pkg-config libssl-dev mold clang protobuf-compiler curl xz-utils \
&& rm -rf /var/lib/apt/lists/*
ARG COMPRESS="true"
RUN set -eux; \
if [ "$COMPRESS" = "true" ]; then \
arch="$(uname -m)"; \
case "$arch" in \
x86_64) upx_arch=amd64; upx_sha=ddc2654063fe4dc80d95b420788494e4db078ebb01a650692d623b5a9906e61e ;; \
aarch64) upx_arch=arm64; upx_sha=100310f74eb6f67694d1d0377f1c729b6a49238ce8c4de21ea2e7d3406186f8b ;; \
*) echo "upx: no prebuilt binary for $arch, skipping compression"; upx_arch="" ;; \
esac; \
if [ -n "$upx_arch" ]; then \
curl -fsSL -o /tmp/upx.tar.xz "https://github.com/upx/upx/releases/download/v5.0.2/upx-5.0.2-${upx_arch}_linux.tar.xz"; \
echo "${upx_sha} /tmp/upx.tar.xz" | sha256sum -c -; \
tar -xJf /tmp/upx.tar.xz -C /tmp; \
install -m0755 "/tmp/upx-5.0.2-${upx_arch}_linux/upx" /usr/local/bin/upx; \
rm -rf /tmp/upx.tar.xz "/tmp/upx-5.0.2-${upx_arch}_linux"; \
fi; \
fi
RUN mkdir -p /stage/var/lib/tranquil-pds/blobs /stage/var/lib/tranquil-pds/store
FROM rust:1.96-alpine AS builder
RUN apk add --no-cache ca-certificates musl-dev pkgconfig openssl-dev openssl-libs-static mold clang protoc
ENV RUSTFLAGS="-C linker=clang -C link-arg=-fuse-ld=mold"
WORKDIR /app
ARG SLIM="false"
@@ -58,23 +36,21 @@ COPY crates/tranquil-store ./crates/tranquil-store
COPY crates/tranquil-signal ./crates/tranquil-signal
COPY crates/tranquil-server ./crates/tranquil-server
COPY migrations ./migrations
RUN --mount=type=cache,id=cargo-registry,target=/usr/local/cargo/registry \
--mount=type=cache,id=cargo-git,target=/usr/local/cargo/git \
--mount=type=cache,id=tranquil-target,target=/app/target,sharing=locked \
RUN --mount=type=cache,target=/usr/local/cargo/registry \
--mount=type=cache,target=/app/target \
if [ "$SLIM" = "true" ]; then \
SQLX_OFFLINE=true cargo build --release -p tranquil-server --no-default-features; \
else \
SQLX_OFFLINE=true cargo build --release -p tranquil-server; \
fi && \
cp target/release/tranquil-server /tmp/tranquil-pds && \
if [ "$COMPRESS" = "true" ] && command -v upx >/dev/null 2>&1; then upx --best --lzma /tmp/tranquil-pds; fi
cp target/release/tranquil-server /tmp/tranquil-pds
FROM ${DISTROLESS_IMAGE}
FROM alpine:3.23
RUN apk add --no-cache ca-certificates
COPY --from=builder /tmp/tranquil-pds /usr/local/bin/tranquil-pds
COPY --from=builder --chown=65532:65532 /stage/var/lib/tranquil-pds /var/lib/tranquil-pds
COPY --from=frontend --chown=65532:65532 /app/dist /var/lib/tranquil-pds/frontend
WORKDIR /var/lib/tranquil-pds
COPY --from=frontend /app/dist /var/lib/tranquil-pds/frontend
WORKDIR /app
ENV SERVER_HOST=[::]
ENV SERVER_PORT=3000
EXPOSE 3000
ENTRYPOINT ["/usr/local/bin/tranquil-pds"]
CMD ["tranquil-pds"]
+7 -6
View File
@@ -1,13 +1,14 @@
FROM rust:1.96-slim-trixie
FROM rust:1.96-alpine
RUN apt-get update && apt-get install -y --no-install-recommends \
RUN apk add --no-cache \
ca-certificates \
pkg-config \
libssl-dev \
musl-dev \
pkgconfig \
openssl-dev \
openssl-libs-static \
mold \
clang \
protobuf-compiler \
&& rm -rf /var/lib/apt/lists/*
protoc
RUN cargo install cargo-watch
-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
-2
View File
@@ -1,7 +1,6 @@
[server]
hostname = "pds.test"
allow_http_proxy = true
allow_private_fetch = true
invite_code_required = false
disable_rate_limiting = true
@@ -11,7 +10,6 @@ dir = "/app/frontend/public"
[database]
url = "postgres://postgres:postgres@db:5432/pds"
max_connections = 20
[storage]
path = "/var/lib/tranquil-pds/blobs"
+4 -4
View File
@@ -8,10 +8,13 @@ license.workspace = true
tranquil-pds = { workspace = true }
tranquil-types = { workspace = true }
tranquil-config = { workspace = true }
tranquil-db = { workspace = true }
tranquil-db-traits = { workspace = true }
tranquil-lexicon = { workspace = true, features = ["resolve"] }
tranquil-scopes = { workspace = true }
tranquil-signal = { workspace = true }
anyhow = { workspace = true }
axum = { workspace = true }
backon = { workspace = true }
base32 = { workspace = true }
@@ -47,7 +50,4 @@ tracing = { workspace = true }
urlencoding = { workspace = true }
uuid = { workspace = true }
webauthn-rs = { workspace = true }
[features]
bsky = ["bsky-support"]
bsky-support = []
zip = { workspace = true }
+3
View File
@@ -0,0 +1,3 @@
mod preferences;
pub use preferences::{get_preferences, put_preferences};
@@ -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,
+10 -10
View File
@@ -8,7 +8,7 @@ use std::collections::HashMap;
use tranquil_pds::api::error::{ApiError, DbResultExt};
use tranquil_pds::auth::{Admin, Auth};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{Did, Handle, InviteCode};
use tranquil_pds::types::{Did, Handle};
#[derive(Deserialize)]
pub struct GetAccountInfoParams {
@@ -39,7 +39,7 @@ pub struct AccountInfo {
#[derive(Serialize, Clone)]
#[serde(rename_all = "camelCase")]
pub struct InviteCodeInfo {
pub code: InviteCode,
pub code: String,
pub available: i32,
pub disabled: bool,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -123,12 +123,12 @@ async fn get_invites_for_user(
return None;
}
let codes: Vec<InviteCode> = invite_codes.iter().map(|ic| ic.code.clone()).collect();
let code_strings: Vec<String> = invite_codes.iter().map(|ic| ic.code.clone()).collect();
let uses = state
.repos
.infra
.get_invite_code_uses_batch(&codes)
.get_invite_code_uses_batch(&code_strings)
.await
.ok()?;
@@ -157,7 +157,7 @@ async fn get_invites_for_user(
}
}
async fn get_invite_code_info(state: &AppState, code: &InviteCode) -> Option<InviteCodeInfo> {
async fn get_invite_code_info(state: &AppState, code: &str) -> Option<InviteCodeInfo> {
let info = state.repos.infra.get_invite_code_info(code).await.ok()??;
let uses = state
@@ -200,11 +200,11 @@ pub async fn get_account_infos(
return Err(ApiError::InvalidRequest("dids is required".into()));
}
let dids: Vec<Did> = dids.iter().filter_map(|d| d.parse().ok()).collect();
let dids_typed: Vec<Did> = dids.iter().filter_map(|d| d.parse().ok()).collect();
let accounts = state
.repos
.infra
.get_admin_account_infos_by_dids(&dids)
.get_admin_account_infos_by_dids(&dids_typed)
.await
.log_db_err("fetching account infos")?;
@@ -217,7 +217,7 @@ pub async fn get_account_infos(
.await
.unwrap_or_default();
let all_codes: Vec<InviteCode> = all_invite_codes
let all_codes: Vec<String> = all_invite_codes
.iter()
.map(|(_, c)| c.code.clone())
.collect();
@@ -233,7 +233,7 @@ pub async fn get_account_infos(
Vec::new()
};
let invited_by_map: HashMap<uuid::Uuid, InviteCode> = state
let invited_by_map: HashMap<uuid::Uuid, String> = state
.repos
.infra
.get_invite_code_uses_by_users(&user_ids)
@@ -249,7 +249,7 @@ pub async fn get_account_infos(
let (codes_by_user, code_info_map): (
HashMap<uuid::Uuid, Vec<InviteCodeInfo>>,
HashMap<InviteCode, InviteCodeInfo>,
HashMap<String, InviteCodeInfo>,
) = all_invite_codes.into_iter().fold(
(HashMap::new(), HashMap::new()),
|(mut by_user, mut by_code), (user_id, ic)| {
@@ -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
@@ -106,7 +106,7 @@ pub async fn update_account_handle(
}
let _ = state
.cache
.delete(&tranquil_pds::cache_keys::handle_key(&handle_for_check))
.delete(&tranquil_pds::cache_keys::handle_key(&handle))
.await;
if let Err(e) = tranquil_pds::repo_ops::sequence_identity_event(
&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
+18 -16
View File
@@ -10,13 +10,12 @@ use tranquil_pds::api::EmptyResponse;
use tranquil_pds::api::error::{ApiError, DbResultExt};
use tranquil_pds::auth::{Admin, Auth};
use tranquil_pds::state::AppState;
use tranquil_types::{Did, InviteCode};
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DisableInviteCodesInput {
pub codes: Option<Vec<InviteCode>>,
pub accounts: Option<Vec<Did>>,
pub codes: Option<Vec<String>>,
pub accounts: Option<Vec<String>>,
}
pub async fn disable_invite_codes(
@@ -29,14 +28,17 @@ pub async fn disable_invite_codes(
{
error!("DB error disabling invite codes: {:?}", e);
}
if let Some(accounts) = &input.accounts
&& let Err(e) = state
if let Some(accounts) = &input.accounts {
let accounts_typed: Vec<tranquil_types::Did> =
accounts.iter().filter_map(|a| a.parse().ok()).collect();
if let Err(e) = state
.repos
.infra
.disable_invite_codes_by_account(accounts)
.disable_invite_codes_by_account(&accounts_typed)
.await
{
error!("DB error disabling invite codes by account: {:?}", e);
{
error!("DB error disabling invite codes by account: {:?}", e);
}
}
Ok(Json(EmptyResponse {}))
}
@@ -51,7 +53,7 @@ pub struct GetInviteCodesParams {
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct InviteCodeInfo {
pub code: InviteCode,
pub code: String,
pub available: i32,
pub disabled: bool,
pub for_account: String,
@@ -70,7 +72,7 @@ pub struct InviteCodeUseInfo {
#[derive(Serialize)]
pub struct GetInviteCodesOutput {
#[serde(skip_serializing_if = "Option::is_none")]
pub cursor: Option<InviteCode>,
pub cursor: Option<String>,
pub codes: Vec<InviteCodeInfo>,
}
@@ -93,9 +95,9 @@ pub async fn get_invite_codes(
.log_db_err("fetching invite codes")?;
let user_ids: Vec<uuid::Uuid> = codes_rows.iter().map(|r| r.created_by_user).collect();
let code_values: Vec<InviteCode> = codes_rows.iter().map(|r| r.code.clone()).collect();
let code_strings: Vec<String> = codes_rows.iter().map(|r| r.code.clone()).collect();
let creator_dids: std::collections::HashMap<uuid::Uuid, Did> = state
let creator_dids: std::collections::HashMap<uuid::Uuid, tranquil_types::Did> = state
.repos
.infra
.get_user_dids_by_ids(&user_ids)
@@ -104,14 +106,14 @@ pub async fn get_invite_codes(
.into_iter()
.collect();
let uses_by_code = if code_values.is_empty() {
let uses_by_code = if code_strings.is_empty() {
std::collections::HashMap::new()
} else {
common::group_invite_uses_by_code(
state
.repos
.infra
.get_invite_code_uses_batch(&code_values)
.get_invite_code_uses_batch(&code_strings)
.await
.unwrap_or_default(),
|u| InviteCodeUseInfo {
@@ -165,7 +167,7 @@ pub async fn disable_account_invites(
if account.is_empty() {
return Err(ApiError::InvalidRequest("account is required".into()));
}
let account_did: Did = account
let account_did: tranquil_types::Did = account
.parse()
.map_err(|_| ApiError::InvalidDid("Invalid DID format".into()))?;
@@ -198,7 +200,7 @@ pub async fn enable_account_invites(
if account.is_empty() {
return Err(ApiError::InvalidRequest("account is required".into()));
}
let account_did: Did = account
let account_did: tranquil_types::Did = account
.parse()
.map_err(|_| ApiError::InvalidDid("Invalid DID format".into()))?;
+49 -82
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};
use tranquil_pds::types::{AtIdentifier, Did, Handle};
pub struct ResolvedRepo {
pub user_id: uuid::Uuid,
@@ -77,7 +76,7 @@ pub async fn resolve_repo_user_id(
pub fn group_invite_uses_by_code<U, F>(
uses: Vec<tranquil_db_traits::InviteCodeUse>,
map_use: F,
) -> HashMap<tranquil_types::InviteCode, Vec<U>>
) -> HashMap<String, Vec<U>>
where
F: Fn(tranquil_db_traits::InviteCodeUse) -> U,
{
@@ -157,80 +156,51 @@ 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,
)
}
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),
},
}
}
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);
}
}
pub fn create_self_hosted_did_web(handle: &str) -> Result<Did, ApiError> {
pub fn create_self_hosted_did_web(handle: &str) -> Result<String, ApiError> {
if !tranquil_pds::util::is_self_hosted_did_web_enabled() {
return Err(ApiError::SelfHostedDidWebDisabled);
}
let encoded_handle = handle.replace(':', "%3A");
Did::new(format!("did:web:{}", encoded_handle))
.map_err(|_| ApiError::InvalidHandle(Some("Handle is not a valid did:web".into())))
Ok(format!("did:web:{}", encoded_handle))
}
pub enum CredentialMatch {
@@ -246,10 +216,10 @@ pub async fn verify_credential(
session_repo: &dyn SessionRepository,
user_id: uuid::Uuid,
password: &str,
password_hash: Option<&PasswordHash>,
password_hash: Option<&str>,
) -> Option<CredentialMatch> {
let main_valid = password_hash
.map(|h| bcrypt::verify(password, h.as_str()).unwrap_or(false))
.map(|h| bcrypt::verify(password, h).unwrap_or(false))
.unwrap_or(false);
if main_valid {
return Some(CredentialMatch::MainPassword);
@@ -260,7 +230,7 @@ pub async fn verify_credential(
.unwrap_or_default();
app_passwords
.into_iter()
.find(|app| bcrypt::verify(password, app.password_hash.as_str()).unwrap_or(false))
.find(|app| bcrypt::verify(password, &app.password_hash).unwrap_or(false))
.map(|app| {
let scopes = app.scopes.unwrap_or_else(|| {
if app.privilege.is_privileged() {
@@ -277,16 +247,14 @@ pub async fn verify_credential(
})
}
pub fn hash_or_internal_error(value: &str) -> Result<PasswordHash, ApiError> {
bcrypt::hash(value, DEFAULT_COST)
.map(PasswordHash::new)
.map_err(|e| {
error!("Bcrypt hash error: {:?}", e);
ApiError::InternalError(None)
})
pub fn hash_or_internal_error(value: &str) -> Result<String, ApiError> {
bcrypt::hash(value, DEFAULT_COST).map_err(|e| {
error!("Bcrypt hash error: {:?}", e);
ApiError::InternalError(None)
})
}
pub async fn hash_password_async(password: &str) -> Result<PasswordHash, ApiError> {
pub async fn hash_password_async(password: &str) -> Result<String, ApiError> {
let password = password.to_string();
tokio::task::spawn_blocking(move || hash(password, DEFAULT_COST))
.await
@@ -294,7 +262,6 @@ pub async fn hash_password_async(password: &str) -> Result<PasswordHash, ApiErro
error!("Failed to spawn blocking task: {:?}", e);
ApiError::InternalError(None)
})?
.map(PasswordHash::new)
.map_err(|e| {
error!("Failed to hash password: {:?}", e);
ApiError::InternalError(None)
+57 -46
View File
@@ -12,12 +12,12 @@ use tranquil_pds::api::{
};
use tranquil_pds::auth::{Active, Auth};
use tranquil_pds::delegation::{
DelegationActionType, IdentityResolutionError, SCOPE_PRESETS, ValidatedDelegationScope,
verify_can_add_controllers, verify_can_control_accounts,
DelegationActionType, SCOPE_PRESETS, ValidatedDelegationScope, verify_can_add_controllers,
verify_can_control_accounts,
};
use tranquil_pds::rate_limit::{AccountCreationLimit, RateLimited};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{CidLink, Did, Handle};
use tranquil_pds::types::{Did, Handle};
pub async fn list_controllers(
State(state): State<AppState>,
@@ -38,10 +38,11 @@ pub async fn list_controllers(
async move {
if c.handle.is_none() {
c.handle = did_resolver
.fetch_did_document(&c.did)
.fetch_did_document(c.did.as_str())
.await
.ok()
.and_then(|doc| tranquil_types::did_doc::extract_handle(&doc));
.and_then(|doc| tranquil_types::did_doc::extract_handle(&doc))
.map(Into::into);
}
c
}
@@ -65,19 +66,19 @@ pub async fn add_controller(
) -> Result<Json<SuccessResponse>, ApiError> {
let resolved = tranquil_pds::delegation::resolve_identity(&state, &input.controller_did)
.await
.map_err(|e| match e {
IdentityResolutionError::PdsEndpoint(_) => ApiError::InvalidDelegation(
"Controller PDS endpoint isn't a usable https URL".into(),
),
IdentityResolutionError::DidResolution(_) => ApiError::ControllerNotFound,
})?;
.map_err(|_| ApiError::ControllerNotFound)?;
if !resolved.is_local
&& let Some(ref pds_url) = resolved.pds_url
{
if !pds_url.starts_with("https://") {
return Err(ApiError::InvalidDelegation(
"Controller PDS must use HTTPS".into(),
));
}
match state
.cross_pds_oauth
.check_remote_is_delegated(pds_url, &input.controller_did)
.check_remote_is_delegated(pds_url, input.controller_did.as_str())
.await
{
Some(true) => {
@@ -173,7 +174,9 @@ pub async fn remove_controller(
.session
.delete_app_passwords_by_controller(&auth.did, &input.controller_did)
.await
.unwrap_or(0);
.unwrap_or(0)
.try_into()
.unwrap_or(0usize);
let revoked_oauth_tokens = state
.repos
@@ -229,20 +232,6 @@ pub async fn update_controller_scopes(
.await
{
Ok(true) => {
let revoked_app_passwords = state
.repos
.session
.delete_app_passwords_by_controller(&auth.did, &input.controller_did)
.await
.unwrap_or(0);
let revoked_oauth_tokens = state
.repos
.oauth
.revoke_tokens_for_controller(&auth.did, &input.controller_did)
.await
.unwrap_or(0);
let _ = state
.repos
.delegation
@@ -252,9 +241,7 @@ pub async fn update_controller_scopes(
Some(&input.controller_did),
DelegationActionType::ScopesModified,
Some(json!({
"new_scopes": input.granted_scopes.as_str(),
"revoked_app_passwords": revoked_app_passwords,
"revoked_oauth_tokens": revoked_oauth_tokens
"new_scopes": input.granted_scopes.as_str()
})),
None,
None,
@@ -343,6 +330,7 @@ pub struct CreateDelegatedAccountInput {
pub handle: String,
pub email: Option<String>,
pub controller_scopes: ValidatedDelegationScope,
pub invite_code: Option<String>,
}
#[derive(Debug, Serialize)]
@@ -368,17 +356,31 @@ 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);
}
let validated_invite_code = if let Some(ref code) = input.invite_code {
match state.repos.infra.validate_invite_code(code).await {
Ok(validated) => Some(validated),
Err(_) => return Err(ApiError::InvalidInviteCode),
}
} else {
let invite_required = tranquil_config::get().server.invite_code_required;
if invite_required {
return Err(ApiError::InviteCodeRequired);
}
None
};
let plc = create_plc_did(&state, &handle).await.map_err(|e| {
tracing::error!("PLC DID creation failed: {:?}", e);
e
})?;
let did = plc.did;
let handle: Handle = handle.parse().map_err(|_| ApiError::InvalidHandle(None))?;
info!(did = %did, handle = %handle, controller = %can_control.did(), "Created DID for delegated account");
let repo = init_genesis_repo(&state, &did, &plc.signing_key, &plc.signing_key_bytes).await?;
@@ -392,18 +394,19 @@ pub async fn create_delegated_account(
controller_scopes: input.controller_scopes.as_str().to_string(),
encrypted_key_bytes: repo.encrypted_key_bytes,
encryption_version: tranquil_pds::config::ENCRYPTION_VERSION,
commit_cid: CidLink::from(&repo.commit_cid),
commit_cid: repo.commit_cid.to_string(),
repo_rev: repo.repo_rev.clone(),
genesis_block_cids: repo.genesis_block_cids,
invite_code: input.invite_code.clone(),
};
match state
let user_id = match state
.repos
.user
.create_delegated_account(&create_input)
.await
{
Ok(_) => {}
Ok(id) => id,
Err(tranquil_db_traits::CreateAccountError::HandleTaken) => {
return Err(ApiError::HandleNotAvailable(None));
}
@@ -414,6 +417,16 @@ pub async fn create_delegated_account(
error!("Error creating delegated account: {:?}", e);
return Err(ApiError::InternalError(None));
}
};
if let Some(validated) = validated_invite_code
&& let Err(e) = state
.repos
.infra
.record_invite_code_use(&validated, user_id)
.await
{
warn!("Failed to record invite code use for {}: {:?}", did, e);
}
crate::identity::provision::sequence_new_account(
@@ -463,26 +476,24 @@ pub async fn resolve_controller(
.parse()
.map_err(|_| ApiError::ControllerNotFound)?
} else {
let handle: Handle = identifier
.parse()
.map_err(|_| ApiError::ControllerNotFound)?;
let local_user = state.repos.user.get_by_handle(&handle).await.ok().flatten();
let local_handle: Option<Handle> = identifier.parse().ok();
let local_user = match local_handle {
Some(ref h) => state.repos.user.get_by_handle(h).await.ok().flatten(),
None => None,
};
match local_user {
Some(user) => user.did,
None => tranquil_pds::handle::resolve_handle(&handle)
None => tranquil_pds::handle::resolve_handle(identifier)
.await
.map_err(|_| ApiError::ControllerNotFound)?
.parse()
.map_err(|_| ApiError::ControllerNotFound)?,
}
};
let resolved = tranquil_pds::delegation::resolve_identity(&state, &did)
.await
.map_err(|e| match e {
IdentityResolutionError::PdsEndpoint(_) => ApiError::InvalidDelegation(
"Controller PDS endpoint isn't a usable https URL".into(),
),
IdentityResolutionError::DidResolution(_) => ApiError::ControllerNotFound,
})?;
.map_err(|_| ApiError::ControllerNotFound)?;
Ok(Json(resolved))
}
+21 -34
View File
@@ -143,40 +143,23 @@ 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();
}
};
let handle = match parse_start_handle(
interaction.data.as_ref().and_then(|d| d.options.as_deref()),
)
.map(Handle::new)
.transpose()
let handle = parse_start_handle(interaction.data.as_ref().and_then(|d| d.options.as_deref()));
if let Some(ref h) = handle
&& Handle::new(h).is_err()
{
Ok(h) => h,
Err(_) => {
return Json(json!({
"type": 4,
"data": {"content": "Invalid handle format. Handle should look like: nel.oyster.cafe", "flags": 64}
}))
.into_response();
}
};
return Json(json!({
"type": 4,
"data": {"content": "Invalid handle format. Handle should look like: alice.example.com", "flags": 64}
}))
.into_response();
}
debug!(
discord_username = %discord_username,
@@ -188,7 +171,7 @@ async fn handle_command(state: AppState, interaction: Interaction) -> Response {
match state
.repos
.user
.store_discord_user_id(&discord_username, &discord_user_id, handle.as_ref())
.store_discord_user_id(&discord_username, &discord_user_id, handle.as_deref())
.await
{
Ok(Some(user_id)) => {
@@ -197,14 +180,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 +295,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()),
);
}
}
+116 -72
View File
@@ -10,11 +10,10 @@ use serde::{Deserialize, Serialize};
use serde_json::json;
use tracing::{debug, error, info};
use tranquil_pds::api::error::ApiError;
use tranquil_pds::api::invite::check_registration_invite;
use tranquil_pds::auth::{ServiceTokenVerifier, extract_auth_token_from_header, is_service_token};
use tranquil_pds::rate_limit::{AccountCreationLimit, RateLimited};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{CidLink, Did, Handle, Nsid, PlainPassword};
use tranquil_pds::types::{Did, Handle, PlainPassword};
use tranquil_pds::validation::validate_password;
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
@@ -25,7 +24,7 @@ pub struct CreateAccountInput {
pub invite_code: Option<String>,
pub did: Option<String>,
pub did_type: Option<String>,
pub signing_key: Option<Did>,
pub signing_key: Option<String>,
pub verification_channel: Option<tranquil_db_traits::CommsChannel>,
pub discord_username: Option<String>,
pub telegram_username: Option<String>,
@@ -47,15 +46,23 @@ pub struct CreateAccountOutput {
async fn try_reactivate_migration(
state: &AppState,
did: &Did,
handle: &Handle,
did: &str,
handle: &str,
email: &Option<String>,
verification_channel: tranquil_db_traits::CommsChannel,
verification_recipient: Option<&tranquil_pds::comms::VerificationTarget>,
verification_recipient: Option<&str>,
) -> Option<Response> {
let did_typed: Did = match did.parse() {
Ok(d) => d,
Err(_) => return Some(ApiError::InternalError(Some("Invalid DID".into())).into_response()),
};
let handle_typed: Handle = match handle.parse() {
Ok(h) => h,
Err(_) => return Some(ApiError::InvalidHandle(None).into_response()),
};
let reactivate_input = tranquil_db_traits::MigrationReactivationInput {
did: did.clone(),
new_handle: handle.clone(),
did: did_typed.clone(),
new_handle: handle_typed.clone(),
new_email: email.clone(),
};
match state
@@ -65,7 +72,7 @@ async fn try_reactivate_migration(
.await
{
Ok(reactivated) => {
info!(did = %did, old_handle = ?reactivated.old_handle, new_handle = %handle, "Preparing existing account for inbound migration");
info!(did = %did, old_handle = %reactivated.old_handle, new_handle = %handle, "Preparing existing account for inbound migration");
let secret_key_bytes = match state
.repos
.user
@@ -112,7 +119,7 @@ async fn try_reactivate_migration(
}
};
let session_data = tranquil_db_traits::SessionTokenCreate {
did: did.clone(),
did: did_typed.clone(),
access_jti: access_meta.jti.clone(),
refresh_jti: refresh_meta.jti.clone(),
access_expires_at: access_meta.expires_at,
@@ -132,7 +139,7 @@ async fn try_reactivate_migration(
super::provision::enqueue_migration_verification(
state,
reactivated.user_id,
did,
&did_typed,
verification_channel,
recipient,
)
@@ -145,9 +152,14 @@ async fn try_reactivate_migration(
(
StatusCode::OK,
Json(CreateAccountOutput {
handle: handle.clone(),
did: did.clone(),
did_doc: state.did_resolver.fetch_did_document(did).await.ok(),
handle: handle.to_string().into(),
did: did_typed.clone(),
did_doc: state
.did_resolver
.fetch_did_document(did)
.await
.ok()
.map(|f| (*f).clone()),
access_jwt: access_meta.token,
refresh_jwt: refresh_meta.token,
verification_required,
@@ -197,10 +209,8 @@ pub async fn create_account(
let token = extracted.token;
if is_service_token(&token) {
let verifier = ServiceTokenVerifier::new();
let create_account_lxm = Nsid::new("com.atproto.server.createAccount")
.expect("com.atproto.server.createAccount is a valid NSID");
match verifier
.verify_service_token(&token, Some(&create_account_lxm))
.verify_service_token(&token, Some("com.atproto.server.createAccount"))
.await
{
Ok(claims) => {
@@ -271,8 +281,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();
}
@@ -297,7 +307,7 @@ pub async fn create_account(
};
let hostname = &cfg.server.hostname;
let key_result =
match super::provision::resolve_signing_key(&state, input.signing_key.as_ref()).await {
match super::provision::resolve_signing_key(&state, input.signing_key.as_deref()).await {
Ok(k) => k,
Err(e) => return e.into_response(),
};
@@ -305,7 +315,7 @@ pub async fn create_account(
let signing_key = key_result.signing_key;
let reserved_key_id = key_result.reserved_key_id;
let did_type = input.did_type.as_deref().unwrap_or("plc");
let did: Did = match did_type {
let did = match did_type {
"web" => {
let self_hosted_did = match common::create_self_hosted_did_web(&handle) {
Ok(d) => d,
@@ -330,42 +340,27 @@ pub async fn create_account(
}
if !is_did_web_byod
&& let Err(e) =
verify_did_web(d, hostname, &input.handle, input.signing_key.as_ref()).await
verify_did_web(d, hostname, &input.handle, input.signing_key.as_deref()).await
{
return ApiError::InvalidDid(e.to_string()).into_response();
}
info!(did = %d, "Creating external did:web account");
match d.parse() {
Ok(d) => d,
Err(_) => return ApiError::InvalidDid("Invalid DID format".into()).into_response(),
}
d.clone()
}
_ => {
if let Some(d) = &input.did {
if d.starts_with("did:plc:") && is_migration {
info!(did = %d, "Migration with existing did:plc");
match d.parse() {
Ok(d) => d,
Err(_) => {
return ApiError::InvalidDid("Invalid DID format".into())
.into_response();
}
}
d.clone()
} else if d.starts_with("did:web:") {
if !is_did_web_byod
&& let Err(e) =
verify_did_web(d, hostname, &input.handle, input.signing_key.as_ref())
verify_did_web(d, hostname, &input.handle, input.signing_key.as_deref())
.await
{
return ApiError::InvalidDid(e.to_string()).into_response();
}
match d.parse() {
Ok(d) => d,
Err(_) => {
return ApiError::InvalidDid("Invalid DID format".into())
.into_response();
}
}
d.clone()
} else if !d.trim().is_empty() {
return ApiError::InvalidDid(
"Only did:web DIDs can be provided; leave empty for did:plc. For migration with existing did:plc, provide service auth.".into()
@@ -393,17 +388,21 @@ pub async fn create_account(
&handle,
&email,
verification_channel,
verification_recipient.as_ref(),
verification_recipient.as_deref(),
)
.await
{
return response;
}
let handle_typed: Handle = match handle.parse() {
Ok(h) => h,
Err(_) => return ApiError::InvalidHandle(None).into_response(),
};
let handle_available = match state
.repos
.user
.check_handle_available_for_new_account(&handle)
.check_handle_available_for_new_account(&handle_typed)
.await
{
Ok(available) => available,
@@ -416,11 +415,40 @@ pub async fn create_account(
return ApiError::HandleTaken.into_response();
}
let invite_registration =
match check_registration_invite(&state, input.invite_code.as_deref()).await {
Ok(outcome) => outcome,
Err(e) => return e.into_response(),
};
let is_bootstrap = state.bootstrap_invite_code.is_some()
&& state.repos.user.count_users().await.unwrap_or(1) == 0;
if is_bootstrap {
match input.invite_code.as_deref() {
Some(code) if Some(code) == state.bootstrap_invite_code.as_deref() => {}
_ => return ApiError::InvalidInviteCode.into_response(),
}
} else {
let invite_code_required = tranquil_config::get().server.invite_code_required;
if invite_code_required
&& input
.invite_code
.as_ref()
.map(|c| c.trim().is_empty())
.unwrap_or(true)
{
return ApiError::InviteCodeRequired.into_response();
}
if let Some(code) = &input.invite_code
&& !code.trim().is_empty()
{
let valid = match state.repos.user.check_and_consume_invite_code(code).await {
Ok(v) => v,
Err(e) => {
error!("Error checking invite code: {:?}", e);
return ApiError::InternalError(None).into_response();
}
};
if !valid {
return ApiError::InvalidInviteCode.into_response();
}
}
}
if let Err(e) = validate_password(&input.password) {
return ApiError::InvalidRequest(e.to_string()).into_response();
@@ -437,15 +465,23 @@ pub async fn create_account(
None
};
let repo =
match super::provision::init_genesis_repo(&state, &did, &signing_key, &secret_key_bytes)
.await
{
Ok(r) => r,
Err(e) => return e.into_response(),
};
let commit_cid = CidLink::from(&repo.commit_cid);
let repo_rev = repo.repo_rev.clone();
let did_for_commit: Did = match did.parse() {
Ok(d) => d,
Err(_) => return ApiError::InternalError(Some("Invalid DID".into())).into_response(),
};
let repo = match super::provision::init_genesis_repo(
&state,
&did_for_commit,
&signing_key,
&secret_key_bytes,
)
.await
{
Ok(r) => r,
Err(e) => return e.into_response(),
};
let commit_cid_str = repo.commit_cid.to_string();
let rev_str = repo.repo_rev.clone();
let birthdate_pref = if tranquil_config::get().server.age_assurance_override {
Some(json!({
@@ -465,9 +501,9 @@ pub async fn create_account(
let repo_for_seq = repo.clone();
let create_input = tranquil_db_traits::CreatePasswordAccountInput {
handle: handle.clone(),
handle: handle_typed.clone(),
email: email.clone(),
did: did.clone(),
did: did_for_commit.clone(),
password_hash,
preferred_comms_channel,
discord_username: comms.discord,
@@ -478,10 +514,14 @@ pub async fn create_account(
encrypted_key_bytes: repo.encrypted_key_bytes,
encryption_version: tranquil_pds::config::ENCRYPTION_VERSION,
reserved_key_id,
commit_cid: commit_cid.clone(),
repo_rev: repo_rev.clone(),
commit_cid: commit_cid_str.clone(),
repo_rev: rev_str.clone(),
genesis_block_cids: repo.genesis_block_cids,
invite_code: invite_registration.into_invite_code(),
invite_code: if is_bootstrap {
None
} else {
input.invite_code.clone()
},
birthdate_pref,
};
@@ -501,9 +541,6 @@ pub async fn create_account(
Err(tranquil_db_traits::CreateAccountError::DidExists) => {
return ApiError::AccountAlreadyExists.into_response();
}
Err(tranquil_db_traits::CreateAccountError::InviteCodeUnavailable) => {
return ApiError::InvalidInviteCode.into_response();
}
Err(e) => {
error!("Error creating password account: {:?}", e);
return ApiError::InternalError(None).into_response();
@@ -511,15 +548,21 @@ pub async fn create_account(
};
let user_id = create_result.user_id;
if !is_migration && !is_did_web_byod {
super::provision::sequence_new_account(&state, &did, &handle, &repo_for_seq, &input.handle)
.await;
super::provision::sequence_new_account(
&state,
&did_for_commit,
&handle_typed,
&repo_for_seq,
&input.handle,
)
.await;
}
if !is_migration {
if let Some(ref recipient) = verification_recipient {
super::provision::enqueue_signup_verification(
&state,
user_id,
&did,
&did_for_commit,
verification_channel,
recipient,
)
@@ -529,7 +572,7 @@ pub async fn create_account(
super::provision::enqueue_migration_verification(
&state,
user_id,
&did,
&did_for_commit,
verification_channel,
recipient,
)
@@ -539,6 +582,7 @@ pub async fn create_account(
let session = match super::provision::create_and_store_session(
&state,
&did,
&did_for_commit,
&secret_key_bytes,
"transition:generic transition:chat.bsky",
None,
@@ -561,9 +605,9 @@ pub async fn create_account(
(
StatusCode::OK,
Json(CreateAccountOutput {
handle: handle.clone(),
did,
did_doc,
handle: handle.clone().into(),
did: did_for_commit,
did_doc: did_doc.map(|f| (*f).clone()),
access_jwt: session.access_jwt,
refresh_jwt: session.refresh_jwt,
verification_required: !is_migration,
+69 -52
View File
@@ -19,7 +19,7 @@ use tranquil_pds::rate_limit::{
HandleUpdateDailyLimit, HandleUpdateLimit, check_user_rate_limit_with_message,
};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{Did, Handle};
use tranquil_pds::types::Handle;
use tranquil_pds::util::get_header_str;
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -44,18 +44,16 @@ pub async fn resolve_handle(
if handle_str.is_empty() {
return ApiError::InvalidRequest("handle is required".into()).into_response();
}
let cache_key = tranquil_pds::cache_keys::handle_key(handle_str);
if let Some(did) = state.cache.get(&cache_key).await {
return DidResponse::response(did).into_response();
}
let handle: Handle = match handle_str.parse() {
Ok(h) => h,
Err(_) => {
return ApiError::InvalidHandle(Some("Invalid handle format".into())).into_response();
}
};
let cache_key = tranquil_pds::cache_keys::handle_key(&handle);
if let Some(cached) = state.cache.get(&cache_key).await
&& let Ok(did) = tranquil_pds::Did::new(cached)
{
return DidResponse::response(did).into_response();
}
let user = state.repos.user.get_by_handle(&handle).await;
match user {
Ok(Some(row)) => {
@@ -65,7 +63,7 @@ pub async fn resolve_handle(
.await;
DidResponse::response(row.did).into_response()
}
Ok(None) => match tranquil_pds::handle::resolve_handle(&handle).await {
Ok(None) => match tranquil_pds::handle::resolve_handle(handle.as_str()).await {
Ok(did) => {
let _ = state
.cache
@@ -132,9 +130,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;
}
@@ -158,15 +159,15 @@ pub async fn well_known_did(State(state): State<AppState>, headers: HeaderMap) -
async fn serve_handle_did_doc(state: &AppState, handle: &str, hostname: &str) -> Response {
let encoded_handle = handle.replace(':', "%3A");
let expected_did: tranquil_pds::types::Did = match format!("did:web:{}", encoded_handle).parse()
{
let expected_did = format!("did:web:{}", encoded_handle);
let expected_did_typed: tranquil_pds::types::Did = match expected_did.parse() {
Ok(d) => d,
Err(_) => return ApiError::InvalidRequest("Invalid DID format".into()).into_response(),
};
let user = match state
.repos
.user
.get_user_for_did_doc_build(&expected_did)
.get_user_for_did_doc_build(&expected_did_typed)
.await
{
Ok(Some(u)) => u,
@@ -211,7 +212,8 @@ async fn serve_handle_did_doc(state: &AppState, handle: &str, hostname: &str) ->
pub async fn user_did_doc(State(state): State<AppState>, Path(handle): Path<String>) -> Response {
let hostname = &tranquil_config::get().server.hostname;
let hostname_for_handles = tranquil_config::get().server.hostname_without_port();
let current_handle: Handle = match format!("{}.{}", handle, hostname_for_handles).parse() {
let current_handle = format!("{}.{}", handle, hostname_for_handles);
let current_handle_typed: Handle = match current_handle.parse() {
Ok(h) => h,
Err(_) => {
return ApiError::InvalidHandle(Some("Invalid handle format".into())).into_response();
@@ -220,7 +222,7 @@ pub async fn user_did_doc(State(state): State<AppState>, Path(handle): Path<Stri
let user = match state
.repos
.user
.get_did_web_info_by_handle(&current_handle)
.get_did_web_info_by_handle(&current_handle_typed)
.await
{
Ok(Some(u)) => u,
@@ -345,7 +347,7 @@ pub async fn verify_did_web(
did: &str,
hostname: &str,
handle: &str,
expected_signing_key: Option<&Did>,
expected_signing_key: Option<&str>,
) -> Result<(), DidWebVerifyError> {
let hostname_for_handles = hostname.split(':').next().unwrap_or(hostname);
let subdomain_host = format!("{}.{}", handle, hostname_for_handles);
@@ -424,15 +426,10 @@ pub async fn verify_did_web(
let expected_multibase = expected_signing_key
.strip_prefix("did:key:")
.ok_or(DidWebVerifyError::InvalidSigningKey)?;
let did_prefixed_key_id = format!("{}#atproto", did);
let has_matching_key = verification_methods.iter().any(|vm| {
let is_atproto_method = vm["id"]
vm["publicKeyMultibase"]
.as_str()
.is_some_and(|id| id == "#atproto" || id == did_prefixed_key_id);
is_atproto_method
&& vm["publicKeyMultibase"]
.as_str()
.is_some_and(|pk| pk == expected_multibase)
.is_some_and(|pk| pk == expected_multibase)
});
if !has_matching_key {
return Err(DidWebVerifyError::KeyMismatch(
@@ -494,10 +491,16 @@ pub async fn get_recommended_did_credentials(
let rotation_keys = if auth.did.starts_with("did:web:") {
vec![]
} else {
tranquil_pds::plc::rotation_keys_for(
tranquil_config::get().secrets.plc_rotation_key.as_deref(),
&signing_key,
)
let server_rotation_key = match &tranquil_config::get().secrets.plc_rotation_key {
Some(key) => key.clone(),
None => {
warn!(
"PLC_ROTATION_KEY not set, falling back to user's signing key for rotation key recommendation"
);
did_key.clone()
}
};
vec![server_rotation_key]
};
Ok(Json(GetRecommendedDidCredentialsOutput {
rotation_keys,
@@ -579,24 +582,35 @@ 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 = 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() {
let handle_typed: Handle = match full_handle.parse() {
Ok(h) => h,
Err(_) => return Err(ApiError::InvalidHandle(None)),
};
if let Err(e) =
tranquil_pds::repo_ops::sequence_identity_event(&state, &did, Some(&handle)).await
tranquil_pds::repo_ops::sequence_identity_event(&state, &did, Some(&handle_typed))
.await
{
warn!("Failed to sequence identity event for handle update: {}", e);
}
@@ -614,21 +628,21 @@ pub async fn update_handle(
return Err(ApiError::InvalidHandle(Some("Handle too long".into())));
}
full_handle
.parse()
.map_err(|_| ApiError::InvalidHandle(Some("Invalid handle format".into())))?
} else {
let handle: Handle = new_handle
.parse()
.map_err(|_| ApiError::InvalidHandle(Some("Invalid handle format".into())))?;
if new_handle == current_handle {
let handle_typed: Handle = match new_handle.parse() {
Ok(h) => h,
Err(_) => return Err(ApiError::InvalidHandle(None)),
};
if let Err(e) =
tranquil_pds::repo_ops::sequence_identity_event(&state, &did, Some(&handle)).await
tranquil_pds::repo_ops::sequence_identity_event(&state, &did, Some(&handle_typed))
.await
{
warn!("Failed to sequence identity event for handle update: {}", e);
}
return Ok(Json(EmptyResponse {}));
}
match tranquil_pds::handle::verify_handle_ownership(&handle, &did).await {
match tranquil_pds::handle::verify_handle_ownership(&new_handle, &did).await {
Ok(()) => {}
Err(tranquil_pds::handle::HandleResolutionError::NotFound) => {
return Err(ApiError::HandleNotAvailable(None));
@@ -647,12 +661,15 @@ pub async fn update_handle(
))));
}
}
handle
new_handle.clone()
};
let handle_typed: Handle = handle
.parse()
.map_err(|_| ApiError::InvalidHandle(Some("Invalid handle format".into())))?;
let handle_exists = state
.repos
.user
.check_handle_exists(&handle, user_id)
.check_handle_exists(&handle_typed, user_id)
.await
.log_db_err("checking handle existence")?;
if handle_exists {
@@ -661,7 +678,7 @@ pub async fn update_handle(
state
.repos
.user
.update_handle(user_id, &handle)
.update_handle(user_id, &handle_typed)
.await
.map_err(|e| {
error!("DB error updating handle: {:?}", e);
@@ -679,11 +696,11 @@ pub async fn update_handle(
.delete(&tranquil_pds::cache_keys::handle_key(&handle))
.await;
if let Err(e) =
tranquil_pds::repo_ops::sequence_identity_event(&state, &did, Some(&handle)).await
tranquil_pds::repo_ops::sequence_identity_event(&state, &did, Some(&handle_typed)).await
{
warn!("Failed to sequence identity event for handle update: {}", e);
}
if let Err(e) = update_plc_handle(&state, &did, &handle).await {
if let Err(e) = update_plc_handle(&state, &did, &handle_typed).await {
warn!("Failed to update PLC handle: {}", e);
}
Ok(Json(EmptyResponse {}))
+3 -2
View File
@@ -26,9 +26,10 @@ pub async fn verify_handle_ownership(
_rate_limit: RateLimited<HandleVerificationLimit>,
Json(input): Json<VerifyHandleOwnershipInput>,
) -> Response {
let handle_str = input.handle.as_str();
let did_str = input.did.as_str();
let dns_mismatch = match tranquil_pds::handle::resolve_handle_dns(&input.handle).await {
let dns_mismatch = match tranquil_pds::handle::resolve_handle_dns(handle_str).await {
Ok(did) if did == did_str => {
return Json(VerifyHandleOwnershipOutput {
verified: true,
@@ -44,7 +45,7 @@ pub async fn verify_handle_ownership(
Err(_) => None,
};
match tranquil_pds::handle::resolve_handle_http(&input.handle).await {
match tranquil_pds::handle::resolve_handle_http(handle_str).await {
Ok(did) if did == did_str => Json(VerifyHandleOwnershipOutput {
verified: true,
method: Some("http".to_string()),
@@ -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 {}))
+12 -7
View File
@@ -67,14 +67,19 @@ pub async fn submit_plc_operation(
})?;
let user_did_key = signing_key_to_did_key(&signing_key);
let server_rotation_key = tranquil_config::get()
.secrets
.plc_rotation_key
.clone()
.unwrap_or_else(|| user_did_key.clone());
if let Some(rotation_keys) = op.get("rotationKeys").and_then(Value::as_array) {
let rotation_key_strs: Vec<&str> = rotation_keys.iter().filter_map(Value::as_str).collect();
if let Some(missing) = tranquil_pds::plc::missing_required_rotation_key(
&rotation_key_strs,
&user_did_key,
tranquil_config::get().secrets.plc_rotation_key.as_deref(),
) {
return Err(ApiError::InvalidRequest(missing.message().into()));
let has_server_key = rotation_keys
.iter()
.any(|k| k.as_str() == Some(&server_rotation_key));
if !has_server_key {
return Err(ApiError::InvalidRequest(
"Rotation keys do not include server's rotation key".into(),
));
}
}
if let Some(services) = op.get("services").and_then(Value::as_object)
+53 -51
View File
@@ -1,4 +1,4 @@
use jacquard_common::types::{integer::LimitedU32, string::Tid as JacquardTid};
use jacquard_common::types::{integer::LimitedU32, string::Tid};
use jacquard_repo::{mst::Mst, storage::BlockStore};
use k256::ecdsa::SigningKey;
use std::sync::Arc;
@@ -6,7 +6,7 @@ use tranquil_db_traits::CommsChannel;
use tranquil_pds::api::error::ApiError;
use tranquil_pds::repo_ops::create_signed_commit;
use tranquil_pds::state::AppState;
use tranquil_pds::types::{CidLink, Did, Handle, Tid};
use tranquil_pds::types::{Did, Handle};
pub struct PlcDidResult {
pub did: Did,
@@ -14,7 +14,7 @@ pub struct PlcDidResult {
pub signing_key: SigningKey,
}
pub async fn create_plc_did(state: &AppState, handle: &Handle) -> Result<PlcDidResult, ApiError> {
pub async fn create_plc_did(state: &AppState, handle: &str) -> Result<PlcDidResult, ApiError> {
use k256::SecretKey;
use rand::rngs::OsRng;
@@ -25,7 +25,10 @@ pub async fn create_plc_did(state: &AppState, handle: &Handle) -> Result<PlcDidR
ApiError::InternalError(None)
})?;
let did = submit_plc_genesis(state, &signing_key, handle).await?;
let did_str = submit_plc_genesis(state, &signing_key, handle).await?;
let did: Did = did_str
.parse()
.map_err(|_| ApiError::InternalError(Some("PLC genesis returned invalid DID".into())))?;
Ok(PlcDidResult {
did,
@@ -37,14 +40,20 @@ pub async fn create_plc_did(state: &AppState, handle: &Handle) -> Result<PlcDidR
pub async fn submit_plc_genesis(
state: &AppState,
signing_key: &SigningKey,
handle: &Handle,
) -> Result<Did, ApiError> {
handle: &str,
) -> Result<String, ApiError> {
let hostname = &tranquil_config::get().server.hostname;
let pds_endpoint = format!("https://{}", hostname);
let rotation_key = tranquil_config::get()
.secrets
.plc_rotation_key
.clone()
.unwrap_or_else(|| tranquil_pds::plc::signing_key_to_did_key(signing_key));
let genesis_result = tranquil_pds::plc::create_genesis_operation(
signing_key,
tranquil_config::get().secrets.plc_rotation_key.as_deref(),
&rotation_key,
handle,
&pds_endpoint,
)
@@ -71,7 +80,7 @@ pub struct GenesisRepo {
pub encrypted_key_bytes: Vec<u8>,
pub commit_cid: cid::Cid,
pub mst_root_cid: cid::Cid,
pub repo_rev: Tid,
pub repo_rev: String,
pub genesis_block_cids: Vec<Vec<u8>>,
}
@@ -93,8 +102,8 @@ pub async fn init_genesis_repo(
ApiError::InternalError(None)
})?;
let rev = JacquardTid::now(LimitedU32::MIN);
let (commit_bytes, _sig) = create_signed_commit(did, mst_root, &rev, None, signing_key)
let rev = Tid::now(LimitedU32::MIN);
let (commit_bytes, _sig) = create_signed_commit(did, mst_root, rev.as_ref(), None, signing_key)
.map_err(|e| {
tracing::error!("Error creating genesis commit: {:?}", e);
ApiError::InternalError(None)
@@ -109,7 +118,7 @@ pub async fn init_genesis_repo(
encrypted_key_bytes,
commit_cid,
mst_root_cid: mst_root,
repo_rev: Tid::from(rev.clone()),
repo_rev: rev.as_ref().to_string(),
genesis_block_cids: vec![mst_root.to_bytes(), commit_cid.to_bytes()],
})
}
@@ -122,7 +131,7 @@ pub struct SigningKeyResult {
pub async fn resolve_signing_key(
state: &AppState,
signing_key_did: Option<&Did>,
signing_key_did: Option<&str>,
) -> Result<SigningKeyResult, ApiError> {
match signing_key_did {
Some(key_did) => {
@@ -164,13 +173,6 @@ pub async fn resolve_signing_key(
}
}
#[cfg_attr(
not(feature = "bsky"),
expect(
unused_variables,
reason = "only the bsky block writes display_name into the default profile record"
)
)]
pub async fn sequence_new_account(
state: &AppState,
did: &Did,
@@ -205,31 +207,27 @@ pub async fn sequence_new_account(
if let Err(e) = tranquil_pds::repo_ops::sequence_sync_event(
state,
did,
&CidLink::from(&repo.commit_cid),
&repo.commit_cid.to_string(),
Some(&repo.repo_rev),
)
.await
{
tracing::warn!("Failed to sequence sync event for {}: {}", did, e);
}
// TODO: make this configurable and also deduplicate with tranquil-oauth-server/src/sso_endpoints.rs:1210
#[cfg(feature = "bsky")]
let profile_record = serde_json::json!({
"$type": "app.bsky.actor.profile",
"displayName": display_name
});
if let Err(e) = tranquil_pds::repo_ops::create_record_internal(
state,
did,
&tranquil_pds::types::PROFILE_COLLECTION,
&tranquil_pds::types::PROFILE_RKEY,
&profile_record,
)
.await
{
let profile_record = serde_json::json!({
"$type": "app.bsky.actor.profile",
"displayName": display_name
});
if let Err(e) = tranquil_pds::repo_ops::create_record_internal(
state,
did,
&tranquil_pds::types::PROFILE_COLLECTION,
&tranquil_pds::types::PROFILE_RKEY,
&profile_record,
)
.await
{
tracing::warn!("Failed to create default profile for {}: {}", did, e);
};
tracing::warn!("Failed to create default profile for {}: {}", did, e);
}
}
@@ -266,23 +264,25 @@ pub struct SessionResult {
pub async fn create_and_store_session(
state: &AppState,
did_str: &str,
did: &Did,
signing_key_bytes: &[u8],
scope: &str,
controller_did: Option<&Did>,
) -> Result<SessionResult, ApiError> {
let access_meta = tranquil_pds::auth::create_access_token_with_metadata(did, signing_key_bytes)
.map_err(|e| {
tracing::error!("Error creating access token: {:?}", e);
ApiError::InternalError(None)
})?;
let refresh_meta =
tranquil_pds::auth::create_refresh_token_with_metadata(did, signing_key_bytes).map_err(
let access_meta =
tranquil_pds::auth::create_access_token_with_metadata(did_str, signing_key_bytes).map_err(
|e| {
tracing::error!("Error creating refresh token: {:?}", e);
tracing::error!("Error creating access token: {:?}", e);
ApiError::InternalError(None)
},
)?;
let refresh_meta =
tranquil_pds::auth::create_refresh_token_with_metadata(did_str, signing_key_bytes)
.map_err(|e| {
tracing::error!("Error creating refresh token: {:?}", e);
ApiError::InternalError(None)
})?;
let session_data = tranquil_db_traits::SessionTokenCreate {
did: did.clone(),
access_jti: access_meta.jti.clone(),
@@ -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,
)
+15 -42
View File
@@ -1,12 +1,5 @@
// BSKY: Bluesky requires PDSs to implement its private preferences API
#[cfg(feature = "bsky-support")]
pub mod actor {
mod preferences;
pub use preferences::{get_preferences, put_preferences};
}
pub mod actor;
pub mod admin;
#[cfg(feature = "bsky")]
pub mod age_assurance;
pub mod common;
pub mod delegation;
@@ -28,7 +21,7 @@ pub fn api_routes() -> axum::Router<AppState> {
let blob_body_limit =
DefaultBodyLimit::max(tranquil_config::get().server.max_blob_size as usize);
let router = axum::Router::new()
axum::Router::new()
.route("/_health", get(server::health))
.route(
"/com.atproto.server.describeServer",
@@ -380,6 +373,14 @@ pub fn api_routes() -> axum::Router<AppState> {
post(admin::update_subject_status),
)
.route("/com.atproto.admin.sendEmail", post(admin::send_email))
.route(
"/app.bsky.actor.getPreferences",
get(actor::get_preferences),
)
.route(
"/app.bsky.actor.putPreferences",
post(actor::put_preferences),
)
.route(
"/com.atproto.temp.checkSignupQueue",
get(temp::check_signup_queue),
@@ -437,21 +438,7 @@ pub fn api_routes() -> axum::Router<AppState> {
.route(
"/_delegation.resolveController",
get(delegation::resolve_controller),
);
#[cfg(feature = "bsky-support")]
let router = router
.route(
"/app.bsky.actor.getPreferences",
get(actor::get_preferences),
)
.route(
"/app.bsky.actor.putPreferences",
post(actor::put_preferences),
);
#[cfg(feature = "bsky")]
let router = router
.route(
"/app.bsky.ageassurance.getState",
get(age_assurance::get_state),
@@ -459,23 +446,15 @@ pub fn api_routes() -> axum::Router<AppState> {
.route(
"/app.bsky.unspecced.getAgeAssuranceState",
get(age_assurance::get_age_assurance_state),
);
router
)
}
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> {
@@ -495,15 +474,9 @@ pub fn webhook_routes() -> axum::Router<AppState> {
pub fn misc_routes() -> axum::Router<AppState> {
use axum::routing::get;
let router = axum::Router::new()
axum::Router::new()
.route("/health", get(server::health))
.route("/robots.txt", get(server::robots_txt))
.route("/favicon.ico", get(server::get_logo))
.route("/u/{handle}/did.json", get(identity::user_did_doc));
if tranquil_config::get().server.rfc_moo_compliance {
router.route("/cow.txt", get(server::cow_txt))
} else {
router
}
.route("/u/{handle}/did.json", get(identity::user_did_doc))
}
+8 -23
View File
@@ -6,16 +6,11 @@ use axum::{
};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::sync::LazyLock;
use tracing::{error, info, warn};
use tracing::{error, info};
use tranquil_pds::api::ApiError;
use tranquil_pds::api::proxy_client::{is_ssrf_safe, proxy_client};
use tranquil_pds::auth::{AnyUser, Auth};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{Did, DidRef, Nsid};
static CREATE_REPORT_NSID: LazyLock<Nsid> =
LazyLock::new(|| "com.atproto.moderation.createReport".parse().unwrap());
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ReportReasonType {
@@ -70,26 +65,17 @@ pub struct CreateReportOutput {
struct ReportServiceConfig {
url: String,
did: Did,
did: String,
}
fn get_report_service_config() -> Option<ReportServiceConfig> {
let cfg = tranquil_config::get();
let url = cfg.moderation.report_service_url.clone()?;
let did_str = cfg.moderation.report_service_did.as_deref()?;
if url.is_empty() || did_str.is_empty() {
let did = cfg.moderation.report_service_did.clone()?;
if url.is_empty() || did.is_empty() {
return None;
}
match did_str.parse::<Did>() {
Ok(did) => Some(ReportServiceConfig { url, did }),
Err(_) => {
warn!(
report_service_did = did_str,
"invalid report_service_did, handling reports locally"
);
None
}
}
Some(ReportServiceConfig { url, did })
}
pub async fn create_report(
@@ -110,7 +96,7 @@ async fn proxy_to_report_service(
state: &AppState,
auth_user: &tranquil_pds::auth::AuthenticatedUser,
service_url: &str,
service_did: &Did,
service_did: &str,
input: &CreateReportInput,
) -> Response {
if let Err(e) = is_ssrf_safe(service_url) {
@@ -151,9 +137,8 @@ async fn proxy_to_report_service(
let service_token = match tranquil_pds::auth::create_service_token(
&auth_user.did,
&DidRef::from(service_did),
Some(&CREATE_REPORT_NSID),
None,
service_did,
Some("com.atproto.moderation.createReport"),
&key_bytes,
) {
Ok(t) => t,
+63 -71
View File
@@ -6,7 +6,7 @@ use tranquil_db_traits::{CommsChannel, CommsStatus, CommsType};
use tranquil_pds::api::error::{ApiError, DbResultExt};
use tranquil_pds::auth::{Active, Auth};
use tranquil_pds::state::AppState;
use tranquil_types::{Did, Handle};
use tranquil_types::Did;
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
@@ -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,30 +136,23 @@ pub async fn request_channel_verification(
user_id: uuid::Uuid,
did: &Did,
channel: CommsChannel,
id: &str,
handle: Option<&Handle>,
identifier: &str,
handle: Option<&str>,
) -> 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 {
CommsChannel::Email => {
let hostname = &tranquil_config::get().server.hostname;
let handle = handle.ok_or_else(|| {
ApiError::InternalError(Some("Email verification requires a handle".into()))
})?;
let new_email = tranquil_types::EmailAddress::new(id)?;
let handle_str = handle.unwrap_or("user");
tranquil_pds::comms::comms_repo::enqueue_email_update(
state.repos.infra.as_ref(),
user_id,
&new_email,
handle,
identifier,
handle_str,
&formatted_token,
hostname,
)
@@ -192,10 +162,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 +187,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 +228,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 +267,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 +369,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 +433,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,
+20 -15
View File
@@ -11,18 +11,14 @@ use multihash::Multihash;
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::pin::Pin;
use std::sync::LazyLock;
use tracing::{debug, error, info, warn};
use tranquil_pds::api::error::{ApiError, DbResultExt};
use tranquil_pds::auth::{Auth, AuthAny, NotTakendown, Permissive, VerifyScope};
use tranquil_pds::delegation::DelegationActionType;
use tranquil_pds::state::AppState;
use tranquil_pds::types::{CidLink, Did, Nsid};
use tranquil_pds::types::{CidLink, Did};
use tranquil_pds::util::get_header_str;
static UPLOAD_BLOB_NSID: LazyLock<Nsid> =
LazyLock::new(|| "com.atproto.repo.uploadBlob".parse().unwrap());
fn detect_mime_type(data: &[u8], client_hint: &str) -> String {
if let Some(kind) = infer::get(data) {
let detected = kind.mime_type().to_string();
@@ -52,7 +48,7 @@ pub async fn upload_blob(
) -> Result<Response, ApiError> {
let (did, controller_did): (Did, Option<Did>) = match &auth {
AuthAny::Service(service) => {
service.require_lxm(&UPLOAD_BLOB_NSID)?;
service.require_lxm("com.atproto.repo.uploadBlob")?;
(service.did.clone(), None)
}
AuthAny::User(user) => {
@@ -148,13 +144,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 +156,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;
+66 -90
View File
@@ -14,35 +14,6 @@ use tranquil_pds::sync::verify::CarVerifier;
use tranquil_pds::types::Did;
use tranquil_types::{AtUri, CidLink};
fn map_car_verify_error(e: tranquil_pds::sync::verify::VerifyError) -> ApiError {
use tranquil_pds::sync::verify::VerifyError;
match e {
VerifyError::DidMismatch {
commit_did,
expected_did,
} => ApiError::InvalidRepo(format!(
"CAR file is for DID {} but you are authenticated as {}",
commit_did, expected_did
)),
VerifyError::InvalidSignature => ApiError::InvalidRequest(
"Repo commit signature does not match the DID document signing key".into(),
),
VerifyError::NoSigningKey => {
ApiError::InvalidRequest("DID document has no atproto signing key".into())
}
VerifyError::DidResolutionFailed(msg) => {
ApiError::InvalidRequest(format!("Could not resolve DID document: {}", msg))
}
VerifyError::MstValidationFailed(msg) => {
ApiError::InvalidRequest(format!("MST validation failed: {}", msg))
}
other => {
error!("CAR verification failed: {:?}", other);
ApiError::InvalidRequest(format!("CAR verification failed: {}", other))
}
}
}
pub async fn import_repo(
State(state): State<AppState>,
auth: Auth<NotTakendown>,
@@ -117,23 +88,41 @@ pub async fn import_repo(
let is_migration = user.inbound_migration && user.deactivated_at.is_some();
if skip_verification {
warn!("Skipping all CAR verification for repo import (SKIP_IMPORT_VERIFICATION=true)");
} else if is_migration {
let verified = CarVerifier::new()
.verify_car_structure_only(&root, &blocks)
.map_err(map_car_verify_error)?;
debug!(
"CAR structure verified for migration import: rev={}, data_cid={}",
verified.rev, verified.data_cid
);
} else {
let verified = CarVerifier::new()
.verify_car(did, &root, &blocks)
.await
.map_err(map_car_verify_error)?;
debug!(
"CAR signature and structure verified: rev={}, data_cid={}",
verified.rev, verified.data_cid
"Verifying CAR file structure for repo import (skipping signature and DID verification)"
);
let verifier = CarVerifier::new();
match verifier.verify_car_structure_only(&root, &blocks) {
Ok(verified) => {
debug!(
"CAR structure verification successful: rev={}, data_cid={}",
verified.rev, verified.data_cid
);
}
Err(tranquil_pds::sync::verify::VerifyError::DidMismatch {
commit_did,
expected_did,
}) => {
return Err(ApiError::InvalidRepo(format!(
"CAR file is for DID {} but you are authenticated as {}",
commit_did, expected_did
)));
}
Err(tranquil_pds::sync::verify::VerifyError::MstValidationFailed(msg)) => {
return Err(ApiError::InvalidRequest(format!(
"MST validation failed: {}",
msg
)));
}
Err(e) => {
error!("CAR structure verification error: {:?}", e);
return Err(ApiError::InvalidRequest(format!(
"CAR verification failed: {}",
e
)));
}
}
}
let max_blocks = tranquil_config::get().import.max_blocks as usize;
let _write_lock = state.repo_write_locks.lock(user_id).await;
@@ -167,11 +156,17 @@ pub async fn import_repo(
.records
.iter()
.flat_map(|record| {
let record_uri = AtUri::from_parts(did, &record.collection, &record.rkey);
record
.blob_refs
.iter()
.map(move |blob_ref| (record_uri.clone(), blob_ref.cid.clone()))
let record_uri =
AtUri::from_parts(did.as_str(), &record.collection, &record.rkey);
record.blob_refs.iter().filter_map(move |blob_ref| {
match CidLink::new(&blob_ref.cid) {
Ok(cid_link) => Some((record_uri.clone(), cid_link)),
Err(_) => {
tracing::warn!(cid = %blob_ref.cid, "skipping unparseable blob CID reference during import");
None
}
}
})
})
.collect();
@@ -221,62 +216,43 @@ pub async fn import_repo(
})?;
let new_rev = Tid::now(LimitedU32::MIN);
let new_rev_str = new_rev.to_string();
let (commit_bytes, _sig) =
create_signed_commit(did, import_result.data_cid, &new_rev, None, &signing_key)
.map_err(|e| {
error!("Failed to create new commit: {}", e);
ApiError::InternalError(None)
})?;
let (commit_bytes, _sig) = create_signed_commit(
did,
import_result.data_cid,
&new_rev_str,
None,
&signing_key,
)
.map_err(|e| {
error!("Failed to create new commit: {}", e);
ApiError::InternalError(None)
})?;
let new_root_cid: cid::Cid =
state.block_store.put(&commit_bytes).await.map_err(|e| {
error!("Failed to store new commit block: {:?}", e);
ApiError::InternalError(None)
})?;
let new_root_cid_link = CidLink::from(&new_root_cid);
let new_rev_tid = tranquil_pds::types::Tid::from(new_rev.clone());
state
.repos
.repo
.update_repo_root(user_id, &new_root_cid_link, &new_rev_tid)
.update_repo_root(user_id, &new_root_cid_link, &new_rev_str)
.await
.map_err(|e| {
error!("Failed to update repo root: {:?}", e);
ApiError::InternalError(None)
})?;
match tranquil_pds::scheduled::collect_current_repo_blocks(
&state.block_store,
&new_root_cid,
)
.await
{
Ok(reachable) => {
if !reachable.is_complete() {
error!(
unreadable = reachable.unreadable,
"scheduling a structural repair because the imported repo walk could \
not read every block"
);
tranquil_pds::repo_ops::schedule_repo_repair(&state, user_id);
}
state
.repos
.repo
.insert_user_blocks(user_id, &reachable.block_cids, &new_rev_tid)
.await
.map_err(|e| {
error!("Failed to insert user_blocks: {:?}", e);
ApiError::InternalError(None)
})?;
}
Err(e) => {
error!(
"Failed to walk the imported repo: {:?}. The root is already updated and \
a scheduled structural repair will rebuild user_blocks",
e
);
tranquil_pds::repo_ops::schedule_repo_repair(&state, user_id);
}
}
let mut all_block_cids: Vec<Vec<u8>> = blocks.keys().map(|c| c.to_bytes()).collect();
all_block_cids.push(new_root_cid.to_bytes());
state
.repos
.repo
.insert_user_blocks(user_id, &all_block_cids, &new_rev_str)
.await
.map_err(|e| {
error!("Failed to insert user_blocks: {:?}", e);
ApiError::InternalError(None)
})?;
let new_root_str = new_root_cid.to_string();
info!(
"Created new commit for imported repo: cid={}, rev={}",
+5 -5
View File
@@ -18,7 +18,7 @@ use tranquil_pds::repo_ops::{
finalize_repo_write, with_repair_retry,
};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{AtIdentifier, AtUri, CidLink, Did, Nsid, Rkey};
use tranquil_pds::types::{AtIdentifier, AtUri, Did, Nsid, Rkey};
use tranquil_pds::validation::ValidationStatus;
const MAX_BATCH_WRITES: usize = 200;
@@ -27,7 +27,7 @@ struct WriteAccumulator {
mst: Mst<TrackingBlockStore>,
results: Vec<WriteResult>,
ops: Vec<RecordOp>,
all_blob_cids: Vec<CidLink>,
all_blob_cids: Vec<String>,
backlinks_to_add: Vec<Backlink>,
backlinks_to_remove: Vec<AtUri>,
}
@@ -353,9 +353,9 @@ pub async fn apply_writes(
&auth,
&input.writes,
|w| match w {
WriteOp::Create { collection, .. } => collection,
WriteOp::Update { collection, .. } => collection,
WriteOp::Delete { collection, .. } => collection,
WriteOp::Create { collection, .. } => collection.as_str(),
WriteOp::Update { collection, .. } => collection.as_str(),
WriteOp::Delete { collection, .. } => collection.as_str(),
},
|w| match w {
WriteOp::Create { .. } => WriteOpKind::Create,
@@ -0,0 +1 @@
pub use tranquil_pds::repo_ops::*;
@@ -9,13 +9,13 @@ pub async fn validate_record_with_status(
require_lexicon: bool,
) -> Result<ValidationStatus, ApiError> {
let registry = tranquil_lexicon::LexiconRegistry::global();
if !registry.has_schema(collection) {
let _ = registry.resolve_dynamic(collection).await;
if !registry.has_schema(collection.as_str()) {
let _ = registry.resolve_dynamic(collection.as_str()).await;
}
let validator = RecordValidator::new().require_lexicon(require_lexicon);
validator
.validate_with_rkey(record, collection, rkey)
.validate_with_rkey(record, collection.as_str(), rkey.map(|v| v.as_str()))
.map_err(validation_error_to_api_error)
}
+9 -19
View File
@@ -7,7 +7,6 @@ use serde::{Deserialize, Serialize};
use serde_json::json;
use std::borrow::Cow;
use std::str::FromStr;
use tracing::warn;
use tranquil_pds::api::error::{ApiError, DbResultExt};
use tranquil_pds::auth::{
Active, Auth, AuthSource, RepoScopeAction, ScopeVerified, VerifyScope, require_not_migrated,
@@ -18,7 +17,7 @@ use tranquil_pds::repo_ops::{
finalize_repo_write, with_repair_retry,
};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{AtIdentifier, AtUri, Did, Nsid, Rkey, Tid};
use tranquil_pds::types::{AtIdentifier, AtUri, Did, Nsid, Rkey};
use tranquil_pds::validation::ValidationStatus;
use uuid::Uuid;
@@ -103,7 +102,7 @@ pub struct CreateRecordInput {
#[serde(rename_all = "camelCase")]
pub struct CommitInfo {
pub cid: String,
pub rev: Tid,
pub rev: String,
}
#[derive(Serialize)]
@@ -182,15 +181,8 @@ async fn create_record_inner(
else {
continue;
};
let (Ok(conflict_rkey), Ok(conflict_collection)) =
(Rkey::new(conflict_rkey_str), Nsid::new(conflict_col_str))
else {
warn!(
uri = %conflict_uri,
"skipping a backlink conflict whose stored URI doesn't parse"
);
continue;
};
let conflict_rkey = Rkey::from(conflict_rkey_str.to_string());
let conflict_collection = Nsid::from(conflict_col_str.to_string());
let conflict_key = format!("{}/{}", conflict_collection, conflict_rkey);
let prev_cid = match mst.get(&conflict_key).await {
@@ -381,8 +373,11 @@ async fn put_record_inner(
let record_ipld = tranquil_pds::util::json_to_ipld(&input.record);
let record_bytes = serde_ipld_dagcbor::to_vec(&record_ipld)
.map_err(|_| ApiError::InvalidRecord("Failed to serialize record".into()))?;
let record_cid = jacquard_repo::mst::util::compute_cid(&record_bytes)
.map_err(|_| ApiError::InvalidRecord("Failed to compute record CID".into()))?;
let record_cid = ctx
.tracking_store
.put(&record_bytes)
.await
.map_err(|_| ApiError::InternalError(Some("Failed to save record block".into())))?;
if existing_cid == Some(record_cid) {
return Ok(PutRecordOutput {
@@ -393,11 +388,6 @@ async fn put_record_inner(
});
}
ctx.tracking_store
.put(&record_bytes)
.await
.map_err(|_| ApiError::InternalError(Some("Failed to save record block".into())))?;
let record_uri = AtUri::from_parts(did, &input.collection, &input.rkey);
let (new_mst, op, is_update, backlinks_to_remove) = match existing_cid {
Some(prev_cid) => {
@@ -10,7 +10,7 @@ use serde_json::Value;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use tracing::{debug, error, info, warn};
use tracing::{error, info, warn};
use tranquil_pds::api::EmptyResponse;
use tranquil_pds::api::error::{ApiError, DbResultExt};
use tranquil_pds::auth::{Auth, NotTakendown, Permissive, require_legacy_session_mfa};
@@ -18,7 +18,7 @@ use tranquil_pds::cache::Cache;
use tranquil_pds::oauth::scopes::{AccountAction, AccountAttr};
use tranquil_pds::plc::PlcClient;
use tranquil_pds::state::AppState;
use tranquil_pds::types::{PlainPassword, Tid};
use tranquil_pds::types::PlainPassword;
use uuid::Uuid;
#[derive(Serialize)]
@@ -27,8 +27,7 @@ pub struct CheckAccountStatusOutput {
pub activated: bool,
pub valid_did: bool,
pub repo_commit: String,
#[serde(serialize_with = "serialize_optional_rev")]
pub repo_rev: Option<Tid>,
pub repo_rev: String,
pub repo_blocks: i64,
pub indexed_records: i64,
pub private_state_values: i64,
@@ -36,13 +35,6 @@ pub struct CheckAccountStatusOutput {
pub imported_blobs: i64,
}
fn serialize_optional_rev<S: serde::Serializer>(
rev: &Option<Tid>,
serializer: S,
) -> Result<S::Ok, S::Error> {
serializer.serialize_str(rev.as_ref().map_or("", Tid::as_str))
}
pub async fn check_account_status(
State(state): State<AppState>,
auth: Auth<Permissive>,
@@ -73,20 +65,23 @@ pub async fn check_account_status(
.count_user_blocks(user_id)
.await
.unwrap_or(0);
let repo_rev = match (repo_rev_from_db, Cid::from_str(&repo_commit)) {
(Some(rev), _) => Some(rev),
(None, Ok(cid)) => state
.block_store
.get(&cid)
.await
.ok()
.flatten()
.and_then(|block| {
let repo_rev = if let Some(rev) = repo_rev_from_db {
rev
} else if !repo_commit.is_empty() {
if let Ok(cid) = Cid::from_str(&repo_commit) {
if let Ok(Some(block)) = state.block_store.get(&cid).await {
Commit::from_cbor(&block)
.ok()
.map(|commit| Tid::from(commit.rev().clone()))
}),
(None, Err(_)) => None,
.map(|c| c.rev().to_string())
.unwrap_or_default()
} else {
String::new()
}
} else {
String::new()
}
} else {
String::new()
};
let record_count: i64 = state.repos.repo.count_records(user_id).await.unwrap_or(0);
let imported_blobs: i64 = state
@@ -138,7 +133,7 @@ async fn assert_valid_did_document_for_service(
if did.as_str().starts_with("did:plc:") {
let max_attempts = if with_retry { 5 } else { 1 };
let cache_for_retry = cache.clone();
let did_owned = did.clone();
let did_owned = did.as_str().to_string();
let expected_owned = expected_endpoint.clone();
let attempt_counter = Arc::new(AtomicUsize::new(0));
@@ -202,20 +197,18 @@ async fn assert_valid_did_document_for_service(
.await
.map_err(ApiError::InvalidRequest)?;
let doc_rotation_keys = doc_data
.get("rotationKeys")
.and_then(Value::as_array)
.map(|arr| arr.iter().filter_map(Value::as_str).collect::<Vec<_>>())
.unwrap_or_default();
let server_rotation_key = tranquil_config::get().secrets.plc_rotation_key.clone();
if let Some(ref expected_rotation_key) = server_rotation_key
&& !doc_rotation_keys.contains(&expected_rotation_key.as_str())
{
debug!(
"DID {} rotation keys {:?} omit the PDS-managed server rotation key {}",
did, doc_rotation_keys, expected_rotation_key
);
if let Some(ref expected_rotation_key) = server_rotation_key {
let rotation_keys = doc_data
.get("rotationKeys")
.and_then(Value::as_array)
.map(|arr| arr.iter().filter_map(Value::as_str).collect::<Vec<_>>())
.unwrap_or_default();
if !rotation_keys.contains(&expected_rotation_key.as_str()) {
return Err(ApiError::InvalidRequest(
"Server rotation key not included in PLC DID data".into(),
));
}
}
let doc_signing_key = doc_data
@@ -250,13 +243,6 @@ async fn assert_valid_did_document_for_service(
"DID document verification method does not match expected signing key".into(),
));
}
if !doc_rotation_keys.contains(&expected_did_key.as_str()) {
debug!(
"DID {} rotation keys {:?} omit the PDS-managed signing key {}",
did, doc_rotation_keys, expected_did_key
);
}
}
} else if let Some(host_and_path) = did.as_str().strip_prefix("did:web:") {
let client = tranquil_pds::api::proxy_client::did_resolution_client();
@@ -390,7 +376,7 @@ pub async fn activate_account(
.cache
.delete(&tranquil_pds::cache_keys::plc_data_key(&did))
.await;
if state.did_resolver.refresh_did(&did).await.is_err() {
if state.did_resolver.refresh_did(did.as_str()).await.is_err() {
warn!(
"[MIGRATION] activateAccount: Failed to refresh DID cache for {}",
did
@@ -426,8 +412,10 @@ pub async fn activate_account(
"[MIGRATION] activateAccount: Sequencing identity event for did={} handle={:?}",
did, handle
);
let handle_typed = handle.clone();
if let Err(e) =
tranquil_pds::repo_ops::sequence_identity_event(&state, &did, handle.as_ref()).await
tranquil_pds::repo_ops::sequence_identity_event(&state, &did, handle_typed.as_ref())
.await
{
warn!(
"[MIGRATION] activateAccount: Failed to sequence identity event for activation: {}",
@@ -450,9 +438,7 @@ pub async fn activate_account(
);
let rev = if let Ok(cid) = Cid::from_str(root_cid_link.as_str()) {
if let Ok(Some(block)) = state.block_store.get(&cid).await {
Commit::from_cbor(&block)
.ok()
.map(|c| Tid::from(c.rev().clone()))
Commit::from_cbor(&block).ok().map(|c| c.rev().to_string())
} else {
None
}
@@ -462,8 +448,8 @@ pub async fn activate_account(
if let Err(e) = tranquil_pds::repo_ops::sequence_sync_event(
&state,
&did,
&root_cid_link,
rev.as_ref(),
root_cid_link.as_str(),
rev.as_deref(),
)
.await
{
@@ -586,27 +572,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 {}))
@@ -651,7 +626,7 @@ pub async fn delete_account(
state.repos.session.as_ref(),
user_id,
password,
password_hash.as_ref(),
password_hash.as_deref(),
)
.await
.is_none()
@@ -116,10 +116,7 @@ pub async fn create_app_password(
.await
.ok()
.flatten();
let granted_scopes = match grant {
Some(g) => g.granted_scopes,
None => return Err(ApiError::InsufficientScope(None)),
};
let granted_scopes = grant.map(|g| g.granted_scopes).unwrap_or_default();
let requested = input.scopes.as_deref().unwrap_or("atproto");
let intersected = intersect_scopes(requested, granted_scopes.as_str());
@@ -191,7 +188,7 @@ pub async fn create_app_password(
}
Ok(Json(CreateAppPasswordOutput {
name: name.to_string(),
password: password.into_inner(),
password,
created_at: created_at.to_rfc3339(),
privileged: privilege.is_privileged(),
scopes: final_scopes,
-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
}
}
}
-57
View File
@@ -1,57 +0,0 @@
..........................
....*o|||||||8#@@@@@@@@@@@@@@@@@@@@@@@###&|o:_..
..*:o|||&8##@###8888888######@#@###########################|*...
.:o|||8#####8888|:::**. *&########################@@################&o_
.*o&8###@#8&o*_. :###@##############@########################@@##&o_
.*o8########& :##@#@##############@############################@###|_
.*o|8##########8o .#######################################################&o_
*&##|_ ..*&##8&o*|88888|_ _#######################################@##################|.
*#####& *&######&o_..*o|o:_ .&##o _###########################################################&_
_##8*##8 .|88|:::|#######8###8|*:_ .&#@@8 _##@@@########################################################&_
_#@8_##8_ *8#8|*_ _:|#####&&####8 .&##############################################################|
_#@8.|##8_ _::o###8&##8 .|##@############################8###########################@@#|_
*###o.|88o ..*&####|..##& _|##########################8|_ .|#############################8
*|###|_ ._&####8|*_ _*_ _::&8888888888888888|::*_ .|##@####@@@##################|
*&###|_ _:_ .&88###8|*_ ..... .|#####@@@##################8
.##@#& _##& .|##o _#@@#@#| .|#######&:_ _|###@####################8
.:8##8o _o:*&##| *##8_.&@@##@#| _::o8#8|::|#####|_ _|#################88###8
.&##&*_ *###o_###| .|##8*&##|*###o _###8####8|_ _:|###|_ .*o|||o:_ _:::&8888888888|_ _##8
.###|_. _###o *###|*&#######8 *##8 .##8_ _:|###|_ _|###|_ .&########o _##&
_|####8|&##8:_ _|#########88o .##8 *##& _|###o .|###o .#########| .o##o
o#8|*:#@@###o _:::*__*_ _##8_ _##8_ _&##|_ *##8_ *8#####8|_ .*oo:_ o##|
*###o.&#####& _oo* .8##& .8##8_ .|##& o##& _::::_.*o|8######|_ .##8.
_###&o&##8_:*_ .###& .###|_&#####| _##8 :###o *ooo&#########@#@#& ....:##&
.|8||###&. _**_ .###88##|*&###|*._&##& *|##8_ *o&####@@####@@######& .*o||||||&#######8_
*&###o _|88##8_ _:8######|*:###|_ _##################88|_ *&#################&
*#####o *&8o *##& _:::*_.&##|_ _#@##############8_ :##################8*
.###&##8_.|88o *&8o _@@& .###| .&####@#########|_ .####@###@@########8*
_##&.|###|_.... .|88o _##8* *&###|_ *###o _:&88######8|_ .*o|||##################o
_##8_ _|########|_ .*o8####&#@@#@##o *#@8 _*:*. .&###@###################|
.|##& _:::::&##& .*&##############@#8_ .###o _#@####################|_
.&##|_ .&##8_ *o&####################8_ *##& .&#####@@############8*
.|##8_.&###&####8_ _########################8**##& _##################|_
.&###&##888888##8_ .|88######@###########|*######o _|8###############|_
.&#####o *###|_ _::::::*o##8**o##8 .|###8o .&##@#############&*
.|####o _|###|_ _##8.*&##& _*_ .#################|
_*_ _|###|_.. .|#####8|_ *&#@#########8###&
*#######&|o:_... ..*:::*. ......._:o&8####888&o:#####8_
_###|&888#####@#####&|||o:_........................._:o||||8##@@@@####8|:*_ _:::*_
.|#@###o _:::o#@#888######@@@@@@@@@@@@@@@@@@@@@@@@#####888|::::::**_
_::*_ :##& *&8|_:::::::::::::::::::::::::**_
.###8||&##8o
_|888888|_
+48 -60
View File
@@ -19,7 +19,6 @@ use tranquil_pds::auth::{Auth, NotTakendown};
use tranquil_pds::oauth::scopes::{AccountAction, AccountAttr};
use tranquil_pds::rate_limit::{EmailUpdateLimit, RateLimited, VerificationCheckLimit};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{AtIdentifier, Did};
const EMAIL_UPDATE_TTL: Duration = Duration::from_secs(30 * 60);
@@ -38,7 +37,7 @@ struct PendingEmailUpdate {
async fn get_pending_email_update(
cache: &dyn tranquil_pds::cache::Cache,
did: &Did,
did: &str,
) -> Option<PendingEmailUpdate> {
cache
.get(&tranquil_pds::cache_keys::email_update_key(did))
@@ -71,7 +70,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(),
));
};
@@ -80,7 +79,7 @@ pub async fn request_email_update(
if token_required {
let token = tranquil_pds::auth::email_token::create_email_token(
state.cache.as_ref(),
&auth.did,
auth.did.as_str(),
tranquil_pds::auth::email_token::EmailTokenPurpose::UpdateEmail,
)
.await
@@ -89,43 +88,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 +149,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 +169,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 +225,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 {
@@ -252,7 +249,7 @@ pub async fn update_email(
tranquil_pds::auth::email_token::validate_email_token(
state.cache.as_ref(),
did,
did.as_str(),
tranquil_pds::auth::email_token::EmailTokenPurpose::UpdateEmail,
token,
)
@@ -285,7 +282,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;
@@ -301,7 +298,7 @@ pub async fn update_email(
let short_token_result = tranquil_pds::auth::email_token::validate_email_token(
state.cache.as_ref(),
did,
did.as_str(),
tranquil_pds::auth::email_token::EmailTokenPurpose::UpdateEmail,
token,
)
@@ -352,26 +349,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,
)
@@ -399,7 +394,7 @@ pub async fn update_email(
#[derive(Deserialize)]
pub struct CheckEmailVerifiedInput {
pub identifier: AtIdentifier,
pub identifier: String,
}
pub async fn check_email_verified(
@@ -569,22 +564,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);
+26 -26
View File
@@ -7,7 +7,6 @@ use tranquil_pds::auth::{Admin, Auth, NotTakendown};
use tranquil_pds::state::AppState;
use tranquil_pds::types::Did;
use tranquil_pds::util::gen_invite_code;
use tranquil_types::InviteCode as InviteCodeValue;
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
@@ -18,7 +17,7 @@ pub struct CreateInviteCodeInput {
#[derive(Serialize)]
pub struct CreateInviteCodeOutput {
pub code: InviteCodeValue,
pub code: String,
}
pub async fn create_invite_code(
@@ -43,7 +42,7 @@ pub async fn create_invite_code(
match state
.repos
.infra
.create_invite_code(&code, input.use_count, &for_account)
.create_invite_code(&code, input.use_count, Some(&for_account))
.await
{
Ok(true) => Ok(Json(CreateInviteCodeOutput { code })),
@@ -73,8 +72,8 @@ pub struct CreateInviteCodesOutput {
#[derive(Serialize)]
pub struct AccountCodes {
pub account: Did,
pub codes: Vec<InviteCodeValue>,
pub account: String,
pub codes: Vec<String>,
}
pub async fn create_invite_codes(
@@ -113,11 +112,14 @@ pub async fn create_invite_codes(
let infra_repo = state.repos.infra.clone();
let use_count = input.use_count;
async move {
let codes: Vec<InviteCodeValue> = (0..code_count).map(|_| gen_invite_code()).collect();
let codes: Vec<String> = (0..code_count).map(|_| gen_invite_code()).collect();
infra_repo
.create_invite_codes_batch(&codes, use_count, admin_user_id, &account)
.create_invite_codes_batch(&codes, use_count, admin_user_id, Some(&account))
.await
.map(|_| AccountCodes { account, codes })
.map(|_| AccountCodes {
account: account.to_string(),
codes,
})
}
}))
.await;
@@ -143,7 +145,7 @@ pub struct GetAccountInviteCodesParams {
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct InviteCode {
pub code: InviteCodeValue,
pub code: String,
pub available: i32,
pub disabled: bool,
pub for_account: String,
@@ -188,24 +190,27 @@ pub async fn get_account_invite_codes(
let codes = futures::future::join_all(filtered_codes.into_iter().map(|info| {
let infra_repo = state.repos.infra.clone();
async move {
let uses: Vec<InviteCodeUse> = infra_repo
let uses = infra_repo
.get_invite_code_uses(&info.code)
.await
.log_db_err("fetching invite code uses")?
.into_iter()
.map(|u| InviteCodeUse {
used_by: u.used_by_did.to_string(),
used_by_handle: u.used_by_handle.map(|h| h.to_string()),
used_at: u.used_at.to_rfc3339(),
.map(|use_rows| {
use_rows
.into_iter()
.map(|u| InviteCodeUse {
used_by: u.used_by_did.to_string(),
used_by_handle: u.used_by_handle.map(|h| h.to_string()),
used_at: u.used_at.to_rfc3339(),
})
.collect::<Vec<_>>()
})
.collect();
.unwrap_or_default();
let use_count = i32::try_from(uses.len()).unwrap_or(i32::MAX);
if !include_used && use_count >= info.available_uses {
return Ok(None);
return None;
}
Ok(Some(InviteCode {
Some(InviteCode {
code: info.code,
available: info.available_uses,
disabled: false,
@@ -216,16 +221,11 @@ pub async fn get_account_invite_codes(
.unwrap_or_else(|| "admin".to_string()),
created_at: info.created_at.to_rfc3339(),
uses,
}))
})
}
}))
.await;
let codes: Vec<InviteCode> = codes
.into_iter()
.collect::<Result<Vec<Option<InviteCode>>, ApiError>>()?
.into_iter()
.flatten()
.collect();
let codes: Vec<InviteCode> = codes.into_iter().flatten().collect();
Ok(Json(GetAccountInviteCodesOutput { codes }))
}
+1 -10
View File
@@ -29,10 +29,6 @@ pub async fn robots_txt() -> impl IntoResponse {
"# Hello!\n\n# Crawling the public API is allowed\nUser-agent: *\nAllow: /\n",
)
}
pub async fn cow_txt() -> &'static str {
include_str!("cow.txt")
}
pub fn is_self_hosted_did_web_enabled() -> bool {
tranquil_config::get().server.enable_pds_hosted_did_web
}
@@ -77,12 +73,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 {
+1 -3
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,14 +22,13 @@ 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,
};
pub use invite::{create_invite_code, create_invite_codes, get_account_invite_codes};
pub use logo::get_logo;
pub use meta::{cow_txt, describe_server, health, robots_txt};
pub use meta::{describe_server, health, robots_txt};
pub use migration::{get_did_document, update_did_document};
pub use passkey_account::{
complete_passkey_setup, create_passkey_account, recover_passkey_account,
@@ -7,14 +7,13 @@ use serde_json::json;
use tracing::{debug, error, info, warn};
use tranquil_db_traits::WebauthnChallengeType;
use tranquil_pds::api::error::ApiError;
use tranquil_pds::api::invite::check_registration_invite;
use tranquil_pds::api::{OptionsResponse, SuccessResponse};
use tranquil_pds::auth::NormalizedLoginIdentifier;
use tranquil_pds::auth::{ServiceTokenVerifier, generate_app_password, is_service_token};
use tranquil_pds::rate_limit::{AccountCreationLimit, PasswordResetLimit, RateLimited};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{Did, Handle, Jti, Nsid, PlainPassword};
use tranquil_pds::types::{Did, Handle, PlainPassword};
use tranquil_pds::validation::validate_password;
fn generate_setup_token() -> String {
@@ -39,7 +38,7 @@ pub struct CreatePasskeyAccountInput {
pub invite_code: Option<String>,
pub did: Option<String>,
pub did_type: Option<String>,
pub signing_key: Option<Did>,
pub signing_key: Option<String>,
pub verification_channel: Option<tranquil_db_traits::CommsChannel>,
pub discord_username: Option<String>,
pub telegram_username: Option<String>,
@@ -69,10 +68,8 @@ pub async fn create_passkey_account(
let token = extracted.token;
if is_service_token(&token) {
let verifier = ServiceTokenVerifier::new();
let create_account_lxm = Nsid::new("com.atproto.server.createAccount")
.expect("com.atproto.server.createAccount is a valid NSID");
match verifier
.verify_service_token(&token, Some(&create_account_lxm))
.verify_service_token(&token, Some("com.atproto.server.createAccount"))
.await
{
Ok(claims) => {
@@ -116,14 +113,32 @@ 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);
}
let invite_registration =
check_registration_invite(&state, input.invite_code.as_deref()).await?;
let is_bootstrap = state.bootstrap_invite_code.is_some()
&& state.repos.user.count_users().await.unwrap_or(1) == 0;
let _validated_invite_code = if is_bootstrap {
match input.invite_code.as_deref() {
Some(code) if Some(code) == state.bootstrap_invite_code.as_deref() => None,
_ => return Err(ApiError::InvalidInviteCode),
}
} else if let Some(ref code) = input.invite_code {
match state.repos.infra.validate_invite_code(code).await {
Ok(validated) => Some(validated),
Err(_) => return Err(ApiError::InvalidInviteCode),
}
} else {
let invite_required = tranquil_config::get().server.invite_code_required;
if invite_required {
return Err(ApiError::InviteCodeRequired);
}
None
};
let verification_channel = input
.verification_channel
@@ -145,7 +160,7 @@ pub async fn create_passkey_account(
let did_type = input.did_type.as_deref().unwrap_or("plc");
let key_result =
match crate::identity::provision::resolve_signing_key(&state, input.signing_key.as_ref())
match crate::identity::provision::resolve_signing_key(&state, input.signing_key.as_deref())
.await
{
Ok(k) => k,
@@ -155,7 +170,7 @@ pub async fn create_passkey_account(
let secret_key = key_result.signing_key;
let reserved_key_id = key_result.reserved_key_id;
let did: Did = match did_type {
let did = match did_type {
"web" => {
let self_hosted_did = match common::create_self_hosted_did_web(&handle) {
Ok(d) => d,
@@ -193,7 +208,7 @@ pub async fn create_passkey_account(
d,
hostname,
&input.handle,
input.signing_key.as_ref(),
input.signing_key.as_deref(),
)
.await
{
@@ -201,8 +216,7 @@ pub async fn create_passkey_account(
}
info!(did = %d, "Creating external did:web passkey account (reserved key)");
}
d.parse()
.map_err(|_| ApiError::InvalidDid("Invalid DID format".into()))?
d.to_string()
}
_ => {
if let Some(ref auth_did) = byod_auth {
@@ -215,9 +229,7 @@ pub async fn create_passkey_account(
)));
}
info!(did = %provided_did, "Creating BYOD did:plc passkey account (migration)");
provided_did
.parse()
.map_err(|_| ApiError::InvalidDid("Invalid DID format".into()))?
provided_did.clone()
} else {
return Err(ApiError::InvalidRequest(
"BYOD migration requires a did:plc or did:web DID".into(),
@@ -229,9 +241,15 @@ pub async fn create_passkey_account(
));
}
} else {
let rotation_key = tranquil_config::get()
.secrets
.plc_rotation_key
.clone()
.unwrap_or_else(|| tranquil_pds::plc::signing_key_to_did_key(&secret_key));
let genesis_result = match tranquil_pds::plc::create_genesis_operation(
&secret_key,
tranquil_config::get().secrets.plc_rotation_key.as_deref(),
&rotation_key,
&handle,
&pds_endpoint,
) {
@@ -273,9 +291,13 @@ pub async fn create_passkey_account(
None
};
let did_typed: Did = match did.parse() {
Ok(d) => d,
Err(_) => return Err(ApiError::InternalError(Some("Invalid DID".into()))),
};
let repo = match crate::identity::provision::init_genesis_repo(
&state,
&did,
&did_typed,
&secret_key,
&secret_key_bytes,
)
@@ -294,6 +316,10 @@ pub async fn create_passkey_account(
None
};
let handle_typed: Handle = match handle.parse() {
Ok(h) => h,
Err(_) => return Err(ApiError::InvalidHandle(None)),
};
let repo_for_seq = repo.clone();
let comms = crate::identity::provision::normalize_comms_usernames(
input.discord_username.as_deref(),
@@ -301,9 +327,9 @@ pub async fn create_passkey_account(
input.signal_username.as_deref(),
);
let create_input = tranquil_db_traits::CreatePasskeyAccountInput {
handle: handle.clone(),
handle: handle_typed.clone(),
email: email.clone().unwrap_or_default(),
did: did.clone(),
did: did_typed.clone(),
preferred_comms_channel: verification_channel,
discord_username: comms.discord,
telegram_username: comms.telegram,
@@ -314,10 +340,14 @@ pub async fn create_passkey_account(
encrypted_key_bytes: repo.encrypted_key_bytes,
encryption_version: tranquil_pds::config::ENCRYPTION_VERSION,
reserved_key_id,
commit_cid: tranquil_types::CidLink::from(&repo.commit_cid),
commit_cid: repo.commit_cid.to_string(),
repo_rev: repo.repo_rev.clone(),
genesis_block_cids: repo.genesis_block_cids,
invite_code: invite_registration.into_invite_code(),
invite_code: if is_bootstrap {
None
} else {
input.invite_code.clone()
},
birthdate_pref,
};
@@ -329,9 +359,6 @@ pub async fn create_passkey_account(
Err(tranquil_db_traits::CreateAccountError::EmailTaken) => {
return Err(ApiError::EmailTaken);
}
Err(tranquil_db_traits::CreateAccountError::InviteCodeUnavailable) => {
return Err(ApiError::InvalidInviteCode);
}
Err(e) => {
error!("Error creating passkey account: {:?}", e);
return Err(ApiError::InternalError(None));
@@ -342,10 +369,10 @@ pub async fn create_passkey_account(
if !is_byod_did_web {
crate::identity::provision::sequence_new_account(
&state,
&did,
&handle,
&did_typed,
&handle_typed,
&repo_for_seq,
handle.as_str(),
&handle,
)
.await;
}
@@ -353,7 +380,7 @@ pub async fn create_passkey_account(
crate::identity::provision::enqueue_signup_verification(
&state,
user_id,
&did,
&did_typed,
verification_channel,
&verification_recipient,
)
@@ -364,10 +391,10 @@ pub async fn create_passkey_account(
let access_jwt = if byod_auth.is_some() {
match tranquil_pds::auth::create_access_token_with_metadata(&did, &secret_key_bytes) {
Ok(token_meta) => {
let refresh_jti = Jti::from(uuid::Uuid::new_v4().to_string());
let refresh_jti = uuid::Uuid::new_v4().to_string();
let refresh_expires = chrono::Utc::now() + chrono::Duration::hours(24);
let session_data = tranquil_db_traits::SessionTokenCreate {
did: did.clone(),
did: did_typed.clone(),
access_jti: token_meta.jti.clone(),
refresh_jti,
access_expires_at: token_meta.expires_at,
@@ -394,8 +421,8 @@ pub async fn create_passkey_account(
};
Ok(Json(CreatePasskeyAccountOutput {
did,
handle,
did: did.into(),
handle: handle.into(),
setup_token,
setup_expires_at,
access_jwt,
@@ -549,7 +576,7 @@ pub async fn complete_passkey_setup(
Ok(Json(CompletePasskeySetupOutput {
did: input.did.clone(),
handle: user.handle,
app_password: app_password.into_inner(),
app_password,
app_password_name,
}))
}
@@ -688,7 +715,7 @@ pub async fn request_passkey_recovery(
if let Err(e) = state
.repos
.user
.set_recovery_token(&user.did, recovery_token_hash.as_str(), expires_at)
.set_recovery_token(&user.did, &recovery_token_hash, expires_at)
.await
{
error!("Error updating recovery token: {:?}", e);
@@ -703,11 +730,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 +803,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 }))
}
+24 -30
View File
@@ -10,7 +10,7 @@ use tranquil_pds::auth::{
};
use tranquil_pds::rate_limit::{PasswordResetLimit, RateLimited, ResetPasswordLimit};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{Handle, PlainPassword};
use tranquil_pds::types::PlainPassword;
use tranquil_pds::validation::validate_password;
#[derive(Deserialize)]
@@ -45,23 +45,12 @@ pub async fn request_password_reset(
None
};
let lookup = match Handle::new(normalized_handle.as_str()) {
Ok(handle) => {
state
.repos
.user
.get_id_by_email_or_handle(normalized, &handle)
.await
}
Err(_) => state
.repos
.user
.get_by_email(normalized)
.await
.map(|user| user.map(|user| user.id)),
};
let user_id = match lookup {
let user_id = match state
.repos
.user
.get_id_by_email_or_handle(normalized, normalized_handle.as_str())
.await
{
Ok(Some(id)) => id,
Ok(None) => {
info!("Password reset requested for unknown identifier");
@@ -90,13 +79,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 +182,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 {}))
}
+2 -2
View File
@@ -74,7 +74,7 @@ pub async fn reauth_password(
.log_db_err("fetching password hash")?
.ok_or(ApiError::AccountNotFound)?;
let password_valid = bcrypt::verify(&input.password, password_hash.as_str()).unwrap_or(false);
let password_valid = bcrypt::verify(&input.password, &password_hash).unwrap_or(false);
if !password_valid {
let app_password_hashes = state
@@ -85,7 +85,7 @@ pub async fn reauth_password(
.unwrap_or_default();
let app_password_valid = app_password_hashes.iter().fold(false, |acc, h| {
acc | bcrypt::verify(&input.password, h.as_str()).unwrap_or(false)
acc | bcrypt::verify(&input.password, h).unwrap_or(false)
});
if !app_password_valid {
@@ -11,7 +11,7 @@ use tracing::{error, info, warn};
use tranquil_pds::api::error::ApiError;
use tranquil_pds::auth::extractor::{Auth, Permissive};
use tranquil_pds::state::AppState;
use tranquil_pds::types::DidRef;
use tranquil_pds::types::Did;
use tranquil_types::Nsid;
static CREATE_ACCOUNT_NSID: LazyLock<Nsid> =
@@ -45,7 +45,7 @@ static PROTECTED_METHODS: LazyLock<HashSet<&'static str>> = LazyLock::new(|| {
#[derive(Deserialize)]
pub struct GetServiceAuthParams {
pub aud: DidRef,
pub aud: Did,
pub lxm: Option<Nsid>,
pub exp: Option<i64>,
}
@@ -118,7 +118,7 @@ pub async fn get_service_auth(
&auth.auth_source,
auth.scope.as_deref(),
params.aud.as_str(),
method,
method.as_str(),
) {
return e.into_response();
}
@@ -171,9 +171,8 @@ pub async fn get_service_auth(
let service_token = match tranquil_pds::auth::create_service_token(
&auth.did,
&params.aud,
lxm,
params.exp,
params.aud.as_str(),
lxm.map(|v| v.as_str()),
&key_bytes,
) {
Ok(t) => t,
+131 -313
View File
@@ -20,7 +20,7 @@ use tranquil_pds::rate_limit::{
check_user_rate_limit_with_message,
};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{AccountState, AtIdentifier, Did, Handle, PlainPassword};
use tranquil_pds::types::{AccountState, Did, Handle, PlainPassword};
use tranquil_types::TokenId;
pub fn verification_blocks_login(channel_verification: &ChannelVerificationStatus) -> bool {
@@ -78,20 +78,10 @@ pub async fn create_session(
"Normalized identifier: {} -> {}",
input.identifier, normalized_identifier
);
let Ok(login_identifier) = AtIdentifier::new(normalized_identifier.as_str()) else {
let _ = verify(
&input.password,
"$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.VTtYw1ZzQKZqmK",
);
warn!("Login identifier is not a valid handle or DID");
return Err(ApiError::AuthenticationFailed(Some(
"Invalid identifier or password".into(),
)));
};
let row = match state
.repos
.user
.get_login_full_by_identifier(&login_identifier)
.get_login_full_by_identifier(normalized_identifier.as_str())
.await
{
Ok(Some(row)) => row,
@@ -122,7 +112,7 @@ pub async fn create_session(
state.repos.session.as_ref(),
row.id,
&input.password,
row.password_hash.as_ref(),
row.password_hash.as_deref(),
)
.await;
let (app_password_name, app_password_scopes, app_password_controller) = match credential {
@@ -220,34 +210,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);
}
@@ -290,7 +267,7 @@ pub async fn create_session(
&row.did,
&key_bytes,
app_password_scopes.as_deref(),
app_password_controller.as_ref(),
app_password_controller.as_deref(),
None,
) {
Ok(m) => m,
@@ -330,39 +307,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();
@@ -380,7 +341,7 @@ pub async fn create_session(
refresh_jwt: refresh_meta.token,
handle,
did: row.did,
did_doc: did_doc.ok(),
did_doc: did_doc.ok().map(|f| (*f).clone()),
email: row.email,
email_confirmed: Some(row.channel_verification.email),
email_auth_factor: email_auth_factor_out,
@@ -473,7 +434,7 @@ pub async fn get_session(
status: account_state.status_for_session().map(String::from),
migrated_to_pds,
migrated_at,
did_doc: did_doc.ok(),
did_doc: did_doc.ok().map(|f| (*f).clone()),
}))
}
Ok(None) => Err(ApiError::AuthenticationFailed(None)),
@@ -491,12 +452,7 @@ pub async fn delete_session(
) -> Result<Json<EmptyResponse>, ApiError> {
let jti = tranquil_pds::auth::extract_jti_from_headers(&headers)
.ok_or(ApiError::AuthenticationRequired)?;
match state
.repos
.session
.delete_session_by_access_jti(&jti, &auth.did)
.await
{
match state.repos.session.delete_session_by_access_jti(&jti).await {
Ok(rows) if rows > 0 => {
let session_cache_key = tranquil_pds::cache_keys::session_key(&auth.did, &jti);
let _ = state.cache.delete(&session_cache_key).await;
@@ -549,15 +505,16 @@ pub async fn refresh_session(
)));
}
};
if let Some(result) = dispatch_refresh_grace(
&state,
&refresh_token,
&refresh_jti,
state.repos.session.lookup_refresh_grace(&refresh_jti).await,
)
.await
if let Ok(Some(_)) = state
.repos
.session
.check_refresh_token_used(&refresh_jti)
.await
{
return result;
warn!("Refresh token reuse detected for jti: {}", refresh_jti);
return Err(ApiError::AuthenticationFailed(Some(
"Refresh token has been revoked due to suspected compromise".into(),
)));
}
let session_row = match state
.repos
@@ -567,18 +524,9 @@ pub async fn refresh_session(
{
Ok(Some(row)) => row,
Ok(None) => {
return dispatch_refresh_grace(
&state,
&refresh_token,
&refresh_jti,
state.repos.session.lookup_refresh_grace(&refresh_jti).await,
)
.await
.unwrap_or_else(|| {
Err(ApiError::AuthenticationFailed(Some(
"Invalid refresh token".into(),
)))
});
return Err(ApiError::AuthenticationFailed(Some(
"Invalid refresh token".into(),
)));
}
Err(e) => {
error!("Database error fetching session: {:?}", e);
@@ -604,7 +552,7 @@ pub async fn refresh_session(
&session_row.did,
&key_bytes,
session_row.scope.as_deref(),
session_row.controller_did.as_ref(),
session_row.controller_did.as_deref(),
None,
) {
Ok(m) => m,
@@ -624,7 +572,6 @@ pub async fn refresh_session(
}
};
let refresh_data = tranquil_db_traits::SessionRefreshData {
did: session_row.did.clone(),
old_refresh_jti: refresh_jti.clone(),
session_id: session_row.id,
new_access_jti: new_access_meta.jti.clone(),
@@ -632,27 +579,22 @@ pub async fn refresh_session(
new_access_expires_at: new_access_meta.expires_at,
new_refresh_expires_at: new_refresh_meta.expires_at,
};
let (access_jwt, refresh_jwt) = match state
match state
.repos
.session
.refresh_session_atomic(&refresh_data)
.await
{
Ok(tranquil_db_traits::RefreshSessionResult::Success) => {
(new_access_meta.token, new_refresh_meta.token)
Ok(tranquil_db_traits::RefreshSessionResult::Success) => {}
Ok(tranquil_db_traits::RefreshSessionResult::TokenAlreadyUsed) => {
warn!("Refresh token reuse detected during atomic operation");
return Err(ApiError::AuthenticationFailed(Some(
"Refresh token has been revoked due to suspected compromise".into(),
)));
}
Ok(tranquil_db_traits::RefreshSessionResult::GraceReplay(replay)) => {
// Lost a benign concurrent rotation; re-mint the winner's tokens
// using this session's signing key (same user, unchanged by rotation).
info!(
"Concurrent refresh within grace window for session_id: {}; replaying tokens",
session_row.id
);
remint_grace_tokens(&replay, &key_bytes)?
}
Ok(tranquil_db_traits::RefreshSessionResult::Compromise) => {
Ok(tranquil_db_traits::RefreshSessionResult::ConcurrentRefresh) => {
warn!(
"Refresh token reuse outside grace window or unreplayable rotation conflict for session_id: {}",
"Concurrent refresh detected for session_id: {}",
session_row.id
);
return Err(ApiError::AuthenticationFailed(Some(
@@ -663,151 +605,12 @@ pub async fn refresh_session(
error!("Database error during session refresh: {:?}", e);
return Err(ApiError::InternalError(None));
}
};
build_refresh_session_output(&state, session_row.did, access_jwt, refresh_jwt).await
}
async fn dispatch_refresh_grace(
state: &AppState,
refresh_token: &str,
presented_jti: &str,
lookup: Result<tranquil_db_traits::RefreshGraceLookup, tranquil_db_traits::DbError>,
) -> Option<Result<Json<RefreshSessionOutput>, ApiError>> {
match lookup {
Ok(tranquil_db_traits::RefreshGraceLookup::NotUsed) => None,
Ok(tranquil_db_traits::RefreshGraceLookup::Replay(replay)) => {
Some(serve_refresh_grace_replay(state, refresh_token, presented_jti, replay).await)
}
Ok(tranquil_db_traits::RefreshGraceLookup::Compromised {
did,
session_id,
key_bytes,
encryption_version,
}) => Some(Err(revoke_compromised_session(
state,
refresh_token,
presented_jti,
did,
session_id,
key_bytes,
encryption_version,
)
.await)),
Err(e) => {
error!("Database error checking refresh token grace: {:?}", e);
Some(Err(ApiError::InternalError(None)))
}
}
}
async fn serve_refresh_grace_replay(
state: &AppState,
refresh_token: &str,
presented_jti: &str,
replay: tranquil_db_traits::RefreshGraceReplay,
) -> Result<Json<RefreshSessionOutput>, ApiError> {
let key =
match tranquil_pds::config::decrypt_key(&replay.key_bytes, Some(replay.encryption_version))
{
Ok(k) => k,
Err(e) => {
error!("Failed to decrypt user key for grace replay: {:?}", e);
return Err(ApiError::InternalError(None));
}
};
if tranquil_pds::auth::verify_refresh_token(refresh_token, &key).is_err() {
return Err(ApiError::AuthenticationFailed(Some(
"Invalid refresh token".into(),
)));
}
info!("Refresh token reuse within grace window for jti: {presented_jti}; replaying tokens");
let (access_jwt, refresh_jwt) = remint_grace_tokens(&replay, &key)?;
build_refresh_session_output(state, replay.did, access_jwt, refresh_jwt).await
}
async fn revoke_compromised_session(
state: &AppState,
refresh_token: &str,
presented_jti: &str,
did: Did,
session_id: SessionId,
key_bytes: Vec<u8>,
encryption_version: i32,
) -> ApiError {
let key = match tranquil_pds::config::decrypt_key(&key_bytes, Some(encryption_version)) {
Ok(k) => k,
Err(e) => {
error!("Failed to decrypt user key for grace check: {:?}", e);
return ApiError::InternalError(None);
}
};
if tranquil_pds::auth::verify_refresh_token(refresh_token, &key).is_err() {
return ApiError::AuthenticationFailed(Some("Invalid refresh token".into()));
}
warn!("Refresh token reuse outside grace window for jti: {presented_jti}; revoking session");
if let Err(e) = state
.repos
.session
.delete_session_by_id(session_id, &did)
.await
{
error!(
"Failed to revoke session {} for refresh token reuse: {:?}",
session_id.as_i32(),
e
);
return ApiError::InternalError(None);
}
ApiError::AuthenticationFailed(Some(
"Refresh token has been revoked due to suspected compromise".into(),
))
}
/// Re-mint the access/refresh JWTs for a grace-window replay from the session's
/// current jtis and signing key. We never persist the signed JWTs; they are
/// reconstructed on demand so a benignly-racing client converges on the same
/// credentials the winning rotation produced. `key_bytes` is the owning user's
/// already-decrypted signing key.
fn remint_grace_tokens(
replay: &tranquil_db_traits::RefreshGraceReplay,
key_bytes: &[u8],
) -> Result<(String, String), ApiError> {
let access_jwt = tranquil_pds::auth::create_access_token_with_jti(
&replay.did,
key_bytes,
replay.scope.as_deref(),
replay.controller_did.as_ref(),
None,
&replay.access_jti,
replay.access_expires_at,
)
.map_err(|e| {
error!("Failed to re-mint access token for grace replay: {:?}", e);
ApiError::InternalError(None)
})?;
let refresh_jwt = tranquil_pds::auth::create_refresh_token_with_jti(
&replay.did,
key_bytes,
&replay.refresh_jti,
replay.refresh_expires_at,
)
.map_err(|e| {
error!("Failed to re-mint refresh token for grace replay: {:?}", e);
ApiError::InternalError(None)
})?;
Ok((access_jwt, refresh_jwt))
}
async fn build_refresh_session_output(
state: &AppState,
did: Did,
access_jwt: String,
refresh_jwt: String,
) -> Result<Json<RefreshSessionOutput>, ApiError> {
let did_for_doc = session_row.did.clone();
let did_resolver = state.did_resolver.clone();
let (db_result, did_doc) = tokio::join!(
state.repos.user.get_session_info_by_did(&did),
did_resolver.fetch_did_document(&did)
state.repos.user.get_session_info_by_did(&session_row.did),
did_resolver.fetch_did_document(&did_for_doc)
);
match db_result {
Ok(Some(u)) => {
@@ -818,10 +621,10 @@ async fn build_refresh_session_output(
let account_state =
AccountState::from_db_fields(u.deactivated_at, u.takedown_ref.clone(), None, None);
Ok(Json(RefreshSessionOutput {
access_jwt,
refresh_jwt,
access_jwt: new_access_meta.token,
refresh_jwt: new_refresh_meta.token,
handle,
did,
did: session_row.did,
email: u.email,
email_confirmed: u.channel_verification.email,
preferred_channel: u.preferred_comms_channel.as_str().to_string(),
@@ -829,12 +632,12 @@ async fn build_refresh_session_output(
preferred_locale: u.preferred_locale,
is_admin: u.is_admin,
active: account_state.is_active(),
did_doc: did_doc.ok(),
did_doc: did_doc.ok().map(|f| (*f).clone()),
status: account_state.status_for_session().map(String::from),
}))
}
Ok(None) => {
error!("User not found for existing session: {}", did);
error!("User not found for existing session: {}", session_row.did);
Err(ApiError::InternalError(None))
}
Err(e) => {
@@ -883,13 +686,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 +702,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 {
@@ -942,6 +747,7 @@ pub async fn confirm_signup(
let session = match crate::identity::provision::create_and_store_session(
&state,
&row.did,
&row.did,
&key_bytes,
"transition:generic transition:chat.bsky",
None,
@@ -953,11 +759,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 +781,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 {
@@ -1013,7 +789,7 @@ pub struct AutoResendResult {
}
pub async fn auto_resend_verification(state: &AppState, did: &Did) -> Option<AutoResendResult> {
let debounce_key = tranquil_pds::cache_keys::auto_verify_sent_key(did);
let debounce_key = tranquil_pds::cache_keys::auto_verify_sent_key(did.as_str());
let debounced = state.cache.get(&debounce_key).await.is_some();
let row = match state.repos.user.get_resend_verification_by_did(did).await {
Ok(Some(row)) => row,
@@ -1033,15 +809,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 +869,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 }))
}
@@ -1154,13 +972,13 @@ pub async fn list_sessions(
Ok(Json(ListSessionsOutput { sessions }))
}
fn extract_client_name(client_id: &tranquil_types::ClientId) -> String {
fn extract_client_name(client_id: &str) -> String {
if client_id.starts_with("http://localhost") || client_id.starts_with("http://127.0.0.1") {
"Localhost App".to_string()
} else if let Ok(parsed) = reqwest::Url::parse(client_id.as_str()) {
} else if let Ok(parsed) = reqwest::Url::parse(client_id) {
parsed.host_str().unwrap_or("Unknown App").to_string()
} else {
client_id.as_str().to_string()
client_id.to_string()
}
}
@@ -1190,7 +1008,7 @@ pub async fn revoke_session(
state
.repos
.session
.delete_session_by_id(session_id, &auth.did)
.delete_session_by_id(session_id)
.await
.log_db_err("deleting session")?;
let cache_key = tranquil_pds::cache_keys::session_key(&auth.did, &access_jti);
@@ -1234,11 +1052,11 @@ pub async fn revoke_all_sessions(
.delete_sessions_by_did(&auth.did)
.await
.log_db_err("revoking JWT sessions")?;
let token_id = TokenId::from(jti.clone().into_inner());
let jti_typed = TokenId::from(jti.clone());
state
.repos
.oauth
.delete_sessions_by_did_except(&auth.did, &token_id)
.delete_sessions_by_did_except(&auth.did, &jti_typed)
.await
.log_db_err("revoking OAuth sessions")?;
} else {
@@ -10,19 +10,17 @@ use serde::{Deserialize, Serialize};
use tracing::{error, info};
use tranquil_pds::api::error::ApiError;
use tranquil_pds::state::AppState;
use tranquil_pds::types::Did;
const SECP256K1_MULTICODEC_PREFIX: [u8; 2] = [0xe7, 0x01];
fn public_key_to_did_key(signing_key: &SigningKey) -> Did {
fn public_key_to_did_key(signing_key: &SigningKey) -> String {
let verifying_key = signing_key.verifying_key();
let compressed_pubkey = verifying_key.to_sec1_bytes();
let mut multicodec_key = Vec::with_capacity(2 + compressed_pubkey.len());
multicodec_key.extend_from_slice(&SECP256K1_MULTICODEC_PREFIX);
multicodec_key.extend_from_slice(&compressed_pubkey);
let encoded = multibase::encode(multibase::Base::Base58Btc, &multicodec_key);
Did::new(format!("did:key:{}", encoded))
.expect("did:key with a multibase-encoded secp256k1 key is a valid DID")
format!("did:key:{}", encoded)
}
#[derive(Deserialize)]
@@ -33,7 +31,7 @@ pub struct ReserveSigningKeyInput {
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ReserveSigningKeyOutput {
pub signing_key: Did,
pub signing_key: String,
}
pub async fn reserve_signing_key(
+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 {}))
}
@@ -52,7 +52,7 @@ impl DeviceTrustState {
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TrustedDevice {
pub id: DeviceId,
pub id: String,
pub user_agent: Option<String>,
pub friendly_name: Option<String>,
pub trusted_at: Option<DateTime<Utc>>,
+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,
+15 -39
View File
@@ -62,27 +62,7 @@ pub async fn handle_telegram_webhook(
&& let Some(from) = message.from
&& let Some(username) = from.username
{
let handle = match parse_start_handle(message.text.as_deref())
.map(tranquil_types::Handle::new)
.transpose()
{
Ok(h) => h,
Err(e) => {
warn!(
telegram_username = %username,
error = %e,
"Ignoring /start with an invalid handle"
);
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;
}
};
let handle = parse_start_handle(message.text.as_deref());
debug!(
telegram_username = %username,
@@ -93,7 +73,7 @@ pub async fn handle_telegram_webhook(
match state
.repos
.user
.store_telegram_chat_id(&username, from.id, handle.as_ref())
.store_telegram_chat_id(&username, from.id, handle.as_deref())
.await
{
Ok(Some(user_id)) => {
@@ -102,21 +82,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 +165,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()),
);
}
}
+9 -10
View File
@@ -58,16 +58,7 @@ pub async fn dereference_scope(
for part in scope_parts {
if let Some(cid_str) = part.strip_prefix("ref:") {
let cid = match Cid::from_str(cid_str) {
Ok(c) => c,
Err(_) => {
tracing::warn!("Invalid CID in scope ref: {}", cid_str);
continue;
}
};
let cache_key =
tranquil_pds::cache_keys::scope_ref_key(&tranquil_types::CidLink::from(&cid));
let cache_key = tranquil_pds::cache_keys::scope_ref_key(cid_str);
if let Some(cached) = state.cache.get(&cache_key).await {
for s in cached.split_whitespace() {
if !resolved_scopes.contains(&s.to_string()) {
@@ -77,6 +68,14 @@ pub async fn dereference_scope(
continue;
}
let cid = match Cid::from_str(cid_str) {
Ok(c) => c,
Err(_) => {
tracing::warn!("Invalid CID in scope ref: {}", cid_str);
continue;
}
};
let block_bytes = match state.block_store.get(&cid).await {
Ok(Some(b)) => b,
Ok(None) => {
+31
View File
@@ -0,0 +1,31 @@
use axum::{
Json,
extract::State,
response::{IntoResponse, Response},
};
use serde::Deserialize;
use tranquil_pds::api::SuccessResponse;
use tranquil_pds::state::AppState;
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ConfirmChannelVerificationInput {
pub channel: tranquil_db_traits::CommsChannel,
pub identifier: String,
pub code: String,
}
pub async fn confirm_channel_verification(
State(state): State<AppState>,
Json(input): Json<ConfirmChannelVerificationInput>,
) -> Response {
let token_input = crate::server::VerifyTokenInput {
token: input.code,
identifier: input.identifier,
};
match crate::server::verify_token_internal(&state, token_input).await {
Ok(_output) => SuccessResponse::ok().into_response(),
Err(e) => e.into_response(),
}
}
-2
View File
@@ -7,7 +7,6 @@ license.workspace = true
[dependencies]
tranquil-config = { workspace = true }
tranquil-crypto = { workspace = true }
tranquil-types = { path = "../tranquil-types", default-features = false }
anyhow = { workspace = true }
base32 = { workspace = true }
@@ -24,4 +23,3 @@ subtle = { workspace = true }
totp-rs = { workspace = true }
urlencoding = { workspace = true }
uuid = { workspace = true }
brotli = "8.0.4"
-183
View File
@@ -1,183 +0,0 @@
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use brotli::{CompressorWriter, Decompressor};
use std::fmt;
use std::io::{Read, Write};
const COMPRESSED_PREFIX: &str = "$br$";
const QUALITY: u32 = 9;
const WINDOW_BITS: u32 = 16;
const BUFFER_SIZE: usize = 4096;
const MAX_SCOPE_LEN: u64 = 64 * 1024;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ScopeDecodeError {
Base64DecodeFailed,
DecompressFailed,
TooLarge,
}
impl fmt::Display for ScopeDecodeError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Base64DecodeFailed => write!(f, "Base64 decode of compressed scope failed"),
Self::DecompressFailed => write!(f, "Brotli decompression of scope failed"),
Self::TooLarge => write!(f, "Decompressed scope exceeds maximum length"),
}
}
}
impl std::error::Error for ScopeDecodeError {}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ScopeEncodeError {
TooLarge,
}
impl fmt::Display for ScopeEncodeError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::TooLarge => write!(f, "Scope exceeds maximum length"),
}
}
}
impl std::error::Error for ScopeEncodeError {}
fn brotli_compress(input: &str) -> Vec<u8> {
let mut writer = CompressorWriter::new(Vec::new(), BUFFER_SIZE, QUALITY, WINDOW_BITS);
writer
.write_all(input.as_bytes())
.expect("writing to a Vec cannot fail");
writer.into_inner()
}
fn brotli_decompress(input: &[u8]) -> Result<String, ScopeDecodeError> {
let mut output = String::new();
Decompressor::new(input, BUFFER_SIZE)
.take(MAX_SCOPE_LEN + 1)
.read_to_string(&mut output)
.map_err(|_| ScopeDecodeError::DecompressFailed)?;
if output.len() as u64 > MAX_SCOPE_LEN {
return Err(ScopeDecodeError::TooLarge);
}
Ok(output)
}
pub fn encode_scope(scope: &str) -> Result<String, ScopeEncodeError> {
if scope.len() as u64 > MAX_SCOPE_LEN {
return Err(ScopeEncodeError::TooLarge);
}
let tagged = format!(
"{COMPRESSED_PREFIX}{}",
URL_SAFE_NO_PAD.encode(brotli_compress(scope))
);
if tagged.len() < scope.len() || scope.starts_with(COMPRESSED_PREFIX) {
Ok(tagged)
} else {
Ok(scope.to_owned())
}
}
pub fn decode_scope(scope: &str) -> Result<String, ScopeDecodeError> {
let Some(encoded) = scope.strip_prefix(COMPRESSED_PREFIX) else {
return Ok(scope.to_owned());
};
let compressed = URL_SAFE_NO_PAD
.decode(encoded)
.map_err(|_| ScopeDecodeError::Base64DecodeFailed)?;
brotli_decompress(&compressed)
}
#[cfg(test)]
mod tests {
use super::*;
fn long_scope() -> String {
let mut scope = String::from("transition:generic transition:chat.bsky");
for collection in [
"social.colibri.message",
"social.colibri.community",
"social.colibri.reaction",
"social.colibri.member",
"social.colibri.channel.read",
] {
scope.push_str(&format!(" repo:{collection}?action=create&action=delete"));
}
scope
}
#[test]
fn long_scope_roundtrips_through_compression() {
let scope = long_scope();
let encoded = encode_scope(&scope).unwrap();
assert!(encoded.starts_with(COMPRESSED_PREFIX));
assert!(encoded.len() < scope.len());
assert_eq!(decode_scope(&encoded).unwrap(), scope);
}
#[test]
fn short_scope_stays_plaintext() {
let encoded = encode_scope("com.atproto.access").unwrap();
assert_eq!(encoded, "com.atproto.access");
assert_eq!(decode_scope(&encoded).unwrap(), "com.atproto.access");
}
#[test]
fn untagged_scope_passes_through() {
assert_eq!(
decode_scope("com.atproto.refresh").unwrap(),
"com.atproto.refresh"
);
assert_eq!(decode_scope("").unwrap(), "");
}
#[test]
fn malformed_compressed_scope_errors_instead_of_panicking() {
assert_eq!(
decode_scope("$br$not valid base64!"),
Err(ScopeDecodeError::Base64DecodeFailed)
);
assert_eq!(
decode_scope("$br$AAAAAAAAAAAAAAAA"),
Err(ScopeDecodeError::DecompressFailed)
);
}
#[test]
fn compression_bomb_is_rejected() {
let bomb = URL_SAFE_NO_PAD.encode(brotli_compress(&"a".repeat(MAX_SCOPE_LEN as usize * 2)));
assert_eq!(
decode_scope(&format!("{COMPRESSED_PREFIX}{bomb}")),
Err(ScopeDecodeError::TooLarge)
);
}
#[test]
fn plaintext_that_looks_compressed_roundtrips() {
let scope = "$br$repo:*";
let encoded = encode_scope(scope).unwrap();
assert!(encoded.starts_with(COMPRESSED_PREFIX));
assert_eq!(decode_scope(&encoded).unwrap(), scope);
}
#[test]
fn encode_rejects_oversized_scope() {
let oversized = "a".repeat(MAX_SCOPE_LEN as usize + 1);
assert_eq!(encode_scope(&oversized), Err(ScopeEncodeError::TooLarge));
assert!(encode_scope(&"a".repeat(MAX_SCOPE_LEN as usize)).is_ok());
}
}
+6 -10
View File
@@ -1,4 +1,3 @@
mod compress;
mod token;
mod totp;
mod types;
@@ -6,15 +5,12 @@ mod verify;
pub use token::{
create_access_token, create_access_token_hs256, create_access_token_hs256_with_metadata,
create_access_token_with_delegation, create_access_token_with_jti,
create_access_token_with_metadata, create_access_token_with_scope_metadata,
create_refresh_token, create_refresh_token_hs256, create_refresh_token_hs256_with_metadata,
create_refresh_token_with_jti, create_refresh_token_with_metadata, create_service_token,
create_service_token_hs256,
create_access_token_with_delegation, create_access_token_with_metadata,
create_access_token_with_scope_metadata, create_refresh_token, create_refresh_token_hs256,
create_refresh_token_hs256_with_metadata, create_refresh_token_with_metadata,
create_service_token, create_service_token_hs256,
};
pub use compress::{ScopeDecodeError, ScopeEncodeError, decode_scope, encode_scope};
pub use totp::{
TotpError, decrypt_totp_secret, encrypt_totp_secret, generate_backup_codes,
generate_qr_png_base64, generate_totp_secret, generate_totp_uri, hash_backup_code,
@@ -28,6 +24,6 @@ pub use types::{
pub use verify::{
get_algorithm_from_token, get_did_from_token, get_jti_from_token, verify_access_token,
verify_access_token_hs256, verify_refresh_token, verify_refresh_token_hs256, verify_token,
verify_token_es256k,
verify_access_token_hs256, verify_access_token_typed, verify_refresh_token,
verify_refresh_token_hs256, verify_token,
};
+47 -115
View File
@@ -1,33 +1,30 @@
use crate::compress::encode_scope;
use super::types::{
ActClaim, Claims, Header, SigningAlgorithm, TokenScope, TokenType, TokenWithMetadata,
};
use anyhow::{Context, Result};
use anyhow::Result;
use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use chrono::{DateTime, Duration, Utc};
use chrono::{Duration, Utc};
use hmac::{Hmac, Mac};
use k256::ecdsa::{Signature, SigningKey, signature::Signer};
use sha2::Sha256;
use tranquil_types::{Did, DidRef, Jti, Nsid};
type HmacSha256 = Hmac<Sha256>;
pub fn create_access_token(did: &Did, key_bytes: &[u8]) -> Result<String> {
pub fn create_access_token(did: &str, key_bytes: &[u8]) -> Result<String> {
Ok(create_access_token_with_metadata(did, key_bytes)?.token)
}
pub fn create_refresh_token(did: &Did, key_bytes: &[u8]) -> Result<String> {
pub fn create_refresh_token(did: &str, key_bytes: &[u8]) -> Result<String> {
Ok(create_refresh_token_with_metadata(did, key_bytes)?.token)
}
pub fn create_access_token_with_metadata(did: &Did, key_bytes: &[u8]) -> Result<TokenWithMetadata> {
pub fn create_access_token_with_metadata(did: &str, key_bytes: &[u8]) -> Result<TokenWithMetadata> {
create_access_token_with_scope_metadata(did, key_bytes, None, None)
}
pub fn create_access_token_with_scope_metadata(
did: &Did,
did: &str,
key_bytes: &[u8],
scopes: Option<&str>,
hostname: Option<&str>,
@@ -44,14 +41,14 @@ pub fn create_access_token_with_scope_metadata(
}
pub fn create_access_token_with_delegation(
did: &Did,
did: &str,
key_bytes: &[u8],
scopes: Option<&str>,
controller_did: Option<&Did>,
controller_did: Option<&str>,
hostname: Option<&str>,
) -> Result<TokenWithMetadata> {
let scope = scopes.unwrap_or(TokenScope::Access.as_str());
let act = controller_did.map(|c| ActClaim { sub: c.clone() });
let act = controller_did.map(|c| ActClaim { sub: c.to_string() });
create_signed_token_with_act(
did,
scope,
@@ -64,7 +61,7 @@ pub fn create_access_token_with_delegation(
}
pub fn create_refresh_token_with_metadata(
did: &Did,
did: &str,
key_bytes: &[u8],
) -> Result<TokenWithMetadata> {
create_signed_token_with_metadata(
@@ -77,80 +74,28 @@ pub fn create_refresh_token_with_metadata(
)
}
/// Re-mint an access token carrying a specific `jti` and expiry. Used by the
/// refresh grace window to reproduce a session's current access token without
/// persisting the signed JWT itself.
pub fn create_access_token_with_jti(
did: &Did,
key_bytes: &[u8],
scopes: Option<&str>,
controller_did: Option<&Did>,
hostname: Option<&str>,
jti: &Jti,
expires_at: DateTime<Utc>,
) -> Result<String> {
let scope = scopes.unwrap_or(TokenScope::Access.as_str());
let act = controller_did.map(|c| ActClaim { sub: c.clone() });
Ok(create_signed_token_pinned(
did,
scope,
TokenType::Access,
key_bytes,
expires_at,
jti.clone(),
act,
hostname,
)?
.token)
}
/// Re-mint a refresh token carrying a specific `jti` and expiry. Counterpart to
/// [`create_access_token_with_jti`] for the refresh grace window.
pub fn create_refresh_token_with_jti(
did: &Did,
key_bytes: &[u8],
jti: &Jti,
expires_at: DateTime<Utc>,
) -> Result<String> {
Ok(create_signed_token_pinned(
did,
TokenScope::Refresh.as_str(),
TokenType::Refresh,
key_bytes,
expires_at,
jti.clone(),
None,
None,
)?
.token)
}
pub fn create_service_token(
did: &Did,
aud: &DidRef,
lxm: Option<&Nsid>,
exp: Option<i64>,
did: &str,
aud: &str,
lxm: Option<&str>,
key_bytes: &[u8],
) -> Result<String> {
let signing_key = SigningKey::from_slice(key_bytes)?;
let expiration = match exp {
Some(exp) => exp,
None => Utc::now()
.checked_add_signed(Duration::seconds(60))
.expect("valid timestamp")
.timestamp(),
};
let expiration = Utc::now()
.checked_add_signed(Duration::seconds(60))
.expect("valid timestamp")
.timestamp();
let claims = Claims {
iss: did.clone(),
sub: did.clone(),
aud: aud.to_string(),
iss: did.to_owned(),
sub: did.to_owned(),
aud: aud.to_owned(),
exp: expiration,
iat: Utc::now().timestamp(),
scope: None,
lxm: lxm.cloned(),
jti: Jti::new(uuid::Uuid::new_v4().to_string()),
lxm: lxm.map(ToOwned::to_owned),
jti: uuid::Uuid::new_v4().to_string(),
act: None,
};
@@ -158,7 +103,7 @@ pub fn create_service_token(
}
fn create_signed_token_with_metadata(
did: &Did,
did: &str,
scope: &str,
typ: TokenType,
key_bytes: &[u8],
@@ -169,7 +114,7 @@ fn create_signed_token_with_metadata(
}
fn create_signed_token_with_act(
did: &Did,
did: &str,
scope: &str,
typ: TokenType,
key_bytes: &[u8],
@@ -177,27 +122,14 @@ fn create_signed_token_with_act(
act: Option<ActClaim>,
hostname: Option<&str>,
) -> Result<TokenWithMetadata> {
let signing_key = SigningKey::from_slice(key_bytes)?;
let expires_at = Utc::now()
.checked_add_signed(duration)
.expect("valid timestamp");
let jti = Jti::new(uuid::Uuid::new_v4().to_string());
create_signed_token_pinned(did, scope, typ, key_bytes, expires_at, jti, act, hostname)
}
#[allow(clippy::too_many_arguments)]
fn create_signed_token_pinned(
did: &Did,
scope: &str,
typ: TokenType,
key_bytes: &[u8],
expires_at: DateTime<Utc>,
jti: Jti,
act: Option<ActClaim>,
hostname: Option<&str>,
) -> Result<TokenWithMetadata> {
let signing_key = SigningKey::from_slice(key_bytes)?;
let expiration = expires_at.timestamp();
let jti = uuid::Uuid::new_v4().to_string();
let aud_hostname = hostname.map(|h| h.to_string()).unwrap_or_else(|| {
tranquil_config::try_get()
@@ -206,12 +138,12 @@ fn create_signed_token_pinned(
});
let claims = Claims {
iss: did.clone(),
sub: did.clone(),
iss: did.to_owned(),
sub: did.to_owned(),
aud: format!("did:web:{}", aud_hostname),
exp: expiration,
iat: Utc::now().timestamp(),
scope: Some(encode_scope(scope).context("Scope too large to encode")?),
scope: Some(scope.to_string()),
lxm: None,
jti: jti.clone(),
act,
@@ -249,16 +181,16 @@ fn sign_claims_with_type(claims: Claims, key: &SigningKey, typ: TokenType) -> Re
Ok(format!("{}.{}", message, signature_b64))
}
pub fn create_access_token_hs256(did: &Did, secret: &[u8]) -> Result<String> {
pub fn create_access_token_hs256(did: &str, secret: &[u8]) -> Result<String> {
Ok(create_access_token_hs256_with_metadata(did, secret)?.token)
}
pub fn create_refresh_token_hs256(did: &Did, secret: &[u8]) -> Result<String> {
pub fn create_refresh_token_hs256(did: &str, secret: &[u8]) -> Result<String> {
Ok(create_refresh_token_hs256_with_metadata(did, secret)?.token)
}
pub fn create_access_token_hs256_with_metadata(
did: &Did,
did: &str,
secret: &[u8],
) -> Result<TokenWithMetadata> {
create_hs256_token_with_metadata(
@@ -271,7 +203,7 @@ pub fn create_access_token_hs256_with_metadata(
}
pub fn create_refresh_token_hs256_with_metadata(
did: &Did,
did: &str,
secret: &[u8],
) -> Result<TokenWithMetadata> {
create_hs256_token_with_metadata(
@@ -284,9 +216,9 @@ pub fn create_refresh_token_hs256_with_metadata(
}
pub fn create_service_token_hs256(
did: &Did,
aud: &Did,
lxm: &Nsid,
did: &str,
aud: &str,
lxm: &str,
secret: &[u8],
) -> Result<String> {
let expiration = Utc::now()
@@ -295,14 +227,14 @@ pub fn create_service_token_hs256(
.timestamp();
let claims = Claims {
iss: did.clone(),
sub: did.clone(),
aud: aud.to_string(),
iss: did.to_owned(),
sub: did.to_owned(),
aud: aud.to_owned(),
exp: expiration,
iat: Utc::now().timestamp(),
scope: None,
lxm: Some(lxm.clone()),
jti: Jti::new(uuid::Uuid::new_v4().to_string()),
lxm: Some(lxm.to_string()),
jti: uuid::Uuid::new_v4().to_string(),
act: None,
};
@@ -310,7 +242,7 @@ pub fn create_service_token_hs256(
}
fn create_hs256_token_with_metadata(
did: &Did,
did: &str,
scope: &str,
typ: TokenType,
secret: &[u8],
@@ -321,11 +253,11 @@ fn create_hs256_token_with_metadata(
.expect("valid timestamp");
let expiration = expires_at.timestamp();
let jti = Jti::new(uuid::Uuid::new_v4().to_string());
let jti = uuid::Uuid::new_v4().to_string();
let claims = Claims {
iss: did.clone(),
sub: did.clone(),
iss: did.to_owned(),
sub: did.to_owned(),
aud: format!(
"did:web:{}",
tranquil_config::try_get()
@@ -334,7 +266,7 @@ fn create_hs256_token_with_metadata(
),
exp: expiration,
iat: Utc::now().timestamp(),
scope: Some(encode_scope(scope).context("Scope too large to encode")?),
scope: Some(scope.to_string()),
lxm: None,
jti: jti.clone(),
act: None,
+10 -11
View File
@@ -2,7 +2,6 @@ use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize, de, ser};
use std::fmt;
use std::str::FromStr;
use tranquil_types::{Did, Jti, Nsid};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TokenType {
@@ -204,21 +203,21 @@ impl std::error::Error for TokenDecodeError {}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ActClaim {
pub sub: Did,
pub sub: String,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct Claims {
pub iss: Did,
pub sub: Did,
pub iss: String,
pub sub: String,
pub aud: String,
pub exp: i64,
pub iat: i64,
#[serde(skip_serializing_if = "Option::is_none")]
pub scope: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub lxm: Option<Nsid>,
pub jti: Jti,
pub lxm: Option<String>,
pub jti: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub act: Option<ActClaim>,
}
@@ -231,8 +230,8 @@ pub struct Header {
#[derive(Debug, Serialize, Deserialize)]
pub struct UnsafeClaims {
pub iss: Did,
pub sub: Option<Did>,
pub iss: String,
pub sub: Option<String>,
}
pub struct TokenData<T> {
@@ -241,21 +240,21 @@ pub struct TokenData<T> {
pub struct TokenWithMetadata {
pub token: String,
pub jti: Jti,
pub jti: String,
pub expires_at: DateTime<Utc>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TokenVerifyError {
Expired,
Invalid(&'static str),
Invalid,
}
impl fmt::Display for TokenVerifyError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Expired => write!(f, "Token expired"),
Self::Invalid(reason) => write!(f, "{}", reason),
Self::Invalid => write!(f, "Token invalid"),
}
}
}
+112 -39
View File
@@ -1,5 +1,3 @@
use crate::compress::decode_scope;
use super::types::{
Claims, Header, SigningAlgorithm, TokenData, TokenDecodeError, TokenScope, TokenType,
TokenVerifyError, UnsafeClaims,
@@ -12,11 +10,10 @@ use hmac::{Hmac, Mac};
use k256::ecdsa::{Signature, SigningKey, VerifyingKey, signature::Verifier};
use sha2::Sha256;
use subtle::ConstantTimeEq;
use tranquil_types::{Did, Jti};
type HmacSha256 = Hmac<Sha256>;
pub fn get_did_from_token(token: &str) -> Result<Did, TokenDecodeError> {
pub fn get_did_from_token(token: &str) -> Result<String, TokenDecodeError> {
let parts: Vec<&str> = token.split('.').collect();
if parts.len() != 3 {
return Err(TokenDecodeError::InvalidFormat);
@@ -32,7 +29,7 @@ pub fn get_did_from_token(token: &str) -> Result<Did, TokenDecodeError> {
Ok(claims.sub.unwrap_or(claims.iss))
}
pub fn get_jti_from_token(token: &str) -> Result<Jti, TokenDecodeError> {
pub fn get_jti_from_token(token: &str) -> Result<String, TokenDecodeError> {
let parts: Vec<&str> = token.split('.').collect();
if parts.len() != 3 {
return Err(TokenDecodeError::InvalidFormat);
@@ -48,7 +45,7 @@ pub fn get_jti_from_token(token: &str) -> Result<Jti, TokenDecodeError> {
claims
.get("jti")
.and_then(|j| j.as_str())
.map(Jti::new)
.map(|s| s.to_string())
.ok_or(TokenDecodeError::MissingClaim)
}
@@ -69,11 +66,11 @@ pub fn get_algorithm_from_token(token: &str) -> Result<SigningAlgorithm, TokenDe
}
pub fn verify_token(token: &str, key_bytes: &[u8]) -> Result<TokenData<Claims>> {
verify_token_es256k(token, key_bytes, None, None).map_err(anyhow::Error::from)
verify_token_internal(token, key_bytes, None, None)
}
pub fn verify_access_token(token: &str, key_bytes: &[u8]) -> Result<TokenData<Claims>> {
verify_token_es256k(
verify_token_internal(
token,
key_bytes,
Some(TokenType::Access),
@@ -83,17 +80,15 @@ pub fn verify_access_token(token: &str, key_bytes: &[u8]) -> Result<TokenData<Cl
TokenScope::AppPassPrivileged,
]),
)
.map_err(anyhow::Error::from)
}
pub fn verify_refresh_token(token: &str, key_bytes: &[u8]) -> Result<TokenData<Claims>> {
verify_token_es256k(
verify_token_internal(
token,
key_bytes,
Some(TokenType::Refresh),
Some(&[TokenScope::Refresh]),
)
.map_err(anyhow::Error::from)
}
pub fn verify_access_token_hs256(token: &str, secret: &[u8]) -> Result<TokenData<Claims>> {
@@ -118,15 +113,15 @@ pub fn verify_refresh_token_hs256(token: &str, secret: &[u8]) -> Result<TokenDat
)
}
pub fn verify_token_es256k(
fn verify_token_internal(
token: &str,
key_bytes: &[u8],
expected_typ: Option<TokenType>,
allowed_scopes: Option<&[TokenScope]>,
) -> Result<TokenData<Claims>, TokenVerifyError> {
) -> Result<TokenData<Claims>> {
let parts: Vec<&str> = token.split('.').collect();
if parts.len() != 3 {
return Err(TokenVerifyError::Invalid("Invalid token format"));
return Err(anyhow!("Invalid token format"));
}
let header_b64 = parts[0];
@@ -135,49 +130,46 @@ pub fn verify_token_es256k(
let header_bytes = URL_SAFE_NO_PAD
.decode(header_b64)
.map_err(|_| TokenVerifyError::Invalid("Base64 decode of header failed"))?;
.context("Base64 decode of header failed")?;
let header: Header = serde_json::from_slice(&header_bytes)
.map_err(|_| TokenVerifyError::Invalid("JSON decode of header failed"))?;
let header: Header =
serde_json::from_slice(&header_bytes).context("JSON decode of header failed")?;
if let Some(expected) = expected_typ
&& header.typ != expected
{
return Err(TokenVerifyError::Invalid("Invalid token type"));
return Err(anyhow!(
"Invalid token type: expected {}, got {}",
expected,
header.typ
));
}
let signature_bytes = URL_SAFE_NO_PAD
.decode(signature_b64)
.map_err(|_| TokenVerifyError::Invalid("Base64 decode of signature failed"))?;
.context("Base64 decode of signature failed")?;
let signature = Signature::from_slice(&signature_bytes)
.map_err(|_| TokenVerifyError::Invalid("Invalid signature format"))?;
.map_err(|e| anyhow!("Invalid signature format: {}", e))?;
let signing_key = SigningKey::from_slice(key_bytes)
.map_err(|_| TokenVerifyError::Invalid("Invalid signing key"))?;
let signing_key = SigningKey::from_slice(key_bytes)?;
let verifying_key = VerifyingKey::from(&signing_key);
let message = format!("{}.{}", header_b64, claims_b64);
verifying_key
.verify(message.as_bytes(), &signature)
.map_err(|_| TokenVerifyError::Invalid("Signature verification failed"))?;
.map_err(|e| anyhow!("Signature verification failed: {}", e))?;
let claims_bytes = URL_SAFE_NO_PAD
.decode(claims_b64)
.map_err(|_| TokenVerifyError::Invalid("Base64 decode of claims failed"))?;
.context("Base64 decode of claims failed")?;
let mut claims: Claims = serde_json::from_slice(&claims_bytes)
.map_err(|_| TokenVerifyError::Invalid("JSON decode of claims failed"))?;
if let Some(scope) = &claims.scope {
claims.scope = Some(
decode_scope(scope).map_err(|_| TokenVerifyError::Invalid("Invalid token scope"))?,
);
}
let claims: Claims =
serde_json::from_slice(&claims_bytes).context("JSON decode of claims failed")?;
let now = Utc::now().timestamp();
if claims.exp < now {
return Err(TokenVerifyError::Expired);
return Err(anyhow!("Token expired"));
}
if let Some(scopes) = allowed_scopes {
@@ -188,7 +180,7 @@ pub fn verify_token_es256k(
.parse()
.unwrap_or_else(|e| match e {});
if !scopes.contains(&token_scope) {
return Err(TokenVerifyError::Invalid("Invalid token scope"));
return Err(anyhow!("Invalid token scope: {}", token_scope));
}
}
@@ -252,13 +244,9 @@ fn verify_token_hs256_internal(
.decode(claims_b64)
.context("Base64 decode of claims failed")?;
let mut claims: Claims =
let claims: Claims =
serde_json::from_slice(&claims_bytes).context("JSON decode of claims failed")?;
if let Some(scope) = &claims.scope {
claims.scope = Some(decode_scope(scope).context("Invalid scope claim encoding")?);
}
let now = Utc::now().timestamp();
if claims.exp < now {
return Err(anyhow!("Token expired"));
@@ -278,3 +266,88 @@ fn verify_token_hs256_internal(
Ok(TokenData { claims })
}
pub fn verify_access_token_typed(
token: &str,
key_bytes: &[u8],
) -> Result<TokenData<Claims>, TokenVerifyError> {
verify_token_typed_internal(token, key_bytes, Some(TokenType::Access), None)
}
fn verify_token_typed_internal(
token: &str,
key_bytes: &[u8],
expected_typ: Option<TokenType>,
allowed_scopes: Option<&[TokenScope]>,
) -> Result<TokenData<Claims>, TokenVerifyError> {
let parts: Vec<&str> = token.split('.').collect();
if parts.len() != 3 {
return Err(TokenVerifyError::Invalid);
}
let header_b64 = parts[0];
let claims_b64 = parts[1];
let signature_b64 = parts[2];
let Ok(header_bytes) = URL_SAFE_NO_PAD.decode(header_b64) else {
return Err(TokenVerifyError::Invalid);
};
let Ok(header) = serde_json::from_slice::<Header>(&header_bytes) else {
return Err(TokenVerifyError::Invalid);
};
if let Some(expected) = expected_typ
&& header.typ != expected
{
return Err(TokenVerifyError::Invalid);
}
let Ok(signature_bytes) = URL_SAFE_NO_PAD.decode(signature_b64) else {
return Err(TokenVerifyError::Invalid);
};
let Ok(signature) = Signature::from_slice(&signature_bytes) else {
return Err(TokenVerifyError::Invalid);
};
let Ok(signing_key) = SigningKey::from_slice(key_bytes) else {
return Err(TokenVerifyError::Invalid);
};
let verifying_key = VerifyingKey::from(&signing_key);
let message = format!("{}.{}", header_b64, claims_b64);
if verifying_key
.verify(message.as_bytes(), &signature)
.is_err()
{
return Err(TokenVerifyError::Invalid);
}
let Ok(claims_bytes) = URL_SAFE_NO_PAD.decode(claims_b64) else {
return Err(TokenVerifyError::Invalid);
};
let Ok(claims) = serde_json::from_slice::<Claims>(&claims_bytes) else {
return Err(TokenVerifyError::Invalid);
};
let now = Utc::now().timestamp();
if claims.exp < now {
return Err(TokenVerifyError::Expired);
}
if let Some(scopes) = allowed_scopes {
let token_scope: TokenScope = claims
.scope
.as_deref()
.unwrap_or("")
.parse()
.unwrap_or_else(|e| match e {});
if !scopes.contains(&token_scope) {
return Err(TokenVerifyError::Invalid);
}
}
Ok(TokenData { claims })
}
+2 -2
View File
@@ -5,16 +5,16 @@ edition.workspace = true
license.workspace = true
[features]
default = []
valkey = ["dep:redis"]
[dependencies]
tranquil-config = { workspace = true }
tranquil-infra = { workspace = true, features = ["cache-keys"] }
tranquil-infra = { workspace = true }
tranquil-ripple = { workspace = true }
async-trait = { workspace = true }
base64 = { workspace = true }
redis = { workspace = true, optional = true }
thiserror = { workspace = true }
tokio-util = { workspace = true }
tracing = { workspace = true }
+33 -20
View File
@@ -1,6 +1,4 @@
pub use tranquil_infra::{
Cache, CacheError, DistributedRateLimiter, cache_keys, cached_json, read_json, write_json,
};
pub use tranquil_infra::{Cache, CacheError, DistributedRateLimiter};
use async_trait::async_trait;
use std::sync::Arc;
@@ -162,28 +160,30 @@ impl Cache for NoOpCache {
}
}
#[derive(Debug, thiserror::Error)]
pub enum CacheInitError {
#[error("ripple config: {0}")]
Config(#[from] tranquil_ripple::RippleConfigError),
#[error("ripple start: {0}")]
Start(#[from] tranquil_ripple::RippleStartError),
pub struct NoOpRateLimiter;
#[async_trait]
impl DistributedRateLimiter for NoOpRateLimiter {
async fn check_rate_limit(&self, _key: &str, _limit: u32, _window_ms: u64) -> bool {
true
}
}
pub async fn create_cache(
shutdown: tokio_util::sync::CancellationToken,
) -> Result<(Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>), CacheInitError> {
) -> (Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>) {
let cache_cfg = tranquil_config::try_get().map(|c| &c.cache);
let backend = cache_cfg.map(|c| c.backend.as_str()).unwrap_or("ripple");
let valkey_url = cache_cfg.and_then(|c| c.valkey_url.as_deref());
#[cfg(feature = "valkey")]
if backend == "valkey" {
if let Some(url) = cache_cfg.and_then(|c| c.valkey_url.as_deref()) {
if let Some(url) = valkey_url {
match ValkeyCache::new(url).await {
Ok(cache) => {
tracing::info!("using valkey cache at {url}");
let rate_limiter = Arc::new(RedisRateLimiter::new(cache.connection()));
return Ok((Arc::new(cache), rate_limiter));
return (Arc::new(cache), rate_limiter);
}
Err(e) => {
tracing::warn!("failed to connect to valkey: {e}. falling back to ripple.");
@@ -201,13 +201,26 @@ pub async fn create_cache(
);
}
let config = tranquil_ripple::RippleConfig::from_config()?;
let peer_count = config.seed_peers.len();
let (cache, rate_limiter, _bound_addr) =
tranquil_ripple::RippleEngine::start(config, shutdown).await?;
match peer_count {
0 => tracing::info!("ripple cache started as a single node"),
n => tracing::info!("ripple cache started with {n} seed peers"),
match tranquil_ripple::RippleConfig::from_config() {
Ok(config) => {
let peer_count = config.seed_peers.len();
match tranquil_ripple::RippleEngine::start(config, shutdown).await {
Ok((cache, rate_limiter, _bound_addr)) => {
match peer_count {
0 => tracing::info!("ripple cache started (single-node)"),
n => tracing::info!("ripple cache started ({n} seed peers)"),
}
(cache, rate_limiter)
}
Err(e) => {
tracing::error!("ripple engine failed to start: {e:#}. running without cache.");
(Arc::new(NoOpCache), Arc::new(NoOpRateLimiter))
}
}
}
Err(e) => {
tracing::error!("ripple config error: {e:#}. running without cache.");
(Arc::new(NoOpCache), Arc::new(NoOpRateLimiter))
}
}
Ok((cache, rate_limiter))
}
+1 -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 }
@@ -20,6 +19,7 @@ reqwest = { workspace = true }
rsa = { workspace = true }
secrecy = { workspace = true }
serde_json = { workspace = true }
sqlx = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true }
tracing = { workspace = true }
+17 -141
View File
@@ -1,43 +1,25 @@
use lettre::Message;
use lettre::message::Mailbox;
use lettre::message::header::{ContentType, MIME_VERSION_1_0};
use lettre::message::header::{Header, HeaderName, HeaderValue};
use lettre::message::header::ContentType;
use uuid::Uuid;
use super::types::EmailDomain;
use crate::sender::SendError;
use crate::{CommsType, QueuedComms};
use crate::types::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()
pub(super) fn build(from: &Mailbox, qc: &QueuedComms) -> Result<Message, SendError> {
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");
let message_id = format!("<{}@{}>", Uuid::new_v4(), from.email.domain());
let builder = Message::builder()
Message::builder()
.from(from.clone())
.to(to)
.subject(subject)
.message_id(Some(message_id))
.header(MIME_VERSION_1_0)
.header(ContentType::TEXT_PLAIN);
let category = apply_atmos_categories
.then(|| atmos_category(qc.comms_type))
.flatten();
let builder = match category {
Some(category) => builder.header(category),
None => builder,
};
builder
.header(ContentType::TEXT_PLAIN)
.body(qc.body.clone())
.map_err(|e| SendError::MessageBuild(e.to_string()))
}
@@ -52,57 +34,10 @@ pub(super) fn recipient_domain(message: &Message) -> Result<EmailDomain, SendErr
.map_err(|e| SendError::InvalidRecipient(format!("invalid recipient domain: {e}")))
}
// for use with comail.at
#[derive(Debug, Clone, PartialEq, Eq, Copy)]
enum AtmosCategory {
PasswordReset,
MfaOtp,
Verification,
}
impl AtmosCategory {
fn as_str(self) -> &'static str {
match self {
Self::PasswordReset => "password-reset",
Self::MfaOtp => "mfa-otp",
Self::Verification => "verification",
}
}
}
impl Header for AtmosCategory {
fn name() -> HeaderName {
HeaderName::new_from_ascii_str("X-Atmos-Category")
}
fn parse(_s: &str) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
//since we're never receiving email, we don't care about parsing
Err("X-Atmos-Category is write-only".into())
}
fn display(&self) -> HeaderValue {
HeaderValue::new(Self::name(), self.as_str().to_string())
}
}
fn atmos_category(comms_type: CommsType) -> Option<AtmosCategory> {
use CommsType::*;
match comms_type {
EmailVerification
| ChannelVerification
| ChannelVerified
| MigrationVerification
| LegacyLoginAlert
| EmailUpdate
| PlcOperation
| AccountDeletion
| Welcome => Some(AtmosCategory::Verification),
PasswordReset | PasskeyRecovery => Some(AtmosCategory::PasswordReset),
TwoFactorCode => Some(AtmosCategory::MfaOtp),
AdminEmail => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{CommsChannel, CommsStatus};
use crate::types::{CommsChannel, CommsStatus, CommsType};
use chrono::Utc;
use uuid::Uuid;
@@ -110,10 +45,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,8 +71,6 @@ mod tests {
let msg = build(
&from_mailbox(),
&fixture("user@nel.pet", Some("Welcome"), "Hello world."),
&to("user@nel.pet"),
false,
)
.unwrap();
let raw = String::from_utf8(msg.formatted()).unwrap();
@@ -149,7 +78,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,8 +87,6 @@ mod tests {
let msg = build(
&from_mailbox(),
&fixture("user@nel.pet", Some("héllo wörld"), "Body"),
&to("user@jola.dev"),
false,
)
.unwrap();
let raw = String::from_utf8(msg.formatted()).unwrap();
@@ -170,8 +96,11 @@ 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"),
);
assert!(matches!(result, Err(SendError::InvalidRecipient(_))));
}
#[test]
@@ -179,14 +108,12 @@ 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: ="),
@@ -196,13 +123,7 @@ mod tests {
#[test]
fn message_id_uses_from_domain() {
let msg = build(
&from_mailbox(),
&fixture("user@nel.pet", Some("s"), "b"),
&to("user@jola.dev"),
false,
)
.unwrap();
let msg = build(&from_mailbox(), &fixture("user@nel.pet", Some("s"), "b")).unwrap();
let raw = String::from_utf8(msg.formatted()).unwrap();
let line = raw
.lines()
@@ -216,60 +137,15 @@ mod tests {
#[test]
fn missing_subject_uses_default() {
let msg = build(
&from_mailbox(),
&fixture("user@nel.pet", None, "Body"),
&to("user@nel.pet"),
false,
)
.unwrap();
let msg = build(&from_mailbox(), &fixture("user@nel.pet", None, "Body")).unwrap();
let raw = String::from_utf8(msg.formatted()).unwrap();
assert!(raw.contains("Subject: Notification"));
}
#[test]
fn recipient_domain_extracted() {
let msg = build(
&from_mailbox(),
&fixture("user@Nel.PET", Some("s"), "b"),
&to("user@nel.pet"),
false,
)
.unwrap();
let msg = build(&from_mailbox(), &fixture("user@Nel.PET", Some("s"), "b")).unwrap();
let d = recipient_domain(&msg).unwrap();
assert_eq!(d.as_str(), "nel.pet");
}
#[test]
fn atmos_category_header_present_when_enabled_and_mapped() {
let qc = QueuedComms {
comms_type: CommsType::PasswordReset,
..fixture("user@nel.pet", Some("s"), "b")
};
let msg = build(&from_mailbox(), &qc, &to("user@jola.dev"), true).unwrap();
let raw = String::from_utf8(msg.formatted()).unwrap();
assert!(raw.contains("X-Atmos-Category: password-reset"));
}
#[test]
fn atmos_category_header_absent_when_disabled() {
let qc = QueuedComms {
comms_type: CommsType::PasswordReset,
..fixture("user@nel.pet", Some("s"), "b")
};
let msg = build(&from_mailbox(), &qc, &to("user@nel.pet"), false).unwrap();
let raw = String::from_utf8(msg.formatted()).unwrap();
assert!(!raw.contains("X-Atmos-Category"));
}
#[test]
fn atmos_category_header_absent_when_unmapped() {
let qc = QueuedComms {
comms_type: CommsType::AdminEmail,
..fixture("user@nel.pet", Some("s"), "b")
};
let msg = build(&from_mailbox(), &qc, &to("user@nel.pet"), true).unwrap();
let raw = String::from_utf8(msg.formatted()).unwrap();
assert!(!raw.contains("X-Atmos-Category"));
}
}
+6 -77
View File
@@ -9,7 +9,6 @@ use std::time::Duration;
use async_trait::async_trait;
use hickory_resolver::TokioAsyncResolver;
use hickory_resolver::config::{ResolverConfig, ResolverOpts};
use lettre::message::Mailbox;
use lettre::transport::smtp::AsyncSmtpTransport;
use lettre::transport::smtp::PoolConfig;
@@ -25,7 +24,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,
@@ -124,18 +123,14 @@ fn build_smarthost(
Ok(SendMode::Smarthost {
transport: Box::new(builder.build()),
total_timeout,
apply_atmos_categories: cfg.email.smarthost.apply_atmos_categories,
})
}
fn build_direct_mx(cfg: &tranquil_config::TranquilConfig) -> Result<SendMode, SendError> {
let helo = resolve_helo(cfg)?;
let resolver = Arc::new(
TokioAsyncResolver::tokio_from_system_conf().unwrap_or_else(|e| {
tracing::warn!("falling back to default DNS resolvers: {}", e);
TokioAsyncResolver::tokio(ResolverConfig::default(), ResolverOpts::default())
}),
);
let resolver = TokioAsyncResolver::tokio_from_system_conf()
.map(Arc::new)
.map_err(|e| config_invalid("system DNS configuration", e))?;
let max_concurrent = cfg.email.direct_mx.max_concurrent_sends.max(1);
Ok(SendMode::DirectMx {
resolver,
@@ -177,38 +172,14 @@ fn build_dkim(cfg: &tranquil_config::DkimConfig) -> Result<Option<DkimSigner>, S
DkimSigner::load(selector, domain, path).map(Some)
}
fn wants_atmos_categories(mode: &SendMode) -> bool {
match mode {
SendMode::Smarthost {
apply_atmos_categories,
..
} => *apply_atmos_categories,
SendMode::DirectMx { .. } => false,
}
}
#[async_trait]
impl CommsSender for EmailSender {
fn channel(&self) -> CommsChannel {
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)?;
if let Some(signer) = &self.dkim {
signer.sign(&mut message);
}
@@ -221,45 +192,3 @@ impl CommsSender for EmailSender {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use lettre::Tokio1Executor;
use std::time::Duration;
fn dummy_smarthost(apply_atmos_categories: bool) -> SendMode {
let transport =
AsyncSmtpTransport::<Tokio1Executor>::builder_dangerous("localhost").build();
SendMode::Smarthost {
transport: Box::new(transport),
total_timeout: Duration::from_secs(10),
apply_atmos_categories,
}
}
fn dummy_direct_mx() -> SendMode {
SendMode::DirectMx {
resolver: Arc::new(TokioAsyncResolver::tokio(
ResolverConfig::default(),
ResolverOpts::default(),
)),
helo: HeloName::parse("mta.nel.pet").unwrap(),
command_timeout: Duration::from_secs(5),
total_timeout: Duration::from_secs(10),
require_tls: false,
inflight: Arc::new(Semaphore::new(1)),
}
}
#[tokio::test]
async fn smarthost_reflects_its_own_flag() {
assert!(wants_atmos_categories(&dummy_smarthost(true)));
assert!(!wants_atmos_categories(&dummy_smarthost(false)));
}
#[test]
fn direct_mx_never_wants_atmos_categories() {
assert!(!wants_atmos_categories(&dummy_direct_mx()));
}
}
+2 -11
View File
@@ -19,7 +19,6 @@ pub enum SendMode {
Smarthost {
transport: Box<AsyncSmtpTransport<Tokio1Executor>>,
total_timeout: Duration,
apply_atmos_categories: bool,
},
DirectMx {
resolver: Arc<TokioAsyncResolver>,
@@ -34,15 +33,8 @@ pub enum SendMode {
impl std::fmt::Debug for SendMode {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Smarthost {
total_timeout,
apply_atmos_categories,
..
} => {
write!(
f,
"SendMode::Smarthost(total_timeout={total_timeout:?}, apply_atmos_categories={apply_atmos_categories:?})"
)
Self::Smarthost { total_timeout, .. } => {
write!(f, "SendMode::Smarthost(total_timeout={total_timeout:?})")
}
Self::DirectMx {
helo, require_tls, ..
@@ -60,7 +52,6 @@ pub async fn dispatch(mode: &SendMode, message: Message) -> Result<(), SendError
SendMode::Smarthost {
transport,
total_timeout,
..
} => with_total_timeout(*total_timeout, run_send(transport, message)).await,
SendMode::DirectMx {
resolver,
+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 -19
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))
}
@@ -57,7 +53,6 @@ fn build_smarthost_sender_with_total_timeout(
SendMode::Smarthost {
transport: Box::new(transport),
total_timeout,
apply_atmos_categories: false,
},
None,
)
@@ -105,9 +100,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 +111,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 +119,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 +145,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) => {}
+1 -2
View File
@@ -5,6 +5,5 @@ edition.workspace = true
license.workspace = true
[dependencies]
serde = { workspace = true }
tranquil-types = { workspace = true }
confique = { workspace = true }
serde = { workspace = true }
+36 -151
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);
@@ -265,9 +259,6 @@ impl TranquilConfig {
// -- tls --------------------------------------------------------------
self.server.tls.validate(&mut errors);
// -- cache ------------------------------------------------------------
self.cache.validate(&mut errors);
// -- SSO providers ----------------------------------------------------
self.validate_sso_provider("sso.github", &self.sso.github, &mut errors);
self.validate_sso_provider("sso.google", &self.sso.google, &mut errors);
@@ -434,7 +425,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,25 +435,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,
/// When set to true, skip age-assurance birthday prompt for all accounts.
#[config(env = "PDS_AGE_ASSURANCE_OVERRIDE", default = false)]
pub age_assurance_override: bool,
@@ -479,10 +458,6 @@ pub struct ServerConfig {
#[config(env = "DISABLE_RATE_LIMITING", default = false)]
pub disable_rate_limiting: bool,
/// Allow outbound fetches to private network addresses. Useful for local development using docker compose.
#[config(env = "ALLOW_PRIVATE_FETCH", default = false)]
pub allow_private_fetch: bool,
/// Skip the verified-comms-channel gate for login and record writes.
/// Please keep this off unless you're an invite-only PDS!
#[config(env = "DISABLE_ACCOUNT_VERIFICATION_GATE", default = false)]
@@ -587,6 +562,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)]
@@ -640,9 +629,7 @@ pub struct SecretsConfig {
#[config(env = "MASTER_KEY")]
pub master_key: Option<String>,
/// Optional operator-held PLC recovery key, as a public `did:key`. The PDS
/// continues to sign PLC operations with the per-account signing key, which
/// always remains in `rotationKeys`.
/// PLC rotation key (DID key). If not set, user-level keys are used.
#[config(env = "PLC_ROTATION_KEY")]
pub plc_rotation_key: Option<String>,
@@ -770,12 +757,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,
}
@@ -803,31 +786,6 @@ pub struct CacheConfig {
pub ripple: RippleCacheConfig,
}
impl CacheConfig {
pub fn validate(&self, errors: &mut Vec<String>) {
let clustered = self
.ripple
.peers
.as_deref()
.unwrap_or(&[])
.iter()
.any(|p| !p.trim().is_empty());
let keyed = self
.ripple
.cluster_key
.as_deref()
.is_some_and(|k| !k.trim().is_empty());
if self.backend == "ripple" && clustered && !keyed && !self.ripple.allow_insecure {
errors.push(
"cache.ripple.peers (RIPPLE_PEERS) is set without cache.ripple.cluster_key \
(RIPPLE_CLUSTER_KEY); set the cluster key to authenticate peers, or set \
cache.ripple.allow_insecure (RIPPLE_ALLOW_INSECURE) for a trusted private network"
.to_string(),
);
}
}
}
#[derive(Debug, Config)]
#[config(layer_attr(serde(deny_unknown_fields)))]
pub struct PlcConfig {
@@ -843,7 +801,7 @@ pub struct PlcConfig {
#[config(env = "PLC_CONNECT_TIMEOUT_SECS", default = 5)]
pub connect_timeout_secs: u64,
/// Seconds to cache DID documents.
/// Seconds to cache DID documents in memory.
#[config(env = "DID_CACHE_TTL_SECS", default = 300)]
pub did_cache_ttl_secs: u64,
}
@@ -865,27 +823,18 @@ pub struct FirehoseConfig {
pub max_concurrent_repo_exports: usize,
/// List of relay / crawler notification URLs.
#[config(env = "CRAWLERS", parse_env = split_comma_list, default = [
// If you know of more relays it makes sense to have here by all means make a PR!
// All we request is that you only add "major" relays.
// What exactly "major" means is up to interpretation and we will make the final call,
// but a good rule of thumb is "most people are likely to add the relay if they know about it"
#[config(env = "CRAWLERS", parse_env = split_comma_list)]
pub crawlers: Option<Vec<String>>,
}
// Microcosm relays
"https://relay.fire.hose.cam",
"https://relay3.fr.hose.cam",
// PBC relay
"https://bsky.network",
// firehose.network relays
"https://northamerica.firehose.network",
"https://europe.firehose.network",
"https://asia.firehose.network",
// Blacksky relay
"https://atproto.africa",
// UpCloud relay
"https://relay.upcloud.world",
])]
pub crawlers: Vec<String>,
impl FirehoseConfig {
/// Returns the list of crawler URLs, falling back to `["https://bsky.network"]`
/// when none are configured.
pub fn crawler_list(&self) -> Vec<String> {
self.crawlers
.clone()
.unwrap_or_else(|| vec!["https://bsky.network".to_string()])
}
}
#[derive(Debug, Config)]
@@ -1132,10 +1081,6 @@ pub struct SmarthostConfig {
/// stuck relay cannot stall the comms queue.
#[config(env = "MAIL_SMARTHOST_TOTAL_TIMEOUT_SECS", default = 60)]
pub total_timeout_secs: u64,
/// Apply Atmos/Comail.at categories for headers to be categorized appropriately.
#[config(env = "MAIL_APPLY_ATMOS_CATEGORIES", default = false)]
pub apply_atmos_categories: bool,
}
#[derive(Debug, Config)]
@@ -1488,21 +1433,18 @@ 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)]
#[config(layer_attr(serde(deny_unknown_fields)))]
pub struct RippleCacheConfig {
/// Address to bind the Ripple gossip protocol listener. With the default
/// value and no cluster_key or peers configured, the listener binds
/// loopback instead and runs as a single node.
/// Address to bind the Ripple gossip protocol listener.
#[config(env = "RIPPLE_BIND", default = "0.0.0.0:0")]
pub bind_addr: String,
@@ -1521,16 +1463,6 @@ pub struct RippleCacheConfig {
/// Maximum cache size in megabytes.
#[config(env = "RIPPLE_CACHE_MAX_MB", default = 256)]
pub cache_max_mb: usize,
/// Pre-shared cluster key authenticating ripple peers. Every node in the
/// cluster must set the same value. When unset, peers are unauthenticated.
#[config(env = "RIPPLE_CLUSTER_KEY")]
pub cluster_key: Option<String>,
/// Allow ripple to bind a non-loopback address without a cluster key.
/// Peers will be unauthenticated. Intended for trusted private networks.
#[config(env = "RIPPLE_ALLOW_INSECURE", default = false)]
pub allow_insecure: bool,
}
#[derive(Debug, Config)]
@@ -1927,52 +1859,6 @@ port = 587
);
}
fn cache_config_for_test(
peers: Option<Vec<String>>,
cluster_key: Option<&str>,
allow_insecure: bool,
) -> CacheConfig {
CacheConfig {
backend: "ripple".to_string(),
valkey_url: None,
ripple: RippleCacheConfig {
bind_addr: "0.0.0.0:0".to_string(),
peers,
machine_id: None,
gossip_interval_ms: 200,
cache_max_mb: 256,
cluster_key: cluster_key.map(str::to_string),
allow_insecure,
},
}
}
#[test]
fn cache_validate_rejects_clustered_keyless_ripple() {
let mut errors = Vec::new();
cache_config_for_test(Some(vec!["10.0.0.7:7000".to_string()]), None, false)
.validate(&mut errors);
assert!(
errors.iter().any(|e| e.contains("RIPPLE_CLUSTER_KEY")),
"expected cluster key error, got {errors:?}"
);
}
#[test]
fn cache_validate_accepts_keyed_insecure_or_standalone() {
let mut errors = Vec::new();
cache_config_for_test(
Some(vec!["10.0.0.7:7000".to_string()]),
Some("nautilus-secret"),
false,
)
.validate(&mut errors);
cache_config_for_test(Some(vec!["10.0.0.7:7000".to_string()]), None, true)
.validate(&mut errors);
cache_config_for_test(None, None, false).validate(&mut errors);
assert!(errors.is_empty(), "expected no errors, got {errors:?}");
}
#[derive(Default)]
struct EmailOverrides {
from_address: Option<&'static str>,
@@ -1998,7 +1884,6 @@ port = 587
pool_size: 4,
command_timeout_secs: 30,
total_timeout_secs: 60,
apply_atmos_categories: false,
},
direct_mx: DirectMxConfig {
command_timeout_secs: 30,
+4 -4
View File
@@ -1,6 +1,6 @@
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use tranquil_types::{AtUri, CidLink, Did, Tid};
use tranquil_types::{AtUri, CidLink, Did};
use uuid::Uuid;
use crate::DbError;
@@ -58,7 +58,7 @@ pub trait BlobRepository: Send + Sync {
limit: i64,
) -> Result<Vec<CidLink>, DbError>;
async fn list_blobs_since_rev(&self, did: &Did, since: &Tid) -> Result<Vec<CidLink>, DbError>;
async fn list_blobs_since_rev(&self, did: &Did, since: &str) -> Result<Vec<CidLink>, DbError>;
async fn count_blobs_by_user(&self, user_id: Uuid) -> Result<i64, DbError>;
@@ -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,
+1 -3
View File
@@ -23,8 +23,7 @@ pub struct DelegationGrant {
#[serde(rename_all = "camelCase")]
pub struct DelegatedAccountInfo {
pub did: Did,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub handle: Option<Handle>,
pub handle: Handle,
pub granted_scopes: DbScope,
pub granted_at: DateTime<Utc>,
}
@@ -33,7 +32,6 @@ pub struct DelegatedAccountInfo {
#[serde(rename_all = "camelCase")]
pub struct ControllerInfo {
pub did: Did,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub handle: Option<Handle>,
pub granted_scopes: DbScope,
pub granted_at: DateTime<Utc>,
-29
View File
@@ -1,31 +1,5 @@
use thiserror::Error;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ColumnRef {
table: &'static str,
column: &'static str,
}
impl ColumnRef {
pub const fn new(table: &'static str, column: &'static str) -> Self {
Self { table, column }
}
pub const fn table(&self) -> &'static str {
self.table
}
pub const fn column(&self) -> &'static str {
self.column
}
}
impl std::fmt::Display for ColumnRef {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}.{}", self.table, self.column)
}
}
#[derive(Debug, Error)]
pub enum DbError {
#[error("Database query error: {0}")]
@@ -55,9 +29,6 @@ pub enum DbError {
#[error("Corrupt data in column: {0}")]
CorruptData(&'static str),
#[error("Column {0} has a value that isn't valid for its type")]
InvalidColumn(ColumnRef),
#[error("Other database error: {0}")]
Other(String),
}
+91 -31
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};
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")]
@@ -95,7 +148,7 @@ pub struct QueuedComms {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InviteCodeInfo {
pub code: InviteCode,
pub code: String,
pub available_uses: i32,
pub state: InviteCodeState,
pub for_account: Option<Did>,
@@ -105,7 +158,7 @@ pub struct InviteCodeInfo {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InviteCodeUse {
pub code: InviteCode,
pub code: String,
pub used_by_did: Did,
pub used_by_handle: Option<Handle>,
pub used_at: DateTime<Utc>,
@@ -113,7 +166,7 @@ pub struct InviteCodeUse {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InviteCodeRow {
pub code: InviteCode,
pub code: String,
pub available_uses: i32,
pub disabled: Option<bool>,
pub created_by_user: Uuid,
@@ -136,7 +189,7 @@ pub struct ReservedSigningKey {
pub struct ReservedSigningKeyFull {
pub id: Uuid,
pub did: Option<Did>,
pub public_key_did_key: Did,
pub public_key_did_key: String,
pub private_key_bytes: Vec<u8>,
pub expires_at: DateTime<Utc>,
pub used_at: Option<DateTime<Utc>>,
@@ -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>,
@@ -194,37 +248,45 @@ pub trait InfraRepository: Send + Sync {
async fn create_invite_code(
&self,
code: &InviteCode,
code: &str,
use_count: i32,
for_account: &Did,
for_account: Option<&Did>,
) -> Result<bool, DbError>;
async fn create_invite_codes_batch(
&self,
codes: &[InviteCode],
codes: &[String],
use_count: i32,
created_by_user: Uuid,
for_account: &Did,
for_account: Option<&Did>,
) -> Result<(), DbError>;
async fn get_invite_code_available_uses(
&self,
code: &InviteCode,
) -> Result<Option<i32>, DbError>;
async fn get_invite_code_available_uses(&self, code: &str) -> Result<Option<i32>, DbError>;
async fn validate_invite_code<'a>(
&self,
code: &'a InviteCode,
code: &'a str,
) -> Result<ValidatedInviteCode<'a>, InviteCodeError>;
async fn decrement_invite_code_uses(
&self,
code: &ValidatedInviteCode<'_>,
) -> Result<(), DbError>;
async fn record_invite_code_use(
&self,
code: &ValidatedInviteCode<'_>,
used_by_user: Uuid,
) -> Result<(), DbError>;
async fn get_invite_codes_for_account(
&self,
for_account: &Did,
) -> Result<Vec<InviteCodeInfo>, DbError>;
async fn get_invite_code_uses(&self, code: &InviteCode) -> Result<Vec<InviteCodeUse>, DbError>;
async fn get_invite_code_uses(&self, code: &str) -> Result<Vec<InviteCodeUse>, DbError>;
async fn disable_invite_codes_by_code(&self, codes: &[InviteCode]) -> Result<(), DbError>;
async fn disable_invite_codes_by_code(&self, codes: &[String]) -> Result<(), DbError>;
async fn disable_invite_codes_by_account(&self, accounts: &[Did]) -> Result<(), DbError>;
@@ -239,7 +301,7 @@ pub trait InfraRepository: Send + Sync {
async fn get_invite_code_uses_batch(
&self,
codes: &[InviteCode],
codes: &[String],
) -> Result<Vec<InviteCodeUse>, DbError>;
async fn get_invites_created_by_user(
@@ -247,20 +309,14 @@ pub trait InfraRepository: Send + Sync {
user_id: Uuid,
) -> Result<Vec<InviteCodeInfo>, DbError>;
async fn get_invite_code_info(
&self,
code: &InviteCode,
) -> Result<Option<InviteCodeInfo>, DbError>;
async fn get_invite_code_info(&self, code: &str) -> Result<Option<InviteCodeInfo>, DbError>;
async fn get_invite_codes_by_users(
&self,
user_ids: &[Uuid],
) -> Result<Vec<(Uuid, InviteCodeInfo)>, DbError>;
async fn get_invite_code_used_by_user(
&self,
user_id: Uuid,
) -> Result<Option<InviteCode>, DbError>;
async fn get_invite_code_used_by_user(&self, user_id: Uuid) -> Result<Option<String>, DbError>;
async fn delete_invite_code_uses_by_user(&self, user_id: Uuid) -> Result<(), DbError>;
@@ -269,14 +325,14 @@ pub trait InfraRepository: Send + Sync {
async fn reserve_signing_key(
&self,
did: Option<&Did>,
public_key_did_key: &Did,
public_key_did_key: &str,
private_key_bytes: &[u8],
expires_at: DateTime<Utc>,
) -> Result<Uuid, DbError>;
async fn get_reserved_signing_key(
&self,
public_key_did_key: &Did,
public_key_did_key: &str,
) -> Result<Option<ReservedSigningKey>, DbError>;
async fn mark_signing_key_used(&self, key_id: Uuid) -> Result<(), DbError>;
@@ -363,6 +419,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,
@@ -376,7 +436,7 @@ pub trait InfraRepository: Send + Sync {
async fn get_invite_code_uses_by_users(
&self,
user_ids: &[Uuid],
) -> Result<Vec<(Uuid, InviteCode)>, DbError>;
) -> Result<Vec<(Uuid, String)>, DbError>;
async fn get_deletion_request_by_did(
&self,
@@ -406,7 +466,7 @@ pub trait InfraRepository: Send + Sync {
async fn get_reserved_signing_key_full(
&self,
public_key_did_key: &Did,
public_key_did_key: &str,
) -> Result<Option<ReservedSigningKeyFull>, DbError>;
async fn get_plc_tokens_by_did(&self, did: &Did) -> Result<Vec<PlcTokenInfo>, DbError>;

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