mxaccess/rust/crates/mxaccess/examples/asb-relay.rs

//! `asb-relay` — TCP middleman that captures both sides of an ASB
//! exchange.
//!
//! Listens on `MX_RELAY_LISTEN` (default `127.0.0.1:8088`) and forwards
//! every connection to `MX_RELAY_UPSTREAM` (default `127.0.0.1:808`,
//! AVEVA's `NetTcpPortSharing` SMSvcHost listener). All bytes both
//! directions are hex-dumped to stderr with direction + offset
//! prefixes, so you can `> capture.log 2>&1` and diff client-vs-server
//! bytes byte-for-byte.
//!
//! Use with the .NET probe:
//!
//! ```powershell
//! # Terminal A: start the relay
//! cargo run -p mxaccess --example asb-relay 2> .\dotnet.log
//!
//! # Terminal B: point the .NET probe at the relay
//! dotnet run --project src\MxAsbClient.Probe -c Release -- `
//!   --endpoint "net.tcp://desktop-6jl3kko:8088/ASBService/Default_ZB_MxDataProvider/IDataV2"
//! ```
//!
//! Then run our Rust client through the relay:
//!
//! ```powershell
//! $env:MX_ASB_HOST = "127.0.0.1:8088"
//! cargo run -p mxaccess --example asb-preamble-probe 2> .\rust.log
//! ```
//!
//! Diff the two logs to find wire-byte deltas. Direction labels are
//! `C->S` (client→server) and `S->C` (server→client).
//!
//! ## Important caveat: SMSvcHost URL matching
//!
//! When the relay forwards to `127.0.0.1:808`, the SMSvcHost dispatcher
//! looks at the NMF `Via` record's URL host segment to pick which
//! registered service to route to. The .NET probe's default URL has
//! the hostname `desktop-6jl3kko`, NOT `127.0.0.1` — and SMSvcHost
//! resolves the registered service by the URL the AVEVA installer
//! recorded (which is the actual hostname). Use the actual hostname
//! in the .NET probe `--endpoint` arg even when sending through the
//! relay; only the TCP socket changes (we listen on `127.0.0.1:8088`,
//! the .NET probe's TCP DNS still resolves to localhost so it
//! connects to us, but the URL inside the preamble routes correctly
//! at SMSvcHost).

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listen_addr =
        std::env::var("MX_RELAY_LISTEN").unwrap_or_else(|_| "127.0.0.1:8088".to_string());
    let upstream_addr =
        std::env::var("MX_RELAY_UPSTREAM").unwrap_or_else(|_| "127.0.0.1:808".to_string());

    let listener = TcpListener::bind(&listen_addr).await?;
    eprintln!(
        "asb-relay listening on {} → forwarding to {}",
        listen_addr, upstream_addr
    );
    eprintln!("hex prefixes: C->S = client→server, S->C = server→client");

    let conn_counter = Arc::new(AtomicUsize::new(0));

    loop {
        let (client_stream, peer) = listener.accept().await?;
        let conn_id = conn_counter.fetch_add(1, Ordering::Relaxed);
        let upstream_addr = upstream_addr.clone();
        eprintln!("[#{conn_id}] accepted {peer}");

        tokio::spawn(async move {
            if let Err(e) = handle_connection(conn_id, client_stream, &upstream_addr).await {
                eprintln!("[#{conn_id}] connection error: {e}");
            }
            eprintln!("[#{conn_id}] closed");
        });
    }
}

async fn handle_connection(
    conn_id: usize,
    mut client: TcpStream,
    upstream_addr: &str,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let mut server = TcpStream::connect(upstream_addr).await?;
    eprintln!("[#{conn_id}] upstream connected");

    // Disable Nagle on both sides so log timing matches actual flushes.
    let _ = client.set_nodelay(true);
    let _ = server.set_nodelay(true);

    let (mut cr, mut cw) = client.split();
    let (mut sr, mut sw) = server.split();

    let cs = forward(conn_id, "C->S", &mut cr, &mut sw);
    let sc = forward(conn_id, "S->C", &mut sr, &mut cw);

    let _ = tokio::join!(cs, sc);
    Ok(())
}

async fn forward<R, W>(
    conn_id: usize,
    tag: &'static str,
    reader: &mut R,
    writer: &mut W,
) -> std::io::Result<()>
where
    R: tokio::io::AsyncRead + Unpin,
    W: tokio::io::AsyncWrite + Unpin,
{
    let mut buf = vec![0u8; 8192];
    let mut total = 0usize;
    loop {
        let n = reader.read(&mut buf).await?;
        if n == 0 {
            break;
        }
        if let Some(slice) = buf.get(..n) {
            print_hex(conn_id, tag, total, slice);
            writer.write_all(slice).await?;
            writer.flush().await?;
        }
        total += n;
    }
    let _ = writer.shutdown().await;
    eprintln!("[#{conn_id}] {tag} EOF after {total} bytes");
    Ok(())
}

fn print_hex(conn_id: usize, tag: &str, base_offset: usize, bytes: &[u8]) {
    for (chunk_idx, chunk) in bytes.chunks(16).enumerate() {
        let offset = base_offset + chunk_idx * 16;
        let hex: Vec<String> = chunk.iter().map(|b| format!("{b:02x}")).collect();
        let ascii: String = chunk
            .iter()
            .map(|b| {
                if b.is_ascii_graphic() || *b == b' ' {
                    *b as char
                } else {
                    '.'
                }
            })
            .collect();
        eprintln!(
            "[#{conn_id}] {tag} {offset:08x}  {:<48}  {}",
            hex.join(" "),
            ascii
        );
    }
}