def _handle_remote_access()

in azurelinuxagent/ga/remoteaccess.py [0:0]

• 37 lines of code
• 18 McCabe index (conditional complexity)

    def _handle_remote_access(self):
        if self._remote_access is not None:
            logger.info("Processing remote access users in goal state.")

            self._check_existing_jit_users = True

            existing_jit_users = self._get_existing_jit_users()
            goal_state_users = set(u.name for u in self._remote_access.user_list.users)

            for acc in self._remote_access.user_list.users:
                try:
                    raw_expiration = acc.expiration
                    account_expiration = datetime.strptime(raw_expiration, REMOTE_USR_EXPIRATION_FORMAT)
                    now = datetime.utcnow()
                    if acc.name not in existing_jit_users and now < account_expiration:
                        self._add_user(acc.name, acc.encrypted_password, account_expiration)
                    elif acc.name in existing_jit_users and now > account_expiration:
                        # user account expired, delete it.
                        logger.info("Remote access user '{0}' expired.", acc.name)
                        self._remove_user(acc.name)
                except Exception as e:
                    logger.error("Error processing remote access user '{0}' - {1}", acc.name, ustr(e))

            for user in existing_jit_users:
                try:
                    if user not in goal_state_users:
                        # user explicitly removed
                        self._remove_user(user)
                except Exception as e:
                    logger.error("Error removing remote access user '{0}' - {1}", user, ustr(e))
        else:
            # There are no JIT users in the goal state; that may mean that they were removed or that they
            # were never added. Enumerating the users on the current vm can be very slow and this path is hit
            # on each goal state; we use self._check_existing_jit_users to avoid enumerating the users
            # every single time.
            if self._check_existing_jit_users:
                logger.info("Looking for existing remote access users.")

                existing_jit_users = self._get_existing_jit_users()

                remove_user_errors = False

                for user in existing_jit_users:
                    try:
                        self._remove_user(user)
                    except Exception as e:
                        logger.error("Error removing remote access user '{0}' - {1}", user, ustr(e))
                        remove_user_errors = True

                if not remove_user_errors:
                    self._check_existing_jit_users = False