[M5] mxaccess-asb: F25 step 6 — Connect/AuthenticateMe handshake

Critical-path piece that turns a fresh TCP stream into an
authenticated session. With this slice landed, an `AsbClient` can
now do `send_preamble().await? -> connect().await? -> register_items()`
end-to-end against a peer.

Operations API additions:
* `build_connect_request_body(connection_id, public_key)` — first op
  on a fresh session. **Unsigned** (no ConnectionValidator header)
  because the authenticator hasn't received the service key yet.
  Wire shape: `<ConnectRequest xmlns="…messages/20111111">
    <ConnectionId>{guid-text}</ConnectionId>
    <ConsumerPublicKey><Data>{pubkey-bytes}</Data></ConsumerPublicKey>
  </ConnectRequest>` per `AsbContracts.cs:78-86`.
* `build_authenticate_me_request_body(data, iv)` — second op,
  **one-way + signed with `forceHmac=true`** per `MxAsbDataClient.cs
  :106-111`. Carries the encrypted `local_pub || remote_pub` blob
  produced by F23's `create_authentication_data()`.
* `ConnectResponse { service_public_key, service_authentication_data,
  connection_lifetime }` + `AuthenticationDataBytes { data, iv }`.
* `decode_connect_response(body, dict)` — extracts ServicePublicKey
  (required), optional ServiceAuthenticationData, optional
  ConnectionLifetime. The lifetime's `:V2` suffix is what F23
  inspects to toggle Apollo (raw AES) vs Baktun (deflate-then-AES)
  encryption.

Client API addition:
* `AsbClient::connect()` — orchestrates the full handshake:
  1. Build + send ConnectRequest (unsigned) carrying our DH public
     key + connection-id GUID.
  2. Decode ConnectResponse.
  3. `authenticator.accept_connect_response(...)` — feeds the
     service public key + lifetime into F23 so it derives the
     shared secret and picks Apollo/Baktun.
  4. `authenticator.create_authentication_data()` — encrypts
     `local_pub || remote_pub` under the derived AES key.
  5. Send AuthenticateMeRequest (one-way, signed with HMAC-SHA1
     forced).
  Returns the `ConnectResponse` so callers can inspect the
  negotiated connection lifetime.

6 new tests:
* ConnectRequest carries hyphenated GUID + raw public-key bytes.
* AuthenticateMe carries Data + IV bytes in order.
* ConnectResponse round-trip with all optional fields populated.
* ConnectResponse round-trip without optional fields.
* ConnectResponse decoder surfaces MissingField when
  ServicePublicKey is absent.
* End-to-end client::connect handshake via `tokio::io::duplex`
  peer that synthesises a ConnectResponse using bob's public key
  (so DH shared-secret derivation actually works) and drains the
  AuthenticateMe one-way SizedEnvelope.

Wire-byte caveat documented inline: WCF XML serialization may add
`xsi:type` attributes / distinct namespaces around <PublicKey> /
<AuthenticationData>; this builder ships the simplest plausible
shape and the live-probe iteration will reconcile.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

rust/crates/mxaccess-asb/src/client.rs

@@ -55,10 +55,11 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use crate::contracts::{ItemIdentity, ItemStatus};
 use crate::envelope::{ConnectionValidator, EnvelopeError, SoapEnvelope};
 use crate::operations::{
-    OperationError, ReadResponse, RegisterItemsResponse, UnregisterItemsResponse,
-    build_keep_alive_request_body, build_read_request_body, build_register_items_request_body,
-    build_unregister_items_request_body, decode_read_response, decode_register_items_response,
-    decode_unregister_items_response,
+    ConnectResponse, OperationError, ReadResponse, RegisterItemsResponse, UnregisterItemsResponse,
+    build_authenticate_me_request_body, build_connect_request_body, build_keep_alive_request_body,
+    build_read_request_body, build_register_items_request_body,
+    build_unregister_items_request_body, decode_connect_response, decode_read_response,
+    decode_register_items_response, decode_unregister_items_response,
 };
 use crate::{actions, decode_envelope, encode_envelope};
 
