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
  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 = common_config('site', 'server');
  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 = common_config('site', 'server');
  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         jabber_send_presence("Send me a message to post a notice", 'available', null, 'available', 100);
 105
 106         return !is_null($this->conn);
 107     }
 108
 109     /**
 110      * Message pump is triggered on socket input, so we only need an idle()
 111      * call often enough to trigger our outgoing pings.
 112      */
 113     function timeout()
 114     {
 115         return self::PING_INTERVAL;
 116     }
 117
 118     /**
 119      * Lists the XMPP connection socket to allow i/o master to wake
 120      * when input comes in here as well as from the queue source.
 121      *
 122      * @return array of resources
 123      */
 124     public function getSockets()
 125     {
 126         if ($this->conn) {
 127             return array($this->conn->getSocket());
 128         } else {
 129             return array();
 130         }
 131     }
 132
 133     /**
 134      * Process XMPP events that have come in over the wire.
 135      * Side effects: may switch site configuration
 136      * @fixme may kill process on XMPP error
 137      * @param resource $socket
 138      */
 139     public function handleInput($socket)
 140     {
 141         $this->switchSite();
 142
 143         # Process the queue for as long as needed
 144         try {
 145             if ($this->conn) {
 146                 assert($socket === $this->conn->getSocket());
 147
 148                 common_log(LOG_DEBUG, "Servicing the XMPP queue.");
 149                 $this->stats('xmpp_process');
 150                 $this->conn->processTime(0);
 151             }
 152         } catch (XMPPHP_Exception $e) {
 153             common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
 154             die($e->getMessage());
 155         }
 156     }
 157
 158     /**
 159      * Idle processing for io manager's execution loop.
 160      * Send keepalive pings to server.
 161      *
 162      * Side effect: kills process on exception from XMPP library.
 163      *
 164      * @fixme non-dying error handling
 165      */
 166     public function idle($timeout=0)
 167     {
 168         if ($this->conn) {
 169             $now = time();
 170             if (empty($this->lastping) || $now - $this->lastping > self::PING_INTERVAL) {
 171                 $this->switchSite();
 172                 try {
 173                     $this->sendPing();
 174                     $this->lastping = $now;
 175                 } catch (XMPPHP_Exception $e) {
 176                     common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
 177                     die($e->getMessage());
 178                 }
 179             }
 180         }
 181     }
 182
 183     /**
 184      * For queue handlers to pass us a message to push out,
 185      * if we're active.
 186      *
 187      * @fixme should this be blocking etc?
 188      *
 189      * @param string $msg XML stanza to send
 190      * @return boolean success
 191      */
 192     public function send($msg)
 193     {
 194         if ($this->conn && !$this->conn->isDisconnected()) {
 195             $bytes = $this->conn->send($msg);
 196             if ($bytes > 0) {
 197                 $this->conn->processTime(0);
 198                 return true;
 199             } else {
 200                 return false;
 201             }
 202         } else {
 203             // Can't send right now...
 204             return false;
 205         }
 206     }
 207
 208     /**
 209      * Send a keepalive ping to the XMPP server.
 210      */
 211     protected function sendPing()
 212     {
 213         $jid = jabber_daemon_address().'/'.$this->resource;
 214         $server = common_config('xmpp', 'server');
 215
 216         if (!isset($this->pingid)) {
 217             $this->pingid = 0;
 218         } else {
 219             $this->pingid++;
 220         }
 221
 222         common_log(LOG_DEBUG, "Sending ping #{$this->pingid}");
 223
 224         $this->conn->send("<iq from='{$jid}' to='{$server}' id='ping_{$this->pingid}' type='get'><ping xmlns='urn:xmpp:ping'/></iq>");
 225     }
 226
 227     /**
 228      * Callback for Jabber reconnect event
 229      * @param $pl
 230      */
 231     function handle_reconnect(&$pl)
 232     {
 233         common_log(LOG_NOTICE, 'XMPP reconnected');
 234
 235         $this->conn->processUntil('session_start');
 236         $this->conn->presence(null, 'available', null, 'available', 100);
 237     }
 238
 239
 240     function get_user($from)
 241     {
 242         $user = User::staticGet('jabber', jabber_normalize_jid($from));
 243         return $user;
 244     }
 245
 246     /**
 247      * XMPP callback for handling message input...
 248      * @param array $pl XMPP payload
 249      */
 250     function handle_message(&$pl)
 251     {
 252         $from = jabber_normalize_jid($pl['from']);
 253
 254         if ($pl['type'] != 'chat') {
 255             $this->log(LOG_WARNING, "Ignoring message of type ".$pl['type']." from $from.");
 256             return;
 257         }
 258
 259         if (mb_strlen($pl['body']) == 0) {
 260             $this->log(LOG_WARNING, "Ignoring message with empty body from $from.");
 261             return;
 262         }
 263
 264         // Forwarded from another daemon for us to handle; this shouldn't
 265         // happen any more but we might get some legacy items.
 266         if ($this->is_self($from)) {
 267             $this->log(LOG_INFO, "Got forwarded notice from self ($from).");
 268             $from = $this->get_ofrom($pl);
 269             $this->log(LOG_INFO, "Originally sent by $from.");
 270             if (is_null($from) || $this->is_self($from)) {
 271                 $this->log(LOG_INFO, "Ignoring notice originally sent by $from.");
 272                 return;
 273             }
 274         }
 275
 276         $user = $this->get_user($from);
 277
 278         // For common_current_user to work
 279         global $_cur;
 280         $_cur = $user;
 281
 282         if (!$user) {
 283             $this->from_site($from, 'Unknown user; go to ' .
 284                              common_local_url('imsettings') .
 285                              ' to add your address to your account');
 286             $this->log(LOG_WARNING, 'Message from unknown user ' . $from);
 287             return;
 288         }
 289         if ($this->handle_command($user, $pl['body'])) {
 290             $this->log(LOG_INFO, "Command message by $from handled.");
 291             return;
 292         } else if ($this->is_autoreply($pl['body'])) {
 293             $this->log(LOG_INFO, 'Ignoring auto reply from ' . $from);
 294             return;
 295         } else if ($this->is_otr($pl['body'])) {
 296             $this->log(LOG_INFO, 'Ignoring OTR from ' . $from);
 297             return;
 298         } else {
 299
 300             $this->log(LOG_INFO, 'Posting a notice from ' . $user->nickname);
 301
 302             $this->add_notice($user, $pl);
 303         }
 304
 305         $user->free();
 306         unset($user);
 307         unset($_cur);
 308
 309         unset($pl['xml']);
 310         $pl['xml'] = null;
 311
 312         $pl = null;
 313         unset($pl);
 314     }
 315
 316
 317     function is_self($from)
 318     {
 319         return preg_match('/^'.strtolower(jabber_daemon_address()).'/', strtolower($from));
 320     }
 321
 322     function get_ofrom($pl)
 323     {
 324         $xml = $pl['xml'];
 325         $addresses = $xml->sub('addresses');
 326         if (!$addresses) {
 327             $this->log(LOG_WARNING, 'Forwarded message without addresses');
 328             return null;
 329         }
 330         $address = $addresses->sub('address');
 331         if (!$address) {
 332             $this->log(LOG_WARNING, 'Forwarded message without address');
 333             return null;
 334         }
 335         if (!array_key_exists('type', $address->attrs)) {
 336             $this->log(LOG_WARNING, 'No type for forwarded message');
 337             return null;
 338         }
 339         $type = $address->attrs['type'];
 340         if ($type != 'ofrom') {
 341             $this->log(LOG_WARNING, 'Type of forwarded message is not ofrom');
 342             return null;
 343         }
 344         if (!array_key_exists('jid', $address->attrs)) {
 345             $this->log(LOG_WARNING, 'No jid for forwarded message');
 346             return null;
 347         }
 348         $jid = $address->attrs['jid'];
 349         if (!$jid) {
 350             $this->log(LOG_WARNING, 'Could not get jid from address');
 351             return null;
 352         }
 353         $this->log(LOG_DEBUG, 'Got message forwarded from jid ' . $jid);
 354         return $jid;
 355     }
 356
 357     function is_autoreply($txt)
 358     {
 359         if (preg_match('/[\[\(]?[Aa]uto[-\s]?[Rr]e(ply|sponse)[\]\)]/', $txt)) {
 360             return true;
 361         } else if (preg_match('/^System: Message wasn\'t delivered. Offline storage size was exceeded.$/', $txt)) {
 362             return true;
 363         } else {
 364             return false;
 365         }
 366     }
 367
 368     function is_otr($txt)
 369     {
 370         if (preg_match('/^\?OTR/', $txt)) {
 371             return true;
 372         } else {
 373             return false;
 374         }
 375     }
 376
 377     function from_site($address, $msg)
 378     {
 379         $text = '['.common_config('site', 'name') . '] ' . $msg;
 380         jabber_send_message($address, $text);
 381     }
 382
 383     function handle_command($user, $body)
 384     {
 385         $inter = new CommandInterpreter();
 386         $cmd = $inter->handle_command($user, $body);
 387         if ($cmd) {
 388             $chan = new XMPPChannel($this->conn);
 389             $cmd->execute($chan);
 390             return true;
 391         } else {
 392             return false;
 393         }
 394     }
 395
 396     function add_notice(&$user, &$pl)
 397     {
 398         $body = trim($pl['body']);
 399         $content_shortened = common_shorten_links($body);
 400         if (Notice::contentTooLong($content_shortened)) {
 401           $from = jabber_normalize_jid($pl['from']);
 402           $this->from_site($from, sprintf(_('Message too long - maximum is %1$d characters, you sent %2$d.'),
 403                                           Notice::maxContent(),
 404                                           mb_strlen($content_shortened)));
 405           return;
 406         }
 407
 408         try {
 409             $notice = Notice::saveNew($user->id, $content_shortened, 'xmpp');
 410         } catch (Exception $e) {
 411             $this->log(LOG_ERR, $e->getMessage());
 412             $this->from_site($user->jabber, $e->getMessage());
 413             return;
 414         }
 415
 416         common_broadcast_notice($notice);
 417         $this->log(LOG_INFO,
 418                    'Added notice ' . $notice->id . ' from user ' . $user->nickname);
 419         $notice->free();
 420         unset($notice);
 421     }
 422
 423     function handle_presence(&$pl)
 424     {
 425         $from = jabber_normalize_jid($pl['from']);
 426         switch ($pl['type']) {
 427          case 'subscribe':
 428             # We let anyone subscribe
 429             $this->subscribed($from);
 430             $this->log(LOG_INFO,
 431                        'Accepted subscription from ' . $from);
 432             break;
 433          case 'subscribed':
 434          case 'unsubscribed':
 435          case 'unsubscribe':
 436             $this->log(LOG_INFO,
 437                        'Ignoring  "' . $pl['type'] . '" from ' . $from);
 438             break;
 439          default:
 440             if (!$pl['type']) {
 441                 $user = User::staticGet('jabber', $from);
 442                 if (!$user) {
 443                     $this->log(LOG_WARNING, 'Presence from unknown user ' . $from);
 444                     return;
 445                 }
 446                 if ($user->updatefrompresence) {
 447                     $this->log(LOG_INFO, 'Updating ' . $user->nickname .
 448                                ' status from presence.');
 449                     $this->add_notice($user, $pl);
 450                 }
 451                 $user->free();
 452                 unset($user);
 453             }
 454             break;
 455         }
 456         unset($pl['xml']);
 457         $pl['xml'] = null;
 458
 459         $pl = null;
 460         unset($pl);
 461     }
 462
 463     function log($level, $msg)
 464     {
 465         $text = 'XMPPDaemon('.$this->resource.'): '.$msg;
 466         common_log($level, $text);
 467     }
 468
 469     function subscribed($to)
 470     {
 471         jabber_special_presence('subscribed', $to);
 472     }
 473
 474     /**
 475      * Make sure we're on the right site configuration
 476      */
 477     protected function switchSite()
 478     {
 479         if ($this->site != common_config('site', 'server')) {
 480             common_log(LOG_DEBUG, __METHOD__ . ": switching to site $this->site");
 481             $this->stats('switch');
 482             StatusNet::init($this->site);
 483         }
 484     }
 485 }