Python Bot for Automatic TR-069 ONU Configuration | Streamline Network Management

Building a Robust TR-069 Database Management System in Python

Managing network configurations and database operations for OLTs (Optical Line Terminals) is a critical task for telecom service providers. The TR-069 Database Management System is a Python-based solution designed to streamline these processes, ensuring secure IP validation, efficient database management, and seamless updates to OLT configurations.

This post will explore the key components of the system, its functionality, and how it can be used to manage OLT configurations effectively.

Features of the TR-069 Database Management System

1. Database Operations:
   • Create and manage MySQL databases and tables for TR-069 configurations.
   • Truncate data from tables when necessary.
   • Insert, update, and retrieve OLT records.
2. IP Validation:
   • Validate the format of OLT IP addresses.
   • Authorize IPs against a predefined list of allowed addresses.
3. Logging:
   • Log all operations to a file and the console for better traceability.
4. User-Friendly Interface:
   • Command-line prompts for user input.
   • Clear and concise feedback for operations performed.

Code Walkthrough

1. Database Configuration

The DatabaseConfig class encapsulates the database connection details such as host, username, password, and the database name.

class DatabaseConfig:
    """Database configuration class"""
    def __init__(self, host: str, user: str, password: str):
        self.host = host
        self.user = user
        self.password = password
        self.database_name = "tr069_bsnl_kasrawad"

2. Database Management

The DatabaseManager class handles all database-related operations, including connecting to the database, creating tables, truncating data, and updating OLT configurations.

Creating Tables

The system ensures that the required tables (tr069_tip and tr69_new) exist before performing any operations:

def create_tr069_tip_table(self):
    """Create tr069_tip table if it doesn't exist"""
    create_query = """
    CREATE TABLE IF NOT EXISTS tr069_tip (
        id INT AUTO_INCREMENT PRIMARY KEY,
        LCOName VARCHAR(255),
        OltLocation VARCHAR(255),
        OltMake VARCHAR(255),
        OltIp VARCHAR(255) UNIQUE,
        Oltuser VARCHAR(255),
        Oltpwd VARCHAR(255),
        Login VARCHAR(255),
        pon VARCHAR(255),
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
    )
    """
    self.cursor.execute(create_query)
    logging.info("tr069_tip table verified/created successfully")

Updating OLT Details

The system can insert new OLT records or update existing ones based on the IP address:

def update_olt_details(self, olt_make: str, olt_ip: str, olt_user: str, olt_password: str):
    """Update or insert OLT details in tr069_tip table"""
    if self.check_olt_exists(olt_ip):
        update_query = """
        UPDATE tr069_tip 
        SET OltMake = %s, 
            Oltuser = %s, 
            Oltpwd = %s,
            updated_at = CURRENT_TIMESTAMP
        WHERE OltIp = %s
        """
        self.cursor.execute(update_query, (olt_make, olt_user, olt_password, olt_ip))
        logging.info(f"Updated existing OLT record for IP: {olt_ip}")
    else:
        insert_query = """
        INSERT INTO tr069_tip (OltMake, OltIp, Oltuser, Oltpwd)
        VALUES (%s, %s, %s, %s)
        """
        self.cursor.execute(insert_query, (olt_make, olt_ip, olt_user, olt_password))
        logging.info(f"Inserted new OLT record for IP: {olt_ip}")

4. User Interaction

The system provides a command-line interface for users to input OLT and database configuration details. The get_user_input function collects this information:

def get_user_input() -> tuple:
    """Get user input for OLT and database configuration"""
    print("=== OLT Configuration ===")
    olt_make = input("Enter OLT make: ").strip()
    olt_ip = input("Enter OLT IP address: ").strip()
    olt_user = input("Enter OLT username: ").strip()
    olt_password = input("Enter OLT password: ").strip()

    print("\n=== Database Configuration ===")
    db_host = input("Enter database host (default: localhost): ").strip() or "localhost"
    db_user = input("Enter database username: ").strip()
    db_password = input("Enter database password: ").strip()

    return (olt_make, olt_ip, olt_user, olt_password, db_host, db_user, db_password)

5. Logging

All operations are logged using Python’s logging module. Logs are written to both a file (tr069_system.log) and the console.

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('tr069_system.log'),
        logging.StreamHandler()
    ]
)

How to Use the System

1. Setup:
   • Ensure Python 3 and MySQL are installed on your system.
   • Install the required Python packages using pip install mysql-connector-python.
