cancel
Showing results for 
Search instead for 
Did you mean: 
cancel
77
Views
0
Helpful
0
Comments
uganesan
Level 1
Level 1

Simple python application to track CUCM configuration changes

Interested in building Cisco Unified Communications Manager (CUCM) change tracking system using python? Feel free to try the code in the below GitHub page and let me know your thoughts.

https://github.com/usharanig92/Cisco-Unified-Communications-CUCM-Config-Tracker

Code Logic

The idea is to maintain a golden template or baseconfig template of all the CUCM system configurations that we would like to monitor, this includes RoutePatterns, TransPatterns, AppUser and so on. In this code, this template is maintained as a csv file for each item in a directory called baseconfig.

Then there is a second directory called runningconfig, which contains the similar set of files as the baseconfig directory. The main difference is that the configurations in the runningconfig directory is being updated by this application upon receiving the change notification through CUCM List Change notification AXL API. Refer here to know more about how this API works.

uganesan_0-1766183033659.png

def list_change(service, history, mode, templates, email_recipient):
    try:
        auto_check(service, history, mode, email_recipient)
        resp = service.listChange()
    except Fault as err:
        if does_last_response_report_credential_error(history):
            raise ServerCredentialError(err)
        else:
            print(f"\nZeep error: polling listChange: {err}")
            sys.exit(1)
    print("Initial listChange response:")
    print()
    print(resp)

    queue_id = resp.queueInfo.queueId
    next_start_change_id = resp.queueInfo.nextStartChangeId

    print()
    print("Starting loop to monitor changes...")
    print("(Press Ctrl+C to exit)")
    print()
    print(f"Action doGet? Type{16 * ' '} UUID{32 * ' '} Field{10 * ' '} Value")
    print(f"{6 * '-'} {6 * '-'} {20 * '-'} {36 * '-'} {15 * '-'} {15 * '-'}")

    actions = {"a": "Add", "u": "Update", "r": "Remove"}

    while True:
        start_change_id = {"queueId": queue_id, "_value_1": next_start_change_id}
        object_list = [{"object": list(templates.keys())}]
        change_types = set()
        # Execute the listChange request
        try:
            resp = service.listChange(start_change_id, object_list)

        except Exception as err:
            print(f"\nZeep error: polling listChange: {err}")
            break

        if resp.changes:
            # Loop through each change in the changes list
            for change in resp.changes.change:
                change_types.add(change.type)
                # If there are any items in the changedTags list...
                if change.changedTags:
                    # Loop through each changedTag
                    for x in range(len(change.changedTags.changedTag)):
                        # If this is the first changedTag, print a full line of data...
                        if x == 0:
                            print(
                                actions[change.action].ljust(6, " "),
                                change.doGet.ljust(6, " "),
                                change.type.ljust(20, " "),
                                change.uuid,
                                change.changedTags.changedTag[x].name.ljust(15, " "),
                                change.changedTags.changedTag[x]._value_1,
                            )

                        # otherwise print just the field/value part of the line
                        else:
                            print(
                                71 * " ",
                                change.changedTags.changedTag[x].name.ljust(15, " "),
                                change.changedTags.changedTag[x]._value_1,
                            )

                # otherwise just print the minimum details
                else:
                    print(
                        actions[change.action].ljust(6, " "),
                        change.doGet.ljust(6, " "),
                        change.type.ljust(20, " "),
                        change.uuid,
                    )
            try:
                for change in change_types:
                    response = execute_sql_query(service, history, templates[change])
                    update_runningconfig(change, response, mode, email_recipient)
            except Exception as e:
                print("Unable to update the running config" + str(e))
                break

        # Update the next highest change Id
        next_start_change_id = resp.queueInfo.nextStartChangeId
        get_presence_server_high_availability_and_save_in_csv(mode)
        time.sleep(600)

    # We should not exit from the "while True" loop above unless there is an
    # error.
    return 1

The list_change function contains two parts:

  1. auto_check This initial auto_check is to first pull all the configuration items defined in the templates dictionary and update in the runningconfig file by calling the update_runningconfig function.
  2. The second part is executing listChange API, which runs in a while loop to continuously monitor for the changes in CUCM and updates the runningconfig files.

Sample output of list_change function

uganesan_5-1766183650970.png

Below is the definition of auto_check.

def auto_check(service, history, mode, email_recipient):
    result = ""
    for template, sql in templates.items():
        try:
            resp = execute_sql_query(service, history, sql)
        except Fault as err:
            if does_last_response_report_credential_error(history):
                raise ServerCredentialError(err)
            else:
                raise
        result += update_runningconfig(template, resp, mode, email_recipient)
    if result:
        print("Base and Running configs has been modified")
    else:
        print("No changes Detected")

# Sample templates dictionary
templates = {
    "RoutePattern": route_pattern_sql,
    "TransPattern": translation_pattern_sql,
    "RouteGroup": route_groups_sql,
    "DevicePool": device_pools_sql,
    "GeoLocation": geolocation_sql,
    "CallManagerGroup": call_manager_group_sql,
    "Css": css_sql,
}

The auto_check function runs the sql query for all the system configurations keys defined under templates dictionary and invokes the update_runningconfig function.

update_runningconfig function updates the csv files in the runningconfig directory with the configurations pulled from CUCM using sql query, compares it with the config in the baseconfig directory and emails if there are any changes between those.

For example: if an admin updated the CSS in CUCM, list_change function receives the notification about the change, which triggers the update_runningconfig function. The function then runs the css_sql query to pull the updated configuration from CUCM and updates the css.csv file in runningconfig. This file is then compared with the css.csv file in the baseconfig directory and notifies admin team about the changes. The admin can update the baseconfig using update_baseconfig argument.

def update_baseconfig(name, username, mode, commit, email_recipient):
    source = get_config_relative_path("runningconfig", name, mode)
    destination = get_config_relative_path("baseconfig", name, mode)
    diff = compare_running_with_base(name, mode)
    if diff:
        body = (
            f"New configs have been committed successfully from the above running config to the base repo <br /><br />"
            f"<strong> commit message </strong> {commit} " + diff
        )
        subject = f"CUCM Configs: Base config updated by {username}"
        email(
            email_recipient=email_recipient,
            subject=subject,
            body=body,
            mode=mode,
        )
        copyfile(source, destination)
        print(
            "New configs have been committed successfully from the above running config to the base repo. "
            f"commit message: {commit}"
        )

update_baseconfig function gets the diff of both the files, updates the baseconfig with the file from the running config and emails the admin team about the changes and the committer of the change.

The below screenshot shows all the configuration items that this application monitors.

uganesan_4-1766183486532.png

 

Tool Highlights

The biggest benefit of this tool is that, it works on push mechanism from CUCM instead of constantly polling CUCM for changes. This significantly reduces API load on the server and improves application efficiency. And this script uses sql query to pull the configurations from CUCM, which is a thin AXL API and does not consume high CPU or memory for execution.

Moreover listChange API is the lightweight API and it has no restriction on polling limit according to Cisco. The polling interval can be modified according to how early we would like to be notified about the changes.

Hope this was useful. Please feel free to comment on any improvements or issues that you run into in using this application.

Getting Started

Find answers to your questions by entering keywords or phrases in the Search bar above. New here? Use these resources to familiarize yourself with the community: