Skip to main content
Loading...

More Python Posts

import subprocess
import json
import re


def parse_bgp_routes(bgp_output):
    """
    Parse BGP route table output and return structured data.
    
    Args:
        bgp_output (str): Raw BGP table output
        
    Returns:
        dict: Structured BGP routes data
    """
    routes = []
    
    # Split into lines and find the route entries
    lines = bgp_output.strip().split('\n')
    
    # Find the header line to identify where routes start
    route_start_idx = None
    for i, line in enumerate(lines):
        if 'Network' in line and 'Next Hop' in line:
            route_start_idx = i + 1
            break
    
    if route_start_idx is None:
        return {"routes": routes}
    
    # Parse each route line
    for line in lines[route_start_idx:]:
        line = line.strip()
        
        # Skip empty lines and summary lines
        if not line or line.startswith('Total number') or line.startswith('IPv6'):
            continue
            
        # Skip lines that don't start with route status indicators
        if not line.startswith('*>'):
            continue
        
        # Parse the route line using regex
        # Pattern matches: *> network next_hop metric [locprf] weight path
        # Note: LocPrf column may be empty, so we'll set it to 0 for all records
        pattern = r'^\*>\s+(\S+)\s+(\S+)\s+(\d+)\s+(\d+)\s+(.+)$'
        match = re.match(pattern, line)
        
        if match:
            network = match.group(1)
            next_hop = match.group(2)
            metric = int(match.group(3))
            loc_prf = 0  # Set to 0 for all records as requested
            weight = int(match.group(4))
            path_info = match.group(5).strip()
            
            # Extract AS path (remove origin code)
            path_parts = path_info.split()
            as_path = []
            for part in path_parts:
                if part.isdigit():
                    as_path.append(part)
            
            path = ' '.join(as_path) if as_path else ""
            
            route = {
                "network": network,
                "nextHop": next_hop,
                "metric": metric,
                "locPrf": loc_prf,
                "weight": weight,
                "path": path
            }
            routes.append(route)
    
    return {"routes": routes}


def get_bgp_routes_json():
    """
    Get BGP routes from sample data and return as JSON.
    
    Returns:
        str: JSON string of BGP routes
    """
    output_text = """IPv4
BGP table version is 0, local router ID is 169.254.0.185
Status codes: s suppressed, d damped, h history, * valid, > best, i - internal,
            r RIB-failure, S Stale, R Removed
Origin codes: i - IGP, e - EGP, ? - incomplete

Network          Next Hop            Metric LocPrf Weight Path
*> 10.0.0.0         169.254.0.186          100             0 65100 ?
*> 10.2.0.0/16      169.254.0.185          100         32768 i
*> 10.5.0.0/16      169.254.0.185          100         32768 i
*> 10.6.0.0/16      169.254.0.185          100         32768 i
*> 10.10.0.0/16     169.254.0.185          100         32768 i
*> 10.42.0.0/16     169.254.0.185          100         32768 i
*> 10.56.0.0/16     169.254.0.185          100         32768 i
*> 10.133.0.0/21    169.254.0.185          100         32768 i
*> 10.192.2.0/23    169.254.0.186          100             0 65100 ?
*> 10.192.6.0/23    169.254.0.186          100             0 65100 ?
*> 10.208.0.0/18    169.254.0.185          100         32768 i
*> 10.210.0.0/20    169.254.0.186          100             0 65100 ?
*> 10.210.64.0/18   169.254.0.186          100             0 65100 ?
*> 10.210.128.0/20  169.254.0.186          100             0 65100 ?
*> 10.211.0.0/17    169.254.0.186          100             0 65100 ?
*> 10.211.128.0/17  169.254.0.186          100             0 65100 ?
*> 10.216.0.0/16    169.254.0.186          100             0 65100 ?
*> 10.224.64.0/19   169.254.0.185          100         32768 i
*> 10.227.128.0/18  169.254.0.185          100         32768 i
*> 10.227.192.0/18  169.254.0.185          100         32768 i
*> 10.228.64.0/18   169.254.0.185          100         32768 i
*> 10.228.128.0/18  169.254.0.185          100         32768 i
*> 10.228.192.0/18  169.254.0.185          100         32768 i
*> 10.229.64.0/18   169.254.0.185          100         32768 i
*> 10.229.128.0/18  169.254.0.185          100         32768 i
*> 10.229.192.0/18  169.254.0.185          100         32768 i
*> 10.230.0.0/18    169.254.0.185          100         32768 i
*> 10.230.64.0/18   169.254.0.185          100         32768 i
*> 10.230.128.0/18  169.254.0.185          100         32768 i
*> 10.230.192.0/18  169.254.0.185          100         32768 i
*> 10.246.48.0/20   169.254.0.185          100         32768 i
*> 10.254.192.0/18  169.254.0.185          100         32768 i
*> 10.255.0.0/18    169.254.0.185          100         32768 i
*> 10.255.64.0/18   169.254.0.185          100         32768 i
*> 10.255.128.0/18  169.254.0.185          100         32768 i
*> 10.255.192.0/19  169.254.0.185          100         32768 i
*> 10.255.224.0/19  169.254.0.185          100         32768 i
*> 100.64.0.64/27   169.254.0.185          100         32768 i
*> 100.64.0.96/27   169.254.0.185          100         32768 i
*> 100.64.0.128/27  169.254.0.185          100         32768 i
*> 100.64.0.160/27  169.254.0.185          100         32768 i
*> 172.16.0.0/12    169.254.0.186          100             0 65100 69999 21222 ?
*> 240.0.0.0/4      169.254.0.186          100             0 65100 69999 21222 ?
*> 253.128.0.0/14   169.254.0.186          100             0 65100 ?
*> 253.132.0.0/20   169.254.0.186          100             0 65100 ?

Total number of prefixes 45

IPv6
No BGP network exists"""
    
    # Parse the BGP output and return as JSON
    parsed_data = parse_bgp_routes(output_text)
    return json.dumps(parsed_data, indent=2)

