add Huawei bgp route structured output

docs/pages/configuration/config/structured-output.mdx

@@ -4,6 +4,7 @@ Devices that support responding to a query with structured or easily parsable da
 
 -   Arista EOS
 -   Juniper Junos
+-   Huawei VRP
 
 When structured output is available, hyperglass checks the RPKI state of each BGP prefix returned using one of two methods:
 

hyperglass/constants.py

@@ -19,7 +19,7 @@
 
 TARGET_JUNIPER_ASPATH = ("juniper", "juniper_junos")
 
-SUPPORTED_STRUCTURED_OUTPUT = ("frr", "juniper", "arista_eos")
+SUPPORTED_STRUCTURED_OUTPUT = ("frr", "juniper", "arista_eos", "huawei", "huawei_vrpv8")
 
 CONFIG_EXTENSIONS = ("py", "yaml", "yml", "json", "toml")
 

hyperglass/defaults/directives/huawei.py

@@ -15,6 +15,7 @@
     "Huawei_BGPRoute",
     "Huawei_Ping",
     "Huawei_Traceroute",
+    "Huawei_BGPRouteTable",
 )
 
 NAME = "Huawei VRP"
@@ -37,6 +38,7 @@
     ],
     field=Text(description="IP Address, Prefix, or Hostname"),
     plugins=["bgp_route_huawei"],
+    table_output="__hyperglass_huawei_bgp_route_table__",
     platforms=PLATFORMS,
 )
 
@@ -111,3 +113,24 @@
     field=Text(description="IP Address, Prefix, or Hostname"),
     platforms=PLATFORMS,
 )
+
+# Table Output Directives
+
+Huawei_BGPRouteTable = BuiltinDirective(
+    id="__hyperglass_huawei_bgp_route_table__",
+    name="BGP Route",
+    rules=[
+        RuleWithIPv4(
+            condition="0.0.0.0/0",
+            action="permit",
+            command="display bgp routing-table {target} | no-more",
+        ),
+        RuleWithIPv6(
+            condition="::/0",
+            action="permit",
+            command="display bgp ipv6 routing-table {target} | no-more",
+        ),
+    ],
+    field=Text(description="IP Address, Prefix, or Hostname"),
+    platforms=PLATFORMS,
+)

hyperglass/models/parsing/huawei.py

@@ -0,0 +1,146 @@
+"""Data Models for Parsing Huawei Response."""
+
+# Standard Library
+import re
+import typing as t
+
+# Third Party
+from pydantic import ConfigDict
+
+# Project
+from hyperglass.log import log
+from hyperglass.models.data import BGPRouteTable
+
+# Local
+from ..main import HyperglassModel
+
+RPKI_STATE_MAP = {
+    "invalid": 0,
+    "valid": 1,
+    "not-found": 2,
+}
+
+WINNING_WEIGHT = "high"
+
+HuaweiPathType = t.Literal["external", "internal", "local"]
+
+HuaweiRPKIState = t.Literal["valid", "invalid", "not-found"]
+
+
+class _HuaweiBase(HyperglassModel):
+    """Base Model for Huawei validation."""
+
+    model_config = ConfigDict(extra="ignore")
+
+
+class HuaweiRoutePath(_HuaweiBase):
+    """Validation model for Huawei BGP route path."""
+
+    prefix: str
+    peer_address: str
+    peer_rid: str
+    duration: str
+    next_hop: str
+    relay_next_hop: t.Optional[str]
+    out_intf: str
+    qos_info: str
+    communities: t.Optional[str]
+    large_communities: t.Optional[str]
+    ext_communities: t.Optional[str]
+    as_path: str
+    origin: str
+    med: t.Optional[int]
+    local_preference: t.Optional[int]
+    preference_value: int
+    path_type: HuaweiPathType
+    is_valid: bool
+    is_best: bool
+    is_preferred: bool
+    route_preference: int
+    rpki_state: t.Optional[HuaweiRPKIState]
+
+
+class HuaweiBGPTable(_HuaweiBase):
+    """Validation model for Huawei BGP routing data."""
+
+    router_id: str = ""
+    local_asn: int = 0
+    vrf: str = "default"
+    paths_num_total: int = 0
+    paths_num_best: int = 0
+    paths_num_selected: int = 0
+    paths_num_best_external: int = 0
+    paths_num_add_path: int = 0
+    bgp_path_entries: list[HuaweiRoutePath] = []
+
+    @staticmethod
+    def _get_route_age(duration: str) -> int:
+        """Parse duration time as sting and return integer."""
+        regex = re.compile(r"(\d+)d(\d+)h(\d+)m(\d+)s")
+        match = regex.search(duration)
+        if match:
+            days, hours, minutes, seconds = [int(n or 0) for n in match.groups()]
+            return days * 24 * 60 * 60 + hours * 60 * 60 + minutes * 60 + seconds
+        return 0
+
+    @staticmethod
+    def _get_as_path(as_path: str) -> list[int]:
+        """Convert AS-path string to list of ASNs."""
+        if as_path == "Nil":
+            return []
+        return [int(asn) for asn in as_path.split() if asn.isdecimal()]
+
+    @staticmethod
+    def _get_communities(community: str | None) -> list[str]:
+        """Convert community string to list of communities."""
+        if isinstance(community, str):
+            return [c.strip("<>") for c in community.split(", ")]
+        return []
+
+    def bgp_table(self: "HuaweiBGPTable") -> BGPRouteTable:
+        """Convert the Huawei-formatted fields to standard parsed data model."""
+        routes = []
+        for route in self.bgp_path_entries:
+            as_path = self._get_as_path(as_path=route.as_path)
+            rpki_state = RPKI_STATE_MAP.get(route.rpki_state, 3)
+            communities = []
+            if route.communities:
+                communities += self._get_communities(route.communities)
+            if route.large_communities:
+                communities += self._get_communities(route.large_communities)
+            if route.ext_communities:
+                communities += self._get_communities(route.ext_communities)
+
+            # iBGP paths contain string "Nil". If the AS_PATH is "Nil", we
+            # set the source_as to the router's local-as.
+            source_as = self.local_asn
+            if len(as_path) != 0:
+                source_as = as_path[-1]
+
+            routes.append(
+                {
+                    "prefix": route.prefix,
+                    "active": route.is_best,
+                    "age": self._get_route_age(route.duration),
+                    "weight": route.preference_value,
+                    "med": route.med or 0,
+                    "local_preference": route.local_preference or 100,
+                    "as_path": as_path,
+                    "communities": communities,
+                    "next_hop": route.next_hop,
+                    "source_as": source_as,
+                    "source_rid": route.peer_rid,
+                    "peer_rid": route.peer_address,
+                    "rpki_state": rpki_state,
+                }
+            )
+
+        serialized = BGPRouteTable(
+            vrf=self.vrf,
+            count=self.paths_num_total,
+            routes=routes,
+            winning_weight=WINNING_WEIGHT,
+        )
+
+        log.bind(platform="huawei", response=repr(serialized)).debug("Serialized response")
+        return serialized

