ceph/src/seastar/demos/tls_echo_server.hh

   1 /*
   2  * This file is open source software, licensed to you under the terms
   3  * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
   4  * distributed with this work for additional information regarding copyright
   5  * ownership.  You may not use this file except in compliance with the License.
   6  *
   7  * You may obtain a copy of the License at
   8  *
   9  *   http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing,
  12  * software distributed under the License is distributed on an
  13  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  14  * KIND, either express or implied.  See the License for the
  15  * specific language governing permissions and limitations
  16  * under the License.
  17  */
  18 /*
  19  * Copyright 2017 ScyllaDB
  20  */
  21 #pragma once
  22
  23 #include <seastar/core/sstring.hh>
  24 #include <seastar/core/do_with.hh>
  25 #include <seastar/core/sharded.hh>
  26 #include <seastar/core/gate.hh>
  27 #include <seastar/core/loop.hh>
  28 #include <seastar/net/tls.hh>
  29 #include <seastar/util/log.hh>
  30 #include <iostream>
  31
  32 using namespace seastar;
  33
  34 struct streams {
  35     connected_socket s;
  36     input_stream<char> in;
  37     output_stream<char> out;
  38
  39     streams(connected_socket cs) : s(std::move(cs)), in(s.input()), out(s.output())
  40     {}
  41 };
  42
  43 class echoserver {
  44     server_socket _socket;
  45     shared_ptr<tls::server_credentials> _certs;
  46     seastar::gate _gate;
  47     bool _stopped = false;
  48     bool _verbose = false;
  49 public:
  50     echoserver(bool verbose = false)
  51             : _certs(make_shared<tls::server_credentials>(make_shared<tls::dh_params>()))
  52             , _verbose(verbose)
  53     {}
  54
  55     future<> listen(socket_address addr, sstring crtfile, sstring keyfile, tls::client_auth ca = tls::client_auth::NONE) {
  56         _certs->set_client_auth(ca);
  57         return _certs->set_x509_key_file(crtfile, keyfile, tls::x509_crt_format::PEM).then([this, addr] {
  58             ::listen_options opts;
  59             opts.reuse_address = true;
  60
  61             _socket = tls::listen(_certs, addr, opts);
  62
  63             // Listen in background.
  64             (void)repeat([this] {
  65                 if (_stopped) {
  66                     return make_ready_future<stop_iteration>(stop_iteration::yes);
  67                 }
  68                 return with_gate(_gate, [this] {
  69                     return _socket.accept().then([this](accept_result ar) {
  70                         ::connected_socket s = std::move(ar.connection);
  71                         socket_address a = std::move(ar.remote_address);
  72                         if (_verbose) {
  73                             std::cout << "Got connection from "<< a << std::endl;
  74                         }
  75                         auto strms = make_lw_shared<streams>(std::move(s));
  76                         return repeat([strms, this]() {
  77                             return strms->in.read().then([this, strms](temporary_buffer<char> buf) {
  78                                 if (buf.empty()) {
  79                                     if (_verbose) {
  80                                         std::cout << "EOM" << std::endl;
  81                                     }
  82                                     return make_ready_future<stop_iteration>(stop_iteration::yes);
  83                                 }
  84                                 sstring tmp(buf.begin(), buf.end());
  85                                 if (_verbose) {
  86                                     std::cout << "Read " << tmp.size() << "B" << std::endl;
  87                                 }
  88                                 return strms->out.write(tmp).then([strms]() {
  89                                     return strms->out.flush();
  90                                 }).then([] {
  91                                     return make_ready_future<stop_iteration>(stop_iteration::no);
  92                                 });
  93                             });
  94                         }).then([strms]{
  95                             return strms->out.close();
  96                         }).handle_exception([](auto ep) {
  97                         }).finally([this, strms]{
  98                             if (_verbose) {
  99                                 std::cout << "Ending session" << std::endl;
 100                             }
 101                             return strms->in.close();
 102                         });
 103                     }).handle_exception([this](auto ep) {
 104                         if (!_stopped) {
 105                             std::cerr << "Error: " << ep << std::endl;
 106                         }
 107                     }).then([this] {
 108                         return make_ready_future<stop_iteration>(_stopped ? stop_iteration::yes : stop_iteration::no);
 109                     });
 110                 });
 111             });
 112             return make_ready_future();
 113         });
 114     }
 115
 116     future<> stop() {
 117         _stopped = true;
 118         _socket.abort_accept();
 119         return _gate.close();
 120     }
 121 };