OAuth2 Client creation management command

Allows programmatic creation of OAuth2 Clients. In particular, allows overriding Client ID and secret, and designating Clients as trusted. Outputs a JSON-encoded representation of newly-created Clients to stdout. XCOM-146.

oauth2_provider/management/commands/create_oauth2_client.py

@@ -0,0 +1,152 @@
+import json
+from optparse import make_option
+
+from django.contrib.auth import get_user_model
+from django.core.exceptions import ValidationError
+from django.core.management.base import BaseCommand, CommandError
+from django.core.validators import URLValidator
+from provider.constants import CONFIDENTIAL, PUBLIC
+from provider.oauth2.models import Client
+
+from oauth2_provider.models import TrustedClient
+
+
+ARG_STRING = '<url> <redirect_uri> <client_type: "confidential" | "public">'
+
+User = get_user_model()
+
+
+class Command(BaseCommand):
+    help = 'Create a new OAuth2 Client. Outputs a serialized representation of the newly-created Client.'
+    args = ARG_STRING
+    fields = None
+
+    option_list = BaseCommand.option_list + (
+        make_option(
+            '-u',
+            '--username',
+            action='store',
+            type='string',
+            dest='username',
+            help="Username of a user to associate with the Client."
+        ),
+        make_option(
+            '-n',
+            '--client_name',
+            action='store',
+            type='string',
+            dest='client_name',
+            help="String to assign as the Client name."
+        ),
+        make_option(
+            '-i',
+            '--client_id',
+            action='store',
+            type='string',
+            dest='client_id',
+            help="String to assign as the Client ID."
+        ),
+        make_option(
+            '-s',
+            '--client_secret',
+            action='store',
+            type='string',
+            dest='client_secret',
+            help="String to assign as the Client secret. Should not be shared."
+        ),
+        make_option(
+            '-t',
+            '--trusted',
+            action='store_true',
+            dest='trusted',
+            default=False,
+            help="Designate the Client as trusted. Trusted Clients bypass the user consent "
+                 "form typically displayed after validating the user's credentials."
+        ),
+    )
+
+    def handle(self, *args, **options):
+        self._clean_args(args)
+        self._parse_options(options)
+
+        trusted = self.fields.pop('trusted')
+        client = Client.objects.create(**self.fields)
+
+        if trusted:
+            TrustedClient.objects.create(client=client)
+
+        serialized = json.dumps(client.serialize(), indent=4)
+        self.stdout.write(serialized)
+
+    def _clean_args(self, args):
+        """Validate and clean the command's arguments.
+
+        These arguments must include the Client application's URL, the Client application's
+        OAuth2 callback URL, and the Client's type, indicating whether the Client application
+        is capable of maintaining the confidentiality of its credentials (e.g., running on a
+        secure server) or is incapable of doing so (e.g., running in a browser).
+
+        Arguments:
+            args (tuple): Arguments with which the command was called.
+
+        Raises:
+            CommandError, if the number of arguments provided is invalid, if the URLs provided
+                are invalid, or if the client type provided is invalid.
+        """
+        if len(args) != 3:
+            raise CommandError(
+                "Number of arguments provided is invalid. "
+                "This command requires the following arguments: {}.".format(ARG_STRING)
+            )
+
+        url, redirect_uri, client_type = args
+
+        # Validate URLs
+        for u in (url, redirect_uri):
+            try:
+                URLValidator()(u)
+            except ValidationError:
+                raise CommandError("URLs provided are invalid. Please provide valid application and redirect URLs.")
+
+        # Validate and map client type to the appropriate django-oauth2-provider constant
+        client_type = client_type.lower()
+        client_type = {
+            'confidential': CONFIDENTIAL,
+            'public': PUBLIC
+        }.get(client_type)
+
+        if client_type is None:
+            raise CommandError("Client type provided is invalid. Please use one of 'confidential' or 'public'.")
+
+        self.fields = {
+            'url': url,
+            'redirect_uri': redirect_uri,
+            'client_type': client_type,
+        }
+
+    def _parse_options(self, options):
+        """Parse the command's options.
+
+        Arguments:
+            options (dict): Options with which the command was called.
+
+        Raises:
+            CommandError, if a user matching the provided username does not exist.
+        """
+        for key in ('username', 'client_name', 'client_id', 'client_secret', 'trusted'):
+            value = options.get(key)
+            if value is not None:
+                self.fields[key] = value
+
+        username = self.fields.pop('username', None)
+        if username is not None:
+            try:
+                self.fields['user'] = User.objects.get(username=username)
+            except User.DoesNotExist:
+                raise CommandError("User matching the provided username does not exist.")
+
+        # The keyword argument 'name' conflicts with that of `call_command()`. We instead
+        # use 'client_name' up to this point, then swap it out for the expected field, 'name'.
+        client_name = self.fields.pop('client_name', None)
+        if client_name is not None:
+            self.fields['name'] = client_name

oauth2_provider/tests/test_create_oauth2_client.py

@@ -0,0 +1,128 @@
+from itertools import product
+import json
+
+import ddt
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+from django.utils.six import StringIO
+from nose.tools import assert_raises
+from provider.oauth2.models import Client
+
+from oauth2_provider.models import TrustedClient
+
+
+User = get_user_model()
+
+
+@ddt.ddt
+class CreateOauth2ClientTests(TestCase):
+    URL = 'https://www.example.com/'
+    REDIRECT_URI = 'https://www.example.com/complete/edx-oidc/'
+    CLIENT_TYPES = ('confidential', 'public')
+    USERNAME = 'username'
+
+    def setUp(self):
+        self.user = User.objects.create(username=self.USERNAME)
+
+    def _call_command(self, args, options=None):
+        """Call the command, capturing its output."""
+        if options is None:
+            options = {}
+
+        out = StringIO()
+        options['stdout'] = out
+
+        call_command('create_oauth2_client', *args, **options)
+
+        return out
+
+    def assert_client_created(self, args, options):
+        """Call the command, verify that the Client was created, and validate the output."""
+        out = self._call_command(args, options)
+        client = Client.objects.get()
+
+        # Verify command output
+        serialized = json.dumps(client.serialize(), indent=4)
+        self.assertIn(serialized, out.getvalue())
+
+        # Verify Client associated with the correct user
+        if options.get('username'):
+            self.assertEqual(self.user, client.user)
+
+        # Verify Client assigned the correct name
+        client_name = options.get('client_name')
+        if client_name:
+            self.assertEqual(client_name, client.name)
+
+        # Verify Client ID and secret overrides
+        for attr in ('client_id', 'client_secret'):
+            value = options.get(attr)
+            if value is not None:
+                self.assertEqual(value, getattr(client, attr))
+
+        # Verify Client designated as trusted
+        if options.get('trusted'):
+            trusted_client = TrustedClient.objects.get()
+            self.assertEqual(client, trusted_client.client)
+
+    # Generate all valid argument and options combinations
+    @ddt.data(*product(
+        # Generate all valid argument combinations
+        product(
+            (URL,),
+            (REDIRECT_URI,),
+            (t for t in CLIENT_TYPES),
+        ),
+        # Generate all valid option combinations
+        (dict(zip(('username', 'client_name', 'client_id', 'client_secret', 'trusted'), p)) for p in product(
+            (USERNAME, None),
+            ('name', None),
+            ('id', None),
+            ('secret', None),
+            (True, False)
+        ))
+    ))
+    @ddt.unpack
+    def test_client_creation(self, args, options):
+        """Verify that the command creates a Client when given valid arguments and options."""
+        self.assert_client_created(args, options)
+
+    @ddt.data(
+        (URL, REDIRECT_URI),
+        (URL, REDIRECT_URI, CLIENT_TYPES[0], CLIENT_TYPES[1]),
+    )
+    def test_argument_cardinality(self, args):
+        """Verify that the command fails when given an incorrect number of arguments."""
+        with assert_raises(CommandError) as e:
+            self._call_command(args, {})
+
+        self.assertIn('Number of arguments provided is invalid.', e.exception.message)
+
+    @ddt.data(
+        ('invalid', REDIRECT_URI, CLIENT_TYPES[0]),
+        (URL, 'invalid', CLIENT_TYPES[0]),
+    )
+    def test_url_validation(self, args):
+        """Verify that the command fails when the provided URLs are invalid."""
+        with assert_raises(CommandError) as e:
+            self._call_command(args)
+
+        self.assertIn('URLs provided are invalid.', e.exception.message)
+
+    def test_client_type_validation(self):
+        """Verify that the command fails when the provided client type is invalid."""
+        with assert_raises(CommandError) as e:
+            self._call_command((self.URL, self.REDIRECT_URI, 'not_a_client_type'))
+
+        self.assertIn('Client type provided is invalid.', e.exception.message)
+
+    def test_username_mismatch(self):
+        with assert_raises(CommandError) as e:
+            self._call_command(
+                (self.URL, self.REDIRECT_URI, self.CLIENT_TYPES[0]),
+                options={'username': 'bad_username'}
+            )
+
+        self.assertIn('User matching the provided username does not exist.', e.exception.message)

oauth2_provider/tests/util.py

@@ -1,5 +1,4 @@
 # pylint: disable=missing-docstring
-
 import os.path
 import urlparse
 

setup.py

@@ -4,7 +4,7 @@
 
 setup(
     name='edx-oauth2-provider',
-    version='0.5.1',
+    version='0.5.2',
     description='Provide OAuth2 access to edX installations',
     author='edX',
     url='https://github.com/edx/edx-oauth2-provider',

test_requirements.txt

@@ -1,3 +1,4 @@
 ddt==0.8.0
+django-nose==1.3
 factory_boy==2.2.1
 mock==1.0.1