def get_specific_network(prefix):
    # Get all BGP routes first
    all_routes = json.loads(get_bgp_routes_json())
    
    # Search through routes for matching prefix
    for route in all_routes["routes"]:
        if route["network"] == prefix:
            return route
            
    # Return None if no matching prefix is found
    return None

def main():
    """Main function to demonstrate the BGP parser."""
    print(get_bgp_routes_json())
    print(get_specific_network("10.42.0.0/16"))


if __name__ == "__main__":
    main()
import re

def _map_ikev2_vendor_capabilities(message_type, input_string):
    # List of acceptable message types
    valid_message_types = ['IKE_SA_INIT', 'IKE_AUTH']
    
    # Validate message type
    if message_type not in valid_message_types:
        raise ValueError(f"Invalid message type: {message_type}. Must be one of {valid_message_types}")
    
    # Mapping dictionary for IKE values with RFC references
    value_map = {
        # IKE_SA_INIT capabilities
        'FRAG_SUP': 'IKE Fragmentation',  # RFC 7383, Section 3
        'REDIR_SUP': 'Redirection',  # RFC 5685, Section 3
        'HASH_ALG': 'Hash Algorithms',  # RFC 7296, Section 3.3.2
        'NATD_S_IP': 'NAT-T (Source IP)',  # RFC 7296, Section 2.23
        'NATD_D_IP': 'NAT-T (Destination IP)',  # RFC 7296, Section 2.23
        'SIGN_HASH_ALGS': 'Signature Hash Algorithms',  # RFC 7296, Section 2.15
        'NON_FIRST_FRAGMENTS': 'Non-First IKE Fragments',  # RFC 7383, Section 3
        'CHILDLESS_IKEV2_SUP': 'Childless IKEv2',  # RFC 6023, Section 3
        'INTERMEDIATE': 'Intermediate Exchange',  # RFC 9242, Section 3
        'COOKIE': 'Cookie-Based DoS Protection',  # RFC 7296, Section 2.6
        # IKE_AUTH capabilities
        'ESP_TFC_PAD_N': 'ESPv3 TFC Padding Not Supported',  # RFC 7296, Section 3.3.1
        'MOBIKE_SUP': 'MOBIKE',  # RFC 4555, Section 3
        'MULT_AUTH': 'Multiple Auth',  # RFC 4739, Section 3
        'EAP_ONLY': 'EAP-Only Auth',  # RFC 5998, Section 3
        'MSG_ID_SYN_SUP': 'Message ID Sync',  # RFC 6311, Section 3
        'IPCOMP_SUPPORTED': 'IP Payload Compression Support',  # RFC 7296, Section 3.3.2
        'ADD_4_ADDR': 'Additional IPv4 Addresses',  # RFC 4555, Section 3.2
        'ADD_6_ADDR': 'Additional IPv6 Addresses',  # RFC 4555, Section 3.2
        'INIT_CONTACT': 'Initial Contact',  # RFC 7296, Section 3.16
        'HTTP_CERT_LOOKUP_SUP': 'HTTP Certificate Lookup',  # RFC 7296, Section 3.7
        'REKEY_SA': 'SA Rekeying'  # RFC 7296, Section 3.16
    }
    
    # Notifications valid for each message type
    ike_sa_init_valid = {
        'FRAG_SUP', 'REDIR_SUP', 'HASH_ALG', 'NATD_S_IP', 'NATD_D_IP',
        'SIGN_HASH_ALGS', 'NON_FIRST_FRAGMENTS', 'CHILDLESS_IKEV2_SUP',
        'INTERMEDIATE', 'COOKIE'
    }
    ike_auth_valid = {
        'ESP_TFC_PAD_N', 'MOBIKE_SUP', 'MULT_AUTH', 'EAP_ONLY', 'MSG_ID_SYN_SUP',
        'IPCOMP_SUPPORTED', 'ADD_4_ADDR', 'ADD_6_ADDR', 'INIT_CONTACT',
        'HTTP_CERT_LOOKUP_SUP', 'REKEY_SA'
    }
    
    # Select valid notifications based on message type
    valid_notifications = ike_sa_init_valid if message_type == 'IKE_SA_INIT' else ike_auth_valid
    
    # Regex to capture N(...) patterns
    pattern = r'N\([^)]+\)'
    matches = re.findall(pattern, input_string)
    
    # Parse matches and create result list
    result = []
    for match in matches:
        key = match[2:-1]  # Extract content inside N(...)
        if key in valid_notifications and key in value_map:
            result.append(value_map[key])
        else:
            # Log unrecognized notifications for debugging
            print(f"Warning: Unrecognized or invalid notification for {message_type}: {key}")
    
    return ', '.join(result) if result else 'None'

