From 4ae760cb62657e68b6b2313e64d2bb59fe264df4 Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Wed, 10 Feb 2010 22:58:39 +0000 Subject: [PATCH] OStatus PuSH fixes: * HMAC now calculated correctly - confirmed interop with Google's public hub * Can optionally use an external PuSH hub, set URL in $config['ostatus']['hub'] (may have issues in replication environment, and will ping the hub for every update rather than just those with subscribers) Internal hub will still function when this is set, but won't be advertised. Warning: setting this, then turning it off later will break subscriptions as that hub will no longer receive pings. --- plugins/OStatus/OStatusPlugin.php | 11 ++- plugins/OStatus/classes/Feedinfo.php | 4 +- plugins/OStatus/classes/HubSub.php | 2 +- .../OStatus/lib/hubdistribqueuehandler.php | 70 +++++++++++++++---- 4 files changed, 67 insertions(+), 20 deletions(-) diff --git a/plugins/OStatus/OStatusPlugin.php b/plugins/OStatus/OStatusPlugin.php index 62ecaf6310..4b9b4d2c32 100644 --- a/plugins/OStatus/OStatusPlugin.php +++ b/plugins/OStatus/OStatusPlugin.php @@ -114,10 +114,15 @@ class OStatusPlugin extends Plugin if ($action instanceof ApiTimelineUserAction || $action instanceof ApiTimelineGroupAction) { $id = $action->arg('id'); if (strval(intval($id)) === strval($id)) { - // Canonical form of id in URL? - // Updates will be handled for our internal PuSH hub. + // Canonical form of id in URL? These are used for OStatus syndication. + + $hub = common_config('ostatus', 'hub'); + if (empty($hub)) { + // Updates will be handled through our internal PuSH hub. + $hub = common_local_url('pushhub'); + } $action->element('link', array('rel' => 'hub', - 'href' => common_local_url('pushhub'))); + 'href' => $hub)); // Also, we'll add in the salmon link $action->element('link', array('rel' => 'salmon', diff --git a/plugins/OStatus/classes/Feedinfo.php b/plugins/OStatus/classes/Feedinfo.php index b4e55c3643..2344a4a0ee 100644 --- a/plugins/OStatus/classes/Feedinfo.php +++ b/plugins/OStatus/classes/Feedinfo.php @@ -160,7 +160,7 @@ class Feedinfo extends Memcached_DataObject function keyTypes() { - return array('id' => 'K'); // @fixme we'll need a profile_id key at least + return array('id' => 'K', 'feeduri' => 'U'); // @fixme we'll need a profile_id key at least } function sequenceKey() @@ -323,7 +323,7 @@ class Feedinfo extends Memcached_DataObject if ($this->secret) { if (preg_match('/^sha1=([0-9a-fA-F]{40})$/', $hmac, $matches)) { $their_hmac = strtolower($matches[1]); - $our_hmac = sha1($xml . $this->secret); + $our_hmac = hash_hmac('sha1', $xml, $this->secret); if ($their_hmac !== $our_hmac) { common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH with bad SHA-1 HMAC: got $their_hmac, expected $our_hmac"); return; diff --git a/plugins/OStatus/classes/HubSub.php b/plugins/OStatus/classes/HubSub.php index 1769f6c941..7071ee5b4f 100644 --- a/plugins/OStatus/classes/HubSub.php +++ b/plugins/OStatus/classes/HubSub.php @@ -242,7 +242,7 @@ class HubSub extends Memcached_DataObject { $headers = array('Content-Type: application/atom+xml'); if ($this->secret) { - $hmac = sha1($atom . $this->secret); + $hmac = hash_hmac('sha1', $atom, $this->secret); $headers[] = "X-Hub-Signature: sha1=$hmac"; } else { $hmac = '(none)'; diff --git a/plugins/OStatus/lib/hubdistribqueuehandler.php b/plugins/OStatus/lib/hubdistribqueuehandler.php index a35b8874c5..245a57f720 100644 --- a/plugins/OStatus/lib/hubdistribqueuehandler.php +++ b/plugins/OStatus/lib/hubdistribqueuehandler.php @@ -49,15 +49,7 @@ class HubDistribQueueHandler extends QueueHandler $feed = common_local_url('ApiTimelineUser', array('id' => $notice->profile_id, 'format' => 'atom')); - $sub = new HubSub(); - $sub->topic = $feed; - if ($sub->find()) { - $atom = $this->userFeedForNotice($notice); - $this->pushFeeds($atom, $sub); - } else { - common_log(LOG_INFO, "No PuSH subscribers for $feed"); - } - return true; + $this->pushFeed($feed, array($this, 'userFeedForNotice'), $notice); } function pushGroup($notice, $group_id) @@ -65,19 +57,69 @@ class HubDistribQueueHandler extends QueueHandler $feed = common_local_url('ApiTimelineGroup', array('id' => $group_id, 'format' => 'atom')); + $this->pushFeed($feed, array($this, 'groupFeedForNotice'), $group_id, $notice); + } + + /** + * @param string $feed URI to the feed + * @param callable $callback function to generate Atom feed update if needed + * any additional params are passed to the callback. + */ + function pushFeed($feed, $callback) + { + $hub = common_config('ostatus', 'hub'); + if ($hub) { + $this->pushFeedExternal($feed, $hub); + } + $sub = new HubSub(); $sub->topic = $feed; if ($sub->find()) { - common_log(LOG_INFO, "Building PuSH feed for $feed"); - $atom = $this->groupFeedForNotice($group_id, $notice); - $this->pushFeeds($atom, $sub); + $args = array_slice(func_get_args(), 2); + $atom = call_user_func_array($callback, $args); + $this->pushFeedInternal($atom, $sub); } else { common_log(LOG_INFO, "No PuSH subscribers for $feed"); } + return true; } - - function pushFeeds($atom, $sub) + /** + * Ping external hub about this update. + * The hub will pull the feed and check for new items later. + * Not guaranteed safe in an environment with database replication. + * + * @param string $feed feed topic URI + * @param string $hub PuSH hub URI + * @fixme can consolidate pings for user & group posts + */ + function pushFeedExternal($feed, $hub) + { + $client = new HTTPClient(); + try { + $data = array('hub.mode' => 'publish', + 'hub.url' => $feed); + $response = $client->post($hub, array(), $data); + if ($response->getStatus() == 204) { + common_log(LOG_INFO, "PuSH ping to hub $hub for $feed ok"); + return true; + } else { + common_log(LOG_ERR, "PuSH ping to hub $hub for $feed failed with HTTP " . + $response->getStatus() . ': ' . + $response->getBody()); + } + } catch (Exception $e) { + common_log(LOG_ERR, "PuSH ping to hub $hub for $feed failed: " . $e->getMessage()); + return false; + } + } + + /** + * Queue up direct feed update pushes to subscribers on our internal hub. + * @param string $atom update feed, containing only new/changed items + * @param HubSub $sub open query of subscribers + */ + function pushFeedInternal($atom, $sub) { common_log(LOG_INFO, "Preparing $sub->N PuSH distribution(s) for $sub->topic"); $qm = QueueManager::get();