Dev compose improvements

1. Set the max connections to 20, I frequently see

db-1            | 2026-08-23 08:57:02.784 UTC [12641] FATAL:  sorry, too many clients already

2. Add a wildcard route in Traefik to serve DID documents locally.

3. Expose the PLC port locally so I can point an app at `http://localhost:2582` and be able to do full OAuth flows.

config.dev.toml

@@ -10,6 +10,7 @@ dir = "/app/frontend/public"
 
 [database]
 url = "postgres://postgres:postgres@db:5432/pds"
+max_connections = 20
 
 [storage]
 path = "/var/lib/tranquil-pds/blobs"

crates/tranquil-api/src/moderation/mod.rs

@@ -12,7 +12,7 @@ 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, Nsid};
+use tranquil_pds::types::{Did, DidRef, Nsid};
 
 static CREATE_REPORT_NSID: LazyLock<Nsid> =
     LazyLock::new(|| "com.atproto.moderation.createReport".parse().unwrap());
@@ -151,8 +151,9 @@ async fn proxy_to_report_service(
 
     let service_token = match tranquil_pds::auth::create_service_token(
         &auth_user.did,
-        service_did,
+        &DidRef::from(service_did),
         Some(&CREATE_REPORT_NSID),
+        None,
         &key_bytes,
     ) {
         Ok(t) => t,

crates/tranquil-api/src/server/service_auth.rs

@@ -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::Did;
+use tranquil_pds::types::DidRef;
 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: Did,
+    pub aud: DidRef,
     pub lxm: Option<Nsid>,
     pub exp: Option<i64>,
 }
@@ -169,14 +169,19 @@ pub async fn get_service_auth(
         }
     }
 
