jessequ
/
AMQP-CPP
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

removed tabs

This commit is contained in:
Emiel Bruijntjes 2018-03-06 08:40:44 +01:00
parent 69c615d0a4
commit 69596e49dc
10 changed files with 1035 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

344
src/linux_tcp/sslconnected.h Normal file
View File

@ -0,0 +1,344 @@
/**
* SslConnected.h
*
* The actual tcp connection over SSL
*
* @copyright 2018 copernica BV
*/
/**
* Include guard
*/
#pragma once
/**
* Dependencies
*/
#include "tcpoutbuffer.h"
#include "tcpinbuffer.h"
#include "wait.h"
#include "sslwrapper.h"
#include "sslshutdown.h"
/**
* Set up namespace
*/
namespace AMQP {
/**
* Class definition
*/
class SslConnected : public TcpState, private Watchable
{
private:
/**
* The SSL structure
* @var SslWrapper
*/
SslWrapper _ssl;
/**
* Socket file descriptor
* @var int
*/
int _socket;
/**
* The outgoing buffer
* @var TcpBuffer
*/
TcpOutBuffer _out;
/**
* The incoming buffer
* @var TcpInBuffer
*/
TcpInBuffer _in;
/**
* Are we now busy with sending or receiving?
* @var enum
*/
enum {
state_idle,
state_sending,
state_receiving
} _state;
/**
* Is the object already closed?
* @var bool
*/
bool _closed = false;
/**
* Cached reallocation instruction
* @var size_t
*/
size_t _reallocate = 0;
/**
* Helper method to report an error
* @return bool Was an error reported?
*/
bool reportError()
{
// we have an error - report this to the user
_handler->onError(_connection, strerror(errno));
// done
return true;
}
/**
* Construct the next state
* @param monitor Object that monitors whether connection still exists
* @return TcpState*
*/
TcpState *nextState(const Monitor &monitor)
{
// if the object is still in a valid state, we can move to the close-state,
// otherwise there is no point in moving to a next state
return monitor.valid() ? new TcpClosed(this) : nullptr;
}
/**
* Proceed with the next operation after the previous operation was
* a success, possibly changing the filedescriptor-monitor
* @return TcpState*
*/
TcpState *proceed()
{
// if we still have an outgoing buffer we want to send out data
if (_out)
{
// we still have a buffer with outgoing data
_state = state_sending;
// let's wait until the socket becomes writable
_handler->monitor(_connection, _socket, readable | writable);
}
else if (_closed)
{
// we forget the current handler to prevent that things are changed
_handler = nullptr;
// start the state that closes the connection
return new SslShutdown(_connection, _socket, _ssl, _handler);
}
else
{
// outgoing buffer is empty, we're idle again waiting for further input
_state = state_idle;
// let's wait until the socket becomes readable
_handler->monitor(_connection, _socket, readable);
}
// done
return this;
}
/**
* Method to repeat the previous call
* @param result result of an earlier openssl operation
* @return TcpState*
*/
TcpState *repeat(int result)
{
// error was returned, so we must investigate what is going on
auto error = SSL_get_error(_ssl, result);
// check the error
switch (error) {
case SSL_ERROR_WANT_READ:
// the operation must be repeated when readable
_handler->monitor(_connection, _socket, readable);
return this;
case SSL_ERROR_WANT_WRITE:
// wait until socket becomes writable again
_handler->monitor(_connection, _socket, readable | writable);
return this;
default:
// @todo check how to handle this
return this;
}
}
/**
* Parse the received buffer
* @param size
* @return TcpState
*/
TcpState *parse(size_t size)
{
// we need a local copy of the buffer - because it is possible that "this"
// object gets destructed halfway through the call to the parse() method
TcpInBuffer buffer(std::move(_in));
// because the object might soon be destructed, we create a monitor to check this
Monitor monitor(this);
// parse the buffer
auto processed = _connection->parse(buffer);
// "this" could be removed by now, check this
if (!monitor.valid()) return nullptr;
// shrink buffer
buffer.shrink(processed);
// restore the buffer as member
_in = std::move(buffer);
// do we have to reallocate?
if (_reallocate) _in.reallocate(_reallocate);
// we can remove the reallocate instruction
_reallocate = 0;
// done
return this;
}
public:
/**
* Constructor
* @param connection Parent TCP connection object
* @param socket The socket filedescriptor
* @param ssl The SSL structure
* @param buffer The buffer that was already built
* @param handler User-supplied handler object
*/
SslConnected(TcpConnection *connection, int socket, const SslWrapper &ssl, TcpOutBuffer &&buffer, TcpHandler *handler) :
TcpState(connection, handler),
_ssl(ssl),
_socket(socket),
_out(std::move(buffer)),
_in(4096),
_state(_out ? state_sending : state_idle)
{
// tell the handler to monitor the socket if there is an out
_handler->monitor(_connection, _socket, _state == state_sending ? writable : readable);
}
/**
* Destructor
*/
virtual ~SslConnected() noexcept
{
// skip if handler is already forgotten
if (_handler == nullptr) return;
// we no longer have to monitor the socket
_handler->monitor(_connection, _socket, 0);
// close the socket
close(_socket);
}
/**
* The filedescriptor of this connection
* @return int
*/
virtual int fileno() const override { return _socket; }
/**
* Process the filedescriptor in the object
* @param fd The filedescriptor that is active
* @param flags AMQP::readable and/or AMQP::writable
* @return New implementation object
*/
virtual TcpState *process(int fd, int flags)
{
// the socket must be the one this connection writes to
if (fd != _socket) return this;
// because the object might soon be destructed, we create a monitor to check this
Monitor monitor(this);
// are we busy with sending or receiving data?
if (_state == state_sending)
{
// try to send more data from the outgoing buffer
auto result = _out.sendto(_ssl);
// if this is a success, we can proceed with the event loop
if (result > 0) return proceed();
// the operation failed, we may have to repeat our call
else return repeat(result);
}
else
{
// read data from ssl into the buffer
auto result = _in.receivefrom(_ssl, _connection->expected());
// if this is a success, we may have to update the monitor
if (result > 0) return parse(result);
// the operation failed, we may have to repeat our call
else return repeat(result);
}
}
/**
* Send data over the connection
* @param buffer buffer to send
* @param size size of the buffer
*/
virtual void send(const char *buffer, size_t size)
{
// put the data in the outgoing buffer
_out.add(buffer, size);
// if we're already busy with sending or receiving, we first have to wait
// for that operation to complete before we can move on
if (_state != state_idle) return;
// object is now busy sending
_state = state_sending;
// let's wait until the socket becomes writable
_handler->monitor(_connection, _socket, readable | writable);
}
/**
* Report that heartbeat negotiation is going on
* @param heartbeat suggested heartbeat
* @return uint16_t accepted heartbeat
*/
virtual uint16_t reportNegotiate(uint16_t heartbeat) override
{
// remember that we have to reallocate (_in member can not be accessed because it is moved away)
_reallocate = _connection->maxFrame();
// pass to base
return TcpState::reportNegotiate(heartbeat);
}
/**
* Report to the handler that the connection was nicely closed
*/
virtual void reportClosed() override
{
// remember that the object is closed
_closed = true;
// if the previous operation is still in progress
if (_state != state_idle) return;
// wait until the connection is writable
_handler->monitor(_connection, _socket, writable);
}
};
/**
* End of namespace
*/
}

