Coverage for api/views/oauth2_views.py: 78%
113 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 06:14 +0000
1import json
2import secrets
3import smtplib
4from textwrap import dedent
6from django.conf import settings
7from django.core.cache import cache
8from django.core.mail import send_mail
9from django.db import DataError
10from rest_framework.exceptions import APIException
11from rest_framework.request import Request
12from rest_framework.response import Response
13from rest_framework.reverse import reverse
14from rest_framework.views import APIView
16import structlog
17from drf_spectacular.utils import extend_schema
18from oauth2_provider.contrib.rest_framework.permissions import TokenHasScope
19from oauth2_provider.generators import generate_client_secret
20from oauth2_provider.models import AccessToken
21from oauth2_provider.signals import app_authorized
22from oauth2_provider.views import TokenView as BaseTokenView
23from redis.exceptions import ConnectionError
25from api.docs.oauth2_docs import key_info, register, token
26from api.models import OAuth2Verification, ThrottledApplication
27from api.serializers.oauth2_serializers import (
28 OAuth2KeyInfoSerializer,
29 OAuth2RegistrationSerializer,
30)
31from api.utils.throttle import OnePerSecond, TenPerDay
34logger = structlog.get_logger(__name__)
37class InvalidCredentials(APIException):
38 status_code = 400
39 default_detail = "Invalid credentials"
40 default_code = "invalid_credentials"
43@extend_schema(tags=["auth"])
44class Register(APIView):
45 throttle_classes = (TenPerDay,)
46 # Registration implicitly does not require authentication
47 authentication_classes = ()
49 @register
50 def post(self, request, format=None):
51 """
52 Register an application for access to the Openverse API via OAuth2.
54 Provide a unique application `name`, a brief `description`, and an
55 `email` address where Openverse can contact you. The application `name`
56 must be unique across all registered Openverse API applications.
58 Upon registering, you will receive a `client_id` and `client_secret`,
59 which you can then use to authenticate using the standard OAuth2 flow.
61 > ⚠️ **WARNINGS:**
62 > - Store your `client_id` and `client_secret` because you will not be
63 > able to retrieve them later.
64 > - You must keep `client_secret` confidential, as anybody with your
65 > `client_secret` can impersonate your application.
67 You must verify your email address by clicking the link sent to you in
68 an email. Until you do that, the application will be subject to the same
69 rate limits as an anonymous user.
70 """
72 # Store the registration information the developer gave us.
73 serialized = OAuth2RegistrationSerializer(data=request.data)
74 if not serialized.is_valid():
75 return Response(status=400, data=serialized.errors)
76 else:
77 serialized.save()
79 # Produce a client ID, client secret, and authorize the application in
80 # the OAuth2 backend.
81 client_secret = generate_client_secret()
82 new_application = ThrottledApplication(
83 name=serialized.validated_data["name"],
84 skip_authorization=False,
85 client_type="Confidential",
86 authorization_grant_type="client-credentials",
87 verified=False,
88 client_secret=client_secret,
89 )
90 new_application.save()
91 # Send a verification email.
92 verification = OAuth2Verification(
93 email=serialized.validated_data["email"],
94 code=secrets.token_urlsafe(64),
95 associated_application=new_application,
96 )
97 verification.save()
98 token = verification.code
99 link = request.build_absolute_uri(reverse("verify-email", [token]))
100 verification_msg = dedent(
101 f"""
102 To verify your Openverse API credentials, click on the following link:
104 {link}
106 If you believe you received this message in error, please disregard it.
107 """
108 )
109 try:
110 send_mail(
111 subject="Verify your API credentials",
112 message=verification_msg,
113 from_email=settings.EMAIL_SENDER,
114 recipient_list=[verification.email],
115 fail_silently=False,
116 )
117 except smtplib.SMTPException as e:
118 logger.error("Failed to send API verification email!", exc=e, exc_info=True)
119 # Give the user their newly created credentials.
120 return Response(
121 status=201,
122 data={
123 "client_id": new_application.client_id,
124 "client_secret": client_secret,
125 "name": new_application.name,
126 "msg": "Check your email for a verification link.",
127 },
128 )
131class VerifyEmail(APIView):
132 """Enable a user's OAuth2 key upon visiting the emailed verification link."""
134 schema = None # Hide this view from the OpenAPI schema.
136 def get(self, request, code, format=None):
137 try:
138 verification = OAuth2Verification.objects.get(code=code)
139 application_pk = verification.associated_application.pk
140 ThrottledApplication.objects.filter(pk=application_pk).update(verified=True)
141 verification.delete()
142 return Response(
143 status=200,
144 data={
145 "msg": "Successfully verified email. Your OAuth2 "
146 "credentials are now active."
147 },
148 )
149 except OAuth2Verification.DoesNotExist:
150 return Response(
151 status=500,
152 data={
153 "msg": "Invalid verification code. Did you validate your "
154 "credentials already?"
155 },
156 )
159@extend_schema(tags=["auth"])
160class TokenView(APIView, BaseTokenView):
161 # Token view is pre-authentication
162 authentication_classes = ()
163 permission_classes = ()
165 @token
166 def post(self, request):
167 """
168 Get an access token using client credentials.
170 To authenticate your requests to the Openverse API, you need to provide
171 an access token as a bearer token in the `Authorization` header of your
172 requests. This endpoint takes your client ID and secret, and issues an
173 access token.
175 > **NOTE:** This endpoint only accepts data as
176 > `application/x-www-form-urlencoded`. Any other encoding will not work.
178 Once your access token expires, you can request another one from this
179 endpoint.
180 """
182 try:
183 res = super().post(request._request)
184 except DataError:
185 raise InvalidCredentials()
186 data = json.loads(res.content)
187 return Response(data, status=res.status_code)
190@app_authorized.connect
191def log_app_authorized(request, token: AccessToken, **kwargs):
192 application = token.application
193 logger.info(
194 "client_application_authorized",
195 application_id=application.id,
196 application_name=application.name,
197 application_verified=application.verified,
198 )
201@extend_schema(tags=["auth"])
202class CheckRates(APIView):
203 throttle_classes = (OnePerSecond,)
204 permission_classes = (TokenHasScope,)
205 required_scopes = ("read",)
207 @key_info
208 def get(self, request: Request, format=None):
209 """
210 Get information about your API key.
212 You can use this endpoint to get information about your API key such as
213 `requests_this_minute`, `requests_today`, and `rate_limit_model`.
215 > ℹ️ **NOTE:** If you get a 401 Unauthorized, it means your token is invalid
216 > (malformed, non-existent, or expired).
217 """
218 application: ThrottledApplication = request.auth.application
220 client_id = application.client_id
222 throttle_type = application.rate_limit_model
223 throttle_key = "throttle_{scope}_{client_id}"
224 if throttle_type == "standard": 224 ↛ 231line 224 didn't jump to line 231 because the condition on line 224 was always true
225 sustained_throttle_key = throttle_key.format(
226 scope="oauth2_client_credentials_sustained", client_id=client_id
227 )
228 burst_throttle_key = throttle_key.format(
229 scope="oauth2_client_credentials_burst", client_id=client_id
230 )
231 elif throttle_type == "enhanced":
232 sustained_throttle_key = throttle_key.format(
233 scope="enhanced_oauth2_client_credentials_sustained",
234 client_id=client_id,
235 )
236 burst_throttle_key = throttle_key.format(
237 scope="enhanced_oauth2_client_credentials_burst", client_id=client_id
238 )
239 elif throttle_type == "exempt":
240 burst_throttle_key = sustained_throttle_key = throttle_key.format(
241 scope="exempt_oauth2_client_credentials_burst", client_id=client_id
242 )
243 else:
244 return APIException("Unknown API key rate limit type")
246 try:
247 sustained_requests_list = cache.get(sustained_throttle_key)
248 sustained_requests = (
249 len(sustained_requests_list) if sustained_requests_list else None
250 )
251 burst_requests_list = cache.get(burst_throttle_key)
252 burst_requests = len(burst_requests_list) if burst_requests_list else None
253 status = 200
254 except ConnectionError:
255 logger.warning("Redis connect failed, cannot get key usage.")
256 burst_requests = None
257 sustained_requests = None
258 status = 424
260 response_data = OAuth2KeyInfoSerializer(
261 {
262 "requests_this_minute": burst_requests,
263 "requests_today": sustained_requests,
264 "rate_limit_model": throttle_type,
265 "verified": application.verified,
266 }
267 )
268 return Response(status=status, data=response_data.data)