Cisco Converged Communications: 2019

Most recent Cisco platforms support the Embedded Packet Capture feature, which allows us to take packet captures directly on a router or switch without having to use SPAN to an attached device. Detailed instructions here: Embedded Packet Capture for Cisco IOS and IOS-XE Configuration Example

Normally you can export the resulting packet capture to an FTP or TFTP server, however if this is blocked by firewalls or you don't have access to a suitable server, this is a workaround. Using the show monitor capture buffer dump command we can view the raw packet capture data, but it isn't in a format that Wireshark understands:

router1#show monitor capture buffer CAP dump
03:57:20.288 EST Nov 27 2019 : IPv4 LES CEF : Gi0/0 None

45ED3640: 3890A5D2 BDE07486 0BAD7BC0 08004500 8.%R=`t..-{@..E.
45ED3650: 0028928F 40003106 8B850A52 7A1E0A53 .(..@.1....Rz..S
45ED3660: 9CF8AD60 00161620 FB112DB5 2E9C5010 .x-`... {.-5..P.
45ED3670: F88C7092 00000000 00000000 00 x.p..........

03:57:20.288 EST Nov 27 2019 : IPv4 LES CEF : Gi0/0 None

45ED3640: 3890A5D2 BDE07486 0BAD7BC0 08004500 8.%R=`t..-{@..E.
45ED3650: 00289290 40003106 8B840A52 7A1E0A53 .(..@.1....Rz..S
45ED3660: 9CF8AD60 00161620 FB112DB5 2E9C5010 .x-`... {.-5..P.
45ED3670: F88C7092 00000000 00000000 00 x.p..........

If you copy this raw data to a text file, there is an open source tool to help with this: ciscoText2pcap

First pipe the saved output into ciscoText2pcap & in turn pipe its output into another text file:

cat input.txt | ./ciscoText2pcap.pl > output.txt

Then use Wireshark's text2pcap tool to convert it to a valid PCAP file:

text2pcap -d output.txt output.pcap

Note that you'll need both Perl & Wireshark installed to be able to do this.

UPDATE: Alternatively I knocked together a Python version that can also handle timestamps, source code below.

#!/usr/bin/env python
# (c) 2019, Chris Perkins
# Converts Cisco EPC "show monitor capture buffer dump" into format usable by text2pcap
# Use text2pcap -d -t "%Y-%m-%d %H:%M:%S." to convert output to PCAP whilst showing parsing info
# Based on ciscoText2pcap https://github.com/mad-ady/ciscoText2pcap

# v1.1 - added time stamp handling, converts into UTC
# v1.0 - initial release

import sys, re, pytz, datetime

if __name__ == "__main__":
    # Parse command line parameters
    if len(sys.argv) != 3:
        print("Please specify source & destination files as parameters.")
        sys.exit(1)
    # Parse input file via regex
    try:
        with open(sys.argv[1]) as in_file:
            with open(sys.argv[2], 'w') as out_file:
                packet_start = 0
                for line in in_file:
                    # Regex to find timestamp, then manipulate into format text2pcap can use, as %z or %Z is failing
                    time_date = re.search(r"^(\d\d:\d\d:\d\d\.\d+) (\w+) ([\w ]+) : ", line)
                    if time_date:
                        # Use pytz to parse timezone, then make datetime object TZ aware & convert into UTC
                        try:
                            tz = pytz.timezone(time_date.group(2))
                            dt = datetime.datetime.strptime(f"{time_date.group(3).rstrip()} {time_date.group(1).rstrip()}",
                                "%b %d %Y %H:%M:%S.%f")
                            dt = dt.replace(tzinfo=tz)
                            dt = dt.astimezone(tz=datetime.timezone.utc)
                            out_file.write(f"{dt.strftime('%Y-%m-%d %H:%M:%S.%f')}\n")
                        except IndexError:
                            pass
                        # Continue to next line in input file
                        continue
                    # Regex to find valid blocks of hexadecimal
                    hex_dump = re.search(r"^[0-9A-F]+:\s+((?:[0-9A-F]+ ){1,4}) (.+)\n", line)
                    if hex_dump:
                        # Iterate through each block of hex & split into sets of 2 digits with spaces inbetween
                        char_list = hex_dump.group(1).split()
                        for chars in char_list:
                            packet_hex = ''
                            for i in range(1,len(chars),2):
                                packet_hex += f"{chars[i-1:i+1]} "
                            packet_hex = packet_hex.rstrip()
                            # Output packet as offset (8 hex digits) + hex string
                            out_file.write(f"{packet_start:08X} {packet_hex}\n")
                            packet_start += len(chars) // 2
                    else:
                        # End of packet
                        packet_start = 0
    except FileNotFoundError:
        print(f"Unable to open file {sys.argv[1]}")
        sys.exit(1)
    except OSError:
        print(f"Unable to write file {sys.argv[2]}")
        sys.exit(1)

Overview
Tool to check the external phone number mask on the primary line of phones (tkclass=1) & device profiles (tkclass=254). Incorrect masks can then optionally be fixed via import from a CSV files.
It shares dial plan configuration in dialplan.json with the Dial Plan Analyser tool.
Requires Python 3 to run, many Linux distros have Python installed by default. For Windows the easiest install is the official Python Windows version, or Miniconda works fine too:
Miniconda distribution of Python: https://conda.io/miniconda.html
Official Python distribution: https://www.python.org/downloads/

The lxml, Requests, urllib3 and Zeep libraries are required to work.

Version History
Written by Chris Perkins in 2019:
v1.1 - fixed CSV output to UTF-8, fixed E.164 mask handling.
v1.0 – initial release.

All testing was done using Windows with CUCM v11.5.

Using the Tool
It connects to CUCM via the AXL API, so the AXL schema for the version of CUCM in use is required, this is downloaded from CUCM via Application > Plugins > Cisco AXL Toolkit. The required files contained within the .zip file are AXLAPI.wsdl, AXLEnums.xsd and AXLSoap.xsd.
Different CUCM servers are defined in JSON formatted files, allowing for multiple CUCM clusters running different versions (and thus different AXL schemas). Load the CSV file via File > Load AXL:

It will then prompt for the password:

If you wish to save the output in a CSV file, enter the filename into the text box:

Click Check Number Masks, the results will be displayed & optionally saved.

To fix external phone number masks, first review the outputted CSV file. Remove any rows that should be left alone & optionally adjust the New Number Mask if desired, then save it.

If you wish to save the failed updates to a CSV file, enter the filename into the text box. Then click Update Number Masks, it will prompt for the CSV file of updates to make. Any failed updates will be displayed & optionally saved.

Customising the Tool
The direct dial ranges to search for can be customised, so that the tool can be used for any CUCM cluster. These settings are stored in dialplan.json (shared with the Dial Plan Analyser) in JSON format, for example:
[
{
"range_start": "87300",
"range_end": "87399",
"partition": "lon_line_pt",
"mask": "0203100XXXX",
"description": "London 020310073XX"
}
]

• The JSON file starts with [ and ends with ].
• Each direct dial range is enclosed within { } and contains parameters for the description, range start, range end, mask and partition. The field headings and values must be enclosed within “”.
• The range end must be greater than the range start.
• The direct dial ranges must have a comma after each, except for the last one.

So to add another range to the above example:
[
{
"range_start": "87300",
"range_end": "87399",
"partition": "lon_line_pt",
"mask": "0203100XXXX",
"description": "London 020310073XX"
},
{
"range_start": "80501",
"range_end": "80700",
"partition": "lon_line_pt",
"mask": "0207170XXXX",
"description": "London 02071700[5-7]XX"
}
]

The parameters for using AXL are also stored in JSON format:
[
{
"fqdn": "cucm-emea-pub.somewhere.com",
"username": "AppAdmin",
"wsdl_file": "file://C://temp//AXLAPI.wsdl"
}
]

• “fqdn” should be the FQDN or IP address of the target CUCM publisher.
• “username” is an application or end user with the Standard AXL API Access role.
• “wsdl_file” points to the location of the AXL schema, note the slightly different path syntax for Windows.

Source Code

#!/usr/bin/env python

# v1.1 - written by Chris Perkins in 2019

# Finds & fixes primary DN's in specified range(s) with an External Phone Number Masks that doesn't match the approved list

# v1.1 - fixed CSV output to UTF-8, fixed E.164 mask handling

# v1.0 – initial release

# Original AXL SQL query code courtesy of Jonathan Els - https://afterthenumber.com/2018/04/27/serializing-thin-axl-sql-query-responses-with-python-zeep/

# To Do:

# Improve the GUI

import sys, json, csv

import tkinter as tk

import requests

from tkinter import ttk

from tkinter import filedialog, simpledialog, messagebox

from collections import OrderedDict

from zeep import Client

from zeep.cache import SqliteCache

from zeep.transports import Transport

from zeep.plugins import HistoryPlugin

from zeep.exceptions import Fault

from zeep.helpers import serialize_object

from requests import Session

from requests.auth import HTTPBasicAuth

from urllib3 import disable_warnings

from urllib3.exceptions import InsecureRequestWarning

from lxml import etree

# GUI and main code

class GUIFrame(tk.Frame):

def __init__(self, parent):

"""Constructor checks parameters and initialise variables"""

self.axl_input_filename = None

self.axl_password = ""

self.csv_input_filename = None

try:

with open("dialplan.json") as f:

self.json_data = json.load(f)

for range_data in self.json_data:

try:

if len(range_data['range_start']) != len(range_data['range_end']):

tk.messagebox.showerror(title="Error", message="The first and last numbers in range must be of equal length.")

sys.exit()

elif int(range_data['range_start']) >= int(range_data['range_end']):

tk.messagebox.showerror(title="Error", message="The last number in range must be greater than the first.")

sys.exit()

except (TypeError, ValueError, KeyError):

tk.messagebox.showerror(title="Error", message="Number range parameters incorrectly formatted.")

sys.exit()

try:

if not range_data['mask']:

tk.messagebox.showerror(title="Error", message="Number mask must be specified.")

sys.exit()

except KeyError:

tk.messagebox.showerror(title="Error", message="Number mask must be specified.")

sys.exit()

try:

if not range_data['partition']:

tk.messagebox.showerror(title="Error", message="Partition must be specified.")

sys.exit()

except KeyError:

tk.messagebox.showerror(title="Error", message="Partition must be specified.")

sys.exit()

except FileNotFoundError:

messagebox.showerror(title="Error", message="Unable to open JSON file.")

sys.exit()

except json.decoder.JSONDecodeError:

messagebox.showerror(title="Error", message="Unable to parse JSON file.")

sys.exit()

tk.Frame.__init__(self, parent)

parent.geometry("320x480")

self.pack(fill=tk.BOTH, expand=True)

menu_bar = tk.Menu(self)

file_menu = tk.Menu(menu_bar, tearoff=0)

file_menu.add_command(label="Load AXL", command=self.open_json_file_dialog)

file_menu.add_separator()

file_menu.add_command(label="Exit", command=self.quit)

menu_bar.add_cascade(label="File", menu=file_menu)

parent.config(menu=menu_bar)

tk.Label(self, text="Output Filename:").place(relx=0.2, rely=0.0, height=22, width=200)

self.output_csv_text = tk.StringVar()

tk.Entry(self, textvariable=self.output_csv_text).place(relx=0.2, rely=0.05, height=22, width=200)

tk.Button(self, text="Check Number Masks", command=self.check_masks).place(relx=0.08, rely=0.12, height=22, width=135)

tk.Button(self, text="Update Number Masks", command=self.update_masks).place(relx=0.52, rely=0.12, height=22, width=135)

self.results_count_text = tk.StringVar()

self.results_count_text.set("Results Found: ")

tk.Label(self, textvariable=self.results_count_text).place(relx=0.20, rely=0.18, height=22, width=210)

list_box_frame = tk.Frame(self, bd=2, relief=tk.SUNKEN)

list_box_scrollbar_y = tk.Scrollbar(list_box_frame)

list_box_scrollbar_x = tk.Scrollbar(list_box_frame, orient=tk.HORIZONTAL)

self.list_box = tk.Listbox(list_box_frame, xscrollcommand=list_box_scrollbar_x.set, yscrollcommand=list_box_scrollbar_y.set)

list_box_frame.place(relx=0.02, rely=0.22, relheight=0.75, relwidth=0.96)

list_box_scrollbar_y.place(relx=0.94, rely=0.0, relheight=1.0, relwidth=0.06)

list_box_scrollbar_x.place(relx=0.0, rely=0.94, relheight=0.06, relwidth=0.94)

self.list_box.place(relx=0.0, rely=0.0, relheight=0.94, relwidth=0.94)

list_box_scrollbar_y.config(command=self.list_box.yview)

list_box_scrollbar_x.config(command=self.list_box.xview)

def element_list_to_ordered_dict(self, elements):

"""Convert list to OrderedDict"""

return [OrderedDict((element.tag, element.text) for element in row) for row in elements]

def sql_query(self, service, sql_statement):

"""Execute SQL query via AXL and return results"""

try:

axl_resp = service.executeSQLQuery(sql=sql_statement)

try:

return self.element_list_to_ordered_dict(serialize_object(axl_resp)["return"]["rows"])

except KeyError:

# Single tuple response

return self.element_list_to_ordered_dict(serialize_object(axl_resp)["return"]["row"])

except TypeError:

# No SQL tuples

return serialize_object(axl_resp)["return"]

except requests.exceptions.ConnectionError as e:

tk.messagebox.showerror(title="Error", message=str(e))

return None

def sql_update(self, service, sql_statement):

"""Execute SQL update via AXL and return rows updated"""

try:

axl_resp = service.executeSQLUpdate(sql=sql_statement)

return serialize_object(axl_resp)["return"]["rowsUpdated"]

except requests.exceptions.ConnectionError as e:

tk.messagebox.showerror(title="Error", message=str(e))

return None

def read_axl(self, output_filename):

"""Check configuration via AXL SQL query"""

try:

self.list_box.delete(0, tk.END)

self.results_count_text.set("Results Found: ")

with open(self.axl_input_filename) as f:

axl_json_data = json.load(f)

for axl_json in axl_json_data:

try:

if not axl_json['fqdn']:

tk.messagebox.showerror(title="Error", message="FQDN must be specified.")

return

except KeyError:

tk.messagebox.showerror(title="Error", message="FQDN must be specified.")

return

try:

if not axl_json['username']:

tk.messagebox.showerror(title="Error", message="Username must be specified.")

return

except KeyError:

tk.messagebox.showerror(title="Error", message="Username must be specified.")

return

try:

if not axl_json['wsdl_file']:

tk.messagebox.showerror(title="Error", message="WSDL file must be specified.")

return

except KeyError:

tk.messagebox.showerror(title="Error", message="WSDL file must be specified.")

return

except FileNotFoundError:

messagebox.showerror(title="Error", message="Unable to open JSON file.")

return

except json.decoder.JSONDecodeError:

messagebox.showerror(title="Error", message="Unable to parse JSON file.")

return

axl_binding_name = "{http://www.cisco.com/AXLAPIService/}AXLAPIBinding"

axl_address = "https://{fqdn}:8443/axl/".format(fqdn=axl_json['fqdn'])

session = Session()

session.verify = False

session.auth = HTTPBasicAuth(axl_json['username'], self.axl_password)

transport = Transport(cache=SqliteCache(), session=session, timeout=60)

history = HistoryPlugin()

try:

client = Client(wsdl=axl_json['wsdl_file'], transport=transport, plugins=[history])

except FileNotFoundError as e:

tk.messagebox.showerror(title="Error", message=str(e))

return

axl = client.create_service(axl_binding_name, axl_address)

# List each primary DN in specified range(s) with an External Phone Number Mask that doesn't match the approved list

cntr = 0

result_list = [["DN", "Partition", "Device Name", "Device Description", "Number Mask", "New Number Mask", "pkid"]]

self.list_box.insert(tk.END, "DN, Partition, Device Name, Device Description, Number Mask, New Number Mask, pkid\n")

sql_statement = "SELECT n.dnorpattern, p.name AS pname, d.name, d.description, dnmap.e164mask, dnmap.pkid FROM device d INNER JOIN devicenumplanmap dnmap ON dnmap.fkdevice=d.pkid INNER JOIN numplan n ON dnmap.fknumplan=n.pkid LEFT JOIN routepartition p ON n.fkroutepartition=p.pkid WHERE (d.tkclass=1 OR d.tkclass=254) AND dnmap.numplanindex=1 ORDER BY n.dnorpattern"

try:

for row in self.sql_query(service=axl, sql_statement=sql_statement):

try:

# Handle None results

if row['pkid'] is None:

dnmap_pkid = ""

else:

dnmap_pkid = row['pkid']

if row['e164mask'] is None:

dnmap_e164mask = ""

else:

dnmap_e164mask = row['e164mask']

if row['description'] is None:

d_description = ""

else:

d_description = row['description']

if row['name'] is None:

d_name = ""

else:

d_name = row['name']

if row['pname'] is None:

p_name = ""

else:

p_name = row['pname']

if row['dnorpattern'] is None:

n_dnorpattern = ""

else:

n_dnorpattern = row['dnorpattern']

is_valid_mask = False

is_in_range = False

correct_mask = ""

for range_data in self.json_data:

try:

range_start = int(range_data['range_start'])

range_end = int(range_data['range_end'])

dn = int(n_dnorpattern)

if p_name.upper() == range_data['partition'].upper() and dn >= range_start and dn <= range_end:

if dnmap_e164mask.upper() == range_data['mask'].upper():

is_valid_mask = True

is_in_range = True

break

else:

is_in_range = True

correct_mask = range_data['mask']

break

except TypeError:

continue

if is_in_range == True and is_valid_mask == False:

self.list_box.insert(tk.END, n_dnorpattern + ', ' + p_name + ', ' + d_name + ', ' + d_description + ', ' + dnmap_e164mask + ', ' + correct_mask + ', ' + dnmap_pkid)

result_list.append([n_dnorpattern, p_name, d_name, d_description, dnmap_e164mask, correct_mask, dnmap_pkid])

cntr += 1

except TypeError:

continue

except TypeError:

pass

except Fault as thin_axl_error:

tk.messagebox.showerror(title="Error", message=thin_axl_error.message)

return

self.results_count_text.set("Results Found: " + str(cntr))

# Output to CSV file if required

try:

if len(output_filename) != 0:

with open(output_filename, 'w', newline='', encoding='utf-8-sig') as csv_file:

writer = csv.writer(csv_file)

writer.writerows(result_list)

except OSError:

tk.messagebox.showerror(title="Error", message="Unable to write CSV file.")

def write_axl(self, output_filename):

"""Update configuration via AXL SQL query"""

try:

self.list_box.delete(0, tk.END)

self.results_count_text.set("Updates Made: ")

with open(self.axl_input_filename) as f:

axl_json_data = json.load(f)

for axl_json in axl_json_data:

try:

if not axl_json['fqdn']:

tk.messagebox.showerror(title="Error", message="FQDN must be specified.")

return

except KeyError:

tk.messagebox.showerror(title="Error", message="FQDN must be specified.")

return

try:

if not axl_json['username']:

tk.messagebox.showerror(title="Error", message="Username must be specified.")

return

except KeyError:

tk.messagebox.showerror(title="Error", message="Username must be specified.")

return

try:

if not axl_json['wsdl_file']:

tk.messagebox.showerror(title="Error", message="WSDL file must be specified.")

return

except KeyError:

tk.messagebox.showerror(title="Error", message="WSDL file must be specified.")

return

except FileNotFoundError:

messagebox.showerror(title="Error", message="Unable to open JSON file.")

return

except json.decoder.JSONDecodeError:

messagebox.showerror(title="Error", message="Unable to parse JSON file.")

return

axl_binding_name = "{http://www.cisco.com/AXLAPIService/}AXLAPIBinding"

axl_address = "https://{fqdn}:8443/axl/".format(fqdn=axl_json['fqdn'])

session = Session()

session.verify = False

session.auth = HTTPBasicAuth(axl_json['username'], self.axl_password)

transport = Transport(cache=SqliteCache(), session=session, timeout=60)

history = HistoryPlugin()

try:

client = Client(wsdl=axl_json['wsdl_file'], transport=transport, plugins=[history])

except FileNotFoundError as e:

tk.messagebox.showerror(title="Error", message=str(e))

return

axl = client.create_service(axl_binding_name, axl_address)

# Update External Phone Number Masks contained in CSV file

cntr = 0

result_list = [["DN", "Partition", "Device Name", "Device Description", "Number Mask", "New Number Mask", "pkid"]]

self.list_box.insert(tk.END, "DN, Partition, Device Name, Device Description, Number Mask, New Number Mask, pkid\n")

# Parse input CSV file & make updates based on the content

try:

with open(self.csv_input_filename, encoding='utf-8-sig') as f:

reader = csv.reader(f)

header_row = next(reader)

if header_row[5] != "New Number Mask" or header_row[6] != "pkid":

tk.messagebox.showerror(title="Error", message="Unable to parse CSV file.")

return

for row in reader:

try:

# Check replacement mask has only valid characters

is_valid = True

for mask_char in row[5]:

if mask_char not in ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'X', '+']:

self.list_box.insert(tk.END, row[0] + ', ' + row[1] + ', ' + row[2] + ', ' + row[3] + ', ' + row[4] + ', ' + row[5] + ', ' + row[6])

result_list.append(row)

is_valid = False

break

if is_valid == False:

continue

sql_statement = "UPDATE devicenumplanmap SET e164mask='" + row[5] + "' WHERE pkid='" + row[6] + "'"

num_results = self.sql_update(service=axl, sql_statement=sql_statement)

# List updates that failed

if num_results < 1:

self.list_box.insert(tk.END, row[0] + ', ' + row[1] + ', ' + row[2] + ', ' + row[3] + ', ' + row[4] + ', ' + row[5] + ', ' + row[6])

result_list.append(row)

else:

cntr += 1

except TypeError:

continue

except Fault as thin_axl_error:

tk.messagebox.showerror(title="Error", message=thin_axl_error.message)

break

except KeyError:

tk.messagebox.showerror(title="Error", message="Unable to parse CSV file.")

pass

except FileNotFoundError:

tk.messagebox.showerror(title="Error", message="Unable to open CSV file.")

return

self.results_count_text.set("Updates Made: " + str(cntr) + " (failures below)")

# Output to CSV file if required

try:

if len(output_filename) != 0:

with open(output_filename, 'w', newline='', encoding='utf-8-sig') as csv_file:

writer = csv.writer(csv_file)

writer.writerows(result_list)

except OSError:

tk.messagebox.showerror(title="Error", message="Unable to write CSV file.")

def check_masks(self):

"""Validate parameters and then call AXL query"""

if not self.axl_input_filename:

tk.messagebox.showerror(title="Error", message="No AXL file selected.")

return

output_string = self.output_csv_text.get()

if len(output_string) == 0:

self.read_axl('')

else:

self.read_axl(output_string)

def update_masks(self):

"""Validate parameters and then call AXL update"""

if not self.axl_input_filename:

tk.messagebox.showerror(title="Error", message="No AXL file selected.")

return

self.open_csv_file_dialog()

if not self.csv_input_filename:

tk.messagebox.showerror(title="Error", message="No CSV file selected.")

return

output_string = self.output_csv_text.get()

if len(output_string) == 0:

self.write_axl('')

else:

self.write_axl(output_string)

def open_json_file_dialog(self):

"""Dialogue to prompt for JSON file to open and AXL password"""

self.axl_input_filename = tk.filedialog.askopenfilename(initialdir="/", filetypes=(("JSON files", "*.json"),("All files", "*.*")))

self.axl_password = tk.simpledialog.askstring("Input", "AXL Password?", show='*')

def open_csv_file_dialog(self):

"""Dialogue to prompt for CSV file to open"""

self.csv_input_filename = tk.filedialog.askopenfilename(initialdir="/", filetypes=(("CSV files", "*.csv"),("All files", "*.*")))

if __name__ == "__main__":

disable_warnings(InsecureRequestWarning)

# Initialise TKinter GUI objects

root = tk.Tk()

root.title("External Number Mask Checker v1.1")

GUIFrame(root)

root.mainloop()

Cisco Converged Communications

Wednesday, 27 November 2019

Converting Monitor Capture Dump to PCAP

Thursday, 5 September 2019

Nexus Switch PTP Intervals

Wednesday, 24 April 2019

External Phone Number Mask Checker