86
src/linux_tcp/sslcontext.h Normal file
View File

@ -0,0 +1,86 @@
/**
* SslContext.h
*
* Class to create and maintain a tcp ssl context
*
* @author Emiel Bruijntjes <emiel.bruijntjes@copernica.com>
* @copyright 2018 Copernica BV
*/
/**
* Include guard
*/
#pragma once
/**
* Begin of namespace
*/
namespace AMQP {
/**
* Class definition
*/
class SslContext
{
private:
/**
* The wrapped context
* @var SSL_CTX
*/
SSL_CTX *_ctx;
public:
/**
* Constructor
* @param method
* @throws std::runtime_error
*/
SslContext(const SSL_METHOD *method) : _ctx(SSL_CTX_new(method))
{
// report error
if (_ctx == nullptr) throw std::runtime_error("failed to construct ssl context");
}
/**
* Constructor that wraps around an existing context
* @param context
*/
SslContext(SSL_CTX *context) : _ctx(context)
{
// increment refcount
// @todo fix this
//SSL_ctx_up_ref(context);
}
/**
* Copy constructor
* @param that
*/
SslContext(SslContext &that) : _ctx(that._ctx)
{
// increment refcount
// @todo fix this
//SSL_ctx_up_ref(context);
}
/**
* Destructor
*/
virtual ~SslContext()
{
// free resource (this updates the refcount -1, and may destruct it)
SSL_CTX_free(_ctx);
}
/**
* Cast to the actual context
* @return SSL_CTX *
*/
operator SSL_CTX * () { return _ctx; }
};
/**
* End of namespace
*/
}

