aceNamePort.py

# Name-Port Mappings Cisco

tcp_ports = {
    'bgp': '179',
    'chargen': '19',
    'cmd': '514',
    'daytime': '13',
    'discard': '9',
    'domain': '53',
    'echo': '7',
    'exec': '512',
    'finger': '79',
    'ftp': '21',
    'ftp-data': '20',
    'gopher': '70',
    'hostname': '101',
    'ident': '113',
    'irc': '194',
    'klogin': '543',
    'kshell': '544',
    'login': '513',
    'lpd': '515',
    'nntp': '119',
    'pim-auto-rp': '496',
    'pop2': '109',
    'pop3': '110',
    'smtp': '25',
    'sunrpc': '111',
    'syslog': '514',
    'tacacs': '49',
    'talk': '517',
    'telnet': '23',
    'time': '37',
    'uucp': '540',
    'whois': '43',
    'www': '80', }

udp_ports = {
    'biff': '512',
    'bootpc': '68',
    'bootps': '67',
    'discard': '9',
    'dnsix': '195',
    'domain': '53',
    'echo': '7',
    'isakmp': '500',
    'mobile-ip': '434',
    'nameserver': '42',
    'netbios-dgm': '138',
    'netbios-ns': '137',
    'netbios-ss': '139',
    'non500-isakmp': '4500',
    'ntp': '123',
    'pim-auto-rp': '496',
    'rip': '520',
    'snmp': '161',
    'snmptrap': '162',
    'sunrpc': '111',
    'syslog': '514',
    'tacacs': '49',
    'talk': '517',
    'tftp': '69',
    'time': '37',
    'who': '513',
    'xdmcp': '177', }


icmp_ports = {
    'administratively-prohibited',
    'alternate-address',
    'conversion-error',
    'dod-host-prohibited',
    'dod-net-prohibited',
    'dscp',
    'echo',
    'echo-reply',
    'fragments',
    'general-parameter-problem',
    'host-isolated',
    'host-precedence-unreachable',
    'host-redirect',
    'host-tos-redirect',
    'host-tos-unreachable',
    'host-unknown',
    'host-unreachable',
    'information-reply',
    'information-request',
    'mask-reply',
    'mask-request',
    'mobile-redirect',
    'net-redirect',
    'net-tos-redirect',
    'net-tos-unreachable',
    'net-unreachable',
    'network-unknown',
    'no-room-for-option',
    'option',
    'option-missing',
    'packet-too-big',
    'pak-len',
    'parameter-problem',
    'port-unreachable',
    'precedence',
    'precedence-unreachable',
    'protocol-unreachable',
    'reassembly-timeout',
    'redirect',
    'reflect',
    'router-advertisement',
    'router-solicitation',
    'source-quench',
    'source-route-failed',
    'time-exceeded',
    'time-range',
    'timestamp-reply',
    'timestamp-request',
    'tos',
    'traceroute',
    'ttl',
    'ttl-exceeded',
    'unreachable', }

lookup = {
    '6': tcp_ports,
    'tcp': tcp_ports,
    '17': udp_ports,
    'udp': udp_ports,
    'icmp': icmp_ports, }

acltests.py

# ACL Tests
'''File to generate mock ace statements for testing parser'''

action = [' permit     ', '  deny  ']
proto = ['ip', 'tcp', 'udp', 'icmp', '16']
src = ['any', 'host 10.10.10.1', '10.10.10.0 0.255.255.255', '99.99.0.1 0.255.255.255']
srcp = ['', 'eq 80', 'eq 443 www', 'eq 88 www 443', 'eq www', 'eq netbios-ns 88 netbios-ns', 'range 20 30', 'lt 42', 'gt 43']
dst = ['any', 'host 20.20.20.1', '20.20.20.0 0.255.255.255', 'host 10.10.10.0']
dstp = ['', 'eq 80', 'eq 443 www', 'eq 88 www 443', 'eq www', 'eq netbios-ns 88 netbios-ns', 'range 20 30', 'lt 42', 'gt 43']
est = ['', 'established']
ref = ['', 'reflect asdf', 'reflect oicu timeout 300']
log = ['', 'log', 'log-input']


def generateACETests():
    for a in action:
        for b in proto:
            for c in src:
                for d in srcp:
                    for e in dst:
                        for f in dstp:
                            for g in est:
                                for h in ref:
                                    for i in log:
                                        yield ' '.join(filter(lambda x: x != '', [a, b, c, d, e, f, g, h, i]))
                                        # print(count, a, b, c, d, e, f, g)


if __name__ == '__main__':
    count = 0
    for a in generateACETests():
        count += 1
        print(f'{count}:', a)

dissectace.py

# flip ace
import re
import aceNamePort
import acltests


