feat(operator): authenticate recovery operators as a distinct peer type

server/crates/arbiter-server/src/peers/operator/auth/mod.rs

@@ -1,4 +1,4 @@
-use super::{Credentials, OperatorConnection};
+use super::{AuthenticatedOperator, OperatorConnection};
 use arbiter_crypto::authn::{self, AuthChallenge};
 use arbiter_proto::transport::Bi;
 
@@ -71,7 +71,7 @@ fn parse_auth_event(payload: Inbound) -> AuthEvents {
 pub async fn authenticate<T>(
     props: &mut OperatorConnection,
     transport: &mut T,
-) -> Result<Credentials, Error>
+) -> Result<AuthenticatedOperator, Error>
 where
     T: Bi<Inbound, Result<Outbound, Error>> + Send + ?Sized,
 {

server/crates/arbiter-server/src/peers/operator/auth/state.rs

@@ -1,12 +1,12 @@
 use super::{
-    super::{Credentials, OperatorConnection},
+    super::{AuthenticatedOperator, Credentials, OperatorConnection, RecoveryCredentials},
     Error,
 };
 use crate::{
     actors::bootstrap::VerifyToken,
     db::{
         DatabasePool,
-        schema::{arbiter_settings, operator_identity},
+        schema::{arbiter_settings, operator_identity, recovery_operator_identity},
     },
     peers::operator::auth::Outbound,
 };
@@ -37,7 +37,7 @@ smlang::statemachine!(
     custom_error: true,
     transitions: {
         *Init + AuthRequest(ChallengeRequest) / async prepare_challenge = SentChallenge(ChallengeContext),
-        SentChallenge(ChallengeContext) + ReceivedSolution(ChallengeSolution) / async verify_solution = AuthOk(Credentials),
+        SentChallenge(ChallengeContext) + ReceivedSolution(ChallengeSolution) / async verify_solution = AuthOk(AuthenticatedOperator),
     }
 );
 
@@ -59,6 +59,27 @@ async fn get_client_id(db: &DatabasePool, pubkey: &authn::PublicKey) -> Result<O
         })
 }
 
