check

#!/usr/bin/env python3

import sys
import ipaddress
import math
from optparse import OptionParser
from math import floor
from filereader import get_communities_data

import requests

ICVPN_NETWORKS = {
    "IPv4": ipaddress.IPv4Network("10.207.0.0/16"),
    "IPv6": ipaddress.IPv6Network("fec0::a:cf:0:0/96"),
}

ANSI_COLOR_ERR = "\x1b[31m"
ANSI_COLOR_WARN = "\x1b[33m"
ANSI_COLOR_RESET = "\x1b[0m"
TLD_LIST = 'http://data.iana.org/TLD/tlds-alpha-by-domain.txt'


def error(*arg):
    print(ANSI_COLOR_ERR, *arg, file=sys.stderr,
          end='%s\n' % ANSI_COLOR_RESET)


def warn(*arg):
    print(ANSI_COLOR_WARN, *arg, file=sys.stderr,
          end='%s\n' % ANSI_COLOR_RESET)


def ip_family_address(family, address):
    obj = getattr(ipaddress, family + "Address")
    return obj(address)


def ip_family_network(family, network):
    obj = getattr(ipaddress, family + "Network")
    return obj(network)


def get_tlds():
    domains = dict()

    try:
        response = requests.get(TLD_LIST, timeout=1.0)
    except requests.exceptions.RequestException:
        pass
    else:
        for line in response.text.split('\n'):
            if line.startswith('#'):
                continue
            domains[line.rstrip().upper()] = "tld(IANA)"

    return domains


def check_dupe(name, v, d, community):
    if v in d:
        error("Duplicate %s (%s):" % (name, v), d[v], community)
        return 1
    else:
        d[v] = community
        return 0


def check_net(family, net, nets, community):
    errcnt = 0

    try:
        net = ip_family_network(family, net)
    except ValueError:
        errcnt += 1
        error("Not an %s network: %s (%s)" % (family, net, community))
    else:
        for other in nets:
            if other.overlaps(net):
                errcnt += 1
                error("%s Network overlap: %s (%s), %s (%s)" %
                      (family, community, net, nets[other], other))

        if errcnt == 0:
            nets[net] = community

    return errcnt


def check_rdns(data):
    errcnt = 0

    allowed_domains = []

    # find allowed rdns-domain-names
    if 'networks' in data:
        if 'ipv6' in data['networks']:
            for net_str in data['networks']['ipv6']:
                try:
                    net = ip_family_network('IPv6', net_str)
                except ValueError:
                    continue

                net_length = math.floor(net.prefixlen / 4)  # 4 bits per character
                # by rounding down instead of correcly handling partially used characters
                # we generate false-passes but no false-warnings for prefixlength % 4 != 0

                net_prefix = net \
                    .network_address \
                    .exploded \
                    .replace(":", "")[:net_length]

                net_rdns = '.'.join(reversed(net_prefix))
                net_rdns = '{}.ip6.arpa'.format(net_rdns)

                allowed_domains.append(net_rdns)

        if 'ipv4' in data['networks']:
            for net_str in data['networks']['ipv4']:
                try:
                    net = ip_family_network('IPv4', net_str)
                except ValueError:
                    continue

                net_length = math.floor(net.prefixlen / 8)  # 8 bits per block
                # by rounding down instead of correcly handling partially used blocks
                # we generate false-passes but no false-warnings for prefixlength % 8 != 0

                net_prefix = net \
                    .network_address \
                    .exploded \
                    .split(".")[:net_length]

                net_rdns = '.'.join(reversed(net_prefix))
                net_rdns = '{}.in-addr.arpa'.format(net_rdns)

                allowed_domains.append(net_rdns)

    if 'domains' in data:
        for domain in data['domains']:
            if '_' in domain:
                errcnt += 1
                error("Domain %s has illegal characters" % domain)
            if domain.endswith('.ip6.arpa') or domain.endswith('.in-addr.arpa'):
                found = False
                for reverse_domain in allowed_domains:
                    if domain.endswith(reverse_domain):  # we want communitys to be able to announce only a part of the net as RDNS-capable
                        found = True

                if not found:
                    errcnt += 1
                    error("Illegal RDNS-Domain: found no matching net for %s (found nets: %s)" % (domain, allowed_domains))

    return errcnt


def check_delegation(nets, asns, dnet, dasn):
    errcnt = 0

    # merge ipv4 / ipv6 nets, flatten the list
    nets = nets.get('ipv4', []) + nets.get('ipv6', [])
    nets = [ipaddress.ip_network(net) for net in nets]

    # check if delegated net is subnet to a given supernetwork allocation
    dnet = ipaddress.ip_network(dnet)
    contained = False
    for net in nets:
        if type(net) is not type(dnet):
            # ip family mismatch
            continue

        # skip supernet check if delegated prefixlen is smaller than the current networks prefixlen
        if dnet.prefixlen < net.prefixlen:
            continue

        try:
            supernet = dnet.supernet(new_prefix=net.prefixlen)
            if net == supernet:
                contained = True
                break
        except ValueError as ex:
            errcnt += 1
            error("Exception during lookup of supernet for {} with prefixlen {}: {}".format(dnet, net.prefixlen, ex))

    if not contained:
        errcnt += 1
        error("Delegated Network {} is not contained in any existing network".format(dnet))

    # checking asn availability depends on having a complete list at the time
    # when we do the check. this is currently not the case as we're getting
    # the data from a generator.
    """
    if dasn not in asns:
        errcnt += 1
        error("Network {} was delegated to unknown ASN {}".format(dnet, dasn))
    """

    return errcnt


