mirror/checktheroads-hyperglass
1
0
Fork 0
mirror of https://github.com/checktheroads/hyperglass synced 2024-05-11 05:55:08 +00:00
Code Releases Activity

all the things

Browse Source
This commit is contained in:
checktheroads
2019-06-10 12:22:38 -07:00
parent 9e94429bb6
commit eca889e866
42 changed files with 2010 additions and 968 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
hyperglass/command/construct.py
View File

@ -3,9 +3,14 @@
Accepts filtered & validated input from execute.py, constructs SSH command for Netmiko library or \
API call parameters for hyperglass-frr
"""
# Module Imports
# Standard Imports
import json
import inspect
import logging
# Module Imports
import logzero
from logzero import logger
from netaddr import IPNetwork, IPAddress # pylint: disable=unused-import
# Dear PyLint, the netaddr library is a special snowflake. You might not see `IPAddress` get used, \
@ -17,7 +22,12 @@ from hyperglass import configuration
# Configuration Imports
codes = configuration.codes()
config = configuration.general()
# Logzero Configuration
if configuration.debug_state():
logzero.loglevel(logging.DEBUG)
else:
logzero.loglevel(logging.INFO)
def current_function():
@ -45,79 +55,90 @@ class Construct:
src = self.d_src_addr_ipv4
if ver == 6:
src = self.d_src_addr_ipv6
logger.debug(f"Source IPv{ver}: {src}")
return src
def ping(self, transport, target):
"""Constructs ping query parameters from pre-validated input"""
cmd = current_function()
query_type = current_function()
logger.debug(f"Constructing {query_type} query for {target} via {transport}...")
query = None
ip_version = IPNetwork(target).ip.version
afi = f"ipv{ip_version}"
source = self.get_src(ip_version)
if transport == "rest":
query = json.dumps(
{"cmd": cmd, "afi": afi, "source": source, "target": target}
{"cmd": query_type, "afi": afi, "source": source, "target": target}
)
if transport == "scrape":
conf_command = self.command[afi][cmd]
conf_command = self.command[afi][query_type]
fmt_command = conf_command.format(target=target, source=source)
query = (self.d_address, self.d_type, fmt_command)
logger.debug(f"Constructed query: {query}")
return query
def traceroute(self, transport, target):
"""Constructs traceroute query parameters from pre-validated input"""
cmd = current_function()
query_type = current_function()
logger.debug(f"Constructing {query_type} query for {target} via {transport}...")
query = None
ip_version = IPNetwork(target).ip.version
afi = f"ipv{ip_version}"
source = self.get_src(ip_version)
if transport == "rest":
query = json.dumps(
{"cmd": cmd, "afi": afi, "source": source, "target": target}
{"cmd": query_type, "afi": afi, "source": source, "target": target}
)
if transport == "scrape":
conf_command = self.command[afi][cmd]
conf_command = self.command[afi][query_type]
fmt_command = conf_command.format(target=target, source=source)
query = (self.d_address, self.d_type, fmt_command)
logger.debug(f"Constructed query: {query}")
return query
def bgp_route(self, transport, target):
"""Constructs bgp_route query parameters from pre-validated input"""
cmd = current_function()
query_type = current_function()
logger.debug(f"Constructing {query_type} query for {target} via {transport}...")
query = None
ip_version = IPNetwork(target).ip.version
afi = f"ipv{ip_version}"
if transport == "rest":
query = json.dumps({"cmd": cmd, "afi": afi, "target": target})
query = json.dumps({"cmd": query_type, "afi": afi, "target": target})
if transport == "scrape":
conf_command = self.command[afi][cmd]
conf_command = self.command[afi][query_type]
fmt_command = conf_command.format(target=target)
query = (self.d_address, self.d_type, fmt_command)
logger.debug(f"Constructed query: {query}")
return query
def bgp_community(self, transport, target):
"""Constructs bgp_community query parameters from pre-validated input"""
cmd = current_function()
query_type = current_function()
logger.debug(f"Constructing {query_type} query for {target} via {transport}...")
afi = "dual"
query = None
if transport == "rest":
query = json.dumps({"cmd": cmd, "afi": afi, "target": target})
query = json.dumps({"cmd": query_type, "afi": afi, "target": target})
if transport == "scrape":
conf_command = self.command[afi][cmd]
conf_command = self.command[afi][query_type]
fmt_command = conf_command.format(target=target)
query = (self.d_address, self.d_type, fmt_command)
logger.debug(f"Constructed query: {query}")
return query
def bgp_aspath(self, transport, target):
"""Constructs bgp_aspath query parameters from pre-validated input"""
cmd = current_function()
query_type = current_function()
logger.debug(f"Constructing {query_type} query for {target} via {transport}...")
afi = "dual"
query = None
if transport == "rest":
query = json.dumps({"cmd": cmd, "afi": afi, "target": target})
query = json.dumps({"cmd": query_type, "afi": afi, "target": target})
if transport == "scrape":
conf_command = self.command[afi][cmd]
conf_command = self.command[afi][query_type]
fmt_command = conf_command.format(target=target)
query = (self.d_address, self.d_type, fmt_command)
logger.debug(f"Constructed query: {query}")
return query