+async fn get_recovery_operator_id(
+    db: &DatabasePool,
+    pubkey: &authn::PublicKey,
+) -> Result<Option<i32>, Error> {
+    let mut conn = db.get().await.map_err(|e| {
+        error!(error = ?e, "Database pool error");
+        Error::internal("Database unavailable")
+    })?;
+
+    recovery_operator_identity::table
+        .filter(recovery_operator_identity::public_key.eq(pubkey.to_bytes()))
+        .select(recovery_operator_identity::id)
+        .first::<i32>(&mut conn)
+        .await
+        .optional()
+        .map_err(|e| {
+            error!(error = ?e, "Database error");
+            Error::internal("Database operation failed")
+        })
+}
+
 async fn register_key(db: &DatabasePool, pubkey: &authn::PublicKey) -> Result<i32, Error> {
     let pubkey_bytes = pubkey.to_bytes();
     let mut conn = db.get().await.map_err(|e| {
@@ -118,12 +139,14 @@ where
             bootstrap_token,
         }: ChallengeRequest,
     ) -> Result<ChallengeContext, Self::Error> {
-        // Verify pubkey is registered (unless bootstrapping)
-        if bootstrap_token.is_none() {
-            let id = get_client_id(&self.conn.db, &pubkey).await?;
-            if id.is_none() {
-                return Err(Error::UnregisteredPublicKey);
-            }
+        // Verify pubkey is registered in either identity table (unless bootstrapping)
+        if bootstrap_token.is_none()
+            && get_client_id(&self.conn.db, &pubkey).await?.is_none()
+            && get_recovery_operator_id(&self.conn.db, &pubkey)
+                .await?
+                .is_none()
+        {
+            return Err(Error::UnregisteredPublicKey);
         }
 
         let challenge = AuthChallenge::generate(&mut rand::rng());
@@ -153,7 +176,7 @@ where
             bootstrap_token,
         }: &ChallengeContext,
         ChallengeSolution { solution }: ChallengeSolution,
-    ) -> Result<Credentials, Self::Error> {
+    ) -> Result<AuthenticatedOperator, Self::Error> {
         let signature = authn::Signature::try_from(solution.as_slice()).map_err(|()| {
             error!("Failed to decode signature in challenge solution");
             Error::InvalidChallengeSolution
@@ -169,8 +192,9 @@ where
             return Err(Error::InvalidChallengeSolution);
         }
 
-        // Resolve client id: bootstrap (verify token, then register) or lookup
-        let id = match bootstrap_token {
+        // Resolve the peer's role: bootstrap (verify token, then register as an ordinary
+        // operator) or look the key up in whichever identity table holds it.
+        let authenticated = match bootstrap_token {
             Some(token) => {
                 let token_ok: bool = self
                     .conn
@@ -194,7 +218,7 @@ where
                     return Err(Error::InvalidBootstrapToken);
                 }
 
-                match register_key(&self.conn.db, pubkey).await {
+                let id = match register_key(&self.conn.db, pubkey).await {
                     Ok(id) => id,
                     // `register_key` refuses a token that verified here but lost the race
                     // against bootstrap. Reported to the peer exactly like the refusal above,
@@ -208,11 +232,38 @@ where
                         return Err(Error::InvalidBootstrapToken);
                     }
                     Err(err) => return Err(err),
+                };
+
+                AuthenticatedOperator::Ordinary(Credentials {
+                    id,
+                    pubkey: pubkey.clone(),
+                })
+            }
+            None => {
+                if let Some(id) = get_client_id(&self.conn.db, pubkey).await? {
+                    AuthenticatedOperator::Ordinary(Credentials {
+                        id,
+                        pubkey: pubkey.clone(),
+                    })
+                } else {
+                    // `prepare_challenge` already found the key in one of the tables, so
+                    // arriving here means it was removed mid-handshake. Reported to the peer
+                    // for the same reason `InvalidBootstrapToken` is above: an operator that
+                    // has sent its solution sees a protocol error rather than a handshake that
+                    // stops with nothing on the wire.
+                    let Some(id) = get_recovery_operator_id(&self.conn.db, pubkey).await? else {
+                        self.transport
+                            .send(Err(Error::UnregisteredPublicKey))
+                            .await
+                            .map_err(|_| Error::Transport)?;
+                        return Err(Error::UnregisteredPublicKey);
+                    };
+                    AuthenticatedOperator::Recovery(RecoveryCredentials {
+                        id,
+                        pubkey: pubkey.clone(),
+                    })
                 }
             }
-            None => get_client_id(&self.conn.db, pubkey)
-                .await?
-                .ok_or(Error::UnregisteredPublicKey)?,
         };
 
         self.transport
@@ -220,9 +271,6 @@ where
             .await
             .map_err(|_| Error::Transport)?;
 
-        Ok(Credentials {
-            id,
-            pubkey: pubkey.clone(),
-        })
+        Ok(authenticated)
     }
 }

server/crates/arbiter-server/src/peers/operator/mod.rs

@@ -33,6 +33,37 @@ impl Integrable for Credentials {
     const KIND: &'static str = "operator_credentials";
 }
 
+/// §3.5: recovery operators are a separate peer type with their own identity table. Their
+/// attestation kind differs from an ordinary operator's so the two id spaces cannot collide.
+#[derive(Debug, Clone, Hashable)]
+pub struct RecoveryCredentials {
+    pub id: i32,
+    pub pubkey: authn::PublicKey,
+}
+
+impl Integrable for RecoveryCredentials {
+    const KIND: &'static str = "recovery_operator_credentials";
+}
+
+/// The outcome of an operator handshake. The variant, not a field, decides what the peer may
+/// do — so no call site can pass an unauthenticated recovery id.
+#[derive(Debug, Clone)]
+pub enum AuthenticatedOperator {
+    Ordinary(Credentials),
+    Recovery(RecoveryCredentials),
+}
+
+impl AuthenticatedOperator {
+    /// The peer's id within its own identity table.
+    #[must_use]
+    pub const fn id(&self) -> i32 {
+        match self {
+            Self::Ordinary(credentials) => credentials.id,
+            Self::Recovery(credentials) => credentials.id,
+        }
+    }
+}
+
 // Messages, sent by operator to connection client without having a request
 #[derive(Debug)]
 pub enum OutOfBand {
@@ -168,7 +199,16 @@ where
     T: Bi<auth::Inbound, Result<auth::Outbound, auth::Error>> + Send,
     T: Bi<vault_gate::Inbound, Result<vault_gate::Outbound, vault_gate::Error>> + Send,
 {
-    let creds = authenticate(props, &mut transport).await?;
+    let authenticated = authenticate(props, &mut transport).await?;
+
+    // A recovery operator has no session of its own yet: everything below this point is written
+    // against an ordinary operator's `Credentials`, so the handshake is refused rather than
+    // silently treated as an ordinary one.
+    let AuthenticatedOperator::Ordinary(creds) = authenticated else {
+        return Err(Error::Internal(
+            "recovery operators have no session yet".into(),
+        ));
+    };
 
     // should run vault gate only if sealed / unbootstrapped
     if should_run_gate(&props.actors.vault).await? {

server/crates/arbiter-server/tests/operator/auth.rs

@@ -8,7 +8,9 @@ use arbiter_server::{
     },
     crypto::integrity,
     db::{self, schema},
-    peers::operator::{self, Credentials, OperatorConnection, auth, vault_gate},
+    peers::operator::{
+        self, AuthenticatedOperator, Credentials, OperatorConnection, auth, vault_gate,
+    },
 };
 
 use async_trait::async_trait;
@@ -196,15 +198,29 @@ pub async fn bootstrap_token_auth() {
         .expect("should receive auth result");
     assert!(matches!(response, Ok(auth::Outbound::AuthSuccess)));
 
-    task.await.unwrap().unwrap();
+    let authenticated = task.await.unwrap().unwrap();
 
     let mut conn = db.get().await.unwrap();
-    let stored_pubkey: Vec<u8> = schema::operator_identity::table
-        .select(schema::operator_identity::public_key)
-        .first::<Vec<u8>>(&mut conn)
+    let (stored_id, stored_pubkey): (i32, Vec<u8>) = schema::operator_identity::table
+        .select((
+            schema::operator_identity::id,
+            schema::operator_identity::public_key,
+        ))
+        .first::<(i32, Vec<u8>)>(&mut conn)
         .await
         .unwrap();
     assert_eq!(stored_pubkey, verifying_key(&new_key).encode().0.to_vec());
+
+    // A key registered through the bootstrap token is an ordinary operator, carrying the id its
+    // registration wrote. Asserted here because this is the file's only bootstrap-arm check on
+    // what `authenticate` actually returns.
+    match authenticated {
+        AuthenticatedOperator::Ordinary(creds) => assert_eq!(creds.id, stored_id),
+        AuthenticatedOperator::Recovery(creds) => panic!(
+            "expected the ordinary role, got a recovery operator with id {}",
+            creds.id
+        ),
+    }
 }
 
 /// A multi-operator committee must all register with the same bootstrap token before bootstrap
@@ -758,3 +774,230 @@ pub async fn challenge_auth_rejects_invalid_signature() {
         Err(auth::Error::InvalidChallengeSolution)
     ));
 }
+
+/// §3.5: a recovery operator is a separate peer type. Its key resolves against
+/// `recovery_operator_identity`, and authentication reports the recovery role.
+///
+/// An ordinary operator is registered alongside it so the recovery key is not simply the only
+/// key on file: the handshake has to reach the recovery table while `operator_identity` is
+/// populated. Both tables autoincrement from 1, so the fixture also pushes the authenticating
+/// recovery operator to id 2 -- with one row in each table an id taken from the wrong table
+/// would still read as 1, and only the variant would be under test.
+#[tokio::test]
+#[test_log::test]
+pub async fn recovery_operator_authenticates_with_its_own_identity() {
+    let db = db::create_test_pool().await;
+    let actors = spawn_actors(db.clone()).await;
+
+    let ordinary_key = MlDsa87::key_gen(&mut rand::rng());
+    let other_recovery_key = MlDsa87::key_gen(&mut rand::rng());
+    let recovery_key = MlDsa87::key_gen(&mut rand::rng());
+    let recovery_pubkey_bytes = authn::PublicKey::from(verifying_key(&recovery_key)).to_bytes();
+
+    let recovery_id: i32 = {
+        let mut conn = db.get().await.unwrap();
+        insert_into(schema::operator_identity::table)
+            .values((schema::operator_identity::public_key
+                .eq(authn::PublicKey::from(verifying_key(&ordinary_key)).to_bytes()),))
+            .execute(&mut conn)
+            .await
+            .unwrap();
+        insert_into(schema::recovery_operator_identity::table)
+            .values((schema::recovery_operator_identity::public_key
+                .eq(authn::PublicKey::from(verifying_key(&other_recovery_key)).to_bytes()),))
+            .execute(&mut conn)
+            .await
+            .unwrap();
+        insert_into(schema::recovery_operator_identity::table)
+            .values((schema::recovery_operator_identity::public_key.eq(recovery_pubkey_bytes),))
+            .returning(schema::recovery_operator_identity::id)
+            .get_result(&mut conn)
+            .await
+            .unwrap()
+    };
+    assert_eq!(
+        recovery_id, 2,
+        "the fixture must give the authenticating recovery operator an id no ordinary \
+         operator holds, or the id assertion below cannot discriminate"
+    );
+
+    let (mut server_transport, mut test_transport) = ChannelTransport::new();
+    let db_for_task = db.clone();
+    let task = tokio::spawn(async move {
+        let mut props = OperatorConnection::new(db_for_task, actors);
+        auth::authenticate(&mut props, &mut server_transport).await
+    });
+
+    test_transport
+        .send(auth::Inbound::AuthChallengeRequest {
+            pubkey: verifying_key(&recovery_key).into(),
+            bootstrap_token: None,
+        })
+        .await
+        .unwrap();
+
+    let response = test_transport
+        .recv()
+        .await
+        .expect("should receive challenge");
+    let challenge = match response {
+        Ok(auth::Outbound::AuthChallenge { challenge }) => challenge,
+        other => panic!("Expected AuthChallenge, got {other:?}"),
+    };
+
+    let signature = sign_operator_challenge(&recovery_key, &challenge);
+    test_transport
+        .send(auth::Inbound::AuthChallengeSolution {
+            signature: signature.to_bytes(),
+        })
+        .await
+        .unwrap();
+
+    let response = test_transport
+        .recv()
+        .await
+        .expect("should receive auth result");
+    assert!(matches!(response, Ok(auth::Outbound::AuthSuccess)));
+
+    let authenticated = task
+        .await
+        .unwrap()
+        .expect("recovery operator should authenticate");
+    match authenticated {
+        AuthenticatedOperator::Recovery(creds) => assert_eq!(creds.id, recovery_id),
+        AuthenticatedOperator::Ordinary(creds) => panic!(
+            "expected the recovery role, got the ordinary operator with id {}",
+            creds.id
+        ),
+    }
+}
+
+/// A key present in neither identity table is still rejected: accepting a key found in either
+/// table must not degrade into accepting any key at all. Both tables hold a row so the refusal
+/// cannot come from an empty lookup.
+#[tokio::test]
+#[test_log::test]
+pub async fn unknown_key_is_rejected_when_both_tables_are_populated() {
+    let db = db::create_test_pool().await;
+    let actors = spawn_actors(db.clone()).await;
+
+    let ordinary_key = MlDsa87::key_gen(&mut rand::rng());
+    let recovery_key = MlDsa87::key_gen(&mut rand::rng());
+    {
+        let mut conn = db.get().await.unwrap();
+        insert_into(schema::operator_identity::table)
+            .values((schema::operator_identity::public_key
+                .eq(authn::PublicKey::from(verifying_key(&ordinary_key)).to_bytes()),))
+            .execute(&mut conn)
+            .await
+            .unwrap();
+        insert_into(schema::recovery_operator_identity::table)
+            .values((schema::recovery_operator_identity::public_key
+                .eq(authn::PublicKey::from(verifying_key(&recovery_key)).to_bytes()),))
+            .execute(&mut conn)
+            .await
+            .unwrap();
+    }
+
+    let (mut server_transport, mut test_transport) = ChannelTransport::new();
+    let db_for_task = db.clone();
+    let task = tokio::spawn(async move {
+        let mut props = OperatorConnection::new(db_for_task, actors);
+        auth::authenticate(&mut props, &mut server_transport).await
+    });
+
+    let unknown_key = MlDsa87::key_gen(&mut rand::rng());
+    test_transport
+        .send(auth::Inbound::AuthChallengeRequest {
+            pubkey: verifying_key(&unknown_key).into(),
+            bootstrap_token: None,
+        })
+        .await
+        .unwrap();
+
+    assert!(matches!(
+        task.await.unwrap(),
+        Err(auth::Error::UnregisteredPublicKey)
+    ));
+}
+
+/// `verify_solution` resolves the recovery id only after the peer has sent its solution, so a
+/// recovery row removed between challenge and solution reaches that refusal. It has to be sent
+/// on the transport, like the `InvalidBootstrapToken` refusals in the arm above: an operator
+/// that has answered the challenge sees a protocol error rather than a handshake that stops
+/// with nothing on the wire.
+#[tokio::test]
+#[test_log::test]
+pub async fn recovery_key_removed_mid_handshake_is_refused_on_the_wire() {
+    let db = db::create_test_pool().await;
+    let actors = spawn_actors(db.clone()).await;
+
+    let recovery_key = MlDsa87::key_gen(&mut rand::rng());
+    let recovery_pubkey_bytes = authn::PublicKey::from(verifying_key(&recovery_key)).to_bytes();
+    {
+        let mut conn = db.get().await.unwrap();
+        insert_into(schema::recovery_operator_identity::table)
+            .values((
+                schema::recovery_operator_identity::public_key.eq(recovery_pubkey_bytes.clone()),
+            ))
+            .execute(&mut conn)
+            .await
+            .unwrap();
+    }
+
+    let (mut server_transport, mut test_transport) = ChannelTransport::new();
+    let db_for_task = db.clone();
+    let task = tokio::spawn(async move {
+        let mut props = OperatorConnection::new(db_for_task, actors);
+        auth::authenticate(&mut props, &mut server_transport).await
+    });
+
+    test_transport
+        .send(auth::Inbound::AuthChallengeRequest {
+            pubkey: verifying_key(&recovery_key).into(),
+            bootstrap_token: None,
+        })
+        .await
+        .unwrap();
+
+    let response = test_transport
+        .recv()
+        .await
+        .expect("should receive challenge");
+    let challenge = match response {
+        Ok(auth::Outbound::AuthChallenge { challenge }) => challenge,
+        other => panic!("Expected AuthChallenge, got {other:?}"),
+    };
+
+    // The challenge has been issued and the solution has not been sent, so the server cannot
+    // have read the table again yet: the row is gone by the time `verify_solution` looks.
+    {
+        let mut conn = db.get().await.unwrap();
+        diesel::delete(
+            schema::recovery_operator_identity::table
+                .filter(schema::recovery_operator_identity::public_key.eq(recovery_pubkey_bytes)),
+        )
+        .execute(&mut conn)
+        .await
+        .unwrap();
+    }
+
+    let signature = sign_operator_challenge(&recovery_key, &challenge);
+    test_transport
+        .send(auth::Inbound::AuthChallengeSolution {
+            signature: signature.to_bytes(),
+        })
+        .await
+        .unwrap();
+
+    let refusal = test_transport
+        .recv()
+        .await
+        .expect("the refusal must be sent to the peer");
+    assert!(matches!(refusal, Err(auth::Error::UnregisteredPublicKey)));
+
+    assert!(matches!(
+        task.await.unwrap(),
+        Err(auth::Error::UnregisteredPublicKey)
+    ));
+}