252
src/linux_tcp/sslhandshake.h Normal file
View File

@ -0,0 +1,252 @@
/**
* SslHandshake.h
*
* Implementation of the TCP state that is responsible for setting
* up the STARTTLS handshake.
*
* @copyright 2018 Copernica BV
*/
/**
* Include guard
*/
#pragma once
/**
* Dependencies
*/
#include "tcpoutbuffer.h"
#include "sslconnected.h"
#include "wait.h"
#include "sslwrapper.h"
#include "sslcontext.h"
/**
* Set up namespace
*/
namespace AMQP {
/**
* Class definition
*/
class SslHandshake : public TcpState, private Watchable
{
private:
/**
* SSL structure
* @var SslWrapper
*/
SslWrapper _ssl;
/**
* The socket file descriptor
* @var int
*/
int _socket;
/**
* The outgoing buffer
* @var TcpOutBuffer
*/
TcpOutBuffer _out;
/**
* Report a new state
* @param state
* @return TcpState
*/
TcpState *nextstate(TcpState *state)
{
// forget the socket to prevent that it is closed by the destructor
_socket = -1;
// done
return state;
}
/**
* Helper method to report an error
* @return TcpState*
*/
TcpState *reportError()
{
// we are no longer interested in any events for this socket
_handler->monitor(_connection, _socket, 0);
// we have an error - report this to the user
_handler->onError(_connection, "failed to setup ssl connection");
// done, go to the closed state
return new TcpClosed(_connection, _handler);
}
/**
* Proceed with the handshake
* @param events the events to wait for on the socket
* @return TcpState
*/
TcpState *proceed(int events)
{
// tell the handler that we want to listen for certain events
_handler->monitor(_connection, _socket, events);
// allow chaining
return this;
}
public:
/**
* Constructor
*
* @todo catch the exception!
*
* @param connection Parent TCP connection object
* @param socket The socket filedescriptor
* @param hostname The hostname to connect to
* @param context SSL context
* @param buffer The buffer that was already built
* @param handler User-supplied handler object
* @throws std::runtime_error
*/
SslHandshake(TcpConnection *connection, int socket, const std::string &hostname, TcpOutBuffer &&buffer, TcpHandler *handler) :
TcpState(connection, handler),
_ssl(SslContext(SSLv23_client_method())),
_socket(socket),
_out(std::move(buffer))
{
// we will be using the ssl context as a client
SSL_set_connect_state(_ssl);
// associate domain name with the connection
SSL_set_tlsext_host_name(_ssl, hostname.data());
// associate the ssl context with the socket filedescriptor
if (SSL_set_fd(_ssl, socket) == 0) throw std::runtime_error("failed to associate filedescriptor with ssl socket");
// we are going to wait until the socket becomes writable before we start the handshake
_handler->monitor(_connection, _socket, writable);
}
/**
* Destructor
*/
virtual ~SslHandshake() noexcept
{
// leap out if socket is invalidated
if (_socket < 0) return;
// close the socket
close(_socket);
}
/**
* The filedescriptor of this connection
* @return int
*/
virtual int fileno() const override { return _socket; }
/**
* Process the filedescriptor in the object
* @param fd Filedescriptor that is active
* @param flags AMQP::readable and/or AMQP::writable
* @return New state object
*/
virtual TcpState *process(int fd, int flags) override
{
// must be the socket
if (fd != _socket) return this;
// start the ssl handshake
int result = SSL_do_handshake(_ssl);
// if the connection succeeds, we can move to the ssl-connected state
if (result == 1) return nextstate(new SslConnected(_connection, _socket, _ssl, std::move(_out), _handler));
// error was returned, so we must investigate what is going on
auto error = SSL_get_error(_ssl, result);
// check the error
switch (error) {
case SSL_ERROR_WANT_READ: return proceed(readable);
case SSL_ERROR_WANT_WRITE: return proceed(readable | writable);
default: return reportError();
}
}
/**
* Send data over the connection
* @param buffer buffer to send
* @param size size of the buffer
*/
virtual void send(const char *buffer, size_t size) override
{
// the handshake is still busy, outgoing data must be cached
_out.add(buffer, size);
}
/**
* Flush the connection, sent all buffered data to the socket
* @return TcpState new tcp state
*/
virtual TcpState *flush() override
{
// create an object to wait for the filedescriptor to becomes active
Wait wait(_socket);
// keep looping
while (true)
{
// start the ssl handshake
int result = SSL_do_handshake(_ssl);
// if the connection succeeds, we can move to the ssl-connected state
if (result == 1) return nextstate(new SslConnected(_connection, _socket, _ssl, std::move(_out), _handler));
// error was returned, so we must investigate what is going on
auto error = SSL_get_error(_ssl, result);
// check the error
switch (error)
{
// if openssl reports that socket readability or writability is needed,
// we wait for that until this situation is reached
case SSL_ERROR_WANT_READ: wait.readable(); break;
case SSL_ERROR_WANT_WRITE: wait.active(); break;
// something is wrong, we proceed to the next state
default: return reportError();
}
}
}
/**
* Report to the handler that the connection was nicely closed
*/
virtual void reportClosed() override
{
// we no longer have to monitor the socket
_handler->monitor(_connection, _socket, 0);
// close the socket
close(_socket);
// socket is closed now
_socket = -1;
// copy the handler (if might destruct this object)
auto *handler = _handler;
// reset member before the handler can make a mess of it
_handler = nullptr;
// notify to handler
handler->onClosed(_connection);
}
};
/**
* End of namespace
*/
}

