Merge branch '3pid-email' into 'next'

3PID email support (including registration UIA)

Closes #381

See merge request famedly/conduit!687

Cargo.toml

@@ -146,6 +146,8 @@ tikv-jemallocator = { version = "0.5.0", features = [
 ], optional = true }
 
 sd-notify = { version = "0.4.1", optional = true }
+async-smtp = "0.9.1"
+tokio-rustls = "0.26.0"
 
 # Used for matrix spec type definitions and helpers
 [dependencies.ruma]
@@ -154,6 +156,7 @@ features = [
   "client-api",
   "compat",
   "federation-api",
+  "identity-service-api",
   "push-gateway-api-c",
   "rand",
   "ring-compat",

src/api/client_server/account.rs

@@ -1,22 +1,37 @@
+use std::{net::SocketAddr, str::FromStr};
+
 use super::{DEVICE_ID_LENGTH, SESSION_ID_LENGTH, TOKEN_LENGTH};
-use crate::{api::client_server, services, utils, Error, Result, Ruma};
+use crate::{
+    api::client_server, service::threepid, services, utils, Error, Result, Ruma, RumaResponse,
+};
+use async_smtp::{Envelope, SendableEmail, SmtpClient, SmtpTransport};
 use ruma::{
-    api::client::{
-        account::{
-            change_password, deactivate, get_3pids, get_username_availability,
-            register::{self, LoginType},
-            request_3pid_management_token_via_email, request_3pid_management_token_via_msisdn,
-            whoami, ThirdPartyIdRemovalStatus,
+    api::{
+        client::{
+            account::{
+                add_3pid, change_password, deactivate, delete_3pid, get_3pids,
+                get_username_availability,
+                register::{self, LoginType},
+                request_3pid_management_token_via_email, request_3pid_management_token_via_msisdn,
+                request_password_change_token_via_email, request_password_change_token_via_msisdn,
+                request_registration_token_via_email, request_registration_token_via_msisdn,
+                whoami, ThirdPartyIdRemovalStatus,
+            },
+            error::ErrorKind,
+            uiaa::{AuthData, AuthFlow, AuthType, UiaaInfo},
         },
-        error::ErrorKind,
-        uiaa::{AuthFlow, AuthType, UiaaInfo},
+        identity_service::association::email::validate_email_by_end_user,
     },
     events::{room::message::RoomMessageEventContent, GlobalAccountDataEventType},
-    push, UserId,
+    push,
+    thirdparty::Medium,
+    OwnedClientSecret, OwnedSessionId, UInt, UserId,
 };
+use tokio::{io::BufStream, net::TcpStream};
 use tracing::{info, warn};
 
 use register::RegistrationKind;
+use url::Url;
 
 const RANDOM_USER_ID_LENGTH: usize = 10;
 
@@ -140,35 +155,29 @@ pub async fn register_route(body: Ruma<register::v3::Request>) -> Result<registe
         ));
     }
 
-    // UIAA
-    let mut uiaainfo;
-    let skip_auth = if services().globals.config.registration_token.is_some() {
-        // Registration token required
-        uiaainfo = UiaaInfo {
-            flows: vec![AuthFlow {
-                stages: vec![AuthType::RegistrationToken],
-            }],
-            completed: Vec::new(),
-            params: Default::default(),
-            session: None,
-            auth_error: None,
-        };
-        body.appservice_info.is_some()
-    } else {
-        // No registration token necessary, but clients must still go through the flow
-        uiaainfo = UiaaInfo {
-            flows: vec![AuthFlow {
-                stages: vec![AuthType::Dummy],
-            }],
-            completed: Vec::new(),
-            params: Default::default(),
-            session: None,
-            auth_error: None,
-        };
-        body.appservice_info.is_some() || is_guest
-    };
+    let mut flows = Vec::new();
+    // Registration token required
+    if services().globals.config.registration_token.is_some() {
+        flows.push(AuthType::RegistrationToken)
+    }
+    // Email verification required
+    if services().globals.config.email_verification.is_some() {
+        flows.push(AuthType::EmailIdentity);
+    }
+    // No registration token or email necessary, but clients must still go through the flow
+    if flows.is_empty() {
+        flows.push(AuthType::Dummy);
+    }
 
