-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathipv6.py
More file actions
1525 lines (1271 loc) · 61.3 KB
/
Copy pathipv6.py
File metadata and controls
1525 lines (1271 loc) · 61.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
IPv6-related formatters.
This module contains formatters for IPv6 commands:
- IPv6BgpNeighborsFormatter: show ipv6 bgp neighbors
- IPv6BgpNetworkFormatter: show ipv6 bgp network (passthrough)
- IPv6BgpSummaryFormatter: show ipv6 bgp summary
- IPv6FibFormatter: show ipv6 fib
- IPv6InterfacesFormatter: show ipv6 interfaces
- IPv6LinkLocalModeFormatter: show ipv6 link-local-mode
- IPv6PrefixListFormatter: show ipv6 prefix-list
- IPv6ProtocolFormatter: show ipv6 protocol
"""
from typing import Dict, List, Optional, Union
from tabulate import tabulate
from gnmi_cli_lib.formatters.base import OutputFormatter
from gnmi_cli_lib.common.types import CommandConfig
from gnmi_cli_lib.utils.sorting import natural_sort_key
class IPv6BgpNeighborsFormatter(OutputFormatter):
"""
Formats IPv6 BGP neighbors output in detailed multi-line format.
Each neighbor gets a detailed block with all properties.
Example output:
BGP neighbor is fc00::2, remote AS 4200300000, local AS 4200200000, confed-external link
Local Role: undefined
Remote Role: undefined
Description: ARISTA01FT2
Member of peer-group FABRICSPINE_V6 for session parameters
BGP version 4, remote router ID 100.1.0.1, local router ID 10.1.0.32
...
"""
# View types that require route table format instead of neighbor details
ROUTE_VIEW_TYPES = {"advertised-routes", "received-routes", "routes"}
def format(
self,
data: Union[Dict, List],
config: CommandConfig,
table_format: str = "simple",
argument: str = "",
options: Optional[Dict] = None,
) -> str:
"""Format BGP neighbors as detailed multi-line output.
Detects view type from the data structure:
- If data contains 'advertisedRoutes', 'receivedRoutes', 'routes',
'bgpTableVersion', or 'tableVersion' keys, delegate to route table formatter
- Otherwise format as neighbor details
"""
if not data:
return ""
# Check if this is a routes view by looking at the data structure
# advertised-routes/received-routes use: bgpTableVersion, advertisedRoutes, etc.
# routes view uses: tableVersion, routes, routerId, etc.
if isinstance(data, dict):
route_view_keys = ["advertisedRoutes", "receivedRoutes", "bgpTableVersion",
"tableVersion", "routerId"]
# Also check for 'routes' key but only if it's a dict of prefixes (not neighbor data)
has_routes_dict = ("routes" in data and isinstance(data.get("routes"), dict) and
any("/" in k or "::" in k for k in data.get("routes", {}).keys()))
if any(key in data for key in route_view_keys) or has_routes_dict:
# This is a routes table view, delegate to routes formatter
routes_formatter = IPv6BgpRoutesTableFormatter()
return routes_formatter.format(data, config, table_format, options)
lines = []
# Sort neighbors by IP address using proper IPv6 sorting
neighbors = sorted(data.keys(), key=self._ipv6_sort_key)
for neighbor_ip in neighbors:
neighbor_data = data[neighbor_ip]
if not isinstance(neighbor_data, dict):
continue
# Build the detailed output for this neighbor
lines.append(self._format_neighbor(neighbor_ip, neighbor_data))
return "\n".join(lines)
def _ipv6_sort_key(self, addr: str) -> tuple:
"""Generate a sortable key for IPv6 addresses."""
import ipaddress
try:
return (0, ipaddress.ip_address(addr).packed)
except ValueError:
# If it's not a valid IP, sort it at the end alphabetically
return (1, addr.encode())
def _format_neighbor(self, neighbor_ip: str, data: Dict) -> str:
"""Format a single neighbor's detailed output."""
lines = []
remote_as = data.get("remoteAs", "N/A")
local_as = data.get("localAs", "N/A")
external_link = "external link" if data.get("nbrExternalLink", False) else "confed-external link"
# First line
lines.append(f"BGP neighbor is {neighbor_ip}, remote AS {remote_as}, local AS {local_as}, {external_link}")
# Local/Remote roles
local_role = data.get("localRole", "undefined")
remote_role = data.get("remoteRole", "undefined")
lines.append(f" Local Role: {local_role}")
lines.append(f" Remote Role: {remote_role}")
# Description
desc = data.get("nbrDesc", "")
if desc:
lines.append(f" Description: {desc}")
# Peer group
peer_group = data.get("peerGroup", "")
if peer_group:
lines.append(f" Member of peer-group {peer_group} for session parameters")
# BGP version, router IDs
bgp_version = data.get("bgpVersion", 4)
remote_router_id = data.get("remoteRouterId", "")
local_router_id = data.get("localRouterId", "")
lines.append(f" BGP version {bgp_version}, remote router ID {remote_router_id}, local router ID {local_router_id}")
# Neighbor under common administration (confederation)
if not data.get("nbrExternalLink", False):
lines.append(" Neighbor under common administration")
# Hostname
hostname = data.get("hostname", "")
if hostname and hostname != "Unknown":
lines.append(f" Hostname: {hostname}")
# Software version
sw_version = data.get("softwareVersion", "")
if sw_version and sw_version != "n/a":
lines.append(f" Software Version: {sw_version}")
# BGP state
bgp_state = data.get("bgpState", "")
up_time = data.get("bgpTimerUpString", "")
bgp_up_msec = data.get("bgpTimerUpMsec", 0)
if bgp_state:
lines.append(f" BGP state = {bgp_state}, up for {up_time}")
# BGP In Update elapsed time
in_update_elapsed = data.get("bgpInUpdateElapsedTimeMsecs", 0)
if in_update_elapsed:
lines.append(f" Last update from {neighbor_ip} {self._format_uptime(in_update_elapsed)} ago")
# Last read/write
last_read = data.get("bgpTimerLastRead", 0)
last_write = data.get("bgpTimerLastWrite", 0)
lines.append(f" Last read {self._format_time(last_read)}, Last write {self._format_time(last_write)}")
# Hold time, keepalive
hold_time = data.get("bgpTimerHoldTimeMsecs", 0) // 1000
keepalive = data.get("bgpTimerKeepAliveIntervalMsecs", 0) // 1000
lines.append(f" Hold time is {hold_time} seconds, keepalive interval is {keepalive} seconds")
# Configured hold/keepalive
cfg_hold = data.get("bgpTimerConfiguredHoldTimeMsecs", 0) // 1000
cfg_keepalive = data.get("bgpTimerConfiguredKeepAliveIntervalMsecs", 0) // 1000
lines.append(f" Configured hold time is {cfg_hold} seconds, keepalive interval is {cfg_keepalive} seconds")
# TCP MSS
tcp_mss_cfg = data.get("bgpTcpMssConfigured", 0)
tcp_mss_synced = data.get("bgpTcpMssSynced", 0)
lines.append(f" Configured tcp-mss is {tcp_mss_cfg}, synced tcp-mss is {tcp_mss_synced}")
# Conditional advertisements interval
cond_adv = data.get("bgpTimerConfiguredConditionalAdvertisementsSec", 60)
lines.append(f" Configured conditional advertisements interval is {cond_adv} seconds")
# Connect retry timer
connect_retry = data.get("connectRetryTimer", 0)
if connect_retry:
lines.append(f" Connect retry timer: {connect_retry} seconds")
# BGP timer up established epoch (Unix timestamp)
up_epoch = data.get("bgpTimerUpEstablishedEpoch", 0)
if up_epoch:
lines.append(f" Peer established epoch: {up_epoch}")
# Extended optional parameters length
ext_opt_params = data.get("extendedOptionalParametersLength", False)
if ext_opt_params:
lines.append(" Extended Optional Parameters Length: Enabled")
# External BGP neighbor max hops
max_hops = data.get("externalBgpNbrMaxHopsAway", 0)
if max_hops and data.get("nbrExternalLink", False):
lines.append(f" External BGP neighbor may be up to {max_hops} hops away.")
# Host local/foreign addresses and ports
host_local = data.get("hostLocal", "")
host_foreign = data.get("hostForeign", "")
port_local = data.get("portLocal", 0)
port_foreign = data.get("portForeign", 0)
if host_local and host_foreign:
lines.append(f" Local host: {host_local}, Local port: {port_local}")
lines.append(f" Foreign host: {host_foreign}, Foreign port: {port_foreign}")
# Nexthop information
nexthop = data.get("nexthop", "")
nexthop_global = data.get("nexthopGlobal", "")
nexthop_local = data.get("nexthopLocal", "")
if nexthop:
lines.append(f" Nexthop: {nexthop}")
if nexthop_global:
lines.append(f" Nexthop global: {nexthop_global}")
if nexthop_local:
lines.append(f" Nexthop local: {nexthop_local}")
# Read/write thread status
read_thread = data.get("readThread", "")
write_thread = data.get("writeThread", "")
if read_thread or write_thread:
lines.append(f" Read thread: {read_thread}, Write thread: {write_thread}")
# Estimated RTT
rtt = data.get("estimatedRttInMsecs", 0)
if rtt:
lines.append(f" Estimated round trip time: {rtt} ms")
# BGP connection type
bgp_conn = data.get("bgpConnection", "")
if bgp_conn:
# Convert camelCase to spaced format (e.g., "sharedNetwork" -> "shared network")
import re as _re
bgp_conn_formatted = _re.sub(r'([a-z])([A-Z])', r'\1 \2', bgp_conn).lower()
lines.append(f" BGP connection: {bgp_conn_formatted}")
# Neighbor capabilities
caps = data.get("neighborCapabilities", {})
if caps:
lines.append(" Neighbor capabilities:")
lines.extend(self._format_capabilities(caps))
# Graceful restart info
gr_info = data.get("gracefulRestartInfo", {})
if gr_info:
lines.extend(self._format_graceful_restart(gr_info))
# Message statistics
msg_stats = data.get("messageStats", {})
if msg_stats:
lines.extend(self._format_message_stats(msg_stats))
# Minimum time between advertisement runs
min_adv_time = data.get("minBtwnAdvertisementRunsTimerMsecs", 0) // 1000
lines.append(f" Minimum time between advertisement runs is {min_adv_time} seconds")
lines.append("")
# Address family info - look inside addressFamilyInfo dict
af_info = data.get("addressFamilyInfo", {})
for af_name, af_data in af_info.items():
if isinstance(af_data, dict):
lines.extend(self._format_address_family(af_name, af_data))
# Connection established/dropped info
conn_est = data.get("connectionsEstablished", 0)
conn_drop = data.get("connectionsDropped", 0)
lines.append(f" Connections established {conn_est}; dropped {conn_drop}")
# Last reset info
last_reset_msec = data.get("lastResetTimerMsecs", 0)
last_reset_reason = data.get("lastResetDueTo", "")
last_reset_code = data.get("lastResetCode", "n/a")
if last_reset_msec or last_reset_reason:
reset_time = self._format_uptime(last_reset_msec)
lines.append(f" Last reset {reset_time}, {last_reset_reason} ({last_reset_code})")
lines.append("") # Blank line between neighbors
return "\n".join(lines)
def _format_time(self, msecs: int) -> str:
"""Format milliseconds as HH:MM:SS."""
seconds = msecs // 1000
hours = seconds // 3600
minutes = (seconds % 3600) // 60
secs = seconds % 60
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
def _format_uptime(self, msecs: int) -> str:
"""Format milliseconds as uptime string like '1d00h33m' or '04:43:30'."""
if not msecs:
return "never"
seconds = msecs // 1000
days = seconds // 86400
hours = (seconds % 86400) // 3600
minutes = (seconds % 3600) // 60
if days > 0:
return f"{days}d{hours:02d}h{minutes:02d}m"
else:
secs = seconds % 60
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
def _format_capabilities(self, caps: Dict) -> List[str]:
"""Format neighbor capabilities."""
lines = []
# 4 byte AS
four_byte = caps.get("4byteAs", "")
if four_byte:
lines.append(f" 4 Byte AS: {self._format_cap_value(four_byte)}")
# Extended message
ext_msg = caps.get("extendedMessage", "")
if ext_msg:
lines.append(f" Extended Message: {ext_msg}")
# Add Path
add_path = caps.get("addPath", {})
if add_path:
lines.append(" AddPath:")
for af, af_data in add_path.items():
if isinstance(af_data, dict):
if af_data.get("rxAdvertisedAndReceived"):
lines.append(f" {self._format_af_name(af)}: RX advertised and received")
# Long-lived GR
llgr = caps.get("longLivedGracefulRestart", "")
if llgr:
lines.append(f" Long-lived Graceful Restart: {llgr}")
# Route refresh
route_ref = caps.get("routeRefresh", "")
if route_ref:
lines.append(f" Route refresh: {self._format_cap_value(route_ref)}")
# Enhanced route refresh
enh_route_ref = caps.get("enhancedRouteRefresh", "")
if enh_route_ref:
lines.append(f" Enhanced Route Refresh: {self._format_cap_value(enh_route_ref)}")
# Multiprotocol extensions
mp_ext = caps.get("multiprotocolExtensions", {})
if mp_ext:
for af, status in mp_ext.items():
if status:
lines.append(f" Address Family {self._format_af_name(af)}: {self._format_cap_value(status)}")
# Hostname capability
hostname = caps.get("hostName", {})
if hostname and isinstance(hostname, dict):
local_name = hostname.get("advHostName", "")
domain = hostname.get("advDomainName", "n/a")
received = hostname.get("rcvHostName", "")
recv_str = "received" if received else "not received"
if local_name:
lines.append(f" Hostname Capability: advertised (name: {local_name},domain name: {domain}) {recv_str}")
# Software version capability
sw_version = caps.get("softwareVersion", {})
if sw_version and isinstance(sw_version, dict):
lines.append(" Version Capability: not advertised not received")
# Graceful restart capability
gr_cap = caps.get("gracefulRestart", "")
if gr_cap:
# gr_cap is a string value like "advertisedAndReceived"
lines.append(f" Graceful Restart Capability: {self._format_cap_value(gr_cap)}")
remote_timer = caps.get("gracefulRestartRemoteTimerMsecs", 0) // 1000
if remote_timer:
lines.append(f" Remote Restart timer is {remote_timer} seconds")
# Address families by peer
af_by_peer = caps.get("addressFamiliesByPeer", {})
if af_by_peer:
lines.append(" Address families by peer:")
if isinstance(af_by_peer, str):
# Value is a string like "none" - output as-is
lines.append(f" {af_by_peer}")
else:
for af in af_by_peer:
lines.append(f" {self._format_af_name(af)}(not preserved)")
return lines
def _format_cap_value(self, value) -> str:
"""Format capability value."""
if isinstance(value, dict):
# Handle dict like {'advertisedAndReceived': True}
if value.get("advertisedAndReceived"):
return "advertised and received"
elif value.get("advertised"):
return "advertised"
elif value.get("received"):
return "received"
return "advertised and received" # Default for true dict values
if value == "advertisedAndReceived":
return "advertised and received"
elif value == "advertised":
return "advertised"
elif value == "received":
return "received"
return str(value)
def _format_af_name(self, af_key: str) -> str:
"""Format address family name."""
mapping = {
"ipv6Unicast": "IPv6 Unicast",
"ipv4Unicast": "IPv4 Unicast",
}
return mapping.get(af_key, af_key)
def _format_graceful_restart(self, gr_info: Dict) -> List[str]:
"""Format graceful restart information."""
lines = [" Graceful restart information:"]
# End-of-RIB info
eor_send = gr_info.get("endOfRibSend", {})
eor_recv = gr_info.get("endOfRibRecv", {})
for af in eor_send:
lines.append(f" End-of-RIB send: {self._format_af_name(af)}")
for af in eor_recv:
lines.append(f" End-of-RIB received: {self._format_af_name(af)}")
# Local/Remote GR mode
local_mode = gr_info.get("localGrMode", "")
remote_mode = gr_info.get("remoteGrMode", "")
if local_mode:
lines.append(f" Local GR Mode: {local_mode}")
lines.append("") # Empty line like CLI output
if remote_mode:
lines.append(f" Remote GR Mode: {remote_mode}")
lines.append("")
# R bit and N bit
r_bit = gr_info.get("rBit", False)
n_bit = gr_info.get("nBit", False)
lines.append(f" R bit: {r_bit}")
lines.append(f" N bit: {n_bit}")
# Timers
timers = gr_info.get("timers", {})
if timers:
lines.append(" Timers:")
cfg_restart = timers.get("configuredRestartTimer", 0)
rcv_restart = timers.get("receivedRestartTimer", 0)
cfg_llgr = timers.get("configuredLlgrStaleTime", 0)
lines.append(f" Configured Restart Time(sec): {cfg_restart}")
lines.append(f" Received Restart Time(sec): {rcv_restart}")
lines.append(f" Configured LLGR Stale Path Time(sec): {cfg_llgr}")
# Per-AFI graceful restart info
for af_key in ["ipv6Unicast", "ipv4Unicast"]:
af_gr = gr_info.get(af_key, {})
if af_gr:
lines.append(f" {self._format_af_name(af_key)}:")
f_bit = af_gr.get("fBit", False)
lines.append(f" F bit: {f_bit}")
eor_status = af_gr.get("endOfRibStatus", {})
if eor_status:
eor_sent = "Yes" if eor_status.get("endOfRibSend", False) else "No"
eor_sent_after = "Yes" if eor_status.get("endOfRibSentAfterUpdate", False) else "No"
eor_recv = "Yes" if eor_status.get("endOfRibRecv", False) else "No"
lines.append(f" End-of-RIB sent: {eor_sent}")
lines.append(f" End-of-RIB sent after update: {eor_sent_after}")
lines.append(f" End-of-RIB received: {eor_recv}")
af_timers = af_gr.get("timers", {})
if af_timers:
lines.append(" Timers:")
stale_path = af_timers.get("stalePathTimer", 0)
llgr_stale = af_timers.get("llgrStaleTime", 0)
lines.append(f" Configured Stale Path Time(sec): {stale_path}")
lines.append(f" LLGR Stale Path Time(sec): {llgr_stale}")
return lines
def _format_message_stats(self, stats: Dict) -> List[str]:
"""Format message statistics."""
lines = [" Message statistics:"]
# Input/output queues (gNMI uses depthInq/depthOutq)
inq = stats.get("depthInq", 0)
outq = stats.get("depthOutq", 0)
lines.append(f" Inq depth is {inq}")
lines.append(f" Outq depth is {outq}")
# Header line
lines.append(" Sent Rcvd")
# Message types - gNMI uses camelCase field names
msg_types = [
("Opens", "opensSent", "opensRecv"),
("Notifications", "notificationsSent", "notificationsRecv"),
("Updates", "updatesSent", "updatesRecv"),
("Keepalives", "keepalivesSent", "keepalivesRecv"),
("Route Refresh", "routeRefreshSent", "routeRefreshRecv"),
("Capability", "capabilitySent", "capabilityRecv"),
("Total", "totalSent", "totalRecv"),
]
for label, sent_key, recv_key in msg_types:
sent = stats.get(sent_key, 0)
recv = stats.get(recv_key, 0)
lines.append(f" {label}:{sent:>13}{recv:>11}")
return lines
def _format_address_family(self, af_name: str, af_data: Dict) -> List[str]:
"""Format address family information."""
lines = [f" For address family: {self._format_af_name(af_name)}"]
# Peer group member
peer_group = af_data.get("peerGroupMember", "")
if peer_group:
lines.append(f" {peer_group} peer-group member")
# Update group, subgroup
update_group = af_data.get("updateGroupId", "")
subgroup = af_data.get("subGroupId", "")
if update_group:
lines.append(f" Update group {update_group}, subgroup {subgroup}")
# Packet queue length
pkt_queue = af_data.get("packetQueueLength", 0)
lines.append(f" Packet Queue length {pkt_queue}")
# Inbound soft reconfig
if af_data.get("inboundSoftConfigPermit", False):
lines.append(" Inbound soft reconfiguration allowed")
# Allow AS in
allow_as_in = af_data.get("allowAsInCount", 0)
if allow_as_in:
lines.append(f" Local AS allowed in path, {allow_as_in} occurrences")
# NEXT_HOP setting (if present)
# Community attribute
comm_attr = af_data.get("commAttriSentToNbr", "")
if comm_attr:
lines.append(f" Community attribute sent to this neighbor({comm_attr})")
# Path policy config
if af_data.get("inboundPathPolicyConfig", False):
lines.append(" Inbound path policy configured")
if af_data.get("outboundPathPolicyConfig", False):
lines.append(" Outbound path policy configured")
# Route maps
in_routemap = af_data.get("routeMapForIncomingAdvertisements", "")
out_routemap = af_data.get("routeMapForOutgoingAdvertisements", "")
if in_routemap:
lines.append(f" Route map for incoming advertisements is *{in_routemap}")
if out_routemap:
lines.append(f" Route map for outgoing advertisements is *{out_routemap}")
# Accepted/sent prefixes
accepted = af_data.get("acceptedPrefixCounter", 0)
sent = af_data.get("sentPrefixCounter", 0)
lines.append(f" {accepted} accepted prefixes, {sent} sent prefixes")
# Maximum prefixes
max_prefix = af_data.get("prefixAllowedMax", 0)
warning_only = af_data.get("prefixAllowedMaxWarning", False)
warning_thresh = af_data.get("prefixAllowedWarningThresh", 0)
if max_prefix:
warn_str = "(warning-only)" if warning_only else ""
lines.append(f" Maximum prefixes allowed {max_prefix} {warn_str}")
if warning_thresh:
lines.append(f" Threshold for warning message {warning_thresh}%")
lines.append("") # Empty line
return lines
class IPv6BgpRoutesTableFormatter(OutputFormatter):
"""
Formats IPv6 BGP routes table output for advertised-routes, received-routes, routes views.
This formatter handles the route table output when a neighbor IP and view type
(advertised-routes, received-routes, routes) are specified.
Example output:
BGP table version is 91647, local router ID is 10.1.0.32, vrf id 0
Default local pref 100, local AS 65100
Status codes: s suppressed, d damped, h history, u unsorted, * valid, > best, = multipath,
i internal, r RIB-failure, S Stale, R Removed
Nexthop codes: @NNN nexthop's vrf id, < announce-nh-self
Origin codes: i - IGP, e - EGP, ? - incomplete
RPKI validation codes: V valid, I invalid, N Not found
Network Next Hop Metric LocPrf Weight Path
*> ::/0 :: 0 64600 65534 6666 6667 i
*> 2064:100:0:41::/128
:: 0 64600 i
"""
def format(
self,
data: Union[Dict, List],
config: CommandConfig,
table_format: str = "simple",
argument: str = "",
options: Optional[Dict] = None,
) -> str:
"""Format BGP routes table."""
if not data:
return ""
lines = []
# Header info - handle both JSON structures
# advertisedRoutes uses bgpTableVersion, routes uses tableVersion
table_version = data.get("bgpTableVersion", data.get("tableVersion", 0))
router_id = data.get("bgpLocalRouterId", data.get("routerId", ""))
local_pref = data.get("defaultLocPrf", 100)
local_as = data.get("localAS", "")
vrf_id = data.get("vrfId", 0)
lines.append(f"BGP table version is {table_version}, local router ID is {router_id}, vrf id {vrf_id}")
lines.append(f"Default local pref {local_pref}, local AS {local_as}")
# Status codes
lines.append("Status codes: s suppressed, d damped, h history, u unsorted, * valid, > best, = multipath,")
lines.append(" i internal, r RIB-failure, S Stale, R Removed")
lines.append("Nexthop codes: @NNN nexthop's vrf id, < announce-nh-self")
lines.append("Origin codes: i - IGP, e - EGP, ? - incomplete")
lines.append("RPKI validation codes: V valid, I invalid, N Not found")
lines.append("")
# Column header
lines.append(" Network Next Hop Metric LocPrf Weight Path")
# Determine which routes dict to use
routes = {}
is_list_format = False # routes view uses list per prefix, others use dict
if "advertisedRoutes" in data:
routes = data["advertisedRoutes"]
elif "receivedRoutes" in data:
routes = data["receivedRoutes"]
elif "routes" in data:
routes = data["routes"]
# If routes dict is empty, CLI shows nothing
if not routes:
return ""
is_list_format = True # routes view has list of paths per prefix
else:
# Try treating the whole data as routes
routes = {k: v for k, v in data.items()
if isinstance(v, (dict, list)) and k not in ["bgpTableVersion", "bgpLocalRouterId",
"defaultLocPrf", "localAS", "tableVersion", "routerId", "vrfId", "vrfName",
"totalPrefixCounter", "filteredPrefixCounter"]}
# Sort routes by network prefix
sorted_prefixes = sorted(routes.keys(), key=self._prefix_sort_key)
for prefix in sorted_prefixes:
route_info = routes[prefix]
if is_list_format and isinstance(route_info, list):
# routes view: each prefix has a list of paths
for path_info in route_info:
if isinstance(path_info, dict):
lines.append(self._format_route(prefix, path_info))
elif isinstance(route_info, dict):
lines.append(self._format_route(prefix, route_info))
# Total counts - use "Displayed X routes" format, add "and Y total paths" only if available
total_prefixes = data.get("totalPrefixCounter", len(routes))
total_paths = data.get("totalPathCounter") # Only include if explicitly provided
lines.append("")
if total_paths is not None:
lines.append(f"Displayed {total_prefixes} routes and {total_paths} total paths")
else:
lines.append(f"Displayed {total_prefixes} routes")
return "\n".join(lines)
def _prefix_sort_key(self, prefix: str) -> tuple:
"""Generate a sortable key for IPv6 prefixes."""
import ipaddress
try:
net = ipaddress.ip_network(prefix, strict=False)
return (0, net.network_address.packed, net.prefixlen)
except ValueError:
return (1, prefix.encode(), 0)
def _format_route(self, prefix: str, route: Dict) -> str:
"""Format a single route entry."""
# Status flags
status = " "
if route.get("valid", False):
status += "*"
else:
status += " "
# bestpath can be 'best' or 'bestpath' depending on JSON source
if route.get("best", False) or route.get("bestpath", False):
status += ">"
elif route.get("multipath", False):
status += "="
else:
status += " "
network = route.get("network", prefix)
# Next hop can be in different fields depending on JSON source
next_hop = route.get("nextHopGlobal", route.get("nextHop", ""))
if not next_hop:
# routes view uses nexthops array
nexthops = route.get("nexthops", [])
if nexthops and isinstance(nexthops, list) and nexthops[0]:
next_hop = nexthops[0].get("ip", "::")
else:
next_hop = "::"
metric = route.get("metric", "")
loc_prf = route.get("locPrf", "")
weight = route.get("weight", 0)
path = route.get("path", "")
origin = self._format_origin(route.get("origin", ""))
# Format the line - network may be long so use continuation
# CLI column layout (determined by header positions):
# Status: 4 chars (positions 0-3: " *= " or similar)
# Network: 17 chars (positions 4-20)
# Next Hop: 20 chars (positions 21-40)
# Metric: 7 chars (positions 41-47)
# LocPrf: 7 chars (positions 48-54)
# Weight: 6 chars (positions 55-60)
# Space: 1 char (position 61)
# Path + Origin: rest (positions 62+)
# Networks with length < 17 fit on one line; networks with length >= 17
# go to continuation (because 17-char network leaves no padding before next_hop)
if len(network) < 17:
# Short prefix - fits on one line
line = f"{status} {network:<17}{next_hop:<20}{metric:>7}{loc_prf:>7}{weight:>6} {path} {origin}"
else:
# Long prefix - network on first line, rest on continuation
# Continuation indent: 20 spaces (aligns with Next Hop at position 21)
line = f"{status} {network}\n {next_hop:<20}{metric:>7}{loc_prf:>7}{weight:>6} {path} {origin}"
return line.rstrip()
def _format_origin(self, origin: str) -> str:
"""Format origin code."""
if origin == "IGP" or origin == "i":
return "i"
elif origin == "EGP" or origin == "e":
return "e"
elif origin == "incomplete" or origin == "?":
return "?"
return origin
class IPv6BgpNetworkFormatter(OutputFormatter):
"""
Formats IPv6 BGP network output (passthrough).
The data contains pre-formatted output from FRR, just extract and return it.
Example input:
{"output": "BGP table version is 135211, local router ID is 10.1.0.32..."}
Example output:
BGP table version is 135211, local router ID is 10.1.0.32, vrf id 0
...
"""
def format(
self,
data: Union[Dict, List],
config: CommandConfig,
table_format: str = "simple",
argument: str = "",
options: Optional[Dict] = None,
) -> str:
"""Format BGP network as passthrough text."""
if not data:
return ""
# Extract the output field - it contains the pre-formatted text
if isinstance(data, dict) and "output" in data:
return data["output"]
return str(data)
class IPv6BgpSummaryFormatter(OutputFormatter):
"""
Formats IPv6 BGP summary output with header and table.
Example output:
IPv6 Unicast Summary:
BGP router identifier 10.1.0.32, local AS number 4200200000 vrf-id 0
BGP table version 135211
RIB entries 32209, using 4122752 bytes of memory
Peers 88, using 1813504 KiB of memory
Peer groups 6, using 384 bytes of memory
Neighbhor V AS MsgRcvd MsgSent TblVer InQ OutQ Up/Down State/PfxRcd NeighborName
----------- --- ---------- --------- --------- -------- ----- ------ --------- -------------- --------------
fc00::1a 4 4200300000 1852 1875 135211 0 0 1d00h33m 1 ARISTA07FT2
"""
def format(
self,
data: Union[Dict, List],
config: CommandConfig,
table_format: str = "simple",
argument: str = "",
options: Optional[Dict] = None,
) -> str:
"""Format BGP summary with header and table."""
if not data:
return ""
lines = []
# Get the ipv6Unicast data
unicast_data = data.get("ipv6Unicast", {})
if not unicast_data:
return ""
# Header section
lines.append("")
lines.append("IPv6 Unicast Summary:")
router_id = unicast_data.get("routerId", "")
local_as = unicast_data.get("as", "")
vrf_id = unicast_data.get("vrfId", 0)
lines.append(f"BGP router identifier {router_id}, local AS number {local_as} vrf-id {vrf_id}")
table_version = unicast_data.get("tableVersion", 0)
lines.append(f"BGP table version {table_version}")
rib_count = unicast_data.get("ribCount", 0)
rib_memory = unicast_data.get("ribMemory", 0)
lines.append(f"RIB entries {rib_count}, using {rib_memory} bytes of memory")
peer_count = unicast_data.get("peerCount", 0)
peer_memory = unicast_data.get("peerMemory", 0)
lines.append(f"Peers {peer_count}, using {peer_memory} KiB of memory")
pg_count = unicast_data.get("peerGroupCount", 0)
pg_memory = unicast_data.get("peerGroupMemory", 0)
lines.append(f"Peer groups {pg_count}, using {pg_memory} bytes of memory")
lines.append("")
lines.append("")
# Peers table
peers = unicast_data.get("peers", {})
if peers:
# Convert "mixed" columns (integers in normal rows, "" in the
# synthetic Total row) to strings so that tabulate's column-type
# detector sees a pure-string column and does not right-align it.
def _s(v):
return "" if v == "" else str(v)
table_data = []
for peer_ip in sorted(peers.keys()):
peer = peers[peer_ip]
row = [
peer_ip,
peer.get("version", 4), # pure-int column → right-align
peer.get("remoteAs", ""), # mixed ("ighbo" + ints) → left
peer.get("msgRcvd", 0), # pure-int column → right-align
_s(peer.get("msgSent", 0)), # mixed ("" + ints) → left
_s(peer.get("tableVersion", 0)), # mixed ("" + ints) → left
_s(peer.get("inq", 0)), # mixed ("" + ints) → left
_s(peer.get("outq", 0)), # mixed ("" + ints) → left
peer.get("peerUptime", ""), # str → left
_s(peer.get("pfxRcd", peer.get("state", ""))), # mixed → left
peer.get("NeighborName", ""), # str → left
]
table_data.append(row)
headers = ["Neighbhor", "V", "AS", "MsgRcvd", "MsgSent", "TblVer",
"InQ", "OutQ", "Up/Down", "State/PfxRcd", "NeighborName"]
# Use disable_numparse=True so tabulate never re-parses string cells
# (e.g. "7915") back to integers. Column alignment is then driven
# entirely by colalign: only V (idx 1) and MsgRcvd (idx 3) are
# right-aligned, matching the sonic-utilities baseline output.
# This produces identical output regardless of tabulate / Python version.
col_align = ("left","right","left","right",
"left","left","left","left","left","left","left")
table_output = tabulate(
table_data, headers=headers, tablefmt=table_format,
disable_numparse=True, colalign=col_align,
)
lines.append(table_output)
return "\n".join(lines)
class IPv6FibFormatter(OutputFormatter):
"""
Formats IPv6 FIB output as a table.
Example input:
{"total": 16195, "entries": [{"index": 1, "route": "2064:100:0:10::", ...}]}
Example output:
No. Vrf Route Nexthop Ifname
----- ----- --------------------- ---------------- -------
1 2064:100:0:10:: fc00::3e Ethernet120
"""
def format(
self,
data: Union[Dict, List],
config: CommandConfig,
table_format: str = "simple",
argument: str = "",
options: Optional[Dict] = None,
) -> str:
"""Format FIB as table."""
if not data:
return ""
entries = data.get("entries", [])
table_data = []
for entry in entries:
if isinstance(entry, dict):
row = [
entry.get("index", ""),
entry.get("vrf", ""), # Vrf column (usually empty)
entry.get("route", ""),
entry.get("nexthop", ""),
entry.get("ifname", ""),
]
table_data.append(row)
headers = ["No.", "Vrf", "Route", "Nexthop", "Ifname"]
table = tabulate(table_data, headers=headers, tablefmt=table_format)
# Add total count footer
total = data.get("total", len(table_data))
table += f"\nTotal number of entries {total}"
return table
class IPv6InterfacesFormatter(OutputFormatter):
"""
Formats IPv6 interfaces output as a table with multi-row support.
Multiple IPv6 addresses per interface are shown on continuation lines.
Link-local addresses (fe80::) include zone identifier (%interface_name).
Example output:
Interface Master IPv6 address/mask Admin/Oper BGP Neighbor Neighbor IP
-------------- ------ ------------------------------------------- ------------ -------------- -------------
Ethernet0 fc00::1/126 up/up ARISTA01FT2 fc00::2
fe80::7a5f:6cff:fe30:d74c%Ethernet0/64 N/A N/A
"""
def _format_ipv6_address(self, address: str, interface_name: str) -> str:
"""
Format IPv6 address, adding zone identifier for link-local addresses.
Link-local addresses (fe80::) get the interface name appended as a zone ID
before the prefix length, e.g., fe80::1234%Ethernet0/64
Args:
address: IPv6 address with prefix (e.g., "fe80::1234/64")
interface_name: Interface name to use as zone ID
Returns:
Formatted address with zone ID if link-local
"""
if not address:
return address
# Check if it's a link-local address (starts with fe80::)
addr_lower = address.lower()
if addr_lower.startswith("fe80:"):
# Split address and prefix length
if "/" in address:
addr_part, prefix = address.rsplit("/", 1)
return f"{addr_part}%{interface_name}/{prefix}"
else:
return f"{address}%{interface_name}"
return address
def format(
self,
data: Union[Dict, List],
config: CommandConfig,
table_format: str = "simple",
argument: str = "",
options: Optional[Dict] = None,
) -> str: