OIDC account linking and login

crates/axum-utils/src/session.rs

@@ -95,7 +95,9 @@ impl<K> SessionInfoExt for PrivateCookieJar<K> {
     }
 
     fn update_session_info(self, info: &SessionInfo) -> Self {
-        let cookie = Cookie::new("session", "");
+        let mut cookie = Cookie::new("session", "");
+        cookie.set_path("/");
+        cookie.set_http_only(true);
         let cookie = cookie.encode(&info);
         self.add(cookie)
     }

crates/data-model/src/upstream_oauth2/mod.rs

@@ -45,6 +45,7 @@ pub struct UpstreamOAuthAuthorizationSession {
     pub nonce: String,
     pub created_at: DateTime<Utc>,
     pub completed_at: Option<DateTime<Utc>>,
+    pub consumed_at: Option<DateTime<Utc>>,
 }
 
 impl UpstreamOAuthAuthorizationSession {
@@ -52,4 +53,9 @@ impl UpstreamOAuthAuthorizationSession {
     pub const fn completed(&self) -> bool {
         self.completed_at.is_some()
     }
+
+    #[must_use]
+    pub const fn consumed(&self) -> bool {
+        self.consumed_at.is_some()
+    }
 }

crates/handlers/src/upstream_oauth2/callback.rs

@@ -81,9 +81,6 @@ pub(crate) enum RouteError {
     #[error("Invalid ID token")]
     InvalidIdToken(#[from] ClaimError),
 
-    #[error("User already linked")]
-    UserAlreadyLinked,
-
     #[error("Error from the provider: {error}")]
     ClientError {
         error: ClientErrorCode,
@@ -293,12 +290,7 @@ pub(crate) async fn get(
         .await
         .to_option()?;
 
-    let link = if let Some((link, maybe_user_id)) = maybe_link {
+    let link = if let Some((link, _maybe_user_id)) = maybe_link {
-        if let Some(_user_id) = maybe_user_id {
-            // TODO: Here we should login if the user is linked
-            return Err(RouteError::UserAlreadyLinked);
-        }
-
         link
     } else {
         add_link(&mut txn, &mut rng, &clock, &provider, subject).await?

crates/handlers/src/upstream_oauth2/link.rs

@@ -24,9 +24,15 @@ use mas_axum_utils::{
     SessionInfoExt,
 };
 use mas_keystore::Encrypter;
+use mas_router::Route;
 use mas_storage::{
-    upstream_oauth2::{lookup_link, lookup_session_on_link},
+    upstream_oauth2::{
-    user::{lookup_user, ActiveSessionLookupError, UserLookupError},
+        associate_link_to_user, consume_session, lookup_link, lookup_session_on_link,
+    },
+    user::{
+        authenticate_session_with_upstream, lookup_user, register_passwordless_user, start_session,
+        ActiveSessionLookupError, UserLookupError,
+    },
     GenericLookupError, LookupResultExt,
 };
 use mas_templates::{
@@ -47,6 +53,10 @@ pub(crate) enum RouteError {
     #[error("Session not found")]
     SessionNotFound,
 
+    /// Session was already consumed
+    #[error("Session already consumed")]
+    SessionConsumed,
+
     #[error("Missing session cookie")]
     MissingCookie,
 
@@ -145,21 +155,33 @@ pub(crate) async fn get(
 
     // This checks that we're in a browser session which is allowed to consume this
     // link: the upstream auth session should have been started in this browser.
-    let _upstream_session = lookup_session_on_link(&mut txn, &link, session_id)
+    let upstream_session = lookup_session_on_link(&mut txn, &link, session_id)
         .await
         .to_option()?
         .ok_or(RouteError::SessionNotFound)?;
 
+    if upstream_session.consumed() {
+        return Err(RouteError::SessionConsumed);
+    }
+
     let (user_session_info, cookie_jar) = cookie_jar.session_info();
-    let (csrf_token, cookie_jar) = cookie_jar.csrf_token(clock.now(), &mut rng);
+    let (csrf_token, mut cookie_jar) = cookie_jar.csrf_token(clock.now(), &mut rng);
     let maybe_user_session = user_session_info.load_session(&mut txn).await?;
 
     let render = match (maybe_user_session, maybe_user_id) {
-        (Some(user_session), Some(user_id)) if user_session.user.data == user_id => {
+        (Some(mut session), Some(user_id)) if session.user.data == user_id => {
             // Session already linked, and link matches the currently logged
-            // user. Do nothing?
+            // user. Mark the session as consumed and renew the authentication.
+            consume_session(&mut txn, &clock, upstream_session).await?;
+            authenticate_session_with_upstream(&mut txn, &mut rng, &clock, &mut session, &link)
+                .await?;
+
+            cookie_jar = cookie_jar.set_session(&session);
+
+            txn.commit().await?;
+
             let ctx = EmptyContext
-                .with_session(user_session)
+                .with_session(session)
                 .with_csrf(csrf_token.form_value());
 
             templates
@@ -217,7 +239,7 @@ pub(crate) async fn post(
     Form(form): Form<ProtectedForm<FormData>>,
 ) -> Result<impl IntoResponse, RouteError> {
     let mut txn = pool.begin().await?;
-    let (clock, _rng) = crate::rng_and_clock()?;
+    let (clock, mut rng) = crate::rng_and_clock()?;
     let form = cookie_jar.verify_form(clock.now(), form)?;
 
     let (link, _provider_id, maybe_user_id) = lookup_link(&mut txn, link_id)
@@ -234,23 +256,45 @@ pub(crate) async fn post(
 
     // This checks that we're in a browser session which is allowed to consume this
     // link: the upstream auth session should have been started in this browser.
-    let _upstream_session = lookup_session_on_link(&mut txn, &link, session_id)
+    let upstream_session = lookup_session_on_link(&mut txn, &link, session_id)
         .await
         .to_option()?
         .ok_or(RouteError::SessionNotFound)?;
 
+    if upstream_session.consumed() {
+        return Err(RouteError::SessionConsumed);
+    }
+
     let (user_session_info, cookie_jar) = cookie_jar.session_info();
     let maybe_user_session = user_session_info.load_session(&mut txn).await?;
 
-    let res = match (maybe_user_session, maybe_user_id, form) {
+    let mut session = match (maybe_user_session, maybe_user_id, form) {
-        (Some(_user_session), None, FormData::Link) => "Linked!".to_owned(),
+        (Some(session), None, FormData::Link) => {
+            associate_link_to_user(&mut txn, &link, &session.user).await?;
+            session
+        }
 
-        (None, Some(_user_id), FormData::Login) => "Logged in!".to_owned(),
+        (None, Some(user_id), FormData::Login) => {
+            let user = lookup_user(&mut txn, user_id).await?;
+            start_session(&mut txn, &mut rng, &clock, user).await?
+        }
 
-        (None, None, FormData::Register { username }) => format!("Registered {username}!"),
+        (None, None, FormData::Register { username }) => {
+            let user = register_passwordless_user(&mut txn, &mut rng, &clock, &username).await?;
+            associate_link_to_user(&mut txn, &link, &user).await?;
+
+            start_session(&mut txn, &mut rng, &clock, user).await?
+        }
 
         _ => return Err(RouteError::InvalidFormAction),
     };
 
-    Ok((cookie_jar, res))
+    consume_session(&mut txn, &clock, upstream_session).await?;
+    authenticate_session_with_upstream(&mut txn, &mut rng, &clock, &mut session, &link).await?;
+
+    let cookie_jar = cookie_jar.set_session(&session);
+
+    txn.commit().await?;
+
+    Ok((cookie_jar, mas_router::Index.go()))
 }

crates/storage/migrations/20221121151402_upstream_oauth.sql

@@ -80,5 +80,9 @@ CREATE TABLE "upstream_oauth_authorization_sessions" (
   "nonce" TEXT NOT NULL,
 
   "created_at" TIMESTAMP WITH TIME ZONE NOT NULL,
-  "completed_at" TIMESTAMP WITH TIME ZONE
+
+  -- When the session turned into a link
+  "completed_at" TIMESTAMP WITH TIME ZONE,
+  -- When the session turned into a user session authentication
+  "consumed_at" TIMESTAMP WITH TIME ZONE
 );

crates/storage/sqlx-data.json

@@ -339,6 +339,76 @@
     },
     "query": "\n            UPDATE user_emails\n            SET confirmed_at = $2\n            WHERE user_email_id = $1\n        "
   },
+  "1758533e0b323452384c8484aee7b0a32fecdd6238d270b68d0fac496816db65": {
+    "describe": {
+      "columns": [
+        {
+          "name": "upstream_oauth_authorization_session_id",
+          "ordinal": 0,
+          "type_info": "Uuid"
+        },
+        {
+          "name": "state",
+          "ordinal": 1,
+          "type_info": "Text"
+        },
+        {
+          "name": "code_challenge_verifier",
+          "ordinal": 2,
+          "type_info": "Text"
+        },
+        {
+          "name": "nonce",
+          "ordinal": 3,
+          "type_info": "Text"
+        },
+        {
+          "name": "created_at",
+          "ordinal": 4,
+          "type_info": "Timestamptz"
+        },
+        {
+          "name": "completed_at",
+          "ordinal": 5,
+          "type_info": "Timestamptz"
+        },
+        {
+          "name": "consumed_at",
+          "ordinal": 6,
+          "type_info": "Timestamptz"
+        }
+      ],
+      "nullable": [
+        false,
+        false,
+        true,
+        false,
+        false,
+        true,
+        true
+      ],
+      "parameters": {
+        "Left": [
+          "Uuid",
+          "Uuid"
+        ]
+      }
+    },
+    "query": "\n            SELECT\n                upstream_oauth_authorization_session_id,\n                state,\n                code_challenge_verifier,\n                nonce,\n                created_at,\n                completed_at,\n                consumed_at\n            FROM upstream_oauth_authorization_sessions\n            WHERE upstream_oauth_authorization_session_id = $1\n              AND upstream_oauth_link_id = $2\n        "
+  },
+  "1e7b1b7e06b5d97d81dc4a8524bb223c3dc7ddbbcce7cc2a142dbfbdd6a2902e": {
+    "describe": {
+      "columns": [],
+      "nullable": [],
+      "parameters": {
+        "Left": [
+          "Uuid",
+          "Uuid"
+        ]
+      }
+    },
+    "query": "\n            UPDATE upstream_oauth_links\n            SET user_id = $1\n            WHERE upstream_oauth_link_id = $2\n        "
+  },
   "1eb6d13e75d8f526c2785749a020731c18012f03e07995213acd38ab560ce497": {
     "describe": {
       "columns": [],
@@ -372,6 +442,20 @@
     },
     "query": "\n            INSERT INTO upstream_oauth_providers (\n                upstream_oauth_provider_id,\n                issuer,\n                scope,\n                token_endpoint_auth_method,\n                token_endpoint_signing_alg,\n                client_id,\n                encrypted_client_secret,\n                created_at\n            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)\n        "
   },
+  "1fec001d1592641695df8806170f82d7667666f4df0b8ae5c614055a6cdaae9d": {
+    "describe": {
+      "columns": [],
+      "nullable": [],
+      "parameters": {
+        "Left": [
+          "Uuid",
+          "Timestamptz",
+          "Uuid"
+        ]
+      }
+    },
+    "query": "\n            UPDATE upstream_oauth_authorization_sessions\n            SET upstream_oauth_link_id = $1,\n                completed_at = $2\n            WHERE upstream_oauth_authorization_session_id = $3\n        "
+  },
   "2153118b364a33582e7f598acce3789fcb8d938948a819b15cf0b6d37edf58b2": {
     "describe": {
       "columns": [],
@@ -998,19 +1082,6 @@
     },
     "query": "\n            INSERT INTO user_passwords (user_password_id, user_id, hashed_password, created_at)\n            VALUES ($1, $2, $3, $4)\n        "
   },
-  "4b6a44d040a0dc849bb4e04abb11a181995b5847917605ef4c160389686a54f5": {
-    "describe": {
-      "columns": [],
-      "nullable": [],
-      "parameters": {
-        "Left": [
-          "Uuid",
-          "Timestamptz"
-        ]
-      }
-    },
-    "query": "\n            UPDATE upstream_oauth_authorization_sessions\n            SET upstream_oauth_link_id = $1,\n                completed_at = $2\n        "
-  },
   "4f8ec19f3f1bfe0268fe102a24e5a9fa542e77eccbebdce65e6deb1c197adf36": {
     "describe": {
       "columns": [
@@ -1148,23 +1219,6 @@
     },
     "query": "\n            SELECT scope_token\n            FROM oauth2_consents\n            WHERE user_id = $1 AND oauth2_client_id = $2\n        "
   },
-  "53a652f0892d25654fe937962913f2f964463fd09f518066fbc83808edc5b394": {
-    "describe": {
-      "columns": [],
-      "nullable": [],
-      "parameters": {
-        "Left": [
-          "Uuid",
-          "Uuid",
-          "Text",
-          "Text",
-          "Text",
-          "Timestamptz"
-        ]
-      }
-    },
-    "query": "\n            INSERT INTO upstream_oauth_authorization_sessions (\n                upstream_oauth_authorization_session_id,\n                upstream_oauth_provider_id,\n                state,\n                code_challenge_verifier,\n                nonce,\n                created_at,\n                completed_at\n            ) VALUES ($1, $2, $3, $4, $5, $6, NULL)\n        "
-  },
   "559a486756d08d101eb7188ef6637b9d24c024d056795b8121f7f04a7f9db6a3": {
     "describe": {
       "columns": [
@@ -1248,57 +1302,6 @@
     },
     "query": "\n            DELETE FROM oauth2_access_tokens\n            WHERE expires_at < $1\n        "
   },
-  "5cb91740580a37044dd37c90a2fadaab9abcd387c7883f47c73c18a8fa260683": {
-    "describe": {
-      "columns": [
-        {
-          "name": "upstream_oauth_authorization_session_id",
-          "ordinal": 0,
-          "type_info": "Uuid"
-        },
-        {
-          "name": "state",
-          "ordinal": 1,
-          "type_info": "Text"
-        },
-        {
-          "name": "code_challenge_verifier",
-          "ordinal": 2,
-          "type_info": "Text"
-        },
-        {
-          "name": "nonce",
-          "ordinal": 3,
-          "type_info": "Text"
-        },
-        {
-          "name": "created_at",
-          "ordinal": 4,
-          "type_info": "Timestamptz"
-        },
-        {
-          "name": "completed_at",
-          "ordinal": 5,
-          "type_info": "Timestamptz"
-        }
-      ],
-      "nullable": [
-        false,
-        false,
-        true,
-        false,
-        false,
-        true
-      ],
-      "parameters": {
-        "Left": [
-          "Uuid",
-          "Uuid"
-        ]
-      }
-    },
-    "query": "\n            SELECT\n                upstream_oauth_authorization_session_id,\n                state,\n                code_challenge_verifier,\n                nonce,\n                created_at,\n                completed_at\n            FROM upstream_oauth_authorization_sessions\n            WHERE upstream_oauth_authorization_session_id = $1\n              AND upstream_oauth_link_id = $2\n        "
-  },
   "5ccde09ee3fe43e7b492d73fa67708b5dcb2b7496c4d05bcfcf0ea63c7576d48": {
     "describe": {
       "columns": [
@@ -1400,20 +1403,7 @@
     },
     "query": "\n            UPDATE user_sessions\n            SET finished_at = $1\n            WHERE user_session_id = $2\n        "
   },
-  "6bf0da5ba3dd07b499193a2e0ddeea6e712f9df8f7f28874ff56a952a9f10e54": {
+  "661c4532f91c7c00d991dadcde6d8440c786777c09b4e75f329522479e6890aa": {
-    "describe": {
-      "columns": [],
-      "nullable": [],
-      "parameters": {
-        "Left": [
-          "Uuid",
-          "Timestamptz"
-        ]
-      }
-    },
-    "query": "\n            UPDATE oauth2_access_tokens\n            SET revoked_at = $2\n            WHERE oauth2_access_token_id = $1\n        "
-  },
-  "6c8816b2618db8d04ab9393429866d9af59ad280949947fc025c89baffe6a455": {
     "describe": {
       "columns": [
         {
@@ -1452,38 +1442,43 @@
           "type_info": "Timestamptz"
         },
         {
-          "name": "provider_issuer",
+          "name": "consumed_at",
           "ordinal": 7,
-          "type_info": "Text"
+          "type_info": "Timestamptz"
         },
         {
-          "name": "provider_scope",
+          "name": "provider_issuer",
           "ordinal": 8,
           "type_info": "Text"
         },
         {
-          "name": "provider_client_id",
+          "name": "provider_scope",
           "ordinal": 9,
           "type_info": "Text"
         },
         {
-          "name": "provider_encrypted_client_secret",
+          "name": "provider_client_id",
           "ordinal": 10,
           "type_info": "Text"
         },
         {
-          "name": "provider_token_endpoint_auth_method",
+          "name": "provider_encrypted_client_secret",
           "ordinal": 11,
           "type_info": "Text"
         },
         {
-          "name": "provider_token_endpoint_signing_alg",
+          "name": "provider_token_endpoint_auth_method",
           "ordinal": 12,
           "type_info": "Text"
         },
         {
-          "name": "provider_created_at",
+          "name": "provider_token_endpoint_signing_alg",
           "ordinal": 13,
+          "type_info": "Text"
+        },
+        {
+          "name": "provider_created_at",
+          "ordinal": 14,
           "type_info": "Timestamptz"
         }
       ],
@@ -1495,6 +1490,7 @@
         false,
         false,
         true,
+        true,
         false,
         false,
         false,
@@ -1509,7 +1505,20 @@
         ]
       }
     },
-    "query": "\n            SELECT\n                ua.upstream_oauth_authorization_session_id,\n                ua.upstream_oauth_provider_id,\n                ua.state,\n                ua.code_challenge_verifier,\n                ua.nonce,\n                ua.created_at,\n                ua.completed_at,\n                up.issuer AS \"provider_issuer\",\n                up.scope AS \"provider_scope\",\n                up.client_id AS \"provider_client_id\",\n                up.encrypted_client_secret AS \"provider_encrypted_client_secret\",\n                up.token_endpoint_auth_method AS \"provider_token_endpoint_auth_method\",\n                up.token_endpoint_signing_alg AS \"provider_token_endpoint_signing_alg\",\n                up.created_at AS \"provider_created_at\"\n            FROM upstream_oauth_authorization_sessions ua\n            INNER JOIN upstream_oauth_providers up\n              USING (upstream_oauth_provider_id)\n            WHERE upstream_oauth_authorization_session_id = $1\n        "
+    "query": "\n            SELECT\n                ua.upstream_oauth_authorization_session_id,\n                ua.upstream_oauth_provider_id,\n                ua.state,\n                ua.code_challenge_verifier,\n                ua.nonce,\n                ua.created_at,\n                ua.completed_at,\n                ua.consumed_at,\n                up.issuer AS \"provider_issuer\",\n                up.scope AS \"provider_scope\",\n                up.client_id AS \"provider_client_id\",\n                up.encrypted_client_secret AS \"provider_encrypted_client_secret\",\n                up.token_endpoint_auth_method AS \"provider_token_endpoint_auth_method\",\n                up.token_endpoint_signing_alg AS \"provider_token_endpoint_signing_alg\",\n                up.created_at AS \"provider_created_at\"\n            FROM upstream_oauth_authorization_sessions ua\n            INNER JOIN upstream_oauth_providers up\n              USING (upstream_oauth_provider_id)\n            WHERE upstream_oauth_authorization_session_id = $1\n        "
+  },
+  "6bf0da5ba3dd07b499193a2e0ddeea6e712f9df8f7f28874ff56a952a9f10e54": {
+    "describe": {
+      "columns": [],
+      "nullable": [],
+      "parameters": {
+        "Left": [
+          "Uuid",
+          "Timestamptz"
+        ]
+      }
+    },
+    "query": "\n            UPDATE oauth2_access_tokens\n            SET revoked_at = $2\n            WHERE oauth2_access_token_id = $1\n        "
   },
   "7262f81a335a984c4051383d2ede7455ff65ed90fbd3151d625f8a21fd26cb05": {
     "describe": {
@@ -2271,6 +2280,23 @@
     },
     "query": "\n            SELECT EXISTS(\n                SELECT 1 FROM users WHERE username = $1\n            ) AS \"exists!\"\n        "
   },
+  "b3a64e41449c3f35e0ad9810eb164e44443034c6895a10367c2a7c6a98437560": {
+    "describe": {
+      "columns": [],
+      "nullable": [],
+      "parameters": {
+        "Left": [
+          "Uuid",
+          "Uuid",
+          "Text",
+          "Text",
+          "Text",
+          "Timestamptz"
+        ]
+      }
+    },
+    "query": "\n            INSERT INTO upstream_oauth_authorization_sessions (\n                upstream_oauth_authorization_session_id,\n                upstream_oauth_provider_id,\n                state,\n                code_challenge_verifier,\n                nonce,\n                created_at,\n                completed_at,\n                consumed_at\n            ) VALUES ($1, $2, $3, $4, $5, $6, NULL, NULL)\n        "
+  },
   "b5b955169ebe6c399e53b74627c11c8219c0736ef2b5b6b44be568a35fd5389f": {
     "describe": {
       "columns": [
@@ -2590,6 +2616,19 @@
     },
     "query": "\n            INSERT INTO upstream_oauth_links (\n                upstream_oauth_link_id,\n                upstream_oauth_provider_id,\n                user_id,\n                subject,\n                created_at\n            ) VALUES ($1, $2, NULL, $3, $4)\n        "
   },
+  "e30562e9637d3a723a91adca6336a8d083657ce6d7fe9551fcd6a9d672453d3c": {
+    "describe": {
+      "columns": [],
+      "nullable": [],
+      "parameters": {
+        "Left": [
+          "Timestamptz",
+          "Uuid"
+        ]
+      }
+    },
+    "query": "\n            UPDATE upstream_oauth_authorization_sessions\n            SET consumed_at = $1\n            WHERE upstream_oauth_authorization_session_id = $2\n        "
+  },
   "e446e37d48c8838ef2e0d0fd82f8f7b04893c84ad46747cdf193ebd83755ceb2": {
     "describe": {
       "columns": [],

crates/storage/src/upstream_oauth2/link.rs

@@ -13,13 +13,13 @@
 // limitations under the License.
 
 use chrono::{DateTime, Utc};
-use mas_data_model::{UpstreamOAuthLink, UpstreamOAuthProvider};
+use mas_data_model::{UpstreamOAuthLink, UpstreamOAuthProvider, User};
 use rand::Rng;
 use sqlx::PgExecutor;
 use ulid::Ulid;
 use uuid::Uuid;
 
-use crate::{Clock, GenericLookupError};
+use crate::{Clock, GenericLookupError, PostgresqlBackend};
 
 struct LinkLookup {
     upstream_oauth_link_id: Uuid,
@@ -158,3 +158,33 @@ pub async fn add_link(
         created_at,
     })
 }
+
+#[tracing::instrument(
+    skip_all,
+    fields(
+        %upstream_oauth_link.id,
+        %upstream_oauth_link.subject,
+        user.id = %user.data,
+        %user.username,
+    ),
+    err,
+)]
+pub async fn associate_link_to_user(
+    executor: impl PgExecutor<'_>,
+    upstream_oauth_link: &UpstreamOAuthLink,
+    user: &User<PostgresqlBackend>,
+) -> Result<(), sqlx::Error> {
+    sqlx::query!(
+        r#"
+            UPDATE upstream_oauth_links
+            SET user_id = $1
+            WHERE upstream_oauth_link_id = $2
+        "#,
+        Uuid::from(user.data),
+        Uuid::from(upstream_oauth_link.id),
+    )
+    .execute(executor)
+    .await?;
+
+    Ok(())
+}

crates/storage/src/upstream_oauth2/mod.rs

@@ -17,9 +17,10 @@ mod provider;
 mod session;
 
 pub use self::{
-    link::{add_link, lookup_link, lookup_link_by_subject},
+    link::{add_link, associate_link_to_user, lookup_link, lookup_link_by_subject},
     provider::{add_provider, lookup_provider, ProviderLookupError},
     session::{
-        add_session, complete_session, lookup_session, lookup_session_on_link, SessionLookupError,
+        add_session, complete_session, consume_session, lookup_session, lookup_session_on_link,
+        SessionLookupError,
     },
 };

crates/storage/src/upstream_oauth2/session.rs

@@ -43,6 +43,7 @@ struct SessionAndProviderLookup {
     nonce: String,
     created_at: DateTime<Utc>,
     completed_at: Option<DateTime<Utc>>,
+    consumed_at: Option<DateTime<Utc>>,
     provider_issuer: String,
     provider_scope: String,
     provider_client_id: String,
@@ -73,6 +74,7 @@ pub async fn lookup_session(
                 ua.nonce,
                 ua.created_at,
                 ua.completed_at,
+                ua.consumed_at,
                 up.issuer AS "provider_issuer",
                 up.scope AS "provider_scope",
                 up.client_id AS "provider_client_id",
@@ -121,6 +123,7 @@ pub async fn lookup_session(
         nonce: res.nonce,
         created_at: res.created_at,
         completed_at: res.completed_at,
+        consumed_at: res.consumed_at,
     };
 
     Ok((provider, session))
@@ -162,8 +165,9 @@ pub async fn add_session(
                 code_challenge_verifier,
                 nonce,
                 created_at,
-                completed_at
+                completed_at,
-            ) VALUES ($1, $2, $3, $4, $5, $6, NULL)
+                consumed_at
+            ) VALUES ($1, $2, $3, $4, $5, $6, NULL, NULL)
         "#,
         Uuid::from(id),
         Uuid::from(upstream_oauth_provider.id),
@@ -182,6 +186,7 @@ pub async fn add_session(
         nonce,
         created_at,
         completed_at: None,
+        consumed_at: None,
     })
 }
 
@@ -206,9 +211,11 @@ pub async fn complete_session(
             UPDATE upstream_oauth_authorization_sessions
             SET upstream_oauth_link_id = $1,
                 completed_at = $2
+            WHERE upstream_oauth_authorization_session_id = $3
         "#,
         Uuid::from(upstream_oauth_link.id),
         completed_at,
+        Uuid::from(upstream_oauth_authorization_session.id),
     )
     .execute(executor)
     .await?;
@@ -218,6 +225,37 @@ pub async fn complete_session(
     Ok(upstream_oauth_authorization_session)
 }
 
+/// Mark a session as consumed
+#[tracing::instrument(
+    skip_all,
+    fields(
+        %upstream_oauth_authorization_session.id,
+    ),
+    err,
+)]
+pub async fn consume_session(
+    executor: impl PgExecutor<'_>,
+    clock: &Clock,
+    mut upstream_oauth_authorization_session: UpstreamOAuthAuthorizationSession,
+) -> Result<UpstreamOAuthAuthorizationSession, sqlx::Error> {
+    let consumed_at = clock.now();
+    sqlx::query!(
+        r#"
+            UPDATE upstream_oauth_authorization_sessions
+            SET consumed_at = $1
+            WHERE upstream_oauth_authorization_session_id = $2
+        "#,
+        consumed_at,
+        Uuid::from(upstream_oauth_authorization_session.id),
+    )
+    .execute(executor)
+    .await?;
+
+    upstream_oauth_authorization_session.consumed_at = Some(consumed_at);
+
+    Ok(upstream_oauth_authorization_session)
+}
+
 struct SessionLookup {
     upstream_oauth_authorization_session_id: Uuid,
     state: String,
@@ -225,6 +263,7 @@ struct SessionLookup {
     nonce: String,
     created_at: DateTime<Utc>,
     completed_at: Option<DateTime<Utc>>,
+    consumed_at: Option<DateTime<Utc>>,
 }
 
 /// Lookup a session, which belongs to a link, by its ID
@@ -250,7 +289,8 @@ pub async fn lookup_session_on_link(
                 code_challenge_verifier,
                 nonce,
                 created_at,
-                completed_at
+                completed_at,
+                consumed_at
             FROM upstream_oauth_authorization_sessions
             WHERE upstream_oauth_authorization_session_id = $1
               AND upstream_oauth_link_id = $2
@@ -271,5 +311,6 @@ pub async fn lookup_session_on_link(
         nonce: res.nonce,
         created_at: res.created_at,
         completed_at: res.completed_at,
+        consumed_at: res.consumed_at,
     })
 }

crates/storage/src/user.rs

@@ -18,7 +18,7 @@ use anyhow::{bail, Context};
 use argon2::Argon2;
 use chrono::{DateTime, Utc};
 use mas_data_model::{
-    Authentication, BrowserSession, User, UserEmail, UserEmailVerification,
+    Authentication, BrowserSession, UpstreamOAuthLink, User, UserEmail, UserEmailVerification,
     UserEmailVerificationState,
 };
 use password_hash::{PasswordHash, PasswordHasher, SaltString};
@@ -439,6 +439,52 @@ pub async fn authenticate_session(
     Ok(())
 }
 
+#[tracing::instrument(
+    skip_all,
+    fields(
+        user.id = %session.user.data,
+        %upstream_oauth_link.id,
+        user_session.id = %session.data,
+        user_session_authentication.id,
+    ),
+    err,
+)]
+pub async fn authenticate_session_with_upstream(
+    executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
+    session: &mut BrowserSession<PostgresqlBackend>,
+    upstream_oauth_link: &UpstreamOAuthLink,
+) -> Result<(), sqlx::Error> {
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
+    tracing::Span::current().record(
+        "user_session_authentication.id",
+        tracing::field::display(id),
+    );
+
+    sqlx::query!(
+        r#"
+            INSERT INTO user_session_authentications
+                (user_session_authentication_id, user_session_id, created_at)
+            VALUES ($1, $2, $3)
+        "#,
+        Uuid::from(id),
+        Uuid::from(session.data),
+        created_at,
+    )
+    .execute(executor)
+    .instrument(tracing::info_span!("Save authentication"))
+    .await?;
+
+    session.last_authentication = Some(Authentication {
+        data: id,
+        created_at,
+    });
+
+    Ok(())
+}
+
 #[tracing::instrument(
     skip_all,
     fields(
@@ -485,6 +531,44 @@ pub async fn register_user(
     Ok(user)
 }
 
+#[tracing::instrument(
+    skip_all,
+    fields(
+        user.username = username,
+        user.id,
+    ),
+    err,
+)]
+pub async fn register_passwordless_user(
+    executor: impl PgExecutor<'_>,
+    mut rng: impl Rng + Send,
+    clock: &Clock,
+    username: &str,
+) -> Result<User<PostgresqlBackend>, sqlx::Error> {
+    let created_at = clock.now();
+    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
+    tracing::Span::current().record("user.id", tracing::field::display(id));
+
+    sqlx::query!(
+        r#"
+            INSERT INTO users (user_id, username, created_at)
+            VALUES ($1, $2, $3)
+        "#,
+        Uuid::from(id),
+        username,
+        created_at,
+    )
+    .execute(executor)
+    .await?;
+
+    Ok(User {
+        data: id,
+        username: username.to_owned(),
+        sub: id.to_string(),
+        primary_email: None,
+    })
+}
+
 #[tracing::instrument(
     skip_all,
     fields(

templates/pages/upstream_oauth2/already_linked.html

@@ -17,6 +17,7 @@ limitations under the License.
 {% extends "base.html" %}
 
 {% block content %}
+  {{ navbar::top() }}
   <section class="flex items-center justify-center flex-1">
     <h1 class="rounded-lg bg-grey-25 dark:bg-grey-450 p-2 flex flex-col font-medium text-lg text-center">
       Your upstream account is already linked.

templates/pages/upstream_oauth2/link_mismatch.html

@@ -17,6 +17,7 @@ limitations under the License.
 {% extends "base.html" %}
 
 {% block content %}
+  {{ navbar::top() }}
   <section class="flex items-center justify-center flex-1">
     <div class="grid grid-cols-1 gap-6 w-96">
       <h1 class="rounded-lg bg-grey-25 dark:bg-grey-450 p-2 flex flex-col font-medium text-lg text-center">

templates/pages/upstream_oauth2/suggest_link.html

@@ -17,6 +17,7 @@ limitations under the License.
 {% extends "base.html" %}
 
 {% block content %}
+  {{ navbar::top() }}
   <section class="flex items-center justify-center flex-1">
     <div class="grid grid-cols-1 gap-6 w-96">
       <h1 class="rounded-lg bg-grey-25 dark:bg-grey-450 p-2 flex flex-col font-medium text-lg text-center">