swadhinbiswas day5 · 2/9/2024
3a91059
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
import os
import jwt
from configparser import ConfigParser
def set_up():
"""Sets up configuration for the app"""
env = os.getenv("ENV", ".config")
if env == ".config":
config = ConfigParser()
config.read(".config")
config = config["AUTH0"]
else:
config = {
"DOMAIN": os.getenv("DOMAIN", "your.domain.com"),
"API_AUDIENCE": os.getenv("API_AUDIENCE", "your.audience.com"),
"ISSUER": os.getenv("ISSUER", "https://your.domain.com/"),
"ALGORITHMS": os.getenv("ALGORITHMS", "RS256"),
}
return config
class VerifyToken():
"""Does all the token verification using PyJWT"""
def __init__(self, token, permissions=None, scopes=None):
self.token = token
self.permissions = permissions
self.scopes = scopes
self.config = set_up()
# This gets the JWKS from a given URL and does processing so you can use any of
# the keys available
jwks_url = f'https://{self.config["DOMAIN"]}/.well-known/jwks.json'
self.jwks_client = jwt.PyJWKClient(jwks_url)
def verify(self):
# This gets the 'kid' from the passed token
try:
self.signing_key = self.jwks_client.get_signing_key_from_jwt(
self.token
).key
except jwt.exceptions.PyJWKClientError as error:
return {"status": "error", "msg": error.__str__()}
except jwt.exceptions.DecodeError as error:
return {"status": "error", "msg": error.__str__()}
try:
payload = jwt.decode(
self.token,
self.signing_key,
algorithms=self.config["ALGORITHMS"],
audience=self.config["API_AUDIENCE"],
issuer=self.config["ISSUER"],
)
except Exception as e:
return {"status": "error", "message": str(e)}
if self.scopes:
result = self._check_claims(payload, 'scope', str, self.scopes.split(' '))
if result.get("error"):
return result
if self.permissions:
result = self._check_claims(payload, 'permissions', list, self.permissions)
if result.get("error"):
return result
return payload
def _check_claims(self, payload, claim_name, claim_type, expected_value):
instance_check = isinstance(payload[claim_name], claim_type)
result = {"status": "success", "status_code": 200}
payload_claim = payload[claim_name]
if claim_name not in payload or not instance_check:
result["status"] = "error"
result["status_code"] = 400
result["code"] = f"missing_{claim_name}"
result["msg"] = f"No claim '{claim_name}' found in token."
return result
if claim_name == 'scope':
payload_claim = payload[claim_name].split(' ')
for value in expected_value:
if value not in payload_claim:
result["status"] = "error"
result["status_code"] = 403
result["code"] = f"insufficient_{claim_name}"
result["msg"] = (f"Insufficient {claim_name} ({value}). You don't have "
"access to this resource")
return result
return result