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

systemprompt_users/repository/
federated_identity.rs
