lib/xmppmanager.php

   1 <?php
   2 /*
   3  * StatusNet - the distributed open-source microblogging tool
   4  * Copyright (C) 2008, 2009, StatusNet, Inc.
   5  *
   6  * This program is free software: you can redistribute it and/or modify
   7  * it under the terms of the GNU Affero General Public License as published by
   8  * the Free Software Foundation, either version 3 of the License, or
   9  * (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14  * GNU Affero General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Affero General Public License
  17  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18  */
  19
  20 if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
  21
  22 /**
  23  * XMPP background connection manager for XMPP-using queue handlers,
  24  * allowing them to send outgoing messages on the right connection.
  25  *
  26  * Input is handled during socket select loop, keepalive pings during idle.
  27  * Any incoming messages will be forwarded to the main XmppDaemon process,
  28  * which handles direct user interaction.
  29  *
  30  * In a multi-site queuedaemon.php run, one connection will be instantiated
  31  * for each site being handled by the current process that has XMPP enabled.
  32  */
  33
  34 class XmppManager extends IoManager
  35 {
  36     protected $site = null;
  37     protected $pingid = 0;
  38     protected $lastping = null;
  39     protected $conn = null;
  40
  41     static protected $singletons = array();
  42
  43     const PING_INTERVAL = 120;
  44
  45     /**
  46      * Fetch the singleton XmppManager for the current site.
  47      * @return mixed XmppManager, or false if unneeded
  48      */
  49     public static function get()
  50     {
  51         if (common_config('xmpp', 'enabled')) {
  52             $site = StatusNet::currentSite();
  53             if (empty(self::$singletons[$site])) {
  54                 self::$singletons[$site] = new XmppManager();
  55             }
  56             return self::$singletons[$site];
  57         } else {
  58             return false;
  59         }
  60     }
  61
  62     /**
  63      * Tell the i/o master we need one instance for each supporting site
  64      * being handled in this process.
  65      */
  66     public static function multiSite()
  67     {
  68         return IoManager::INSTANCE_PER_SITE;
  69     }
  70
  71     function __construct()
  72     {
  73         $this->site = StatusNet::currentSite();
  74         $this->resource = common_config('xmpp', 'resource') . 'daemon';
  75     }
  76
  77     /**
  78      * Initialize connection to server.
  79      * @return boolean true on success
  80      */
  81     public function start($master)
  82     {
  83         parent::start($master);
  84         $this->switchSite();
  85
  86         require_once INSTALLDIR . "/lib/jabber.php";
  87
  88         # Low priority; we don't want to receive messages
  89
  90         common_log(LOG_INFO, "INITIALIZE");
  91         $this->conn = jabber_connect($this->resource);
  92
  93         if (empty($this->conn)) {
  94             common_log(LOG_ERR, "Couldn't connect to server.");
  95             return false;
  96         }
  97
  98         $this->log(LOG_DEBUG, "Initializing stanza handlers.");
  99
 100         $this->conn->addEventHandler('message', 'handle_message', $this);
 101         $this->conn->addEventHandler('presence', 'handle_presence', $this);
 102         $this->conn->addEventHandler('reconnect', 'handle_reconnect', $this);
 103
 104         $this->conn->setReconnectTimeout(600);
 105         jabber_send_presence("Send me a message to post a notice", 'available', null, 'available', 100);
 106
 107         return !is_null($this->conn);
 108     }
 109
 110     /**
 111      * Message pump is triggered on socket input, so we only need an idle()
 112      * call often enough to trigger our outgoing pings.
 113      */
 114     function timeout()
 115     {
 116         return self::PING_INTERVAL;
 117     }
 118
 119     /**
 120      * Lists the XMPP connection socket to allow i/o master to wake
 121      * when input comes in here as well as from the queue source.
 122      *
 123      * @return array of resources
 124      */
 125     public function getSockets()
 126     {
 127         if ($this->conn) {
 128             return array($this->conn->getSocket());
 129         } else {
 130             return array();
 131         }
 132     }
 133
 134     /**
 135      * Process XMPP events that have come in over the wire.
 136      * Side effects: may switch site configuration
 137      * @fixme may kill process on XMPP error
 138      * @param resource $socket
 139      */
 140     public function handleInput($socket)
 141     {
 142         $this->switchSite();
 143
 144         # Process the queue for as long as needed
 145         try {
 146             if ($this->conn) {
 147                 assert($socket === $this->conn->getSocket());
 148
 149                 common_log(LOG_DEBUG, "Servicing the XMPP queue.");
 150                 $this->stats('xmpp_process');
 151                 $this->conn->processTime(0);
 152             }
 153         } catch (XMPPHP_Exception $e) {
 154             common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
 155             die($e->getMessage());
 156         }
 157     }
 158
 159     /**
 160      * Idle processing for io manager's execution loop.
 161      * Send keepalive pings to server.
 162      *
 163      * Side effect: kills process on exception from XMPP library.
 164      *
 165      * @fixme non-dying error handling
 166      */
 167     public function idle($timeout=0)
 168     {
 169         if ($this->conn) {
 170             $now = time();
 171             if (empty($this->lastping) || $now - $this->lastping > self::PING_INTERVAL) {
 172                 $this->switchSite();
 173                 try {
 174                     $this->sendPing();
 175                     $this->lastping = $now;
 176                 } catch (XMPPHP_Exception $e) {
 177                     common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
 178                     die($e->getMessage());
 179                 }
 180             }
 181         }
 182     }
 183
 184     /**
 185      * For queue handlers to pass us a message to push out,
 186      * if we're active.
 187      *
 188      * @fixme should this be blocking etc?
 189      *
 190      * @param string $msg XML stanza to send
 191      * @return boolean success
 192      */
 193     public function send($msg)
 194     {
 195         if ($this->conn && !$this->conn->isDisconnected()) {
 196             $bytes = $this->conn->send($msg);
 197             if ($bytes > 0) {
 198                 $this->conn->processTime(0);
 199                 return true;
 200             } else {
 201                 return false;
 202             }
 203         } else {
 204             // Can't send right now...
 205             return false;
 206         }
 207     }
 208
 209     /**
 210      * Send a keepalive ping to the XMPP server.
 211      */
 212     protected function sendPing()
 213     {
 214         $jid = jabber_daemon_address().'/'.$this->resource;
 215         $server = common_config('xmpp', 'server');
 216
 217         if (!isset($this->pingid)) {
 218             $this->pingid = 0;
 219         } else {
 220             $this->pingid++;
 221         }
 222
 223         common_log(LOG_DEBUG, "Sending ping #{$this->pingid}");
 224
 225         $this->conn->send("<iq from='{$jid}' to='{$server}' id='ping_{$this->pingid}' type='get'><ping xmlns='urn:xmpp:ping'/></iq>");
 226     }
 227
 228     /**
 229      * Callback for Jabber reconnect event
 230      * @param $pl
 231      */
 232     function handle_reconnect(&$pl)
 233     {
 234         common_log(LOG_NOTICE, 'XMPP reconnected');
 235
 236         $this->conn->processUntil('session_start');
 237         $this->conn->presence(null, 'available', null, 'available', 100);
 238     }
 239
 240
 241     function get_user($from)
 242     {
 243         $user = User::staticGet('jabber', jabber_normalize_jid($from));
 244         return $user;
 245     }
 246
 247     /**
 248      * XMPP callback for handling message input...
 249      * @param array $pl XMPP payload
 250      */
 251     function handle_message(&$pl)
 252     {
 253         $from = jabber_normalize_jid($pl['from']);
 254
 255         if ($pl['type'] != 'chat') {
 256             $this->log(LOG_WARNING, "Ignoring message of type ".$pl['type']." from $from: " . $pl['xml']->toString());
 257             return;
 258         }
 259
 260         if (mb_strlen($pl['body']) == 0) {
 261             $this->log(LOG_WARNING, "Ignoring message with empty body from $from: "  . $pl['xml']->toString());
 262             return;
 263         }
 264
 265         // Forwarded from another daemon for us to handle; this shouldn't
 266         // happen any more but we might get some legacy items.
 267         if ($this->is_self($from)) {
 268             $this->log(LOG_INFO, "Got forwarded notice from self ($from).");
 269             $from = $this->get_ofrom($pl);
 270             $this->log(LOG_INFO, "Originally sent by $from.");
 271             if (is_null($from) || $this->is_self($from)) {
 272                 $this->log(LOG_INFO, "Ignoring notice originally sent by $from.");
 273                 return;
 274             }
 275         }
 276
 277         $user = $this->get_user($from);
 278
 279         // For common_current_user to work
 280         global $_cur;
 281         $_cur = $user;
 282
 283         if (!$user) {
 284             $this->from_site($from, 'Unknown user; go to ' .
 285                              common_local_url('imsettings') .
 286                              ' to add your address to your account');
 287             $this->log(LOG_WARNING, 'Message from unknown user ' . $from);
 288             return;
 289         }
 290         if ($this->handle_command($user, $pl['body'])) {
 291             $this->log(LOG_INFO, "Command message by $from handled.");
 292             return;
 293         } else if ($this->is_autoreply($pl['body'])) {
 294             $this->log(LOG_INFO, 'Ignoring auto reply from ' . $from);
 295             return;
 296         } else if ($this->is_otr($pl['body'])) {
 297             $this->log(LOG_INFO, 'Ignoring OTR from ' . $from);
 298             return;
 299         } else {
 300
 301             $this->log(LOG_INFO, 'Posting a notice from ' . $user->nickname);
 302
 303             $this->add_notice($user, $pl);
 304         }
 305
 306         $user->free();
 307         unset($user);
 308         unset($_cur);
 309
 310         unset($pl['xml']);
 311         $pl['xml'] = null;
 312
 313         $pl = null;
 314         unset($pl);
 315     }
 316
 317
 318     function is_self($from)
 319     {
 320         return preg_match('/^'.strtolower(jabber_daemon_address()).'/', strtolower($from));
 321     }
 322
 323     function get_ofrom($pl)
 324     {
 325         $xml = $pl['xml'];
 326         $addresses = $xml->sub('addresses');
 327         if (!$addresses) {
 328             $this->log(LOG_WARNING, 'Forwarded message without addresses');
 329             return null;
 330         }
 331         $address = $addresses->sub('address');
 332         if (!$address) {
 333             $this->log(LOG_WARNING, 'Forwarded message without address');
 334             return null;
 335         }
 336         if (!array_key_exists('type', $address->attrs)) {
 337             $this->log(LOG_WARNING, 'No type for forwarded message');
 338             return null;
 339         }
 340         $type = $address->attrs['type'];
 341         if ($type != 'ofrom') {
 342             $this->log(LOG_WARNING, 'Type of forwarded message is not ofrom');
 343             return null;
 344         }
 345         if (!array_key_exists('jid', $address->attrs)) {
 346             $this->log(LOG_WARNING, 'No jid for forwarded message');
 347             return null;
 348         }
 349         $jid = $address->attrs['jid'];
 350         if (!$jid) {
 351             $this->log(LOG_WARNING, 'Could not get jid from address');
 352             return null;
 353         }
 354         $this->log(LOG_DEBUG, 'Got message forwarded from jid ' . $jid);
 355         return $jid;
 356     }
 357
 358     function is_autoreply($txt)
 359     {
 360         if (preg_match('/[\[\(]?[Aa]uto[-\s]?[Rr]e(ply|sponse)[\]\)]/', $txt)) {
 361             return true;
 362         } else if (preg_match('/^System: Message wasn\'t delivered. Offline storage size was exceeded.$/', $txt)) {
 363             return true;
 364         } else {
 365             return false;
 366         }
 367     }
 368
 369     function is_otr($txt)
 370     {
 371         if (preg_match('/^\?OTR/', $txt)) {
 372             return true;
 373         } else {
 374             return false;
 375         }
 376     }
 377
 378     function from_site($address, $msg)
 379     {
 380         $text = '['.common_config('site', 'name') . '] ' . $msg;
 381         jabber_send_message($address, $text);
 382     }
 383
 384     function handle_command($user, $body)
 385     {
 386         $inter = new CommandInterpreter();
 387         $cmd = $inter->handle_command($user, $body);
 388         if ($cmd) {
 389             $chan = new XMPPChannel($this->conn);
 390             $cmd->execute($chan);
 391             return true;
 392         } else {
 393             return false;
 394         }
 395     }
 396
 397     function add_notice(&$user, &$pl)
 398     {
 399         $body = trim($pl['body']);
 400         $content_shortened = common_shorten_links($body);
 401         if (Notice::contentTooLong($content_shortened)) {
 402           $from = jabber_normalize_jid($pl['from']);
 403           $this->from_site($from, sprintf(_('Message too long - maximum is %1$d characters, you sent %2$d.'),
 404                                           Notice::maxContent(),
 405                                           mb_strlen($content_shortened)));
 406           return;
 407         }
 408
 409         try {
 410             $notice = Notice::saveNew($user->id, $content_shortened, 'xmpp');
 411         } catch (Exception $e) {
 412             $this->log(LOG_ERR, $e->getMessage());
 413             $this->from_site($user->jabber, $e->getMessage());
 414             return;
 415         }
 416
 417         common_broadcast_notice($notice);
 418         $this->log(LOG_INFO,
 419                    'Added notice ' . $notice->id . ' from user ' . $user->nickname);
 420         $notice->free();
 421         unset($notice);
 422     }
 423
 424     function handle_presence(&$pl)
 425     {
 426         $from = jabber_normalize_jid($pl['from']);
 427         switch ($pl['type']) {
 428          case 'subscribe':
 429             # We let anyone subscribe
 430             $this->subscribed($from);
 431             $this->log(LOG_INFO,
 432                        'Accepted subscription from ' . $from);
 433             break;
 434          case 'subscribed':
 435          case 'unsubscribed':
 436          case 'unsubscribe':
 437             $this->log(LOG_INFO,
 438                        'Ignoring  "' . $pl['type'] . '" from ' . $from);
 439             break;
 440          default:
 441             if (!$pl['type']) {
 442                 $user = User::staticGet('jabber', $from);
 443                 if (!$user) {
 444                     $this->log(LOG_WARNING, 'Presence from unknown user ' . $from);
 445                     return;
 446                 }
 447                 if ($user->updatefrompresence) {
 448                     $this->log(LOG_INFO, 'Updating ' . $user->nickname .
 449                                ' status from presence.');
 450                     $this->add_notice($user, $pl);
 451                 }
 452                 $user->free();
 453                 unset($user);
 454             }
 455             break;
 456         }
 457         unset($pl['xml']);
 458         $pl['xml'] = null;
 459
 460         $pl = null;
 461         unset($pl);
 462     }
 463
 464     function log($level, $msg)
 465     {
 466         $text = 'XMPPDaemon('.$this->resource.'): '.$msg;
 467         common_log($level, $text);
 468     }
 469
 470     function subscribed($to)
 471     {
 472         jabber_special_presence('subscribed', $to);
 473     }
 474
 475     /**
 476      * Make sure we're on the right site configuration
 477      */
 478     protected function switchSite()
 479     {
 480         if ($this->site != StatusNet::currentSite()) {
 481             common_log(LOG_DEBUG, __METHOD__ . ": switching to site $this->site");
 482             $this->stats('switch');
 483             StatusNet::switchSite($this->site);
 484         }
 485     }
 486 }