Add globus auth code

changelog/46.feature.rst

@@ -0,0 +1 @@
+Add utilities for doing OAuth with Globus.

dkist/utils/globus/__init__.py

@@ -0,0 +1,4 @@
+"""
+Utilities and Helpers for dealing with Globus.
+"""
+from .auth import get_refresh_token_authorizer

dkist/utils/globus/auth.py

@@ -0,0 +1,191 @@
+"""
+Globus Auth Helpers.
+"""
+# A lot of this code is copied from the Globus Auth Example repo:
+# https://github.com/globus/native-app-examples
+
+import json
+import stat
+import queue
+import threading
+import webbrowser
+from pathlib import Path
+from http.server import HTTPServer, BaseHTTPRequestHandler
+from urllib.parse import parse_qs, urlparse
+
+import appdirs
+import globus_sdk
+import globus_sdk.auth.token_response
+
+CLIENT_ID = 'dd2d62af-0b44-4e2e-9454-1092c94b46b3'
+SCOPES = ('urn:globus:auth:scope:transfer.api.globus.org:all',)
+
+
+__all__ = ['get_refresh_token_authorizer']
+
+
+class AuthenticationError(Exception):
+    """
+    An error to be raised if authentication fails.
+    """
+
+
+class RedirectHTTPServer(HTTPServer):
+    def __init__(self, listen, handler_class):
+        super().__init__(listen, handler_class)
+
+        self._auth_code_queue = queue.Queue()
+
+    def return_code(self, code):
+        self._auth_code_queue.put_nowait(code)
+
+    def wait_for_code(self):
+        return self._auth_code_queue.get(block=True)
+
+
+class RedirectHandler(BaseHTTPRequestHandler):
+    def do_GET(self):
+        self.send_response(200)
+        self.send_header('Content-type', 'text/html')
+        self.end_headers()
+        self.wfile.write(b'You\'re all set, you can close this window!')
+
+        code = parse_qs(urlparse(self.path).query).get('code', [''])[0]
+        self.server.return_code(code)
+
+    def log_message(self, format, *args):
+        return
+
+
+def start_local_server(listen=('localhost', 0)):
+    """
+    Start a server which will listen for the OAuth2 callback.
+
+    Parameters
+    ----------
+    listen: `tuple`, optional
+        ``(address, port)`` tuple, defaults to localhost and port 0, which
+        leads to the system choosing a free port.
+    """
+    server = RedirectHTTPServer(listen, RedirectHandler)
+    thread = threading.Thread(target=server.serve_forever)
+    thread.daemon = True
+    thread.start()
+
+    return server
+
+
+def get_cache_file_path():
+    """
+    Use appdirs to get the cache path for the user and add the filename.
+    """
+    cache_dir = Path(appdirs.user_cache_dir("dkist"))
+    return cache_dir / "globus_auth_cache.json"
+
+
+def get_cache_contents():
+    """
+    Read the cache file, return an empty dict if not found or invalid.
+    """
+    cache_file = get_cache_file_path()
+    if not cache_file.exists():
+        return {}
+    else:
+        try:
+            with open(cache_file) as fd:
+                return json.load(fd)
+        except (IOError, json.JSONDecodeError):
+            return {}
+
+
+def save_auth_cache(auth_cache):
+    """
+    Write the auth cache to the cache file.
+
+    Parameters
+    ----------
+    auth_cache: `dict` or `~globus_sdk.auth.token_response.OAuthTokenResponse`
+        The auth cache to save.
+
+    """
+    if isinstance(auth_cache, globus_sdk.auth.token_response.OAuthTokenResponse):
+        auth_cache = auth_cache.by_resource_server
+
+    cache_file = get_cache_file_path()
+
+    # Ensure the cache dir exists.
+    cache_dir = cache_file.parent
+    if not cache_dir.exists():
+        cache_dir.mkdir()
+
+    # Write the token to the cache file.
+    with open(cache_file, "w") as fd:
+        json.dump(auth_cache, fd)
+
+    # Ensure the token file has minimal permissions.
+    cache_file.chmod(stat.S_IRUSR | stat.S_IWUSR)
+
+
+def do_native_app_authentication(client_id, requested_scopes=None):  # pragma: no cover
+    """
+    Does a Native App authentication flow and returns a
+    dict of tokens keyed by service name.
+    """
+    server = start_local_server()
+    redirect_uri = "http://{a[0]}:{a[1]}".format(a=server.server_address)
+
+    client = globus_sdk.NativeAppAuthClient(client_id=client_id)
+    client.oauth2_start_flow(requested_scopes=SCOPES,
+                             redirect_uri=redirect_uri,
+                             refresh_tokens=True)
+    url = client.oauth2_get_authorize_url()
+
+    webbrowser.open(url, new=1)
+
+    print("Waiting for completion of Globus Authentication in your webbrowser...")
+    try:
+        auth_code = server.wait_for_code()
+    except KeyboardInterrupt:
+        raise AuthenticationError("Failed to authenticate with Globus.")
+    finally:
+        server.shutdown()
+
+    token_response = client.oauth2_exchange_code_for_tokens(auth_code)
+
+    # return a set of tokens, organized by resource server name
+    return token_response.by_resource_server
+
+
+def get_refresh_token_authorizer(force_reauth=False):
+    """
+    Perform OAuth2 Authentication to Globus.
+
+    Parameters
+    ----------
+    force_reauth: `bool`, optional
+        If `True` ignore any cached credentials and reauth with Globus. This is
+        useful if the cache is corrupted or the refresh token has expired.
+
+    Returns
+    -------
+    `globus_sdk.RefreshTokenAuthorizer`
+
+    """
+    tokens = None
+    if not force_reauth:
+        tokens = get_cache_contents()
+    if not tokens:
+        tokens = do_native_app_authentication(CLIENT_ID, SCOPES)
+        save_auth_cache(tokens)
+
+    transfer_tokens = tokens['transfer.api.globus.org']
+
+    auth_client = globus_sdk.NativeAppAuthClient(client_id=CLIENT_ID)
+    authorizer = globus_sdk.RefreshTokenAuthorizer(
+        transfer_tokens['refresh_token'],
+        auth_client,
+        access_token=transfer_tokens['access_token'],
+        expires_at=transfer_tokens['expires_at_seconds'],
+        on_refresh=save_auth_cache)
+
+    return authorizer

