Repository: ansible/awx
File: awx/playbooks/action_plugins/insights.py (Python)

Code before:
        return result

    def run(self, tmp=None, task_vars=None):
        self._supports_check_mode = False

        result = super(ActionModule, self).run(tmp, task_vars)

        insights_url = self._task.args.get('insights_url', None)
        proj_path = self._task.args.get('project_path', None)
        license = self._task.args.get('awx_license_type', None)
        awx_version = self._task.args.get('awx_version', None)
        authentication = self._task.args.get('authentication', None)
        username = self._task.args.get('username', None)
        password = self._task.args.get('password', None)
        client_id = self._task.args.get('client_id', None)
        client_secret = self._task.args.get('client_secret', None)
        oidc_endpoint = self._task.args.get('oidc_endpoint', None)

        headers = {
            'Content-Type': 'application/json',
            'User-Agent': '{} {} ({})'.format('AWX' if license == 'open' else 'Red Hat Ansible Automation Platform', awx_version, license),
        }

        session = requests.Session()
        if authentication == 'service_account' or (client_id and client_secret):
            self._obtain_auth_token(oidc_endpoint, client_id, client_secret, result)
            if 'token' not in result:
                return result
            headers['Authorization'] = f'{result['token_type']} {result['token']}'
        elif authentication == 'basic' or (username and password):
            session.auth = requests.auth.HTTPBasicAuth(username, password)

        url = '/api/remediations/v1/remediations'
        while url:
            res = session.get('{}{}'.format(insights_url, url), headers=headers, timeout=120)

            if res.status_code != 200:
                result['failed'] = True
                result['msg'] = 'Expected {} to return a status code of 200 but returned status code "{}" instead with content "{}".'.format(
                    url, res.status_code, res.content
                )
                return result

            # FIXME: ETags are (maybe?) not yet supported in the new
            # API, and even if they are we'll need to put some thought
            # into how to deal with them in combination with pagination.
            if 'ETag' in res.headers:
                version = res.headers['ETag']
                if version.startswith('"') and version.endswith('"'):
                    version = version[1:-1]
            else:

Review comment (security):
I don't love passing the dict `result` into `_obtain_auth_token()`. Instead, can `_obtain_auth_token` return a tuple `(token_type, token)` or `None`.

Right now, with `result` passed in and being returned from the module, I'm thinking the `result['token']` might can end up in the Ansible output long with `msg`. We wouldn't want that behavior.

Code after:
            data['token_type'] = response.json().get('token_type', None)
        return data

    def run(self, tmp=None, task_vars=None):
        self._supports_check_mode = False

        session = requests.Session()
        result = super(ActionModule, self).run(tmp, task_vars)

        insights_url = self._task.args.get('insights_url', None)
        proj_path = self._task.args.get('project_path', None)
        license = self._task.args.get('awx_license_type', None)
        awx_version = self._task.args.get('awx_version', None)
        authentication = self._task.args.get('authentication', None)
        username = self._task.args.get('username', None)
        password = self._task.args.get('password', None)
        client_id = self._task.args.get('client_id', None)
        client_secret = self._task.args.get('client_secret', None)
        oidc_endpoint = self._task.args.get('oidc_endpoint', None)

        session.headers.update(
            {
                'Content-Type': 'application/json',
                'User-Agent': '{} {} ({})'.format('AWX' if license == 'open' else 'Red Hat Ansible Automation Platform', awx_version, license),
            }
        )

        if authentication == 'service_account' or (client_id and client_secret):
            data = self._obtain_auth_token(oidc_endpoint, client_id, client_secret)
            if 'token' not in data:
                result['failed'] = data['failed']
                result['msg'] = data['msg']
                return result
            session.headers.update({'Authorization': f'{result['token_type']} {result['token']}'})
        elif authentication == 'basic' or (username and password):
            session.auth = requests.auth.HTTPBasicAuth(username, password)

        url = '/api/remediations/v1/remediations'
        while url:
            res = session.get('{}{}'.format(insights_url, url), timeout=120)

            if res.status_code != 200:
                result['failed'] = True
                result['msg'] = 'Expected {} to return a status code of 200 but returned status code "{}" instead with content "{}".'.format(
                    url, res.status_code, res.content
                )
                return result

            # FIXME: ETags are (maybe?) not yet supported in the new
            # API, and even if they are we'll need to put some thought
            # into how to deal with them in combination with pagination.