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 find_or_create_federated(
30        &self,
31        issuer: &str,
32        external_sub: &str,
33        claims: &FederatedIdentityClaims,
34    ) -> Result<User> {
35        let mut conn = self.write_pool.acquire().await?;
36        let mut tx = conn.begin().await?;
37
38        if let Some(existing) = sqlx::query!(
39            "UPDATE federated_identities SET last_seen_at = CURRENT_TIMESTAMP WHERE issuer = $1 \
40             AND external_sub = $2 RETURNING user_id",
41            issuer,
42            external_sub
43        )
44        .fetch_optional(&mut *tx)
45        .await?
46        {
47            let user = sqlx::query_as!(
48                UserRow,
49                r#"
50                SELECT id, name, email, full_name, display_name, status,
51                       email_verified, roles, avatar_url, is_bot, is_scanner,
52                       created_at, updated_at
53                FROM users WHERE id = $1
54                "#,
55                existing.user_id
56            )
57            .fetch_one(&mut *tx)
58            .await
59            .map_err(UserError::from)
60            .and_then(User::try_from)?;
61            tx.commit().await?;
62            return Ok(user);
63        }
64
65        if let Some(existing) =
66            link_by_verified_email(&mut tx, issuer, external_sub, claims).await?
67        {
68            tx.commit().await?;
69            return Ok(existing);
70        }
71
72        let fields = NewFederatedUser::derive(issuer, external_sub, claims);
73
74        let user = sqlx::query_as!(
75            UserRow,
76            r#"
77            INSERT INTO users (
78                id, name, email, full_name, display_name,
79                status, email_verified, roles, is_bot,
80                created_at, updated_at
81            )
82            VALUES ($1, $2, $3, $4, $5, $6, false, $7::TEXT[], false, $8, $8)
83            RETURNING id, name, email, full_name, display_name, status, email_verified,
84                      roles, avatar_url, is_bot, is_scanner, created_at, updated_at
85            "#,
86            fields.id.as_str(),
87            fields.name,
88            fields.email,
89            fields.display_name.as_deref(),
90            fields.display_name.as_deref(),
91            fields.status.as_str(),
92            &fields.roles,
93            fields.now,
94        )
95        .fetch_one(&mut *tx)
96        .await
97        .map_err(UserError::from)
98        .and_then(User::try_from)?;
99
100        sqlx::query!(
101            "INSERT INTO federated_identities (issuer, external_sub, user_id) VALUES ($1, $2, $3)",
102            issuer,
103            external_sub,
104            user.id.as_str()
105        )
106        .execute(&mut *tx)
107        .await?;
108
109        tx.commit().await?;
110        Ok(user)
111    }
112}
113
114async fn link_by_verified_email(
115    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
116    issuer: &str,
117    external_sub: &str,
118    claims: &FederatedIdentityClaims,
119) -> Result<Option<User>> {
120    if !claims.email_verified {
121        return Ok(None);
122    }
123    let Some(addr) = claims.email.as_deref() else {
124        return Ok(None);
125    };
126    let email = normalise_email(addr);
127    let deleted_status = UserStatus::Deleted.as_str();
128    let Some(existing) = sqlx::query_as!(
129        UserRow,
130        r#"
131        SELECT id, name, email, full_name, display_name, status,
132               email_verified, roles, avatar_url, is_bot, is_scanner,
133               created_at, updated_at
134        FROM users WHERE email = $1 AND status != $2
135        "#,
136        email,
137        deleted_status
138    )
139    .fetch_optional(&mut **tx)
140    .await?
141    .map(User::try_from)
142    .transpose()?
143    else {
144        return Ok(None);
145    };
146
147    sqlx::query!(
148        "INSERT INTO federated_identities (issuer, external_sub, user_id) VALUES ($1, $2, $3)",
149        issuer,
150        external_sub,
151        existing.id.as_str()
152    )
153    .execute(&mut **tx)
154    .await?;
155    Ok(Some(existing))
156}
157
158struct NewFederatedUser {
159    id: UserId,
160    name: String,
161    email: String,
162    display_name: Option<String>,
163    status: UserStatus,
164    roles: Vec<String>,
165    now: chrono::DateTime<Utc>,
166}
167
168impl NewFederatedUser {
169    fn derive(issuer: &str, external_sub: &str, claims: &FederatedIdentityClaims) -> Self {
170        let name = claims
171            .preferred_username
172            .clone()
173            .or_else(|| claims.name.clone())
174            .unwrap_or_else(|| format!("fed_{}_{}", short_hash(issuer), short_hash(external_sub)));
175        let synthetic_email = || {
176            format!(
177                "{}@{}.federated.local",
178                short_hash(external_sub),
179                short_host(issuer)
180            )
181        };
182        let email = match (claims.email.as_deref(), claims.email_verified) {
183            (Some(addr), true) => normalise_email(addr),
184            (Some(addr), false) => {
185                tracing::warn!(
186                    issuer,
187                    external_sub,
188                    upstream_email = addr,
189                    "upstream IdP did not assert email_verified; using synthetic local email to \
190                     prevent account-claim attacks"
191                );
192                synthetic_email()
193            },
194            (None, _) => synthetic_email(),
195        };
196
197        Self {
198            id: UserId::generate(),
199            name,
200            email,
201            display_name: claims.name.clone(),
202            status: UserStatus::Active,
203            roles: normalised_roles(&claims.roles),
204            now: Utc::now(),
205        }
206    }
207}
208
209fn normalised_roles(claim_roles: &[String]) -> Vec<String> {
210    if claim_roles.is_empty() {
211        vec![UserRole::User.as_str().to_owned()]
212    } else {
213        claim_roles.to_vec()
214    }
215}
216
217fn short_hash(s: &str) -> String {
218    use sha2::{Digest, Sha256};
219    let digest = Sha256::digest(s.as_bytes());
220    hex::encode(&digest[..6])
221}
222
223fn short_host(issuer: &str) -> String {
224    issuer
225        .trim_start_matches("https://")
226        .trim_start_matches("http://")
227        .split('/')
228        .next()
229        .unwrap_or("issuer")
230        .replace(['.', ':'], "-")
231}
systemprompt_users/repository/federated_identity.rs

systemprompt_users/repository/
federated_identity.rs
