natas_hw
/
paho
mirror of https://github.com/eclipse/paho.mqtt.c
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Proxy support for websocket. 

Signed-off-by: Tamotsu Kanoh <kanoh@plathome.co.jp>
This commit is contained in:
Tamotsu Kanoh 2018-09-16 00:40:45 +09:00
parent 9f715d0862
commit 0dc0049721
6 changed files with 128 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/Clients.h
View File

@ -82,7 +82,9 @@ typedef struct
#if defined(OPENSSL)
SSL* ssl;
SSL_CTX* ctx;
char *https_proxy;
#endif
char *http_proxy;
int websocket; /**< socket has been upgraded to use web sockets */
char *websocket_key;
} networkHandles;
@ -99,6 +101,8 @@ typedef struct
#define WEBSOCKET_IN_PROGRESS 0x3
/** TCP completed, waiting for MQTT ACK */
#define WAIT_FOR_CONNACK 0x4
/** Proxy connection in progress */
#define PROXY_CONNECY_IN_PROGRESS 0x5
/** Disconnecting */
#define DISCONNECTING -2

12
src/MQTTAsync.c
View File

@ -3531,6 +3531,12 @@ static int MQTTAsync_connecting(MQTTAsyncs* m)
size_t hostname_len;
int setSocketForSSLrc = 0;
if (m->websocket && m->c->net.https_proxy) {
m->c->connect_state = PROXY_CONNECY_IN_PROGRESS;
if ((rc = WebSocket_proxy_connect( &m->c->net, m->serverURI)) == SOCKET_ERROR )
goto exit;
}
hostname_len = MQTTProtocol_addressPort(m->serverURI, &port, NULL);
setSocketForSSLrc = SSLSocket_setSocketForSSL(&m->c->net, m->c->sslopts,
m->serverURI, hostname_len);
@ -3589,6 +3595,12 @@ static int MQTTAsync_connecting(MQTTAsyncs* m)
#endif
if ( m->websocket )
{
if (m->c->net.http_proxy) {
m->c->connect_state = PROXY_CONNECY_IN_PROGRESS;
if ((rc = WebSocket_proxy_connect( &m->c->net, m->serverURI)) == SOCKET_ERROR )
goto exit;
}
m->c->connect_state = WEBSOCKET_IN_PROGRESS;
if ((rc = WebSocket_connect(&m->c->net, m->serverURI)) == SOCKET_ERROR )
goto exit;

18
src/MQTTClient.c
View File

@ -897,6 +897,7 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
}
}
#endif
else if (m->c->connect_state == WEBSOCKET_IN_PROGRESS)
{
Log(TRACE_MIN, -1, "Posting websocket handshake for client %s rc %d", m->c->clientID, m->rc);
@ -1132,6 +1133,12 @@ static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_c
const char *topic;
int setSocketForSSLrc = 0;
if (m->websocket && m->c->net.https_proxy) {
m->c->connect_state = PROXY_CONNECY_IN_PROGRESS;
if ((rc = WebSocket_proxy_connect( &m->c->net, m->serverURI)) == SOCKET_ERROR )
goto exit;
}
hostname_len = MQTTProtocol_addressPort(m->serverURI, &port, &topic);
setSocketForSSLrc = SSLSocket_setSocketForSSL(&m->c->net, m->c->sslopts,
m->serverURI, hostname_len);
@ -1185,6 +1192,12 @@ static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_c
#endif
else if (m->websocket)
{
if (m->c->net.http_proxy) {
m->c->connect_state = PROXY_CONNECY_IN_PROGRESS;
if ((rc = WebSocket_proxy_connect( &m->c->net, m->serverURI)) == SOCKET_ERROR )
goto exit;
}
m->c->connect_state = WEBSOCKET_IN_PROGRESS;
if ( WebSocket_connect(&m->c->net, m->serverURI) == SOCKET_ERROR )
{
@ -2434,6 +2447,11 @@ static MQTTPacket* MQTTClient_waitfor(MQTTClient handle, int packet_type, int* r
*rc = 1;
break;
}
else if (m->c->connect_state == PROXY_CONNECY_IN_PROGRESS )
{
*rc = 1;
break;
}
else if (m->c->connect_state == WAIT_FOR_CONNACK)
{
int error;

40
src/MQTTProtocolOut.c
View File

@ -110,12 +110,36 @@ int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int websocket
{
int rc, port;
size_t addr_len;
char *p;
FUNC_ENTRY;
aClient->good = 1;
addr_len = MQTTProtocol_addressPort(ip_address, &port, NULL);
rc = Socket_new(ip_address, addr_len, port, &(aClient->net.socket));
aClient->net.http_proxy = NULL;
if ((p = getenv("http_proxy")))
aClient->net.http_proxy = strchr(p, ':') + 3;
#if defined(OPENSSL)
aClient->net.https_proxy = NULL;
if ((p = getenv("https_proxy")))
aClient->net.https_proxy = strchr(p, ':') + 3;
if (!ssl && websocket && aClient->net.http_proxy) {
#else
if (websocket && aClient->net.http_proxy) {
#endif
addr_len = MQTTProtocol_addressPort(aClient->net.http_proxy, &port, NULL);
rc = Socket_new(aClient->net.http_proxy, addr_len, port, &(aClient->net.socket));
}
#if defined(OPENSSL)
else if (ssl && websocket && aClient->net.https_proxy) {
addr_len = MQTTProtocol_addressPort(aClient->net.https_proxy, &port, NULL);
rc = Socket_new(aClient->net.https_proxy, addr_len, port, &(aClient->net.socket));
}
#endif
else {
addr_len = MQTTProtocol_addressPort(ip_address, &port, NULL);
rc = Socket_new(ip_address, addr_len, port, &(aClient->net.socket));
}
if (rc == EINPROGRESS || rc == EWOULDBLOCK)
aClient->connect_state = TCP_IN_PROGRESS; /* TCP connect called - wait for connect completion */
else if (rc == 0)
@ -123,7 +147,11 @@ int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int websocket
#if defined(OPENSSL)
if (ssl)
{
if (SSLSocket_setSocketForSSL(&aClient->net, aClient->sslopts, ip_address, addr_len) == 1)
if (websocket && aClient->net.https_proxy) {
aClient->connect_state = PROXY_CONNECY_IN_PROGRESS;
rc = WebSocket_proxy_connect( &aClient->net, ip_address);
}
if (rc == 0 && SSLSocket_setSocketForSSL(&aClient->net, aClient->sslopts, ip_address, addr_len) == 1)
{
rc = aClient->sslopts->struct_version >= 3 ?
SSLSocket_connect(aClient->net.ssl, aClient->net.socket, ip_address,
@ -136,7 +164,13 @@ int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int websocket
else
rc = SOCKET_ERROR;
}
else if (websocket && aClient->net.http_proxy) {
#else
if (websocket && aClient->net.http_proxy) {
#endif
aClient->connect_state = PROXY_CONNECY_IN_PROGRESS;
rc = WebSocket_proxy_connect( &aClient->net, ip_address);
}
if ( websocket )
{
rc = WebSocket_connect( &aClient->net, ip_address );

55
src/WebSocket.c
View File

@ -89,6 +89,8 @@
/** @brief raw uuid type */
typedef unsigned char uuid_t[16];
#define HTTP_PROTOCOL(x) x ? "https" : "http"
/**
* @brief generates a uuid, compatible with RFC 4122, version 4 (random)
* @note Uses a very insecure algorithm but no external dependencies
@ -345,12 +347,18 @@ int WebSocket_connect( networkHandles *net, const char *uri )
"Host: %.*s:%d\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
"Origin: http://%.*s:%d\r\n"
"Origin: %s://%.*s:%d\r\n"
"Sec-WebSocket-Key: %s\r\n"
"Sec-WebSocket-Version: 13\r\n"
"Sec-WebSocket-Protocol: mqtt\r\n"
"\r\n", topic,
(int)hostname_len, uri, port,
#if defined(OPENSSL)
HTTP_PROTOCOL(net->ssl),
#else
HTTP_PROTOCOL(0),
#endif
(int)hostname_len, uri, port,
net->websocket_key );
@ -1064,3 +1072,48 @@ exit:
return rc;
}
/**
* Notify the IP address and port of the endpoint to proxy, and wait connection to endpoint.
*
* @param[in] net network connection to proxy.
* @param[in] hostname hostname of endpoint.
*
* @retval SOCKET_ERROR failed to network connection
* @retval 0 connection to endpoint
*
*/
int WebSocket_proxy_connect( networkHandles *net, const char *hostname)
{
int port, i, rc = 0, buf_len=0;
char *buf = NULL;
size_t hostname_len, actual_len = 0;
FUNC_ENTRY;
hostname_len = MQTTProtocol_addressPort(hostname, &port, NULL);
for ( i = 0; i < 2; ++i ) {
buf_len = snprintf( buf, (size_t)buf_len, "CONNECT %.*s:%d HTTP/1.1\r\n\r\n",
(int)hostname_len, hostname, port);
if ( i==0 && buf_len > 0 ) {
++buf_len;
buf = malloc( buf_len );
}
}
Socket_putdatas( net->socket, buf, buf_len, 0, NULL, NULL, NULL );
free(buf);
buf = NULL;
while(!actual_len)
buf = Socket_getdata(net->socket, (size_t)12, &actual_len);
if ( strcmp( buf, "HTTP/1.1 200" ) != 0 )
rc = SOCKET_ERROR;
/* flash the SocketBuffer */
actual_len = 1;
while(actual_len)
buf = Socket_getdata(net->socket, (size_t)1, &actual_len);
FUNC_EXIT_RC(rc);
return rc;
}

3
src/WebSocket.h
View File

@ -74,4 +74,7 @@ void WebSocket_terminate(void);
/* handles websocket upgrade request */
int WebSocket_upgrade(networkHandles *net);
/* Notify the IP address and port of the endpoint to proxy, and wait connection to endpoint */
int WebSocket_proxy_connect( networkHandles *net, const char *hostname);
#endif /* WEBSOCKET_H */