protwrap 0.4.3

Thin protocol wrapper for network applications.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

#[cfg(all(feature = "tokio", feature = "tls"))]
mod tokio_tests {
  use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    sync::oneshot,
    task
  };

  use protwrap::tokio::{
    client::{Connector, TlsTcpConnInfo},
    server::listener::{
      async_trait, Acceptor, KillSwitch, Listener, SockAddr,
      TlsTcpListenerInfo
    },
    ServerStream
  };

  struct MyServer {
    tx_port: Option<oneshot::Sender<u16>>,
    ks: KillSwitch
  }

  #[async_trait]
  impl Acceptor for MyServer {
    async fn bound(&mut self, _listener: &Listener, sa: SockAddr) {
      let port = sa.unwrap_std().port();
      let Some(tx) = self.tx_port.take() else {
        panic!("Channel end-point missing");
      };

      // Transmit assigned port number
      tx.send(port).unwrap();
    }

    async fn unbound(&mut self, _listener: &Listener) {}

    async fn connected(&mut self, _sa: SockAddr, strm: ServerStream) {
      tokio::task::spawn(Self::handle_connection(strm, self.ks.clone()))
        .await
        .unwrap();
    }
  }

  impl MyServer {
    async fn handle_connection(mut strm: ServerStream, ks: KillSwitch) {
      let mut buf = [0u8; 5];

      let n = strm.read_exact(&mut buf[..]).await.unwrap();
      assert_eq!(n, 5);
      assert_eq!(buf, *b"hello");

      strm.write_all(b"world").await.unwrap();
      strm.flush().await.unwrap();

      ks.trigger();
    }
  }

  #[tokio::test]
  async fn client_server() {
    //
    // Generate pki files
    //
    let (ca, srv) = task::spawn_blocking(init_pki).await.unwrap();
    let (srv_key_pem, srv_cert_pem) = srv.to_pem().unwrap();

    //
    // Set up TLS listener
    //
    let tlsinfo = TlsTcpListenerInfo {
      addr: "127.0.0.1:0".into(),
      srv_key_pem,
      srv_cert_pem
    };
    let listener = Listener::from(tlsinfo);

    let ks = KillSwitch::new();

    // One-shot channel used to pass the assigned port number
    let (tx, rx) = oneshot::channel();

    //
    // Kick off the listener in a semarate task
    //
    let acceptor = MyServer {
      tx_port: Some(tx),
      ks: ks.clone()
    };
    let killswitch = ks.clone();
    let jh_server = tokio::task::spawn(async move {
      listener.run(killswitch, acceptor).await.unwrap();
    });

    let (_, ca_cert_pem) = ca.to_pem().unwrap();
    let jh_client = tokio::task::spawn(async move {
      // Use side-channel to receive port number from server
      let port = rx.await.unwrap();

      //
      // Set up connector
      //
      let addr = format!("127.0.0.1:{port}");
      let conninfo = TlsTcpConnInfo {
        addr,
        host: "localhost".into(),
        ca_cert_pem
      };
      let c = Connector::from(conninfo);

      //
      // Connect to server
      //
      let mut strm = c.connect().await.unwrap();

      //
      // Send 'hello' to client
      //
      let n = strm.write(b"hello").await.unwrap();
      assert_eq!(n, 5);
      strm.flush().await.unwrap();

      //
      // Receive 'world' from client
      //
      let mut buf = [0u8; 5];
      let n = strm.read_exact(&mut buf[..]).await.unwrap();
      assert_eq!(n, 5);
      assert_eq!(buf, *b"world");
    });

    ks.wait().await;

    jh_client.await.unwrap();
    jh_server.await.unwrap();
  }

  fn init_pki() -> (quickcert::PkiIdent, quickcert::PkiIdent) {
    let ca = quickcert::mk_ca().unwrap();
    let srv = quickcert::mk_server(&ca, "server", ["localhost"]).unwrap();
    (ca, srv)
  }
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :