src/telepathy/telepathy-transport.c

   1 /**
   2  * @file telepathy-transport.c
   3  *
   4  * pidgin-sipe
   5  *
   6  * Copyright (C) 2012-2016 SIPE Project <http://sipe.sourceforge.net/>
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License as published by
  10  * the Free Software Foundation; either version 2 of the License, or
  11  * (at your option) any later version.
  12  *
  13  * This program is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  * GNU General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program; if not, write to the Free Software
  20  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  21  */
  22
  23 #ifdef HAVE_CONFIG_H
  24 #include "config.h"
  25 #endif
  26
  27 #include <string.h>
  28
  29 #include <glib.h>
  30 #include <gio/gio.h>
  31
  32 #include "sipe-backend.h"
  33 #include "sipe-common.h"
  34 #include "sipe-core.h"
  35 #include "sipe-nls.h"
  36
  37 #include "telepathy-private.h"
  38
  39 struct sipe_transport_telepathy {
  40         /* public part shared with core */
  41         struct sipe_transport_connection public;
  42
  43         /* telepathy private part */
  44         transport_connected_cb *connected;
  45         transport_input_cb *input;
  46         transport_error_cb *error;
  47         gchar *hostname;
  48         struct sipe_tls_info *tls_info;
  49         struct sipe_backend_private *private;
  50         GCancellable *cancel;
  51         GSocketConnection *socket;
  52         GInputStream *istream;
  53         GOutputStream *ostream;
  54         GSList *buffers; /* != NULL -> write operation in progress */
  55         guint port;
  56         gboolean do_flush;
  57 };
  58
  59 #define TELEPATHY_TRANSPORT ((struct sipe_transport_telepathy *) conn)
  60 #define SIPE_TRANSPORT_CONNECTION ((struct sipe_transport_connection *) transport)
  61
  62 #define BUFFER_SIZE_INCREMENT 4096
  63
  64 static void read_completed(GObject *stream,
  65                            GAsyncResult *result,
  66                            gpointer data)
  67 {
  68         struct sipe_transport_telepathy *transport = data;
  69         struct sipe_transport_connection *conn = SIPE_TRANSPORT_CONNECTION;
  70
  71         do {
  72                 if (conn->buffer_length < conn->buffer_used + BUFFER_SIZE_INCREMENT) {
  73                         conn->buffer_length += BUFFER_SIZE_INCREMENT;
  74                         conn->buffer = g_realloc(conn->buffer, conn->buffer_length);
  75                         SIPE_DEBUG_INFO("read_completed: new buffer length %" G_GSIZE_FORMAT,
  76                                         conn->buffer_length);
  77                 }
  78
  79                 /* callback result is valid */
  80                 if (result) {
  81                         GError *error = NULL;
  82                         gssize len    = g_input_stream_read_finish(G_INPUT_STREAM(stream),
  83                                                                    result,
  84                                                                    &error);
  85
  86                         if (len < 0) {
  87                                 const gchar *msg = error ? error->message : "UNKNOWN";
  88                                 SIPE_DEBUG_ERROR("read_completed: error: %s", msg);
  89                                 if (transport->error)
  90                                         transport->error(conn, msg);
  91                                 g_error_free(error);
  92                                 return;
  93                         } else if (len == 0) {
  94                                 SIPE_DEBUG_ERROR_NOFORMAT("read_completed: server has disconnected");
  95                                 transport->error(conn, _("Server has disconnected"));
  96                                 return;
  97                         } else if (transport->do_flush) {
  98                                 /* read completed while disconnected transport is flushing */
  99                                 SIPE_DEBUG_INFO_NOFORMAT("read_completed: ignored during flushing");
 100                                 return;
 101                         } else if (g_cancellable_is_cancelled(transport->cancel)) {
 102                                 /* read completed when transport was disconnected */
 103                                 SIPE_DEBUG_INFO_NOFORMAT("read_completed: cancelled");
 104                                 return;
 105                         }
 106
 107                         /* Forward data to core */
 108                         conn->buffer_used               += len;
 109                         conn->buffer[conn->buffer_used]  = '\0';
 110                         transport->input(conn);
 111
 112                         /* we processed the result */
 113                         result = NULL;
 114                 }
 115
 116                 /* buffer too short? */
 117         } while (conn->buffer_length - conn->buffer_used - 1 == 0);
 118
 119         /* setup next read */
 120         g_input_stream_read_async(G_INPUT_STREAM(stream),
 121                                   conn->buffer + conn->buffer_used,
 122                                   conn->buffer_length - conn->buffer_used - 1,
 123                                   G_PRIORITY_DEFAULT,
 124                                   transport->cancel,
 125                                   read_completed,
 126                                   transport);
 127 }
 128
 129 static gboolean internal_connect(gpointer data);
 130 static void certificate_result(SIPE_UNUSED_PARAMETER GObject *unused,
 131                                GAsyncResult *result,
 132                                gpointer data)
 133 {
 134         struct sipe_transport_telepathy *transport = data;
 135         GError *error = NULL;
 136
 137         g_simple_async_result_propagate_error(G_SIMPLE_ASYNC_RESULT(result),
 138                                               &error);
 139         if (error) {
 140                 SIPE_DEBUG_INFO("certificate_result: %s", error->message);
 141                 if (transport->error)
 142                         transport->error(SIPE_TRANSPORT_CONNECTION,
 143                                          error->message);
 144                 g_error_free(error);
 145         } else {
 146                 SIPE_DEBUG_INFO("certificate_result: trigger reconnect %p", transport);
 147                 g_idle_add(internal_connect, transport);
 148         }
 149 }
 150
 151 static void socket_connected(GObject *client,
 152                              GAsyncResult *result,
 153                              gpointer data)
 154 {
 155         struct sipe_transport_telepathy *transport = data;
 156         GError *error = NULL;
 157
 158         transport->socket = g_socket_client_connect_finish(G_SOCKET_CLIENT(client),
 159                                                            result,
 160                                                            &error);
 161
 162         if (transport->socket == NULL) {
 163                 if (transport->tls_info) {
 164                         SIPE_DEBUG_INFO_NOFORMAT("socket_connected: need to wait for user interaction");
 165                         sipe_telepathy_tls_verify_async(G_OBJECT(transport->private->connection),
 166                                                         transport->tls_info,
 167                                                         certificate_result,
 168                                                         transport);
 169                 } else {
 170                         const gchar *msg = error ? error->message : "UNKNOWN";
 171                         SIPE_DEBUG_ERROR("socket_connected: failed: %s", msg);
 172                         if (transport->error)
 173                                 transport->error(SIPE_TRANSPORT_CONNECTION, msg);
 174                         g_error_free(error);
 175                 }
 176         } else if (g_cancellable_is_cancelled(transport->cancel)) {
 177                 /* connect already succeeded when transport was disconnected */
 178                 g_object_unref(transport->socket);
 179                 transport->socket = NULL;
 180                 SIPE_DEBUG_INFO_NOFORMAT("socket_connected: succeeded, but cancelled");
 181         } else {
 182                 GSocketAddress *saddr = g_socket_connection_get_local_address(transport->socket,
 183                                                                               &error);
 184
 185                 if (saddr) {
 186                         SIPE_DEBUG_INFO_NOFORMAT("socket_connected: success");
 187
 188                         transport->public.client_port = g_inet_socket_address_get_port(G_INET_SOCKET_ADDRESS(saddr));
 189                         g_object_unref(saddr);
 190
 191                         transport->istream = g_io_stream_get_input_stream(G_IO_STREAM(transport->socket));
 192                         transport->ostream = g_io_stream_get_output_stream(G_IO_STREAM(transport->socket));
 193
 194                         /* the first connection is always to the server */
 195                         if (transport->private->transport == NULL)
 196                                 transport->private->transport = transport;
 197
 198                         /* this sets up the async read handler */
 199                         read_completed(G_OBJECT(transport->istream), NULL, transport);
 200                         transport->connected(SIPE_TRANSPORT_CONNECTION);
 201
 202                 } else {
 203                         g_object_unref(transport->socket);
 204                         transport->socket = NULL;
 205                         SIPE_DEBUG_ERROR("socket_connected: failed: %s", error->message);
 206                         transport->error(SIPE_TRANSPORT_CONNECTION, error->message);
 207                         g_error_free(error);
 208                 }
 209         }
 210 }
 211
 212 static gboolean accept_certificate_signal(SIPE_UNUSED_PARAMETER GTlsConnection *tls,
 213                                           GTlsCertificate *peer_cert,
 214                                           SIPE_UNUSED_PARAMETER GTlsCertificateFlags errors,
 215                                           gpointer user_data)
 216 {
 217         struct sipe_transport_telepathy *transport = user_data;
 218
 219         SIPE_DEBUG_INFO("accept_certificate_signal: %p", transport);
 220
 221         /* second connection attempt after feedback from user? */
 222         if (transport->tls_info) {
 223                 /* user accepted certificate */
 224                 sipe_telepathy_tls_info_free(transport->tls_info);
 225                 transport->tls_info = NULL;
 226                 return(TRUE);
 227         } else {
 228                 /* retry after user accepted certificate */
 229                 transport->tls_info = sipe_telepathy_tls_info_new(transport->hostname,
 230                                                                   peer_cert);
 231                 return(FALSE);
 232         }
 233 }
 234
 235 static void tls_handshake_starts(SIPE_UNUSED_PARAMETER GSocketClient *client,
 236                                  GSocketClientEvent event,
 237                                  SIPE_UNUSED_PARAMETER GSocketConnectable *connectable,
 238                                  GIOStream *connection,
 239                                  gpointer user_data)
 240 {
 241         if (event == G_SOCKET_CLIENT_TLS_HANDSHAKING) {
 242                 SIPE_DEBUG_INFO("tls_handshake_starts: %p", connection);
 243                 g_signal_connect(connection, /* is a GTlsConnection */
 244                                  "accept-certificate",
 245                                  G_CALLBACK(accept_certificate_signal),
 246                                  user_data);
 247         }
 248 }
 249
 250 static gboolean internal_connect(gpointer data)
 251 {
 252         struct sipe_transport_telepathy *transport = data;
 253         GSocketClient *client = g_socket_client_new();
 254
 255         SIPE_DEBUG_INFO("internal_connect - hostname: %s port: %d",
 256                         transport->hostname, transport->port);
 257
 258         /* request TLS connection */
 259         if (transport->public.type == SIPE_TRANSPORT_TLS) {
 260                 SIPE_DEBUG_INFO_NOFORMAT("using TLS");
 261                 g_socket_client_set_tls(client, TRUE);
 262                 g_signal_connect(client,
 263                                  "event",
 264                                  G_CALLBACK(tls_handshake_starts),
 265                                  transport);
 266         } else
 267                 SIPE_DEBUG_INFO_NOFORMAT("using TCP");
 268
 269         g_socket_client_connect_async(client,
 270                                       g_network_address_new(transport->hostname,
 271                                                             transport->port),
 272                                       transport->cancel,
 273                                       socket_connected,
 274                                       transport);
 275         g_object_unref(client);
 276
 277         return(FALSE);
 278 }
 279
 280 struct sipe_transport_connection *sipe_backend_transport_connect(struct sipe_core_public *sipe_public,
 281                                                                  const sipe_connect_setup *setup)
 282 {
 283         struct sipe_transport_telepathy *transport = g_new0(struct sipe_transport_telepathy, 1);
 284
 285         transport->public.type      = setup->type;
 286         transport->public.user_data = setup->user_data;
 287         transport->connected        = setup->connected;
 288         transport->input            = setup->input;
 289         transport->error            = setup->error;
 290         transport->hostname         = g_strdup(setup->server_name);
 291         transport->tls_info         = NULL;
 292         transport->private          = sipe_public->backend_private;
 293         transport->cancel           = g_cancellable_new();
 294         transport->buffers          = NULL;
 295         transport->port             = setup->server_port;
 296         transport->do_flush         = FALSE;
 297
 298         if ((setup->type == SIPE_TRANSPORT_TLS) ||
 299             (setup->type == SIPE_TRANSPORT_TCP)) {
 300
 301                 internal_connect(transport);
 302                 return(SIPE_TRANSPORT_CONNECTION);
 303
 304         } else {
 305                 setup->error(SIPE_TRANSPORT_CONNECTION,
 306                              "This should not happen...");
 307                 sipe_backend_transport_disconnect(SIPE_TRANSPORT_CONNECTION);
 308                 return(NULL);
 309         }
 310 }
 311
 312 static gboolean free_transport(gpointer data)
 313 {
 314         struct sipe_transport_telepathy *transport = data;
 315         GSList *entry;
 316
 317         SIPE_DEBUG_INFO("free_transport %p", transport);
 318
 319         if (transport->tls_info)
 320                 sipe_telepathy_tls_info_free(transport->tls_info);
 321         g_free(transport->hostname);
 322
 323         /* free unflushed buffers */
 324         for (entry = transport->buffers; entry; entry = entry->next)
 325                 g_free(entry->data);
 326         g_slist_free(transport->buffers);
 327
 328         if (transport->cancel)
 329                 g_object_unref(transport->cancel);
 330
 331         g_free(transport);
 332
 333         return(FALSE);
 334 }
 335
 336 static void close_completed(GObject *stream,
 337                             GAsyncResult *result,
 338                             gpointer data)
 339 {
 340         struct sipe_transport_telepathy *transport = data;
 341         SIPE_DEBUG_INFO("close_completed: transport %p", data);
 342         g_io_stream_close_finish(G_IO_STREAM(stream), result, NULL);
 343         g_idle_add(free_transport, transport);
 344 }
 345
 346 static void do_close(struct sipe_transport_telepathy *transport)
 347 {
 348         SIPE_DEBUG_INFO("do_close: %p", transport);
 349
 350         /* cancel outstanding asynchronous operations */
 351         transport->do_flush = FALSE;
 352         g_cancellable_cancel(transport->cancel);
 353         g_io_stream_close_async(G_IO_STREAM(transport->socket),
 354                                 G_PRIORITY_DEFAULT,
 355                                 NULL,
 356                                 close_completed,
 357                                 transport);
 358 }
 359
 360 void sipe_backend_transport_disconnect(struct sipe_transport_connection *conn)
 361 {
 362         struct sipe_transport_telepathy *transport = TELEPATHY_TRANSPORT;
 363
 364         if (!transport) return;
 365
 366         SIPE_DEBUG_INFO("sipe_backend_transport_disconnect: %p", transport);
 367
 368         /* error callback is invalid now, do no longer call! */
 369         transport->error = NULL;
 370
 371         /* dropping connection to the server? */
 372         if (transport->private->transport == transport)
 373                 transport->private->transport = NULL;
 374
 375         /* already connected? */
 376         if (transport->socket) {
 377
 378                 /* flush required? */
 379                 if (transport->do_flush && transport->buffers)
 380                         SIPE_DEBUG_INFO("sipe_backend_transport_disconnect: %p needs flushing",
 381                                         transport);
 382                 else
 383                         do_close(transport);
 384
 385         } else {
 386                 /* cancel outstanding connect operation */
 387                 if (transport->cancel)
 388                         g_cancellable_cancel(transport->cancel);
 389
 390                 /* queue transport to be deleted */
 391                 g_idle_add(free_transport, transport);
 392         }
 393 }
 394
 395 gchar *sipe_backend_transport_ip_address(struct sipe_transport_connection *conn)
 396 {
 397         struct sipe_transport_telepathy *transport = TELEPATHY_TRANSPORT;
 398         gchar *ipstr = NULL;
 399
 400         if (transport && transport->socket) {
 401                 GSocketAddress *saddr = g_socket_connection_get_local_address(transport->socket,
 402                                                                               NULL);
 403
 404                 if (saddr) {
 405                         GInetAddress *iaddr = g_inet_socket_address_get_address(G_INET_SOCKET_ADDRESS(saddr));
 406
 407                         if (iaddr) {
 408                                 ipstr = g_inet_address_to_string(iaddr);
 409                                 SIPE_DEBUG_INFO("sipe_backend_transport_ip_address: %s", ipstr);
 410                         }
 411                         g_object_unref(saddr);
 412                 }
 413         }
 414
 415         return(ipstr ? ipstr : g_strdup("0.0.0.0"));
 416 }
 417
 418 static void do_write(struct sipe_transport_telepathy *transport);
 419 static void write_completed(GObject *stream,
 420                             GAsyncResult *result,
 421                             gpointer data)
 422 {
 423         struct sipe_transport_telepathy *transport = data;
 424         gchar                           *buffer;
 425         GError                          *error     = NULL;
 426         gssize written = g_output_stream_write_finish(G_OUTPUT_STREAM(stream),
 427                                                       result,
 428                                                       &error);
 429
 430         /* free the buffer that has just been written. */
 431         buffer = transport->buffers->data;
 432         transport->buffers = g_slist_remove(transport->buffers, buffer);
 433         g_free(buffer);
 434
 435         if ((written < 0) || error) {
 436                 const gchar *msg = error ? error->message : "UNKNOWN";
 437                 SIPE_DEBUG_ERROR("write_completed: error: %s", msg);
 438                 if (transport->error)
 439                         transport->error(SIPE_TRANSPORT_CONNECTION, msg);
 440                 g_error_free(error);
 441
 442                 /* error during flush: give up and close transport */
 443                 if (transport->do_flush)
 444                         do_close(transport);
 445
 446         } else if (g_cancellable_is_cancelled(transport->cancel)) {
 447                 /* write completed when transport was disconnected */
 448                 SIPE_DEBUG_INFO_NOFORMAT("write_completed: cancelled");
 449         } else {
 450                 /* more to write? */
 451                 if (transport->buffers) {
 452                         do_write(transport);
 453                 /* flush completed? */
 454                 } else if (transport->do_flush) {
 455                         do_close(transport);
 456                 }
 457         }
 458 }
 459
 460 static void do_write(struct sipe_transport_telepathy *transport)
 461 {
 462         g_output_stream_write_async(transport->ostream,
 463                                     transport->buffers->data,
 464                                     strlen(transport->buffers->data),
 465                                     G_PRIORITY_DEFAULT,
 466                                     transport->cancel,
 467                                     write_completed,
 468                                     transport);
 469 }
 470
 471 void sipe_backend_transport_message(struct sipe_transport_connection *conn,
 472                                     const gchar *buffer)
 473 {
 474         struct sipe_transport_telepathy *transport = TELEPATHY_TRANSPORT;
 475         gboolean can_write = (transport->buffers == NULL);
 476
 477         transport->buffers = g_slist_append(transport->buffers, g_strdup(buffer));
 478
 479         if (can_write)
 480                 do_write(transport);
 481 }
 482
 483 void sipe_backend_transport_flush(struct sipe_transport_connection *conn)
 484 {
 485         struct sipe_transport_telepathy *transport = TELEPATHY_TRANSPORT;
 486         transport->do_flush = TRUE;
 487 }
 488
 489
 490 /*
 491   Local Variables:
 492   mode: c
 493   c-file-style: "bsd"
 494   indent-tabs-mode: t
 495   tab-width: 8
 496   End:
 497 */