How to create static forwarder

[Ranger11Danger]

Ive been trying to figure out how to create a simple forwarder and client that creates a forward address

Forwarder:

#[ockam::node]
async fn main(ctx: Context) -> Result<()> {
    let tcp_listener_options = TcpListenerOptions::new();
    
    let options = ForwardingServiceOptions::new()
        .service_as_consumer(
            &tcp_listener_options.spawner_flow_control_id(),
            FlowControlPolicy::SpawnerAllowMultipleMessages,
        )
        .forwarder_as_consumer(
            &tcp_listener_options.spawner_flow_control_id(),
            FlowControlPolicy::SpawnerAllowMultipleMessages,
        );
    ForwardingService::create(&ctx, "static_forwarding_service", options).await?;

    let node = node(ctx);
    let tcp = node.create_tcp_transport().await?;

    let _listener = tcp
        .listen("127.0.0.1:4444", tcp_listener_options)
        .await?;
    Ok(())
} 

Client:

struct Echoer;
#[ockam::worker]
impl Worker for Echoer {
    type Context = Context;
    type Message = String;

    async fn handle_message(&mut self, ctx: &mut Context, msg: Routed<String>) -> Result<()> {
        println!("\n[✓] Address: {}, Received: {:?}, Route: {:?}", ctx.address(), msg, msg.return_route());
        ctx.send(msg.return_route(), msg.body()).await
    }
}

#[ockam::node]
async fn main(ctx: Context) -> Result<()> {
    
    let mut node = node(ctx);
    
    node.start_worker("echoer", Echoer, AllowAll, AllowAll).await?;
    let tcp = node.create_tcp_transport().await?;

    let node_in_hub = tcp
        .connect("127.0.0.1:4444", TcpConnectionOptions::new())
        .await?;

    let forwarder = node
        .create_static_forwarder(node_in_hub.clone(), "custom_address", RemoteForwarderOptions::new())
        .await?; 
    println!("{:?}", forwarder.remote_address());
Ok(())
}

Ive gotten this to work using the normal 'create_forwarder' method but when I try and create a static one the logs from the forwarder say this:

