164 lines
No EOL
6 KiB
Python
Executable file
164 lines
No EOL
6 KiB
Python
Executable file
import base64
|
|
import binascii
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from urllib.parse import unquote_plus
|
|
|
|
import requests
|
|
# import service_clients
|
|
from django.contrib.auth import get_user_model
|
|
from django.utils.timezone import make_aware
|
|
from oauth2_provider.models import get_access_token_model
|
|
from oauth2_provider.oauth2_validators import OAuth2Validator as BaseOAuth2Validator
|
|
import base64
|
|
import binascii
|
|
import http.client
|
|
import inspect
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from collections import OrderedDict
|
|
from datetime import datetime, timedelta
|
|
from urllib.parse import unquote_plus
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
from django.contrib.auth import authenticate, get_user_model
|
|
from django.contrib.auth.hashers import check_password, identify_hasher
|
|
from django.core.exceptions import ObjectDoesNotExist
|
|
from django.db import transaction
|
|
from django.db.models import Q
|
|
from django.http import HttpRequest
|
|
from django.utils import dateformat, timezone
|
|
from django.utils.crypto import constant_time_compare
|
|
from django.utils.timezone import make_aware
|
|
from django.utils.translation import gettext_lazy as _
|
|
from jwcrypto import jws, jwt
|
|
from jwcrypto.common import JWException
|
|
from jwcrypto.jwt import JWTExpired
|
|
from oauthlib.oauth2.rfc6749 import utils
|
|
from oauthlib.openid import RequestValidator
|
|
|
|
from oauth2_provider.models import (
|
|
AbstractApplication,
|
|
get_access_token_model,
|
|
get_application_model,
|
|
get_grant_model,
|
|
get_id_token_model,
|
|
get_refresh_token_model,
|
|
)
|
|
from oauth2_provider.settings import oauth2_settings
|
|
from oauth2_provider.utils import get_timezone
|
|
|
|
Application = get_application_model()
|
|
AccessToken = get_access_token_model()
|
|
IDToken = get_id_token_model()
|
|
Grant = get_grant_model()
|
|
RefreshToken = get_refresh_token_model()
|
|
UserModel = get_user_model()
|
|
|
|
|
|
|
|
log = logging.getLogger("oauth2_provider")
|
|
|
|
|
|
class OAuth2Validator(BaseOAuth2Validator): # pylint: disable=w0223
|
|
def get_or_create_user_from_content(self, content):
|
|
"""
|
|
An optional layer to define where to store the profile in `UserModel` or a separate model.
|
|
For example `UserOAuth`, where `user = models.OneToOneField(UserModel)` .
|
|
|
|
The function is called after checking that username is in the content.
|
|
|
|
Returns an UserModel instance;
|
|
"""
|
|
user, _ = UserModel.objects.get_or_create(pk=content["username"])
|
|
return user
|
|
|
|
def _get_token_from_authentication_server(
|
|
self, token, introspection_url, introspection_token, introspection_credentials
|
|
):
|
|
# NOTICE: onlu change from orginal method is that we create application here
|
|
"""Use external introspection endpoint to "crack open" the token.
|
|
:param introspection_url: introspection endpoint URL
|
|
:param introspection_token: Bearer token
|
|
:param introspection_credentials: Basic Auth credentials (id,secret)
|
|
:return: :class:`models.AccessToken`
|
|
|
|
Some RFC 7662 implementations (including this one) use a Bearer token while others use Basic
|
|
Auth. Depending on the external AS's implementation, provide either the introspection_token
|
|
or the introspection_credentials.
|
|
|
|
If the resulting access_token identifies a username (e.g. Authorization Code grant), add
|
|
that user to the UserModel. Also cache the access_token up until its expiry time or a
|
|
configured maximum time.
|
|
|
|
"""
|
|
headers = None
|
|
if introspection_token:
|
|
headers = {"Authorization": "Bearer {}".format(introspection_token)}
|
|
elif introspection_credentials:
|
|
client_id = introspection_credentials[0].encode("utf-8")
|
|
client_secret = introspection_credentials[1].encode("utf-8")
|
|
basic_auth = base64.b64encode(client_id + b":" + client_secret)
|
|
headers = {"Authorization": "Basic {}".format(basic_auth.decode("utf-8"))}
|
|
|
|
try:
|
|
response = requests.post(introspection_url, data={"token": token}, headers=headers)
|
|
except requests.exceptions.RequestException:
|
|
log.exception("Introspection: Failed POST to %r in token lookup", introspection_url)
|
|
return None
|
|
|
|
# Log an exception when response from auth server is not successful
|
|
if response.status_code != http.client.OK:
|
|
log.exception(
|
|
"Introspection: Failed to get a valid response "
|
|
"from authentication server. Status code: {}, "
|
|
"Reason: {}.".format(response.status_code, response.reason)
|
|
)
|
|
return None
|
|
|
|
try:
|
|
content = response.json()
|
|
except ValueError:
|
|
log.exception("Introspection: Failed to parse response as json")
|
|
return None
|
|
|
|
if "active" in content and content["active"] is True:
|
|
if "username" in content:
|
|
user = self.get_or_create_user_from_content(content)
|
|
else:
|
|
user = None
|
|
|
|
max_caching_time = datetime.now() + timedelta(
|
|
seconds=oauth2_settings.RESOURCE_SERVER_TOKEN_CACHING_SECONDS
|
|
)
|
|
|
|
if "exp" in content:
|
|
expires = datetime.utcfromtimestamp(content["exp"])
|
|
if expires > max_caching_time:
|
|
expires = max_caching_time
|
|
else:
|
|
expires = max_caching_time
|
|
|
|
scope = content.get("scope", "")
|
|
|
|
if settings.USE_TZ:
|
|
expires = make_aware(
|
|
expires, timezone=get_timezone(oauth2_settings.AUTHENTICATION_SERVER_EXP_TIME_ZONE)
|
|
)
|
|
|
|
# TODO: get application owner and put it here
|
|
access_token, _created = AccessToken.objects.update_or_create(
|
|
token=token,
|
|
defaults={
|
|
"user": user,
|
|
"client_id": content.get("client_id", ""),
|
|
"client_owner": user,
|
|
"application": None,
|
|
"scope": scope,
|
|
"expires": expires,
|
|
},
|
|
)
|
|
|
|
return access_token |