Initial implementation of internal PuSH server in Friendica. It has been tested with GNU Social/StatusNet, and subscribe, unsubscribe and pushing new items seem to work.

boot.php

@@ -14,7 +14,7 @@ require_once('include/features.php');
 define ( 'FRIENDICA_PLATFORM',     'Friendica');
 define ( 'FRIENDICA_VERSION',      '3.2.1745' );
 define ( 'DFRN_PROTOCOL_VERSION',  '2.23'    );
-define ( 'DB_UPDATE_VERSION',      1165      );
+define ( 'DB_UPDATE_VERSION',      1166      );
 define ( 'EOL',                    "<br />\r\n"     );
 define ( 'ATOM_TIME',              'Y-m-d\TH:i:s\Z' );
 

include/notifier.php

@@ -966,9 +966,18 @@ function notifier_run(&$argv, &$argc){
 					$h = trim($h);
 					if(! strlen($h))
 						continue;
-					$params = 'hub.mode=publish&hub.url=' . urlencode($a->get_baseurl() . '/dfrn_poll/' . $owner['nickname'] );
+
+					if ($h === '[internal]') {
+						// Set push flag for PuSH subscribers to this topic,
+						// they will be notified in queue.php
+						q("UPDATE `push_subscriber` SET `push` = 1 " . 
+						  "WHERE `nickname` = '%s'", dbesc($owner['nickname']));
+					} else {
+
+						$params = 'hub.mode=publish&hub.url=' . urlencode( $a->get_baseurl() . '/dfrn_poll/' . $owner['nickname'] );
 						post_url($h,$params);
 						logger('pubsub: publish: ' . $h . ' ' . $params . ' returned ' . $a->get_curl_code());
+					}
 					if(count($hubs) > 1)
 						sleep(7);				// try and avoid multiple hubs responding at precisely the same time
 				}

include/queue.php

@@ -2,6 +2,51 @@
 require_once("boot.php");
 require_once('include/queue_fn.php');
 
