Merge pull request #33 from jaywink/fetch-hcard

Add utilities to fetch Diaspora user discovery documents

.coveragerc

@@ -0,0 +1,5 @@
+[run]
+omit =
+    setup.py
+    federation/__init__.py
+    */tests/*

federation/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.3.2"

federation/tests/utils/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

federation/tests/utils/test_diaspora.py

@@ -0,0 +1,84 @@
+# -*- coding: utf-8 -*-
+from unittest.mock import patch
+from urllib.parse import quote
+
+from federation.hostmeta.generators import DiasporaWebFinger, DiasporaHostMeta
+from federation.utils.diaspora import retrieve_diaspora_hcard, retrieve_diaspora_webfinger, retrieve_diaspora_host_meta
+
+
+class TestRetrieveDiasporaHCard(object):
+    @patch("federation.utils.diaspora.retrieve_diaspora_webfinger", return_value=None)
+    def test_retrieve_webfinger_is_called(self, mock_retrieve):
+        retrieve_diaspora_hcard("bob@localhost")
+        assert mock_retrieve.called_with("bob@localhost")
+
+    @patch("federation.utils.diaspora.fetch_document")
+    @patch("federation.utils.diaspora.retrieve_diaspora_webfinger")
+    def test_fetch_document_is_called(self, mock_retrieve, mock_fetch):
+        mock_retrieve.return_value = DiasporaWebFinger(
+            "bob@localhost", "https://localhost", "123", "456"
+        ).xrd
+        mock_fetch.return_value = "document", None, None
+        document = retrieve_diaspora_hcard("bob@localhost")
+        mock_fetch.assert_called_with("https://localhost/hcard/users/123")
+        assert document == "document"
+
+    @patch("federation.utils.diaspora.fetch_document")
+    @patch("federation.utils.diaspora.retrieve_diaspora_webfinger")
+    def test_returns_none_on_fetch_document_exception(self, mock_retrieve, mock_fetch):
+        mock_retrieve.return_value = DiasporaWebFinger(
+            "bob@localhost", "https://localhost", "123", "456"
+        ).xrd
+        mock_fetch.return_value = None, None, ValueError()
+        document = retrieve_diaspora_hcard("bob@localhost")
+        mock_fetch.assert_called_with("https://localhost/hcard/users/123")
+        assert document == None
+
+
+class TestRetrieveDiasporaWebfinger(object):
+    @patch("federation.utils.diaspora.retrieve_diaspora_host_meta", return_value=None)
+    def test_retrieve_host_meta_is_called(self, mock_retrieve):
+        retrieve_diaspora_webfinger("bob@localhost")
+        mock_retrieve.assert_called_with("localhost")
+
+    @patch("federation.utils.diaspora.XRD.parse_xrd")
+    @patch("federation.utils.diaspora.fetch_document")
+    @patch("federation.utils.diaspora.retrieve_diaspora_host_meta", return_value=None)
+    def test_retrieve_fetch_document_is_called(self, mock_retrieve, mock_fetch, mock_xrd):
+        mock_retrieve.return_value = DiasporaHostMeta(
+            webfinger_host="https://localhost"
+        ).xrd
+        mock_fetch.return_value = "document", None, None
+        mock_xrd.return_value = "document"
+        document = retrieve_diaspora_webfinger("bob@localhost")
+        mock_fetch.assert_called_with("https://localhost/webfinger?q=%s" % quote("bob@localhost"))
+        assert document == "document"
+
+    @patch("federation.utils.diaspora.fetch_document")
+    @patch("federation.utils.diaspora.retrieve_diaspora_host_meta", return_value=None)
+    def test_returns_none_on_fetch_document_exception(self, mock_retrieve, mock_fetch):
+        mock_retrieve.return_value = DiasporaHostMeta(
+            webfinger_host="https://localhost"
+        ).xrd
+        mock_fetch.return_value = None, None, ValueError()
+        document = retrieve_diaspora_webfinger("bob@localhost")
+        mock_fetch.assert_called_with("https://localhost/webfinger?q=%s" % quote("bob@localhost"))
+        assert document == None
+
+
+class TestRetrieveDiasporaHostMeta(object):
+    @patch("federation.utils.diaspora.XRD.parse_xrd")
+    @patch("federation.utils.diaspora.fetch_document")
+    def test_fetch_document_is_called(self, mock_fetch, mock_xrd):
+        mock_fetch.return_value = "document", None, None
+        mock_xrd.return_value = "document"
+        document = retrieve_diaspora_host_meta("localhost")
+        mock_fetch.assert_called_with(host="localhost", path="/.well-known/host-meta")
+        assert document == "document"
+
+    @patch("federation.utils.diaspora.fetch_document")
+    def test_returns_none_on_fetch_document_exception(self, mock_fetch):
+        mock_fetch.return_value = None, None, ValueError()
+        document = retrieve_diaspora_host_meta("localhost")
+        mock_fetch.assert_called_with(host="localhost", path="/.well-known/host-meta")
+        assert document == None

federation/tests/utils/test_network.py

@@ -0,0 +1,88 @@
+# -*- coding: utf-8 -*-
+from unittest.mock import patch, Mock, call
+
+import pytest
+from requests import HTTPError
+from requests.exceptions import SSLError, RequestException
+
+from federation.utils.network import fetch_document, USER_AGENT
+
+
+class TestFetchDocument(object):
+    call_args = {"timeout": 10, "headers": {'user-agent': USER_AGENT}}
+
+    def test_raises_without_url_and_host(self):
+        with pytest.raises(ValueError):
+            fetch_document()
+
+    @patch("federation.utils.network.requests.get")
+    def test_url_is_called(self, mock_get):
+        mock_get.return_value = Mock(status_code=200, text="foo")
+        fetch_document("https://localhost")
+        assert mock_get.called
+
+    @patch("federation.utils.network.requests.get")
+    def test_host_is_called_with_https_first_then_http(self, mock_get):
+        def mock_failing_https_get(url, *args, **kwargs):
+            if url.find("https://") > -1:
+                raise HTTPError()
+            return Mock(status_code=200, text="foo")
+        mock_get.side_effect = mock_failing_https_get
+        fetch_document(host="localhost")
+        assert mock_get.call_count == 2
+        assert mock_get.call_args_list == [
+            call("https://localhost/", **self.call_args),
+            call("http://localhost/", **self.call_args),
+        ]
+
+    @patch("federation.utils.network.requests.get")
+    def test_host_is_sanitized(self, mock_get):
+        mock_get.return_value = Mock(status_code=200, text="foo")
+        fetch_document(host="http://localhost")
+        assert mock_get.call_args_list == [
+            call("https://localhost/", **self.call_args)
+        ]
+
+    @patch("federation.utils.network.requests.get")
+    def test_path_is_sanitized(self, mock_get):
+        mock_get.return_value = Mock(status_code=200, text="foo")
+        fetch_document(host="localhost", path="foobar/bazfoo")
+        assert mock_get.call_args_list == [
+            call("https://localhost/foobar/bazfoo", **self.call_args)
+        ]
+
+    @patch("federation.utils.network.requests.get")
+    def test_exception_is_raised_if_both_protocols_fail(self, mock_get):
+        mock_get.side_effect = HTTPError
+        doc, code, exc = fetch_document(host="localhost")
+        assert mock_get.call_count == 2
+        assert doc == None
+        assert code == None
+        assert exc.__class__ == HTTPError
+
+    @patch("federation.utils.network.requests.get")
+    def test_exception_is_raised_if_url_fails(self, mock_get):
+        mock_get.side_effect = HTTPError
+        doc, code, exc = fetch_document("localhost")
+        assert mock_get.call_count == 1
+        assert doc == None
+        assert code == None
+        assert exc.__class__ == HTTPError
+
+    @patch("federation.utils.network.requests.get")
+    def test_exception_is_raised_if_http_fails_and_raise_ssl_errors_true(self, mock_get):
+        mock_get.side_effect = SSLError
+        doc, code, exc = fetch_document("localhost")
+        assert mock_get.call_count == 1
+        assert doc == None
+        assert code == None
+        assert exc.__class__ == SSLError
+
+    @patch("federation.utils.network.requests.get")
+    def test_exception_is_raised_on_network_error(self, mock_get):
+        mock_get.side_effect = RequestException
+        doc, code, exc = fetch_document(host="localhost")
+        assert mock_get.call_count == 1
+        assert doc == None
+        assert code == None
+        assert exc.__class__ == RequestException

federation/utils/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

federation/utils/diaspora.py

@@ -0,0 +1,61 @@
+# -*- coding: utf-8 -*-
+from urllib.parse import quote
+
+from xrd import XRD
+
+from federation.utils.network import fetch_document
+
+
+def retrieve_diaspora_hcard(handle):
+    """Retrieve a remote Diaspora hCard document.
+
+    Args:
+        handle (str) - Remote handle to retrieve
+
+    Returns:
+        str (HTML document)
+    """
+    webfinger = retrieve_diaspora_webfinger(handle)
+    if not webfinger:
+        return None
+    url = webfinger.find_link(rels="http://microformats.org/profile/hcard").href
+    document, code, exception = fetch_document(url)
+    if exception:
+        return None
+    return document
+
+
+def retrieve_diaspora_webfinger(handle):
+    """Retrieve a remote Diaspora webfinger document.
+
+    Args:
+        handle (str) - Remote handle to retrieve
+
+    Returns:
+        XRD
+    """
+    hostmeta = retrieve_diaspora_host_meta(handle.split("@")[1])
+    if not hostmeta:
+        return None
+    url = hostmeta.find_link(rels="lrdd").template.replace("{uri}", quote(handle))
+    document, code, exception = fetch_document(url)
+    if exception:
+        return None
+    xrd = XRD.parse_xrd(document)
+    return xrd
+
+
+def retrieve_diaspora_host_meta(host):
+    """Retrieve a remote Diaspora host-meta document.
+
+    Args:
+        host (str) - Host to retrieve from
+
+    Returns:
+        XRD
+    """
+    document, code, exception = fetch_document(host=host, path="/.well-known/host-meta")
+    if exception:
+        return None
+    xrd = XRD.parse_xrd(document)
+    return xrd

federation/utils/network.py

@@ -0,0 +1,76 @@
+# -*- coding: utf-8 -*-
+import logging
+
+import requests
+from requests.exceptions import RequestException, HTTPError, SSLError
+
+from federation import __version__
+
+logger = logging.getLogger("social-federation")
+
+USER_AGENT = "python/social-federation/%s" % __version__
+
+
+def fetch_document(url=None, host=None, path="/", timeout=10, raise_ssl_errors=True):
+    """Helper method to fetch remote document.
+
+    Must be given either the `url` or `host`.
+    If `url` is given, only that will be tried without falling back to http from https.
+    If `host` given, `path` will be added to it. Will fall back to http on non-success status code.
+
+    Args:
+        url (str) - Full url to fetch, including protocol
+        host (str) - Domain part only without path or protocol
+        path (str) - Path without domain (defaults to "/")
+        timeout (int) - Seconds to wait for response (defaults to 10)
+        raise_ssl_errors (bool) - Pass False if you want to try HTTP even for sites with SSL errors (default True)
+
+    Returns:
+        document (str) - The full document or None
+        status_code (int) - Status code returned or None
+        error (obj) - Exception raised, if any
+    """
+    if not url and not host:
+        raise ValueError("Need url or host.")
+
+    logger.debug("fetch_document: url=%s, host=%s, path=%s, timeout=%s, raise_ssl_errors=%s",
+                 url, host, path, timeout, raise_ssl_errors)
+    headers = {'user-agent': USER_AGENT}
+    if url:
+        # Use url since it was given
+        logger.debug("fetch_document: trying %s", url)
+        try:
+            response = requests.get(url, timeout=timeout, headers=headers)
+            logger.debug("fetch_document: found document, code %s", response.status_code)
+            return response.text, response.status_code, None
+        except RequestException as ex:
+            logger.debug("fetch_document: exception %s", ex)
+            return None, None, ex
+    # Build url with some little sanitizing
+    host_string = host.replace("http://", "").replace("https://", "").strip("/")
+    path_string = path if path.startswith("/") else "/%s" % path
+    url = "https://%s%s" % (host_string, path_string)
+    logger.debug("fetch_document: trying %s", url)
+    try:
+        response = requests.get(url, timeout=timeout, headers=headers)
+        logger.debug("fetch_document: found document, code %s", response.status_code)
+        response.raise_for_status()
+        return response.text, response.status_code, None
+    except (HTTPError, SSLError) as ex:
+        if isinstance(ex, SSLError) and raise_ssl_errors:
+            logger.debug("fetch_document: exception %s", ex)
+            return None, None, ex
+        # Try http then
+        url = url.replace("https://", "http://")
+        logger.debug("fetch_document: trying %s", url)
+        try:
+            response = requests.get(url, timeout=timeout, headers=headers)
+            logger.debug("fetch_document: found document, code %s", response.status_code)
+            response.raise_for_status()
+            return response.text, response.status_code, None
+        except RequestException as ex:
+            logger.debug("fetch_document: exception %s", ex)
+            return None, None, ex
+    except RequestException as ex:
+        logger.debug("fetch_document: exception %s", ex)
+        return None, None, ex

setup.py

@@ -1,13 +1,15 @@
 #!/usr/bin/env python
 from setuptools import setup, find_packages
 
+from federation import __version__
+
 
 description = 'Python library for abstracting social federation protocols'
 
 
 setup(
     name='Social-Federation',
-    version='0.3.2',
+    version=__version__,
     description=description,
     long_description=description,
     author='Jason Robinson',
@@ -25,6 +27,7 @@ setup(
         "pycrypto>=2.6.0, <3.0.0",
         "python-dateutil>=2.4.0, <3.0.0",
         "python-xrd==0.1",
+        "requests>=2.8.0",
     ],
     include_package_data=True,
     classifiers=[