Script to rewrite an Ansible playbook or tasks file to use plugin FQCN

rewrite_fqcn.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# (c) 2020 Matt Martz <[email protected]>
# GNU General Public License v3.0+
#     (see https://www.gnu.org/licenses/gpl-3.0.txt)

import argparse
import io
import os
import re
import sys

import ansible.config

from ansible.errors import AnsibleParserError
from ansible.inventory.manager import InventoryManager
from ansible.parsing.dataloader import DataLoader
from ansible.playbook.task import Task
from ansible.vars.manager import VariableManager

try:
    import ruamel.yaml
    import ruamel.yaml.util
except ImportError:
    raise SystemExit('ruamel.yaml is required')

MISSING_MODULE_RE = re.compile(r"couldn't resolve module/action '([^']+)'\.")
MISSING_LOOKUP_RE = re.compile(
    r"'([^']+)' is not a valid attribute for a Task"
)
FILTER_RE = re.compile(r'((.+?)\s*([\w \.\'"]+)(\s*)\|(\s*)(\w+))')
LOOKUP_RE = re.compile(r'((q|query|lookup)\((\s*[\'"])([^\'"]+))')
WITH_RE = re.compile(r'((\s+with_)([^ :]+):)')

yaml = ruamel.yaml.YAML()
yaml.preserve_quotes = True

# Migrated lookups used in with_ would cause an
# AnsibleParserError without this
Task._validate_attributes = lambda *args: None


@ruamel.yaml.yaml_object(yaml)
class Vault:
    yaml_tag = u'!vault'

    def __init__(self, value, style, anchor):
        self.value = value
        self.style = style
        self.anchor = anchor

    @classmethod
    def to_yaml(cls, representer, node):
        return representer.represent_scalar(
            cls.yaml_tag,
            u'{.value}'.format(node),
            style=node.style,
            anchor=node.anchor
        )

    @classmethod
    def from_yaml(cls, constructor, node):
        return cls(node.value, node.style, node.anchor)


@ruamel.yaml.yaml_object(yaml)
class Unsafe:
    yaml_tag = u'!unsafe'

    def __init__(self, value, style, anchor):
        self.value = value
        self.style = style
        self.anchor = anchor

    @classmethod
    def to_yaml(cls, representer, node):
        return representer.represent_scalar(
            cls.yaml_tag,
            u'{.value}'.format(node),
            style=node.style,
            anchor=node.anchor
        )

    @classmethod
    def from_yaml(cls, constructor, node):
        return cls(node.value, node.style, node.anchor)


def get_tasks(obj):
    if isinstance(obj, list):
        for o in obj:
            yield from get_tasks(o)
    else:
        found = False
        for key in ('tasks', 'pre_tasks', 'post_tasks', 'handlers',
                    'block', 'rescue', 'always'):
            if obj.get(key):
                found = True
                yield from get_tasks(obj.get(key))
        if not found:
            yield obj


parser = argparse.ArgumentParser()
parser.add_argument('file', help='Playbook or task file')
args = parser.parse_args()


loader = DataLoader()
inventory = InventoryManager(loader=loader, sources='localhost,')
variable_manager = VariableManager(loader=loader, inventory=inventory)


routing_yml = os.path.join(
    os.path.dirname(ansible.config.__file__),
    'ansible_builtin_runtime.yml'
)
try:
    with open(routing_yml) as f:
        routing = yaml.load(f)['plugin_routing']
except FileNotFoundError:
    raise SystemExit(
        'This script can only operate with ansible-base installed'
    )

with open(args.file) as f:
    data = yaml.load(f)

for ds in get_tasks(data):
    try:
        task = Task.load(ds, variable_manager=variable_manager, loader=loader)
    except AnsibleParserError as e:
        match = MISSING_MODULE_RE.search(str(e)).group(1)
        action = routing['action'].get(match, {}).get('redirect')
        module = routing['modules'].get(match, {}).get('redirect')
        if action or module:
            value = ds[match]
            del ds[match]
            ds[action or module] = value
    except Exception as e:
        print('WARNING: %s' % e, file=sys.stderr)

with io.StringIO() as f:
    yaml.dump(data, f)
    new = f.getvalue()

for match in FILTER_RE.finditer(new):
    redirect = routing['filter'].get(match.group(6), {}).get('redirect')
    if redirect:
        new = re.sub(
            re.escape(match.group(1)),
            '%s%s|%s%s%s' % (match.groups()[1:-1] + (redirect,)),
            new
        )

for match in LOOKUP_RE.finditer(new):
    redirect = routing['lookup'].get(match.group(4), {}).get('redirect')
    if redirect:
        new = re.sub(
            re.escape(match.group(1)),
            '%s(%s%s' % (match.groups()[1:-1] + (redirect,)),
            new
        )

for match in WITH_RE.finditer(new):
    redirect = routing['lookup'].get(match.group(3), {}).get('redirect')
    if redirect:
        new = re.sub(
            re.escape(match.group(1)),
            '%s%s:' % (match.groups()[1:-1] + (redirect,)),
            new
        )

print(new.rstrip())

I added this before line 85:

        if 'block' in obj:
            found = True
            yield from get_tasks(obj['block'])
            for key in ('always', 'rescue'):
                if obj.get(key):
                    yield from get_tasks(obj.get(key))

Also nicer with:

yaml.preserve_quotes = True

The string in line 116 needs to be changed to ansible_builtin_runtime.yml.