2. Run the Script:
   • Execute the script from the command line: python tr069_system.py.
   • Provide the necessary OLT and database configuration details when prompted.

python dbsetup.py

1. Perform Operations:
   • The system will validate the OLT IP, connect to the database, and perform the requested operations (e.g., updating OLT details, truncating tables).

1. Script Metadata and Imports

Purpose:

The script begins with metadata and necessary imports to clarify its purpose and include required libraries.

#!/usr/bin/env python3
"""
TR-069 Database Management System
Description: Manages OLT configurations with IP validation and MySQL database operations
Author: Assistant
Version: 1.2
"""

---

• Metadata: Provides a high-level overview of the script’s purpose and version.
• Imports:
  • mysql.connector: For connecting to and interacting with the MySQL database.
  • ipaddress: For validating IP address formats.
  • logging: For logging system events.
  • sys, os: For system-related operations like clearing the screen.
  • typing.List and Optional: Used for type hints in function definitions.
  • datetime: For handling timestamps.

2. Screen Clearing Function

Purpose:

Clears the terminal screen at the start of the program for better readability.

def clear_screen():
    """Clear the terminal screen."""
    os.system('cls' if os.name == 'nt' else 'clear')

• os.system: Executes a system command to clear the screen. It uses cls for Windows and clear for Linux/Mac.
• Why?: To provide a clean slate for the user when they run the script.

---

3. Logging Configuration

Purpose:

Configures logging to record events in both a log file (tr069_system.log) and the console.

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('tr069_system.log'),
        logging.StreamHandler()
    ]
)

• Log Levels: Logs are recorded at the INFO level. Errors are logged as ERROR.
• Handlers:
  • FileHandler: Writes logs to a file for persistent storage.
  • StreamHandler: Outputs logs to the console for real-time feedback.

5. Database Configuration Class

Purpose:

Encapsulates database connection details in a reusable object.

class DatabaseConfig:
    """Database configuration class"""
    def __init__(self, host: str, user: str, password: str):
        self.host = host
        self.user = user
        self.password = password
        self.database_name = "tr069_bsnl_kasrawad"

• Attributes:
  • host, user, password: Database connection credentials.
  • database_name: The name of the database being managed.
• Why?: Centralizes database connection details for easy reuse.

Database Manager Class

Purpose:

Handles all database-related operations, including connecting, creating tables, truncating data, and updating records.

a) Initialization and Connection

class DatabaseManager:
    """Handles all database operations"""
    def __init__(self, config: DatabaseConfig):
        self.config = config
        self.connection = None
        self.cursor = None

    def connect(self) -> bool:
        """Establish database connection"""
        try:
            self.connection = mysql.connector.connect(
                host=self.config.host,
                user=self.config.user,
                password=self.config.password,
                auth_plugin='mysql_native_password'
            )
            self.cursor = self.connection.cursor(buffered=True)
            logging.info("Successfully connected to MySQL server")
            return True
        except Error as e:
            logging.error(f"Database connection error: {e}")
            return False

• connect Method:
  • Establishes a connection to the MySQL database.
  • Creates a cursor for executing SQL queries.
  • Logs success or failure.

b) Disconnecting

def disconnect(self):
    """Close database connection"""
    try:
        if self.cursor:
            self.cursor.close()
        if self.connection and self.connection.is_connected():
            self.connection.close()
            logging.info("Database connection closed successfully")
    except Error as e:
        logging.error(f"Error disconnecting from database: {e}")

• loses the cursor and connection to free resources.

c) Creating Tables

The create_tr069_tip_table and create_tr69_new_table methods define the structure of the required tables.

def create_tr069_tip_table(self):
    """Create tr069_tip table if it doesn't exist"""
    create_query = """
    CREATE TABLE IF NOT EXISTS tr069_tip (
        id INT AUTO_INCREMENT PRIMARY KEY,
        ...
    )
    """
    self.cursor.execute(create_query)
    logging.info("tr069_tip table verified/created successfully")

• Why?: Ensures the necessary tables exist before performing operations.

d) Truncating Data

def truncate_tr69_new_table(self) -> bool:
    """Delete all data from tr69_new table"""
    try:
        self.cursor.execute("SET FOREIGN_KEY_CHECKS = 0")
        self.cursor.execute("TRUNCATE TABLE tr69_new")
        self.cursor.execute("SET FOREIGN_KEY_CHECKS = 1")
        self.connection.commit()
        logging.info("Successfully truncated tr69_new table")
        return True
    except Error as e:
        self.connection.rollback()
        logging.error(f"Error truncating tr69_new table: {e}")
        return False