-    let service_token =
-        match tranquil_pds::auth::create_service_token(&auth.did, &params.aud, lxm, &key_bytes) {
-            Ok(t) => t,
-            Err(e) => {
-                error!("Failed to create service token: {:?}", e);
-                return ApiError::InternalError(None).into_response();
-            }
-        };
+    let service_token = match tranquil_pds::auth::create_service_token(
+        &auth.did,
+        &params.aud,
+        lxm,
+        params.exp,
+        &key_bytes,
+    ) {
+        Ok(t) => t,
+        Err(e) => {
+            error!("Failed to create service token: {:?}", e);
+            return ApiError::InternalError(None).into_response();
+        }
+    };
     (
         StatusCode::OK,
         Json(GetServiceAuthOutput {

crates/tranquil-auth/src/token.rs

@@ -10,7 +10,7 @@ use chrono::{DateTime, Duration, Utc};
 use hmac::{Hmac, Mac};
 use k256::ecdsa::{Signature, SigningKey, signature::Signer};
 use sha2::Sha256;
-use tranquil_types::{Did, Jti, Nsid};
+use tranquil_types::{Did, DidRef, Jti, Nsid};
 
 type HmacSha256 = Hmac<Sha256>;
 
@@ -127,16 +127,20 @@ pub fn create_refresh_token_with_jti(
 
 pub fn create_service_token(
     did: &Did,
-    aud: &Did,
+    aud: &DidRef,
     lxm: Option<&Nsid>,
+    exp: Option<i64>,
     key_bytes: &[u8],
 ) -> Result<String> {
     let signing_key = SigningKey::from_slice(key_bytes)?;
 
-    let expiration = Utc::now()
-        .checked_add_signed(Duration::seconds(60))
-        .expect("valid timestamp")
-        .timestamp();
+    let expiration = match exp {
+        Some(exp) => exp,
+        None => Utc::now()
+            .checked_add_signed(Duration::seconds(60))
+            .expect("valid timestamp")
+            .timestamp(),
+    };
 
     let claims = Claims {
         iss: did.clone(),

crates/tranquil-comms/src/email/message.rs

@@ -1,25 +1,41 @@
 use lettre::Message;
 use lettre::message::Mailbox;
 use lettre::message::header::ContentType;
+use lettre::message::header::{Header, HeaderName, HeaderValue};
 use uuid::Uuid;
 
 use super::types::EmailDomain;
 use crate::sender::SendError;
-use crate::types::QueuedComms;
+use crate::types::{CommsType, QueuedComms};
 
-pub(super) fn build(from: &Mailbox, qc: &QueuedComms) -> Result<Message, SendError> {
+pub(super) fn build(
+    from: &Mailbox,
+    qc: &QueuedComms,
+    apply_atmos_categories: bool,
+) -> 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());
-    Message::builder()
+    let builder = Message::builder()
         .from(from.clone())
         .to(to)
         .subject(subject)
         .message_id(Some(message_id))
-        .header(ContentType::TEXT_PLAIN)
+        .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
         .body(qc.body.clone())
         .map_err(|e| SendError::MessageBuild(e.to_string()))
 }
@@ -34,10 +50,57 @@ 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::types::{CommsChannel, CommsStatus, CommsType};
+    use crate::types::{CommsChannel, CommsStatus};
     use chrono::Utc;
     use uuid::Uuid;
 
@@ -71,6 +134,7 @@ mod tests {
         let msg = build(
             &from_mailbox(),
             &fixture("user@nel.pet", Some("Welcome"), "Hello world."),
+            false,
         )
         .unwrap();
         let raw = String::from_utf8(msg.formatted()).unwrap();
@@ -87,6 +151,7 @@ mod tests {
         let msg = build(
             &from_mailbox(),
             &fixture("user@nel.pet", Some("héllo wörld"), "Body"),
+            false,
         )
         .unwrap();
         let raw = String::from_utf8(msg.formatted()).unwrap();
@@ -99,6 +164,7 @@ mod tests {
         let result = build(
             &from_mailbox(),
             &fixture("x@nel.pet\r\nBcc: evil@x", Some("s"), "b"),
+            false,
         );
         assert!(matches!(result, Err(SendError::InvalidRecipient(_))));
     }
@@ -108,6 +174,7 @@ mod tests {
         let msg = build(
             &from_mailbox(),
             &fixture("user@nel.pet", Some("hi\r\nBcc: evil@nel.pet"), "body"),
+            false,
         )
         .expect("subject CRLF should be encoded, not rejected");
         let raw = String::from_utf8(msg.formatted()).unwrap();
@@ -123,7 +190,12 @@ mod tests {
 
     #[test]
     fn message_id_uses_from_domain() {
-        let msg = build(&from_mailbox(), &fixture("user@nel.pet", Some("s"), "b")).unwrap();
+        let msg = build(
+            &from_mailbox(),
+            &fixture("user@nel.pet", Some("s"), "b"),
+            false,
+        )
+        .unwrap();
         let raw = String::from_utf8(msg.formatted()).unwrap();
         let line = raw
             .lines()
@@ -137,15 +209,58 @@ mod tests {
 
     #[test]
     fn missing_subject_uses_default() {
-        let msg = build(&from_mailbox(), &fixture("user@nel.pet", None, "Body")).unwrap();
+        let msg = build(
+            &from_mailbox(),
+            &fixture("user@nel.pet", None, "Body"),
+            false,
+        )
+        .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")).unwrap();
+        let msg = build(
+            &from_mailbox(),
+            &fixture("user@Nel.PET", Some("s"), "b"),
+            false,
+        )
+        .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, 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, 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, true).unwrap();
+        let raw = String::from_utf8(msg.formatted()).unwrap();
+        assert!(!raw.contains("X-Atmos-Category"));
+    }
 }

crates/tranquil-comms/src/email/mod.rs

@@ -124,6 +124,7 @@ fn build_smarthost(
     Ok(SendMode::Smarthost {
         transport: Box::new(builder.build()),
         total_timeout,
+        apply_atmos_categories: cfg.email.smarthost.apply_atmos_categories,
     })
 }
 
@@ -176,6 +177,16 @@ 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 {
@@ -183,7 +194,8 @@ impl CommsSender for EmailSender {
     }
 
     async fn send(&self, notification: &QueuedComms) -> Result<(), SendError> {
-        let mut message = message::build(&self.from, notification)?;
+        let mut message =
+            message::build(&self.from, notification, wants_atmos_categories(&self.mode))?;
         if let Some(signer) = &self.dkim {
             signer.sign(&mut message);
         }
@@ -196,3 +208,45 @@ 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()));
+    }
+}

crates/tranquil-comms/src/email/transport.rs

@@ -19,6 +19,7 @@ pub enum SendMode {
     Smarthost {
         transport: Box<AsyncSmtpTransport<Tokio1Executor>>,
         total_timeout: Duration,
+        apply_atmos_categories: bool,
     },
     DirectMx {
         resolver: Arc<TokioAsyncResolver>,
@@ -33,8 +34,15 @@ 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, .. } => {
-                write!(f, "SendMode::Smarthost(total_timeout={total_timeout:?})")
+            Self::Smarthost {
+                total_timeout,
+                apply_atmos_categories,
+                ..
+            } => {
+                write!(
+                    f,
+                    "SendMode::Smarthost(total_timeout={total_timeout:?}, apply_atmos_categories={apply_atmos_categories:?})"
+                )
             }
             Self::DirectMx {
                 helo, require_tls, ..
@@ -52,6 +60,7 @@ 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,

crates/tranquil-comms/tests/email_smtp.rs

@@ -53,6 +53,7 @@ fn build_smarthost_sender_with_total_timeout(
         SendMode::Smarthost {
             transport: Box::new(transport),
             total_timeout,
+            apply_atmos_categories: false,
         },
         None,
     )

crates/tranquil-config/src/lib.rs

@@ -1124,6 +1124,10 @@ 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)]
@@ -1985,6 +1989,7 @@ port = 587
                 pool_size: 4,
                 command_timeout_secs: 30,
                 total_timeout_secs: 60,
+                apply_atmos_categories: false,
             },
             direct_mx: DirectMxConfig {
                 command_timeout_secs: 30,

crates/tranquil-oauth/src/client.rs

@@ -142,7 +142,7 @@ impl ClientMetadataCache {
             response_types: vec!["code".into()],
             scope,
             token_endpoint_auth_method: Some("none".into()),
-            dpop_bound_access_tokens: Some(false),
+            dpop_bound_access_tokens: Some(true),
             jwks: None,
             jwks_uri: None,
             application_type: Some("native".into()),

crates/tranquil-pds/src/api/proxy.rs

@@ -5,7 +5,7 @@ use std::sync::LazyLock;
 use crate::api::error::ApiError;
 use crate::api::proxy_client::proxy_client;
 use crate::state::AppState;
-use crate::types::{Did, Nsid};
+use crate::types::{Did, DidRef, Nsid};
 use crate::util::get_header_str;
 use axum::{
     body::Bytes,
@@ -361,8 +361,9 @@ async fn proxy_handler(
 
                 match crate::auth::create_service_token(
                     &auth_user.did,
-                    &token_aud,
+                    &DidRef::from(token_aud),
                     Some(&token_lxm),
+                    None,
                     &key_bytes,
                 ) {
                     Ok(new_token) => {

crates/tranquil-pds/tests/jwt_security.rs

@@ -14,7 +14,7 @@ use tranquil_pds::auth::{
     get_did_from_token, get_jti_from_token, verify_access_token, verify_refresh_token,
     verify_token,
 };
-use tranquil_types::{Did, Nsid};
+use tranquil_types::{Did, DidRef, Nsid};
 
 fn generate_user_key() -> Vec<u8> {
     let secret_key = SecretKey::random(&mut OsRng);
@@ -169,8 +169,9 @@ fn test_token_type_confusion() {
 
     let service_token = create_service_token(
         &did,
-        &Did::new("did:web:nel.pet").expect("valid DID"),
+        &DidRef::new("did:web:nel.pet").expect("valid DID reference"),
         Some(&Nsid::new("cafe.oyster.method").expect("valid NSID")),
+        None,
         &key_bytes,
     )
     .unwrap();

crates/tranquil-pds/tests/oauth.rs

@@ -1270,6 +1270,63 @@ async fn test_granular_scope_rpc_specific_method() {
     );
 }
 
+#[tokio::test]
+async fn test_granular_scope_rpc_aud_with_service_id() {
+    let url = base_url().await;
+    let http_client = client();
+    let (token, _, _) = get_oauth_token_with_scope(
+        "rpc:app.bsky.feed.getTimeline?aud=did:web:api.bsky.app#bsky_appview",
+    )
+    .await;
+    let allowed_res = http_client
+        .get(format!("{}/xrpc/com.atproto.server.getServiceAuth", url))
+        .bearer_auth(&token)
+        .query(&[
+            ("aud", "did:web:api.bsky.app#bsky_appview"),
+            ("lxm", "app.bsky.feed.getTimeline"),
+        ])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(
+        allowed_res.status(),
+        StatusCode::OK,
+        "the granted service id must cover a request naming it"
+    );
+    let body: Value = allowed_res.json().await.unwrap();
+    let service_token = body["token"].as_str().unwrap();
+    let payload = service_token.split('.').nth(1).unwrap();
+    let claims: Value = serde_json::from_slice(&URL_SAFE_NO_PAD.decode(payload).unwrap()).unwrap();
+    assert_eq!(
+        claims["aud"], "did:web:api.bsky.app#bsky_appview",
+        "the service id must reach the signed claim even on the granular scope path"
+    );
+
+    for (aud, reason) in [
+        (
+            "did:web:api.bsky.app#atproto_labeler",
+            "a scope for the appview must not mint tokens for the labeler on the same DID",
+        ),
+        (
+            "did:web:api.bsky.app",
+            "a scope for one service must not widen to the whole DID",
+        ),
+        (
+            "did:web:other.example#bsky_appview",
+            "a service id must not smuggle in a different audience",
+        ),
+    ] {
+        let blocked_res = http_client
+            .get(format!("{}/xrpc/com.atproto.server.getServiceAuth", url))
+            .bearer_auth(&token)
+            .query(&[("aud", aud), ("lxm", "app.bsky.feed.getTimeline")])
+            .send()
+            .await
+            .unwrap();
+        assert_eq!(blocked_res.status(), StatusCode::FORBIDDEN, "{reason}");
+    }
+}
+
 #[tokio::test]
 async fn test_oauth_metadata_includes_prompt_values_supported() {
     let url = base_url().await;

crates/tranquil-pds/tests/scope_edge_cases.rs

@@ -181,6 +181,52 @@ fn test_permissions_rpc_lxm_wildcard_prefix() {
     assert!(!perms.allows_rpc("did:web:api.bsky.app", &c("app.bsky.actor.getProfile")));
 }
 
+#[test]
+fn test_permissions_rpc_aud_service_id_must_match_verbatim() {
+    let perms =
+        ScopePermissions::from_scope_string(Some("rpc:app.bsky.feed.*?aud=did:web:api.bsky.app"));
+    assert!(
+        perms.allows_rpc("did:web:api.bsky.app", &c("app.bsky.feed.getTimeline")),
+        "the granted audience must cover itself"
+    );
+    assert!(
+        !perms.allows_rpc(
+            "did:web:api.bsky.app#bsky_appview",
+            &c("app.bsky.feed.getTimeline")
+        ),
+        "a bare DID grants nothing to the services listed under it"
+    );
+    assert!(
+        !perms.allows_rpc(
+            "did:web:other.example#bsky_appview",
+            &c("app.bsky.feed.getTimeline")
+        ),
+        "a service id must not smuggle in a different audience"
+    );
+
+    let fragment_scope = ScopePermissions::from_scope_string(Some(
+        "rpc:app.bsky.feed.*?aud=did:web:api.bsky.app%23bsky_appview",
+    ));
+    assert!(
+        fragment_scope.allows_rpc(
+            "did:web:api.bsky.app#bsky_appview",
+            &c("app.bsky.feed.getTimeline")
+        ),
+        "the granted service id must cover itself"
+    );
+    assert!(
+        !fragment_scope.allows_rpc("did:web:api.bsky.app", &c("app.bsky.feed.getTimeline")),
+        "a scope granted for one service must not widen to the whole DID"
+    );
+    assert!(
+        !fragment_scope.allows_rpc(
+            "did:web:api.bsky.app#atproto_labeler",
+            &c("app.bsky.feed.getTimeline")
+        ),
+        "the appview and the labeler are different audiences even on one DID"
+    );
+}
+
 #[test]
 fn test_delegation_intersect_mismatched_params_empty() {
     let result = intersect_scopes("repo:*?action=create", "repo:*?action=delete");

crates/tranquil-pds/tests/server.rs

@@ -132,6 +132,46 @@ async fn test_service_auth() {
     let lxm_payload = URL_SAFE_NO_PAD.decode(lxm_parts[1]).unwrap();
     let lxm_claims: Value = serde_json::from_slice(&lxm_payload).unwrap();
     assert_eq!(lxm_claims["lxm"], "com.atproto.repo.getRecord");
+    let fragment_res = client
+        .get(format!("{}/xrpc/com.atproto.server.getServiceAuth", base))
+        .bearer_auth(&access_jwt)
+        .query(&[
+            ("aud", "did:web:example.com#colibri_appview"),
+            ("lxm", "com.atproto.repo.getRecord"),
+        ])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(fragment_res.status(), StatusCode::OK);
+    let fragment_body: Value = fragment_res.json().await.unwrap();
+    let fragment_token = fragment_body["token"].as_str().unwrap();
+    let fragment_parts: Vec<&str> = fragment_token.split('.').collect();
+    let fragment_payload = URL_SAFE_NO_PAD.decode(fragment_parts[1]).unwrap();
+    let fragment_claims: Value = serde_json::from_slice(&fragment_payload).unwrap();
+    assert_eq!(
+        fragment_claims["aud"], "did:web:example.com#colibri_appview",
+        "the service id must survive into the signed claim so the receiver can match it \
+         against its own DID document"
+    );
+
+    let empty_fragment = client
+        .get(format!("{}/xrpc/com.atproto.server.getServiceAuth", base))
+        .bearer_auth(&access_jwt)
+        .query(&[("aud", "did:web:example.com#")])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(empty_fragment.status(), StatusCode::BAD_REQUEST);
+
+    let double_fragment = client
+        .get(format!("{}/xrpc/com.atproto.server.getServiceAuth", base))
+        .bearer_auth(&access_jwt)
+        .query(&[("aud", "did:web:example.com#a#b")])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(double_fragment.status(), StatusCode::BAD_REQUEST);
+
     let unauth = client
         .get(format!("{}/xrpc/com.atproto.server.getServiceAuth", base))
         .query(&[("aud", "did:web:example.com")])

crates/tranquil-scopes/src/permissions.rs

@@ -177,8 +177,6 @@ impl ScopePermissions {
             return Ok(());
         }
 
-        let aud_base = aud.split('#').next().unwrap_or(aud);
-
         let has_permission = self.find_rpc_scopes().any(|rpc_scope| {
             let lxm_matches = match &rpc_scope.lxm {
                 None => true,
@@ -193,10 +191,7 @@ impl ScopePermissions {
             let aud_matches = match &rpc_scope.aud {
                 None => true,
                 Some(scope_aud) if scope_aud == "*" => true,
-                Some(scope_aud) => {
-                    let scope_aud_base = scope_aud.split('#').next().unwrap_or(scope_aud);
-                    scope_aud_base == aud_base
-                }
+                Some(scope_aud) => scope_aud == aud,
             };
 
             lxm_matches && aud_matches
@@ -558,27 +553,47 @@ mod tests {
         let perms = ScopePermissions::from_scope_string(Some(
             "rpc:app.bsky.feed.getAuthorFeed?aud=did:web:api.bsky.app#bsky_appview",
         ));
-        assert!(perms.allows_rpc("did:web:api.bsky.app", &c("app.bsky.feed.getAuthorFeed")));
         assert!(perms.allows_rpc(
             "did:web:api.bsky.app#bsky_appview",
             &c("app.bsky.feed.getAuthorFeed")
         ));
-        assert!(perms.allows_rpc(
-            "did:web:api.bsky.app#other_service",
-            &c("app.bsky.feed.getAuthorFeed")
-        ));
+        assert!(
+            !perms.allows_rpc("did:web:api.bsky.app", &c("app.bsky.feed.getAuthorFeed")),
+            "a scope naming one service must not cover the whole DID"
+        );
+        assert!(
+            !perms.allows_rpc(
+                "did:web:api.bsky.app#atproto_labeler",
+                &c("app.bsky.feed.getAuthorFeed")
+            ),
+            "a scope naming one service must not cover a sibling service on the same DID"
+        );
         assert!(!perms.allows_rpc("did:web:other.app", &c("app.bsky.feed.getAuthorFeed")));
         assert!(!perms.allows_rpc("did:web:api.bsky.app", &c("app.bsky.feed.getTimeline")));
     }
 
     #[test]
-    fn test_rpc_scope_without_fragment_matches_with_fragment() {
+    fn test_rpc_scope_without_fragment_does_not_cover_service_ids() {
         let perms = ScopePermissions::from_scope_string(Some(
             "rpc:app.bsky.feed.getAuthorFeed?aud=did:web:api.bsky.app",
         ));
         assert!(perms.allows_rpc("did:web:api.bsky.app", &c("app.bsky.feed.getAuthorFeed")));
+        assert!(
+            !perms.allows_rpc(
+                "did:web:api.bsky.app#bsky_appview",
+                &c("app.bsky.feed.getAuthorFeed")
+            ),
+            "the audience is compared verbatim, so a bare DID grants nothing to its services"
+        );
+    }
+
+    #[test]
+    fn test_rpc_scope_aud_wildcard_covers_service_ids() {
+        let perms =
+            ScopePermissions::from_scope_string(Some("rpc:app.bsky.feed.getAuthorFeed?aud=*"));
+        assert!(perms.allows_rpc("did:web:api.bsky.app", &c("app.bsky.feed.getAuthorFeed")));
         assert!(perms.allows_rpc(
-            "did:web:api.bsky.app#bsky_appview",
+            "did:web:api.bsky.app#atproto_labeler",
             &c("app.bsky.feed.getAuthorFeed")
         ));
     }

crates/tranquil-types/src/lib.rs

@@ -225,6 +225,52 @@ impl Did {
     }
 }
 
+const DID_REF_MAX_LEN: usize = 2048;
+const SERVICE_ID_MAX_LEN: usize = 128;
+
+const fn is_pchar(b: u8) -> bool {
+    b.is_ascii_alphanumeric() || matches!(b, b'-' | b'.' | b'_' | b'~')
+}
+
+fn is_service_id(s: &str) -> bool {
+    !s.is_empty() && s.len() <= SERVICE_ID_MAX_LEN && s.bytes().all(is_pchar)
+}
+
+validated_string_newtype! {
+    pub struct DidRef;
+    error = DidRefError;
+    label = "DID reference";
+    validator = |s| {
+        if s.len() > DID_REF_MAX_LEN {
+            return Err(());
+        }
+        match s.split_once('#') {
+            None => jacquard_common::types::string::Did::new(s)
+                .map(|v| v.as_str().to_owned())
+                .map_err(|_| ()),
+            Some((did, service_id)) => {
+                if !is_service_id(service_id) {
+                    return Err(());
+                }
+                let base = jacquard_common::types::string::Did::new(did).map_err(|_| ())?;
+                Ok(format!("{}#{}", base.as_str(), service_id))
+            }
+        }
+    };
+}
+
+impl From<Did> for DidRef {
+    fn from(did: Did) -> Self {
+        Self(did.0)
+    }
+}
+
+impl From<&Did> for DidRef {
+    fn from(did: &Did) -> Self {
+        Self(did.0.clone())
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, sqlx::Type)]
 #[serde(transparent)]
 #[sqlx(transparent)]
@@ -1588,6 +1634,126 @@ mod validated_newtype_tests {
         );
     }
 
+    #[test]
+    fn a_bare_did_ref_names_no_service() {
+        let aud = DidRef::new("did:plc:abc").unwrap();
+        assert_eq!(
+            aud.as_str(),
+            "did:plc:abc",
+            "an absent fragment is not the same as an empty one, so nothing may be appended"
+        );
+    }
+
+    #[test]
+    fn a_did_ref_keeps_the_service_id_it_was_given() {
+        let aud = DidRef::new("did:web:api.colibri.social#colibri_appview").unwrap();
+        assert_eq!(
+            aud.as_str(),
+            "did:web:api.colibri.social#colibri_appview",
+            "the fragment is what tells the receiver which of its services was audienced, \
+             so it must survive entirely"
+        );
+    }
+
+    #[test]
+    fn a_did_ref_normalizes_its_did_half_the_way_a_did_does() {
+        assert_eq!(
+            DidRef::new("at://did:plc:abc#colibri_appview")
+                .unwrap()
+                .as_str(),
+            "did:plc:abc#colibri_appview"
+        );
+        assert_eq!(
+            DidRef::new("did:plc:def").unwrap().as_str(),
+            Did::new("did:plc:def").unwrap().as_str(),
+            "a fragmentless DidRef must be byte-identical to the Did it replaces"
+        );
+    }
+
+    #[test]
+    fn a_did_ref_rejects_anything_that_cannot_name_one_service() {
+        for bad in [
+            "did:web:oyster.cafe#",
+            "did:web:oyster.cafe#a#b",
+            "did:web:oyster.cafe# whelk",
+            "did:web:oyster.cafe#a/b",
+            "did:web:oyster.cafe#a?b",
+            "did:web:oyster.cafe#<script>",
+            "did:web:oyster.cafe#a%20b",
+            "did:web:oyster.cafe#a%5Fb",
+            "did:web:oyster.cafe#a\"b",
+            "did:web:oyster.cafe#a[b]",
+            "did:web:oyster.cafe#atproto_p\u{200b}ds",
+            "did:web:oyster.cafe#\u{feff}atproto_pds",
+            "not-a-did#colibri_appview",
+            "#colibri_appview",
+        ] {
+            assert!(
+                DidRef::new(bad).is_err(),
+                "{bad} should not parse as a DID reference"
+            );
+        }
+    }
+
+    #[test]
+    fn a_did_ref_does_not_second_guess_the_service_ids_it_has_not_seen() {
+        for good in [
+            "did:web:oyster.cafe#atproto_pds",
+            "did:web:oyster.cafe#atproto_labeler",
+            "did:plc:abc#bsky_chat",
+            "did:web:oyster.cafe#whelk.v2",
+            "did:web:oyster.cafe#atproto~pds",
+        ] {
+            assert!(
+                DidRef::new(good).is_ok(),
+                "{good} names a service the receiver resolves in its own DID document, \
+                 so rejecting it here would recreate the bug this type exists to fix"
+            );
+        }
+    }
+
+    #[test]
+    fn an_over_long_service_id_is_rejected_before_the_did_ref_bound_bites() {
+        let did = "did:web:oyster.cafe";
+        let longest_accepted = format!("{did}#{}", "a".repeat(SERVICE_ID_MAX_LEN));
+        assert!(DidRef::new(&longest_accepted).is_ok());
+        let one_too_long = format!("{did}#{}", "a".repeat(SERVICE_ID_MAX_LEN + 1));
+        assert!(
+            DidRef::new(&one_too_long).is_err(),
+            "no service names itself in more than {SERVICE_ID_MAX_LEN} bytes, and the 2048-byte \
+             aud bound is far too loose to catch a fragment used as a payload"
+        );
+    }
+
+    #[test]
+    fn an_over_long_did_ref_is_rejected() {
+        let did = format!("did:plc:{}", "a".repeat(DID_REF_MAX_LEN - "did:plc:".len()));
+        assert_eq!(did.len(), DID_REF_MAX_LEN);
+        assert!(
+            Did::new(&did).is_ok(),
+            "the DID half has to stand on its own, or the bound below proves nothing"
+        );
+        assert!(
+            DidRef::new(format!("{did}#x")).is_err(),
+            "the lexicon bounds aud at {DID_REF_MAX_LEN} bytes, which a whole DID plus the shortest service id already exceeds"
+        );
+    }
+
+    #[test]
+    fn a_did_ref_built_from_a_did_names_no_service() {
+        let did = Did::new("did:plc:def").unwrap();
+        assert_eq!(
+            DidRef::from(&did).as_str(),
+            did.as_str(),
+            "a DID that named no service must not gain one on the way in"
+        );
+        assert_eq!(
+            DidRef::from(did.clone()).as_str(),
+            did.as_str(),
+            "the owned conversion must land on the same bytes as the borrowed one"
+        );
+    }
+
     #[test]
     fn the_earliest_tid_sorts_below_every_generated_tid() {
         let earliest = Tid::earliest();

docker-compose.yaml

@@ -96,6 +96,8 @@ services:
     depends_on:
       db:
         condition: service_healthy
+    ports:
+      - "2582:2582"
 
   mailpit:
     profiles: [dev]

docs/3_DEBUGGING_YOUR_SETUP.md

@@ -19,4 +19,4 @@ Fixing this isn't too hard. You want to figure out what `seq` the relays think y
 >
 > #protip double check that the `seq` is actually misaligned with your PDS by getting the latest `seq` from your PDS using PDSls's firehose feature and a cursor value of `0` when connecting. If the relays don't have a `seq` that's bigger than your PDSs `seq` then this isn't your issue!
 
-Now you need to update the PDS to use a `seq` that's *at least* one above the highest `seq` any of the relays have. Currently doing this on Tranquil depends on your used storage backend. For the default Postgres repo store you want to shutdown Tranquil itself, open the DB in `psql` and run `SELECT setval('firehose_seq, <new updated seq value>');` and the start Tranquil back up. Take a few actions to make sure new events get sent out. Everything should work once all the downstream consumers have had a chance to resync. You can double check with debug.hose.cam that all the relays show your PDS as `active`. Some might need a lil push with a request crawl, luckily debug.hose.cam also makes that easy :p (ignore that it says it failed to issue a request crawl, CORS is fickle).
+Now you need to update the PDS to use a `seq` that's *at least* one above the highest `seq` any of the relays have. Currently doing this on Tranquil depends on your used storage backend. For the default Postgres repo store you want to shutdown Tranquil itself, open the DB in `psql` and run `SELECT setval('firehose_seq', <new updated seq integer value>);` and the start Tranquil back up. Take a few actions to make sure new events get sent out. Everything should work once all the downstream consumers have had a chance to resync. You can double check with debug.hose.cam that all the relays show your PDS as `active`. Some might need a lil push with a request crawl, luckily debug.hose.cam also makes that easy :p (ignore that it says it failed to issue a request crawl, CORS is fickle).

example.toml

@@ -501,6 +501,13 @@
 # Default value: 60
 #total_timeout_secs = 60
 
+# Apply Atmos/Comail.at categories for headers to be categorized appropriately.
+#
+# Can also be specified via environment variable `MAIL_APPLY_ATMOS_CATEGORIES`.
+#
+# Default value: false
+#apply_atmos_categories = false
+
 [email.direct_mx]
 # Per-command SMTP timeout in seconds.
 #

traefik.dev.yaml

@@ -18,6 +18,12 @@ http:
       service: frontend
       priority: 1
       tls: {}
+    handles:
+      rule: 'HostRegexp(`^.+\.pds\.test$`)'
+      entryPoints:
+        - websecure
+      service: backend
+      tls: {}
 
   services:
     backend: