import requests from django.conf import settings from django.db import models from django.db.models import JSONField from oauth2_provider.models import AbstractApplication, AbstractAccessToken, AbstractGrant, AbstractRefreshToken from oauth2_provider.scopes import get_scopes_backend from oauth2_provider.settings import oauth2_settings import uuid class Resource(models.Model): uuid = models.UUIDField(primary_key=True, editable=False, default=uuid.uuid4, unique=True, db_index=True) name = models.CharField(max_length=255) user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True, related_name="resources" ) expires = models.DateTimeField() def __str__(self): return self.name class Application(AbstractApplication): """ Application model for use with Django OAuth Toolkit that allows the scopes available to an application to be restricted on a per-application basis. """ user = models.ForeignKey( settings.AUTH_USER_MODEL, related_name="%(app_label)s_%(class)s", on_delete=models.PROTECT ) allowed_scope = models.TextField(blank=True) resource = models.OneToOneField( Resource, models.PROTECT, blank=True, null=True, help_text='The resource of application.', related_name='application' ) @property def allowed_scopes(self): """ Returns the set of allowed scope names for this application. """ all_scopes = set(get_scopes_backend().get_all_scopes().keys()) app_scopes = set(self.allowed_scope.split()) return app_scopes.intersection(all_scopes) class Scope(models.Model): """ Django model for an OAuth scope. """ #: The application that created the scope # NOTE: This is not used to limit access to the scope in any way - we want the # scope to be available to other applications in order to request access # to the resource it protects! application = models.ForeignKey( oauth2_settings.APPLICATION_MODEL, models.CASCADE, # This field is nullable because it is only set for scopes created by # external resource servers, which have a corresponding OAuth application # record on the authorisation server blank=True, null=True, help_text='The application to which the scope belongs.', related_name='scopes' ) resource = models.ForeignKey( Resource, models.PROTECT, blank=True, null=True, help_text='The resource of scope.', related_name='scopes' ) #: The name of the scope name = models.CharField( max_length=255, unique=True, help_text='The name of the scope.' ) #: A brief description of the scope description = models.TextField( help_text='A brief description of the scope. This text is displayed ' 'to users when authorising access for the scope.' ) #: Indicates if the scope should be included in the default scopes is_default = models.BooleanField( default=False, help_text='Indicates if this scope should be included in the default scopes.' ) @property def final_name(self): args = [] if self.resource: args.append(self.resource.name) args.append(self.name) return '.'.join(args) @property def final_description(self): resource_name = self.resource and self.resource.name if resource_name: return f"{resource_name} -> {self.description}" return self.description @classmethod def register(cls, name, description, is_default=False): """ Registers a scope with the given values. It always creates an instance in the local database, but if this resource server has an external authorisation server, it will also register the scope there. Returns ``True`` on success. Should raise on failure. """ endpoint = settings.RESOURCE_SERVER_REGISTER_SCOPE_URL if endpoint: #  If the endpoint is set, make the callout to the authz server token = "Bearer {}".format(oauth2_settings.RESOURCE_SERVER_AUTH_TOKEN) #  Let any failures bubble up # The idea is to call this method during deployment as a post-migrate #  hook, so we want failures to halt the deployment response = requests.post( endpoint, json={ 'name': name, 'description': description, 'is_default': is_default }, headers={"Authorization": token} ) #  Raise the exception for anything other than 20x responses response.raise_for_status() #  Always create/update the scope record locally _ = Scope.objects.update_or_create( name=name, defaults={'description': description, 'is_default': is_default} ) return True class Grant(AbstractGrant): application = models.ForeignKey( oauth2_settings.APPLICATION_MODEL, on_delete=models.CASCADE, related_name='grants' ) class RefreshToken(AbstractRefreshToken): application = models.ForeignKey( oauth2_settings.APPLICATION_MODEL, on_delete=models.CASCADE, related_name='refresh_tokens') class AccessToken(AbstractAccessToken): detail = JSONField(null=True, blank=True) application = models.ForeignKey( oauth2_settings.APPLICATION_MODEL, on_delete=models.CASCADE, blank=True, null=True, related_name='access_tokens' )