feat(appview): Bearer-Auth für Timeline und Notifications

Die AppView hatte keinerlei Authentifizierung: jeder konnte
/api/notifications?did=<beliebig> lesen und per /seen als gelesen
markieren. Mit Phase 8 sind das die ersten privaten Daten im System.

Das Access-JWT der PDS trug von Anfang an sub, scope
"com.atproto.access" und aud "did:web:appview…" — es war für die
AppView ausgestellt, nur hat sie es nie geprüft. Neu ist deshalb vor
allem die Schlüsselbeschaffung: auth.rs holt das DID-Dokument der PDS
(PDS_INTERNAL_URL, sonst PDS_PUBLIC_URL), cached den Schlüssel und lädt
ihn bei einem Verifikationsfehler nach — höchstens einmal pro Minute,
damit Müll-Tokens kein Werkzeug werden, die PDS zu fluten. Ein
Schlüsselwechsel braucht damit keinen Neustart.

Ist die PDS beim Start weg, warnt die AppView nur und startet trotzdem
(sie indiziert den Firehose, der von der lokalen PDS unabhängig ist).
Ist der Schlüssel beim Prüfen eines Tokens nicht zu beschaffen, gibt es
503 — fail closed.

Geschützt: /api/timeline/home und die drei Notification-Endpoints, jeweils
mit sub == did. Öffentlich bleiben Profile, Suche, Posts, Threads und die
Follower-Listen; das sind in AT Proto öffentliche Records.

401 AuthMissing / 401 TokenInvalid / 403 Forbidden / 503 AuthUnavailable.
TokenInvalid ist ein Vertrag mit dem Client: daran erkennt er, dass er
sein Token erneuern und einmal wiederholen muss.

Dazu CORS: statt Any für alles jetzt eine Allowlist über
APPVIEW_CORS_ORIGINS (unset = altes Verhalten plus Warnung), und
/internal/ingest-commit liegt außerhalb der CORS-Schicht — die Route
wird server-zu-server aufgerufen, ein Allow-Origin darauf würde nur
einer Webseite helfen, in den Index zu schreiben.

APPVIEW_AUTH_REQUIRED=false stellt das alte Verhalten her (VPN-Instanz,
fail-open-Tests) und warnt beim Start in Großbuchstaben.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX

crates/appview/Cargo.toml

@@ -35,6 +35,7 @@ at-shared = { workspace = true }
 at-firehose = { workspace = true }
 at-crypto = { workspace = true }
 at-identity = { workspace = true }
+reqwest = { workspace = true }
 uuid = { workspace = true }
 rustls = { version = "0.23", default-features = false, features = ["ring", "std", "logging", "tls12"] }
 base64 = { workspace = true }
@@ -45,3 +46,7 @@ tokio = { workspace = true }
 reqwest = { workspace = true }
 serde_json = { workspace = true }
 uuid = { workspace = true }
+# Unit tests in `src/auth.rs` build a P-256 verification key in the
+# same `0x8012 + uncompressed point` encoding the PDS publishes, which
+# needs the curve's `ToEncodedPoint`.
+p256 = { workspace = true }

crates/appview/src/auth.rs