def do_checks(srcdir):
    """
    Check files for sanity.
    """

    asns = dict()
    bgp_gw = dict()
    bgp_gw_ip = dict()
    networks = {
        'IPv4': {},
        'IPv6': {},
    }
    warncnt = 0
    errcnt = 0
    domains = get_tlds()

    if len(domains) == 0:
        warn("Unable to load IANA Toplevel Domainlist from %s, skipping checks..." % TLD_LIST)
        warncnt += 1

    def filereader_error_handler(community):
        global errcnt
        error("Invalid YAML: %s" % community)
        errcnt += 1

    for community, data in get_communities_data(srcdir, [],
                                                filereader_error_handler):
        print("Checking", community)

        if 'asn' in data:
            errcnt += check_dupe("ASN", data['asn'], asns, community)
        else:
            if 'bgp' in data:
                errcnt += 1
                error("BGP block found, but no ASN")

        for bgp in data.get('bgp', []):
            errcnt += check_dupe("BGP peer name", bgp, bgp_gw, community)

            # prevent dashes in peernames, this breaks
            # a) bird config when not using quotation marks
            # b) bird cli
            if '-' in bgp:
                error("BGP Peers may not use dashes (breaks bird/bird cli): {peername}".format(peername=bgp))
                errcnt += 1

            lastbytes = set()

            for ipclass in data['bgp'][bgp]:
                upipclass = ipclass[:2].upper() + ipclass[2:]
                try:
                    ip = ip_family_address(upipclass,
                                           data['bgp'][bgp][ipclass])
                except ValueError:
                    errcnt += 1
                    error("Not an %s BGP address: %s (%s)" %
                          (upipclass, data['bgp'][bgp][ipclass], community))
                    continue

                errcnt += check_dupe("BGP IP", ip, bgp_gw_ip, community)
                try:
                    network = ICVPN_NETWORKS[upipclass]
                except KeyError:
                    errcnt += 1
                    error("Unknown BGP protocol class: %s (%s)" %
                          (upipclass, community))
                    continue

                if ip not in network:
                    errcnt += 1
                    error("%s BGP address in wrong subnet: %s (%s)" %
                          (upipclass, data['bgp'][bgp][ipclass], community))

                # we want that the last half of the host part of the
                # address spaces contains the same numeric value for all
                # protocol version
                # i.e. last 8 bits of IPv4 = last 16 bits of IPv6
                # i.e. IPv6 has to be padded with 8 zeroes
                matchbits = floor((ip.max_prefixlen - network.prefixlen) / 2)
                lastbytes.add(int(ip) % (1 << matchbits))

            if len(lastbytes) != 1:
                warncnt += 1
                warn("Last part of BGP addresses differs: " +
                     "%s (%s)" % (lastbytes, community))

        if 'networks' in data:
            for family in ('IPv4', 'IPv6'):
                try:
                    for net in data['networks'][family.lower()]:
                        errcnt += check_net(family,
                                            net,
                                            networks[family],
                                            community)
                except KeyError:
                    pass

        if 'domains' in data:
            errcnt += check_rdns(data)
            for domain in data.get('domains', []):
                errcnt += check_dupe("Domain", domain.upper(), domains, community)
                try:
                    domain.encode("ascii").decode("idna")
                except (UnicodeEncodeError, UnicodeError):
                    errcnt += 1
                    error("Domain not IDNA-encoded: %s" % (domain))

        if 'delegate' in data:
            for dasn, dnets in data['delegate'].items():
                for dnet in dnets:
                    errcnt += check_delegation(data['networks'], asns, dnet, dasn)

        if 'nameservers' in data:
            for nameserver in data['nameservers']:
                try:
                    ipaddress.ip_address(nameserver)
                except ValueError:
                    errcnt += 1
                    error("Not an IP-Adress: %s" % nameserver)

        if 'nameserver' in data:
            warncnt += 1
            warn("found unknown tag nameserver. Perhaps typo of nameservers?")

    print(ANSI_COLOR_WARN, "%d warning(s)" % warncnt, ANSI_COLOR_RESET)
    print(ANSI_COLOR_ERR, "%d error(s)" % errcnt, ANSI_COLOR_RESET)

    return 0 if errcnt == 0 else 1


if __name__ == "__main__":
    parser = OptionParser()
    parser.add_option("-s", "--sourcedir", dest="src",
                      help="Use files in DIR as input files. Default: ../icvpn-meta/",
                      metavar="DIR",
                      default="../icvpn-meta/")

    (options, args) = parser.parse_args()

    ret = do_checks(options.src)

    sys.exit(ret)