161
src/linux_tcp/sslshutdown.h Normal file
View File

@ -0,0 +1,161 @@
/**
* SslShutdown.h
*
* Class that takes care of the final handshake to close a SSL connection
*
* @author Emiel Bruijntjes <emiel.bruijntjes@copernica.com>
* @copyright 2018 Copernica BV
*/
/**
* Include guard
*/
#pragma once
/**
* Begin of namespace
*/
namespace AMQP {
/**
* Class definition
*/
class SslShutdown : public TcpState, private Watchable
{
private:
/**
* The SSL context
* @var SslWrapper
*/
SslWrapper _ssl;
/**
* Socket file descriptor
* @var int
*/
int _socket;
/**
* Proceed with the next operation after the previous operation was
* a success, possibly changing the filedescriptor-monitor
* @return TcpState*
*/
TcpState *proceed()
{
// construct monitor to prevent that we access members if object is destructed
Monitor monitor(this);
// we're no longer interested in events
_handler->monitor(_connection, _socket, 0);
// stop if object was destructed
if (!monitor) return nullptr;
// close the socket
close(_socket);
// forget the socket
_socket = -1;
// go to the closed state
return new TcpClosed(_connection, _handler);
}
/**
* Method to repeat the previous call
* @param result result of an earlier openssl operation
* @return TcpState*
*/
TcpState *repeat(int result)
{
// error was returned, so we must investigate what is going on
auto error = SSL_get_error(_ssl, result);
// check the error
switch (error) {
case SSL_ERROR_WANT_READ:
// the operation must be repeated when readable
_handler->monitor(_connection, _socket, readable);
return this;
case SSL_ERROR_WANT_WRITE:
// wait until socket becomes writable again
_handler->monitor(_connection, _socket, readable | writable);
return this;
default:
// @todo check how to handle this
return this;
}
}
public:
/**
* Constructor
* @param connection Parent TCP connection object
* @param socket The socket filedescriptor
* @param ssl The SSL structure
* @param handler User-supplied handler object
*/
SslShutdown(TcpConnection *connection, int socket, const SslWrapper &ssl, TcpHandler *handler) :
TcpState(connection, handler),
_ssl(ssl),
_socket(socket)
{
// tell the handler to monitor the socket if there is an out
_handler->monitor(_connection, _socket, readable);
}
/**
* Destructor
*/
virtual ~SslShutdown() noexcept
{
// skip if handler is already forgotten
if (_handler == nullptr) return;
// we no longer have to monitor the socket
_handler->monitor(_connection, _socket, 0);
// close the socket
close(_socket);
}
/**
* The filedescriptor of this connection
* @return int
*/
virtual int fileno() const override { return _socket; }
/**
* Process the filedescriptor in the object
* @param fd The filedescriptor that is active
* @param flags AMQP::readable and/or AMQP::writable
* @return New implementation object
*/
virtual TcpState *process(int fd, int flags)
{
// the socket must be the one this connection writes to
if (fd != _socket) return this;
// because the object might soon be destructed, we create a monitor to check this
Monitor monitor(this);
// close the connection
auto result = SSL_shutdown(_ssl);
// if this is a success, we can proceed with the event loop
if (result > 0) return proceed();
// the operation failed, we may have to repeat our call
else return repeat(result);
}
};
/**
* End of namespace
*/
}