@@ -0,0 +1,659 @@
+//! Bearer-token authentication for the AppView's private endpoints.
+//!
+//! ## What is being proven
+//!
+//! The PDS already issues an ES256 access JWT on
+//! `com.atproto.server.createSession` / `createAccount`. That token
+//! carries `sub = <did>`, `scope = "com.atproto.access"` and
+//! `aud = "did:web:appview.maarcadetweet.local"` — it was always meant
+//! to be presented *here*. All the AppView has to do is verify the
+//! signature and read `sub`.
+//!
+//! ## Where the key comes from
+//!
+//! Verifying an ES256 signature needs only the public half of the PDS's
+//! P-256 key. The PDS publishes it in its DID document at
+//! `GET /.well-known/did.json` (see `pds-server/src/main.rs`), so
+//! `PDS_JWT_SECRET` never leaves the PDS process. We fetch that
+//! document once, cache the `publicKeyMultibase` behind an `RwLock`,
+//! and re-fetch on a verification failure so an operator can rotate
+//! `PDS_JWT_SECRET` without restarting the AppView.
+//!
+//! Two failure modes are deliberately handled differently:
+//!
+//!  - **PDS unreachable at startup.** The AppView has always booted
+//!    independently of the PDS (it indexes the firehose, which has
+//!    nothing to do with the local PDS), and a crash-loop on a
+//!    colocated service that happens to boot second would be a
+//!    self-inflicted outage. So startup only *warns*; the first
+//!    authenticated request retries the fetch.
+//!  - **PDS unreachable when a token must be checked.** There is no
+//!    safe way to guess, so the request gets a `503`. Fail closed —
+//!    never fail open.
+//!
+//! ## Error contract
+//!
+//! The body shape matches every other AppView error
+//! (`{"error": …, "message": …}`). The `error` codes are load-bearing:
+//!
+//! | case                                   | status | `error`           |
+//! |----------------------------------------|--------|-------------------|
+//! | no / malformed `Authorization` header  | 401    | `AuthMissing`     |
+//! | bad signature, expired, wrong `scope`  | 401    | `TokenInvalid`    |
+//! | valid token, but `sub` ≠ requested did | 403    | `Forbidden`       |
+//! | PDS key not obtainable                 | 503    | `AuthUnavailable` |
+//!
+//! **`TokenInvalid` is a contract with the desktop client**: seeing it,
+//! the client refreshes its access JWT (`com.atproto.server.refreshSession`)
+//! and retries the request once. Renaming it silently logs every user
+//! out an hour after login.
+
+use at_crypto::jwt::JwtClaims;
+use axum::async_trait;
+use axum::extract::FromRequestParts;
+use axum::http::request::Parts;
+use axum::http::{HeaderMap, StatusCode};
+use axum::Json;
+use serde_json::{json, Value};
+use std::time::{Duration, Instant};
+use tokio::sync::RwLock;
+use tracing::{debug, warn};
+
+use crate::state::AppState;
+
+/// The scope an *access* token carries. Refresh tokens are minted with
+/// `com.atproto.refresh` by the same key, so without this check a
+/// refresh token — which lives for 90 days instead of an hour — would
+/// be accepted as a session credential everywhere.
+const ACCESS_SCOPE: &str = "com.atproto.access";
+
+/// Minimum spacing between two key re-fetches triggered by a failed
+/// verification.
+///
+/// Re-fetching on failure is what makes key rotation work without a
+/// restart. Doing it on *every* failure would also hand anyone who can
+/// reach the AppView a free amplifier: a stream of garbage tokens
+/// becomes a stream of requests to the PDS. One re-fetch per minute is
+/// far quicker than any plausible rotation cadence needs and costs the
+/// PDS nothing.
+const KEY_REFETCH_MIN_INTERVAL: Duration = Duration::from_secs(60);
+
+/// HTTP timeout for the DID-document fetch. The PDS is colocated; if it
+/// doesn't answer in two seconds it isn't answering, and a request
+/// blocked on auth is a request the user is staring at.
+const DID_DOC_TIMEOUT: Duration = Duration::from_secs(2);
+
+// -- error type --------------------------------------------------------------
+
+/// Everything that can go wrong between "a request arrived" and "we know
+/// which DID it belongs to". Converted into the AppView's standard error
+/// body by [`AuthError::into_response_parts`].
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum AuthError {
+    /// No `Authorization` header, or one that isn't `Bearer <token>`.
+    Missing,
+    /// Signature, expiry or scope check failed. The client's cue to
+    /// refresh its access token and retry.
+    Invalid(String),
+    /// Authenticated fine, but the token belongs to somebody else.
+    Forbidden,
+    /// We could not obtain the PDS's public key, so we cannot decide.
+    Unavailable(String),
+}
+
+impl AuthError {
+    pub fn into_response_parts(self) -> (StatusCode, Json<Value>) {
+        let (status, code, message) = match self {
+            AuthError::Missing => (
+                StatusCode::UNAUTHORIZED,
+                "AuthMissing",
+                "missing Authorization: Bearer header".to_string(),
+            ),
+            AuthError::Invalid(detail) => {
+                (StatusCode::UNAUTHORIZED, "TokenInvalid", detail)
+            }
+            AuthError::Forbidden => (
+                StatusCode::FORBIDDEN,
+                "Forbidden",
+                "token sub does not match the requested did".to_string(),
+            ),
+            AuthError::Unavailable(detail) => (
+                StatusCode::SERVICE_UNAVAILABLE,
+                "AuthUnavailable",
+                format!("cannot verify tokens: {detail}"),
+            ),
+        };
+        (
+            status,
+            Json(json!({
+                "error": code,
+                "message": message,
+            })),
+        )
+    }
+}
+
+impl From<AuthError> for (StatusCode, Json<Value>) {
+    fn from(e: AuthError) -> Self {
+        e.into_response_parts()
+    }
+}
+
+// -- key cache ---------------------------------------------------------------
+
+#[derive(Default)]
+struct CachedKey {
+    multibase: Option<String>,
+    /// When we last *attempted* a fetch — successful or not. Throttles
+    /// the on-failure refresh path; see [`KEY_REFETCH_MIN_INTERVAL`].
+    last_attempt: Option<Instant>,
+}
+
+/// The PDS signing key, lazily fetched and cached.
+///
+/// Lives in [`AppState`] behind an `Arc`, so all handlers share one
+/// cache and one HTTP client.
+pub struct PdsKeys {
+    http: reqwest::Client,
+    /// Fully-qualified URL of the PDS's DID document.
+    did_doc_url: String,
+    inner: RwLock<CachedKey>,
+}
+
+impl PdsKeys {
+    /// Build a cache pointed at `base_url` (no trailing slash required).
+    pub fn new(base_url: &str) -> Self {
+        let http = reqwest::Client::builder()
+            .timeout(DID_DOC_TIMEOUT)
+            .build()
+            .expect("reqwest client build should never fail");
+        Self {
+            http,
+            did_doc_url: format!(
+                "{}/.well-known/did.json",
+                base_url.trim_end_matches('/')
+            ),
+            inner: RwLock::new(CachedKey::default()),
+        }
+    }
+
+    /// Same PDS the handle-sync worker talks to: `PDS_INTERNAL_URL`
+    /// when set, else `PDS_PUBLIC_URL`.
+    pub fn from_config(cfg: &at_shared::config::AppConfig) -> Self {
+        Self::new(&cfg.pds_base_url())
+    }
+
+    pub fn did_doc_url(&self) -> &str {
+        &self.did_doc_url
+    }
+
+    /// The cached key, if we have ever fetched one.
+    pub async fn cached(&self) -> Option<String> {
+        self.inner.read().await.multibase.clone()
+    }
+
+    /// Fetch the DID document and replace the cached key.
+    ///
+    /// Called once at startup (best effort), on the first authenticated
+    /// request if startup failed, and — throttled — after a failed
+    /// verification.
+    pub async fn refresh(&self) -> anyhow::Result<String> {
+        // Record the attempt before the await so two concurrent
+        // failures can't both decide they're the first one.
+        self.inner.write().await.last_attempt = Some(Instant::now());
+
+        let resp = self.http.get(&self.did_doc_url).send().await?;
+        if !resp.status().is_success() {
+            anyhow::bail!(
+                "{} returned {}",
+                self.did_doc_url,
+                resp.status().as_u16()
+            );
+        }
+        let doc: Value = resp.json().await?;
+        let key = extract_public_key_multibase(&doc)?;
+        {
+            let mut guard = self.inner.write().await;
+            guard.multibase = Some(key.clone());
+        }
+        debug!(url = %self.did_doc_url, "loaded PDS signing key");
+        Ok(key)
+    }
+
+    /// The key to verify with, fetching it if we don't have one yet.
+    async fn key_or_fetch(&self) -> Result<String, AuthError> {
+        if let Some(k) = self.cached().await {
+            return Ok(k);
+        }
+        self.refresh().await.map_err(|e| {
+            warn!(url = %self.did_doc_url, error = %e, "PDS signing key unavailable");
+            AuthError::Unavailable(format!(
+                "PDS did document at {} not reachable: {e}",
+                self.did_doc_url
+            ))
+        })
+    }
+
+    /// Re-fetch after a verification failure, at most once per
+    /// [`KEY_REFETCH_MIN_INTERVAL`]. Returns the new key only when it
+    /// actually differs from `stale` — re-running the same failing
+    /// verification against an unchanged key proves nothing.
+    async fn refetch_if_stale(&self, stale: &str) -> Option<String> {
+        {
+            let guard = self.inner.read().await;
+            if let Some(last) = guard.last_attempt {
+                if last.elapsed() < KEY_REFETCH_MIN_INTERVAL {
+                    return None;
+                }
+            }
+        }
+        match self.refresh().await {
+            Ok(fresh) if fresh != stale => {
+                warn!("PDS signing key changed; re-verifying with the rotated key");
+                Some(fresh)
+            }
+            Ok(_) => None,
+            Err(e) => {
+                warn!(error = %e, "PDS key re-fetch after verification failure failed");
+                None
+            }
+        }
+    }
+
+    /// Verify an access token and return its claims.
+    ///
+    /// Retries exactly once against a freshly fetched key, so a rotated
+    /// `PDS_JWT_SECRET` costs one extra HTTP round trip rather than a
+    /// restart.
+    pub async fn verify_access_token(&self, token: &str) -> Result<JwtClaims, AuthError> {
+        let key = self.key_or_fetch().await?;
+        match verify_with_key(token, &key) {
+            Ok(claims) => Ok(claims),
+            Err(first) => {
+                let Some(fresh) = self.refetch_if_stale(&key).await else {
+                    return Err(first);
+                };
+                verify_with_key(token, &fresh).map_err(|_| first)
+            }
+        }
+    }
+}
+
+/// Pull `verificationMethod[].publicKeyMultibase` out of a DID document.
+///
+/// We take the first entry that carries a `publicKeyMultibase` rather
+/// than insisting on a fragment name: the PDS writes `#atproto`, but a
+/// document served by a proxy or a future PDS version may order or name
+/// its methods differently, and any key in the document is a key the
+/// controller published for itself. A document with none is an error,
+/// not an empty key — silently caching `""` would turn every later
+/// verification into a confusing signature failure.
+fn extract_public_key_multibase(doc: &Value) -> anyhow::Result<String> {
+    let methods = doc
+        .get("verificationMethod")
+        .and_then(|v| v.as_array())
+        .ok_or_else(|| anyhow::anyhow!("did document has no verificationMethod array"))?;
+    for m in methods {
+        if let Some(k) = m.get("publicKeyMultibase").and_then(|v| v.as_str()) {
+            if !k.is_empty() {
+                return Ok(k.to_string());
+            }
+        }
+    }
+    anyhow::bail!("did document has no verificationMethod with publicKeyMultibase")
+}
+
+/// Signature + expiry + scope check against one specific key.
+///
+/// `verify_jwt` handles the ES256 signature and `exp` (with a 30 s
+/// leeway for clock skew); the scope check is ours, and it is the line
+/// that keeps a 90-day refresh token from working as a session
+/// credential.
+fn verify_with_key(token: &str, pubkey_multibase: &str) -> Result<JwtClaims, AuthError> {
+    let claims = at_crypto::jwt::verify_jwt(token, pubkey_multibase)
+        .map_err(|e| AuthError::Invalid(format!("invalid token: {e}")))?;
+    match claims.scope.as_deref() {
+        Some(ACCESS_SCOPE) => Ok(claims),
+        other => Err(AuthError::Invalid(format!(
+            "token scope {:?} is not {ACCESS_SCOPE}",
+            other.unwrap_or("<none>")
+        ))),
+    }
+}
+
+/// Extract the bearer token from an `Authorization` header.
+///
+/// The scheme match is case-insensitive (RFC 7235 says it is) — some
+/// HTTP clients send `bearer`.
+fn bearer_token(headers: &HeaderMap) -> Result<String, AuthError> {
+    let raw = headers
+        .get("authorization")
+        .and_then(|v| v.to_str().ok())
+        .ok_or(AuthError::Missing)?;
+    let (scheme, token) = raw.split_once(' ').ok_or(AuthError::Missing)?;
+    if !scheme.eq_ignore_ascii_case("bearer") {
+        return Err(AuthError::Missing);
+    }
+    let token = token.trim();
+    if token.is_empty() {
+        return Err(AuthError::Missing);
+    }
+    Ok(token.to_string())
+}
+
+// -- extractor ---------------------------------------------------------------
+
+/// The authenticated DID of the caller.
+///
+/// `None` means auth is switched off for this instance
+/// (`APPVIEW_AUTH_REQUIRED=false`) — *not* "anonymous but allowed
+/// through". When auth is on, this is always `Some` by construction:
+/// the extractor rejects the request otherwise.
+///
+/// Handlers must still call [`AuthedDid::ensure_matches`] with the DID
+/// the request asks about. Proving *who you are* is not the same as
+/// proving *whose notifications you may read*.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AuthedDid(pub Option<String>);
+
+impl AuthedDid {
+    /// 403 unless the token's `sub` is the DID the request targets.
+    ///
+    /// With auth disabled this is a no-op, which is exactly what
+    /// `APPVIEW_AUTH_REQUIRED=false` means.
+    pub fn ensure_matches(&self, did: &str) -> Result<(), (StatusCode, Json<Value>)> {
+        match self.0.as_deref() {
+            None => Ok(()),
+            Some(sub) if sub == did => Ok(()),
+            Some(_) => Err(AuthError::Forbidden.into_response_parts()),
+        }
+    }
+
+    /// The caller's DID, if authenticated.
+    ///
+    /// No handler needs it yet — every private endpoint takes the DID
+    /// as a parameter and compares it — but an endpoint that acts
+    /// purely on "whoever is calling" would read it here instead of
+    /// trusting a query parameter.
+    #[allow(dead_code)]
+    pub fn did(&self) -> Option<&str> {
+        self.0.as_deref()
+    }
+}
+
+#[async_trait]
+impl FromRequestParts<AppState> for AuthedDid {
+    type Rejection = (StatusCode, Json<Value>);
+
+    async fn from_request_parts(
+        parts: &mut Parts,
+        state: &AppState,
+    ) -> Result<Self, Self::Rejection> {
+        if !state.cfg.appview_auth_required {
+            // Opt-out instance. The startup log says so in capitals.
+            return Ok(AuthedDid(None));
+        }
+        let token = bearer_token(&parts.headers).map_err(AuthError::into_response_parts)?;
+        let claims = state
+            .pds_keys
+            .verify_access_token(&token)
+            .await
+            .map_err(AuthError::into_response_parts)?;
+        Ok(AuthedDid(Some(claims.sub)))
+    }
+}
+
+// -- startup posture ---------------------------------------------------------
+
+/// Log, once at boot, every way this instance is configured to be less
+/// strict than the defaults.
+///
+/// All three of these are legitimate configurations — a VPN-only
+/// deployment, a dev box, a single-machine setup — and all three are
+/// also exactly what an accidentally-public instance looks like. The
+/// only defence that survives a hurried deployment is a log line the
+/// operator cannot miss, so each one is a `warn!` naming the variable
+/// that turns it back on.
+pub fn log_startup_posture(cfg: &at_shared::config::AppConfig) {
+    if !cfg.appview_auth_required {
+        warn!(
+            "APPVIEW_AUTH_REQUIRED=false — /api/notifications* and \
+             /api/timeline/home are served to ANY caller for ANY did. \
+             Only safe when this instance is unreachable from untrusted \
+             networks."
+        );
+    }
+    if cfg.appview_cors_origins.is_empty() {
+        warn!(
+            "APPVIEW_CORS_ORIGINS unset — sending Access-Control-Allow-Origin: * \
+             so any web page can call this AppView from a browser. Set it to the \
+             origins your client actually uses, e.g. \
+             'tauri://localhost,http://127.0.0.1:1430'."
+        );
+    }
+    if cfg.appview_ingest_secret.is_none() {
+        warn!(
+            "APPVIEW_INGEST_SECRET unset — POST /internal/ingest-commit accepts \
+             unauthenticated writes into the index (anyone who can reach this port \
+             can forge posts, follows and notifications). Set the same value here \
+             and on the PDS."
+        );
+    }
+}
+
+// -- tests -------------------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use at_crypto::ecdsa::P256Keypair;
+    use at_crypto::jwt::{issue_jwt, JwtClaims};
+    use axum::http::HeaderValue;
+
+    /// A throwaway server key plus its `publicKeyMultibase`, in the
+    /// exact encoding the PDS publishes: `0x8012` (P-256) followed by
+    /// the *uncompressed* affine coordinates. `P256Keypair::generate`
+    /// stores a compressed point, which `verify_jwt` cannot decode — so
+    /// this mirrors `pds-server`'s `server_p256_keypair`.
+    fn test_key() -> (P256Keypair, String) {
+        let kp = P256Keypair::generate().unwrap();
+        let vk = kp.verifying_key().unwrap();
+        let pt = vk.to_encoded_point(false);
+        let mut raw = vec![0x80u8, 0x12u8];
+        raw.extend_from_slice(pt.x().unwrap());
+        raw.extend_from_slice(pt.y().unwrap());
+        let multibase = at_crypto::multibase_util::encode_b58btc(&raw);
+        (kp, multibase)
+    }
+
+    fn mint(kp: &P256Keypair, did: &str, scope: &str, ttl_secs: i64) -> String {
+        let now = chrono::Utc::now().timestamp();
+        issue_jwt(
+            kp,
+            &JwtClaims {
+                iss: "did:web:127.0.0.1%3A2583".into(),
+                sub: did.into(),
+                aud: "did:web:appview.maarcadetweet.local".into(),
+                iat: now - 1,
+                exp: now + ttl_secs,
+                jti: None,
+                scope: Some(scope.into()),
+            },
+        )
+        .unwrap()
+    }
+
+    fn header_map(value: &str) -> HeaderMap {
+        let mut h = HeaderMap::new();
+        h.insert("authorization", HeaderValue::from_str(value).unwrap());
+        h
+    }
+
+    #[test]
+    fn missing_header_is_auth_missing() {
+        assert_eq!(bearer_token(&HeaderMap::new()), Err(AuthError::Missing));
+    }
+
+    #[test]
+    fn malformed_header_is_auth_missing() {
+        // No scheme, wrong scheme, and an empty token all read as
+        // "the client never presented a credential" — 401 AuthMissing,
+        // not TokenInvalid, so the client re-authenticates instead of
+        // burning a refresh round trip.
+        assert_eq!(bearer_token(&header_map("abc.def.ghi")), Err(AuthError::Missing));
+        assert_eq!(bearer_token(&header_map("Basic dXNlcjpwdw==")), Err(AuthError::Missing));
+        assert_eq!(bearer_token(&header_map("Bearer   ")), Err(AuthError::Missing));
+    }
+
+    #[test]
+    fn bearer_scheme_is_case_insensitive() {
+        assert_eq!(bearer_token(&header_map("bearer tok")).unwrap(), "tok");
+        assert_eq!(bearer_token(&header_map("Bearer tok")).unwrap(), "tok");
+    }
+
+    #[test]
+    fn error_bodies_carry_the_documented_codes() {
+        // These strings are a contract: the desktop client keys its
+        // "refresh and retry" behaviour off `TokenInvalid`.
+        let (s, b) = AuthError::Missing.into_response_parts();
+        assert_eq!(s, StatusCode::UNAUTHORIZED);
+        assert_eq!(b.0["error"], "AuthMissing");
+        let (s, b) = AuthError::Invalid("x".into()).into_response_parts();
+        assert_eq!(s, StatusCode::UNAUTHORIZED);
+        assert_eq!(b.0["error"], "TokenInvalid");
+        let (s, b) = AuthError::Forbidden.into_response_parts();
+        assert_eq!(s, StatusCode::FORBIDDEN);
+        assert_eq!(b.0["error"], "Forbidden");
+        let (s, b) = AuthError::Unavailable("pds down".into()).into_response_parts();
+        assert_eq!(s, StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(b.0["error"], "AuthUnavailable");
+        // Every body carries both fields the client parses.
+        assert!(b.0["message"].as_str().unwrap().contains("pds down"));
+    }
+
+    #[test]
+    fn valid_access_token_verifies() {
+        let (kp, mb) = test_key();
+        let token = mint(&kp, "did:plc:alice", ACCESS_SCOPE, 3600);
+        let claims = verify_with_key(&token, &mb).unwrap();
+        assert_eq!(claims.sub, "did:plc:alice");
+    }
+
+    #[test]
+    fn garbage_token_is_invalid() {
+        let (_, mb) = test_key();
+        let err = verify_with_key("not-a-jwt", &mb).unwrap_err();
+        assert!(matches!(err, AuthError::Invalid(_)));
+    }
+
+    #[test]
+    fn token_signed_by_another_key_is_invalid() {
+        let (kp, _) = test_key();
+        let (_, other_mb) = test_key();
+        let token = mint(&kp, "did:plc:alice", ACCESS_SCOPE, 3600);
+        assert!(matches!(
+            verify_with_key(&token, &other_mb).unwrap_err(),
+            AuthError::Invalid(_)
+        ));
+    }
+
+    #[test]
+    fn refresh_scope_is_rejected() {
+        // The refresh token is signed by the same key and lives 90
+        // days. Without the scope check it would be a session token.
+        let (kp, mb) = test_key();
+        let token = mint(&kp, "did:plc:alice", "com.atproto.refresh", 3600);
+        let err = verify_with_key(&token, &mb).unwrap_err();
+        match err {
+            AuthError::Invalid(msg) => assert!(msg.contains("com.atproto.refresh")),
+            other => panic!("expected Invalid, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn expired_token_is_rejected() {
+        // Beyond `verify_jwt`'s 30 s leeway.
+        let (kp, mb) = test_key();
+        let token = mint(&kp, "did:plc:alice", ACCESS_SCOPE, -120);
+        assert!(matches!(
+            verify_with_key(&token, &mb).unwrap_err(),
+            AuthError::Invalid(_)
+        ));
+    }
+
+    #[test]
+    fn ensure_matches_enforces_sub_equals_did() {
+        let me = AuthedDid(Some("did:plc:alice".into()));
+        assert!(me.ensure_matches("did:plc:alice").is_ok());
+        let (status, body) = me.ensure_matches("did:plc:bob").unwrap_err();
+        assert_eq!(status, StatusCode::FORBIDDEN);
+        assert_eq!(body.0["error"], "Forbidden");
+        // A prefix of the real DID must not pass.
+        assert!(me.ensure_matches("did:plc:ali").is_err());
+    }
+
+    #[test]
+    fn ensure_matches_is_a_noop_when_auth_disabled() {
+        let off = AuthedDid(None);
+        assert!(off.ensure_matches("did:plc:anyone").is_ok());
+        assert_eq!(off.did(), None);
+    }
+
+    #[test]
+    fn did_doc_key_extraction() {
+        let doc = json!({
+            "id": "did:web:127.0.0.1%3A2583",
+            "verificationMethod": [{
+                "id": "did:web:127.0.0.1%3A2583#atproto",
+                "type": "Multikey",
+                "controller": "did:web:127.0.0.1%3A2583",
+                "publicKeyMultibase": "zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme"
+            }]
+        });
+        assert_eq!(
+            extract_public_key_multibase(&doc).unwrap(),
+            "zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme"
+        );
+
+        // A document without a usable key must error rather than cache
+        // an empty string.
+        assert!(extract_public_key_multibase(&json!({})).is_err());
+        assert!(extract_public_key_multibase(&json!({
+            "verificationMethod": [{"id": "#x", "type": "Multikey"}]
+        }))
+        .is_err());
+        assert!(extract_public_key_multibase(&json!({
+            "verificationMethod": [{"publicKeyMultibase": ""}]
+        }))
+        .is_err());
+    }
+
+    #[test]
+    fn did_doc_url_is_built_from_the_base_url() {
+        assert_eq!(
+            PdsKeys::new("http://127.0.0.1:2583").did_doc_url(),
+            "http://127.0.0.1:2583/.well-known/did.json"
+        );
+        // A trailing slash must not produce a double slash — some
+        // servers 404 on it.
+        assert_eq!(
+            PdsKeys::new("http://pds:3000/").did_doc_url(),
+            "http://pds:3000/.well-known/did.json"
+        );
+    }
+
+    #[tokio::test]
+    async fn verification_fails_closed_when_the_pds_is_unreachable() {
+        // Port 1 on loopback: nothing listens there, so the fetch fails
+        // fast. The result must be a 503, never a pass-through.
+        let keys = PdsKeys::new("http://127.0.0.1:1");
+        let err = keys.verify_access_token("whatever").await.unwrap_err();
+        assert!(
+            matches!(err, AuthError::Unavailable(_)),
+            "expected Unavailable, got {err:?}"
+        );
+        let (status, body) = err.into_response_parts();
+        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(body.0["error"], "AuthUnavailable");
+    }
+}

crates/appview/src/ingest.rs

@@ -15,8 +15,31 @@
 //! }
 //! ```
 //!
-//! In production this endpoint would be protected with mTLS and a token
-//! minted by the PDS; for now it's open inside the cluster.
+//! ## Who may call this
+//!
+//! This is the AppView's only write path, and it is not a browser
+//! endpoint: it is excluded from the CORS layer in
+//! [`crate::routes::router`], because an `Access-Control-Allow-Origin`
+//! header here would only ever help a web page forge index entries.
+//!
+//! Authentication is the shared secret `APPVIEW_INGEST_SECRET`,
+//! compared in constant time against the caller's `X-Ingest-Secret`
+//! header:
+//!
+//! - **set** → enforced. A missing or wrong header is `401
+//!   AuthenticationRequired`.
+//! - **unset** → anonymous writes are accepted, and the AppView shouts
+//!   about it once at startup (see
+//!   [`crate::auth::log_startup_posture`]). Refusing to start would
+//!   break every existing single-machine dev setup for a service that,
+//!   in that configuration, is bound to loopback anyway; accepting
+//!   silently is how an internet-facing deployment ends up letting
+//!   anyone forge posts, follows and notifications. So: keep working,
+//!   but never quietly.
+//!
+//! A future hardening step is mTLS or a PDS-minted token, at which
+//! point the shared secret becomes the fallback rather than the only
+//! line.
 
 use crate::indexer;
 use crate::state::AppState;
@@ -51,8 +74,13 @@ pub struct IngestCommitReq {
 }
 
 /// Authenticate internal ingest requests.
-/// - If `APPVIEW_INGEST_SECRET` env var is unset: dev mode, accept anything.
-/// - If set: require `X-Ingest-Secret: <value>` header to match.
+///
+/// - `APPVIEW_INGEST_SECRET` unset: accept anonymous writes (dev mode —
+///   the startup log warns, see the module docs for why this isn't a
+///   hard failure).
+/// - Set: require a matching `X-Ingest-Secret` header. The comparison
+///   is constant-time so a caller can't recover the secret byte by byte
+///   from response timings.
 pub fn check_ingest_secret(
     headers: &HeaderMap,
     configured: Option<&str>,

crates/appview/src/lib.rs

@@ -3,6 +3,7 @@
 //! tests under `tests/` import from here so they can build a worker
 //! against a stub resolver without booting the binary.
 
+pub mod auth;
 pub mod firehose;
 pub mod handle_sync;
 pub mod indexer;

crates/appview/src/main.rs

@@ -7,6 +7,7 @@ use tokio::sync::mpsc;
 use tracing::info;
 use tracing_subscriber::EnvFilter;
 
+mod auth;
 mod firehose;
 mod handle_sync;
 mod indexer;
@@ -83,6 +84,33 @@ async fn main() -> Result<()> {
 
     let state = AppState::new(cfg.clone(), db.clone(), stats.clone());
 
+    // Announce every relaxed security switch before we serve anything.
+    auth::log_startup_posture(&cfg);
+
+    // Pre-load the PDS's published signing key so the first
+    // authenticated request doesn't pay for the round trip.
+    //
+    // Best effort on purpose: the AppView has always started
+    // independently of the PDS, and in a compose file the two race. A
+    // hard failure here would turn "the PDS booted two seconds later"
+    // into "the AppView is in a crash loop". If the fetch fails, the
+    // first authenticated request retries it — and answers `503
+    // AuthUnavailable` if the PDS is still unreachable. Never open.
+    if cfg.appview_auth_required {
+        match state.pds_keys.refresh().await {
+            Ok(_) => info!(
+                url = %state.pds_keys.did_doc_url(),
+                "loaded PDS signing key for token verification"
+            ),
+            Err(e) => tracing::warn!(
+                url = %state.pds_keys.did_doc_url(),
+                error = %e,
+                "could not load the PDS signing key at startup; will retry on the \
+                 first authenticated request (which fails with 503 until it works)"
+            ),
+        }
+    }
+
     // Back-fill the `handle` column on posts that the Jetstream
     // indexer inserted with an empty placeholder. The worker dispatches
     // by DID method: `did:plc:` → PLC directory, `did:web:` → a
@@ -102,12 +130,12 @@ async fn main() -> Result<()> {
     // inside docker compose) — `pds_public_url` may not be reachable
     // from inside the cluster when TLS / DNS is set up for outside
     // clients only.
-    let pds_base_url = cfg
-        .pds_internal_url
-        .clone()
-        .unwrap_or_else(|| cfg.pds_public_url.clone());
+    //
+    // `AppConfig::pds_base_url()` owns that fallback so the handle
+    // resolver and the signing-key fetch in `auth.rs` can never end up
+    // pointed at different PDS instances.
     let pds_resolver: Arc<dyn at_identity::DidHandleResolver> = Arc::new(
-        at_identity::pds_handle::PdsHandleResolver::new(pds_base_url),
+        at_identity::pds_handle::PdsHandleResolver::new(cfg.pds_base_url()),
     );
 
     let handle_sync = handle_sync::HandleSyncWorker {

crates/appview/src/routes.rs

@@ -8,6 +8,30 @@
 //! - `ingest_commit`: the internal-only writer used by the PDS, owned
 //!   in `crate::ingest`.
 //!
+//! ## Public vs. private
+//!
+//! Most of what the AppView serves is public by construction: in AT
+//! Proto a post, a profile, a follow edge and a like are records in a
+//! public repo, replicated over the firehose. Those endpoints
+//! (`/api/profile*`, `/api/search`, `/api/post/*`, `/api/thread*`,
+//! `/api/followers`, `/api/following`) need no credential.
+//!
+//! Two things are *not* public, and they are the reason this service
+//! has authentication at all:
+//!
+//!  - `/api/notifications*` — who interacted with you, and the read
+//!    state of that list. Nothing in the protocol makes it readable by
+//!    anyone but the recipient.
+//!  - `/api/timeline/home` — the timeline is assembled from the
+//!    viewer's follow graph, so serving it to an arbitrary `did`
+//!    parameter answers "what does this person's feed look like" for
+//!    any DID a caller cares to type.
+//!
+//! Both require a valid PDS-issued access token whose `sub` equals the
+//! `did` in the request ([`crate::auth`]). The check is two steps on
+//! purpose: the extractor proves *who* the caller is, and
+//! `ensure_matches` proves they are asking about themselves.
+//!
 //! The cursor format used by `timeline_home` is opaque: it's a
 //! `base64url(micros):uri` pair, which is what [`cursor::encode`] and
 //! [`cursor::decode`] produce/consume.
@@ -19,11 +43,14 @@ use axum::{
     routing::{get, post},
     Json, Router,
 };
+use axum::http::{header, HeaderValue, Method};
 use chrono::{DateTime, TimeZone, Utc};
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
-use tower_http::cors::{Any, CorsLayer};
+use tower_http::cors::{AllowOrigin, Any, CorsLayer};
+use tracing::warn;
 
+use crate::auth::AuthedDid;
 use crate::state::AppState;
 
 pub mod cursor;
@@ -35,25 +62,73 @@ use types::{
     ProfileResponse, SearchResponse, ThreadFullResponse, TimelineResponse,
 };
 
-pub fn router(state: AppState) -> Router {
-    // CORS: the Tauri webview's origin is the Vite dev server
-    // (`http://127.0.0.1:1430`) in dev or the bundled `tauri://` /
-    // `asset://` origin in production. Either way it's a cross-origin
-    // fetch against this service's `http://127.0.0.1:2584` listen
-    // address, so the browser blocks the response without an explicit
-    // allow-origin header. We allow any origin — the AppView's
-    // public read endpoints (`/api/...`) carry no auth cookie and
-    // the AppView runs alongside the user's own PDS, not on the
-    // open internet; production deployments behind a reverse proxy
-    // can tighten this via the proxy itself.
-    let cors = CorsLayer::new()
-        .allow_origin(Any)
-        .allow_methods(Any)
-        .allow_headers(Any);
+/// Build the CORS layer for the browser-facing routes.
+///
+/// The Tauri webview is a cross-origin caller: in dev its origin is the
+/// Vite server (`http://127.0.0.1:1430`), in a packaged build it is a
+/// platform-specific scheme — `tauri://localhost` on macOS/iOS,
+/// `http://tauri.localhost` on Windows. Either way the browser drops
+/// the response unless we send `Access-Control-Allow-Origin`.
+///
+/// `APPVIEW_CORS_ORIGINS` is a comma-separated allowlist, e.g.
+/// `tauri://localhost,http://127.0.0.1:1430`. When it is unset we keep
+/// the historic wildcard so no existing deployment breaks on upgrade —
+/// [`crate::auth::log_startup_posture`] warns about that at startup.
+///
+/// `Authorization` has to be in `allow_headers`: it is not a
+/// CORS-safelisted header, so without it the browser's preflight fails
+/// and the authenticated endpoints become unreachable from the webview
+/// — with an error that looks nothing like an auth problem.
+///
+/// `allow_credentials` stays off. We authenticate with a bearer token
+/// the client attaches deliberately, never with an ambient cookie, so
+/// there is nothing for a hostile page to replay — and turning it on
+/// would additionally make the wildcard origin illegal.
+fn cors_layer(cfg: &at_shared::config::AppConfig) -> CorsLayer {
+    let base = CorsLayer::new()
+        .allow_methods([Method::GET, Method::POST, Method::OPTIONS])
+        .allow_headers([
+            header::AUTHORIZATION,
+            header::CONTENT_TYPE,
+            header::ACCEPT,
+        ]);
 
-    Router::new()
+    if cfg.appview_cors_origins.is_empty() {
+        return base.allow_origin(Any);
+    }
+
+    // Anything unparseable as a header value is dropped with a warning
+    // rather than panicking the process — a stray quote in the env var
+    // must not take the service down. If *every* entry is bad we fall
+    // back to the wildcard and say so, because an empty allowlist would
+    // silently break every browser client.
+    let mut origins: Vec<HeaderValue> = Vec::new();
+    for raw in &cfg.appview_cors_origins {
+        match HeaderValue::from_str(raw) {
+            Ok(v) => origins.push(v),
+            Err(_) => warn!(origin = %raw, "APPVIEW_CORS_ORIGINS: ignoring unparseable origin"),
+        }
+    }
+    if origins.is_empty() {
+        warn!("APPVIEW_CORS_ORIGINS contained no usable origin; falling back to allow-any");
+        return base.allow_origin(Any);
+    }
+    base.allow_origin(AllowOrigin::list(origins))
+}
+
+pub fn router(state: AppState) -> Router {
+    let cors = cors_layer(&state.cfg);
+
+    // Browser-facing surface. Everything here answers a `fetch()` from
+    // the desktop client, so it carries the CORS layer.
+    let api = Router::new()
         .route("/", get(root))
+        // --- private: token required, `sub` must equal `did` ---
         .route("/api/timeline/home", get(timeline_home))
+        .route("/api/notifications", get(notifications))
+        .route("/api/notifications/count", get(notifications_count))
+        .route("/api/notifications/seen", post(notifications_seen))
+        // --- public: AT Proto public records ---
         .route("/api/profile", get(profile_query))
         .route("/api/profile/:handle", get(profile_path))
         .route("/api/search", get(search))
@@ -66,15 +141,20 @@ pub fn router(state: AppState) -> Router {
         // implementation, so they can't drift.
         .route("/api/thread", get(thread_query))
         .route("/api/thread/*uri", get(thread_path))
-        .route("/api/notifications", get(notifications))
-        .route("/api/notifications/count", get(notifications_count))
-        .route("/api/notifications/seen", post(notifications_seen))
         .route("/api/followers", get(followers))
         .route("/api/following", get(following))
         .route("/healthz", get(healthz))
-        .route("/internal/ingest-commit", post(crate::ingest::ingest_commit))
-        .layer(cors)
-        .with_state(state)
+        .layer(cors);
+
+    // Server-to-server surface. `/internal/ingest-commit` is called by
+    // the PDS with a shared secret, never by a browser, so it stays
+    // outside the CORS layer: handing it an
+    // `Access-Control-Allow-Origin` header would only ever help a web
+    // page try to write to the index.
+    let internal = Router::new()
+        .route("/internal/ingest-commit", post(crate::ingest::ingest_commit));
+
+    api.merge(internal).with_state(state)
 }
 
 async fn root() -> Json<Value> {
@@ -115,13 +195,21 @@ const MAX_LIMIT: i64 = 100;
 /// surface.
 const MAX_FOLLOWED_DIDS: usize = 1000;
 
+/// `GET /api/timeline/home?did=…`
+///
+/// **Authenticated.** The timeline is derived from the viewer's follow
+/// graph, so `did` must be the caller's own DID — otherwise this
+/// endpoint would answer "what does this account's feed look like" for
+/// any DID at all.
 async fn timeline_home(
     State(state): State<AppState>,
+    auth: AuthedDid,
     Query(q): Query<TimelineQuery>,
 ) -> Result<Json<TimelineResponse>, (StatusCode, Json<Value>)> {
     if q.did.is_empty() {
         return Err(bad_request("did is required"));
     }
+    auth.ensure_matches(&q.did)?;
     let limit = clamp_limit(q.limit);
 
     // Look up the set of DIDs this user follows, then build the
@@ -1046,13 +1134,19 @@ struct NotificationsQuery {
 /// of the list. The tiebreak here is the row's `id` rather than a URI
 /// (a notification has no URI of its own), which the shared
 /// [`cursor`] codec carries in its string slot.
+///
+/// **Authenticated**: `did` is the recipient, so the caller has to be
+/// that recipient. This is the endpoint that made authentication
+/// necessary in the first place.
 async fn notifications(
     State(state): State<AppState>,
+    auth: AuthedDid,
     Query(q): Query<NotificationsQuery>,
 ) -> Result<Json<NotificationsResponse>, (StatusCode, Json<Value>)> {
     if q.did.is_empty() {
         return Err(bad_request("did is required"));
     }
+    auth.ensure_matches(&q.did)?;
     let limit = clamp_limit(q.limit);
     let keyset = decode_cursor(q.cursor.as_deref())?;
     // The cursor's string slot holds the row id. A client that hands
@@ -1121,13 +1215,18 @@ struct NotificationCountQuery {
 /// scales with the number of *unread* rows, not the user's lifetime
 /// notification history. That matters because the client polls this
 /// for its tray badge.
+///
+/// **Authenticated**, same rule as the list itself: an unread count is
+/// still information about someone else's inbox.
 async fn notifications_count(
     State(state): State<AppState>,
+    auth: AuthedDid,
     Query(q): Query<NotificationCountQuery>,
 ) -> Result<Json<NotificationCountResponse>, (StatusCode, Json<Value>)> {
     if q.did.is_empty() {
         return Err(bad_request("did is required"));
     }
+    auth.ensure_matches(&q.did)?;
     let count: i64 = sqlx::query_scalar(
         "SELECT COUNT(*)::BIGINT FROM notifications \
           WHERE recipient_did = $1 AND read_at IS NULL",
@@ -1165,13 +1264,20 @@ struct NotificationsSeenReq {
 /// updates nothing and reports `updated: 0`. `read_at` is set to
 /// `now()` (when we recorded the ack), not to `seenAt` (which is a
 /// client-supplied watermark and could be arbitrarily far in the past).
+///
+/// **Authenticated**, and the only *write* among the private
+/// endpoints: without the check anyone could clear another user's
+/// unread badge. `AuthedDid` runs before `Json` because the body
+/// extractor consumes the request — axum requires body extractors last.
 async fn notifications_seen(
     State(state): State<AppState>,
+    auth: AuthedDid,
     Json(req): Json<NotificationsSeenReq>,
 ) -> Result<Json<NotificationSeenResponse>, (StatusCode, Json<Value>)> {
     if req.did.is_empty() {
         return Err(bad_request("did is required"));
     }
+    auth.ensure_matches(&req.did)?;
     let res = sqlx::query(
         r#"UPDATE notifications
               SET read_at = now()

crates/appview/src/state.rs

@@ -2,18 +2,30 @@ use at_shared::config::AppConfig;
 use sqlx::PgPool;
 use std::sync::Arc;
 
+use crate::auth::PdsKeys;
 use crate::firehose::Stats;
 
 #[derive(Clone)]
 pub struct AppState {
-    #[allow(dead_code)]
     pub cfg: AppConfig,
     pub db: PgPool,
     pub stats: Arc<Stats>,
+    /// Cache of the PDS's published signing key, used by the
+    /// [`crate::auth::AuthedDid`] extractor. Shared (`Arc`) so every
+    /// handler verifies against the same cached key and one HTTP
+    /// client, and so a key rotation picked up by one request is
+    /// immediately visible to the rest.
+    pub pds_keys: Arc<PdsKeys>,
 }
 
 impl AppState {
     pub fn new(cfg: AppConfig, db: PgPool, stats: Arc<Stats>) -> Self {
-        Self { cfg, db, stats }
+        let pds_keys = Arc::new(PdsKeys::from_config(&cfg));
+        Self {
+            cfg,
+            db,
+            stats,
+            pds_keys,
+        }
     }
 }

crates/appview/tests/api_integration.rs

@@ -7,6 +7,9 @@
 //! rather than panicking — so `cargo test --workspace` stays green in
 //! environments where the appview hasn't been started.
 
+mod common;
+
+use common::TestAuth;
 use serde_json::{json, Value};
 use std::time::Duration;
 
@@ -46,6 +49,36 @@ async fn db_reachable() -> bool {
     )
 }
 
+/// How this suite authenticates against `/api/timeline/home`, which is
+/// no longer public. `None` means the AppView enforces auth but the
+/// test process cannot mint a token (no `PDS_JWT_SECRET`), in which
+/// case the test skips like it does for a missing service.
+async fn auth_or_skip() -> Option<TestAuth> {
+    TestAuth::probe(&client().await, APPVIEW_URL).await
+}
+
+/// `GET /api/timeline/home` as `did`, with the bearer token attached
+/// when the instance requires one. The seeded DIDs are synthetic, so
+/// the token is minted from the PDS's own signing secret — see
+/// `tests/common/mod.rs`.
+async fn get_timeline(
+    c: &reqwest::Client,
+    auth: &TestAuth,
+    did: &str,
+    extra: &[(&str, &str)],
+) -> reqwest::Response {
+    let mut params: Vec<(&str, &str)> = vec![("did", did)];
+    params.extend_from_slice(extra);
+    auth.apply(
+        c.get(format!("{APPVIEW_URL}/api/timeline/home"))
+            .query(&params),
+        did,
+    )
+    .send()
+    .await
+    .unwrap()
+}
+
 async fn post_ingest(c: &reqwest::Client, body: Value) -> reqwest::Response {
     c.post(format!("{APPVIEW_URL}/internal/ingest-commit"))
         .json(&body)
@@ -120,6 +153,7 @@ async fn timeline_returns_seeded_posts() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let did = did_for_test("tl");
 
     // Seed 3 posts with distinct rkeys.
@@ -157,12 +191,7 @@ async fn timeline_returns_seeded_posts() {
     // machine that has run this suite twice) the three rows we just
     // seeded fall outside a 10-row window and the assertions below
     // fail for reasons that have nothing to do with the timeline.
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", did.as_str()), ("limit", "100")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "100")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().expect("posts is array");
@@ -222,6 +251,7 @@ async fn timeline_paginates_with_cursor() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let did = did_for_test("pg");
 
     // Seed 50 posts.
@@ -246,28 +276,14 @@ async fn timeline_paginates_with_cursor() {
     tokio::time::sleep(Duration::from_millis(100)).await;
 
     // Page 1: limit=20.
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", did.as_str()), ("limit", "20")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "20")]).await;
     let body: Value = resp.json().await.unwrap();
     let page1 = body["posts"].as_array().unwrap().clone();
     let cursor1 = body["cursor"].as_str().expect("page1 cursor");
     assert_eq!(page1.len(), 20, "page1 should be exactly 20");
 
     // Page 2: with cursor.
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[
-            ("did", did.as_str()),
-            ("limit", "20"),
-            ("cursor", cursor1),
-        ])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "20"), ("cursor", cursor1)]).await;
     let body: Value = resp.json().await.unwrap();
     let page2 = body["posts"].as_array().unwrap().clone();
     assert_eq!(page2.len(), 20, "page2 should be exactly 20");
@@ -285,16 +301,7 @@ async fn timeline_paginates_with_cursor() {
 
     // Page 3: tail — fewer than 20 expected, cursor=null.
     let cursor2 = body["cursor"].as_str().expect("page2 cursor");
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[
-            ("did", did.as_str()),
-            ("limit", "20"),
-            ("cursor", cursor2),
-        ])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "20"), ("cursor", cursor2)]).await;
     let body: Value = resp.json().await.unwrap();
     let page3 = body["posts"].as_array().unwrap().clone();
     assert!(page3.len() <= 20, "page3 should be <= 20");
@@ -484,6 +491,7 @@ async fn timeline_filters_to_followees() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let url = std::env::var("DATABASE_URL_APPVIEW").unwrap();
     let pool = sqlx::PgPool::connect(&url).await.unwrap();
 
@@ -503,12 +511,7 @@ async fn timeline_filters_to_followees() {
 
     tokio::time::sleep(Duration::from_millis(100)).await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", alice.as_str()), ("limit", "100")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &alice, &[("limit", "100")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().expect("posts is array");
@@ -555,18 +558,14 @@ async fn timeline_includes_own_posts() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let alice = did_for_test("alone");
 
     // Alice posts without seeding any follows.
     seed_posts(&c, &alice, &["alice's first post", "alice's second post"]).await;
     tokio::time::sleep(Duration::from_millis(100)).await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", alice.as_str()), ("limit", "100")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &alice, &[("limit", "100")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().expect("posts is array");
@@ -605,6 +604,7 @@ async fn timeline_caps_followee_list() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let url = std::env::var("DATABASE_URL_APPVIEW").unwrap();
     let pool = sqlx::PgPool::connect(&url).await.unwrap();
 
@@ -624,12 +624,7 @@ async fn timeline_caps_followee_list() {
     seed_posts(&c, &alice, &["poweruser post"]).await;
     tokio::time::sleep(Duration::from_millis(100)).await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", alice.as_str()), ("limit", "50")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &alice, &[("limit", "50")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().expect("posts is array");

crates/appview/tests/auth_integration.rs

@@ -0,0 +1,311 @@
+//! End-to-end tests for AppView authentication.
+//!
+//! Unlike the other suites in this directory these use a **real**
+//! token: the test creates an account on the running PDS
+//! (`com.atproto.server.createAccount`, the same call the desktop
+//! client makes at signup) and presents the access JWT it gets back to
+//! the AppView. That is the whole point — it exercises the actual key
+//! distribution path (PDS signs → publishes its public key at
+//! `/.well-known/did.json` → AppView fetches and verifies), which a
+//! self-minted token would bypass.
+//!
+//! Fail-open like the rest of the suites: if the PDS or the AppView
+//! isn't running, or the AppView runs with `APPVIEW_AUTH_REQUIRED=false`,
+//! the test prints a notice and returns successfully.
+
+use serde_json::{json, Value};
+use std::time::Duration;
+
+const PDS_URL: &str = "http://127.0.0.1:2583";
+
+fn appview_url() -> String {
+    std::env::var("APPVIEW_TEST_URL").unwrap_or_else(|_| "http://127.0.0.1:2584".to_string())
+}
+
+/// 30 s, not the 5 s the sibling suites use.
+///
+/// The authenticated happy path for `/api/timeline/home` runs the
+/// cold-start query (a brand-new account follows nobody, so the handler
+/// falls back to the global recent feed). On a developer machine whose
+/// AppView has been indexing the public firehose for a while that scan
+/// takes seconds — a timeout there would look like an auth failure and
+/// is nothing of the sort.
+fn client() -> reqwest::Client {
+    reqwest::Client::builder()
+        .timeout(Duration::from_secs(30))
+        .build()
+        .unwrap()
+}
+
+async fn up(c: &reqwest::Client, base: &str) -> bool {
+    for _ in 0..20 {
+        if let Ok(r) = c.get(format!("{base}/healthz")).send().await {
+            if r.status().is_success() {
+                return true;
+            }
+        }
+        tokio::time::sleep(Duration::from_millis(250)).await;
+    }
+    false
+}
+
+/// A freshly created PDS account.
+struct Account {
+    did: String,
+    access_jwt: String,
+    refresh_jwt: String,
+}
+
+/// Guard for every test here: both services up **and** the AppView
+/// actually enforcing auth. Returns the client plus a new account.
+async fn ready() -> Option<(reqwest::Client, Account)> {
+    let c = client();
+    if !up(&c, &appview_url()).await {
+        eprintln!("appview not running, skipping");
+        return None;
+    }
+    if !up(&c, PDS_URL).await {
+        eprintln!("pds not running, skipping");
+        return None;
+    }
+    // Probe: does this instance enforce auth? An operator running with
+    // `APPVIEW_AUTH_REQUIRED=false` (the mode the other suites use)
+    // would otherwise see every assertion here fail for the one reason
+    // that isn't a bug.
+    let probe = c
+        .get(format!("{}/api/notifications/count", appview_url()))
+        .query(&[("did", "did:plc:auth_probe")])
+        .send()
+        .await
+        .ok()?;
+    if probe.status().as_u16() != 401 {
+        eprintln!(
+            "appview does not enforce auth (probe returned {}), skipping",
+            probe.status()
+        );
+        return None;
+    }
+    let acc = create_account(&c).await?;
+    Some((c, acc))
+}
+
+async fn create_account(c: &reqwest::Client) -> Option<Account> {
+    let handle = format!("auth_{}.maarcadetweet.local", uuid::Uuid::new_v4().simple());
+    let r: Value = c
+        .post(format!("{PDS_URL}/xrpc/com.atproto.server.createAccount"))
+        .json(&json!({ "handle": handle, "password": "hunter2hunter2" }))
+        .send()
+        .await
+        .ok()?
+        .json()
+        .await
+        .ok()?;
+    Some(Account {
+        did: r["did"].as_str()?.to_string(),
+        access_jwt: r["access_jwt"].as_str()?.to_string(),
+        refresh_jwt: r["refresh_jwt"].as_str()?.to_string(),
+    })
+}
+
+/// `GET /api/notifications?did=…` with an optional bearer token.
+async fn get_notifications(
+    c: &reqwest::Client,
+    did: &str,
+    token: Option<&str>,
+) -> reqwest::Response {
+    let mut rb = c
+        .get(format!("{}/api/notifications", appview_url()))
+        .query(&[("did", did)]);
+    if let Some(t) = token {
+        rb = rb.bearer_auth(t);
+    }
+    rb.send().await.unwrap()
+}
+
+/// Assert the AppView's standard error envelope: the status, and the
+/// `error` code the desktop client branches on.
+async fn assert_error(r: reqwest::Response, status: u16, code: &str) {
+    let got = r.status().as_u16();
+    let body: Value = r.json().await.unwrap();
+    assert_eq!(got, status, "unexpected status; body = {body}");
+    assert_eq!(body["error"], json!(code), "unexpected error code: {body}");
+    assert!(
+        body["message"].is_string(),
+        "error body must carry a message: {body}"
+    );
+}
+
+#[tokio::test]
+async fn own_token_reads_own_notifications() {
+    let Some((c, acc)) = ready().await else { return };
+    let r = get_notifications(&c, &acc.did, Some(&acc.access_jwt)).await;
+    assert_eq!(r.status().as_u16(), 200);
+    let body: Value = r.json().await.unwrap();
+    // A brand-new account has no notifications, but the shape must be
+    // the normal list response, not an error.
+    assert!(body["notifications"].is_array(), "body = {body}");
+}
+
+#[tokio::test]
+async fn no_token_is_401_auth_missing() {
+    let Some((c, acc)) = ready().await else { return };
+    let r = get_notifications(&c, &acc.did, None).await;
+    assert_error(r, 401, "AuthMissing").await;
+}
+
+#[tokio::test]
+async fn foreign_did_is_403() {
+    let Some((c, acc)) = ready().await else { return };
+    // A valid token, but asking about somebody else's inbox. This is
+    // the case the endpoint used to answer with a 200.
+    let r = get_notifications(&c, "did:plc:somebodyelse", Some(&acc.access_jwt)).await;
+    assert_error(r, 403, "Forbidden").await;
+}
+
+#[tokio::test]
+async fn garbage_token_is_401_token_invalid() {
+    let Some((c, acc)) = ready().await else { return };
+    let r = get_notifications(&c, &acc.did, Some("not.a.jwt")).await;
+    assert_error(r, 401, "TokenInvalid").await;
+}
+
+#[tokio::test]
+async fn refresh_token_is_not_accepted() {
+    let Some((c, acc)) = ready().await else { return };
+    // Signed by the same key and valid for 90 days — only the `scope`
+    // claim separates it from a session credential.
+    let r = get_notifications(&c, &acc.did, Some(&acc.refresh_jwt)).await;
+    assert_error(r, 401, "TokenInvalid").await;
+}
+
+#[tokio::test]
+async fn seen_write_requires_matching_token() {
+    let Some((c, acc)) = ready().await else { return };
+    let url = format!("{}/api/notifications/seen", appview_url());
+
+    // Somebody else's unread badge: 403, nothing written.
+    let r = c
+        .post(&url)
+        .bearer_auth(&acc.access_jwt)
+        .json(&json!({ "did": "did:plc:somebodyelse" }))
+        .send()
+        .await
+        .unwrap();
+    assert_error(r, 403, "Forbidden").await;
+
+    // No credential at all: 401.
+    let r = c
+        .post(&url)
+        .json(&json!({ "did": &acc.did }))
+        .send()
+        .await
+        .unwrap();
+    assert_error(r, 401, "AuthMissing").await;
+
+    // Own inbox: allowed (zero rows updated — the account is new).
+    let r = c
+        .post(&url)
+        .bearer_auth(&acc.access_jwt)
+        .json(&json!({ "did": &acc.did }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(r.status().as_u16(), 200);
+    let body: Value = r.json().await.unwrap();
+    assert_eq!(body["ok"], json!(true));
+}
+
+#[tokio::test]
+async fn timeline_home_is_authenticated() {
+    let Some((c, acc)) = ready().await else { return };
+    let base = appview_url();
+
+    let r = c
+        .get(format!("{base}/api/timeline/home"))
+        .query(&[("did", acc.did.as_str())])
+        .send()
+        .await
+        .unwrap();
+    assert_error(r, 401, "AuthMissing").await;
+
+    let r = c
+        .get(format!("{base}/api/timeline/home"))
+        .query(&[("did", "did:plc:somebodyelse")])
+        .bearer_auth(&acc.access_jwt)
+        .send()
+        .await
+        .unwrap();
+    assert_error(r, 403, "Forbidden").await;
+
+    let r = c
+        .get(format!("{base}/api/timeline/home"))
+        .query(&[("did", acc.did.as_str())])
+        .bearer_auth(&acc.access_jwt)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(r.status().as_u16(), 200);
+}
+
+#[tokio::test]
+async fn public_endpoints_stay_public() {
+    let Some((c, acc)) = ready().await else { return };
+    let base = appview_url();
+
+    // Profiles, search, follower lists and the health probe are public
+    // records in AT Proto. Putting them behind auth would be a
+    // behaviour change no protocol rule asks for — assert they still
+    // answer without a token.
+    for (path, params) in [
+        ("/api/profile", vec![("did", acc.did.as_str())]),
+        ("/api/search", vec![("q", "hello")]),
+        ("/api/followers", vec![("did", acc.did.as_str())]),
+        ("/api/following", vec![("did", acc.did.as_str())]),
+        ("/healthz", vec![]),
+    ] {
+        let r = c
+            .get(format!("{base}{path}"))
+            .query(&params)
+            .send()
+            .await
+            .unwrap();
+        assert_eq!(
+            r.status().as_u16(),
+            200,
+            "{path} must remain public, got {}",
+            r.status()
+        );
+    }
+}
+
+/// The AppView can only verify anything because the PDS publishes its
+/// public key. If this document ever stops carrying a usable key, every
+/// authenticated request degrades to `503 AuthUnavailable` — so assert
+/// the shape the AppView parses.
+#[tokio::test]
+async fn pds_publishes_a_usable_signing_key() {
+    let c = client();
+    if !up(&c, PDS_URL).await {
+        eprintln!("pds not running, skipping");
+        return;
+    }
+    let doc: Value = c
+        .get(format!("{PDS_URL}/.well-known/did.json"))
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let id = doc["id"].as_str().expect("did document needs an id");
+    assert!(id.starts_with("did:web:"), "id = {id}");
+    let vm = &doc["verificationMethod"][0];
+    assert_eq!(vm["type"], json!("Multikey"));
+    assert_eq!(vm["controller"], json!(id));
+    let key = vm["publicKeyMultibase"]
+        .as_str()
+        .expect("verificationMethod needs publicKeyMultibase");
+    // base58-btc multibase: the `z` prefix is what the AppView's
+    // decoder expects.
+    assert!(key.starts_with('z'), "key = {key}");
+}

crates/appview/tests/common/mod.rs

@@ -0,0 +1,167 @@
+//! Shared test support for the AppView integration suites.
+//!
+//! ## Why the suites need this
+//!
+//! `/api/notifications*` and `/api/timeline/home` require a
+//! PDS-issued access token whose `sub` equals the `did` in the request.
+//! The suites, however, seed synthetic DIDs (`did:plc:ntf_…`) through
+//! `/internal/ingest-commit` — accounts the PDS has never heard of, so
+//! there is no `createSession` that would hand out a token for them.
+//!
+//! The way out is that a token is just an ES256 JWT signed with the
+//! server key derived from `PDS_JWT_SECRET`. A test that can read that
+//! secret (from the process environment, or from the repo `.env` the
+//! dev stack itself was started with) can mint a token for any DID it
+//! likes — the same thing `pds-server/src/jwt_issuer.rs` does.
+//!
+//! ## Fail-open, like the rest of the suites
+//!
+//! [`TestAuth::probe`] asks the running AppView whether it enforces
+//! auth at all:
+//!
+//! - not enforcing (`APPVIEW_AUTH_REQUIRED=false`) → no header needed;
+//! - enforcing and we have the secret → mint per-DID tokens;
+//! - enforcing and we don't → `None`, and the caller skips, exactly as
+//!   it already skips when the service or the database is down.
+
+#![allow(dead_code)]
+
+use at_crypto::ecdsa::P256Keypair;
+use at_crypto::jwt::{issue_jwt, JwtClaims};
+
+/// Audience the PDS stamps into access tokens. Not validated by
+/// `verify_jwt` today (`validate_aud = false`), but minting a token
+/// that differs from the real thing would make this helper a poor
+/// stand-in for the client.
+const APPVIEW_AUD: &str = "did:web:appview.maarcadetweet.local";
+
+/// The scope the AppView insists on. A token with any other scope —
+/// `com.atproto.refresh`, say — is rejected with `TokenInvalid`.
+pub const ACCESS_SCOPE: &str = "com.atproto.access";
+
+/// How the suite should authenticate against the AppView under test.
+///
+/// `Clone` because a test that pages through results in a closure has
+/// to hand each iteration its own copy, exactly like the client and the
+/// base URL next to it.
+#[derive(Clone)]
+pub enum TestAuth {
+    /// `APPVIEW_AUTH_REQUIRED=false`: send no `Authorization` header.
+    Disabled,
+    /// Auth is enforced; mint tokens with this hex secret.
+    Secret(String),
+}
+
+impl TestAuth {
+    /// Decide how (or whether) this suite can talk to the AppView.
+    ///
+    /// Returns `None` when the AppView enforces auth but no
+    /// `PDS_JWT_SECRET` is reachable — the caller should print a notice
+    /// and return, keeping `cargo test --workspace` green on a machine
+    /// without the dev stack's environment.
+    pub async fn probe(c: &reqwest::Client, base_url: &str) -> Option<Self> {
+        // An unauthenticated probe against a private endpoint. We only
+        // look at the status: 401 means the extractor is active. A DID
+        // that doesn't exist is fine — the auth check runs first.
+        let status = c
+            .get(format!("{base_url}/api/notifications/count"))
+            .query(&[("did", "did:plc:auth_probe")])
+            .send()
+            .await
+            .ok()?
+            .status()
+            .as_u16();
+        if status != 401 && status != 503 {
+            return Some(TestAuth::Disabled);
+        }
+        match pds_jwt_secret() {
+            Some(secret) => Some(TestAuth::Secret(secret)),
+            None => {
+                eprintln!(
+                    "appview enforces auth (probe returned {status}) but PDS_JWT_SECRET \
+                     is not set and no .env was found — skipping"
+                );
+                None
+            }
+        }
+    }
+
+    /// Attach an `Authorization: Bearer` header for `did`, if needed.
+    pub fn apply(&self, rb: reqwest::RequestBuilder, did: &str) -> reqwest::RequestBuilder {
+        match self {
+            TestAuth::Disabled => rb,
+            TestAuth::Secret(secret) => match mint_access_jwt(secret, did, ACCESS_SCOPE, 3600) {
+                Some(token) => rb.bearer_auth(token),
+                None => rb,
+            },
+        }
+    }
+
+    /// A token for `did` — for tests that want to send a *wrong* one on
+    /// purpose. `None` when auth is disabled, in which case the test
+    /// that needs it should skip.
+    pub fn token_for(&self, did: &str) -> Option<String> {
+        match self {
+            TestAuth::Disabled => None,
+            TestAuth::Secret(secret) => mint_access_jwt(secret, did, ACCESS_SCOPE, 3600),
+        }
+    }
+}
+
+/// `PDS_JWT_SECRET` from the environment, falling back to the repo
+/// `.env` — the same file the running dev stack loaded at startup, so
+/// the minted tokens verify against the key the PDS actually publishes.
+pub fn pds_jwt_secret() -> Option<String> {
+    if let Ok(v) = std::env::var("PDS_JWT_SECRET") {
+        if !v.trim().is_empty() {
+            return Some(v);
+        }
+    }
+    // `dotenvy::dotenv` walks up from the current directory, which for
+    // a test binary is the crate root — so this finds the workspace
+    // `.env` two levels up. It never overrides a real env var.
+    let _ = dotenvy::dotenv();
+    std::env::var("PDS_JWT_SECRET")
+        .ok()
+        .filter(|v| !v.trim().is_empty())
+}
+
+/// Mint an access JWT for `did`, signed with the PDS's server key.
+///
+/// Mirrors `pds-server/src/jwt_issuer.rs`: the P-256 secret scalar is
+/// the **first 32 bytes** of `PDS_JWT_SECRET` (the config allows a
+/// longer secret), i.e. the first 64 hex characters.
+///
+/// `ttl_secs` may be negative to build a deliberately expired token.
+pub fn mint_access_jwt(
+    secret_hex: &str,
+    did: &str,
+    scope: &str,
+    ttl_secs: i64,
+) -> Option<String> {
+    let hex = secret_hex.trim().trim_start_matches("0x");
+    if hex.len() < 64 || !hex.chars().all(|c| c.is_ascii_hexdigit()) {
+        eprintln!("PDS_JWT_SECRET is not ≥32 bytes of hex; cannot mint a test token");
+        return None;
+    }
+    let kp = P256Keypair {
+        secret_hex: hex[..64].to_string(),
+        // Only the signing half is used by `issue_jwt`; the verifier
+        // fetches the public key from the PDS's DID document.
+        public_multibase: String::new(),
+    };
+    let now = chrono::Utc::now().timestamp();
+    issue_jwt(
+        &kp,
+        &JwtClaims {
+            iss: "did:web:test".into(),
+            sub: did.to_string(),
+            aud: APPVIEW_AUD.into(),
+            iat: now - 1,
+            exp: now + ttl_secs,
+            jti: None,
+            scope: Some(scope.to_string()),
+        },
+    )
+    .ok()
+}

crates/appview/tests/embeds_integration.rs

@@ -15,6 +15,9 @@
 //! and returns rather than panicking. The point of the tests is to
 //! catch regressions in CI where the service IS up.
 
+mod common;
+
+use common::TestAuth;
 use serde_json::{json, Value};
 use std::time::Duration;
 
@@ -50,6 +53,32 @@ async fn db_reachable() -> bool {
     )
 }
 
+/// `/api/timeline/home` requires a token whose `sub` is the requested
+/// DID. The DIDs here are synthetic, so the token is minted from the
+/// PDS signing secret — see `tests/common/mod.rs`. `None` → skip.
+async fn auth_or_skip() -> Option<TestAuth> {
+    TestAuth::probe(&client().await, APPVIEW_URL).await
+}
+
+/// `GET /api/timeline/home` as `did`, authenticated when required.
+async fn get_timeline(
+    c: &reqwest::Client,
+    auth: &TestAuth,
+    did: &str,
+    extra: &[(&str, &str)],
+) -> reqwest::Response {
+    let mut params: Vec<(&str, &str)> = vec![("did", did)];
+    params.extend_from_slice(extra);
+    auth.apply(
+        c.get(format!("{APPVIEW_URL}/api/timeline/home"))
+            .query(&params),
+        did,
+    )
+    .send()
+    .await
+    .unwrap()
+}
+
 async fn post_ingest(c: &reqwest::Client, body: Value) -> reqwest::Response {
     c.post(format!("{APPVIEW_URL}/internal/ingest-commit"))
         .json(&body)
@@ -125,6 +154,7 @@ async fn timeline_includes_embed() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let did = did_for_test("img");
     let uri = seed_post(
         &c,
@@ -160,12 +190,7 @@ async fn timeline_includes_embed() {
     )
     .await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", did.as_str()), ("limit", "10")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "10")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().unwrap();
@@ -199,6 +224,7 @@ async fn timeline_includes_external_embed() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let did = did_for_test("ext");
     let uri = seed_post(
         &c,
@@ -222,12 +248,7 @@ async fn timeline_includes_external_embed() {
     )
     .await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", did.as_str()), ("limit", "10")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "10")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().unwrap();
@@ -241,12 +262,7 @@ async fn timeline_includes_external_embed() {
             break;
         }
         tokio::time::sleep(Duration::from_millis(50)).await;
-        let resp = c
-            .get(format!("{APPVIEW_URL}/api/timeline/home"))
-            .query(&[("did", did.as_str()), ("limit", "10")])
-            .send()
-            .await
-            .unwrap();
+        let resp = get_timeline(&c, &auth, &did, &[("limit", "10")]).await;
         let body: Value = resp.json().await.unwrap();
         our = body["posts"]
             .as_array()
@@ -410,6 +426,7 @@ async fn timeline_post_without_embed_has_null_embed() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let did = did_for_test("plain");
     let uri = seed_post(
         &c,
@@ -421,12 +438,7 @@ async fn timeline_post_without_embed_has_null_embed() {
     )
     .await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", did.as_str()), ("limit", "10")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "10")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let our = body["posts"]

crates/appview/tests/notifications_integration.rs

@@ -14,6 +14,9 @@
 //! API. That's deliberate: it's the only way to catch a mismatch
 //! between what the write path stores and what the read path joins.
 
+mod common;
+
+use common::TestAuth;
 use serde_json::{json, Value};
 use std::time::Duration;
 
@@ -55,8 +58,12 @@ async fn db_pool() -> Option<sqlx::PgPool> {
 }
 
 /// Guard used at the top of every test. Returns `None` (→ skip) unless
-/// both the HTTP service and the database are up.
-async fn ready() -> Option<(reqwest::Client, sqlx::PgPool)> {
+/// the HTTP service is up, the database is reachable, **and** we know
+/// how to authenticate: `/api/notifications*` now requires a token
+/// whose `sub` is the requested DID, and the DIDs seeded here are
+/// synthetic, so the token has to be minted from the PDS's signing
+/// secret. See `tests/common/mod.rs`.
+async fn ready() -> Option<(reqwest::Client, sqlx::PgPool, TestAuth)> {
     if !wait_for_appview_db().await {
         eprintln!("appview not running, skipping");
         return None;
@@ -65,7 +72,32 @@ async fn ready() -> Option<(reqwest::Client, sqlx::PgPool)> {
         eprintln!("appview DB unreachable, skipping");
         return None;
     };
-    Some((client().await, pool))
+    let c = client().await;
+    let auth = TestAuth::probe(&c, &appview_url()).await?;
+    Some((c, pool, auth))
+}
+
+/// `GET <url>` against a private endpoint, carrying the token for
+/// `did` when the instance enforces auth.
+fn authed_get(
+    c: &reqwest::Client,
+    auth: &TestAuth,
+    url: String,
+    did: &str,
+) -> reqwest::RequestBuilder {
+    auth.apply(c.get(url), did)
+}
+
+/// `POST <url>` against a private endpoint. Same rule as
+/// [`authed_get`] — `/api/notifications/seen` is a write into one
+/// user's read state.
+fn authed_post(
+    c: &reqwest::Client,
+    auth: &TestAuth,
+    url: String,
+    did: &str,
+) -> reqwest::RequestBuilder {
+    auth.apply(c.post(url), did)
 }
 
 async fn post_ingest(c: &reqwest::Client, body: Value) -> reqwest::Response {
@@ -181,7 +213,7 @@ async fn seed_follow(c: &reqwest::Client, follower: &str, subject: &str) {
 #[tokio::test]
 async fn notifications_list_count_and_seen() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, auth)) = ready().await else {
         return;
     };
     let alice = did_for_test("alice");
@@ -196,8 +228,7 @@ async fn notifications_list_count_and_seen() {
     let reply_uri = seed_reply(&c, &carol, &post_uri, &post_uri, "carol's reply").await;
     seed_follow(&c, &bob, &alice).await;
 
-    let resp = c
-        .get(format!("{base}/api/notifications"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
         .query(&[("did", alice.as_str()), ("limit", "50")])
         .send()
         .await
@@ -253,8 +284,7 @@ async fn notifications_list_count_and_seen() {
     }
 
     // The unread count agrees with the list.
-    let resp = c
-        .get(format!("{base}/api/notifications/count"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications/count"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -264,8 +294,7 @@ async fn notifications_list_count_and_seen() {
     assert_eq!(body["count"], json!(3));
 
     // Mark everything seen.
-    let resp = c
-        .post(format!("{base}/api/notifications/seen"))
+    let resp = authed_post(&c, &auth, format!("{base}/api/notifications/seen"), &alice)
         .json(&json!({ "did": alice }))
         .send()
         .await
@@ -276,8 +305,7 @@ async fn notifications_list_count_and_seen() {
     assert_eq!(body["updated"], json!(3));
 
     // Idempotent: a second call updates nothing and still succeeds.
-    let resp = c
-        .post(format!("{base}/api/notifications/seen"))
+    let resp = authed_post(&c, &auth, format!("{base}/api/notifications/seen"), &alice)
         .json(&json!({ "did": alice }))
         .send()
         .await
@@ -286,8 +314,7 @@ async fn notifications_list_count_and_seen() {
     assert_eq!(body["updated"], json!(0));
 
     // Count is now zero and the rows carry a read_at.
-    let resp = c
-        .get(format!("{base}/api/notifications/count"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications/count"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -295,8 +322,7 @@ async fn notifications_list_count_and_seen() {
     let body: Value = resp.json().await.unwrap();
     assert_eq!(body["count"], json!(0));
 
-    let resp = c
-        .get(format!("{base}/api/notifications"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -311,8 +337,7 @@ async fn notifications_list_count_and_seen() {
         "/api/notifications",
         "/api/notifications/count",
     ] {
-        let resp = c
-            .get(format!("{base}{path}"))
+        let resp = authed_get(&c, &auth, format!("{base}{path}"), &alice)
             .query(&[("did", "")])
             .send()
             .await
@@ -326,7 +351,7 @@ async fn notifications_list_count_and_seen() {
 #[tokio::test]
 async fn notifications_skip_self_interactions() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, auth)) = ready().await else {
         return;
     };
     let alice = did_for_test("solo");
@@ -335,8 +360,7 @@ async fn notifications_skip_self_interactions() {
     seed_reply(&c, &alice, &post_uri, &post_uri, "and replying too").await;
     seed_follow(&c, &alice, &alice).await;
 
-    let resp = c
-        .get(format!("{base}/api/notifications"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -355,7 +379,7 @@ async fn notifications_skip_self_interactions() {
 #[tokio::test]
 async fn notifications_paginate_with_cursor() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, auth)) = ready().await else {
         return;
     };
     let alice = did_for_test("popular");
@@ -372,9 +396,9 @@ async fn notifications_paginate_with_cursor() {
         let c = c.clone();
         let alice = alice.clone();
         let base = base.clone();
+        let auth = auth.clone();
         async move {
-            let mut req = c
-                .get(format!("{base}/api/notifications"))
+            let mut req = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
                 .query(&[("did", alice.as_str()), ("limit", "5")]);
             if let Some(cur) = cursor {
                 req = req.query(&[("cursor", cur.as_str())]);
@@ -417,8 +441,7 @@ async fn notifications_paginate_with_cursor() {
     assert_eq!(all.len(), 12);
 
     // A mangled cursor is a 400, not a silent restart at page 1.
-    let resp = c
-        .get(format!("{base}/api/notifications"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
         .query(&[("did", alice.as_str()), ("cursor", "!!!garbage!!!")])
         .send()
         .await
@@ -431,7 +454,7 @@ async fn notifications_paginate_with_cursor() {
 #[tokio::test]
 async fn notifications_seen_respects_watermark() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, auth)) = ready().await else {
         return;
     };
     let alice = did_for_test("watermark");
@@ -442,8 +465,7 @@ async fn notifications_seen_respects_watermark() {
 
     // Read back the first notification's indexed_at — that's the
     // watermark a client would echo after rendering page 1.
-    let body: Value = c
-        .get(format!("{base}/api/notifications"))
+    let body: Value = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -461,8 +483,7 @@ async fn notifications_seen_respects_watermark() {
     let second = did_for_test("late");
     seed_like(&c, &second, &post_uri).await;
 
-    let resp = c
-        .post(format!("{base}/api/notifications/seen"))
+    let resp = authed_post(&c, &auth, format!("{base}/api/notifications/seen"), &alice)
         .json(&json!({ "did": alice, "seenAt": watermark }))
         .send()
         .await
@@ -475,8 +496,7 @@ async fn notifications_seen_respects_watermark() {
     );
 
     // The later one is still unread.
-    let body: Value = c
-        .get(format!("{base}/api/notifications/count"))
+    let body: Value = authed_get(&c, &auth, format!("{base}/api/notifications/count"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -487,8 +507,7 @@ async fn notifications_seen_respects_watermark() {
     assert_eq!(body["count"], json!(1));
 
     // snake_case spelling must work identically.
-    let resp = c
-        .post(format!("{base}/api/notifications/seen"))
+    let resp = authed_post(&c, &auth, format!("{base}/api/notifications/seen"), &alice)
         .json(&json!({ "did": alice, "seen_at": null }))
         .send()
         .await
@@ -502,7 +521,7 @@ async fn notifications_seen_respects_watermark() {
 #[tokio::test]
 async fn followers_and_following_lists() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, _auth)) = ready().await else {
         return;
     };
     let hub = did_for_test("hub");
@@ -591,7 +610,8 @@ async fn followers_and_following_lists() {
     assert_eq!(unique.len(), seen.len(), "paged followers repeat: {seen:?}");
     assert_eq!(seen.len(), 3, "paging lost a follower: {seen:?}");
 
-    // `did` is mandatory.
+    // `did` is mandatory. These two stay public — a follow edge is a
+    // public record — so no token is involved.
     for path in ["/api/followers", "/api/following"] {
         let resp = c
             .get(format!("{base}{path}"))
@@ -610,7 +630,7 @@ async fn followers_and_following_lists() {
 #[tokio::test]
 async fn thread_returns_parents_and_replies() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, _auth)) = ready().await else {
         return;
     };
     let a = did_for_test("root");
@@ -689,7 +709,7 @@ async fn thread_returns_parents_and_replies() {
 #[tokio::test]
 async fn post_by_uri_stays_backwards_compatible() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, _auth)) = ready().await else {
         return;
     };
     let a = did_for_test("compat_a");