• Temporarily disables foreign key checks to safely truncate the table.
• Rolls back changes if an error occurs.

e) Updating OLT Details

def update_olt_details(self, olt_make: str, olt_ip: str, olt_user: str, olt_password: str):
    """Update or insert OLT details in tr069_tip table"""
    if self.check_olt_exists(olt_ip):
        update_query = """
        UPDATE tr069_tip 
        SET OltMake = %s, 
            Oltuser = %s, 
            Oltpwd = %s,
            updated_at = CURRENT_TIMESTAMP
        WHERE OltIp = %s
        """
        self.cursor.execute(update_query, (olt_make, olt_user, olt_password, olt_ip))
    else:
        insert_query = """
        INSERT INTO tr069_tip (OltMake, OltIp, Oltuser, Oltpwd)
        VALUES (%s, %s, %s, %s)
        """
        self.cursor.execute(insert_query, (olt_make, olt_ip, olt_user, olt_password))
    self.connection.commit()

• Updates existing records or inserts new ones based on whether the IP already exists.

---

8. User Interaction

Header Display

def display_header():
    """Display application header"""
    print("\n" + "="*50)
    print("TR-069 Database Management System")
    print("Created by Assistant")
    print("TAKE BACKUP BEFORE RUNNING THE SCRIPT")
    print("="*50 + "\n")

• Displays an informative header to the user.

Input Collection

def get_user_input() -> tuple:
    """Get user input for OLT and database configuration"""
    olt_make = input("Enter OLT make: ").strip()
    olt_ip = input("Enter OLT IP address: ").strip()
    ...
    return (olt_make, olt_ip, olt_user, olt_password, db_host, db_user, db_password)

• Collects OLT and database configuration details via the command line.

---

9. Main Setup Function

Purpose:

Combines all the above components into a cohesive workflow.

def setup_database(config: DatabaseConfig, olt_make: str, olt_ip: str, 
                  olt_user: str, olt_password: str) -> bool:
    """Main database setup function"""
    if not IPValidator.validate_ip(olt_ip):
        logging.error(f"Invalid IP address format: {olt_ip}")
        return False

    if not IPValidator.is_authorized(olt_ip):
        logging.error(f"Unauthorized IP address: {olt_ip}")
        return False

    db_manager = DatabaseManager(config)
    if db_manager.connect():
        db_manager.create_database()
        db_manager.create_tr069_tip_table()
        db_manager.update_olt_details(olt_make, olt_ip, olt_user, olt_password)
        db_manager.disconnect()
        return True
    return False

1. Script Initialization

Purpose:

The script starts by importing required libraries, clearing the terminal screen, and displaying a header.

pythonCopyimport sys
import re
import telnetlib
import socket
import time
import mysql.connector
import os

• Libraries:
  • sys: For system-level operations (e.g., exiting the script).
  • re: For regular expressions to parse Telnet command outputs.
  • telnetlib: To establish Telnet sessions with OLT devices.
  • mysql.connector: To connect to and interact with the MySQL database.
  • os: For clearing the terminal screen.

Clearing the Screen:

pythonCopydef clear_screen():
    """Clear the terminal screen."""
    os.system('cls' if os.name == 'nt' else 'clear')

# Clear the screen at the start of the program
clear_screen()

• Why?:
• Provides a clean interface for the user when running the script.

Header Display:

print("\n" + "="*50)
print("Created by Farhan Mansuri JTO KASRAWAD")
print("TAKE BACKUP BEFORE RUN THE SCRIPT")
print("VISIT https://fresherhub.com for More Info")
print("="*50 + "\n")

• Displays essential information, including a warning to take a backup before running the script.

---

• Why?: Prevents unauthorized access by ensuring only predefined OLT IPs are processed.

User Input for OLT IP:

inolt = input("Enter OLT IP: ").strip()  # Take input for OLT IP and remove any whitespace

# Validate OLT IP - exit silently if unauthorized
if not check_olt_ip(inolt):
    sys.exit(0)

• The script:
  • Prompts the user to input an OLT IP.
  • Exits silently (sys.exit(0)) if the IP is not authorized.

---

3. Database Connection

Purpose:

Prompts the user for database credentials and attempts to establish a connection.

db_host = input("Enter database host (default: localhost): ").strip() or "localhost"
db_user = input("Enter database username: ").strip()
db_password = input("Enter database password: ").strip()
db_name = "tr069_bsnl_kasrawad"  # Fixed database name, no input needed

