pandaharvester/harvestercredmanager/token_key_cred_manager.py

0001 import datetime
0002 import os.path
0003 import subprocess
0004
0005 from pandaharvester.harvestercore import core_utils
0006 from pandaharvester.harvestercore.communicator_pool import CommunicatorPool
0007
0008 from .base_cred_manager import BaseCredManager
0009
0010 # logger
0011 _logger = core_utils.setup_logger("token_key_cred_manager")
0012
0013
0014 # credential manager for token key
0015 class TokenKeyCredManager(BaseCredManager):
0016     # constructor
0017     def __init__(self, **kwarg):
0018         BaseCredManager.__init__(self, **kwarg)
0019         # set default
0020         if not hasattr(self, "refresh_interval"):
0021             self.refresh_interval = 60
0022
0023     # check token key
0024     def check_credential(self):
0025         # make logger
0026         main_log = self.make_logger(_logger, method_name="check_credential")
0027         main_log.debug("start")
0028         try:
0029             # check if the key file exists
0030             if not os.path.exists(self.token_key_file):
0031                 main_log.debug(f"token key file {self.token_key_file} not found")
0032                 return False
0033             # check if the key file is fresh
0034             mod_time = datetime.datetime.fromtimestamp(os.stat(self.token_key_file).st_mtime, datetime.timezone.utc)
0035             if datetime.datetime.now(datetime.timezone.utc) - mod_time < datetime.timedelta(minutes=self.refresh_interval):
0036                 main_log.debug(f"token key {self.token_key_file} is fresh")
0037                 return True
0038             main_log.debug(f"token key {self.token_key_file} is outdated")
0039         except Exception:
0040             core_utils.dump_error_message(main_log)
0041         return False
0042
0043     # renew proxy
0044     def renew_credential(self):
0045         # make logger
0046         main_log = self.make_logger(_logger, method_name="renew_credential")
0047         main_log.debug("start")
0048         # make communication channel to PanDA
0049         com = CommunicatorPool()
0050         token_key, msg = com.get_token_key(self.client_name)
0051         if token_key:
0052             is_ok = True
0053             with open(self.token_key_file, "w") as f:
0054                 f.write(token_key)
0055             main_log.debug(f"got new token key in {self.token_key_file}")
0056         else:
0057             is_ok = False
0058             main_log.error(f"failed to renew token key with a server message : {msg}")
0059         return is_ok, msg