systemprompt_users/repository/
federated_identity.rs

1//! Repository for `federated_identities` — the `{issuer, external_sub} ->
2//! users.id` mapping used by RFC 8693 token-exchange first-touch.
3//!
4//! Copyright (c) systemprompt.io — Business Source License 1.1.
5//! See <https://systemprompt.io> for licensing details.
6
7use chrono::Utc;
8use sqlx::Acquire;
9use systemprompt_identifiers::UserId;
10use systemprompt_traits::FederatedIdentityClaims;
11
12use crate::error::{Result, UserError};
13use crate::models::{User, UserRole, UserRow, UserStatus, normalise_email};
14use crate::repository::UserRepository;
15
16impl UserRepository {
17    pub async fn find_federated(&self, issuer: &str, external_sub: &str) -> Result<Option<UserId>> {
18        let row = sqlx::query!(
19            "SELECT user_id FROM federated_identities WHERE issuer = $1 AND external_sub = $2",
20            issuer,
21            external_sub
22        )
23        .fetch_optional(&*self.pool)
24        .await?;
25
26        Ok(row.map(|r| UserId::new(r.user_id)))
27    }
28
29    pub async fn link_federated(
30        &self,
31        issuer: &str,
32        external_sub: &str,
33        user_id: &UserId,
34    ) -> Result<()> {
35        sqlx::query!(
36            "INSERT INTO federated_identities (issuer, external_sub, user_id) VALUES ($1, $2, $3) \
37             ON CONFLICT (issuer, external_sub) DO UPDATE SET user_id = EXCLUDED.user_id, \
38             last_seen_at = CURRENT_TIMESTAMP",
39            issuer,
40            external_sub,
41            user_id.as_str()
42        )
43        .execute(&*self.write_pool)
44        .await?;
45        Ok(())
46    }
47
48    pub async fn find_or_create_federated(
49        &self,
50        issuer: &str,
51        external_sub: &str,
52        claims: &FederatedIdentityClaims,
53    ) -> Result<User> {
54        let mut conn = self.write_pool.acquire().await?;
55        let mut tx = conn.begin().await?;
56
57        lock_first_touch(&mut tx, &format!("federated:{issuer}\n{external_sub}")).await?;
58        if let (true, Some(addr)) = (claims.email_verified, claims.email.as_deref()) {
59            lock_first_touch(&mut tx, &format!("email:{}", normalise_email(addr))).await?;
60        }
61
62        if let Some(existing) = touch_existing_link(&mut tx, issuer, external_sub).await? {
63            tx.commit().await?;
64            return Ok(existing);
65        }
66
67        if let Some(existing) =
68            link_by_verified_email(&mut tx, issuer, external_sub, claims).await?
69        {
70            tx.commit().await?;
71            return Ok(existing);
72        }
73
74        let user = insert_federated_user(&mut tx, issuer, external_sub, claims).await?;
75        tx.commit().await?;
76        Ok(user)
77    }
78}
79
80async fn lock_first_touch(tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, key: &str) -> Result<()> {
81    sqlx::query!("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", key)
82        .fetch_one(&mut **tx)
83        .await?;
84    Ok(())
85}
86
87async fn touch_existing_link(
88    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
89    issuer: &str,
90    external_sub: &str,
91) -> Result<Option<User>> {
92    let Some(existing) = sqlx::query!(
93        "UPDATE federated_identities SET last_seen_at = CURRENT_TIMESTAMP WHERE issuer = $1 \
94         AND external_sub = $2 RETURNING user_id",
95        issuer,
96        external_sub
97    )
98    .fetch_optional(&mut **tx)
99    .await?
100    else {
101        return Ok(None);
102    };
103    let user = sqlx::query_as!(
104        UserRow,
105        r#"
106        SELECT id, name, email, full_name, display_name, status,
107               email_verified, roles, avatar_url, is_bot, is_scanner,
108               created_at, updated_at, account_kind
109        FROM users WHERE id = $1
110        "#,
111        existing.user_id
112    )
113    .fetch_one(&mut **tx)
114    .await
115    .map_err(UserError::from)
116    .and_then(User::try_from)?;
117    Ok(Some(user))
118}
119
120async fn insert_federated_user(
121    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
122    issuer: &str,
123    external_sub: &str,
124    claims: &FederatedIdentityClaims,
125) -> Result<User> {
126    let fields = NewFederatedUser::derive(issuer, external_sub, claims);
127
128    let user = sqlx::query_as!(
129        UserRow,
130        r#"
131        INSERT INTO users (
132            id, name, email, full_name, display_name,
133            status, email_verified, roles, is_bot,
134            created_at, updated_at
135        )
136        VALUES ($1, $2, $3, $4, $5, $6, false, $7::TEXT[], false, $8, $8)
137        RETURNING id, name, email, full_name, display_name, status, email_verified,
138                  roles, avatar_url, is_bot, is_scanner, created_at, updated_at, account_kind
139        "#,
140        fields.id.as_str(),
141        fields.name,
142        fields.email,
143        fields.display_name.as_deref(),
144        fields.display_name.as_deref(),
145        fields.status.as_str(),
146        &fields.roles,
147        fields.now,
148    )
149    .fetch_one(&mut **tx)
150    .await
151    .map_err(UserError::from)
152    .and_then(User::try_from)?;
153
154    sqlx::query!(
155        "INSERT INTO federated_identities (issuer, external_sub, user_id) VALUES ($1, $2, $3)",
156        issuer,
157        external_sub,
158        user.id.as_str()
159    )
160    .execute(&mut **tx)
161    .await?;
162    Ok(user)
163}
164
165async fn link_by_verified_email(
166    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
167    issuer: &str,
168    external_sub: &str,
169    claims: &FederatedIdentityClaims,
170) -> Result<Option<User>> {
171    if !claims.email_verified {
172        return Ok(None);
173    }
174    let Some(addr) = claims.email.as_deref() else {
175        return Ok(None);
176    };
177    let email = normalise_email(addr);
178    let deleted_status = UserStatus::Deleted.as_str();
179    let Some(existing) = sqlx::query_as!(
180        UserRow,
181        r#"
182        SELECT id, name, email, full_name, display_name, status,
183               email_verified, roles, avatar_url, is_bot, is_scanner,
184               created_at, updated_at, account_kind
185        FROM users WHERE email = $1 AND status != $2
186        "#,
187        email,
188        deleted_status
189    )
190    .fetch_optional(&mut **tx)
191    .await?
192    .map(User::try_from)
193    .transpose()?
194    else {
195        return Ok(None);
196    };
197
198    sqlx::query!(
199        "INSERT INTO federated_identities (issuer, external_sub, user_id) VALUES ($1, $2, $3)",
200        issuer,
201        external_sub,
202        existing.id.as_str()
203    )
204    .execute(&mut **tx)
205    .await?;
206    Ok(Some(existing))
207}
208
209struct NewFederatedUser {
210    id: UserId,
211    name: String,
212    email: String,
213    display_name: Option<String>,
214    status: UserStatus,
215    roles: Vec<String>,
216    now: chrono::DateTime<Utc>,
217}
218
219impl NewFederatedUser {
220    fn derive(issuer: &str, external_sub: &str, claims: &FederatedIdentityClaims) -> Self {
221        let name = claims
222            .preferred_username
223            .clone()
224            .or_else(|| claims.name.clone())
225            .unwrap_or_else(|| format!("fed_{}_{}", short_hash(issuer), short_hash(external_sub)));
226        let synthetic_email = || {
227            format!(
228                "{}@{}.federated.local",
229                short_hash(external_sub),
230                short_host(issuer)
231            )
232        };
233        let email = match (claims.email.as_deref(), claims.email_verified) {
234            (Some(addr), true) => normalise_email(addr),
235            (Some(addr), false) => {
236                tracing::warn!(
237                    issuer,
238                    external_sub,
239                    upstream_email = addr,
240                    "upstream IdP did not assert email_verified; using synthetic local email to \
241                     prevent account-claim attacks"
242                );
243                synthetic_email()
244            },
245            (None, _) => synthetic_email(),
246        };
247
248        Self {
249            id: UserId::generate(),
250            name,
251            email,
252            display_name: claims.name.clone(),
253            status: UserStatus::Active,
254            roles: normalised_roles(&claims.roles),
255            now: Utc::now(),
256        }
257    }
258}
259
260fn normalised_roles(claim_roles: &[String]) -> Vec<String> {
261    if claim_roles.is_empty() {
262        vec![UserRole::User.as_str().to_owned()]
263    } else {
264        claim_roles.to_vec()
265    }
266}
267
268fn short_hash(s: &str) -> String {
269    use sha2::{Digest, Sha256};
270    let digest = Sha256::digest(s.as_bytes());
271    hex::encode(&digest[..6])
272}
273
274fn short_host(issuer: &str) -> String {
275    issuer
276        .trim_start_matches("https://")
277        .trim_start_matches("http://")
278        .split('/')
279        .next()
280        .unwrap_or("issuer")
281        .replace(['.', ':'], "-")
282}
systemprompt_users/repository/federated_identity.rs

systemprompt_users/repository/
federated_identity.rs