• User Input:
  • db_host: Defaults to localhost if no input is provided.
  • db_user and db_password: Required credentials for connecting to the database.

---

Validation and Connection:

if not db_user:
    print("Error: Database username cannot be empty")
    sys.exit(1)

try:
    db = mysql.connector.connect(
        host=db_host,
        user=db_user,
        passwd=db_password,
        database=db_name,
        auth_plugin='mysql_native_password'
    )
    print("Successfully connected to database")
except mysql.connector.Error as e:
    print(f"Error connecting to MySQL: {e}")
    sys.exit(1)

• Validation:
  • Ensures the username is not empty.
• Connection:
  • Attempts to connect to the database using the provided credentials.
  • Exits with an error message if the connection fails.

---

4. Fetching OLT Details

Purpose:

Retrieves details of the OLT from the tr069_tip table based on the provided IP.

mycursor = db.cursor()
try:
    mycursor.execute("SELECT * FROM tr069_tip WHERE OltIp = %s", (inolt,))
    results = mycursor.fetchall()

• SQL Query:
  • Fetches all rows where the OltIp matches the user-provided IP.
• Why?: Identifies the OLT and retrieves its configuration details.

---

5. Telnet Session Management

Purpose:

Establishes a Telnet connection to the OLT and executes commands to retrieve or update its configuration.

Telnet Connection:

tn = telnetlib.Telnet(ip, timeout=3)
tn.read_until(b"Login:", timeout=5)
tn.write(user.encode('ascii') + b"\r")
if password:
    tn.read_until(b"Password:", timeout=10)
    tn.write(password.encode('ascii') + b"\r")

• Connects to the OLT using its IP address and credentials (user and password).

Command Execution:

cmd = "en"
cmd2 = "configure terminal"
cmd3 = "show onu basic-info all"  # Basic info command
tn.write(cmd.encode('ascii') + b"\r")
tn.write(cmd2.encode('ascii') + b"\r")
tn.write(cmd3.encode('ascii') + b"\r")

• Sends commands to the OLT to retrieve basic information about connected ONUs (Optical Network Units).

---

Parsing Telnet Output:

Regex Matching:

totallist = re.findall(r'EPON0/(\d+):(\d+)\s+(\w{4})\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)', line.decode('utf-8'), re.DOTALL)

• Uses regular expressions to extract details such as:
  • Port number.
  • ONU ID.
  • Serial number.
  • Hardware and software versions.

Database Update:

sql2 = "INSERT INTO tr69_new (id, user, pwd, oltip, oltmake, port, onuid, sn, serial, hw, sw) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
values2 = (None, user, password, ip, make, port_number, onuid, sn, serial, hw, sw)
mycursor.execute(sql2, values2)
db.commit()

• Inserts new ONU details into the tr69_new table if they do not already exist.

---

6. WAN and ACS Configuration

Purpose:

Retrieves and updates WAN and ACS configurations for ONUs.

WAN Configuration:

cmd5 = "show onu " + str(current_onuid) + " pri wan_adv"
tn.write(cmd5.encode('ascii') + b"\r")

• Retrieves WAN configuration details for a specific ONU.

ACS Configuration:

cmd9 = "onu " + str(current_onuid) + " pri tr069_mng enable acs_server url http://acs.bsnl.in:7547 username acs password acs@bsnl certificate disable inform enable inform_interval 100"
tn.write(cmd9.encode('ascii') + b"\r")

• Enables ACS management and updates the ACS server URL, username, and password.

Database Update:

sql_upd = "UPDATE `tr69_new` SET `acs_status` = %s, `acs_url` = %s, `acs_interval` = %s WHERE id = %s"
values2 = (acsstatus, acsurl, acsinterval, id)
mycursor.execute(sql_upd, values2)
db.commit()

• Updates the database with the new ACS configuration details.

---

7. Final Output

Purpose:

Displays the updated configuration and status of ONUs.

mycursor.execute("SELECT * FROM tr69_new WHERE OltIp = %s", (ip,))
myresult = mycursor.fetchall()
for row in myresult:
    print(f"ONU ID: {row[6]}, WAN Mode: {row[10]}, ACS Status: {row[30]}")

• Fetches and prints the updated details of ONUs managed by the OLT.

---

Conclusion

This script automates the management of OLTs and ONUs by:

1. Validating OLT IPs.
2. Establishing Telnet sessions to retrieve and update configurations.
3. Storing and managing data in a MySQL database.