Adding Digital Ocean to Flask-Dance
Posted on Fri 13 November 2020 in conveyor
In the last article we learned how to set up and use a Digital Ocean Oauth app. Now that we understand how that works, let's make all of that hard work more accessible to the rest of the Flask community. Flask-Dance is an extension that is used to complete the "OAuth Dance" with various services. Digital Ocean is not supported, yet. Fortunately, Flask-Dance has great documentation about how a new provider can be added.
Flask-Dance provider requirements
- You must create a Python file in this directory named after the provider: for example, my_provider.py.
- The file must declare a maintainer variable, with the name and email address of the maintainer of this configuration.
- The file must have a factory function that returns an instance of OAuth1ConsumerBlueprint or an instance of OAuth2ConsumerBlueprint. The factory function must be named after the provider: for example, make_my_provider_blueprint().
- The file must expose a variable named after the provider, which is a local proxy to the session attribute on the blueprint returned by the factory function.
- You must create a Python file in the tests/contrib directory named after your provider with a prefix of test_: for example, test_my_provider. This file must contain tests for the file you created in this directory.
- You must add your provider to the docs/providers.rst file, and ensure that your factory function has a RST-formatted docstring that the documentation can pick up.
- All automated tests must pass, and the test coverage must not drop.
- You must update the CHANGELOG.rst file to indicate that this provider was added.
Digital Ocean as a provider
The digitalocean.py file will take care of the first 4 requirements
from __future__ import unicode_literals
from flask_dance.consumer import OAuth2ConsumerBlueprint
from functools import partial
from flask.globals import LocalProxy, _lookup_app_object
try:
from flask import _app_ctx_stack as stack
except ImportError:
from flask import _request_ctx_stack as stack
__maintainer__ = "Michael Abrahamsen [email protected]>"
def make_digitalocean_blueprint(
client_id=None,
client_secret=None,
scope=None,
redirect_url=None,
redirect_to=None,
login_url=None,
authorized_url=None,
session_class=None,
storage=None,
):
"""
Make a blueprint for authenticating with Digital Ocean using OAuth 2.
This requires a client ID and client secret from Digital Ocean.
You should either pass them to this constructor, or make sure that your
Flask application config defines them, using the variables
:envvar:`DIGITALOCEAN_OAUTH_CLIENT_ID` and
:envvar:`DIGITALOCEAN_OAUTH_CLIENT_SECRET`.
Args:
client_id (str): Client ID for your application on Digital Ocean
client_secret (str): Client secret for your Digital Ocean application
scope (str, optional): comma-separated list of scopes for the OAuth
token. Digital Ocean uses space delimitation but comma deliminated
values are used for library consistency and the commas are replaced
with spaces when the blueprint is declared below
redirect_url (str): the URL to redirect to after the authentication
dance is complete
redirect_to (str): if ``redirect_url`` is not defined, the name of the
view to redirect to after the authentication dance is complete.
The actual URL will be determined by :func:`flask.url_for`
login_url (str, optional): the URL path for the ``login`` view.
Defaults to ``/digitalocean``
authorized_url (str, optional): the URL path for the ``authorized`` view.
Defaults to ``/digitalocean/authorized``.
session_class (class, optional): The class to use for creating a
Requests session. Defaults to
:class:`~flask_dance.consumer.requests.OAuth2Session`.
storage: A token storage class, or an instance of a token storage
class, to use for this blueprint. Defaults to
:class:`~flask_dance.consumer.storage.session.SessionStorage`.
:rtype: :class:`~flask_dance.consumer.OAuth2ConsumerBlueprint`
:returns: A :ref:`blueprint <flask:blueprints>` to attach to your Flask app.
"""
digitalocean_bp = OAuth2ConsumerBlueprint(
"digitalocean",
__name__,
client_id=client_id,
client_secret=client_secret,
scope=scope.replace(",", " ") if scope else None,
base_url="https://cloud.digitalocean.com/v1/oauth",
authorization_url="https://cloud.digitalocean.com/v1/oauth/authorize",
token_url="https://cloud.digitalocean.com/v1/oauth/token",
redirect_url=redirect_url,
redirect_to=redirect_to,
login_url=login_url,
authorized_url=authorized_url,
session_class=session_class,
storage=storage,
)
digitalocean_bp.from_config["client_id"] = "DIGITALOCEAN_OAUTH_CLIENT_ID"
digitalocean_bp.from_config["client_secret"] = "DIGITALOCEAN_OAUTH_CLIENT_SECRET"
@digitalocean_bp.before_app_request
def set_applocal_session():
ctx = stack.top
ctx.digitalocean_oauth = digitalocean_bp.session
return digitalocean_bp
digitalocean = LocalProxy(partial(_lookup_app_object, "digitalocean_oauth"))
Creating tests
Most of the tests were copied from another provider. Two additional tests were added that were specific to the digital ocean provider. The scope is passed in as a comma deliminated string to keep with the standards of the Flask-Dance library. Digital Ocean expects the scope to be deliminated by spaces. This is something I plan to ask the project maintainer about. By default the scope is not set but it may make sense to have the default set to "read".
def test_scope_list_is_valid_with_single_scope():
digitalocean_bp = make_digitalocean_blueprint(client_id="foobar",
client_secret="supersecret",
scope="read")
assert digitalocean_bp.session.scope == "read"
def test_scope_list_is_converted_to_space_delimited():
digitalocean_bp = make_digitalocean_blueprint(client_id="foobar",
client_secret="supersecret",
scope="read,write")
assert digitalocean_bp.session.scope == "read write"
Testing the new provider with an example application
An example application has been created and can be found at https://github.com/mikeabrahamsen/flask-dance-digitalocean-oauth-example
import os
import requests
from flask import Flask, request, render_template, redirect, url_for
from flask_dance.contrib.digitalocean import (make_digitalocean_blueprint,
digitalocean)
app = Flask(__name__)
app.secret_key = os.environ.get("SECRET_KEY", "supersecretkey")
app.config["DIGITALOCEAN_OAUTH_CLIENT_ID"] = os.environ.get("DIGITALOCEAN_OAUTH_CLIENT_ID")
app.config["DIGITALOCEAN_OAUTH_CLIENT_SECRET"] = os.environ.get("DIGITALOCEAN_OAUTH_CLIENT_SECRET")
digitalocean_bp = make_digitalocean_blueprint(
scope="read",
redirect_url="http://localhost:5000/digitalocean")
app.register_blueprint(digitalocean_bp, url_prefix="/")
@app.route('/', methods=['GET'])
def index():
oauth_url = None
if not digitalocean.authorized:
oauth_url = url_for("digitalocean.login")
return redirect(oauth_url)
else:
return redirect(url_for("droplets"))
error = request.args.get('error', None)
return render_template(
'index.html',
oauth_url=oauth_url,
error=error
)
@app.route('/droplets', methods=['GET'])
def droplets():
error = None
try:
token = digitalocean.access_token
headers = {"Authorization": f"Bearer {token}"}
api_droplet_list_url = "https://api.digitalocean.com/v2/droplets"
servers = requests.get(api_droplet_list_url, headers=headers,
timeout=3).json()
return render_template(
'server_list.html',
servers=servers.get('droplets', None)
)
except TypeError as e:
error = f'Error: {e}'
print(error)
return redirect(url_for('.index', error=error))