2023-05-28T22:03:16.696577Z  INFO ockam::forwarding_service::forwarder: Created new alias 0#custom_address for 0#a73fbe06ff7256f409233c5a1a457480 => 0#6f10348855ec9d0071eaf0a8090346b2
2023-05-28T22:03:16.696610Z  INFO ockam_node::worker_builder: Initializing ockam worker '0#custom_address' with access control in:AllowAll out:AllowOnwardAddress(0#a73fbe06ff7256f409233c5a1a457480)
2023-05-28T22:03:21.700588Z  INFO ockam::forwarding_service::forwarder: Created new alias 0#custom_address for 0#a73fbe06ff7256f409233c5a1a457480 => 0#6f10348855ec9d0071eaf0a8090346b2
2023-05-28T22:03:21.700686Z  INFO ockam_node::worker_builder: Initializing ockam worker '0#custom_address' with access control in:AllowAll out:AllowOnwardAddress(0#a73fbe06ff7256f409233c5a1a457480)
2023-05-28T22:03:21.701056Z ERROR ockam_node::relay::worker_relay: Error encountered during '0#static_forwarding_service' message handling: operation failed for address 0#custom_address

I was just wondering if anyone else as any experience with this and could point me i the right direction thanks!

[etorreborre]

Hi @Ranger11Danger this appears to be an issue with the heartbeat implementation that gets triggered when you use a static forwarder.

I don't know exactly yet how to fix it but I think that this should not prevent the static forwarder from working properly despite the error messages. Those messages are meaning to say that a new forwarder can not be created because there is already a forwarder at that address.

If you want to get rid of those messages you can use a lower-level API and write in the client:

let forwarder = RemoteForwarder::create_static_without_heartbeats(
    node.context(),
    node_in_hub.clone(),
    "custom_address",
    RemoteForwarderOptions::new(),
).await?;

While starting the server as (note the forwarding_service instead of static_forwarding_service):

ForwardingService::create(&ctx, "forwarding_service", options).await?;

We are going to better diagnose this issue and come back to you. Thanks for reporting it!

[Ranger11Danger]

that solution got rid of the error messages, Thanks!

[Ranger11Danger]

Hey @etorreborre Ive been building off of the example above and setup a basic scenario where a client registers a forward address with the server, and then the client sends a message to the servers "echoer" The server then echos back like normal and the message is received, however in the servers "echoer" worker after it sends a response it also tries to send a message using the forward address the client created to send message to an "echoer" worker running on the client. Essentially creating a ping -> pong -> ping -> pong situation. however when the client receives the message through the forward address and hitting the "echoer" address i recieve and error saying it did not pass outgoing access control. Is there a way to allow that message to come back through?
Server:

struct Echoer;
#[ockam::worker]
impl Worker for Echoer {
    type Context = Context;
    type Message = String;

    async fn handle_message(&mut self, ctx: &mut Context, msg: Routed<String>) -> Result<()> {
        println!("\n[✓] Address: {}, Received: {:?}, Route: {:?}", ctx.address(), msg, msg.return_route());
        ctx.send(msg.return_route(), msg.body()).await;
        ctx.send(route!["custom_address", "echoer"], "test".to_string()).await
    }
}

#[ockam::node]
async fn main(ctx: Context) -> Result<()> {
    
    let tcp_listener_options = TcpListenerOptions::new();
    
    let options = ForwardingServiceOptions::new()
        .service_as_consumer(
            &tcp_listener_options.spawner_flow_control_id(),
            FlowControlPolicy::SpawnerAllowMultipleMessages,
        )
        .forwarder_as_consumer(
            &tcp_listener_options.spawner_flow_control_id(),
            FlowControlPolicy::SpawnerAllowMultipleMessages,
        );
    ForwardingService::create(&ctx, "forwarding_service", options).await?;

    let node = node(ctx);
    
    node.start_worker("echoer", Echoer, AllowAll, AllowAll).await?;
    
    let tcp = node.create_tcp_transport().await?;
    // start the worker that is going to echo back messages

    // start the listener for the implant
    let listener = tcp
        .listen("127.0.0.1:4444", tcp_listener_options)
        .await?;

    let bob = node.create_identity().await?;
    let secure_channel_listener = node
        .create_secure_channel_listener(
            &bob,
            "bob_listener",
            SecureChannelListenerOptions::new().as_consumer(
                listener.flow_control_id(),
                FlowControlPolicy::SpawnerAllowMultipleMessages
            ),
        )
        .await?;

    node.flow_controls().add_consumer(
        "echoer",
        secure_channel_listener.flow_control_id(),
        FlowControlPolicy::SpawnerAllowMultipleMessages,
    );

    Ok(())
}

Client:

struct Echoer;
#[ockam::worker]
impl Worker for Echoer {
    type Context = Context;
    type Message = String;

    async fn handle_message(&mut self, ctx: &mut Context, msg: Routed<String>) -> Result<()> {
        println!("\n[✓] Address: {}, Received: {:?}, Route: {:?}", ctx.address(), msg, msg.return_route());
        ctx.send(msg.return_route(), msg.body()).await
    }
}

#[ockam::node]
async fn main(ctx: Context) -> Result<()> {
    
    let mut node = node(ctx); 
    node.start_worker("echoer", Echoer, AllowAll, AllowAll).await?;

    //Setup Connection to Server
    let tcp = node.create_tcp_transport().await?;
    let node_in_hub = tcp
        .connect("127.0.0.1:4444", TcpConnectionOptions::new())
        .await?;

    let forwarder = RemoteForwarder::create_static_without_heartbeats(
    node.context(),
    node_in_hub.clone(),
    "custom_address",
    RemoteForwarderOptions::new())
    .await?;
    println!("{:?}", forwarder.remote_address());

    //Setup Secure Channel
    let alice = node.create_identity().await?;
    let r = route![node_in_hub,  "bob_listener"];
    let channel = node
        .create_secure_channel(&alice, r, SecureChannelOptions::new())
        .await?;

        //Add echoer as consumer
    node.flow_controls().add_consumer(
        "echoer",
        channel.flow_control_id(),
        FlowControlPolicy::ProducerAllowMultiple,
    );
    let reply = node
        .send_and_receive::<String>(route![channel, "echoer"], "Hello Ockam!".to_string())
        .await?;
    println!("App Received: {}", reply); // should print "Hello Ockam!"

    Ok(())
}

Client Error:

WARN ockam_node::context::send_message: Message forwarded from 0#a95e27256baf33609dbdf02ef3afda24 to 0#echoer did not pass outgoing access control

[etorreborre]

Hi @Ranger11Danger.

Is there a way to allow that message to come back through?

The answer is yes! But you need to modify a bit the client code

...
let tcp_options = TcpConnectionOptions::new();
let tcp_flow_control_id = tcp_options.producer_flow_control_id();
let node_in_hub = tcp.connect("127.0.0.1:4444", tcp_options).await?;
...
// Add echoer as consumer
node.flow_controls()
    .add_consumer("echoer", &tcp_flow_control_id, FlowControlPolicy::ProducerAllowMultiple);

The warning message tells you that the TCPReceiverWorker could send a message to the echoer because there was no flow control established between the 2. This is not obvious to see with random addresses but if you run the same code with --features ockam_core/debugger you will get better names for the workers addresses.

This is however not what you would want in production since sending a message back from the Server Echoer via the forwarding address does not encrypt the message! But you should be able to see a ping-pong-ping-pong-... going on with that setup.

[Ranger11Danger]

Hi @etorreborre Thanks for the help, i was able to get it working through the secure channel as well! is there any documentation of the difference between spawners, producers, and consumers. or How the flow control policies work?

[etorreborre]

is there any documentation of the difference between spawners, producers, and consumers. or How the flow control policies work?

This is still work in progress. Since Ockam's routing is based on workers sending messages with routes determined by the message itself we need to have a way to control which worker is accessible from which other worker. The whole situation become a bit more complicated when you have workers spawning other workers :-). Typically when a ForwardingService creates a Forwarder.

Our first step was to make the system secure with flow controls in order to make sure that a malicious message could not reach a worker that it is not supposed to reach. We are fully aware that the UX of that feature is not the best. This is something that we actively continue to work on. Our hope is to be able to better encapsulate and simplify so this area of the code might change in the next months (ideas are welcome if you have a knack for APIs and DSLs :-)). Stay tuned!

[etorreborre]

is there any documentation of the difference between spawners, producers, and consumers. or How the flow control policies work?

Maybe I should still attempt a quick explanation though :-).

When you want to establish a "chain of communication between some workers" you generate a flow id. Attached to this flow id are:

• spawners: workers which can spawn other workers
• producers: workers which are sending messages
• consumers: workers which are receiving messages

Then the API of the FlowControls struct allows you to define, for example, which consumer can receive a message from which producer. In practice this means that:

• most producers on a secure node cannot send messages to anyone (that's a safe default)
• a consumer specified for a given flow id will appear in the list of recipients that a producer for that flow id can send messages to