-    if !skip_auth {
+    // UIAA
+    if body.appservice_info.is_none() && !is_guest {
+        let mut uiaainfo = UiaaInfo {
+            flows: flows.into_iter().map(|f| AuthFlow::new(vec![f])).collect(),
+            completed: Vec::new(),
+            params: Default::default(),
+            session: None,
+            auth_error: None,
+        };
         if let Some(auth) = &body.auth {
             let (worked, uiaainfo) = services().uiaa.try_auth(
                 &UserId::parse_with_server_name("", services().globals.server_name())
@@ -230,6 +239,18 @@ pub async fn register_route(body: Ruma<register::v3::Request>) -> Result<registe
         .expect("to json always works"),
     )?;
 
+    if let Some(AuthData::EmailIdentity(data)) = &body.auth {
+        let threepid = services()
+            .threepid
+            .find_validated_token(
+                &data.thirdparty_id_creds.client_secret,
+                &data.thirdparty_id_creds.sid,
+            )?
+            .expect("we just validated this");
+
+        services().threepid.add_threepid(&user_id, &threepid)?;
+    }
+
     // Inhibit login does not work for guests
     if !is_guest && body.inhibit_login {
         return Ok(register::v3::Response {
@@ -313,19 +334,29 @@ pub async fn register_route(body: Ruma<register::v3::Request>) -> Result<registe
 /// - Forgets to-device events
 /// - Triggers device list updates
 pub async fn change_password_route(
-    body: Ruma<change_password::v3::Request>,
+    mut body: Ruma<change_password::v3::Request>,
 ) -> Result<change_password::v3::Response> {
+    if services().globals.config.email_verification.is_some() {
+        body.sender_user = Some(
+            UserId::parse_with_server_name("", services().globals.server_name())
+                .expect("we know this is valid"),
+        );
+        body.sender_device = Some("".into());
+    }
+
     let sender_user = body
         .sender_user
         .as_ref()
-        // In the future password changes could be performed with UIA with 3PIDs, but we don't support that currently
         .ok_or_else(|| Error::BadRequest(ErrorKind::MissingToken, "Missing access token."))?;
     let sender_device = body.sender_device.as_ref().expect("user is authenticated");
 
+    let mut flows = vec![AuthFlow::new(vec![AuthType::Password])];
+    if services().globals.config.email_verification.is_some() {
+        flows.push(AuthFlow::new(vec![AuthType::EmailIdentity]));
+    }
+
     let mut uiaainfo = UiaaInfo {
-        flows: vec![AuthFlow {
-            stages: vec![AuthType::Password],
-        }],
+        flows,
         completed: Vec::new(),
         params: Default::default(),
         session: None,
@@ -413,10 +444,13 @@ pub async fn deactivate_route(
         .ok_or_else(|| Error::BadRequest(ErrorKind::MissingToken, "Missing access token."))?;
     let sender_device = body.sender_device.as_ref().expect("user is authenticated");
 
+    let mut flows = vec![AuthFlow::new(vec![AuthType::Password])];
+    if services().globals.config.email_verification.is_some() {
+        flows.push(AuthFlow::new(vec![AuthType::EmailIdentity]));
+    }
+
     let mut uiaainfo = UiaaInfo {
-        flows: vec![AuthFlow {
-            stages: vec![AuthType::Password],
-        }],
+        flows,
         completed: Vec::new(),
         params: Default::default(),
         session: None,
@@ -465,12 +499,176 @@ pub async fn deactivate_route(
 /// Get a list of third party identifiers associated with this account.
 ///
 /// - Currently always returns empty list
-pub async fn third_party_route(
+pub async fn get_3pids_route(
     body: Ruma<get_3pids::v3::Request>,
 ) -> Result<get_3pids::v3::Response> {
-    let _sender_user = body.sender_user.as_ref().expect("user is authenticated");
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+    let threepids = services().threepid.get_threepids(sender_user)?;
 
-    Ok(get_3pids::v3::Response::new(Vec::new()))
+    threepids
+        .collect::<Result<_>>()
+        .map(get_3pids::v3::Response::new)
+}
+
+pub async fn add_3pid_route(body: Ruma<add_3pid::v3::Request>) -> Result<add_3pid::v3::Response> {
+    if services()
+        .threepid
+        .find_validated_token(&body.client_secret, &body.sid)?
+        .and_then(|t| {
+            services()
+                .threepid
+                .user_from_threepid(t.medium, &t.address)
+                .transpose()
+        })
+        .transpose()?
+        .is_some()
+    {
+        Err(Error::BadRequest(
+            ErrorKind::ThreepidInUse,
+            "Email is already in use.",
+        ))
+    } else {
+        let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+        let sender_device = body.sender_device.as_ref().expect("user is authenticated");
+
+        let mut uiaainfo = UiaaInfo {
+            flows: vec![AuthFlow {
+                stages: vec![AuthType::Password],
+            }],
+            completed: Vec::new(),
+            params: Default::default(),
+            session: None,
+            auth_error: None,
+        };
+
+        if let Some(auth) = &body.auth {
+            let (worked, uiaainfo) =
+                services()
+                    .uiaa
+                    .try_auth(sender_user, sender_device, auth, &uiaainfo)?;
+            if !worked {
+                return Err(Error::Uiaa(uiaainfo));
+            }
+        // Success!
+        } else if let Some(json) = body.json_body {
+            uiaainfo.session = Some(utils::random_string(SESSION_ID_LENGTH));
+            services()
+                .uiaa
+                .create(sender_user, sender_device, &uiaainfo, &json)?;
+            return Err(Error::Uiaa(uiaainfo));
+        } else {
+            return Err(Error::BadRequest(ErrorKind::NotJson, "Not json."));
+        }
+
+        if let Some(threepid) = services()
+            .threepid
+            .find_validated_token(&body.client_secret, &body.sid)?
+        {
+            services()
+                .threepid
+                .add_threepid(sender_user, &threepid)
+                .map(|_| add_3pid::v3::Response::new())
+        } else {
+            Err(Error::BadRequest(
+                ErrorKind::ThreepidAuthFailed,
+                "No valid token has been submitted yet.",
+            ))
+        }
+    }
+}
+
+pub async fn delete_3pid_route(
+    body: Ruma<delete_3pid::v3::Request>,
+) -> Result<delete_3pid::v3::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    let Some(true) = services()
+        .threepid
+        .user_from_threepid(body.medium.clone(), &body.address)?
+        .as_ref()
+        .map(|other| other == sender_user)
+    else {
+        return Err(Error::BadRequest(
+            ErrorKind::ThreepidNotFound,
+            "Third-party identifier does not belong to this user.",
+        ));
+    };
+
+    services()
+        .threepid
+        .remove_threepid(sender_user, body.medium.clone(), &body.address)
+        .map(|_| delete_3pid::v3::Response {
+            id_server_unbind_result: ThirdPartyIdRemovalStatus::NoSupport,
+        })
+}
+
+async fn request_3pid_token_helper(
+    TokenRequest {
+        client_secret,
+        medium,
+        address,
+        send_attempt,
+    }: TokenRequest,
+    path: &str,
+) -> Result<(OwnedSessionId, Option<String>)> {
+    let (session_id, token, send_verification) =
+        services()
+            .threepid
+            .request_token(&client_secret, send_attempt, medium, address)?;
+    let access_token = threepid::MAGIC_ACCESS_TOKEN.to_owned();
+
+    let mut submit_url: Url = services()
+        .globals
+        .well_known_client()
+        .parse()
+        .map_err(|_| Error::bad_config("Invalid well_known_client in configuration."))?;
+    submit_url.set_path(path);
+    submit_url.set_query(Some(&format!(
+        "sid={session_id}&token={token}&client_secret={client_secret}&access_token={access_token}"
+    )));
+
+    if send_verification {
+        let smtp = services()
+            .globals
+            .config
+            .email_verification
+            .as_ref()
+            .unwrap();
+
+        // let config = ClientConfig::builder()
+        //     .with_root_certificates(RootCertStore::empty())
+        //     .with_no_client_auth();
+        // let connector = TlsConnector::from(Arc::new(config));
+
+        let stream = TcpStream::connect(SocketAddr::from_str(&smtp.address).unwrap())
+            .await
+            .unwrap();
+        // let stream = connector
+        //     .connect(smtp.address.ip().to_string(), stream)
+        //     .await
+        //     .unwrap();
+
+        let client = SmtpClient::new().smtp_utf8(true);
+        let mut transport = SmtpTransport::new(client, BufStream::new(stream))
+            .await
+            .unwrap();
+
+        let email = SendableEmail::new(
+            Envelope::new(
+                Some("user@localhost".parse().unwrap()),
+                vec!["root@localhost".parse().unwrap()],
+            )
+            .unwrap(),
+            format!(
+                "Subject: {}\r\nContent-Type: text/plain\r\n\r\n{}",
+                "Matrix verification code",
+                format!("Click here: {submit_url}",)
+            ),
+        );
+        transport.send(email).await.unwrap();
+    }
+
+    Ok((session_id.parse().expect(""), None))
 }
 
 /// # `POST /_matrix/client/v3/account/3pid/email/requestToken`
@@ -478,13 +676,142 @@ pub async fn third_party_route(
 /// "This API should be used to request validation tokens when adding an email address to an account"
 ///
 /// - 403 signals that The homeserver does not allow the third party identifier as a contact option.
+pub async fn request_registration_token_via_email_route(
+    body: Ruma<request_registration_token_via_email::v3::Request>,
+) -> Result<request_registration_token_via_email::v3::Response> {
+    if services()
+        .threepid
+        .user_from_threepid(Medium::Email, &body.email)?
+        .is_some()
+    {
+        Err(Error::BadRequest(
+            ErrorKind::ThreepidInUse,
+            "Email is already in use.",
+        ))
+    } else {
+        let (sid, submit_url) = request_3pid_token_helper(
+            body.body.into(),
+            "_matrix/client/unstable/register/email/submitToken",
+        )
+        .await?;
+        Ok(request_registration_token_via_email::v3::Response { sid, submit_url })
+    }
+}
+
 pub async fn request_3pid_management_token_via_email_route(
-    _body: Ruma<request_3pid_management_token_via_email::v3::Request>,
+    body: Ruma<request_3pid_management_token_via_email::v3::Request>,
 ) -> Result<request_3pid_management_token_via_email::v3::Response> {
-    Err(Error::BadRequest(
-        ErrorKind::ThreepidDenied,
-        "Third party identifiers are currently unsupported by this server implementation",
-    ))
+    if services()
+        .threepid
+        .user_from_threepid(Medium::Email, &body.email)?
+        .is_some()
+    {
+        Err(Error::BadRequest(
+            ErrorKind::ThreepidInUse,
+            "Email is already in use.",
+        ))
+    } else {
+        let (sid, submit_url) = request_3pid_token_helper(
+            body.body.into(),
+            "_matrix/client/unstable/3pid/email/submitToken",
+        )
+        .await?;
+        Ok(request_3pid_management_token_via_email::v3::Response { sid, submit_url })
+    }
+}
+
+pub async fn request_password_change_token_via_email_route(
+    body: Ruma<request_password_change_token_via_email::v3::Request>,
+) -> Result<request_password_change_token_via_email::v3::Response> {
+    let (sid, submit_url) = request_3pid_token_helper(
+        body.body.into(),
+        "_matrix/client/unstable/password/email/submitToken",
+    )
+    .await?;
+    Ok(request_password_change_token_via_email::v3::Response { sid, submit_url })
+}
+
+pub async fn submit_registration_token_via_email_route(
+    body: Ruma<validate_email_by_end_user::v2::Request>,
+) -> Result<RumaResponse<validate_email_by_end_user::v2::Response>> {
+    if services()
+        .threepid
+        .find_validated_token(&body.client_secret, &body.sid)?
+        .and_then(|t| {
+            services()
+                .threepid
+                .user_from_threepid(t.medium, &t.address)
+                .transpose()
+        })
+        .transpose()?
+        .is_some()
+    {
+        Err(Error::BadRequest(
+            ErrorKind::ThreepidInUse,
+            "Email is already in use.",
+        ))
+    } else if services()
+        .threepid
+        .validate_token(&body.client_secret, &body.sid, &body.token)?
+        .is_some()
+    {
+        Ok(validate_email_by_end_user::v2::Response::new()).map(RumaResponse)
+    } else {
+        Err(Error::BadRequest(
+            ErrorKind::ThreepidAuthFailed,
+            "Invalid token.",
+        ))
+    }
+}
+
+pub async fn submit_3pid_management_token_via_email_route(
+    body: Ruma<validate_email_by_end_user::v2::Request>,
+) -> Result<RumaResponse<validate_email_by_end_user::v2::Response>> {
+    if services()
+        .threepid
+        .find_validated_token(&body.client_secret, &body.sid)?
+        .and_then(|t| {
+            services()
+                .threepid
+                .user_from_threepid(t.medium, &t.address)
+                .transpose()
+        })
+        .transpose()?
+        .is_some()
+    {
+        Err(Error::BadRequest(
+            ErrorKind::ThreepidInUse,
+            "Email is already in use.",
+        ))
+    } else if services()
+        .threepid
+        .validate_token(&body.client_secret, &body.sid, &body.token)?
+        .is_some()
+    {
+        Ok(validate_email_by_end_user::v2::Response::new()).map(RumaResponse)
+    } else {
+        Err(Error::BadRequest(
+            ErrorKind::ThreepidAuthFailed,
+            "Invalid token.",
+        ))
+    }
+}
+
+pub async fn submit_password_change_token_via_email_route(
+    body: Ruma<validate_email_by_end_user::v2::Request>,
+) -> Result<RumaResponse<validate_email_by_end_user::v2::Response>> {
+    if services()
+        .threepid
+        .validate_token(&body.client_secret, &body.sid, &body.token)?
+        .is_some()
+    {
+        Ok(validate_email_by_end_user::v2::Response::new()).map(RumaResponse)
+    } else {
+        Err(Error::BadRequest(
+            ErrorKind::ThreepidAuthFailed,
+            "Invalid token.",
+        ))
+    }
 }
 
 /// # `POST /_matrix/client/v3/account/3pid/msisdn/requestToken`
@@ -492,11 +819,89 @@ pub async fn request_3pid_management_token_via_email_route(
 /// "This API should be used to request validation tokens when adding an phone number to an account"
 ///
 /// - 403 signals that The homeserver does not allow the third party identifier as a contact option.
+pub async fn request_registration_token_via_msisdn_route(
+    _: Ruma<request_registration_token_via_msisdn::v3::Request>,
+) -> Result<request_registration_token_via_msisdn::v3::Response> {
+    Err(Error::BadRequest(
+        ErrorKind::ThreepidDenied,
+        "Third party MSISDNs are currently unsupported by this server implementation",
+    ))
+}
+
 pub async fn request_3pid_management_token_via_msisdn_route(
-    _body: Ruma<request_3pid_management_token_via_msisdn::v3::Request>,
+    _: Ruma<request_3pid_management_token_via_msisdn::v3::Request>,
 ) -> Result<request_3pid_management_token_via_msisdn::v3::Response> {
     Err(Error::BadRequest(
         ErrorKind::ThreepidDenied,
-        "Third party identifiers are currently unsupported by this server implementation",
+        "Third party MSISDNs are currently unsupported by this server implementation",
     ))
 }
+pub async fn request_password_change_token_via_msisdn_route(
+    _: Ruma<request_password_change_token_via_msisdn::v3::Request>,
+) -> Result<request_password_change_token_via_msisdn::v3::Response> {
+    Err(Error::BadRequest(
+        ErrorKind::ThreepidDenied,
+        "Third party MSISDNs are currently unsupported by this server implementation",
+    ))
+}
+
+pub struct TokenRequest {
+    client_secret: OwnedClientSecret,
+    medium: Medium,
+    address: String,
+    send_attempt: UInt,
+}
+
+impl From<request_registration_token_via_email::v3::Request> for TokenRequest {
+    fn from(
+        request_registration_token_via_email::v3::Request {
+            client_secret,
+            email: address,
+            send_attempt,
+            ..
+        }: request_registration_token_via_email::v3::Request,
+    ) -> Self {
+        Self {
+            client_secret,
+            medium: Medium::Email,
+            address,
+            send_attempt,
+        }
+    }
+}
+
+impl From<request_3pid_management_token_via_email::v3::Request> for TokenRequest {
+    fn from(
+        request_3pid_management_token_via_email::v3::Request {
+            client_secret,
+            email: address,
+            send_attempt,
+            ..
+        }: request_3pid_management_token_via_email::v3::Request,
+    ) -> Self {
+        Self {
+            client_secret,
+            medium: Medium::Email,
+            address,
+            send_attempt,
+        }
+    }
+}
+
+impl From<request_password_change_token_via_email::v3::Request> for TokenRequest {
+    fn from(
+        request_password_change_token_via_email::v3::Request {
+            client_secret,
+            email: address,
+            send_attempt,
+            ..
+        }: request_password_change_token_via_email::v3::Request,
+    ) -> Self {
+        Self {
+            client_secret,
+            medium: Medium::Email,
+            address,
+            send_attempt,
+        }
+    }
+}

src/config/mod.rs

@@ -46,9 +46,13 @@ pub struct Config {
     pub max_fetch_prev_events: u16,
     #[serde(default = "false_fn")]
     pub allow_registration: bool,
+    #[serde(default)]
+    pub email_verification: Option<SmtpConfig>,
     pub registration_token: Option<String>,
     #[serde(default = "default_openid_token_ttl")]
     pub openid_token_ttl: u64,
+    #[serde(default = "default_threepid_token_ttl")]
+    pub threepid_token_ttl: u64,
     #[serde(default = "true_fn")]
     pub allow_encryption: bool,
     #[serde(default = "false_fn")]
@@ -101,6 +105,11 @@ pub struct WellKnownConfig {
     pub server: Option<OwnedServerName>,
 }
 
+#[derive(Clone, Debug, Deserialize)]
+pub struct SmtpConfig {
+    pub address: String,
+}
+
 const DEPRECATED_KEYS: &[&str] = &["cache_capacity"];
 
 impl Config {
@@ -308,6 +317,10 @@ fn default_openid_token_ttl() -> u64 {
     60 * 60
 }
 
+fn default_threepid_token_ttl() -> u64 {
+    60 * 15
+}
+
 // I know, it's a great name
 pub fn default_default_room_version() -> RoomVersionId {
     RoomVersionId::V10

src/database/key_value/mod.rs

@@ -8,6 +8,7 @@ mod media;
 mod pusher;
 mod rooms;
 mod sending;
+mod threepid;
 mod transaction_ids;
 mod uiaa;
 mod users;

src/database/key_value/threepid.rs

@@ -0,0 +1,247 @@
+use std::str::FromStr;
+
+use ruma::{
+    thirdparty::{Medium, ThirdPartyIdentifier, ThirdPartyIdentifierInit},
+    ClientSecret, MilliSecondsSinceUnixEpoch, OwnedUserId, SessionId, UInt, UserId,
+};
+
+use crate::{
+    api::client_server::{SESSION_ID_LENGTH, TOKEN_LENGTH},
+    service, services, utils, Error, KeyValueDatabase, Result,
+};
+
+impl service::threepid::Data for KeyValueDatabase {
+    fn get_threepids<'a>(
+        &'a self,
+        user_id: &UserId,
+    ) -> Result<Box<dyn Iterator<Item = Result<ThirdPartyIdentifier>> + 'a>> {
+        let mut prefix = user_id.as_bytes().to_vec();
+        prefix.push(0xff);
+
+        Ok(Box::new(
+            self.userthreepid_metadata
+                .scan_prefix(prefix)
+                .map(|(_, json)| {
+                    serde_json::from_slice::<ThirdPartyIdentifier>(&json).map_err(|_| {
+                        Error::bad_database(
+                            "ThirdPartyIdentifier in userid_threepids is invalid JSON.",
+                        )
+                    })
+                }),
+        ))
+    }
+
+    fn user_from_threepid(&self, medium: Medium, address: &str) -> Result<Option<OwnedUserId>> {
+        let mut key = medium.as_str().as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(address.as_bytes());
+
+        let Some(v) = self.threepid_userid.get(&key)? else {
+            return Ok(None);
+        };
+
+        Some(
+            OwnedUserId::from_str(utils::string_from_bytes(&v).as_deref().map_err(|_| {
+                Error::bad_database("provider in userid_providersubjectid is invalid unicode.")
+            })?)
+            .map_err(|_| Error::bad_database("provider in userid_providersubjectid is invalid.")),
+        )
+        .transpose()
+    }
+
+    fn add_threepid(&self, user_id: &UserId, threepid: &ThirdPartyIdentifier) -> Result<()> {
+        tracing::warn!(
+            "adding third-party identifier {} for {}",
+            &threepid.address,
+            user_id,
+        );
+
+        let mut key = user_id.as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(threepid.medium.as_str().as_bytes());
+        key.push(0xff);
+        key.extend_from_slice(threepid.address.as_bytes());
+
+        let value = serde_json::to_vec(&threepid).expect("");
+
+        self.userthreepid_metadata.insert(&key, &value)?;
+
+        let mut key = threepid.medium.as_str().as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(threepid.address.as_bytes());
+
+        let value = user_id.as_bytes();
+
+        self.threepid_userid.insert(&key, value)
+    }
+
+    fn remove_threepid(&self, user_id: &UserId, medium: Medium, address: &str) -> Result<()> {
+        let mut key = user_id.as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(medium.as_str().as_bytes());
+        key.push(0xff);
+        key.extend_from_slice(address.as_bytes());
+
+        self.userthreepid_metadata.remove(&key)?;
+
+        let mut key = medium.as_str().as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(address.as_bytes());
+
+        self.threepid_userid.remove(&key)
+    }
+
+    fn request_token(
+        &self,
+        client_secret: &ClientSecret,
+        send_attempt: UInt,
+        medium: Medium,
+        address: String,
+    ) -> Result<(String, String, bool)> {
+        let mut session_id = utils::random_string(SESSION_ID_LENGTH);
+
+        let mut expires_at = utils::millis_since_unix_epoch()
+            .checked_add(services().globals.config.threepid_token_ttl * 1000)
+            .expect("time overflow");
+        let mut last_attempt = u64::default();
+        let mut token = utils::random_string(TOKEN_LENGTH);
+        let mut threepid = ThirdPartyIdentifierInit {
+            address,
+            medium,
+            validated_at: MilliSecondsSinceUnixEpoch(UInt::default()),
+            added_at: MilliSecondsSinceUnixEpoch::now(),
+        }
+        .into();
+
+        let prefix = client_secret.as_bytes().to_vec();
+        if let Some((key, value)) = self
+            .clientsecretsessionid_session
+            .scan_prefix(prefix)
+            .next()
+        {
+            session_id =
+                utils::string_from_bytes(&key[client_secret.as_bytes().len()..]).expect("");
+
+            let (v, rem) = value.split_at(std::mem::size_of::<u64>());
+            expires_at = u64::from_be_bytes(v.try_into().expect(""));
+
+            let (v, rem) = rem.split_at(std::mem::size_of::<u64>());
+            last_attempt = u64::from_be_bytes(v.try_into().expect(""));
+
+            let (v, rem) = rem.split_at(TOKEN_LENGTH);
+            token = utils::string_from_bytes(v).map_err(|_| {
+                Error::bad_database("token in clientsecretsessionid_session is invalid unicode.")
+            })?;
+
+            threepid = serde_json::from_slice::<ThirdPartyIdentifier>(rem).map_err(|_| {
+                Error::bad_database(
+                    "ThirdPartyIdentifier in clientsecretsessionid_session is invalid JSON.",
+                )
+            })?;
+
+            tracing::warn!(
+                "updated registration token for {}, (attempt {last_attempt}): {token}",
+                threepid.address
+            );
+        };
+
+        let mut key = client_secret.as_bytes().to_vec();
+        key.extend_from_slice(session_id.as_bytes());
+
+        let mut value = expires_at.to_be_bytes().to_vec();
+        value.extend_from_slice(&last_attempt.max(send_attempt.into()).to_be_bytes());
+        value.extend_from_slice(token.as_bytes());
+        value.extend_from_slice(&serde_json::to_vec(&threepid).expect(""));
+
+        self.clientsecretsessionid_session.insert(&key, &value)?;
+        Ok((session_id, token, last_attempt < send_attempt.into()))
+    }
+
+    fn validate_token(
+        &self,
+        client_secret: &ClientSecret,
+        session_id: &SessionId,
+        token: &str,
+    ) -> Result<Option<ThirdPartyIdentifier>> {
+        let mut key = client_secret.as_bytes().to_vec();
+        key.extend_from_slice(session_id.as_bytes());
+
+        let Some(value) = self.clientsecretsessionid_session.get(&key)? else {
+            tracing::warn!(
+                "unrecognized third-party credentials for client secret {client_secret}"
+            );
+
+            return Ok(None);
+        };
+
+        let (v, rem) = value.split_at(std::mem::size_of::<u64>());
+        let (_, rem) = rem.split_at(std::mem::size_of::<u64>());
+        let expires_at = u64::from_be_bytes(v.try_into().expect(""));
+
+        let (v, rem) = rem.split_at(TOKEN_LENGTH);
+        let expected_token = utils::string_from_bytes(v).map_err(|_| {
+            Error::bad_database("token in clientsecretsessionid_session is invalid unicode.")
+        })?;
+
+        let mut threepid = serde_json::from_slice::<ThirdPartyIdentifier>(rem).map_err(|_| {
+            Error::bad_database(
+                "ThirdPartyIdentifier in clientsecretsessionid_session is invalid JSON.",
+            )
+        })?;
+
+        if token != expected_token || expires_at < utils::millis_since_unix_epoch() {
+            tracing::warn!("invalid or expired token for client secret {client_secret}: {token} != {expected_token}");
+
+            return Ok(None);
+        } else if threepid.validated_at == MilliSecondsSinceUnixEpoch(UInt::default()) {
+            tracing::warn!(
+                "successfully validated third-party identifier for client secret {client_secret}"
+            );
+
+            threepid.validated_at = MilliSecondsSinceUnixEpoch::now();
+        }
+
+        let mut value = expires_at.to_be_bytes().to_vec();
+        value.extend_from_slice(&u64::default().to_be_bytes());
+        value.extend_from_slice(token.as_bytes());
+        value.extend_from_slice(&serde_json::to_vec(&threepid).expect(""));
+
+        self.clientsecretsessionid_session.insert(&key, &value)?;
+        Ok(Some(threepid))
+    }
+
+    fn find_validated_token(
+        &self,
+        client_secret: &ClientSecret,
+        session_id: &SessionId,
+    ) -> Result<Option<ThirdPartyIdentifier>> {
+        let mut key = client_secret.as_bytes().to_vec();
+        key.extend_from_slice(session_id.as_bytes());
+
+        let Some(value) = self.clientsecretsessionid_session.get(&key)? else {
+            tracing::warn!(
+                "unrecognized third-party credentials for client secret {client_secret}"
+            );
+
+            return Ok(None);
+        };
+
+        let (_, rem) = value.split_at(std::mem::size_of::<u64>() * 2 + TOKEN_LENGTH);
+
+        let threepid = serde_json::from_slice::<ThirdPartyIdentifier>(rem).map_err(|_| {
+            Error::bad_database(
+                "ThirdPartyIdentifier in clientsecretsessionid_session is invalid JSON.",
+            )
+        })?;
+
+        if threepid.validated_at == MilliSecondsSinceUnixEpoch(UInt::default()) {
+            tracing::warn!(
+                "third-party identifier for client secret {client_secret} has not been validated yet"
+            );
+
+            return Ok(None);
+        }
+
+        Ok(Some(threepid))
+    }
+}

src/database/key_value/users.rs

@@ -13,7 +13,7 @@ use tracing::warn;
 use crate::{
     api::client_server::TOKEN_LENGTH,
     database::KeyValueDatabase,
-    service::{self, users::clean_signatures},
+    service::{self, threepid, users::clean_signatures},
     services, utils, Error, Result,
 };
 
@@ -42,6 +42,14 @@ impl service::users::Data for KeyValueDatabase {
 
     /// Find out which user an access token belongs to.
     fn find_from_token(&self, token: &str) -> Result<Option<(OwnedUserId, String)>> {
+        if token == threepid::MAGIC_ACCESS_TOKEN {
+            return Ok(Some((
+                UserId::parse_with_server_name("", &services().globals.config.server_name)
+                    .expect("we know this is valid"),
+                String::default(),
+            )));
+        }
+
         self.token_userdeviceid
             .get(token.as_bytes())?
             .map_or(Ok(None), |bytes| {

src/database/mod.rs

@@ -50,6 +50,9 @@ pub struct KeyValueDatabase {
     pub(super) userdeviceid_metadata: Arc<dyn KvTree>, // This is also used to check if a device exists
     pub(super) userid_devicelistversion: Arc<dyn KvTree>, // DevicelistVersion = u64
     pub(super) token_userdeviceid: Arc<dyn KvTree>,
+    pub(super) userthreepid_metadata: Arc<dyn KvTree>,
+    pub(super) threepid_userid: Arc<dyn KvTree>,
+    pub(super) clientsecretsessionid_session: Arc<dyn KvTree>,
 
     pub(super) onetimekeyid_onetimekeys: Arc<dyn KvTree>, // OneTimeKeyId = UserId + DeviceKeyId
     pub(super) userid_lastonetimekeyupdate: Arc<dyn KvTree>, // LastOneTimeKeyUpdate = Count
@@ -287,6 +290,11 @@ impl KeyValueDatabase {
             userdeviceid_metadata: builder.open_tree("userdeviceid_metadata")?,
             userid_devicelistversion: builder.open_tree("userid_devicelistversion")?,
             token_userdeviceid: builder.open_tree("token_userdeviceid")?,
+
+            userthreepid_metadata: builder.open_tree("userid_threepids")?,
+            threepid_userid: builder.open_tree("threepid_userid")?,
+            clientsecretsessionid_session: builder.open_tree("clientsecretsessionid_session")?,
+
             onetimekeyid_onetimekeys: builder.open_tree("onetimekeyid_onetimekeys")?,
             userid_lastonetimekeyupdate: builder.open_tree("userid_lastonetimekeyupdate")?,
             keychangeid_userid: builder.open_tree("keychangeid_userid")?,

src/main.rs

@@ -289,9 +289,15 @@ fn routes(config: &Config) -> Router {
         .ruma_route(client_server::logout_all_route)
         .ruma_route(client_server::change_password_route)
         .ruma_route(client_server::deactivate_route)
-        .ruma_route(client_server::third_party_route)
+        .ruma_route(client_server::get_3pids_route)
+        .ruma_route(client_server::add_3pid_route)
+        .ruma_route(client_server::delete_3pid_route)
+        .ruma_route(client_server::request_registration_token_via_email_route)
         .ruma_route(client_server::request_3pid_management_token_via_email_route)
+        .ruma_route(client_server::request_password_change_token_via_email_route)
+        .ruma_route(client_server::request_registration_token_via_msisdn_route)
         .ruma_route(client_server::request_3pid_management_token_via_msisdn_route)
+        .ruma_route(client_server::request_password_change_token_via_msisdn_route)
         .ruma_route(client_server::get_capabilities_route)
         .ruma_route(client_server::get_pushrules_all_route)
         .ruma_route(client_server::set_pushrule_route)
@@ -364,6 +370,20 @@ fn routes(config: &Config) -> Router {
         .ruma_route(client_server::send_state_event_for_key_route)
         .ruma_route(client_server::get_state_events_route)
         .ruma_route(client_server::get_state_events_for_key_route)
+        // The specification does not define endpoints for token submission, as a workaround
+        // we use custom endpoints which are invoked via out-of-bound verification
+        .route(
+            "/_matrix/client/unstable/register/email/submitToken",
+            get(client_server::submit_registration_token_via_email_route),
+        )
+        .route(
+            "/_matrix/client/unstable/3pid/email/submitToken",
+            get(client_server::submit_3pid_management_token_via_email_route),
+        )
+        .route(
+            "/_matrix/client/unstable/password/email/submitToken",
+            get(client_server::submit_password_change_token_via_email_route),
+        )
         // Ruma doesn't have support for multiple paths for a single endpoint yet, and these routes
         // share one Ruma request / response type pair with {get,send}_state_event_for_key_route
         .route(

src/service/mod.rs

@@ -19,6 +19,7 @@ pub mod pdu;
 pub mod pusher;
 pub mod rooms;
 pub mod sending;
+pub mod threepid;
 pub mod transaction_ids;
 pub mod uiaa;
 pub mod users;
@@ -36,6 +37,7 @@ pub struct Services {
     pub key_backups: key_backups::Service,
     pub media: media::Service,
     pub sending: Arc<sending::Service>,
+    pub threepid: threepid::Service,
 }
 
 impl Services {
@@ -51,6 +53,7 @@ impl Services {
             + key_backups::Data
             + media::Data
             + sending::Data
+            + threepid::Data
             + 'static,
     >(
         db: &'static D,
@@ -110,6 +113,7 @@ impl Services {
                 user: rooms::user::Service { db },
             },
             transaction_ids: transaction_ids::Service { db },
+            threepid: threepid::Service { db },
             uiaa: uiaa::Service { db },
             users: users::Service {
                 db,

src/service/threepid/data.rs

@@ -0,0 +1,40 @@
+use ruma::{
+    thirdparty::{Medium, ThirdPartyIdentifier},
+    ClientSecret, OwnedUserId, SessionId, UInt, UserId,
+};
+
+use crate::Result;
+
+pub trait Data: Send + Sync {
+    fn get_threepids<'a>(
+        &'a self,
+        user_id: &UserId,
+    ) -> Result<Box<dyn Iterator<Item = Result<ThirdPartyIdentifier>> + 'a>>;
+
+    fn user_from_threepid(&self, medium: Medium, address: &str) -> Result<Option<OwnedUserId>>;
+
+    fn add_threepid(&self, user_id: &UserId, threepid: &ThirdPartyIdentifier) -> Result<()>;
+
+    fn remove_threepid(&self, user_id: &UserId, medium: Medium, address: &str) -> Result<()>;
+
+    fn request_token(
+        &self,
+        client_secret: &ClientSecret,
+        send_attempt: UInt,
+        medium: Medium,
+        address: String,
+    ) -> Result<(String, String, bool)>;
+
+    fn validate_token(
+        &self,
+        client_secret: &ClientSecret,
+        session_id: &SessionId,
+        token: &str,
+    ) -> Result<Option<ThirdPartyIdentifier>>;
+
+    fn find_validated_token(
+        &self,
+        client_secret: &ClientSecret,
+        session_id: &SessionId,
+    ) -> Result<Option<ThirdPartyIdentifier>>;
+}

src/service/threepid/mod.rs

@@ -0,0 +1,64 @@
+use ruma::{
+    thirdparty::{Medium, ThirdPartyIdentifier},
+    ClientSecret, OwnedUserId, SessionId, UInt, UserId,
+};
+
+use crate::Result;
+
+mod data;
+pub use data::Data;
+
+pub struct Service {
+    pub db: &'static dyn Data,
+}
+
+pub const MAGIC_ACCESS_TOKEN: &'static str = "THREEPID";
+
+impl Service {
+    pub fn get_threepids<'a>(
+        &'a self,
+        user_id: &UserId,
+    ) -> Result<Box<dyn Iterator<Item = Result<ThirdPartyIdentifier>> + 'a>> {
+        self.db.get_threepids(user_id)
+    }
+
+    pub fn user_from_threepid(&self, medium: Medium, address: &str) -> Result<Option<OwnedUserId>> {
+        self.db.user_from_threepid(medium, address)
+    }
+
+    pub fn add_threepid(&self, user_id: &UserId, threepid: &ThirdPartyIdentifier) -> Result<()> {
+        self.db.add_threepid(user_id, threepid)
+    }
+
+    pub fn remove_threepid(&self, user_id: &UserId, medium: Medium, address: &str) -> Result<()> {
+        self.db.remove_threepid(user_id, medium, address)
+    }
+
+    pub fn request_token(
+        &self,
+        client_secret: &ClientSecret,
+        send_attempt: UInt,
+        medium: Medium,
+        address: String,
+    ) -> Result<(String, String, bool)> {
+        self.db
+            .request_token(client_secret, send_attempt, medium, address)
+    }
+
+    pub fn validate_token(
+        &self,
+        client_secret: &ClientSecret,
+        session_id: &SessionId,
+        token: &str,
+    ) -> Result<Option<ThirdPartyIdentifier>> {
+        self.db.validate_token(client_secret, session_id, token)
+    }
+
+    pub fn find_validated_token(
+        &self,
+        client_secret: &ClientSecret,
+        session_id: &SessionId,
+    ) -> Result<Option<ThirdPartyIdentifier>> {
+        self.db.find_validated_token(client_secret, session_id)
+    }
+}

src/service/uiaa/mod.rs

@@ -5,7 +5,10 @@ pub use data::Data;
 use ruma::{
     api::client::{
         error::ErrorKind,
-        uiaa::{AuthData, AuthType, Password, UiaaInfo, UserIdentifier},
+        uiaa::{
+            AuthData, AuthType, EmailIdentity, Password, ThirdpartyIdCredentials, UiaaInfo,
+            UserIdentifier,
+        },
     },
     CanonicalJsonValue, DeviceId, UserId,
 };
@@ -107,6 +110,29 @@ impl Service {
                     return Ok((false, uiaainfo));
                 }
             }
+            AuthData::EmailIdentity(EmailIdentity {
+                thirdparty_id_creds:
+                    ThirdpartyIdCredentials {
+                        sid: session_id,
+                        client_secret,
+                        ..
+                    },
+                ..
+            }) => {
+                if !services()
+                    .threepid
+                    .find_validated_token(client_secret, session_id)?
+                    .is_some()
+                {
+                    uiaainfo.auth_error = Some(ruma::api::client::error::StandardErrorBody {
+                        kind: ErrorKind::ThreepidAuthFailed,
+                        message: "No valid token has been submitted yet.".to_owned(),
+                    });
+                    return Ok((false, uiaainfo));
+                };
+
+                uiaainfo.completed.push(AuthType::EmailIdentity);
+            }
             AuthData::Dummy(_) => {
                 uiaainfo.completed.push(AuthType::Dummy);
             }