pandaharvester/harvestermisc/frontend_utils.py

0001 import time
0002
0003 import jwt
0004 from pandaharvester.harvesterconfig import harvester_config
0005
0006
0007 class HarvesterToken:
0008     """
0009     Methods of JSON Web Token used in harvester frontend
0010     """
0011
0012     algorithm = "HS256"
0013
0014     def __init__(self, **kwarg):
0015         # load secret from file
0016         with open(harvester_config.frontend.secretFile) as _f:
0017             self.secret = _f.read()
0018         # wheter verify the token
0019         self.verify = True
0020         if harvester_config.frontend.verifyToken is False:
0021             self.verify = False
0022         # default token lifetime in second. 4 days.
0023         self.default_lifetime = 345600
0024         # default payload spec
0025         self.default_payload_dict = {
0026             "sub": "Subject",
0027             "exp": 0,
0028             "iss": harvester_config.master.harvester_id,
0029             "iat": 0,
0030         }
0031
0032     def generate(self, payload=None, header=None):
0033         """
0034         Generate a harvester token.
0035         Additional payload can be upadated, argument as a dict.
0036         """
0037         timestamp_now = int(time.time())
0038         payload_dict = self.default_payload_dict.copy()
0039         payload_dict["iat"] = timestamp_now
0040         payload_dict["exp"] = timestamp_now + self.default_lifetime
0041         if payload:
0042             payload_dict.update(payload)
0043         token = jwt.encode(payload_dict, key=self.secret, algorithm=self.algorithm, headers=header)
0044         return token
0045
0046     def get_payload(self, token):
0047         """
0048         Decode a harvester token to a python object, typically dict.
0049         One can set verify=False if other proxy/gateway service already verifies token.
0050         """
0051         payload = jwt.decode(token, key=self.secret, algorithms=self.algorithm, verify=self.verify)
0052         return payload