hyperglass/plugins/_builtin/__init__.py

@@ -4,7 +4,7 @@
 from .bgp_route_frr import BGPRoutePluginFrr
 from .remove_command import RemoveCommand
 from .bgp_route_arista import BGPRoutePluginArista
-from .bgp_route_huawei import BGPRoutePluginHuawei
+from .bgp_route_huawei import BGPRoutePluginHuawei, BGPRouteTablePluginHuawei
 from .bgp_route_juniper import BGPRoutePluginJuniper
 from .mikrotik_garbage_output import MikrotikGarbageOutput
 
@@ -13,6 +13,7 @@
     "BGPRoutePluginFrr",
     "BGPRoutePluginJuniper",
     "BGPRoutePluginHuawei",
+    "BGPRouteTablePluginHuawei",
     "MikrotikGarbageOutput",
     "RemoveCommand",
 )

hyperglass/plugins/_builtin/bgp_route_huawei.py

@@ -1,27 +1,138 @@
+"""Huawei buildin plugins."""
+
 # Standard Library
+import re
 import typing as t
 from ipaddress import ip_network
 
 # Third Party
-from pydantic import PrivateAttr
+from pydantic import PrivateAttr, ValidationError
+
+# Project
+from hyperglass.log import log
+from hyperglass.exceptions.private import ParsingError
+from hyperglass.models.parsing.huawei import HuaweiBGPTable
 
 # Local
 from .._input import InputPlugin
+from .._output import OutputPlugin
 
 if t.TYPE_CHECKING:
     # Project
     from hyperglass.models.api.query import Query
+    from hyperglass.models.data import OutputDataModel
+
+    # Local
+    from .._output import OutputType
 
 InputPluginTransformReturn = t.Union[t.Sequence[str], str]
 
 