85
src/linux_tcp/sslwrapper.h Normal file
View File

@ -0,0 +1,85 @@
/**
* SslWrapper.h
*
* Wrapper around a SSL pointer
*
* @author Emiel Bruijntjes <emiel.bruijntjes@copernica.com>
* @copyright 2018 Copernica BV
*/
/**
* Include guard
*/
#pragma once
/**
* Begin of namespace
*/
namespace AMQP {
/**
* Class definition
*/
class SslWrapper
{
private:
/**
* The wrapped object
* @var SSL*
*/
SSL *_ssl;
public:
/**
* Constructor
* @param ctx
*/
SslWrapper(SSL_CTX *ctx) : _ssl(SSL_new(ctx))
{
// report error
if (_ssl == nullptr) throw std::runtime_error("failed to construct ssl structure");
}
/**
* Wrapper constructor
* @param ssl
*/
SslWrapper(SSL *ssl) : _ssl(ssl)
{
// one more reference
// @todo fix this
//CRYPTO_add(_ssl);
}
/**
* Copy constructor
* @param that
*/
SslWrapper(const SslWrapper &that) : _ssl(that._ssl)
{
// one more reference
// @todo fix this
//SSL_up_ref(_ssl);
}
/**
* Destructor
*/
virtual ~SslWrapper()
{
// destruct object
SSL_free(_ssl);
}
/**
* Cast to the SSL*
* @return SSL *
*/
operator SSL * () const { return _ssl; }
};
/**
* End of namespace
*/
}