Pythonで短縮系組み込み関数入りCloudFormationのYAMLを解析

こんにちは、SWX3人目の熊谷（悠）です。
CloudFormation（以下CFn）に使用するテンプレートファイルをプログラムで読み込んで色々するツールを作りたくなる事は往々にしてあると思います。
また、CFnには必須とも言える便利な組み込み関数が用意されています。

問題

通常、PythonでYAMLを読み込む場合はPyYAMLやruamel.yaml等を用いて解析しますが、
!RefなどのCFn独自の組み込み関数の短縮形を表す感嘆符("!")から始まる文字列は、タグ¹としてYAMLで定義されているため解析時に以下のようなエラーが発生してしまいます。

yaml.constructor.ConstructorError: could not determine a constructor for the tag '!Ref'

解決策

aws-cliパッケージに、この問題を解決できる関数が実装されています。
aws-cli/yamlhelper.py at develop · aws/aws-cli
※AWS CLIのaws cloudformation validate-templateコマンド²にて使用されています。

CFnにて正しく読み取れる形のYAML文字列を渡すと、順序付き辞書型（collections.OrderedDict）に変換してくれます。

def yaml_parse(yamlstr):
    """Parse a yaml string"""
    try:
        # PyYAML doesn't support json as well as it should, so if the input
        # is actually just json it is better to parse it with the standard
        # json parser.
        return json.loads(yamlstr, object_pairs_hook=OrderedDict)
    except ValueError:
        loader = SafeLoaderWrapper
        loader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, 
                               _dict_constructor)
        loader.add_multi_constructor("!", intrinsics_multi_constructor)
        return yaml.load(yamlstr, loader)

変換時には短縮形のCFn組み込み関数が完全な関数名の構文に置き換えられます。
※感嘆符!をFn ::に変換したりGetAttを標準的な配列へ変換したりなど
※コメントにあるようにRefやConditionは完全な関数名にFn ::が付かないので、感嘆符!のみ取り除かれます。
このような変換処理をローダーに読ませて、PyYAMLで改めてロードしています。

    # Some intrinsic functions doesn't support prefix "Fn::"
    prefix = "Fn::"
    if tag in ["Ref", "Condition"]:
        prefix = ""

    if tag == "GetAtt" and isinstance(node.value, six.string_types):
        # ShortHand notation for !GetAtt accepts Resource.Attribute format
        # while the standard notation is to use an array
        # [Resource, Attribute]. Convert shorthand to standard format
        value = node.value.split(".", 1)

実装

環境

$ python -V
Python 3.8.5
$ aws --version
aws-cli/1.19.57 Python/3.8.5 Linux/4.14.209-160.335.amzn2.x86_64 botocore/1.20.57

pipenvで管理していますが、awscliのバージョンは1.19.2です。

        "awscli": {
            "hashes": [
                "sha256:7ca82e21bba8e1c08fef5f8c2161e1a390ddc19da69214eca8db249328ebd204",
                "sha256:8b79284e7fc018708afe2ad18ace37abb6921352cd079c0be6d15eabeabe5169"
            ],
            "index": "pypi",
            "version": "==1.19.2"
        },

例

import yaml

from awscli.customizations.cloudformation.yamlhelper import yaml_parse

if __name__ == '__main__':
    try:
        yaml_str = open('cfn-template.yaml').read()
        print(f'yaml_str：{yaml_str}')
        
        # YAML文字列を解析し、順序付き辞書型のオブジェクトを返却する
        yaml_dict = yaml_parse(yaml_str)
        print(f'yaml_dict：{yaml_dict}')
    except yaml.parser.ParserError as e:
        print(e)
        print('YAML形式として解析できない文字列です。（例：キーや:が無い）')
    except yaml.scanner.ScannerError as e:
        print(e)
        print('YAML形式として読み取れない値が含まれています。（例：CFnの組み込み関数の構文誤り）')

Resources:
    ExampleVpc:
        Type: AWS::EC2::VPC
        Properties:
            CidrBlock: "10.0.0.0/16"
    IPv6CidrBlock:
        Type: AWS::EC2::VPCCidrBlock
        Properties:
            AmazonProvidedIpv6CidrBlock: true
            VpcId: !Ref ExampleVpc
    ExampleSubnet:
        Type: AWS::EC2::Subnet
        DependsOn: IPv6CidrBlock
        Properties:
            AssignIpv6AddressOnCreation: true
            CidrBlock: !Select [ 0, !Cidr [ !GetAtt ExampleVpc.CidrBlock, 1, 8 ]]
            Ipv6CidrBlock: !Select [ 0, !Cidr [ !Select [ 0, !GetAtt ExampleVpc.Ipv6CidrBlocks], 1, 64 ]]
            VpcId: !Ref ExampleVpc

OrderedDict([('Resources', 
    OrderedDict([
        ('ExampleVpc', 
            OrderedDict([
                ('Type', 'AWS::EC2::VPC'), 
                ('Properties', 
                    OrderedDict([
                        ('CidrBlock', '10.0.0.0/16')
                    ])
                )
            ])
        ), 
        ('IPv6CidrBlock', 
            OrderedDict([
                ('Type', 'AWS::EC2::VPCCidrBlock'), 
                ('Properties', 
                    OrderedDict([
                        ('AmazonProvidedIpv6CidrBlock', True), 
                        ('VpcId', 
                            {'Ref': 'ExampleVpc'})
                    ])
                )
            ])
        ), 
        ('ExampleSubnet', 
            OrderedDict([
                ('Type', 'AWS::EC2::Subnet'), 
                ('DependsOn', 'IPv6CidrBlock'), 
                ('Properties', 
                    OrderedDict([
                        ('AssignIpv6AddressOnCreation', True), 
                        ('CidrBlock', {
                            'Fn::Select': [
                                0, 
                                {'Fn::Cidr': [
                                    {'Fn::GetAtt': [
                                        'ExampleVpc', 'CidrBlock'
                                        ]
                                    }, 
                                    1, 
                                    8
                                ]}
                            ]
                        }), 
                        ('Ipv6CidrBlock', {
                            'Fn::Select': [
                                0, 
                                {'Fn::Cidr': [
                                    {'Fn::Select': [
                                        0, 
                                        {'Fn::GetAtt': [
                                            'ExampleVpc', 'Ipv6CidrBlocks'
                                            ]
                                        }
                                    ]}, 
                                    1, 
                                    64
                                ]}
                            ]
                        }), 
                        ('VpcId', {
                            'Ref': 'ExampleVpc'
                        })
                    ])
                )
            ])
        )
    ])
)])