fix(oauth): improve oauth token validation and error handling

- add application status check in grant validation
- better error handling for lock acquisition failures
- add tests for concurrent grant usage

Author: mdtro

src/sentry/web/frontend/oauth_token.py

@@ -18,6 +18,7 @@
 from sentry.models.apitoken import ApiToken
 from sentry.sentry_apps.token_exchange.util import GrantTypes
 from sentry.utils import json, metrics
+from sentry.utils.locking import UnableToAcquireLock
 from sentry.web.frontend.base import control_silo_view
 from sentry.web.frontend.openidtoken import OpenIDToken
 
@@ -128,7 +129,9 @@ def post(self, request: Request) -> HttpResponse:
     def get_access_tokens(self, request: Request, application: ApiApplication) -> dict:
         code = request.POST.get("code")
         try:
-            grant = ApiGrant.objects.get(application=application, code=code)
+            grant = ApiGrant.objects.get(
+                application=application, application__status=ApiApplicationStatus.active, code=code
+            )
         except ApiGrant.DoesNotExist:
             return {"error": "invalid_grant", "reason": "invalid grant"}
 
@@ -141,7 +144,12 @@ def get_access_tokens(self, request: Request, application: ApiApplication) -> di
         elif grant.redirect_uri != redirect_uri:
             return {"error": "invalid_grant", "reason": "invalid redirect URI"}
 
-        token_data = {"token": ApiToken.from_grant(grant=grant)}
+        try:
+            token_data = {"token": ApiToken.from_grant(grant=grant)}
+        except UnableToAcquireLock:
+            # TODO(mdtro): we should return a 409 status code here
+            return {"error": "invalid_grant", "reason": "invalid grant"}
+
         if grant.has_scope("openid") and options.get("codecov.signing_secret"):
             open_id_token = OpenIDToken(
                 request.POST.get("client_id"),
@@ -150,6 +158,7 @@ def get_access_tokens(self, request: Request, application: ApiApplication) -> di
                 nonce=request.POST.get("nonce"),
             )
             token_data["id_token"] = open_id_token.get_signed_id_token(grant=grant)
+
         return token_data
 
     def get_refresh_token(self, request: Request, application: ApiApplication) -> dict:

tests/sentry/web/frontend/test_oauth_token.py

@@ -2,6 +2,7 @@
 
 from django.utils import timezone
 
+from sentry.locks import locks
 from sentry.models.apiapplication import ApiApplication
 from sentry.models.apigrant import ApiGrant
 from sentry.models.apitoken import ApiToken
@@ -203,6 +204,28 @@ def test_one_time_use_grant(self):
         )
         assert resp.status_code == 400
 
+    def test_grant_lock(self):
+        self.login_as(self.user)
+
+        # Simulate a concurrent request by using an existing grant
+        # that has its grant lock taken out.
+        lock = locks.get(ApiGrant.get_lock_key(self.grant.id), duration=10, name="api_grant")
+        lock.acquire()
+
+        # Attempt to create a token with the same grant
+        # This should fail because the lock is held by the previous request
+        resp = self.client.post(
+            self.path,
+            {
+                "grant_type": "authorization_code",
+                "code": self.grant.code,
+                "client_id": self.application.client_id,
+                "client_secret": self.client_secret,
+            },
+        )
+        assert resp.status_code == 400
+        assert resp.json() == {"error": "invalid_grant"}
+
     def test_invalid_redirect_uri(self):
         self.login_as(self.user)