1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187
|
import json
import time
from authlib.oauth2.rfc6749.grants import RefreshTokenGrant as _RefreshTokenGrant
from .models import Client
from .models import OAuth2Token
from .models import User
from .oauth2_server import TestCase
class RefreshTokenGrant(_RefreshTokenGrant):
def authenticate_refresh_token(self, refresh_token):
try:
item = OAuth2Token.objects.get(refresh_token=refresh_token)
if item.is_refresh_token_active():
return item
except OAuth2Token.DoesNotExist:
return None
def authenticate_user(self, credential):
return credential.user
def revoke_old_credential(self, credential):
now = int(time.time())
credential.access_token_revoked_at = now
credential.refresh_token_revoked_at = now
credential.save()
return credential
class RefreshTokenTest(TestCase):
def create_server(self):
server = super().create_server()
server.register_grant(RefreshTokenGrant)
return server
def prepare_client(self, grant_type="refresh_token", scope=""):
user = User(username="foo")
user.save()
client = Client(
user_id=user.pk,
client_id="client",
client_secret="secret",
scope=scope,
grant_type=grant_type,
token_endpoint_auth_method="client_secret_basic",
default_redirect_uri="https://a.b",
)
client.save()
def prepare_token(self, scope="profile", user_id=1):
token = OAuth2Token(
user_id=user_id,
client_id="client",
token_type="bearer",
access_token="a1",
refresh_token="r1",
scope=scope,
expires_in=3600,
)
token.save()
def test_invalid_client(self):
server = self.create_server()
self.prepare_client()
request = self.factory.post(
"/oauth/token",
data={"grant_type": "refresh_token", "refresh_token": "foo"},
)
resp = server.create_token_response(request)
assert resp.status_code == 401
data = json.loads(resp.content)
assert data["error"] == "invalid_client"
request = self.factory.post(
"/oauth/token",
data={"grant_type": "refresh_token", "refresh_token": "foo"},
HTTP_AUTHORIZATION=self.create_basic_auth("invalid", "secret"),
)
resp = server.create_token_response(request)
assert resp.status_code == 401
data = json.loads(resp.content)
assert data["error"] == "invalid_client"
def test_invalid_refresh_token(self):
self.prepare_client()
server = self.create_server()
auth_header = self.create_basic_auth("client", "secret")
request = self.factory.post(
"/oauth/token",
data={"grant_type": "refresh_token"},
HTTP_AUTHORIZATION=auth_header,
)
resp = server.create_token_response(request)
assert resp.status_code == 400
data = json.loads(resp.content)
assert data["error"] == "invalid_request"
assert "Missing" in data["error_description"]
request = self.factory.post(
"/oauth/token",
data={"grant_type": "refresh_token", "refresh_token": "invalid"},
HTTP_AUTHORIZATION=auth_header,
)
resp = server.create_token_response(request)
assert resp.status_code == 400
data = json.loads(resp.content)
assert data["error"] == "invalid_grant"
def test_invalid_scope(self):
server = self.create_server()
server.scopes_supported = ["profile"]
self.prepare_client()
self.prepare_token()
request = self.factory.post(
"/oauth/token",
data={
"grant_type": "refresh_token",
"refresh_token": "r1",
"scope": "invalid",
},
HTTP_AUTHORIZATION=self.create_basic_auth("client", "secret"),
)
resp = server.create_token_response(request)
assert resp.status_code == 400
data = json.loads(resp.content)
assert data["error"] == "invalid_scope"
def test_authorize_tno_scope(self):
server = self.create_server()
self.prepare_client()
self.prepare_token()
request = self.factory.post(
"/oauth/token",
data={
"grant_type": "refresh_token",
"refresh_token": "r1",
},
HTTP_AUTHORIZATION=self.create_basic_auth("client", "secret"),
)
resp = server.create_token_response(request)
assert resp.status_code == 200
data = json.loads(resp.content)
assert "access_token" in data
def test_authorize_token_scope(self):
server = self.create_server()
self.prepare_client()
self.prepare_token()
request = self.factory.post(
"/oauth/token",
data={
"grant_type": "refresh_token",
"refresh_token": "r1",
"scope": "profile",
},
HTTP_AUTHORIZATION=self.create_basic_auth("client", "secret"),
)
resp = server.create_token_response(request)
assert resp.status_code == 200
data = json.loads(resp.content)
assert "access_token" in data
def test_revoke_old_token(self):
server = self.create_server()
self.prepare_client()
self.prepare_token()
request = self.factory.post(
"/oauth/token",
data={
"grant_type": "refresh_token",
"refresh_token": "r1",
"scope": "profile",
},
HTTP_AUTHORIZATION=self.create_basic_auth("client", "secret"),
)
resp = server.create_token_response(request)
assert resp.status_code == 200
data = json.loads(resp.content)
assert "access_token" in data
resp = server.create_token_response(request)
assert resp.status_code == 400
|