This blog post aims to assist Ansible users in providing documentation to security teams when using privileged escalation (become) in Ansible playbooks.
Ansible does not use a specific command when using privileged escalation on remote machines see here for more details. As an example, if you wanted to restart the httpd service, the command you run on the system would be sudo systemctl restart httpd but when Ansible executes the task, you would see something like /bin/sh BECOME-SUCCESS ; /usr/bin/python /home/myuser/.ansible/tmp/ansible-tmp-abcdef/AnsiballZ_12345.py. Here is a link explaining the Ansiballz framework. This presents two concerns to most security teams. First, they cannot limit the commands used by Ansible. Second, they cannot see which commands Ansible executes. We will be addressing the second concern.
Ansible provides Callback Plugins that can be used when Ansible responds to various events inside a playbook. Ansible provides a small set of default callback plugins out of the box, but none that address our problem.
We will be creating a custom callback plugin that hooks into the use of become and logs to stdout. This will not log the actual command used by Ansible but the module and arguments passed to the module. As an example, instead of systemctl restart httpd the log would show the ansible.builtin.service module with the name set to httpd and the state set to restarted.
First we need to build out our directory structure. We are choosing the name “become” for our plugin, but you could choose your own name. By convention, the filename and plugin name should match, but they do not have to. This assumes the callback plugin will be added to an existing Ansible project:
├── ansible.cfg // Ansible Configuration file.
├── example_pb.yaml // Ansible playbook.
├── callback_plugins
│ └── become.py // Our Become Callback Plugin.
mkdir callback_plugins
touch callback_plugins/become.py
Then add the callback plugin to your ansible.cfg file.
# ansible.cfg
[defaults]
callback_whitelist = become
From there we will create a sample playbook that restarts httpd:
# example_pb.yaml
---
- hosts: all
tasks:
- name: "ESCALATED TASK: RESTART HTTPD"
ansible.builtin.service:
name: httpd
state: restarted
become: true
- name: "NON-ESCALATED TASK: DEBUG"
ansible.builtin.debug:
msg: "This task does not require escalation"
Next we will add the basic structure to our callback plugin. Ansible provides an example plugin that we will modify.
# become.py
# Make coding more python3-ish, this is required for contributions to Ansible
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
# not only visible to ansible-doc, it also 'declares' the options the plugin requires and how to configure them.
DOCUMENTATION = '''
callback: become
requirements:
- whitelist in configuration
short_description: When become is used, log to stdout.
description:
- This callback will log the become, become method, task name, action and arguments to stdout.
'''
from ansible.plugins.callback import CallbackBase
class CallbackModule(CallbackBase):
"""
This callback module tells you when become is used.
"""
CALLBACK_VERSION = 2.0
CALLBACK_NAME = 'become'
# only needed if you ship it and don't want to enable by default
CALLBACK_NEEDS_WHITELIST = True
def __init__(self):
# make sure the expected objects are present, calling the base's __init__
super(CallbackModule, self).__init__()
We have now added the skeleton that Ansible requires, but we have not added any new functionality to our plugin. Callback plugins listen for events that Ansible emits. Check out the callback init file in the Ansible repository for a list of all the events supported. The event that we will be looking at hooking into is the v2_runner_on_start, since it passes not only the task but also the host, and it executes at the start of a task. Let’s add our first callback now.
# become.py
class CallbackModule(CallbackBase):
...
def v2_runner_on_start(self, host, task):
"""
Process the runner on start event.
"""
if task.become:
print(f"Hostname: {host.name}")
print(f"Task: {task.name}")
print(f"Action: {task.action}")
print(f"Arguments: {task.args}")
We are accessing attributes of task and host here. In this example we are showing just a few, but you can use dir(task) or dir(host) to find many others. When we execute the playbook, we should see the following:
TASK [Gathering Facts]****************************
ok: [my_hostname]
TASK [ESCALATED TASK: RESTART HTTPD]**************
Hostname: my_hostname
Task: ESCALATED TASK: RESTART HTTPD
Action: ansible.builtin.service
Arguments: {'name': 'httpd', 'state': 'restarted'}
changed: [my_hostname]
TASK [NON-ESCALATED TASK: DEBUG]
ok: [my_hostname] => {
"msg": "This task does not require escalation"
}
At this point, We have a callback plugin that will log to stdout the use of become in our playbooks. But if our playbook uses variables, we will not see those variables. Let’s modify our example to show:
# example_pb.yaml
---
- hosts: all
tasks:
- name: "ESCALATED TASK: RESTART HTTPD"
ansible.builtin.service:
name: "{{ service }}"
state: restarted
become: true
vars:
service: httpd
- name: "NON-ESCALATED TASK: DEBUG"
ansible.builtin.debug:
msg: "This task does not require escalation"
Now let’s run the playbook:
TASK [Gathering Facts]****************************
ok: [my_hostname]
TASK [ESCALATED TASK: RESTART HTTPD]**************
Hostname: my_hostname
Task: ESCALATED TASK: RESTART HTTPD
Action: ansible.builtin.service
Arguments: {'name': '{{ service }}', 'state': 'restarted'}
changed: [my_hostname]
TASK [NON-ESCALATED TASK: DEBUG]
ok: [my_hostname] => {
"msg": "This task does not require escalation"
}
In order to access variables passed into a task (or from group vars), we need to add the Templar class to our callback plugin. We will need to add a few more hooks to get all of the play data and the hostvars.
# become.py
from ansible.plugins.callback import CallbackBase
from ansible.template import Templar
class CallbackModule(CallbackBase):
...
def v2_playbook_on_start(self, playbook):
"""
Initialize self.playbook.
"""
self.playbook = playbook
def v2_playbook_on_play_start(self, play):
"""
Get the variable manager of the current play from the playbook.
"""
self.play = play
self.vm = play.get_variable_manager()
def _all_vars(self, host=None, task=None):
"""
Load all variables for the given inputs.
"""
return self.vm.get_vars(
play = self.play,
host = host,
task = task
)
def v2_runner_on_start(self, host, task):
"""
Process the runner on start event.
"""
templar = Templar(loader=self.playbook.get_loader(),
variables=self._vars(host=host, task=task))
if task.become:
print(f"Hostname: {host.name}")
print(f"Task: {task.name}")
print(f"Action: {task.action}")
print(f"Arguments: {templar.template(task.args, fail_on_undefined=False)}")
With the above changes, running the playbook will now fill in the variables:
TASK [Gathering Facts]****************************
ok: [my_hostname]
TASK [ESCALATED TASK: RESTART HTTPD]**************
Hostname: my_hostname
Task: ESCALATED TASK: RESTART HTTPD
Action: ansible.builtin.service
Arguments: {'name': 'httpd', 'state': 'restarted'}
changed: [my_hostname]
TASK [NON-ESCALATED TASK: DEBUG]
ok: [my_hostname] => {
"msg": "This task does not require escalation"
}
Now that we know how to access the task attributes and variables, we could extend any of the other popular callback plugins, such as syslog_json, Logstash, or Splunk, to send the data directly to the security team.
There are a few Ansible playbook options that are not covered here, such as loop or with_items, become_method, or become_user, but this should be enough to get you started. Another suggested change would be to set CALLBACK_NEEDS_WHITELIST = False. With this change, we would no longer need to whitelist become in the ansible.cfg file. This, paired with Ansible Tower or AWX, would allow the security team to ensure conformance to standards. For this to work, the plugin would need to be moved to the environment created by Ansible Tower or AWX. See Adding a plugin locally for more details.
Ansible callback plugins give us the ability to provide detailed evidence of our privileged escalation to answer some of the security team’s concerns.
Thanks for taking the time to read!
-Stephen
Here is the full code from become.py:
# become.py
# Make coding more python3-ish, this is required for contributions to Ansible
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
# not only visible to ansible-doc, it also 'declares' the options the plugin requires and how to configure them.
DOCUMENTATION = '''
callback: become
requirements:
- whitelist in configuration
short_description: Logs become use to stdout
description:
- This callback will log the become, become method, task name, action and arguments to stdout.
'''
from ansible.plugins.callback import CallbackBase
from ansible.template import Templar
class CallbackModule(CallbackBase):
"""
This callback module tells you when become is used.
"""
CALLBACK_VERSION = 2.0
CALLBACK_NAME = 'become'
# only needed if you ship it and don't want to enable by default
CALLBACK_NEEDS_WHITELIST = True
def __init__(self):
# make sure the expected objects are present, calling the base's __init__
super(CallbackModule, self).__init__()
def v2_playbook_on_start(self, playbook):
"""
Initialize self.playbook.
"""
self.playbook = playbook
def v2_playbook_on_play_start(self, play):
"""
Get the variable manager of the current play from the playbook.
"""
self.play = play
self.vm = play.get_variable_manager()
def _all_vars(self, host=None, task=None):
"""
Load all variables for the given inputs.
"""
return self.vm.get_vars(
play = self.play,
host = host,
task = task
)
def v2_runner_on_start(self, host, task):
"""
Process the runner on start event.
"""
templar = Templar(loader=self.playbook.get_loader(),
variables=self._vars(host=host, task=task))
if task.become:
print(f"Hostname: {host.name}")
print(f"Task: {task.name}")
print(f"Action: {task.action}")
print(f"Arguments: {templar.template(task.args, fail_on_undefined=False)}")
Does this all sound amazing? Want to know more about how Network to Code can help you do this, reach out to our sales team. If you want to help make this a reality for our clients, check out our careers page.
Share details about yourself & someone from our team will reach out to you ASAP!
