257 lines
8.6 KiB
Python
257 lines
8.6 KiB
Python
import base64
|
|
import json
|
|
import secrets
|
|
from urllib.parse import urlparse, urlunparse
|
|
|
|
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
from django.http import (
|
|
HttpResponse,
|
|
HttpResponseForbidden,
|
|
HttpResponseRedirect,
|
|
JsonResponse,
|
|
)
|
|
from django.shortcuts import render
|
|
from django.utils import timezone
|
|
from django.utils.decorators import method_decorator
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from django.views.generic import View
|
|
|
|
from api.models import Application, Authorization, Token
|
|
|
|
|
|
class OauthRedirect(HttpResponseRedirect):
|
|
def __init__(self, redirect_uri, **kwargs):
|
|
url_parts = urlparse(redirect_uri)
|
|
self.allowed_schemes = [url_parts.scheme]
|
|
# Either add or join the query section
|
|
url_parts = list(url_parts)
|
|
|
|
query_string = url_parts[4]
|
|
|
|
for key, value in kwargs.items():
|
|
if value is None:
|
|
continue
|
|
if not query_string:
|
|
query_string = f"{key}={value}"
|
|
else:
|
|
query_string += f"&{key}={value}"
|
|
|
|
url_parts[4] = query_string
|
|
super().__init__(urlunparse(url_parts))
|
|
|
|
|
|
def get_json_and_formdata(request):
|
|
# Did they submit JSON?
|
|
if request.content_type == "application/json" and request.body.strip():
|
|
return json.loads(request.body)
|
|
# Fall back to form data
|
|
value = {}
|
|
for key, item in request.POST.items():
|
|
value[key] = item
|
|
for key, item in request.GET.items():
|
|
value[key] = item
|
|
return value
|
|
|
|
|
|
class AuthorizationView(LoginRequiredMixin, View):
|
|
"""
|
|
Asks the user to authorize access.
|
|
|
|
Could maybe be a FormView, but things are weird enough we just handle the
|
|
POST manually.
|
|
"""
|
|
|
|
def get(self, request):
|
|
redirect_uri = self.request.GET["redirect_uri"]
|
|
scope = self.request.GET.get("scope", "read")
|
|
state = self.request.GET.get("state")
|
|
|
|
response_type = self.request.GET.get("response_type")
|
|
if response_type != "code":
|
|
return render(
|
|
request,
|
|
"api/oauth_error.html",
|
|
{"error": f"Invalid response type '{response_type}'"},
|
|
)
|
|
|
|
application = Application.objects.filter(
|
|
client_id=self.request.GET.get("client_id"),
|
|
).first()
|
|
|
|
if application is None:
|
|
return render(
|
|
request, "api/oauth_error.html", {"error": "Invalid client_id"}
|
|
)
|
|
|
|
if application.redirect_uris and redirect_uri not in application.redirect_uris:
|
|
return render(
|
|
request,
|
|
"api/oauth_error.html",
|
|
{"error": "Invalid application redirect URI"},
|
|
)
|
|
|
|
context = {
|
|
"application": application,
|
|
"state": state,
|
|
"redirect_uri": redirect_uri,
|
|
"scope": scope,
|
|
"identities": self.request.user.identities.all(),
|
|
}
|
|
return render(request, "api/oauth_authorize.html", context)
|
|
|
|
def post(self, request):
|
|
post_data = get_json_and_formdata(request)
|
|
# Grab the application and other details again
|
|
redirect_uri = post_data["redirect_uri"]
|
|
scope = post_data["scope"]
|
|
application = Application.objects.get(client_id=post_data["client_id"])
|
|
# Get the identity
|
|
identity = self.request.user.identities.get(pk=post_data["identity"])
|
|
|
|
extra_args = {}
|
|
if post_data.get("state"):
|
|
extra_args["state"] = post_data["state"]
|
|
|
|
# Make a token
|
|
token = Authorization.objects.create(
|
|
application=application,
|
|
user=self.request.user,
|
|
identity=identity,
|
|
code=secrets.token_urlsafe(43),
|
|
redirect_uri=redirect_uri,
|
|
scopes=scope.split(),
|
|
)
|
|
# If it's an out of band request, show the code
|
|
if redirect_uri == "urn:ietf:wg:oauth:2.0:oob":
|
|
return render(request, "api/oauth_code.html", {"code": token.code})
|
|
# Redirect with the token's code
|
|
return OauthRedirect(redirect_uri, code=token.code, **extra_args)
|
|
|
|
|
|
def extract_client_info_from_basic_auth(request):
|
|
if "authorization" in request.headers:
|
|
auth = request.headers["authorization"].split()
|
|
if len(auth) == 2:
|
|
if auth[0].lower() == "basic":
|
|
client_id, client_secret = (
|
|
base64.b64decode(auth[1]).decode("utf8").split(":", 1)
|
|
)
|
|
|
|
return client_id, client_secret
|
|
return None, None
|
|
|
|
|
|
@method_decorator(csrf_exempt, name="dispatch")
|
|
class TokenView(View):
|
|
def verify_code(
|
|
self, authorization: Authorization, client_id, client_secret, redirect_uri
|
|
):
|
|
application = authorization.application
|
|
return (
|
|
application.client_id == client_id
|
|
and application.client_secret == client_secret
|
|
and authorization.redirect_uri == redirect_uri
|
|
)
|
|
|
|
def post(self, request):
|
|
post_data = get_json_and_formdata(request)
|
|
auth_client_id, auth_client_secret = extract_client_info_from_basic_auth(
|
|
request
|
|
)
|
|
post_data.setdefault("client_id", auth_client_id)
|
|
post_data.setdefault("client_secret", auth_client_secret)
|
|
|
|
grant_type = post_data.get("grant_type")
|
|
if grant_type not in (
|
|
"authorization_code",
|
|
"client_credentials",
|
|
):
|
|
return JsonResponse({"error": "invalid_grant_type"}, status=400)
|
|
|
|
if grant_type == "client_credentials":
|
|
# TODO: Implement client credentials flow
|
|
return JsonResponse(
|
|
{
|
|
"error": "invalid_grant_type",
|
|
"error_description": "client credential flow not implemented",
|
|
},
|
|
status=400,
|
|
)
|
|
elif grant_type == "authorization_code":
|
|
code = post_data.get("code")
|
|
if not code:
|
|
return JsonResponse(
|
|
{
|
|
"error": "invalid_request",
|
|
"error_description": "Required param : code",
|
|
},
|
|
status=400,
|
|
)
|
|
|
|
authorization = Authorization.objects.get(code=code)
|
|
if (
|
|
not authorization
|
|
or timezone.now() - authorization.created
|
|
> timezone.timedelta(seconds=authorization.valid_for_seconds)
|
|
):
|
|
return JsonResponse({"error": "access_denied"}, status=401)
|
|
|
|
application = Application.objects.filter(
|
|
client_id=post_data["client_id"],
|
|
client_secret=post_data["client_secret"],
|
|
).first()
|
|
|
|
code_verified = self.verify_code(
|
|
authorization,
|
|
client_id=post_data.get("client_id"),
|
|
client_secret=post_data.get("client_secret"),
|
|
redirect_uri=post_data.get("redirect_uri"),
|
|
)
|
|
|
|
if not application or authorization.token or not code_verified:
|
|
# this authorization code has already been used
|
|
return JsonResponse({"error": "access_denied"}, status=401)
|
|
|
|
token = Token.objects.create(
|
|
application=application,
|
|
user=authorization.user,
|
|
identity=authorization.identity,
|
|
token=secrets.token_urlsafe(43),
|
|
scopes=authorization.scopes,
|
|
)
|
|
token.save()
|
|
# Return them the token
|
|
return JsonResponse(
|
|
{
|
|
"access_token": token.token,
|
|
"token_type": "Bearer",
|
|
"scope": " ".join(token.scopes),
|
|
"created_at": int(token.created.timestamp()),
|
|
}
|
|
)
|
|
|
|
|
|
@method_decorator(csrf_exempt, name="dispatch")
|
|
class RevokeTokenView(View):
|
|
def post(self, request):
|
|
post_data = get_json_and_formdata(request)
|
|
auth_client_id, auth_client_secret = extract_client_info_from_basic_auth(
|
|
request
|
|
)
|
|
post_data.setdefault("client_id", auth_client_id)
|
|
post_data.setdefault("client_secret", auth_client_secret)
|
|
token_str = post_data["token"]
|
|
|
|
application = Application.objects.filter(
|
|
client_id=post_data["client_id"],
|
|
client_secret=post_data["client_secret"],
|
|
).first()
|
|
|
|
token = Token.objects.filter(application=application, token=token_str).first()
|
|
if token is None:
|
|
return HttpResponseForbidden()
|
|
|
|
token.revoked = timezone.now()
|
|
token.save()
|
|
return HttpResponse("")
|