@@ -196,6 +197,62 @@ impl<T: AsyncRead + AsyncWrite + Unpin + Send> AsbClient<T> {
         self.send_envelope(&signed_env).await
     }
 
+    /// Run the full DH `Connect` + `AuthenticateMe` handshake. Mirrors
+    /// `MxAsbDataClient.cs:84-112`:
+    ///
+    /// 1. Send `ConnectRequest` (unsigned — the authenticator hasn't
+    ///    received the service key yet) carrying our connection ID +
+    ///    public key.
+    /// 2. Receive `ConnectResponse` containing the service public key
+    ///    + optional connection lifetime + optional service auth data.
+    /// 3. Call `authenticator.accept_connect_response(...)` so it can
+    ///    derive the shared secret + decide on Apollo vs Baktun
+    ///    encryption based on the `:V2` lifetime suffix.
+    /// 4. Build encrypted `ConsumerAuthenticationData` via
+    ///    `authenticator.create_authentication_data()` (this is
+    ///    `Encrypt(local_pub || remote_pub)` — see F23).
+    /// 5. Send signed `AuthenticateMeRequest` with `forceHmac=true`
+    ///    (one-way, no response expected).
+    ///
+    /// Caller must have called [`Self::send_preamble`] first. Returns
+    /// the `ConnectResponse` so callers can inspect the negotiated
+    /// connection lifetime.
+    pub async fn connect(&mut self) -> Result<ConnectResponse, ClientError> {
+        if !self.preamble_sent {
+            return Err(ClientError::PreambleNotSent);
+        }
+
+        // Step 1: ConnectRequest (unsigned)
+        let connection_id = self.authenticator.connection_id();
+        let public_key = self.authenticator.local_public_key().to_vec();
+        let connect_body = build_connect_request_body(connection_id, &public_key);
+        let unsigned_env = SoapEnvelope::new(actions::CONNECT).with_body_tokens(connect_body);
+
+        // Step 2: send + receive ConnectResponse
+        let response_env = self.send_envelope(&unsigned_env).await?;
+        let connect_response =
+            decode_connect_response(&response_env.body_tokens, &self.read_dictionary)?;
+
+        // Step 3: feed the service public key + lifetime into the
+        // authenticator so it can derive the shared secret.
+        self.authenticator.accept_connect_response(
+            &connect_response.service_public_key,
+            connect_response.connection_lifetime.as_deref(),
+        );
+
+        // Step 4: build encrypted authentication data (local_pub ||
+        // remote_pub, encrypted under the derived AES key). Errors
+        // surface through ClientError::Auth.
+        let auth_data = self.authenticator.create_authentication_data()?;
+
+        // Step 5: AuthenticateMe one-way, signed with HMAC-SHA1 forced.
+        let auth_body = build_authenticate_me_request_body(&auth_data.ciphertext, &auth_data.iv);
+        self.send_signed_envelope_one_way(actions::AUTHENTICATE_ME, auth_body, true)
+            .await?;
+
+        Ok(connect_response)
+    }
+
     /// One-way send: encode + frame + write, but do **not** read a
     /// response. Mirrors WCF's `[OperationContract(IsOneWay = true)]`
     /// semantics — `KeepAlive`, `Disconnect`, and `AuthenticateMe` all
@@ -766,6 +823,113 @@ mod tests {
         let _ = peer_task.await.unwrap();
     }
 
+    #[tokio::test]
+    async fn connect_handshake_round_trips_through_in_memory_peer() {
+        let (client_end, peer_end) = tokio::io::duplex(8192);
+        let peer_task = spawn_peer(peer_end, |mut peer| async move {
+            // 1. Drain preamble + send PreambleAck
+            let mut buf = vec![0u8; 256];
+            let _n = peer.read(&mut buf).await.unwrap();
+            peer.write_all(&[0x0Bu8]).await.unwrap();
+            peer.flush().await.unwrap();
+
+            // 2. Drain ConnectRequest SizedEnvelope
+            let mut typebyte = [0u8; 1];
+            peer.read_exact(&mut typebyte).await.unwrap();
+            assert_eq!(typebyte[0], 0x06);
+            let mut lenbuf = Vec::new();
+            for _ in 0..5 {
+                let mut b = [0u8; 1];
+                peer.read_exact(&mut b).await.unwrap();
+                lenbuf.push(b[0]);
+                if b[0] & 0x80 == 0 {
+                    break;
+                }
+            }
+            let mut cursor = 0;
+            let len = mxaccess_asb_nettcp::nmf::decode_multibyte_int31(&lenbuf, &mut cursor)
+                .unwrap() as usize;
+            let _connect_request = read_n(&mut peer, len).await;
+
+            // 3. Build a synthetic ConnectResponse: service_public_key
+            // = matching `bob` so the shared-secret derivation works.
+            let bob = make_authenticator();
+            let svc_pubkey = bob.local_public_key().to_vec();
+            let body = synthesise_connect_response_body(svc_pubkey);
+            let envelope = SoapEnvelope::new(actions::CONNECT).with_body_tokens(body);
+            let mut response_dict = DynamicDictionary::new();
+            let envelope_bytes = encode_envelope(&envelope, &mut response_dict).unwrap();
+
+            let mut frame = vec![0x06u8];
+            mxaccess_asb_nettcp::nmf::encode_multibyte_int31(
+                &mut frame,
+                envelope_bytes.len() as i32,
+            )
+            .unwrap();
+            frame.extend_from_slice(&envelope_bytes);
+            peer.write_all(&frame).await.unwrap();
+            peer.flush().await.unwrap();
+
+            // 4. Drain AuthenticateMe one-way SizedEnvelope.
+            let mut typebyte = [0u8; 1];
+            peer.read_exact(&mut typebyte).await.unwrap();
+            assert_eq!(typebyte[0], 0x06);
+            let mut lenbuf = Vec::new();
+            for _ in 0..5 {
+                let mut b = [0u8; 1];
+                peer.read_exact(&mut b).await.unwrap();
+                lenbuf.push(b[0]);
+                if b[0] & 0x80 == 0 {
+                    break;
+                }
+            }
+            let mut cursor = 0;
+            let len = mxaccess_asb_nettcp::nmf::decode_multibyte_int31(&lenbuf, &mut cursor)
+                .unwrap() as usize;
+            let _authenticate_me = read_n(&mut peer, len).await;
+
+            peer
+        });
+
+        let mut client = AsbClient::new(client_end, make_authenticator(), "test://h/p");
+        client.send_preamble().await.unwrap();
+        let response = client.connect().await.unwrap();
+        // Smoke-check that the response carries our synthesized public
+        // key bytes (length matches a real DH key, ~129 bytes).
+        assert!(!response.service_public_key.is_empty());
+        assert!(response.connection_lifetime.is_none());
+
+        let _ = peer_task.await.unwrap();
+    }
+
+    fn synthesise_connect_response_body(
+        service_public_key: Vec<u8>,
+    ) -> Vec<mxaccess_asb_nettcp::nbfx::NbfxToken> {
+        use mxaccess_asb_nettcp::nbfx::{NbfxName, NbfxText, NbfxToken};
+        const MESSAGES_NS: &str = "http://asb.contracts.messages/20111111";
+        vec![
+            NbfxToken::Element {
+                prefix: None,
+                name: NbfxName::Inline("ConnectResponse".to_string()),
+            },
+            NbfxToken::DefaultNamespace {
+                value: NbfxText::Chars(MESSAGES_NS.to_string()),
+            },
+            NbfxToken::Element {
+                prefix: None,
+                name: NbfxName::Inline("ServicePublicKey".to_string()),
+            },
+            NbfxToken::Element {
+                prefix: None,
+                name: NbfxName::Inline("Data".to_string()),
+            },
+            NbfxToken::Text(NbfxText::Bytes(service_public_key)),
+            NbfxToken::EndElement, // </Data>
+            NbfxToken::EndElement, // </ServicePublicKey>
+            NbfxToken::EndElement, // </ConnectResponse>
+        ]
+    }
+
     fn synthesise_read_response_body(
         status_payload: Vec<u8>,
         values_payload: Vec<u8>,