# Example usage
ike_sa_init_string = '2025-07-18 20:16:22.839 15[ENC] <4> parsed IKE_SA_INIT request 0 [ SA KE No N(NATD_S_IP) N(NATD_D_IP) N(FRAG_SUP) N(HASH_ALG) N(REDIR_SUP) N(SIGN_HASH_ALGS) N(NON_FIRST_FRAGMENTS) N(CHILDLESS_IKEV2_SUP) N(INTERMEDIATE) N(COOKIE) ]'
ike_auth_string = '2025-07-18 20:16:22.898 07[ENC] <4> parsed IKE_AUTH request 1 [ IDi N(INIT_CONTACT) IDr AUTH N(ESP_TFC_PAD_N) SA TSi TSr N(MOBIKE_SUP) N(ADD_4_ADDR) N(MULT_AUTH) N(EAP_ONLY) N(MSG_ID_SYN_SUP) N(IPCOMP_SUPPORTED) N(ADD_6_ADDR) N(HTTP_CERT_LOOKUP_SUP) N(REKEY_SA) ]'

# Test with IKE_SA_INIT
print("IKE_SA_INIT Results:")
print(_map_ikev2_vendor_capabilities("IKE_SA_INIT", ike_sa_init_string))

# Test with IKE_AUTH
print("\nIKE_AUTH Results:")
print(_map_ikev2_vendor_capabilities("IKE_AUTH", ike_auth_string))