def ace_elements(ace1):
    if ace1 is None or type(ace1) != str:
        return None
    '''Dissect ACE statement into it's parts.
    returns and array of each part as follows
    [action protocol src_address src_ports dst_address dst_ports established]
    Any empty element returns None. For example:
    [permit, tcp, host 10.10.10.1, None, Host 20.20.20.1, None, None]
    Anything that it trips up on returns a None in that location
    grade it using the valACE function to ensure it's correct '''
    statement = re.match('(?:permit|deny)', ace1)
    if not statement:
        return None

    '''statement protocol src_address src_ports dst_address dst_ports established
       The search array follows that order, and I create a search for each element.
    '''

    hst = (
        r'\b(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.' +
        r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.' +
        r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.' +
        r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b)')

    net = (
        r'\b(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.' +
        r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.' +
        r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.' +
        r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?) ' +
        r'\b(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.' +
        r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.' +
        r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.' +
        r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b')


    ACESearchs = [r'\b(?:permit|deny)\b',
                  r'\b(?:ip|tcp|udp|icmp|[0-9]+)\b',
                  r'\b(?:any|host (?:' + hst + ')|' + net,
                  r'\b(?:eq(?: [\w\-]+\b)+?|range [\w\-]+ [\w\-]+\b|(?:lt|gt) [\w\-]+\b)(?= (?:\d+\.|host|any))',
                  r'\b(?:any|host (?:' + hst + ')|' + net,
                  r'\b(?:eq(?: [\w\-]+\b)+?|range [\w\-]+ [\w\-]+\b|(?:lt|gt) [\w\-]+\b)((?= (?:established|reflect))|$)',
                  r'\b(?:established)(?:$)?',
                  r'\b(?:reflect)\b',
                  r'\b(?:[\w\-]+)(?= timeout|$)',
                  r'\b(?:timeout \d+)$']
    res = []
    loc = 0
    for search in ACESearchs:
        append = re.match(search, ace1[loc:])
        if append:
            loc += append.end() + 1
            res.append(append.group())
        else:
            res.append(append)
    return res


def keyReplace(ace):
    ace2 = ace.copy()
    if ace2 is None:
        return None
    if ace2[3]:
        newace3 = []
        plist = ace2[3].split(' ')
        lookup = aceNamePort.lookup.get(ace2[1])
        for item in plist[1:]:
            try:
                int(item)
                newace3.append(item)
            except (ValueError, TypeError) as e:
                if lookup and lookup.__class__ is dict:
                    newkey = lookup.get(item)
                    if newkey:
                        newace3.append(newkey)
                elif lookup:
                    if item in lookup:
                        pass
                else:
                    # print(ace2[3])
                    return None
        ace2[3] = ' '.join(newace3)
    if ace2[5]:
        newace5 = []
        plist = ace2[5].split(' ')
        lookup = aceNamePort.lookup.get(ace2[1])
        for item in plist[1:]:
            try:
                int(item)
                newace5.append(item)
            except (ValueError, TypeError) as e:
                if lookup and lookup.__class__ is dict:
                    newkey = lookup.get(item)
                    if newkey:
                        newace5.append(newkey)
                elif lookup:
                    if item in lookup:
                        pass
                else:
                    # print(ace2[5])
                    return None
        ace2[5] = ' '.join(newace5)
    return ace2


def valACE(ace3):
    if ace3.__class__ is not list or len(ace3) != 10:
        return False, 'bad list'
    if not ace3[2]:
        return False, 'missing source'
    if not ace3[4]:
        return False, 'missing destination'
    if ace3[6] and (ace3[1] != 'tcp' or ace3[1] != '6'):
        return False, 'established without tcp'
    if ace3[1] == 'icmp' and ace3[3]:
        return False, 'icmp with source ports'
    if ace3[3] and ace3[1] == 'tcp' and not ace3[6]:
        return False, 'missing established'
    if ace3[5] and ace3[6]:
        return False, 'established wrong side'
    if ace3[1] == 'ip' and (ace3[5] or ace3[3]):
        return False, 'ports with ip'
    if ace3[1] == 'icmp' and ace3[3]:
        return False, 'source ports with ICMP'
    if ace3[7]:
        return False, 'reflect needs removed'
    validports = aceNamePort.lookup.get(ace3[1])
    if validports is None and (ace3[3] or ace3[5]):
        return False, 'ports with unknown protocol'
    if ace3[1] == 'icmp' and ace3[5]:
        for port in ace3[5].split(' '):
            if port not in validports:
                return False, 'invalid ICMP-PORT'
    if ace3[3]:
        for port in ace3[3].split(' ')[1:]:
            try:
                int(port)
            except ValueError:
                if port not in validports:
                    return False, f'Port: {port} not valid'
    if ace3[5]:
        for port in ace3[5].split(' ')[1:]:
            try:
                int(port)
            except ValueError:
                if port not in validports:
                    return False, f'Port: {port} not valid'
    if ace3[3] and ace3[5]:
        return True, 'Warn: ports both sides'
    return True, 'pass'


def flipACE(ace4):
    if ace4 is None:
        return None
    if ace4.__class__ is not list or len(ace4) != 10:
        return None
    if ace4[6]:
        return (ace4[0], ace4[1], ace4[4], ace4[5], ace4[2], ace4[3], None)
    elif ace4[1] == 'tcp' and not ace4[6] and not ace4[3] and ace4[5]:
        return (ace4[0], ace4[1], ace4[4], ace4[5], ace4[2], ace4[3], 'established')
    else:
        return (ace4[0], ace4[1], ace4[4], ace4[5], ace4[2], ace4[3], None)


def printACE(ace5):
    return ' '.join([x for x in ace5 if x])


if __name__ == '__main__':
    total = 0
    fail = 0
    keyfail = 0
    for item in acltests.generateACETests():
        total += 1
        splitACE = ace_elements(item)
        if not splitACE:
            print('failed test:', item)
            break
        test, result = valACE(splitACE)
        if test:
            fail += 1
            print(result, item)
            continue
        elif not test:
            continue
        printnorm = printACE(splitACE)
        if printnorm != item:
            print('failed to parse', item, '!=', printnorm)
            break
        keyrep = keyReplace(splitACE)
        if not keyrep:
            keyfail += 1
            print('key fail:', item)
        flip = flipACE(splitACE)
        if not flip:
            print('flip fail:', item)
            break

    print(f'Total: {total}, TestFail: {fail}, KeyFail: {keyfail}')