Timing/race condition with API Token refresh

[omni1504]

Hi all,

We are using FMCAPI to migrate from Checkpoint firewall, which involved bulk addition of objects to a given rule. The nature of FMCAPI is that when you "add" a destination/source network, it iterates through all FMC objects via /networkaddresses , which is quite expensive for a busy FMC with a lot of objects (but is fine). More severe issue is with token refresh -I've included a code to get token refreshed before I call something via FMCAPI , but in case of "add" operation, there are multiple GETs which you cannot control (paginated retrieval), so you get 401 errors in the middle of pagination.

Proposed fix which I have implemented on my local copy - adding a check for 401 error and refreshing token/repeating afterwards:

def send_to_api(self, method='', url='', headers='', json_data=None, more_items=[]):
---snip---
try:
            while status_code == 429:
                if method == 'get':
                    response = requests.get(url, headers=headers, verify=self.VERIFY_CERT)
                elif method == 'post':
                    response = requests.post(url, json=json_data, headers=headers, verify=self.VERIFY_CERT)
                elif method == 'put':
                    response = requests.put(url, json=json_data, headers=headers, verify=self.VERIFY_CERT)
                elif method == 'delete':
                    response = requests.delete(url, headers=headers, verify=self.VERIFY_CERT)
                else:
                    logging.error("No request method given.  Returning nothing.")
                    return
                status_code = response.status_code

                if status_code == 429:
                    logging.warning("Too many connections to the FMC.  Waiting 30 seconds and trying again.")
                    time.sleep(30)
                if status_code == 401:
                    logging.warning("Token has expired. Trying to refresh.")
                    self.mytoken.access_token=self.mytoken.get_token()
                    headers = {'Content-Type': 'application/json', 'X-auth-access-token': self.mytoken.access_token}
                    status_code = 429
            json_response = json.loads(response.text)
            if status_code > 301 or 'error' in json_response:
                response.raise_for_status()

Regards, Amir

[daxm]

Thanks for the code snippet! I've integrated it into the published pip package. Please update your pacakage from pip and let me know if it works. (I can't test it as I don't have an FMC anymore to test against.)

[omni1504]

Hi Dax,

Works well me for on FMC 6.3.0 - tested on a large real rule-set and I can see token properly refreshing. I would consider this issue closed, thanks for quick reply.

Amir