+function handle_pubsubhubbub() {
+	global $a, $db;
+
+	logger('queue [pubsubhubbub]: start');
+
+	// We'll push to each subscriber that has the push flag set,
+	// i.e. there has been an update (set in notifier.php).
+
+	$r = q("SELECT * FROM `push_subscriber` WHERE `push` = 1");
+
+	foreach($r as $rr) {
+		$params = get_feed_for($a, '', $rr['nickname'], $rr['last_update']);
+		$hmac_sig = hash_hmac("sha1", $params, $rr['secret']);
+
+		$headers = array("Content-type: application/atom+xml",
+						 sprintf("Link: <%s>;rel=hub," .
+								 "<%s>;rel=self",
+								 $a->get_baseurl() . '/pubsubhubbub',
+								 $rr['topic']),
+						 "X-Hub-Signature: sha1=" . $hmac_sig);
+
+		logger('queue [pubsubhubbub]: POST', $headers);
+
+		post_url($rr['callback_url'], $params, $headers);
+		$ret = $a->get_curl_code();
+
+		if ($ret >= 200 && $ret <= 299) {
+			logger('queue [pubsubhubbub]: successfully pushed to ' .
+				   $rr['callback_url']);
+			// here we should set push = 0 and update last_update to 'now'
+			$date_now = datetime_convert('UTC','UTC','now','Y-m-d H:i:s');
+			q("UPDATE `push_subscriber` SET `push` = 0, last_update = '%s' " .
+			  "WHERE id = %d",
+			  dbesc($date_now),
+			  intval($rr['id']));
+		} else {
+			logger('queue [pubsubhubbub]: error when pushing to ' .
+				   $rr['callback_url'] . 'HTTP: ', $ret);
+			// here we should set update some retry counter
+			// or cancel if counter is too high, remove subscription?
+		}
+	}
+}
+
+
 function queue_run(&$argv, &$argc){
 	global $a, $db;
 
@@ -38,6 +83,8 @@ function queue_run(&$argv, &$argc){
 
 	logger('queue: start');
 
+	handle_pubsubhubbub();
+
 	$interval = ((get_config('system','delivery_interval') === false) ? 2 : intval(get_config('system','delivery_interval')));
 
 	$r = q("select * from deliverq where 1");

include/text.php

@@ -1554,7 +1554,7 @@ if(! function_exists('feed_hublinks')) {
  * @return string hub link xml elements
  */
 function feed_hublinks() {
-
+	$a = get_app();
 	$hub = get_config('system','huburl');
 
 	$hubxml = '';
@@ -1565,6 +1565,8 @@ function feed_hublinks() {
 				$h = trim($h);
 				if(! strlen($h))
 					continue;
+				if ($h === '[internal]')
+					$h = $a->get_baseurl() . '/pubsubhubbub';
 				$hubxml .= '<link rel="hub" href="' . xmlify($h) . '" />' . "\n" ;
 			}
 		}

mod/pubsubhubbub.php

@@ -0,0 +1,162 @@
+<?php
+
+function post_var($name) {
+	return (x($_POST, $name)) ? notags(trim($_POST[$name])) : '';
+}
+
+function pubsubhubbub_init(&$a) {
+	// PuSH subscription must be considered "public" so just block it
+	// if public access isn't enabled.
+	if (get_config('system', 'block_public')) {
+		http_status_exit(403);
+	}
+
+	// Subscription request from subscriber
+	// https://pubsubhubbub.googlecode.com/git/pubsubhubbub-core-0.4.html#anchor4
+	// Example from GNU Social:
+    // [hub_mode] => subscribe
+    // [hub_callback] => http://status.local/main/push/callback/1
+    // [hub_verify] => sync
+    // [hub_verify_token] => af11...
+    // [hub_secret] => af11...
+    // [hub_topic] => http://friendica.local/dfrn_poll/sazius
+
+	if($_SERVER['REQUEST_METHOD'] === 'POST') {
+		$hub_mode = post_var('hub_mode');
+		$hub_callback = post_var('hub_callback');
+		$hub_verify = post_var('hub_verify');
+		$hub_verify_token = post_var('hub_verify_token');
+		$hub_secret = post_var('hub_secret');
+		$hub_topic = post_var('hub_topic');
+
+		// check for valid hub_mode
+		if ($hub_mode === 'subscribe') {
+			$subscribe = 1;
+		} else if ($hub_mode === 'unsubscribe') {
+			$subscribe = 0;
+		} else {
+			logger("pubsubhubbub: invalid hub_mode=$hub_mode, ignoring.");
+			http_status_exit(404);
+		}
+
+		logger("pubsubhubbub: $hub_mode request from " . 
+			   $_SERVER['REMOTE_ADDR']);
+
+		// get the nick name from the topic, a bit hacky but needed
+		$nick = substr(strrchr($hub_topic, "/"), 1);
+
+		if (!$nick) {
+			logger('pubsubhubbub: bad hub_topic=$hub_topic, ignoring.');
+			http_status_exit(404);
+		}
+
+		// fetch user from database given the nickname
+		$r = q("SELECT * FROM `user` WHERE `nickname` = '%s'" .
+			   " AND `account_expired` = 0 AND `account_removed` = 0 LIMIT 1", 
+			   dbesc($nick));
+		
+		if(!count($r)) {
+			logger('pubsubhubbub: local account not found: ' . $nick);
+			http_status_exit(404);
+		}
+
+		$owner = $r[0];
+
+		// abort if user's wall is supposed to be private
+		if ($r[0]['hidewall']) {
+			logger('pubsubhubbub: local user ' . $nick .
+				   'has chosen to hide wall, ignoring.');
+			http_status_exit(403);
+		}
+
+		// get corresponding row from contact table
+		$r = q("SELECT * FROM `contact` WHERE `uid` = %d AND `blocked` = 0" .
+			   " AND `pending` = 0 LIMIT 1",
+			   intval($owner['uid']));
+		if(!count($r)) {
+			logger('pubsubhubbub: contact not found.');
+			http_status_exit(404);
+		}
+
+		$contact = $r[0];
+
+		// sanity check that topic URLs are the same
+		if(!link_compare($hub_topic, $contact['poll'])) {
+			logger('pubsubhubbub: hub topic ' . $hub_topic . ' != ' . 
+				   $contact['poll']);
+			http_status_exit(404);
+		}
+
+		// do subscriber verification according to the PuSH protocol
+		$hub_challenge = random_string(40);
+		$params = 'hub.mode=' . 
+			($subscribe == 1 ? 'subscribe' : 'unsubscribe') .
+			'&hub.topic=' . urlencode($hub_topic) .
+			'&hub.challenge=' . $hub_challenge .
+			'&hub.lease_seconds=604800' .
+			'&hub.verify_token=' . $hub_verify_token;
+		
+		// lease time is hard coded to one week (in seconds)
+		// we don't actually enforce the lease time because GNU
+		// Social/StatusNet doesn't honour it (yet)
+
+		$body = fetch_url($hub_callback . "?" . $params);
+		$ret = $a->get_curl_code();
+
+		// give up if the HTTP return code wasn't a success (2xx)
+		if ($ret < 200 || $ret > 299) {
+			logger("pubsubhubbub: subscriber verification at $hub_callback ".
+				   "returned $ret, ignoring.");
+			http_status_exit(404);
+		}
+
+		// check that the correct hub_challenge code was echoed back
+		if (trim($body) !== $hub_challenge) {
+			logger("pubsubhubbub: subscriber did not echo back ".
+				   "hub.challenge, ignoring.");
+			logger("\"$hub_challenge\" != \"".trim($body)."\"");
+			http_status_exit(404);
+		}
+
+		// fetch the old subscription if it exists
+		$r = q("SELECT * FROM `push_subscriber` WHERE `callback_url` = '%s'",
+		  dbesc($hub_callback));
+
+		// delete old subscription if it exists
+		q("DELETE FROM `push_subscriber` WHERE `callback_url` = '%s'",
+		  dbesc($hub_callback));
+
+		if ($subscribe) {
+			$last_update = datetime_convert('UTC','UTC','now','Y-m-d H:i:s');
+			$push_flag = 0;
+
+			// if we are just updating an old subscription, keep the
+			// old values for push and last_update
+			if (count($r)) {
+				$last_update = $r[0]['last_update'];
+				$push_flag = $r[0]['push'];
+			}
+
+			// subscribe means adding the row to the table
+			q("INSERT INTO `push_subscriber` (`uid`, `callback_url`, " .
+			  "`topic`, `nickname`, `push`, `last_update`, `secret`) values " .
+			  "(%d, '%s', '%s', '%s', %d, '%s', '%s')",
+			  intval($owner['uid']),
+			  dbesc($hub_callback),
+			  dbesc($hub_topic),
+			  dbesc($nick),
+			  intval($push_flag),
+			  dbesc($last_update),
+			  dbesc($hub_secret));
+			logger("pubsubhubbub: successfully subscribed [$hub_callback].");
+		} else {
+			logger("pubsubhubbub: successfully unsubscribed [$hub_callback].");
+			// we do nothing here, since the row was already deleted
+		}
+		http_status_exit(202);
+	}
+
+	killme();
+}
+
+?>

update.php

@@ -1,6 +1,6 @@
 <?php
 
-define( 'UPDATE_VERSION' , 1165 );
+define( 'UPDATE_VERSION' , 1166 );
 
 /**
  *
@@ -1509,3 +1509,20 @@ function update_1164() {
 
 	return UPDATE_SUCCESS;
 }
+
+function update_1165() {
+	$r = q("CREATE TABLE IF NOT EXISTS `push_subscriber` (
+			`id` INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+		    `uid` INT NOT NULL,
+	        `callback_url` CHAR( 255 ) NOT NULL,
+            `topic` CHAR( 255 ) NOT NULL,
+            `nickname` CHAR( 255 ) NOT NULL,
+            `push` INT NOT NULL,
+            `last_update` DATETIME NOT NULL,
+            `secret` CHAR( 255 ) NOT NULL
+		  ) ENGINE = MYISAM DEFAULT CHARSET=utf8 ");
+	if (!$r)
+		return UPDATE_FAILED;
+
+	return UPDATE_SUCCESS;
+}