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 class XmppManager extends IoManager
  34 {
  35     protected $site = null;
  36     protected $pingid = 0;
  37     protected $lastping = null;
  38     protected $conn = null;
  39
  40     static protected $singletons = array();
  41
  42     const PING_INTERVAL = 120;
  43
  44     /**
  45      * Fetch the singleton XmppManager for the current site.
  46      * @return mixed XmppManager, or false if unneeded
  47      */
  48     public static function get()
  49     {
  50         if (common_config('xmpp', 'enabled')) {
  51             $site = StatusNet::currentSite();
  52             if (empty(self::$singletons[$site])) {
  53                 self::$singletons[$site] = new XmppManager();
  54             }
  55             return self::$singletons[$site];
  56         } else {
  57             return false;
  58         }
  59     }
  60
  61     /**
  62      * Tell the i/o master we need one instance for each supporting site
  63      * being handled in this process.
  64      */
  65     public static function multiSite()
  66     {
  67         return IoManager::INSTANCE_PER_SITE;
  68     }
  69
  70     function __construct()
  71     {
  72         $this->site = StatusNet::currentSite();
  73         $this->resource = common_config('xmpp', 'resource') . 'daemon';
  74     }
  75
  76     /**
  77      * Initialize connection to server.
  78      * @return boolean true on success
  79      */
  80     public function start($master)
  81     {
  82         parent::start($master);
  83         $this->switchSite();
  84
  85         require_once INSTALLDIR . "/lib/jabber.php";
  86
  87         # Low priority; we don't want to receive messages
  88
  89         common_log(LOG_INFO, "INITIALIZE");
  90         $this->conn = jabber_connect($this->resource);
  91
  92         if (empty($this->conn)) {
  93             common_log(LOG_ERR, "Couldn't connect to server.");
  94             return false;
  95         }
  96
  97         $this->log(LOG_DEBUG, "Initializing stanza handlers.");
  98
  99         $this->conn->addEventHandler('message', 'handle_message', $this);
 100         $this->conn->addEventHandler('presence', 'handle_presence', $this);
 101         $this->conn->addEventHandler('reconnect', 'handle_reconnect', $this);
 102
 103         $this->conn->setReconnectTimeout(600);
 104         // @todo Needs i18n?
 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                 common_log(LOG_ERR, __METHOD__ . ' failed: 0 bytes sent');
 202                 return false;
 203             }
 204         } else {
 205             // Can't send right now...
 206             common_log(LOG_ERR, __METHOD__ . ' failed: XMPP server connection currently down');
 207             return false;
 208         }
 209     }
 210
 211     /**
 212      * Send a keepalive ping to the XMPP server.
 213      */
 214     protected function sendPing()
 215     {
 216         $jid = jabber_daemon_address().'/'.$this->resource;
 217         $server = common_config('xmpp', 'server');
 218
 219         if (!isset($this->pingid)) {
 220             $this->pingid = 0;
 221         } else {
 222             $this->pingid++;
 223         }
 224
 225         common_log(LOG_DEBUG, "Sending ping #{$this->pingid}");
 226
 227         $this->conn->send("<iq from='{$jid}' to='{$server}' id='ping_{$this->pingid}' type='get'><ping xmlns='urn:xmpp:ping'/></iq>");
 228     }
 229
 230     /**
 231      * Callback for Jabber reconnect event
 232      * @param $pl
 233      */
 234     function handle_reconnect(&$pl)
 235     {
 236         common_log(LOG_NOTICE, 'XMPP reconnected');
 237
 238         $this->conn->processUntil('session_start');
 239         $this->conn->presence(null, 'available', null, 'available', 100);
 240     }
 241
 242
 243     function get_user($from)
 244     {
 245         $user = User::staticGet('jabber', jabber_normalize_jid($from));
 246         return $user;
 247     }
 248
 249     /**
 250      * XMPP callback for handling message input...
 251      * @param array $pl XMPP payload
 252      */
 253     function handle_message(&$pl)
 254     {
 255         $from = jabber_normalize_jid($pl['from']);
 256
 257         if ($pl['type'] != 'chat') {
 258             $this->log(LOG_WARNING, "Ignoring message of type ".$pl['type']." from $from: " . $pl['xml']->toString());
 259             return;
 260         }
 261
 262         if (mb_strlen($pl['body']) == 0) {
 263             $this->log(LOG_WARNING, "Ignoring message with empty body from $from: "  . $pl['xml']->toString());
 264             return;
 265         }
 266
 267         // Forwarded from another daemon for us to handle; this shouldn't
 268         // happen any more but we might get some legacy items.
 269         if ($this->is_self($from)) {
 270             $this->log(LOG_INFO, "Got forwarded notice from self ($from).");
 271             $from = $this->get_ofrom($pl);
 272             $this->log(LOG_INFO, "Originally sent by $from.");
 273             if (is_null($from) || $this->is_self($from)) {
 274                 $this->log(LOG_INFO, "Ignoring notice originally sent by $from.");
 275                 return;
 276             }
 277         }
 278
 279         $user = $this->get_user($from);
 280
 281         // For common_current_user to work
 282         global $_cur;
 283         $_cur = $user;
 284
 285         if (!$user) {
 286             // TRANS: %s is the URL to the StatusNet site's Instant Messaging settings.
 287             $this->from_site($from, sprintf(_('Unknown user. Go to %s ' .
 288                              'to add your address to your account'),common_local_url('imsettings')));
 289             $this->log(LOG_WARNING, 'Message from unknown user ' . $from);
 290             return;
 291         }
 292         if ($this->handle_command($user, $pl['body'])) {
 293             $this->log(LOG_INFO, "Command message by $from handled.");
 294             return;
 295         } else if ($this->is_autoreply($pl['body'])) {
 296             $this->log(LOG_INFO, 'Ignoring auto reply from ' . $from);
 297             return;
 298         } else if ($this->is_otr($pl['body'])) {
 299             $this->log(LOG_INFO, 'Ignoring OTR from ' . $from);
 300             return;
 301         } else {
 302
 303             $this->log(LOG_INFO, 'Posting a notice from ' . $user->nickname);
 304
 305             $this->add_notice($user, $pl);
 306         }
 307
 308         $user->free();
 309         unset($user);
 310         unset($_cur);
 311
 312         unset($pl['xml']);
 313         $pl['xml'] = null;
 314
 315         $pl = null;
 316         unset($pl);
 317     }
 318
 319     function is_self($from)
 320     {
 321         return preg_match('/^'.strtolower(jabber_daemon_address()).'/', strtolower($from));
 322     }
 323
 324     function get_ofrom($pl)
 325     {
 326         $xml = $pl['xml'];
 327         $addresses = $xml->sub('addresses');
 328         if (!$addresses) {
 329             $this->log(LOG_WARNING, 'Forwarded message without addresses');
 330             return null;
 331         }
 332         $address = $addresses->sub('address');
 333         if (!$address) {
 334             $this->log(LOG_WARNING, 'Forwarded message without address');
 335             return null;
 336         }
 337         if (!array_key_exists('type', $address->attrs)) {
 338             $this->log(LOG_WARNING, 'No type for forwarded message');
 339             return null;
 340         }
 341         $type = $address->attrs['type'];
 342         if ($type != 'ofrom') {
 343             $this->log(LOG_WARNING, 'Type of forwarded message is not ofrom');
 344             return null;
 345         }
 346         if (!array_key_exists('jid', $address->attrs)) {
 347             $this->log(LOG_WARNING, 'No jid for forwarded message');
 348             return null;
 349         }
 350         $jid = $address->attrs['jid'];
 351         if (!$jid) {
 352             $this->log(LOG_WARNING, 'Could not get jid from address');
 353             return null;
 354         }
 355         $this->log(LOG_DEBUG, 'Got message forwarded from jid ' . $jid);
 356         return $jid;
 357     }
 358
 359     function is_autoreply($txt)
 360     {
 361         if (preg_match('/[\[\(]?[Aa]uto[-\s]?[Rr]e(ply|sponse)[\]\)]/', $txt)) {
 362             return true;
 363         } else if (preg_match('/^System: Message wasn\'t delivered. Offline storage size was exceeded.$/', $txt)) {
 364             return true;
 365         } else {
 366             return false;
 367         }
 368     }
 369
 370     function is_otr($txt)
 371     {
 372         if (preg_match('/^\?OTR/', $txt)) {
 373             return true;
 374         } else {
 375             return false;
 376         }
 377     }
 378
 379     function from_site($address, $msg)
 380     {
 381         $text = '['.common_config('site', 'name') . '] ' . $msg;
 382         jabber_send_message($address, $text);
 383     }
 384
 385     function handle_command($user, $body)
 386     {
 387         $inter = new CommandInterpreter();
 388         $cmd = $inter->handle_command($user, $body);
 389         if ($cmd) {
 390             $chan = new XMPPChannel($this->conn);
 391             $cmd->execute($chan);
 392             return true;
 393         } else {
 394             return false;
 395         }
 396     }
 397
 398     function add_notice(&$user, &$pl)
 399     {
 400         $body = trim($pl['body']);
 401         $content_shortened = $user->shortenLinks($body);
 402         if (Notice::contentTooLong($content_shortened)) {
 403           $from = jabber_normalize_jid($pl['from']);
 404           // TRANS: Response to XMPP source when it sent too long a message.
 405           // TRANS: %1$d the maximum number of allowed characters (used for plural), %2$d is the sent number.
 406           $this->from_site($from, sprintf(_m('Message too long. Maximum is %1$d character, you sent %2$d.',
 407                                              'Message too long. Maximum is %1$d characters, you sent %2$d.',
 408                                              Notice::maxContent()),
 409                                           Notice::maxContent(),
 410                                           mb_strlen($content_shortened)));
 411           return;
 412         }
 413
 414         try {
 415             $notice = Notice::saveNew($user->id, $content_shortened, 'xmpp');
 416         } catch (Exception $e) {
 417             $this->log(LOG_ERR, $e->getMessage());
 418             $this->from_site($user->jabber, $e->getMessage());
 419             return;
 420         }
 421
 422         common_broadcast_notice($notice);
 423         $this->log(LOG_INFO,
 424                    'Added notice ' . $notice->id . ' from user ' . $user->nickname);
 425         $notice->free();
 426         unset($notice);
 427     }
 428
 429     function handle_presence(&$pl)
 430     {
 431         $from = jabber_normalize_jid($pl['from']);
 432         switch ($pl['type']) {
 433          case 'subscribe':
 434             # We let anyone subscribe
 435             $this->subscribed($from);
 436             $this->log(LOG_INFO,
 437                        'Accepted subscription from ' . $from);
 438             break;
 439          case 'subscribed':
 440          case 'unsubscribed':
 441          case 'unsubscribe':
 442             $this->log(LOG_INFO,
 443                        'Ignoring  "' . $pl['type'] . '" from ' . $from);
 444             break;
 445          default:
 446             if (!$pl['type']) {
 447                 $user = User::staticGet('jabber', $from);
 448                 if (!$user) {
 449                     $this->log(LOG_WARNING, 'Presence from unknown user ' . $from);
 450                     return;
 451                 }
 452                 if ($user->updatefrompresence) {
 453                     $this->log(LOG_INFO, 'Updating ' . $user->nickname .
 454                                ' status from presence.');
 455                     $this->add_notice($user, $pl);
 456                 }
 457                 $user->free();
 458                 unset($user);
 459             }
 460             break;
 461         }
 462         unset($pl['xml']);
 463         $pl['xml'] = null;
 464
 465         $pl = null;
 466         unset($pl);
 467     }
 468
 469     function log($level, $msg)
 470     {
 471         $text = 'XMPPDaemon('.$this->resource.'): '.$msg;
 472         common_log($level, $text);
 473     }
 474
 475     function subscribed($to)
 476     {
 477         jabber_special_presence('subscribed', $to);
 478     }
 479
 480     /**
 481      * Make sure we're on the right site configuration
 482      */
 483     protected function switchSite()
 484     {
 485         if ($this->site != StatusNet::currentSite()) {
 486             common_log(LOG_DEBUG, __METHOD__ . ": switching to site $this->site");
 487             $this->stats('switch');
 488             StatusNet::switchSite($this->site);
 489         }
 490     }
 491 }