src/telepathy/telepathy-transport.c

   1 /**
   2  * @file telepathy-transport.c
   3  *
   4  * pidgin-sipe
   5  *
   6  * Copyright (C) 2012-2013 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;
  55         guint port;
  56         gboolean is_writing;
  57         gboolean do_flush;
  58 };
  59
  60 #define TELEPATHY_TRANSPORT ((struct sipe_transport_telepathy *) conn)
  61 #define SIPE_TRANSPORT_CONNECTION ((struct sipe_transport_connection *) transport)
  62
  63 #define BUFFER_SIZE_INCREMENT 4096
  64
  65 static void read_completed(GObject *stream,
  66                            GAsyncResult *result,
  67                            gpointer data)
  68 {
  69         struct sipe_transport_telepathy *transport = data;
  70         struct sipe_transport_connection *conn = SIPE_TRANSPORT_CONNECTION;
  71
  72         do {
  73                 if (conn->buffer_length < conn->buffer_used + BUFFER_SIZE_INCREMENT) {
  74                         conn->buffer_length += BUFFER_SIZE_INCREMENT;
  75                         conn->buffer = g_realloc(conn->buffer, conn->buffer_length);
  76                         SIPE_DEBUG_INFO("read_completed: new buffer length %" G_GSIZE_FORMAT,
  77                                         conn->buffer_length);
  78                 }
  79
  80                 /* callback result is valid */
  81                 if (result) {
  82                         GError *error = NULL;
  83                         gssize len    = g_input_stream_read_finish(G_INPUT_STREAM(stream),
  84                                                                    result,
  85                                                                    &error);
  86
  87                         if (len < 0) {
  88                                 const gchar *msg = error ? error->message : "UNKNOWN";
  89                                 SIPE_DEBUG_ERROR("read_completed: error: %s", msg);
  90                                 if (transport->error)
  91                                         transport->error(conn, msg);
  92                                 g_error_free(error);
  93                                 return;
  94                         } else if (len == 0) {
  95                                 SIPE_DEBUG_ERROR_NOFORMAT("read_completed: server has disconnected");
  96                                 transport->error(conn, _("Server has disconnected"));
  97                                 return;
  98                         } else if (transport->do_flush) {
  99                                 /* read completed while disconnected transport is flushing */
 100                                 SIPE_DEBUG_INFO_NOFORMAT("read_completed: ignored during flushing");
 101                                 return;
 102                         } else if (g_cancellable_is_cancelled(transport->cancel)) {
 103                                 /* read completed when transport was disconnected */
 104                                 SIPE_DEBUG_INFO_NOFORMAT("read_completed: cancelled");
 105                                 return;
 106                         }
 107
 108                         /* Forward data to core */
 109                         conn->buffer_used               += len;
 110                         conn->buffer[conn->buffer_used]  = '\0';
 111                         transport->input(conn);
 112
 113                         /* we processed the result */
 114                         result = NULL;
 115                 }
 116
 117                 /* buffer too short? */
 118         } while (conn->buffer_length - conn->buffer_used - 1 == 0);
 119
 120         /* setup next read */
 121         g_input_stream_read_async(G_INPUT_STREAM(stream),
 122                                   conn->buffer + conn->buffer_used,
 123                                   conn->buffer_length - conn->buffer_used - 1,
 124                                   G_PRIORITY_DEFAULT,
 125                                   transport->cancel,
 126                                   read_completed,
 127                                   transport);
 128 }
 129
 130 static gboolean internal_connect(gpointer data);
 131 static void certificate_result(SIPE_UNUSED_PARAMETER GObject *unused,
 132                                GAsyncResult *result,
 133                                gpointer data)
 134 {
 135         struct sipe_transport_telepathy *transport = data;
 136         GError *error = NULL;
 137
 138         g_simple_async_result_propagate_error(G_SIMPLE_ASYNC_RESULT(result),
 139                                               &error);
 140         if (error) {
 141                 SIPE_DEBUG_INFO("certificate_result: %s", error->message);
 142                 if (transport->error)
 143                         transport->error(SIPE_TRANSPORT_CONNECTION,
 144                                          error->message);
 145                 g_error_free(error);
 146         } else {
 147                 SIPE_DEBUG_INFO("certificate_result: trigger reconnect %p", transport);
 148                 g_idle_add(internal_connect, transport);
 149         }
 150 }
 151
 152 static void socket_connected(GObject *client,
 153                              GAsyncResult *result,
 154                              gpointer data)
 155 {
 156         struct sipe_transport_telepathy *transport = data;
 157         GError *error = NULL;
 158
 159         transport->socket = g_socket_client_connect_finish(G_SOCKET_CLIENT(client),
 160                                                            result,
 161                                                            &error);
 162
 163         if (transport->socket == NULL) {
 164                 if (transport->tls_info) {
 165                         SIPE_DEBUG_INFO_NOFORMAT("socket_connected: need to wait for user interaction");
 166                         sipe_telepathy_tls_verify_async(G_OBJECT(transport->private->connection),
 167                                                         transport->tls_info,
 168                                                         certificate_result,
 169                                                         transport);
 170                 } else {
 171                         const gchar *msg = error ? error->message : "UNKNOWN";
 172                         SIPE_DEBUG_ERROR("socket_connected: failed: %s", msg);
 173                         if (transport->error)
 174                                 transport->error(SIPE_TRANSPORT_CONNECTION, msg);
 175                         g_error_free(error);
 176                 }
 177         } else if (g_cancellable_is_cancelled(transport->cancel)) {
 178                 /* connect already succeeded when transport was disconnected */
 179                 g_object_unref(transport->socket);
 180                 transport->socket = NULL;
 181                 SIPE_DEBUG_INFO_NOFORMAT("socket_connected: succeeded, but cancelled");
 182         } else {
 183                 GSocketAddress *saddr = g_socket_connection_get_local_address(transport->socket,
 184                                                                               &error);
 185
 186                 if (saddr) {
 187                         SIPE_DEBUG_INFO_NOFORMAT("socket_connected: success");
 188
 189                         transport->public.client_port = g_inet_socket_address_get_port(G_INET_SOCKET_ADDRESS(saddr));
 190                         g_object_unref(saddr);
 191
 192                         transport->istream = g_io_stream_get_input_stream(G_IO_STREAM(transport->socket));
 193                         transport->ostream = g_io_stream_get_output_stream(G_IO_STREAM(transport->socket));
 194
 195                         /* the first connection is always to the server */
 196                         if (transport->private->transport == NULL)
 197                                 transport->private->transport = transport;
 198
 199                         /* this sets up the async read handler */
 200                         read_completed(G_OBJECT(transport->istream), NULL, transport);
 201                         transport->connected(SIPE_TRANSPORT_CONNECTION);
 202
 203                 } else {
 204                         g_object_unref(transport->socket);
 205                         transport->socket = NULL;
 206                         SIPE_DEBUG_ERROR("socket_connected: failed: %s", error->message);
 207                         transport->error(SIPE_TRANSPORT_CONNECTION, error->message);
 208                         g_error_free(error);
 209                 }
 210         }
 211 }
 212
 213 static gboolean accept_certificate_signal(SIPE_UNUSED_PARAMETER GTlsConnection *tls,
 214                                           GTlsCertificate *peer_cert,
 215                                           SIPE_UNUSED_PARAMETER GTlsCertificateFlags errors,
 216                                           gpointer user_data)
 217 {
 218         struct sipe_transport_telepathy *transport = user_data;
 219
 220         SIPE_DEBUG_INFO("accept_certificate_signal: %p", transport);
 221
 222         /* second connection attempt after feedback from user? */
 223         if (transport->tls_info) {
 224                 /* user accepted certificate */
 225                 sipe_telepathy_tls_info_free(transport->tls_info);
 226                 transport->tls_info = NULL;
 227                 return(TRUE);
 228         } else {
 229                 /* retry after user accepted certificate */
 230                 transport->tls_info = sipe_telepathy_tls_info_new(transport->hostname,
 231                                                                   peer_cert);
 232                 return(FALSE);
 233         }
 234 }
 235
 236 static void tls_handshake_starts(SIPE_UNUSED_PARAMETER GSocketClient *client,
 237                                  GSocketClientEvent event,
 238                                  SIPE_UNUSED_PARAMETER GSocketConnectable *connectable,
 239                                  GIOStream *connection,
 240                                  gpointer user_data)
 241 {
 242         if (event == G_SOCKET_CLIENT_TLS_HANDSHAKING) {
 243                 SIPE_DEBUG_INFO("tls_handshake_starts: %p", connection);
 244                 g_signal_connect(connection, /* is a GTlsConnection */
 245                                  "accept-certificate",
 246                                  G_CALLBACK(accept_certificate_signal),
 247                                  user_data);
 248         }
 249 }
 250
 251 static gboolean internal_connect(gpointer data)
 252 {
 253         struct sipe_transport_telepathy *transport = data;
 254         GSocketClient *client = g_socket_client_new();
 255
 256         SIPE_DEBUG_INFO("internal_connect - hostname: %s port: %d",
 257                         transport->hostname, transport->port);
 258
 259         /* request TLS connection */
 260         if (transport->public.type == SIPE_TRANSPORT_TLS) {
 261                 SIPE_DEBUG_INFO_NOFORMAT("using TLS");
 262                 g_socket_client_set_tls(client, TRUE);
 263                 g_signal_connect(client,
 264                                  "event",
 265                                  G_CALLBACK(tls_handshake_starts),
 266                                  transport);
 267         } else
 268                 SIPE_DEBUG_INFO_NOFORMAT("using TCP");
 269
 270         g_socket_client_connect_async(client,
 271                                       g_network_address_new(transport->hostname,
 272                                                             transport->port),
 273                                       transport->cancel,
 274                                       socket_connected,
 275                                       transport);
 276         g_object_unref(client);
 277
 278         return(FALSE);
 279 }
 280
 281 struct sipe_transport_connection *sipe_backend_transport_connect(struct sipe_core_public *sipe_public,
 282                                                                  const sipe_connect_setup *setup)
 283 {
 284         struct sipe_transport_telepathy *transport = g_new0(struct sipe_transport_telepathy, 1);
 285
 286         transport->public.type      = setup->type;
 287         transport->public.user_data = setup->user_data;
 288         transport->connected        = setup->connected;
 289         transport->input            = setup->input;
 290         transport->error            = setup->error;
 291         transport->hostname         = g_strdup(setup->server_name);
 292         transport->tls_info         = NULL;
 293         transport->private          = sipe_public->backend_private;
 294         transport->cancel           = g_cancellable_new();
 295         transport->buffers          = NULL;
 296         transport->port             = setup->server_port;
 297         transport->is_writing       = FALSE;
 298         transport->do_flush         = FALSE;
 299
 300         if ((setup->type == SIPE_TRANSPORT_TLS) ||
 301             (setup->type == SIPE_TRANSPORT_TCP)) {
 302
 303                 internal_connect(transport);
 304                 return(SIPE_TRANSPORT_CONNECTION);
 305
 306         } else {
 307                 setup->error(SIPE_TRANSPORT_CONNECTION,
 308                              "This should not happen...");
 309                 sipe_backend_transport_disconnect(SIPE_TRANSPORT_CONNECTION);
 310                 return(NULL);
 311         }
 312 }
 313
 314 static gboolean free_transport(gpointer data)
 315 {
 316         struct sipe_transport_telepathy *transport = data;
 317         GSList *entry;
 318
 319         SIPE_DEBUG_INFO("free_transport %p", transport);
 320
 321         if (transport->tls_info)
 322                 sipe_telepathy_tls_info_free(transport->tls_info);
 323         g_free(transport->hostname);
 324
 325         /* free unflushed buffers */
 326         for (entry = transport->buffers; entry; entry = entry->next)
 327                 g_free(entry->data);
 328         g_slist_free(transport->buffers);
 329
 330         if (transport->cancel)
 331                 g_object_unref(transport->cancel);
 332
 333         g_free(transport);
 334
 335         return(FALSE);
 336 }
 337
 338 static void close_completed(GObject *stream,
 339                             GAsyncResult *result,
 340                             gpointer data)
 341 {
 342         struct sipe_transport_telepathy *transport = data;
 343         SIPE_DEBUG_INFO("close_completed: transport %p", data);
 344         g_io_stream_close_finish(G_IO_STREAM(stream), result, NULL);
 345         g_idle_add(free_transport, transport);
 346 }
 347
 348 static void do_close(struct sipe_transport_telepathy *transport)
 349 {
 350         SIPE_DEBUG_INFO("do_close: %p", transport);
 351
 352         /* cancel outstanding asynchronous operations */
 353         transport->do_flush = FALSE;
 354         g_cancellable_cancel(transport->cancel);
 355         g_io_stream_close_async(G_IO_STREAM(transport->socket),
 356                                 G_PRIORITY_DEFAULT,
 357                                 NULL,
 358                                 close_completed,
 359                                 transport);
 360 }
 361
 362 void sipe_backend_transport_disconnect(struct sipe_transport_connection *conn)
 363 {
 364         struct sipe_transport_telepathy *transport = TELEPATHY_TRANSPORT;
 365
 366         if (!transport) return;
 367
 368         SIPE_DEBUG_INFO("sipe_backend_transport_disconnect: %p", transport);
 369
 370         /* error callback is invalid now, do no longer call! */
 371         transport->error = NULL;
 372
 373         /* dropping connection to the server? */
 374         if (transport->private->transport == transport)
 375                 transport->private->transport = NULL;
 376
 377         /* already connected? */
 378         if (transport->socket) {
 379
 380                 /* flush required? */
 381                 if (transport->do_flush && transport->is_writing)
 382                         SIPE_DEBUG_INFO("sipe_backend_transport_disconnect: %p needs flushing",
 383                                         transport);
 384                 else
 385                         do_close(transport);
 386
 387         } else {
 388                 /* cancel outstanding connect operation */
 389                 if (transport->cancel)
 390                         g_cancellable_cancel(transport->cancel);
 391
 392                 /* queue transport to be deleted */
 393                 g_idle_add(free_transport, transport);
 394         }
 395 }
 396
 397 static void do_write(struct sipe_transport_telepathy *transport,
 398                      const gchar *buffer);
 399 static void write_completed(GObject *stream,
 400                             GAsyncResult *result,
 401                             gpointer data)
 402 {
 403         struct sipe_transport_telepathy *transport = data;
 404         GError                          *error     = NULL;
 405         gssize written = g_output_stream_write_finish(G_OUTPUT_STREAM(stream),
 406                                                       result,
 407                                                       &error);
 408
 409         if ((written < 0) || error) {
 410                 const gchar *msg = error ? error->message : "UNKNOWN";
 411                 SIPE_DEBUG_ERROR("write_completed: error: %s", msg);
 412                 if (transport->error)
 413                         transport->error(SIPE_TRANSPORT_CONNECTION, msg);
 414                 g_error_free(error);
 415
 416                 /* error during flush: give up and close transport */
 417                 if (transport->do_flush)
 418                         do_close(transport);
 419
 420         } else if (g_cancellable_is_cancelled(transport->cancel)) {
 421                 /* write completed when transport was disconnected */
 422                 SIPE_DEBUG_INFO_NOFORMAT("write_completed: cancelled");
 423                 transport->is_writing = FALSE;
 424         } else {
 425                 /* more to write? */
 426                 if (transport->buffers) {
 427                         /* yes */
 428                         gchar *buffer = transport->buffers->data;
 429                         transport->buffers = g_slist_remove(transport->buffers,
 430                                                             buffer);
 431                         do_write(transport, buffer);
 432                         g_free(buffer);
 433                 } else {
 434                         /* no, we're done for now... */
 435                         transport->is_writing = FALSE;
 436
 437                         /* flush completed */
 438                         if (transport->do_flush)
 439                                 do_close(transport);
 440                 }
 441         }
 442 }
 443
 444 static void do_write(struct sipe_transport_telepathy *transport,
 445                      const gchar *buffer)
 446 {
 447         transport->is_writing = TRUE;
 448         g_output_stream_write_async(transport->ostream,
 449                                     buffer,
 450                                     strlen(buffer),
 451                                     G_PRIORITY_DEFAULT,
 452                                     transport->cancel,
 453                                     write_completed,
 454                                     transport);
 455 }
 456
 457 void sipe_backend_transport_message(struct sipe_transport_connection *conn,
 458                                     const gchar *buffer)
 459 {
 460         struct sipe_transport_telepathy *transport = TELEPATHY_TRANSPORT;
 461
 462         /* currently writing? */
 463         if (transport->is_writing) {
 464                 /* yes, append copy of buffer to list */
 465                 transport->buffers = g_slist_append(transport->buffers,
 466                                                     g_strdup(buffer));
 467         } else
 468                 /* no, write directly to stream */
 469                 do_write(transport, buffer);
 470 }
 471
 472 void sipe_backend_transport_flush(struct sipe_transport_connection *conn)
 473 {
 474         struct sipe_transport_telepathy *transport = TELEPATHY_TRANSPORT;
 475         transport->do_flush = TRUE;
 476 }
 477
 478 const gchar *sipe_backend_network_ip_address(struct sipe_core_public *sipe_public)
 479 {
 480         struct sipe_backend_private *telepathy_private = sipe_public->backend_private;
 481         const gchar *ipstr = telepathy_private->ipaddress;
 482
 483         /* address cached? */
 484         if (!ipstr) {
 485                 struct sipe_transport_telepathy *transport = telepathy_private->transport;
 486
 487                 /* default if everything should fail */
 488                 ipstr = "127.0.0.1";
 489
 490                 /* connection to server established - get local IP from socket */
 491                 if (transport && transport->socket) {
 492                         GSocketAddress *saddr = g_socket_connection_get_local_address(transport->socket,
 493                                                                                       NULL);
 494
 495                         if (saddr) {
 496                                 GInetAddress *iaddr = g_inet_socket_address_get_address(G_INET_SOCKET_ADDRESS(saddr));
 497
 498                                 if (iaddr) {
 499                                         /* cache address string */
 500                                         ipstr = telepathy_private->ipaddress = g_inet_address_to_string(iaddr);
 501                                         SIPE_DEBUG_INFO("sipe_backend_network_ip_address: %s", ipstr);
 502                                 }
 503                                 g_object_unref(saddr);
 504                         }
 505                 }
 506         }
 507
 508         return(ipstr);
 509 }
 510
 511
 512 /*
 513   Local Variables:
 514   mode: c
 515   c-file-style: "bsd"
 516   indent-tabs-mode: t
 517   tab-width: 8
 518   End:
 519 */