Last active
May 2, 2018 20:08
-
-
Save confiks/6d7b0ee3e137df01b758 to your computer and use it in GitHub Desktop.
A simple ansible action plugin to ask the user to input a key, in the middle of a role
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# Not extensively tested | |
# Put this script in the action_plugins directory of your playbook directory | |
# If you have issues, please report it in the comments (or fork and fix) | |
# Usage: | |
# - name: "Ask the user if we should continue." | |
# action: ask_key prompt="Continue? Yes / No / Random (y/n/r)?" accepted_keys="['y', 'n', 'r']" | |
# register: answer | |
# | |
# The pressed key is now in answer.key | |
from __future__ import (absolute_import, division, print_function) | |
__metaclass__ = type | |
import termios | |
import tty | |
from os import isatty | |
from ansible.plugins.action import ActionBase | |
try: | |
from __main__ import display | |
except ImportError: | |
from ansible.utils.display import Display | |
display = Display() | |
class ActionModule(ActionBase): | |
''' Ask the user to input a key ''' | |
VALID_ARGS = set(["prompt", "accepted_keys"]) | |
REQUIRED_ARGS = set(["prompt", "accepted_keys"]) | |
BYPASS_HOST_LOOP = True | |
def run(self, tmp=None, task_vars=None): | |
if task_vars is None: | |
task_vars = dict() | |
for arg in self._task.args: | |
if arg not in self.VALID_ARGS: | |
return {"failed": True, "msg": "'%s' is not a valid option in ask" % arg} | |
for arg in self.REQUIRED_ARGS: | |
if arg not in self._task.args: | |
return {"failed": True, "msg": "'%s' is required in ask" % arg} | |
result = super(ActionModule, self).run(tmp, task_vars) | |
fd = None | |
old_settings = None | |
try: | |
fd = self._connection._new_stdin.fileno() | |
except ValueError: | |
pass | |
if fd is not None: | |
if isatty(fd): | |
old_settings = termios.tcgetattr(fd) | |
tty.setraw(fd) | |
termios.tcflush(self._connection._new_stdin, termios.TCIFLUSH) | |
else: | |
return {"failed": True, "msg": "For some reason, we couldn't access the connection tty."} | |
display.display(self._task.args["prompt"] + "\r") | |
while True: | |
key = self._connection._new_stdin.read(1) | |
if key == '\x03': | |
result["failed"] = True | |
result["msg"] = "User requested to cancel." | |
break | |
elif key in self._task.args["accepted_keys"]: | |
result["key"] = key | |
break | |
if old_settings != None and isatty(fd): | |
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) | |
return result |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
I have thought of a couple of features that I would like to see:
I think that these would be incredibly useful - however my Python is not (currently!) up to it.... :(
Any chance you might consider adding?