+HUAWEI_BGP_HEADER_REGEX = (
+    r" BGP local router ID : (?P<router_id>\S+)\n"
+    r" Local AS number : (?P<local_asn>\d+)\n.*?"
+    r" Paths:   (?P<paths_num_total>\d+) available, (?P<paths_num_best>\d+) best,"
+    r" (?P<paths_num_selected>\d+) select, (?P<paths_num_best_external>\d+) best-external,"
+    r" (?P<paths_num_add_path>\d+) add-path"
+)
+HUAWEI_BGP_PATH_REGEX = (
+    r"BGP routing table entry information of (?P<prefix>\S+):.+?"
+    r"From: (?P<peer_address>\S+) \((?P<peer_rid>\S+)\).+?"
+    r"Route Duration: (?P<duration>\S+).+?"
+    r"(?: Relay IP Nexthop: (?P<relay_next_hop>\S+))?.+?"
+    r"(?:Relay IP|Direct) Out-[Ii]nterface: (?P<out_intf>\S+).+?"
+    r"Original nexthop: (?P<next_hop>\S+).+?"
+    r"Qos information : (?P<qos_info>\S+).+?"
+    r"(?: Community: (?P<communities>[<>\d:, ]+)\n)?"
+    r"(?: Large-Community: (?P<large_communities>[<>\d:, ]+)\n)?"
+    r"(?: Ext-Community: (?P<ext_communities>[\S ]+)\n)?"
+    r" AS-path (?P<as_path>[\w ]+), origin (?P<origin>\S+), "
+    r"(?:MED (?P<med>\d+), )?(?:localpref (?P<local_preference>\d+), )?"
+    r"pref-val (?P<preference_value>\d+), (?:(?P<is_valid>valid), )?"
+    r"(?P<path_type>(?:external|internal|local)), (?:(?P<is_best>best), )?"
+    r"(?:(?P<is_preferred>select), )?pre (?P<route_preference>\d+).*?"
+    r"(?:, validation (?P<rpki_state>(?:valid|invalid|not-found)))?.*?"
+)
+
+
+def parse_bgp_route_header_huawei(output: str) -> dict[str, str | None] | None:
+    """Parse Huawei BGP routing-table header."""
+    header = output[: output.index("\n BGP routing table")]
+    regex = re.compile(HUAWEI_BGP_HEADER_REGEX, flags=re.DOTALL)
+    match_iter = regex.finditer(header)
+    for match in match_iter:
+        return match.groupdict()
+    return None
+
+
+def parse_bgp_path_entries_huawei(output: str) -> list[dict[str, str | None]]:
+    """Parse Huawei BGP routing-table entries."""
+    routes: list[dict[str, str | None]] = []
+    regex = re.compile(HUAWEI_BGP_PATH_REGEX, flags=re.DOTALL)
+    match_iter = regex.finditer(output)
+    for match in match_iter:
+        route = match.groupdict()
+        route["is_valid"] = True if route["is_valid"] == "valid" else False
+        route["is_best"] = True if route["is_best"] == "best" else False
+        route["is_preferred"] = True if route["is_preferred"] == "select" else False
+        routes.append(route)
+    return routes
+
+
+def parse_bgp_route_output_huawei(response: str):
+    """Parse Huawei BGP routing-table response."""
+    if "network does not exist" in response:
+        return {}
+
+    header = parse_bgp_route_header_huawei(output=response)
+    if not header:
+        raise ParsingError("Error parsing response data")
+
+    paths = parse_bgp_path_entries_huawei(output=response)
+    return {**header, "bgp_path_entries": paths}
+
+
+def parse_huawei(output: t.Sequence[str]) -> "OutputDataModel":
+    """Parse a Arista BGP JSON response."""
+    result = None
+
+    _log = log.bind(plugin=BGPRouteTablePluginHuawei.__name__)
+
+    for response in output:
+        try:
+            parsed: t.Dict = parse_bgp_route_output_huawei(response)
+
+            _log.debug("Parsed data", data=parsed)
+
+            validated = HuaweiBGPTable(**parsed)
+            bgp_table = validated.bgp_table()
+
+            if result is None:
+                result = bgp_table
+            else:
+                result += bgp_table
+
+        except ValueError as err:
+            _log.critical(err)
+            raise ParsingError("Error parsing response data") from err
+
+        except ValidationError as err:
+            _log.critical(err)
+            raise ParsingError(err.errors()) from err
+
+    return result
+
+
 class BGPRoutePluginHuawei(InputPlugin):
     _hyperglass_builtin: bool = PrivateAttr(True)
     platforms: t.Sequence[str] = (
         "huawei",
         "huawei_vrpv8",
     )
-    directives: t.Sequence[str] = ("__hyperglass_huawei_bgp_route__",)
+    directives: t.Sequence[str] = (
+        "__hyperglass_huawei_bgp_route__",
+        "__hyperglass_huawei_bgp_route_table__",
+    )
     """
     Huawei BGP Route Input Plugin
 
@@ -45,3 +156,28 @@ def transform(self, query: "Query") -> InputPluginTransformReturn:
         target_network = ip_network(target)
 
         return f"{target_network.network_address!s} {target_network.prefixlen!s}"
+
+
+class BGPRouteTablePluginHuawei(OutputPlugin):
+    """Coerce a Huawei route table string output to a standard BGP Table structure."""
+
+    _hyperglass_builtin: bool = PrivateAttr(True)
+    platforms: t.Sequence[str] = (
+        "huawei",
+        "huawei_vrpv8",
+    )
+    directives: t.Sequence[str] = ("__hyperglass_huawei_bgp_route_table__",)
+
+    def process(self, *, output: "OutputType", query: "Query") -> "OutputType":
+        """Parse Huawei BGP routing-table response if data is a string."""
+        should_process = all(
+            (
+                isinstance(output, (list, tuple)),
+                query.device.platform in self.platforms,
+                query.device.structured_output is True,
+                query.device.has_directives(*self.directives),
+            )
+        )
+        if should_process:
+            return parse_huawei(output)
+        return output