HubSub.php 9.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274
  1. <?php
  2. /*
  3. * StatusNet - the distributed open-source microblogging tool
  4. * Copyright (C) 2010, 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. if (!defined('STATUSNET')) {
  20. exit(1);
  21. }
  22. /**
  23. * PuSH feed subscription record
  24. * @package Hub
  25. * @author Brion Vibber <brion@status.net>
  26. */
  27. class HubSub extends Managed_DataObject
  28. {
  29. public $__table = 'hubsub';
  30. public $hashkey; // sha1(topic . '|' . $callback); (topic, callback) key is too long for myisam in utf8
  31. public $topic;
  32. public $callback;
  33. public $secret;
  34. public $lease;
  35. public $sub_start;
  36. public $sub_end;
  37. public $created;
  38. public $modified;
  39. protected static function hashkey($topic, $callback)
  40. {
  41. return sha1($topic . '|' . $callback);
  42. }
  43. public static function getByHashkey($topic, $callback)
  44. {
  45. return self::getKV('hashkey', self::hashkey($topic, $callback));
  46. }
  47. public static function schemaDef()
  48. {
  49. return array(
  50. 'fields' => array(
  51. 'hashkey' => array('type' => 'char', 'not null' => true, 'length' => 40, 'description' => 'HubSub hashkey'),
  52. 'topic' => array('type' => 'varchar', 'not null' => true, 'length' => 255, 'description' => 'HubSub topic'),
  53. 'callback' => array('type' => 'varchar', 'not null' => true, 'length' => 255, 'description' => 'HubSub callback'),
  54. 'secret' => array('type' => 'text', 'description' => 'HubSub stored secret'),
  55. 'lease' => array('type' => 'int', 'description' => 'HubSub leasetime'),
  56. 'sub_start' => array('type' => 'datetime', 'description' => 'subscription start'),
  57. 'sub_end' => array('type' => 'datetime', 'description' => 'subscription end'),
  58. 'created' => array('type' => 'datetime', 'not null' => true, 'description' => 'date this record was created'),
  59. 'modified' => array('type' => 'timestamp', 'not null' => true, 'description' => 'date this record was modified'),
  60. ),
  61. 'primary key' => array('hashkey'),
  62. 'indexes' => array(
  63. 'hubsub_topic_idx' => array('topic'),
  64. ),
  65. );
  66. }
  67. /**
  68. * Validates a requested lease length, sets length plus
  69. * subscription start & end dates.
  70. *
  71. * Does not save to database -- use before insert() or update().
  72. *
  73. * @param int $length in seconds
  74. */
  75. function setLease($length)
  76. {
  77. assert(is_int($length));
  78. $min = 86400;
  79. $max = 86400 * 30;
  80. if ($length == 0) {
  81. // We want to garbage collect dead subscriptions!
  82. $length = $max;
  83. } elseif( $length < $min) {
  84. $length = $min;
  85. } else if ($length > $max) {
  86. $length = $max;
  87. }
  88. $this->lease = $length;
  89. $this->start_sub = common_sql_now();
  90. $this->end_sub = common_sql_date(time() + $length);
  91. }
  92. /**
  93. * Schedule a future verification ping to the subscriber.
  94. * If queues are disabled, will be immediate.
  95. *
  96. * @param string $mode 'subscribe' or 'unsubscribe'
  97. * @param string $token hub.verify_token value, if provided by client
  98. */
  99. function scheduleVerify($mode, $token=null, $retries=null)
  100. {
  101. if ($retries === null) {
  102. $retries = intval(common_config('ostatus', 'hub_retries'));
  103. }
  104. $data = array('sub' => clone($this),
  105. 'mode' => $mode,
  106. 'token' => $token, // let's put it in there if remote uses PuSH <0.4
  107. 'retries' => $retries);
  108. $qm = QueueManager::get();
  109. $qm->enqueue($data, 'hubconf');
  110. }
  111. /**
  112. * Send a verification ping to subscriber, and if confirmed apply the changes.
  113. * This may create, update, or delete the database record.
  114. *
  115. * @param string $mode 'subscribe' or 'unsubscribe'
  116. * @param string $token hub.verify_token value, if provided by client
  117. * @throws ClientException on failure
  118. */
  119. function verify($mode, $token=null)
  120. {
  121. assert($mode == 'subscribe' || $mode == 'unsubscribe');
  122. $challenge = common_random_hexstr(32);
  123. $params = array('hub.mode' => $mode,
  124. 'hub.topic' => $this->topic,
  125. 'hub.challenge' => $challenge);
  126. if ($mode == 'subscribe') {
  127. $params['hub.lease_seconds'] = $this->lease;
  128. }
  129. if ($token !== null) { // TODO: deprecated in PuSH 0.4
  130. $params['hub.verify_token'] = $token; // let's put it in there if remote uses PuSH <0.4
  131. }
  132. // Any existing query string parameters must be preserved
  133. $url = $this->callback;
  134. if (strpos($url, '?') !== false) {
  135. $url .= '&';
  136. } else {
  137. $url .= '?';
  138. }
  139. $url .= http_build_query($params, '', '&');
  140. $request = new HTTPClient();
  141. $response = $request->get($url);
  142. $status = $response->getStatus();
  143. if ($status >= 200 && $status < 300) {
  144. common_log(LOG_INFO, "Verified {$mode} of {$this->callback}:{$this->topic}");
  145. } else {
  146. // TRANS: Client exception. %s is a HTTP status code.
  147. throw new ClientException(sprintf(_m('Hub subscriber verification returned HTTP %s.'),$status));
  148. }
  149. $old = HubSub::getByHashkey($this->topic, $this->callback);
  150. if ($mode == 'subscribe') {
  151. if ($old instanceof HubSub) {
  152. $this->update($old);
  153. } else {
  154. $ok = $this->insert();
  155. }
  156. } else if ($mode == 'unsubscribe') {
  157. if ($old instanceof HubSub) {
  158. $old->delete();
  159. } else {
  160. // That's ok, we're already unsubscribed.
  161. }
  162. }
  163. }
  164. /**
  165. * Insert wrapper; transparently set the hash key from topic and callback columns.
  166. * @return mixed success
  167. */
  168. function insert()
  169. {
  170. $this->hashkey = self::hashkey($this->topic, $this->callback);
  171. $this->created = common_sql_now();
  172. $this->modified = common_sql_now();
  173. return parent::insert();
  174. }
  175. /**
  176. * Schedule delivery of a 'fat ping' to the subscriber's callback
  177. * endpoint. If queues are disabled, this will run immediately.
  178. *
  179. * @param string $atom well-formed Atom feed
  180. * @param int $retries optional count of retries if POST fails; defaults to hub_retries from config or 0 if unset
  181. */
  182. function distribute($atom, $retries=null)
  183. {
  184. if ($retries === null) {
  185. $retries = intval(common_config('ostatus', 'hub_retries'));
  186. }
  187. // We dare not clone() as when the clone is discarded it'll
  188. // destroy the result data for the parent query.
  189. // @fixme use clone() again when it's safe to copy an
  190. // individual item from a multi-item query again.
  191. $sub = HubSub::getByHashkey($this->topic, $this->callback);
  192. $data = array('sub' => $sub,
  193. 'atom' => $atom,
  194. 'retries' => $retries);
  195. common_log(LOG_INFO, "Queuing PuSH: $this->topic to $this->callback");
  196. $qm = QueueManager::get();
  197. $qm->enqueue($data, 'hubout');
  198. }
  199. /**
  200. * Queue up a large batch of pushes to multiple subscribers
  201. * for this same topic update.
  202. *
  203. * If queues are disabled, this will run immediately.
  204. *
  205. * @param string $atom well-formed Atom feed
  206. * @param array $pushCallbacks list of callback URLs
  207. */
  208. function bulkDistribute($atom, $pushCallbacks)
  209. {
  210. $data = array('atom' => $atom,
  211. 'topic' => $this->topic,
  212. 'pushCallbacks' => $pushCallbacks);
  213. common_log(LOG_INFO, "Queuing PuSH batch: $this->topic to " .
  214. count($pushCallbacks) . " sites");
  215. $qm = QueueManager::get();
  216. $qm->enqueue($data, 'hubprep');
  217. }
  218. /**
  219. * Send a 'fat ping' to the subscriber's callback endpoint
  220. * containing the given Atom feed chunk.
  221. *
  222. * Determination of which items to send should be done at
  223. * a higher level; don't just shove in a complete feed!
  224. *
  225. * @param string $atom well-formed Atom feed
  226. * @throws Exception (HTTP or general)
  227. */
  228. function push($atom)
  229. {
  230. $headers = array('Content-Type: application/atom+xml');
  231. if ($this->secret) {
  232. $hmac = hash_hmac('sha1', $atom, $this->secret);
  233. $headers[] = "X-Hub-Signature: sha1=$hmac";
  234. } else {
  235. $hmac = '(none)';
  236. }
  237. common_log(LOG_INFO, "About to push feed to $this->callback for $this->topic, HMAC $hmac");
  238. $request = new HTTPClient();
  239. $request->setBody($atom);
  240. $response = $request->post($this->callback, $headers);
  241. if ($response->isOk()) {
  242. return true;
  243. } else {
  244. // TRANS: Exception. %1$s is a response status code, %2$s is the body of the response.
  245. throw new Exception(sprintf(_m('Callback returned status: %1$s. Body: %2$s'),
  246. $response->getStatus(),trim($response->getBody())));
  247. }
  248. }
  249. }