Coverage for src/birdplan/plugins/cmdline/bgp/peer/show.py: 22%
138 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-25 07:38 +0000
1#
2# SPDX-License-Identifier: GPL-3.0-or-later
3#
4# Copyright (c) 2019-2025, AllWorldIT
5#
6# This program is free software: you can redistribute it and/or modify
7# it under the terms of the GNU General Public License as published by
8# the Free Software Foundation, either version 3 of the License, or
9# (at your option) any later version.
10#
11# This program is distributed in the hope that it will be useful,
12# but WITHOUT ANY WARRANTY; without even the implied warranty of
13# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14# GNU General Public License for more details.
15#
16# You should have received a copy of the GNU General Public License
17# along with this program. If not, see <http://www.gnu.org/licenses/>.
19"""BirdPlan commandline options for BGP peer show <peer>."""
21import argparse
22import io
23from typing import TYPE_CHECKING, Any
25from .....cmdline import BirdPlanCommandLine, BirdPlanCommandlineResult
26from .....console.colors import colored
27from ...cmdline_plugin import BirdPlanCmdlinePluginBase
29if TYPE_CHECKING:
30 from birdplan import BirdPlanBGPPeerShow
32__all__ = ["BirdPlanCmdlineBGPPeerShowPeerArg"]
35class BirdPlanCmdlineBGPPeerShowPeerArgResult(BirdPlanCommandlineResult):
36 """BirdPlan BGP peer show peer result."""
38 def as_text(self) -> str: # noqa: C901, PLR0915, PLR0912
39 """
40 Return data in text format.
42 Returns
43 -------
44 str
45 Return data in text format.
47 """
49 ob = io.StringIO()
51 # Work out filter strings to use
52 aspath_filters = "none"
53 origin_filters = "none"
54 as_sets_filter = "none"
55 if "import_filter" in self.data:
56 # Check for aspath_asns in our filter
57 aspath_strs = []
58 if "aspath_asns" in self.data["import_filter"]:
59 if "static" in self.data["import_filter"]["aspath_asns"]:
60 count = len(self.data["import_filter"]["aspath_asns"]["static"])
61 aspath_strs.append(f"{count} manual")
62 if "calculated" in self.data["import_filter"]["aspath_asns"]:
63 count = len(self.data["import_filter"]["aspath_asns"]["calculated"])
64 aspath_strs.append(f"{count} calculated")
65 aspath_filters = ", ".join(aspath_strs)
66 # Check for origin filters
67 origin_strs = []
68 if "origin_asns" in self.data["import_filter"]:
69 if "static" in self.data["import_filter"]["origin_asns"]:
70 count = len(self.data["import_filter"]["origin_asns"]["static"])
71 origin_strs.append(f"{count} manual")
72 if "irr" in self.data["import_filter"]["origin_asns"]:
73 count = len(self.data["import_filter"]["origin_asns"]["irr"])
74 origin_strs.append(f"{count} from IRR")
75 origin_filters = ", ".join(origin_strs)
76 # Check for AS-SET filters
77 if "as_sets" in self.data["import_filter"]:
78 if isinstance(self.data["import_filter"]["as_sets"], list):
79 as_sets_filter = ", ".join(self.data["import_filter"]["as_sets"])
80 elif isinstance(self.data["import_filter"]["as_sets"], str):
81 as_sets_filter = self.data["import_filter"]["as_sets"]
83 ob.write(f"ASN.............: {self.data['asn']}\n")
84 ob.write(f"Type............: {self.data['type']}\n")
85 ob.write(f"Name............: {self.data['name']}\n")
86 ob.write(f"Description.....: {self.data['description']}\n")
87 ob.write(f"AS-SET..........: {as_sets_filter}\n")
88 ob.write(f"Origin filters..: {origin_filters}\n")
89 ob.write(f"AS-Path filters.: {aspath_filters}\n")
90 if self.data.get("use_rpki"):
91 ob.write("RPKI ROV........: enabled\n")
93 # Loop with protocols and output self.data
94 for protocol, protocol_data in self.data["protocols"].items():
95 # Work out better protocol string
96 protocol_str = ""
97 if protocol == "ipv4":
98 protocol_str = "IPv4"
99 elif protocol == "ipv6":
100 protocol_str = "IPv6"
102 # Check for import prefix filters
103 prefix_filters = "none"
104 if "import_filter" in self.data: # noqa: SIM102
105 if "prefixes" in self.data["import_filter"]:
106 prefix_filter_strs = []
107 if "irr" in self.data["import_filter"]["prefixes"]: # noqa: SIM102
108 if protocol in self.data["import_filter"]["prefixes"]["irr"]:
109 count = len(self.data["import_filter"]["prefixes"]["irr"][protocol])
110 prefix_filter_strs.append(f"{count} from IRR")
111 if "static" in self.data["import_filter"]["prefixes"]: # noqa: SIM102
112 if protocol in self.data["import_filter"]["prefixes"]["static"]:
113 count = len(self.data["import_filter"]["prefixes"]["static"][protocol])
114 prefix_filter_strs.append(f"{count} manual")
115 prefix_filters = ", ".join(prefix_filter_strs)
117 ob.write(f"\n Protocol: {protocol_str}\n")
119 # Setup o the protocol_status so we can copy-paste the below colors
120 protocol_status = protocol_data["status"]
121 # Set the state and info with no color
122 state = protocol_status["state"]
123 info = protocol_status["info"]
125 # NK - Update in peer_arg show too
126 # Check how we're going to color entries based on their state and info
127 if protocol_status["state"] == "down":
128 state = colored(protocol_status["state"], "red")
129 if "last_error" in protocol_status:
130 info += " - " + colored(protocol_status["last_error"], "red")
131 elif protocol_status["state"] == "up":
132 if protocol_status["info"] == "established":
133 state = colored(protocol_status["state"], "green")
134 info = colored(protocol_status["info"], "green")
136 # Check for quarantine flag
137 quarantined = "no"
138 if self.data.get("quarantine"):
139 quarantined = colored("yes", "red")
141 # Check for graceful shutdown flag
142 graceful_shutdown = "no"
143 if self.data.get("graceful_shutdown"):
144 graceful_shutdown = colored("yes", "red")
146 prefix_limit_str = ""
147 if "prefix_limit" in self.data:
148 if "peeringdb" in self.data["prefix_limit"]:
149 if protocol in self.data["prefix_limit"]["peeringdb"]:
150 prefix_limit_str = " from PeeringDB"
151 elif "static" in self.data["prefix_limit"]: # noqa: SIM102
152 if protocol in self.data["prefix_limit"]["static"]:
153 prefix_limit_str = " manual"
155 ob.write(f" Mode..............: {protocol_data['mode']}\n")
156 ob.write(f" State.............: {state} ({info}) since {protocol_status['since']}\n")
157 ob.write(f" Local AS..........: {protocol_status['local_as']}\n")
159 # Work out our source address, depending if peer is up or not
160 source_address = protocol_status.get("source_address", protocol_data["source_address"])
161 ob.write(f" Source IP.........: {source_address}\n")
163 ob.write(f" Neighbor AS.......: {protocol_status['neighbor_as']}\n")
164 ob.write(f" Neighbor IP.......: {protocol_status['neighbor_address']}\n")
166 if "neighbor_id" in protocol_status:
167 ob.write(f" Neighbor ID.......: {protocol_status['neighbor_id']}\n")
169 # Check if we have an import limit
170 if "import_limit" in protocol_status:
171 import_limit = protocol_status["import_limit"]
172 import_limit_action = protocol_status["import_limit_action"]
173 ob.write(f" Import limit......: {import_limit}{prefix_limit_str} (action: {import_limit_action})\n")
174 else:
175 ob.write(" Import limit......: none\n")
177 ob.write(f" Prefix filters....: {prefix_filters}\n")
179 # Check if we have route information
180 if "routes_imported" in protocol_status and "routes_exported" in protocol_status:
181 routes_imported = protocol_status["routes_imported"]
182 routes_exported = protocol_status["routes_exported"]
183 ob.write(f" Prefixes..........: {routes_imported} imported, {routes_exported} exported\n")
185 if self.data.get("security"):
186 ob.write(f" BGP security......: {', '.join(sorted(self.data['security']))}\n")
188 ob.write(f" Quarantined.......: {quarantined}\n")
189 ob.write(f" Graceful shutdown.: {graceful_shutdown}\n")
191 ob.write("\n")
193 ob.write("\n")
195 return ob.getvalue()
198class BirdPlanCmdlineBGPPeerShowPeerArg(BirdPlanCmdlinePluginBase):
199 """BirdPlan "bgp peer show <peer>" command."""
201 def __init__(self) -> None:
202 """Initialize object."""
204 super().__init__()
206 # Plugin setup
207 self.plugin_description = "birdplan bgp peer show <peer>"
208 self.plugin_order = 30
210 def register_parsers(self, args: dict[str, Any]) -> None:
211 """
212 Register commandline parsers.
214 Parameters
215 ----------
216 args : Dict[str, Any]
217 Method argument(s).
219 """
221 plugins = args["plugins"]
223 parent_subparsers = plugins.call_plugin("birdplan.plugins.cmdline.bgp.peer", "get_subparsers", {})
225 # CMD: bgp peer show <peer>
226 subparser = parent_subparsers.add_parser("show", help="BGP peer show commands")
228 subparser.add_argument(
229 "--action",
230 action="store_const",
231 const="bgp_peer_show",
232 default="bgp_peer_show",
233 help=argparse.SUPPRESS,
234 )
236 subparser.add_argument(
237 "peer",
238 nargs=1,
239 metavar="PEER",
240 help="Peer to show (its BirdPlan name)",
241 )
243 # Set our internal subparser property
244 self._subparser = subparser
245 self._subparsers = None
247 def cmd_bgp_peer_show(self, args: dict[str, Any]) -> BirdPlanCmdlineBGPPeerShowPeerArgResult:
248 """
249 Commandline handler for "bgp peer show <peer>" action.
251 Parameters
252 ----------
253 args : Dict[str, Any]
254 Method argument(s).
256 """
258 if not self._subparser: # pragma: no cover
259 raise RuntimeError
261 cmdline: BirdPlanCommandLine = args["cmdline"]
263 # Grab Bird control socket
264 bird_socket = cmdline.args.bird_socket[0]
266 # Grab the peer
267 peer = cmdline.args.peer[0]
269 # Suppress info output
270 cmdline.birdplan.birdconf.birdconfig_globals.suppress_info = True
272 # Load BirdPlan configuration using the cache
273 cmdline.birdplan_load_config(ignore_irr_changes=True, ignore_peeringdb_changes=True, use_cached=True)
275 # Try grab peer info
276 res: BirdPlanBGPPeerShow = cmdline.birdplan.state_bgp_peer_show(peer, bird_socket=bird_socket)
278 return BirdPlanCmdlineBGPPeerShowPeerArgResult(res)