Validate correlated authority response envelopes in both clients

account-manager/core/src/account_authority/mod.rs

@@ -34,8 +34,8 @@ pub use operation::{
 pub use protocol::{
     authority_id_for, decode_request, dispatch, map_core_error, AuthorityContext,
     AuthorityResponse, DecodedRequest, ErrorCode, LoginPort, ProtocolBounds, ProtocolFailure,
-    ADVERTISED_OPERATIONS, MAX_REQUEST_BYTES, MAX_RESPONSE_BYTES, PROTOCOL_VERSION,
-    REQUEST_IO_DEADLINE,
+    ResponseValidationError, ADVERTISED_OPERATIONS, MAX_REQUEST_BYTES, MAX_RESPONSE_BYTES,
+    PROTOCOL_VERSION, REQUEST_IO_DEADLINE,
 };
 pub use registry::StoredAccountRegistry;
 pub use server::{

account-manager/core/src/account_authority/protocol.rs

@@ -87,6 +87,23 @@ pub struct AuthorityResponse {
 }
 
 impl AuthorityResponse {
+    /// Validate correlation and the exclusive result/error shape before consuming a response.
+    pub fn validate_for_request(&self, request_id: &str) -> Result<(), ResponseValidationError> {
+        if request_id.is_empty() || self.request_id.is_empty() {
+            return Err(ResponseValidationError::EmptyRequestId);
+        }
+        if self.request_id != request_id {
+            return Err(ResponseValidationError::RequestIdMismatch);
+        }
+        if self.protocol_version != PROTOCOL_VERSION {
+            return Err(ResponseValidationError::UnsupportedVersion);
+        }
+        if self.result.is_some() == self.error.is_some() {
+            return Err(ResponseValidationError::InvalidResultShape);
+        }
+        Ok(())
+    }
+
     pub fn result(request_id: impl Into<String>, result: Value) -> Self {
         Self {
             protocol_version: PROTOCOL_VERSION,
@@ -117,6 +134,19 @@ impl AuthorityResponse {
     }
 }
 
+/// Closed response failures whose diagnostics never include remote values.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
+pub enum ResponseValidationError {
+    #[error("authority response requires nonempty request IDs")]
+    EmptyRequestId,
+    #[error("authority response request ID mismatch")]
+    RequestIdMismatch,
+    #[error("authority response protocol version is unsupported")]
+    UnsupportedVersion,
+    #[error("authority response requires exactly one result or error")]
+    InvalidResultShape,
+}
+
 /// Decode or dispatch failure that still has a closed wire shape.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ProtocolFailure {
@@ -1307,3 +1337,66 @@ fn hex_lower(bytes: &[u8]) -> String {
     }
     out
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn response_validation_requires_correlated_exclusive_envelope() {
+        let valid = AuthorityResponse::result("rpc", json!({}));
+        assert_eq!(valid.validate_for_request("rpc"), Ok(()));
+        assert_eq!(
+            AuthorityResponse::error("rpc", ErrorCode::Busy, "busy").validate_for_request("rpc"),
+            Ok(())
+        );
+        assert_eq!(
+            valid.validate_for_request(""),
+            Err(ResponseValidationError::EmptyRequestId)
+        );
+        for (response, expected) in [
+            (
+                AuthorityResponse::result("", json!({})),
+                ResponseValidationError::EmptyRequestId,
+            ),
+            (
+                AuthorityResponse::result("secret-sentinel", json!({})),
+                ResponseValidationError::RequestIdMismatch,
+            ),
+            (
+                AuthorityResponse {
+                    protocol_version: 2,
+                    ..valid.clone()
+                },
+                ResponseValidationError::UnsupportedVersion,
+            ),
+            (
+                AuthorityResponse {
+                    result: None,
+                    ..valid.clone()
+                },
+                ResponseValidationError::InvalidResultShape,
+            ),
+            (
+                AuthorityResponse {
+                    result: Some(json!({})),
+                    ..AuthorityResponse::error("rpc", ErrorCode::Busy, "secret-sentinel")
+                },
+                ResponseValidationError::InvalidResultShape,
+            ),
+        ] {
+            let error = response.validate_for_request("rpc").unwrap_err();
+            assert_eq!(error, expected);
+            assert!(!error.to_string().contains("secret-sentinel"));
+        }
+        assert!(serde_json::from_value::<AuthorityResponse>(json!({
+            "protocolVersion": PROTOCOL_VERSION, "result": {}
+        }))
+        .is_err());
+        assert_eq!(
+            AuthorityResponse::result("", json!({})).validate_for_request(""),
+            Err(ResponseValidationError::EmptyRequestId)
+        );
+    }
+}

account-manager/src-tauri/src/authority_client.rs

@@ -610,14 +610,9 @@ fn decode_response(bytes: &[u8], request_id: &str) -> Result<AuthorityResponse>
             .transpose()
             .map_err(|_| authority_decode_error("invalid authority error envelope"))?,
     };
-    if response.protocol_version != PROTOCOL_VERSION
-        || response.request_id != request_id
-        || response.result.is_some() == response.error.is_some()
-    {
-        return Err(authority_decode_error(
-            "authority response envelope mismatch",
-        ));
-    }
+    response
+        .validate_for_request(request_id)
+        .map_err(|error| authority_decode_error(&error.to_string()))?;
     Ok(response)
 }
 
@@ -1456,9 +1451,20 @@ mod tests {
     fn response_refuses_uncorrelated_ambiguous_unknown_and_secret_decode_errors() {
         let valid = json!({"protocolVersion":1,"requestId":"rpc","result":{}});
         assert!(decode_response(&serde_json::to_vec(&valid).unwrap(), "rpc").is_ok());
+        assert!(decode_response(
+            br#"{"protocolVersion":1,"requestId":"rpc","error":{"code":"busy","message":"busy"}}"#,
+            "rpc"
+        )
+        .is_ok());
+        assert!(decode_response(&serde_json::to_vec(&valid).unwrap(), "").is_err());
+        assert!(
+            decode_response(br#"{"protocolVersion":1,"requestId":"","result":{}}"#, "").is_err()
+        );
         for invalid in [
             json!({"protocolVersion":2,"requestId":"rpc","result":{}}),
             json!({"protocolVersion":1,"requestId":"other","result":{}}),
+            json!({"protocolVersion":1,"result":{}}),
+            json!({"protocolVersion":1,"requestId":"","result":{}}),
             json!({"protocolVersion":1,"requestId":"rpc"}),
             json!({"protocolVersion":1,"requestId":"rpc","result":{},"error":{"code":"busy","message":"busy"}}),
             json!({"protocolVersion":1,"requestId":"rpc","result":{},"command":"arbitrary"}),
@@ -1547,6 +1553,65 @@ mod tests {
             std::env::remove_var(CAM_ACCOUNT_AUTHORITY_SOCKET_ENV);
         }
 
+        #[test]
+        fn exchange_refuses_invalid_envelopes_without_changing_local_selection() {
+            let (_registry_dir, registry) = isolated_registry();
+            registry
+                .begin_add(
+                    "codex-cli",
+                    "work",
+                    "Work",
+                    AuthKind::OAuth,
+                    StoredAccountMaterial::VendorHome,
+                )
+                .unwrap();
+            registry.complete_add("codex-cli", "work").unwrap();
+            registry
+                .select_complete_revision("codex-cli", "work", None)
+                .unwrap();
+            let original = registry.selected_binding("codex-cli").unwrap();
+            for case in 0..6 {
+                let (_guard, root) = private_tempdir();
+                let socket = root.join("authority.sock");
+                let listener = std::os::unix::net::UnixListener::bind(&socket).unwrap();
+                let client = AuthorityClient::connect(&socket).unwrap();
+                let server = thread::spawn(move || {
+                    let (mut stream, _) = listener.accept().unwrap();
+                    let deadline = std::time::Instant::now() + REQUEST_IO_DEADLINE;
+                    let request = read_framed(&mut stream, MAX_REQUEST_BYTES, deadline).unwrap();
+                    let request: Value = serde_json::from_slice(&request).unwrap();
+                    assert_eq!(request["operation"], "selection.get");
+                    let mut response = json!({"protocolVersion": PROTOCOL_VERSION,
+                        "requestId": request["requestId"], "result": {"selected": false}});
+                    match case {
+                        0 => response["requestId"] = json!("secret-sentinel"),
+                        1 => {
+                            response.as_object_mut().unwrap().remove("requestId");
+                        }
+                        2 => response["requestId"] = json!(""),
+                        3 => {
+                            response["error"] =
+                                json!({"code": "busy", "message": "secret-sentinel"});
+                        }
+                        4 => {
+                            response.as_object_mut().unwrap().remove("result");
+                        }
+                        _ => response["protocolVersion"] = json!(2),
+                    }
+                    write_framed(
+                        &mut stream,
+                        &serde_json::to_vec(&response).unwrap(),
+                        deadline,
+                    )
+                    .unwrap();
+                });
+                let error = client.selected_binding("codex-cli").unwrap_err();
+                assert!(!error.to_string().contains("secret-sentinel"));
+                assert_eq!(registry.selected_binding("codex-cli").unwrap(), original);
+                server.join().unwrap();
+            }
+        }
+
         #[test]
         fn client_selection_get_round_trips() {
             let (_registry_dir, registry) = isolated_registry();

src/account_authority/socket_client.rs

@@ -149,14 +149,14 @@ impl AuthoritySocketClient {
         let deadline = Instant::now() + REQUEST_IO_DEADLINE;
         write_framed(&mut stream, &body, deadline)?;
         let response_bytes = read_framed(&mut stream, MAX_RESPONSE_BYTES, deadline)?;
-        let response: AuthorityResponse =
-            serde_json::from_slice(&response_bytes).context("decode CAM authority response")?;
-        if response.protocol_version != PROTOCOL_VERSION {
-            return Err(anyhow!(
-                "CAM authority response protocolVersion {} is not supported",
-                response.protocol_version
-            ));
-        }
+        let response: AuthorityResponse = serde_json::from_slice(&response_bytes)
+            .map_err(|_| anyhow!("invalid CAM authority response envelope"))?;
+        response.validate_for_request(
+            request
+                .get("requestId")
+                .and_then(Value::as_str)
+                .unwrap_or(""),
+        )?;
         Ok(response)
     }
 }
@@ -397,6 +397,62 @@ mod tests {
         (dir, listener)
     }
 
+    #[test]
+    fn exchange_refuses_uncorrelated_or_ambiguous_response() {
+        use std::os::unix::net::UnixListener;
+
+        let (_registry_dir, registry) = isolated_registry();
+        seed_selected_account(&registry);
+        let original = registry.selected_binding("codex-cli").unwrap();
+        for case in 0..7 {
+            let (_guard, root) = private_tempdir();
+            let socket = root.join("account.sock");
+            let listener = UnixListener::bind(&socket).unwrap();
+            let client = connect_authority_socket(&socket).unwrap();
+            let server = thread::spawn(move || {
+                let (mut stream, _) = listener.accept().unwrap();
+                let deadline = Instant::now() + REQUEST_IO_DEADLINE;
+                let request = read_framed(&mut stream, MAX_REQUEST_BYTES, deadline).unwrap();
+                let request: Value = serde_json::from_slice(&request).unwrap();
+                assert_eq!(request["operation"], "selection.get");
+                let mut response = json!({
+                    "protocolVersion": PROTOCOL_VERSION,
+                    "requestId": request["requestId"],
+                    "result": {"selected": false},
+                });
+                match case {
+                    0 => response["requestId"] = json!("secret-sentinel"),
+                    1 => {
+                        response.as_object_mut().unwrap().remove("requestId");
+                    }
+                    2 => response["requestId"] = json!(""),
+                    3 => response["error"] = json!({"code": "busy", "message": "secret-sentinel"}),
+                    4 => {
+                        response.as_object_mut().unwrap().remove("result");
+                    }
+                    5 => response["protocolVersion"] = json!(2),
+                    _ => {
+                        response.as_object_mut().unwrap().remove("result");
+                        response["error"] = json!({"code": "busy", "message": "busy"});
+                    }
+                }
+                write_framed(
+                    &mut stream,
+                    &serde_json::to_vec(&response).unwrap(),
+                    deadline,
+                )
+                .unwrap();
+            });
+            let error = client.selection_get("codex-cli").unwrap_err();
+            assert!(!format!("{error:#}").contains("secret-sentinel"));
+            if case == 6 {
+                assert!(error.to_string().contains("busy"));
+            }
+            assert_eq!(registry.selected_binding("codex-cli").unwrap(), original);
+            server.join().unwrap();
+        }
+    }
+
     #[test]
     fn framing_helpers_round_trip_payload() {
         use std::os::unix::net::UnixListener;