dkist/utils/globus/tests/test_auth.py

@@ -0,0 +1,101 @@
+import json
+import stat
+import pathlib
+from unittest import mock
+
+import requests
+import globus_sdk
+
+from dkist.utils.globus.auth import (get_cache_contents, get_cache_file_path, save_auth_cache,
+                                     start_local_server, get_refresh_token_authorizer)
+
+
+def test_http_server():
+    server = start_local_server()
+    redirect_uri = "http://{a[0]}:{a[1]}".format(a=server.server_address)
+    inp_code = "wibble"
+
+    requests.get(redirect_uri + f"?code={inp_code}")
+
+    code = server.wait_for_code()
+
+    assert code == inp_code
+
+
+@mock.patch("appdirs.user_cache_dir", return_value="/tmp/test/")
+def test_get_cache_file_path(mock_appdirs):
+    path = get_cache_file_path()
+    assert isinstance(path, pathlib.Path)
+
+    assert path.parent == pathlib.Path("/tmp/test")
+    assert path.name == "globus_auth_cache.json"
+
+
+def test_get_no_cache(tmpdir):
+    with mock.patch("appdirs.user_cache_dir", return_value=str(tmpdir)):
+        # Test file not exists
+        cache = get_cache_contents()
+        assert isinstance(cache, dict)
+        assert not cache
+
+
+def test_get_cache(tmpdir):
+    with mock.patch("appdirs.user_cache_dir", return_value=str(tmpdir)):
+        with open(tmpdir / "globus_auth_cache.json", "w") as fd:
+            json.dump({"hello": "world"}, fd)
+
+        cache = get_cache_contents()
+        assert isinstance(cache, dict)
+        assert len(cache) == 1
+        assert cache == {"hello": "world"}
+
+
+def test_get_cache_not_json(tmpdir):
+    with mock.patch("appdirs.user_cache_dir", return_value=str(tmpdir)):
+        with open(tmpdir / "globus_auth_cache.json", "w") as fd:
+            fd.write("aslkjdasdjjdlsajdjklasjdj, akldjaskldjasd, lkjasdkljasldkjas")
+
+        cache = get_cache_contents()
+        assert isinstance(cache, dict)
+        assert not cache
+
+
+def test_save_auth_cache(tmpdir):
+    filename = tmpdir / "globus_auth_cache.json"
+    assert not filename.exists()  # Sanity check
+    with mock.patch("appdirs.user_cache_dir", return_value=str(tmpdir)):
+        save_auth_cache({"hello": "world"})
+
+    assert filename.exists()
+    statinfo = filename.stat()
+
+    # Test that the user can read and write
+    assert bool(statinfo.mode & stat.S_IRUSR)
+    assert bool(statinfo.mode & stat.S_IWUSR)
+    # Test that neither "Group" or "Other" have read permissions
+    assert not bool(statinfo.mode & stat.S_IRGRP)
+    assert not bool(statinfo.mode & stat.S_IROTH)
+
+
+def test_get_refresh_token_authorizer():
+    # An example cache without real tokens
+    cache = {
+        "transfer.api.globus.org": {
+            "scope": "urn:globus:auth:scope:transfer.api.globus.org:all",
+            "access_token": "buscVeATmhfB0v1tzu8VmTfFRB1nwlF8bn1R9rQTI3Q",
+            "refresh_token": "YSbLZowAHfmhxehUqeOF3lFvoC0FlTT11QGupfWAOX4",
+            "token_type": "Bearer",
+            "expires_at_seconds": 1553362861,
+            "resource_server": "transfer.api.globus.org"
+        }
+    }
+
+    with mock.patch("dkist.utils.globus.auth.get_cache_contents", return_value=cache):
+        auth = get_refresh_token_authorizer()
+        assert isinstance(auth, globus_sdk.RefreshTokenAuthorizer)
+        assert auth.access_token == cache["transfer.api.globus.org"]["access_token"]
+
+    with mock.patch("dkist.utils.globus.auth.do_native_app_authentication", return_value=cache):
+        auth = get_refresh_token_authorizer(force_reauth=True)
+        assert isinstance(auth, globus_sdk.RefreshTokenAuthorizer)
+        assert auth.access_token == cache["transfer.api.globus.org"]["access_token"]

setup.cfg

@@ -24,6 +24,8 @@ install_requires =
   numpy
   scipy
   matplotlib
+  globus-sdk
+  appdirs
   # Git deps
   astropy
   asdf

tox.ini

@@ -6,15 +6,10 @@ setenv =
     MPLBACKEND = agg
 passenv = CC
 deps = 
-    numpy
-    scipy
-    matplotlib
-    dask[array]
     coverage
     pytest-astropy
     pytest-cov
     pytest-mpl
-    importlib_resources;python_version<"3.7"
     git+https://github.com/astropy/astropy
     git+https://github.com/spacetelescope/asdf
     git+https://github.com/Cadair/gwcs@dkist