bundlewrap/bundles/iptables/metadata.py

from bundlewrap.exceptions import BundleError

@metadata_reactor.provides(
    'iptables/bundle_rules/iptables',
)
def port_rules_to_iptables(metadata):
    # Using this, bundles can simply set up port based rules. This
    # reactor will then take care of converting those rules to actual
    # iptables rules
    ruleset = set()

    # Plese note we do not set any defaults for ports. Bundles are
    # expected to know themselves which default to use.
    for portdef, targets in metadata.get('iptables/port_rules', {}).items():
        if '/' in portdef:
            port, proto = portdef.split('/', 2)

            if proto not in {'udp'}:
                raise BundleError(f'iptables/port_rules: illegal identifier {portdef} in metadata for {node.name}')
        else:
            port = portdef
            proto = 'tcp'

        for target in targets:
            if port == '*' and target == '*':
                raise BundleError('iptables/port_rules: setting both port and target to * is unsupported')

            comment = f'-m comment --comment "iptables port_rules {target}"'

            if port != '*':
                port_str = f'--dport {port}'
            else:
                port_str = ''

            if target == '*':
                ruleset.add(f'iptables_both -A INPUT -p {proto} {port_str} {comment} -j ACCEPT')
            else:
                resolved = repo.libs.tools.resolve_identifier(repo, target)

                for address in resolved['ipv4']:
                    ruleset.add(f'iptables -A INPUT -p {proto} -s {address} {port_str} {comment} -j ACCEPT')

                for address in resolved['ipv6']:
                    ruleset.add(f'ip6tables -A INPUT -p {proto} -s {address} {port_str} {comment} -j ACCEPT')

    return {
        'iptables': {
            'bundle_rules': {
                # order does not matter here.
                'iptables': list(sorted(ruleset)),
            },
        },
    }