lib.frr

   1"""FRR / OSPF / BGP / RIP grading helpers.
   2
   3Every ``get_*`` function registers one command through ``grade.test()`` -- so it
   4follows the two-pass contract of ``Grade.grade()`` (placeholder result on the
   5registration pass, real output afterwards) -- and hands the text to a pure
   6``parse_*`` function.  The parsers are unit-tested on captured FRR 9 output in
   7``tests/test_frr.py``.  The OSPF and RIP helpers parse the text output of ``vtysh``
   8(``show ip rip`` and ``show ip rip status`` have no json form), the BGP helpers its
   9``json`` output (``show bgp ipv4 unicast ... json``).  On the placeholder output
  10every parser returns its empty shape (``{}``/``[]``), never an exception.  What a
  11RIP router really announces on a link is read by the passive probe of ``lib/rip.py``.
  12
  13Pass ``allow_error=True`` when the router may not run OSPF/BGP/RIP yet (``vtysh``
  14then exits non-zero and the error would otherwise be recorded in the archive).
  15"""
  16import json
  17import re
  18from ipaddress import IPv4Address
  19from typing import Any, Dict, List, Optional
  20
  21from SRE.lib_sre import Grade0
  22
  23_IPV4_RE = re.compile(r'^\d{1,3}(?:\.\d{1,3}){3}$')
  24
  25#: kernel route protocol values that mean "installed by ospfd" (name from
  26#: /etc/iproute2/rt_protos.d/frr.conf, or the raw number when that file is missing)
  27OSPF_ROUTE_PROTOCOLS = ('ospf', '188')
  28#: kernel route protocol values that mean "installed by bgpd"
  29BGP_ROUTE_PROTOCOLS = ('bgp', '186')
  30#: kernel route protocol values that mean "installed by ripd" / "by ripngd"
  31RIP_ROUTE_PROTOCOLS = ('rip', '189')
  32RIPNG_ROUTE_PROTOCOLS = ('ripng', '190')
  33
  34
  35def _test(grade: Grade0, machine_name: str, command: str, step: int, allow_error: bool):
  36    if allow_error:
  37        return grade.test(machine_name, command, step=step, allow_error=True)
  38    return grade.test(machine_name, command, step=step)
  39
  40
  41def normalize_area(area: str) -> str:
  42    """Return an OSPF area id in dotted form: ``'1'`` -> ``'0.0.0.1'``, ``'0.0.0.1'`` unchanged."""
  43    area = str(area).strip()
  44    if area.isdigit():
  45        return str(IPv4Address(int(area)))
  46    return area
  47
  48
  49def is_ospf_protocol(protocol) -> bool:
  50    """True when a kernel route ``protocol`` field (``ip -j route``) denotes an OSPF route."""
  51    return str(protocol) in OSPF_ROUTE_PROTOCOLS
  52
  53
  54def is_bgp_protocol(protocol) -> bool:
  55    """True when a kernel route ``protocol`` field (``ip -j route``) denotes a BGP route."""
  56    return str(protocol) in BGP_ROUTE_PROTOCOLS
  57
  58
  59def is_rip_protocol(protocol) -> bool:
  60    """True when a kernel route ``protocol`` field (``ip -j route``) denotes a RIP route."""
  61    return str(protocol) in RIP_ROUTE_PROTOCOLS
  62
  63
  64def is_ripng_protocol(protocol) -> bool:
  65    """True when a kernel route ``protocol`` field (``ip -j -6 route``) denotes a RIPng route."""
  66    return str(protocol) in RIPNG_ROUTE_PROTOCOLS
  67
  68
  69# ---------------------------------------------------------------------------
  70# show ip ospf interface
  71# ---------------------------------------------------------------------------
  72
  73def parse_ospf_interfaces(output: str) -> Dict[str, Dict[str, Any]]:
  74    """Parse the text of ``show ip ospf interface`` (see :func:`get_ospf_interfaces`)."""
  75    result: Dict[str, Dict[str, Any]] = {}
  76    current: Dict[str, Any] | None = None
  77
  78    for line in output.splitlines():
  79        # Interface header: "ethX is up"
  80        m = re.match(r'^(\S+) is (up|down)', line)
  81        if m:
  82            current = {
  83                'state': m.group(2),
  84                'ospf_enabled': True,
  85                'mtu': None, 'bandwidth': None,
  86                'internet_address': None, 'broadcast': None, 'area': None,
  87                'router_id': None, 'network_type': None, 'cost': None,
  88                'ospf_state': None, 'priority': None, 'passive': False,
  89                'hello_interval': None, 'dead_interval': None, 'retransmit_interval': None,
  90                'neighbor_count': None, 'adjacent_neighbor_count': None,
  91                'dr_id': None, 'dr_address': None,
  92                'bdr_id': None, 'bdr_address': None,
  93            }
  94            result[m.group(1)] = current
  95            continue
  96
  97        if current is None:
  98            continue
  99
 100        if 'OSPF not enabled on this interface' in line:
 101            current['ospf_enabled'] = False
 102            continue
 103
 104        # ifindex 13, MTU 1500 bytes, BW 10000 Mbit <UP,LOWER_UP,BROADCAST,RUNNING,MULTICAST>
 105        m = re.search(r'MTU (\d+) bytes,\s*BW (\d+) Mbit', line)
 106        if m:
 107            current['mtu'] = int(m.group(1))
 108            current['bandwidth'] = int(m.group(2))
 109            continue
 110
 111        # Internet Address 10.150.180.224/28, Broadcast 10.150.180.239, Area 0.0.0.0
 112        m = re.search(r'Internet Address (\S+?),\s*Broadcast (\S+?),\s*Area (\S+)', line)
 113        if m:
 114            current['internet_address'] = m.group(1)
 115            current['broadcast'] = m.group(2)
 116            current['area'] = m.group(3)
 117            continue
 118
 119        # Internet Address 10.0.0.1/30, Area 0.0.0.0   (point-to-point, no broadcast)
 120        m = re.search(r'Internet Address (\S+?),\s*Area (\S+)', line)
 121        if m:
 122            current['internet_address'] = m.group(1)
 123            current['area'] = m.group(2)
 124            continue
 125
 126        # Router ID 10.150.180.224, Network Type BROADCAST, Cost: 10
 127        m = re.search(r'Router ID (\S+?),\s*Network Type (\S+?),\s*Cost:\s*(\d+)', line)
 128        if m:
 129            current['router_id'] = m.group(1)
 130            current['network_type'] = m.group(2)
 131            current['cost'] = int(m.group(3))
 132            continue
 133
 134        # Transmit Delay is 1 sec, State Backup, Priority 1
 135        m = re.search(r'State (\S+?),\s*Priority (\d+)', line)
 136        if m:
 137            current['ospf_state'] = m.group(1).rstrip(',')
 138            current['priority'] = int(m.group(2))
 139            continue
 140
 141        # Designated Router (ID) 172.23.52.228 Interface Address 10.150.180.230/28
 142        m = re.search(r'Designated Router \(ID\) (\S+)\s+Interface Address (\S+)', line)
 143        if m and 'Backup' not in line:
 144            current['dr_id'] = m.group(1)
 145            current['dr_address'] = m.group(2)
 146            continue
 147
 148        # Backup Designated Router (ID) 10.150.180.224, Interface Address 10.150.180.224
 149        m = re.search(r'Backup Designated Router \(ID\) (\S+?),\s*Interface Address (\S+)', line)
 150        if m:
 151            current['bdr_id'] = m.group(1)
 152            current['bdr_address'] = m.group(2)
 153            continue
 154
 155        # Timer intervals configured, Hello 10s, Dead 40s, Wait 40s, Retransmit 5
 156        m = re.search(r'Hello (\d+)s,\s*Dead (\d+)s,.*Retransmit (\d+)', line)
 157        if m:
 158            current['hello_interval'] = int(m.group(1))
 159            current['dead_interval'] = int(m.group(2))
 160            current['retransmit_interval'] = int(m.group(3))
 161            continue
 162
 163        # No Hellos (Passive interface)
 164        if 'No Hellos (Passive interface)' in line:
 165            current['passive'] = True
 166            continue
 167
 168        # Neighbor Count is 1, Adjacent neighbor count is 1
 169        m = re.search(r'Neighbor Count is (\d+),\s*Adjacent neighbor count is (\d+)', line)
 170        if m:
 171            current['neighbor_count'] = int(m.group(1))
 172            current['adjacent_neighbor_count'] = int(m.group(2))
 173            continue
 174
 175    return result
 176
 177
 178def get_ospf_interfaces(grade: Grade0, machine_name: str, step: int = 1,
 179                        allow_error: bool = False) -> Dict[str, Dict[str, Any]]:
 180    """Run 'vtysh -c "show ip ospf interface"' and parse the output.
 181
 182    Returns a dict mapping interface name -> dict of parsed fields:
 183      state            : 'up' | 'down'
 184      ospf_enabled     : bool  (False for "OSPF not enabled on this interface" entries)
 185      mtu              : int
 186      bandwidth        : int   (Mbit, the "BW" field: interface speed or `bandwidth` setting)
 187      internet_address : str  e.g. '10.150.180.224/28'
 188      broadcast        : str  e.g. '10.150.180.239'
 189      area             : str  e.g. '0.0.0.0'
 190      router_id        : str
 191      network_type     : str  e.g. 'BROADCAST'
 192      cost             : int
 193      ospf_state       : str  e.g. 'Backup', 'DR', 'DROther'
 194      priority         : int
 195      passive          : bool ("No Hellos (Passive interface)")
 196      hello_interval   : int  (seconds)
 197      dead_interval    : int  (seconds)
 198      retransmit_interval : int (seconds)
 199      neighbor_count          : int
 200      adjacent_neighbor_count : int
 201      dr_id            : str | None
 202      dr_address       : str | None
 203      bdr_id           : str | None
 204      bdr_address      : str | None
 205
 206    Returns {} on error or empty output.
 207    """
 208    output, code = _test(grade, machine_name, 'vtysh -c "show ip ospf interface"', step, allow_error)
 209    if code != 0 or not output:
 210        return {}
 211    return parse_ospf_interfaces(output)
 212
 213
 214# ---------------------------------------------------------------------------
 215# show ip ospf neighbor
 216# ---------------------------------------------------------------------------
 217
 218def parse_ospf_neighbors(output: str) -> List[Dict[str, Any]]:
 219    """Parse the text of ``show ip ospf neighbor``.
 220
 221    Data rows look like (the "Up Time" column exists in FRR >= 7.5 only)::
 222
 223        Neighbor ID     Pri State           Up Time         Dead Time Address         Interface                        RXmtL RqstL DBsmL
 224        10.0.0.2          1 Full/DR         2m34s             36.245s 10.0.12.2       eth0:10.0.12.1                       0     0     0
 225
 226    Rows are parsed from the end (three counters, interface, address, dead time) so
 227    both layouts work.  Non-data lines are skipped.
 228    """
 229    result: List[Dict[str, Any]] = []
 230    for line in output.splitlines():
 231        fields = line.split()
 232        if len(fields) < 9 or not _IPV4_RE.match(fields[0]):
 233            continue
 234        state, _, role = fields[2].partition('/')
 235        interface, _, local_address = fields[-4].partition(':')
 236        try:
 237            priority: Optional[int] = int(fields[1])
 238        except ValueError:
 239            priority = None
 240        result.append({
 241            'neighbor_id': fields[0],
 242            'priority': priority,
 243            'state': state,
 244            'role': role or None,
 245            'up_time': fields[3] if len(fields) >= 10 else None,
 246            'dead_time': fields[-6],
 247            'address': fields[-5],
 248            'interface': interface,
 249            'local_address': local_address or None,
 250        })
 251    return result
 252
 253
 254def get_ospf_neighbors(grade: Grade0, machine_name: str, step: int = 1,
 255                       allow_error: bool = False) -> List[Dict[str, Any]]:
 256    """Run 'vtysh -c "show ip ospf neighbor"' and return one dict per neighbour:
 257    ``neighbor_id, priority, state ('Full', '2-Way', ...), role ('DR', 'Backup',
 258    'DROther', '-'), up_time, dead_time, address, interface, local_address``.
 259    Returns [] on error or empty output."""
 260    output, code = _test(grade, machine_name, 'vtysh -c "show ip ospf neighbor"', step, allow_error)
 261    if code != 0 or not output:
 262        return []
 263    return parse_ospf_neighbors(output)
 264
 265
 266def find_neighbor(neighbors: List[Dict[str, Any]], neighbor_id: str = None, interface: str = None,
 267                  address: str = None) -> Optional[Dict[str, Any]]:
 268    """Return the first neighbour of :func:`get_ospf_neighbors` matching every given criterion."""
 269    for n in neighbors:
 270        if neighbor_id is not None and n['neighbor_id'] != str(neighbor_id):
 271            continue
 272        if interface is not None and n['interface'] != interface:
 273            continue
 274        if address is not None and n['address'] != str(address):
 275            continue
 276        return n
 277    return None
 278
 279
 280def neighbor_is_full(neighbors: List[Dict[str, Any]], neighbor_id: str = None, interface: str = None,
 281                     address: str = None) -> bool:
 282    """True when a neighbour matching the criteria exists and is in state Full."""
 283    n = find_neighbor(neighbors, neighbor_id=neighbor_id, interface=interface, address=address)
 284    return n is not None and n['state'] == 'Full'
 285
 286
 287# ---------------------------------------------------------------------------
 288# show ip ospf
 289# ---------------------------------------------------------------------------
 290
 291def parse_ospf_info(output: str) -> Dict[str, Any]:
 292    """Parse the text of ``show ip ospf``.
 293
 294    Returns ``{}`` when no router id is found (OSPF not running), else::
 295
 296        {'router_id': '10.0.0.4', 'abr': True, 'asbr': False,
 297         'areas': {'0.0.0.0': {'type': 'backbone', 'no_summary': False, 'interfaces': 3,
 298                               'active_interfaces': 3, 'full_neighbors': 2,
 299                               'authentication': 'no'}, ...}}
 300
 301    ``type`` is ``'backbone'``, ``'stub'``, ``'nssa'`` or ``'normal'``;
 302    ``authentication`` is ``'no'``, ``'simple password'`` or ``'message digest'``.
 303    """
 304    info: Dict[str, Any] = {'router_id': None, 'abr': False, 'asbr': False, 'areas': {}}
 305    current: Dict[str, Any] | None = None
 306
 307    for line in output.splitlines():
 308        m = re.search(r'OSPF Routing Process, Router ID:\s*(\S+)', line)
 309        if m:
 310            info['router_id'] = m.group(1)
 311            continue
 312        if 'This router is an ABR' in line:
 313            info['abr'] = True
 314            continue
 315        if 'This router is an ASBR' in line:
 316            info['asbr'] = True
 317            continue
 318        m = re.match(r'^\s*Area ID:\s*(\S+)(?:\s*\((.*?)\))?', line)
 319        if m:
 320            desc = (m.group(2) or '').strip()
 321            low = desc.lower()
 322            if low.startswith('backbone'):
 323                area_type = 'backbone'
 324            elif low.startswith('stub'):
 325                area_type = 'stub'
 326            elif low.startswith('nssa'):
 327                area_type = 'nssa'
 328            else:
 329                area_type = 'normal'
 330            current = {
 331                'type': area_type,
 332                'no_summary': 'no summary' in low,
 333                'interfaces': None, 'active_interfaces': None,
 334                'full_neighbors': None, 'authentication': None,
 335            }
 336            info['areas'][m.group(1)] = current
 337            continue
 338        if current is None:
 339            continue
 340        m = re.search(r'Number of interfaces in this area:\s*Total:\s*(\d+),\s*Active:\s*(\d+)', line)
 341        if m:
 342            current['interfaces'] = int(m.group(1))
 343            current['active_interfaces'] = int(m.group(2))
 344            continue
 345        m = re.search(r'Number of fully adjacent neighbors in this area:\s*(\d+)', line)
 346        if m:
 347            current['full_neighbors'] = int(m.group(1))
 348            continue
 349        m = re.search(r'Area has (.+?) authentication', line)
 350        if m:
 351            current['authentication'] = m.group(1).strip()
 352            continue
 353
 354    if info['router_id'] is None:
 355        return {}
 356    return info
 357
 358
 359def get_ospf_info(grade: Grade0, machine_name: str, step: int = 1, allow_error: bool = False) -> Dict[str, Any]:
 360    """Run 'vtysh -c "show ip ospf"' and return the router-level state (see :func:`parse_ospf_info`).
 361    Returns {} when OSPF is not running."""
 362    output, code = _test(grade, machine_name, 'vtysh -c "show ip ospf"', step, allow_error)
 363    if code != 0 or not output:
 364        return {}
 365    return parse_ospf_info(output)
 366
 367
 368def router_id_map(grade: Grade0, machine_names, step: int = 1, allow_error: bool = True) -> Dict[str, str]:
 369    """Return ``{router_id: machine_name}`` for every machine whose OSPF process answers."""
 370    result: Dict[str, str] = {}
 371    for m in machine_names:
 372        rid = get_ospf_info(grade, m, step=step, allow_error=allow_error).get('router_id')
 373        if rid:
 374            result[rid] = m
 375    return result
 376
 377
 378# ---------------------------------------------------------------------------
 379# show ip ospf route
 380# ---------------------------------------------------------------------------
 381
 382# "N IA 10.0.5.0/24 [30] area: 0.0.0.0", "D IA 10.37.4.0/22 Discard entry",
 383# "N E2 203.0.113.0/24 [10/20] tag: 0" and, in the router table, "R 1.1.1.1 IA [400] area: 0.0.0.2, ASBR"
 384# (the type marker follows the router id there).
 385_OSPF_ROUTE_RE = re.compile(
 386    r'^(?P<kind>[NRD])\s+(?P<type>IA|E1|E2|N1|N2)?\s*'
 387    r'(?P<dest>\d{1,3}(?:\.\d{1,3}){3}(?:/\d{1,2})?)\s+'
 388    r'(?:(?P<type2>IA|E1|E2|N1|N2)\s+)?'
 389    r'(?:\[(?P<cost>\d+)(?:/(?P<ext_cost>\d+))?\]|(?P<discard>Discard entry))'
 390    r'(?P<rest>.*)$'
 391)
 392
 393
 394def parse_ospf_routes(output: str) -> Dict[str, Dict[str, Any]]:
 395    """Parse the text of ``show ip ospf route``.
 396
 397    Network entries (``N``) and discard entries (``D``, created by ``area range``) are
 398    keyed by prefix, router entries (``R``) by router id::
 399
 400        {'10.0.5.0/24': {'kind': 'N', 'type': 'IA', 'cost': 30, 'ext_cost': None, 'tag': None,
 401                         'area': '0.0.0.0', 'abr': False, 'asbr': False,
 402                         'nexthops': [('10.0.12.3', 'eth0')]},
 403         '203.0.113.0/24': {'kind': 'N', 'type': 'E2', 'cost': 10, 'ext_cost': 20, 'tag': 0, ...},
 404         '10.0.0.3': {'kind': 'R', 'type': 'intra', 'cost': 10, 'area': '0.0.0.0', 'abr': True, ...}}
 405
 406    ``type`` is ``'intra'`` for plain intra-area entries, ``'IA'``, ``'E1'``, ``'E2'``,
 407    ``'N1'`` or ``'N2'``.  ``nexthops`` holds ``(ip, interface)`` pairs, ``ip`` being
 408    ``None`` for ``directly attached to`` entries.
 409    """
 410    result: Dict[str, Dict[str, Any]] = {}
 411    current: Dict[str, Any] | None = None
 412
 413    for line in output.splitlines():
 414        m = _OSPF_ROUTE_RE.match(line.strip())
 415        if m:
 416            rest = m.group('rest') or ''
 417            area_m = re.search(r'area:\s*(\S+?)(?:,|\s|$)', rest)
 418            tag_m = re.search(r'tag:\s*(\d+)', rest)
 419            current = {
 420                'kind': m.group('kind'),
 421                'type': m.group('type') or m.group('type2') or 'intra',
 422                'cost': int(m.group('cost')) if m.group('cost') else None,
 423                'ext_cost': int(m.group('ext_cost')) if m.group('ext_cost') else None,
 424                'tag': int(tag_m.group(1)) if tag_m else None,
 425                'area': area_m.group(1) if area_m else None,
 426                'abr': 'ABR' in rest,
 427                'asbr': 'ASBR' in rest,
 428                'discard': bool(m.group('discard')),
 429                'nexthops': [],
 430            }
 431            result[m.group('dest')] = current
 432            continue
 433        if current is None:
 434            continue
 435        m = re.search(r'via (\S+?),\s*(\S+)', line)
 436        if m:
 437            current['nexthops'].append((m.group(1), m.group(2)))
 438            continue
 439        m = re.search(r'directly attached to (\S+)', line)
 440        if m:
 441            current['nexthops'].append((None, m.group(1)))
 442            continue
 443
 444    return result
 445
 446
 447def get_ospf_routes(grade: Grade0, machine_name: str, step: int = 1, allow_error: bool = False) -> Dict[str, Dict[str, Any]]:
 448    """Run 'vtysh -c "show ip ospf route"' and parse it (see :func:`parse_ospf_routes`).
 449    Returns {} on error or empty output."""
 450    output, code = _test(grade, machine_name, 'vtysh -c "show ip ospf route"', step, allow_error)
 451    if code != 0 or not output:
 452        return {}
 453    return parse_ospf_routes(output)
 454
 455
 456# ---------------------------------------------------------------------------
 457# ip -j route  (kernel routing table)
 458# ---------------------------------------------------------------------------
 459
 460def parse_kernel_routes(output: str) -> Dict[str, Dict[str, Any]]:
 461    """Parse the JSON of ``ip -j route``.
 462
 463    Returns ``{prefix: {'protocol': str, 'metric': int, 'scope': str | None,
 464    'nexthops': [(gateway | None, dev)]}}``; ``default`` becomes ``0.0.0.0/0`` and
 465    host routes get ``/32``.  ECMP routes list every next hop.  When the same prefix
 466    appears several times (different metrics) the first entry wins.
 467    """
 468    try:
 469        routes = json.loads(output)
 470    except (ValueError, TypeError):
 471        return {}
 472    if not isinstance(routes, list):
 473        return {}
 474    result: Dict[str, Dict[str, Any]] = {}
 475    for r in routes:
 476        if not isinstance(r, dict):
 477            continue
 478        dst = r.get('dst')
 479        if not dst:
 480            continue
 481        if dst == 'default':
 482            dst = '0.0.0.0/0'
 483        elif '/' not in dst:
 484            dst += '/32'
 485        if dst in result:
 486            continue
 487        if isinstance(r.get('nexthops'), list):
 488            nexthops = [(n.get('gateway'), n.get('dev')) for n in r['nexthops'] if isinstance(n, dict)]
 489        else:
 490            nexthops = [(r.get('gateway'), r.get('dev'))]
 491        result[dst] = {
 492            'protocol': str(r.get('protocol', '')),
 493            'metric': int(r.get('metric', 0) or 0),
 494            'scope': r.get('scope'),
 495            'nexthops': nexthops,
 496        }
 497    return result
 498
 499
 500def get_kernel_routes(grade: Grade0, machine_name: str, step: int = 1, allow_error: bool = False) -> Dict[str, Dict[str, Any]]:
 501    """Run 'ip -j route' and parse it (see :func:`parse_kernel_routes`). Returns {} on error."""
 502    output, code = _test(grade, machine_name, 'ip -j route', step, allow_error)
 503    if code != 0 or not output:
 504        return {}
 505    return parse_kernel_routes(output)
 506
 507
 508def kernel_route_gateways(routes: Dict[str, Dict[str, Any]], prefix, ospf_only: bool = True,
 509                          protocols=None) -> List[str]:
 510    """Return the gateway list of ``prefix`` in a :func:`parse_kernel_routes` result.
 511
 512    ``[]`` when the prefix is absent or when its protocol is not accepted: by default
 513    only OSPF routes are (``ospf_only``); ``protocols`` (e.g. ``BGP_ROUTE_PROTOCOLS``)
 514    replaces that filter, ``ospf_only=False`` and ``protocols=None`` accept any route.
 515    """
 516    entry = routes.get(str(prefix))
 517    if entry is None:
 518        return []
 519    if protocols is not None:
 520        if str(entry['protocol']) not in protocols:
 521            return []
 522    elif ospf_only and not is_ospf_protocol(entry['protocol']):
 523        return []
 524    return [gw for gw, _dev in entry['nexthops'] if gw]
 525
 526
 527# ---------------------------------------------------------------------------
 528# show running-config
 529# ---------------------------------------------------------------------------
 530
 531def _empty_interface_config() -> Dict[str, Any]:
 532    return {
 533        'cost': None, 'priority': None, 'hello_interval': None, 'dead_interval': None,
 534        'authentication': None, 'message_digest_keys': {}, 'authentication_key': None,
 535        'area': None, 'bandwidth': None, 'passive': None, 'network_type': None,
 536        'addresses': [],
 537    }
 538
 539
 540def _empty_area_config() -> Dict[str, Any]:
 541    return {'stub': False, 'nssa': False, 'no_summary': False, 'ranges': [], 'authentication': None,
 542            'default_cost': None}
 543
 544
 545def _empty_bgp_neighbor_config() -> Dict[str, Any]:
 546    return {
 547        'remote_as': None, 'peer_group': None, 'description': None, 'update_source': None,
 548        'timers': None, 'shutdown': False, 'password': None, 'ebgp_multihop': None,
 549        'next_hop_self': False, 'route_reflector_client': False, 'default_originate': False,
 550        'default_originate_route_map': None, 'route_map_in': None, 'route_map_out': None,
 551        'prefix_list_in': None, 'prefix_list_out': None, 'filter_list_in': None, 'filter_list_out': None,
 552        'soft_reconfiguration': False, 'send_community': True, 'activate': None,
 553    }
 554
 555
 556def _empty_router_bgp_config() -> Dict[str, Any]:
 557    return {
 558        'present': False, 'as': None, 'router_id': None, 'ebgp_requires_policy': True,
 559        'network_import_check': True, 'default_ipv4_unicast': True, 'bestpath': [], 'cluster_id': None,
 560        'default_local_preference': None, 'timers': None, 'neighbors': {}, 'networks': [],
 561        'aggregate_addresses': [], 'redistribute': [],
 562    }
 563
 564
 565def _parse_router_bgp_line(rb: Dict[str, Any], words: List[str], neighbor_cfg) -> None:
 566    """Apply one indented ``router bgp`` statement (router level or address-family) to ``rb``."""
 567    if not words:
 568        return
 569    negated = words[0] == 'no'
 570    w = words[1:] if negated else words
 571    if not w:
 572        return
 573    if w[0] in ('address-family', 'exit-address-family'):
 574        return
 575    if w[0] == 'neighbor' and len(w) >= 3:
 576        n = neighbor_cfg(w[1])
 577        kw, arg = w[2], (w[3] if len(w) >= 4 else None)
 578        if kw == 'remote-as' and arg:
 579            n['remote_as'] = int(arg) if arg.isdigit() else arg
 580        elif kw == 'peer-group' and arg:
 581            n['peer_group'] = arg
 582        elif kw == 'description':
 583            n['description'] = " ".join(w[3:]) or None
 584        elif kw == 'update-source' and arg:
 585            n['update_source'] = arg
 586        elif kw == 'timers' and len(w) >= 5 and w[3].isdigit() and w[4].isdigit():
 587            n['timers'] = (int(w[3]), int(w[4]))
 588        elif kw == 'shutdown':
 589            n['shutdown'] = not negated
 590        elif kw == 'password':
 591            n['password'] = None if negated else arg
 592        elif kw == 'ebgp-multihop':
 593            n['ebgp_multihop'] = None if negated else (int(arg) if arg and arg.isdigit() else 255)
 594        elif kw == 'next-hop-self':
 595            n['next_hop_self'] = not negated
 596        elif kw == 'route-reflector-client':
 597            n['route_reflector_client'] = not negated
 598        elif kw == 'default-originate':
 599            n['default_originate'] = not negated
 600            if not negated and len(w) >= 5 and w[3] == 'route-map':
 601                n['default_originate_route_map'] = w[4]
 602        elif kw == 'route-map' and len(w) >= 5 and w[4] in ('in', 'out'):
 603            n[f'route_map_{w[4]}'] = None if negated else w[3]
 604        elif kw == 'prefix-list' and len(w) >= 5 and w[4] in ('in', 'out'):
 605            n[f'prefix_list_{w[4]}'] = None if negated else w[3]
 606        elif kw == 'filter-list' and len(w) >= 5 and w[4] in ('in', 'out'):
 607            n[f'filter_list_{w[4]}'] = None if negated else w[3]
 608        elif kw == 'soft-reconfiguration':
 609            n['soft_reconfiguration'] = not negated
 610        elif kw == 'send-community':
 611            n['send_community'] = not negated
 612        elif kw == 'activate':
 613            n['activate'] = not negated
 614        return
 615    if w[0] == 'bgp' and len(w) >= 2:
 616        if w[1] == 'router-id' and len(w) >= 3:
 617            rb['router_id'] = None if negated else w[2]
 618        elif w[1] == 'ebgp-requires-policy':
 619            rb['ebgp_requires_policy'] = not negated
 620        elif w[1] == 'network' and len(w) >= 3 and w[2] == 'import-check':
 621            rb['network_import_check'] = not negated
 622        elif w[1] == 'default' and len(w) >= 3 and w[2] == 'ipv4-unicast':
 623            rb['default_ipv4_unicast'] = not negated
 624        elif w[1] == 'default' and len(w) >= 4 and w[2] == 'local-preference':
 625            rb['default_local_preference'] = None if negated else int(w[3])
 626        elif w[1] == 'bestpath' and len(w) >= 3:
 627            rb['bestpath'].append(" ".join(w[2:]))
 628        elif w[1] == 'cluster-id' and len(w) >= 3:
 629            rb['cluster_id'] = None if negated else w[2]
 630        return
 631    if w[0] == 'network' and len(w) >= 2:
 632        if negated:
 633            if w[1] in rb['networks']:
 634                rb['networks'].remove(w[1])
 635        else:
 636            rb['networks'].append(w[1])
 637    elif w[0] == 'aggregate-address' and len(w) >= 2 and not negated:
 638        rb['aggregate_addresses'].append((w[1], w[2:]))
 639    elif w[0] == 'redistribute' and len(w) >= 2 and not negated:
 640        rb['redistribute'].append(w[1])
 641    elif w[0] == 'timers' and len(w) >= 4 and w[1] == 'bgp' and w[2].isdigit() and w[3].isdigit():
 642        rb['timers'] = (int(w[2]), int(w[3]))
 643
 644
 645def parse_frr_config(text: str) -> Dict[str, Any]:
 646    """Parse an FRR configuration (``show running-config`` or ``frr.conf``).
 647
 648    Returns::
 649
 650        {'hostname': str | None,
 651         'static_routes': [(prefix, nexthop)],       # nexthop may be 'blackhole' / 'Null0'
 652         'router_ospf': {'present': bool, 'router_id': str | None,
 653                         'networks': [(prefix, area)], 'passive_interfaces': [...],
 654                         'passive_default': bool,
 655                         'areas': {area: {'stub', 'nssa', 'no_summary', 'ranges': [prefix],
 656                                          'authentication', 'default_cost'}},
 657                         'redistribute': [protocol, ...],
 658                         'default_information_originate': bool,
 659                         'default_information_always': bool,
 660                         'auto_cost_reference_bandwidth': int | None},
 661         'interfaces': {name: {'cost', 'priority', 'hello_interval', 'dead_interval',
 662                               'authentication' ('message-digest' | 'simple' | 'null' | 'key-chain'),
 663                               'message_digest_keys': {id: key}, 'authentication_key',
 664                               'area', 'bandwidth', 'passive' (True/False/None), 'network_type',
 665                               'addresses': ['10.0.0.1/32', ...]}},
 666         'router_bgp': {'present': bool, 'as': int | None, 'router_id': str | None,
 667                        'ebgp_requires_policy': bool, 'network_import_check': bool,
 668                        'default_ipv4_unicast': bool, 'bestpath': ['compare-routerid', ...],
 669                        'cluster_id', 'default_local_preference', 'timers': (keepalive, hold) | None,
 670                        'neighbors': {peer: {'remote_as': int | str | None, 'peer_group', 'description',
 671                                             'update_source', 'timers': (keepalive, hold) | None,
 672                                             'shutdown': bool, 'password', 'ebgp_multihop': int | None,
 673                                             'next_hop_self': bool, 'route_reflector_client': bool,
 674                                             'default_originate': bool, 'default_originate_route_map',
 675                                             'route_map_in', 'route_map_out', 'prefix_list_in',
 676                                             'prefix_list_out', 'filter_list_in', 'filter_list_out',
 677                                             'soft_reconfiguration': bool, 'send_community': bool,
 678                                             'activate': bool | None}},
 679                        'networks': [prefix], 'aggregate_addresses': [(prefix, [option, ...])],
 680                        'redistribute': [protocol, ...]},
 681         'prefix_lists': {name: [{'seq': int | None, 'action': 'permit' | 'deny', 'prefix': str,
 682                                  'ge': int | None, 'le': int | None}]},
 683         'route_maps': {name: [{'seq': int, 'action': 'permit' | 'deny',
 684                                'match': ['ip address prefix-list P', ...],
 685                                'set': ['local-preference 200', ...]}]},
 686         'as_path_lists': {name: [(seq | None, action, regexp)]},
 687         'community_lists': {name: [(seq | None, action, 'community ...')]}}
 688
 689    Area ids are normalised to dotted form (``area 1`` -> ``'0.0.0.1'``).  Both the
 690    router-level ``passive-interface X`` and the interface-level ``ip ospf passive``
 691    syntaxes are recognised; ``interfaces[X]['passive']`` reflects only the
 692    interface-level statement, ``router_ospf['passive_interfaces']`` the router-level one.
 693    In ``router bgp`` the statements of the ``address-family ipv4 unicast`` block and the
 694    router-level ones are merged (FRR moves ``network``, ``next-hop-self``, ``route-map``...
 695    into the address-family block in ``show running-config``).
 696    """
 697    cfg: Dict[str, Any] = {
 698        'hostname': None,
 699        'static_routes': [],
 700        'router_ospf': {
 701            'present': False, 'router_id': None, 'networks': [], 'passive_interfaces': [],
 702            'passive_default': False, 'areas': {}, 'redistribute': [],
 703            'default_information_originate': False, 'default_information_always': False,
 704            'auto_cost_reference_bandwidth': None,
 705        },
 706        'interfaces': {},
 707        'router_bgp': _empty_router_bgp_config(),
 708        'prefix_lists': {},
 709        'route_maps': {},
 710        'as_path_lists': {},
 711        'community_lists': {},
 712    }
 713    section: Optional[str] = None  # 'router_ospf' | 'router_bgp' | 'interface' | 'route_map' | 'other' | None
 714    iface: Dict[str, Any] | None = None
 715    rm_entry: Dict[str, Any] | None = None
 716    ro = cfg['router_ospf']
 717    rb = cfg['router_bgp']
 718
 719    def neighbor_cfg(peer: str) -> Dict[str, Any]:
 720        if peer not in rb['neighbors']:
 721            rb['neighbors'][peer] = _empty_bgp_neighbor_config()
 722        return rb['neighbors'][peer]
 723
 724    def seq_of(w: List[str]):
 725        """Return (seq, rest) for a 'seq N ...' prefix of the word list."""
 726        if len(w) >= 2 and w[0] == 'seq' and w[1].isdigit():
 727            return int(w[1]), w[2:]
 728        return None, w
 729
 730    def area_cfg(area: str) -> Dict[str, Any]:
 731        key = normalize_area(area)
 732        if key not in ro['areas']:
 733            ro['areas'][key] = _empty_area_config()
 734        return ro['areas'][key]
 735
 736    for raw in text.splitlines():
 737        line = raw.rstrip()
 738        if not line.strip() or line.strip() == '!':
 739            continue
 740        stripped = line.strip()
 741        words = stripped.split()
 742
 743        if not line[0].isspace():
 744            # top-level line: opens a section or is a global statement
 745            if stripped in ('exit', 'end', 'exit-address-family'):
 746                section, iface, rm_entry = None, None, None
 747                continue
 748            if words[0] == 'interface' and len(words) >= 2:
 749                section = 'interface'
 750                iface = cfg['interfaces'].setdefault(words[1], _empty_interface_config())
 751                continue
 752            if words[0] == 'router' and len(words) >= 2 and words[1] == 'ospf':
 753                section = 'router_ospf'
 754                iface = None
 755                ro['present'] = True
 756                continue
 757            if words[0] == 'router' and len(words) >= 3 and words[1] == 'bgp':
 758                section, iface = 'router_bgp', None
 759                rb['present'] = True
 760                rb['as'] = int(words[2]) if words[2].isdigit() else words[2]
 761                continue
 762            if words[0] == 'route-map' and len(words) >= 4 and words[2] in ('permit', 'deny'):
 763                section, iface = 'route_map', None
 764                rm_entry = {'seq': int(words[3]) if words[3].isdigit() else words[3], 'action': words[2],
 765                            'match': [], 'set': []}
 766                cfg['route_maps'].setdefault(words[1], []).append(rm_entry)
 767                continue
 768            if words[0] == 'router' or words[0] in ('vrf', 'route-map', 'key', 'line', 'segment-routing'):
 769                section, iface = 'other', None
 770                continue
 771            section, iface, rm_entry = None, None, None
 772            if words[0] == 'hostname' and len(words) >= 2:
 773                cfg['hostname'] = words[1]
 774            elif words[0] == 'ip' and len(words) >= 4 and words[1] == 'route':
 775                cfg['static_routes'].append((words[2], words[3]))
 776            elif words[:2] == ['ip', 'prefix-list'] and len(words) >= 5:
 777                # ip prefix-list NAME [seq N] permit|deny PREFIX [ge X] [le Y]
 778                seq, rest = seq_of(words[3:])
 779                if len(rest) >= 2 and rest[0] in ('permit', 'deny'):
 780                    entry = {'seq': seq, 'action': rest[0], 'prefix': rest[1], 'ge': None, 'le': None}
 781                    for i in range(2, len(rest) - 1):
 782                        if rest[i] in ('ge', 'le') and rest[i + 1].isdigit():
 783                            entry[rest[i]] = int(rest[i + 1])
 784                    cfg['prefix_lists'].setdefault(words[2], []).append(entry)
 785            elif words[:3] == ['bgp', 'as-path', 'access-list'] and len(words) >= 6:
 786                # bgp as-path access-list NAME [seq N] permit|deny REGEXP
 787                seq, rest = seq_of(words[4:])
 788                if len(rest) >= 2 and rest[0] in ('permit', 'deny'):
 789                    cfg['as_path_lists'].setdefault(words[3], []).append((seq, rest[0], " ".join(rest[1:])))
 790            elif words[:2] == ['bgp', 'community-list'] and len(words) >= 5:
 791                # bgp community-list [standard|expanded] NAME [seq N] permit|deny COMMUNITY...
 792                rest = words[2:]
 793                if rest and rest[0] in ('standard', 'expanded'):
 794                    rest = rest[1:]
 795                if len(rest) >= 3:
 796                    name = rest[0]
 797                    seq, rest2 = seq_of(rest[1:])
 798                    if len(rest2) >= 2 and rest2[0] in ('permit', 'deny'):
 799                        cfg['community_lists'].setdefault(name, []).append((seq, rest2[0], " ".join(rest2[1:])))
 800            continue
 801
 802        # indented line: belongs to the current section
 803        if section == 'interface' and iface is not None:
 804            if words[:2] == ['ip', 'ospf'] and len(words) >= 3:
 805                sub = words[2]
 806                arg = words[3] if len(words) >= 4 else None
 807                if sub == 'cost' and arg:
 808                    iface['cost'] = int(arg)
 809                elif sub == 'priority' and arg:
 810                    iface['priority'] = int(arg)
 811                elif sub == 'hello-interval' and arg:
 812                    iface['hello_interval'] = int(arg)
 813                elif sub == 'dead-interval' and arg:
 814                    iface['dead_interval'] = int(arg) if arg.isdigit() else arg
 815                elif sub == 'authentication':
 816                    iface['authentication'] = arg if arg else 'simple'
 817                elif sub == 'authentication-key' and arg:
 818                    iface['authentication_key'] = arg
 819                elif sub == 'message-digest-key' and len(words) >= 6 and words[4] == 'md5':
 820                    iface['message_digest_keys'][int(arg)] = words[5]
 821                elif sub == 'area' and arg:
 822                    iface['area'] = normalize_area(arg)
 823                elif sub == 'passive':
 824                    iface['passive'] = True
 825                elif sub == 'network' and arg:
 826                    iface['network_type'] = arg
 827            elif words[:3] == ['no', 'ip', 'ospf'] and len(words) >= 4 and words[3] == 'passive':
 828                iface['passive'] = False
 829            elif words[0] == 'bandwidth' and len(words) >= 2:
 830                iface['bandwidth'] = int(words[1])
 831            elif words[:2] == ['ip', 'address'] and len(words) >= 3:
 832                iface['addresses'].append(words[2])
 833            continue
 834
 835        if section == 'route_map' and rm_entry is not None:
 836            if words[0] == 'match' and len(words) >= 2:
 837                rm_entry['match'].append(" ".join(words[1:]))
 838            elif words[0] == 'set' and len(words) >= 2:
 839                rm_entry['set'].append(" ".join(words[1:]))
 840            continue
 841
 842        if section == 'router_bgp':
 843            _parse_router_bgp_line(rb, words, neighbor_cfg)
 844            continue
 845
 846        if section == 'router_ospf':
 847            if words[0] in ('ospf', 'router-id') and 'router-id' in words:
 848                ro['router_id'] = words[-1]
 849            elif words[0] == 'network' and len(words) >= 4 and words[2] == 'area':
 850                ro['networks'].append((words[1], normalize_area(words[3])))
 851            elif words[0] == 'passive-interface' and len(words) >= 2:
 852                if words[1] == 'default':
 853                    ro['passive_default'] = True
 854                else:
 855                    ro['passive_interfaces'].append(words[1])
 856            elif words[0] == 'no' and len(words) >= 3 and words[1] == 'passive-interface':
 857                if words[2] in ro['passive_interfaces']:
 858                    ro['passive_interfaces'].remove(words[2])
 859            elif words[0] == 'area' and len(words) >= 3:
 860                a = area_cfg(words[1])
 861                kw = words[2]
 862                if kw == 'stub':
 863                    a['stub'] = True
 864                    a['no_summary'] = 'no-summary' in words[3:]
 865                elif kw == 'nssa':
 866                    a['nssa'] = True
 867                    a['no_summary'] = 'no-summary' in words[3:]
 868                elif kw == 'range' and len(words) >= 4:
 869                    a['ranges'].append(words[3])
 870                elif kw == 'authentication':
 871                    a['authentication'] = words[3] if len(words) >= 4 else 'simple'
 872                elif kw == 'default-cost' and len(words) >= 4:
 873                    a['default_cost'] = int(words[3])
 874            elif words[0] == 'redistribute' and len(words) >= 2:
 875                ro['redistribute'].append(words[1])
 876            elif words[:2] == ['default-information', 'originate']:
 877                ro['default_information_originate'] = True
 878                ro['default_information_always'] = 'always' in words[2:]
 879            elif words[:2] == ['auto-cost', 'reference-bandwidth'] and len(words) >= 3:
 880                ro['auto_cost_reference_bandwidth'] = int(words[2])
 881            continue
 882
 883    return cfg
 884
 885
 886def get_frr_running_config(grade: Grade0, machine_name: str, step: int = 1, allow_error: bool = False) -> Dict[str, Any]:
 887    """Run 'vtysh -c "show running-config"' and parse it (see :func:`parse_frr_config`).
 888    Returns {} on error or empty output."""
 889    output, code = _test(grade, machine_name, 'vtysh -c "show running-config"', step, allow_error)
 890    if code != 0 or not output:
 891        return {}
 892    return parse_frr_config(output)
 893
 894
 895def expected_auto_cost(reference_bandwidth: int, interface_bandwidth: int) -> int:
 896    """OSPF cost FRR computes for an interface: ``round(reference / bandwidth)`` clamped to
 897    ``[1, 65535]`` (both in Mbit/s; the default reference is 100000 and a veth reports 10000,
 898    hence the default cost of 10)."""
 899    if not interface_bandwidth:
 900        return 10
 901    cost = int(reference_bandwidth / interface_bandwidth + 0.5)
 902    return max(1, min(65535, cost))
 903
 904
 905# ---------------------------------------------------------------------------
 906# BGP (json output of vtysh)
 907# ---------------------------------------------------------------------------
 908
 909def _vtysh_json(grade: Grade0, machine_name: str, command: str, step: int = 1, allow_error: bool = False) -> Any:
 910    """Run ``vtysh -c "<command> json"`` and return the decoded JSON object.
 911
 912    Returns ``{}`` on the registration-pass placeholder, on a non-zero exit code, when
 913    BGP is not configured (``% ...`` error line) or when the output is not a JSON object.
 914    """
 915    output, code = _test(grade, machine_name, f'vtysh -c "{command} json"', step, allow_error)
 916    if code != 0 or not output:
 917        return {}
 918    start = output.find('{')
 919    if start < 0:
 920        return {}
 921    try:
 922        data = json.loads(output[start:])
 923    except (ValueError, TypeError):
 924        return {}
 925    return data if isinstance(data, dict) else {}
 926
 927
 928def _int_or_none(value) -> Optional[int]:
 929    try:
 930        return int(value)
 931    except (TypeError, ValueError):
 932        return None
 933
 934
 935def as_list_from_string(path) -> List[int]:
 936    """``'65100 65000'`` -> ``[65100, 65000]`` (AS sets ``{65100,65000}`` keep their members)."""
 937    if not path:
 938        return []
 939    return [int(tok) for tok in re.findall(r'\d+', str(path))]
 940
 941
 942# -- show bgp ipv4 unicast summary ------------------------------------------
 943
 944def parse_bgp_summary(data: Any) -> Dict[str, Any]:
 945    """Parse the JSON of ``show bgp ipv4 unicast summary``.
 946
 947    Accepts the flat address-family object or the ``{'ipv4Unicast': {...}}`` wrapper of
 948    ``show bgp summary json``.  Returns ``{}`` when no BGP process answers, else::
 949
 950        {'router_id': '10.0.0.1', 'as': 65001,
 951         'peers': {'10.1.1.2': {'remote_as': 65100, 'state': 'Established', 'established': True,
 952                                'policy_blocked': False,   # eBGP peer without in/out policy (RFC 8212)
 953                                'pfx_rcd': 3, 'pfx_snt': 1, 'description': 'ISP-A',
 954                                'uptime': '00:12:34'}}}
 955    """
 956    if not isinstance(data, dict):
 957        return {}
 958    if 'ipv4Unicast' in data and isinstance(data['ipv4Unicast'], dict):
 959        data = data['ipv4Unicast']
 960    if 'routerId' not in data and 'peers' not in data:
 961        return {}
 962    peers: Dict[str, Dict[str, Any]] = {}
 963    for ip, p in (data.get('peers') or {}).items():
 964        if not isinstance(p, dict):
 965            continue
 966        state = p.get('state')
 967        peers[ip] = {
 968            'remote_as': _int_or_none(p.get('remoteAs')),
 969            'state': state,
 970            'established': state == 'Established',
 971            'policy_blocked': p.get('peerState') == 'Policy',
 972            'pfx_rcd': _int_or_none(p.get('pfxRcd')) or 0,
 973            'pfx_snt': _int_or_none(p.get('pfxSnt')) or 0,
 974            'description': p.get('desc'),
 975            'uptime': p.get('peerUptime'),
 976        }
 977    return {'router_id': data.get('routerId'), 'as': _int_or_none(data.get('as')), 'peers': peers}
 978
 979
 980def get_bgp_summary(grade: Grade0, machine_name: str, step: int = 1, allow_error: bool = False) -> Dict[str, Any]:
 981    """Run 'vtysh -c "show bgp ipv4 unicast summary json"' (see :func:`parse_bgp_summary`).
 982    Returns {} when BGP is not running."""
 983    return parse_bgp_summary(_vtysh_json(grade, machine_name, 'show bgp ipv4 unicast summary', step, allow_error))
 984
 985
 986# -- show bgp neighbors X -----------------------------------------------------
 987
 988def parse_bgp_neighbor(data: Any, peer: str = None) -> Dict[str, Any]:
 989    """Parse the JSON of ``show bgp neighbors <peer>``.
 990
 991    The output is keyed by peer address; ``peer`` selects an entry (the first one when
 992    omitted).  Returns ``{}`` when the neighbour does not exist, else::
 993
 994        {'peer': '10.1.1.2', 'remote_as': 65100, 'local_as': 65001, 'state': 'Established',
 995         'established': True, 'link': 'external' | 'internal', 'description': 'ISP-A',
 996         'remote_router_id': '100.100.100.100', 'local_router_id': '10.0.3.1',
 997         'hold_time': 30, 'keepalive': 10,                    # negotiated, seconds
 998         'configured_hold_time': 30, 'configured_keepalive': 10,
 999         'local_host': '10.0.3.1', 'foreign_host': '10.0.3.2', 'update_source': 'lo' | None,
1000         'accepted_prefixes': 3, 'sent_prefixes': 1,
1001         'route_reflector_client': False, 'next_hop_self': False, 'default_originate': False,
1002         'default_originate_route_map': None,
1003         'route_map_in': 'LP-IN', 'route_map_out': None, 'prefix_list_in': None, 'prefix_list_out': None,
1004         'send_community': True, 'af': {...ipv4Unicast...}, 'raw': {...}}
1005    """
1006    if not isinstance(data, dict) or not data:
1007        return {}
1008    if 'bgpNoSuchNeighbor' in data or 'warning' in data:
1009        return {}
1010    if peer is not None:
1011        entry = data.get(str(peer))
1012    else:
1013        entry = next((v for v in data.values() if isinstance(v, dict)), None)
1014    if not isinstance(entry, dict) or 'bgpState' not in entry:
1015        return {}
1016    af = ((entry.get('addressFamilyInfo') or {}).get('ipv4Unicast')) or {}
1017    if not isinstance(af, dict):
1018        af = {}
1019    peer_ip = peer if peer is not None else next(iter(data))
1020
1021    def msecs(key) -> Optional[int]:
1022        v = _int_or_none(entry.get(key))
1023        return None if v is None else v // 1000
1024
1025    if entry.get('nbrExternalLink'):
1026        link = 'external'
1027    elif entry.get('nbrInternalLink'):
1028        link = 'internal'
1029    else:
1030        link = None
1031    return {
1032        'peer': peer_ip,
1033        'remote_as': _int_or_none(entry.get('remoteAs')),
1034        'local_as': _int_or_none(entry.get('localAs')),
1035        'state': entry.get('bgpState'),
1036        'established': entry.get('bgpState') == 'Established',
1037        'link': link,
1038        'description': entry.get('nbrDesc'),
1039        'remote_router_id': entry.get('remoteRouterId'),
1040        'local_router_id': entry.get('localRouterId'),
1041        'hold_time': msecs('bgpTimerHoldTimeMsecs'),
1042        'keepalive': msecs('bgpTimerKeepAliveIntervalMsecs'),
1043        'configured_hold_time': msecs('bgpTimerConfiguredHoldTimeMsecs'),
1044        'configured_keepalive': msecs('bgpTimerConfiguredKeepAliveIntervalMsecs'),
1045        'local_host': entry.get('hostLocal'),
1046        'foreign_host': entry.get('hostForeign'),
1047        'update_source': entry.get('updateSource'),
1048        'accepted_prefixes': _int_or_none(af.get('acceptedPrefixCounter')) or 0,
1049        'sent_prefixes': _int_or_none(af.get('sentPrefixCounter')) or 0,
1050        'route_reflector_client': bool(af.get('routeReflectorClient')),
1051        'next_hop_self': bool(af.get('routerAlwaysNextHop') or af.get('nextHopSelf')),
1052        'default_originate': bool(af.get('defaultSent') or af.get('defaultNotSent')),
1053        'default_originate_route_map': af.get('defaultRouteMap'),
1054        'route_map_in': af.get('routeMapForIncomingAdvertisements'),
1055        'route_map_out': af.get('routeMapForOutgoingAdvertisements'),
1056        'prefix_list_in': af.get('incomingUpdatePrefixFilterList'),
1057        'prefix_list_out': af.get('outgoingUpdatePrefixFilterList'),
1058        'send_community': 'commAttriSentToNbr' in af if af else True,
1059        'af': af,
1060        'raw': entry,
1061    }
1062
1063
1064def get_bgp_neighbor(grade: Grade0, machine_name: str, peer, step: int = 1, allow_error: bool = False) -> Dict[str, Any]:
1065    """Run 'vtysh -c "show bgp neighbors <peer> json"' (see :func:`parse_bgp_neighbor`).
1066    Returns {} when BGP is not running or the neighbour is unknown."""
1067    return parse_bgp_neighbor(_vtysh_json(grade, machine_name, f'show bgp neighbors {peer}', step, allow_error),
1068                              peer=str(peer))
1069
1070
1071# -- show bgp ipv4 unicast (whole table) --------------------------------------
1072
1073def _parse_bgp_path_entry(e: Dict[str, Any]) -> Dict[str, Any]:
1074    """Common fields of a path in ``show bgp ipv4 unicast json`` and ``show bgp ipv4 unicast X json``."""
1075    aspath = e.get('aspath')
1076    if isinstance(aspath, dict):
1077        aspath_str = aspath.get('string') or ''
1078    else:
1079        aspath_str = e.get('path') or ''
1080    if aspath_str == 'Local':
1081        aspath_str = ''
1082    best = e.get('bestpath')
1083    if isinstance(best, dict):
1084        reason = best.get('selectionReason')
1085        best = bool(best.get('overall'))
1086    else:
1087        reason = e.get('selectionReason')
1088        best = bool(best)
1089    nexthops = [n for n in (e.get('nexthops') or []) if isinstance(n, dict)]
1090    used = next((n for n in nexthops if n.get('used')), nexthops[0] if nexthops else {})
1091    peer = e.get('peer')
1092    if isinstance(peer, dict):
1093        peer_id, peer_router_id = peer.get('peerId'), peer.get('routerId')
1094        path_from = peer.get('type') or e.get('pathFrom')
1095    else:
1096        peer_id, peer_router_id = e.get('peerId'), None
1097        path_from = e.get('pathFrom')
1098    community = e.get('community')
1099    if isinstance(community, dict):
1100        communities = list(community.get('list') or [])
1101        if not communities and community.get('string'):
1102            communities = str(community['string']).split()
1103    else:
1104        communities = []
1105    cluster = e.get('clusterList')
1106    if isinstance(cluster, dict):
1107        cluster_list = list(cluster.get('list') or [])
1108    elif isinstance(cluster, list):
1109        cluster_list = list(cluster)
1110    else:
1111        cluster_list = []
1112    return {
1113        'valid': bool(e.get('valid')),
1114        'best': best,
1115        'reason': reason,
1116        'path_from': path_from,
1117        'nexthop': used.get('ip'),
1118        'nexthop_used': bool(used.get('used')),
1119        'accessible': used.get('accessible'),
1120        'peer': peer_id,
1121        'peer_router_id': peer_router_id,
1122        'aspath': aspath_str,
1123        'as_list': as_list_from_string(aspath_str),
1124        'locprf': _int_or_none(e.get('locPrf')),
1125        'med': _int_or_none(e.get('metric')),
1126        'weight': _int_or_none(e.get('weight')) or 0,
1127        'origin': e.get('origin'),
1128        'local': bool(e.get('local') or e.get('sourced') or peer_id in ('(unspec)', '0.0.0.0')),
1129        'originator_id': e.get('originatorId'),
1130        'cluster_list': cluster_list,
1131        'communities': communities,
1132    }
1133
1134
1135def parse_bgp_routes(data: Any) -> Dict[str, List[Dict[str, Any]]]:
1136    """Parse the JSON of ``show bgp ipv4 unicast`` (the whole BGP table).
1137
1138    Returns ``{prefix: [path, ...]}`` (``{}`` when BGP is not running), each path being::
1139
1140        {'valid': True, 'best': True, 'reason': 'Local Pref', 'path_from': 'internal',
1141         'nexthop': '10.0.3.3', 'nexthop_used': True, 'accessible': None,
1142         'peer': '10.0.3.2', 'peer_router_id': None, 'aspath': '65200 65000',
1143         'as_list': [65200, 65000], 'locprf': 200, 'med': 0, 'weight': 0, 'origin': 'IGP',
1144         'local': False, 'originator_id': None, 'cluster_list': [], 'communities': []}
1145
1146    A ``network`` statement whose prefix has no route in the RIB (``bgp network
1147    import-check``) yields a path with ``valid`` and ``best`` both ``False``.
1148    """
1149    if not isinstance(data, dict):
1150        return {}
1151    routes = data.get('routes')
1152    if not isinstance(routes, dict):
1153        return {}
1154    result: Dict[str, List[Dict[str, Any]]] = {}
1155    for prefix, entries in routes.items():
1156        if isinstance(entries, dict):
1157            entries = [entries]
1158        if not isinstance(entries, list):
1159            continue
1160        result[prefix] = [_parse_bgp_path_entry(e) for e in entries if isinstance(e, dict)]
1161    return result
1162
1163
1164def get_bgp_routes(grade: Grade0, machine_name: str, step: int = 1, allow_error: bool = False) -> Dict[str, List[Dict[str, Any]]]:
1165    """Run 'vtysh -c "show bgp ipv4 unicast json"' (see :func:`parse_bgp_routes`). Returns {} on error."""
1166    return parse_bgp_routes(_vtysh_json(grade, machine_name, 'show bgp ipv4 unicast', step, allow_error))
1167
1168
1169def bgp_best_path(routes: Dict[str, List[Dict[str, Any]]], prefix) -> Dict[str, Any]:
1170    """Return the best path of ``prefix`` in a :func:`parse_bgp_routes` result (``{}`` when none)."""
1171    for p in routes.get(str(prefix), []):
1172        if p.get('best'):
1173            return p
1174    return {}
1175
1176
1177def bgp_paths(routes: Dict[str, List[Dict[str, Any]]], prefix, nexthop=None, peer=None, first_as: int = None,
1178              ends_with=None, path_from: str = None, valid: bool = None) -> List[Dict[str, Any]]:
1179    """Return the paths of ``prefix`` matching every given criterion (``ends_with`` is a
1180    list of AS numbers compared with the end of ``as_list``)."""
1181    result = []
1182    for p in routes.get(str(prefix), []):
1183        if nexthop is not None and p.get('nexthop') != str(nexthop):
1184            continue
1185        if peer is not None and p.get('peer') != str(peer):
1186            continue
1187        if first_as is not None and (not p['as_list'] or p['as_list'][0] != int(first_as)):
1188            continue
1189        if ends_with is not None:
1190            tail = [int(a) for a in ends_with]
1191            if p['as_list'][-len(tail):] != tail:
1192                continue
1193        if path_from is not None and p.get('path_from') != path_from:
1194            continue
1195        if valid is not None and bool(p.get('valid')) != valid:
1196            continue
1197        result.append(p)
1198    return result
1199
1200
1201# -- show bgp ipv4 unicast PREFIX ---------------------------------------------
1202
1203def parse_bgp_prefix(data: Any) -> Dict[str, Any]:
1204    """Parse the JSON of ``show bgp ipv4 unicast <prefix>`` (detailed paths).
1205
1206    Returns ``{'prefix': None, 'paths': []}`` when the prefix is unknown, else
1207    ``{'prefix': '10.1.0.0/24', 'paths': [path, ...]}`` with the fields of
1208    :func:`parse_bgp_routes` plus ``accessible``, ``peer_router_id``, ``originator_id``,
1209    ``cluster_list`` (route reflection) and ``communities`` (``['65001:300', ...]``).
1210    """
1211    if not isinstance(data, dict) or not isinstance(data.get('paths'), list):
1212        return {'prefix': None, 'paths': []}
1213    return {'prefix': data.get('prefix'),
1214            'paths': [_parse_bgp_path_entry(e) for e in data['paths'] if isinstance(e, dict)]}
1215
1216
1217def get_bgp_prefix(grade: Grade0, machine_name: str, prefix, step: int = 1, allow_error: bool = False) -> Dict[str, Any]:
1218    """Run 'vtysh -c "show bgp ipv4 unicast <prefix> json"' (see :func:`parse_bgp_prefix`)."""
1219    return parse_bgp_prefix(_vtysh_json(grade, machine_name, f'show bgp ipv4 unicast {prefix}', step, allow_error))
1220
1221
1222# -- show bgp ipv4 unicast neighbors X advertised-routes ------------------------
1223
1224def parse_bgp_advertised_routes(data: Any) -> Dict[str, Dict[str, Any]]:
1225    """Parse the JSON of ``show bgp ipv4 unicast neighbors <peer> advertised-routes``
1226    (the routes sent to that peer, after the outbound policy).
1227
1228    Returns ``{prefix: {'nexthop', 'aspath', 'as_list', 'locprf', 'med', 'weight', 'origin'}}``
1229    (``{}`` on error or when nothing is advertised).
1230    """
1231    if not isinstance(data, dict):
1232        return {}
1233    adv = data.get('advertisedRoutes')
1234    if not isinstance(adv, dict):
1235        return {}
1236    result: Dict[str, Dict[str, Any]] = {}
1237    for prefix, e in adv.items():
1238        if not isinstance(e, dict):
1239            continue
1240        path = e.get('path') or ''
1241        result[prefix] = {
1242            'nexthop': e.get('nextHop'),
1243            'aspath': path,
1244            'as_list': as_list_from_string(path),
1245            'locprf': _int_or_none(e.get('locPrf')),
1246            'med': _int_or_none(e.get('metric')),
1247            'weight': _int_or_none(e.get('weight')) or 0,
1248            'origin': e.get('bgpOriginCode') or e.get('origin'),
1249        }
1250    return result
1251
1252
1253def get_bgp_advertised_routes(grade: Grade0, machine_name: str, peer, step: int = 1,
1254                              allow_error: bool = False) -> Dict[str, Dict[str, Any]]:
1255    """Run 'vtysh -c "show bgp ipv4 unicast neighbors <peer> advertised-routes json"'
1256    (see :func:`parse_bgp_advertised_routes`)."""
1257    return parse_bgp_advertised_routes(
1258        _vtysh_json(grade, machine_name, f'show bgp ipv4 unicast neighbors {peer} advertised-routes', step, allow_error))
1259
1260
1261# ---------------------------------------------------------------------------
1262# RIP (text output of vtysh: ripd has no json form)
1263# ---------------------------------------------------------------------------
1264
1265#: metric of an unreachable route
1266RIP_INFINITY = 16
1267#: sub-codes of `show ip rip`: where the route of the RIP table comes from
1268RIP_SUB_CODES = {'n': 'normal', 's': 'static', 'd': 'default', 'r': 'redistribute', 'i': 'interface'}
1269
1270_RIP_ROUTE_RE = re.compile(
1271    r'^(?P<code>[A-Z])\((?P<sub>[a-z])\)\s+(?P<prefix>\d{1,3}(?:\.\d{1,3}){3}/\d{1,2})\s+'
1272    r'(?P<nexthop>\d{1,3}(?:\.\d{1,3}){3}|blackhole|V6 Address Hidden)\s+(?P<metric>\d+)\s+'
1273    r'(?P<source>self(?:\s+\((?P<external>[^)]*)\))?|\d{1,3}(?:\.\d{1,3}){3})\s+(?P<tag>\d+)'
1274    r'(?:\s+(?P<time>\d\d:\d\d(?::\d\d)?))?\s*$')
1275_RIP_VERSION = r'(?:1 2|1|2|none)'
1276_RIP_INTERFACE_RE = re.compile(
1277    rf'^\s+(?P<name>\S+)\s+(?P<send>{_RIP_VERSION})\s+(?P<receive>{_RIP_VERSION})(?:\s+(?P<key_chain>\S+))?\s*$')
1278_RIP_SOURCE_RE = re.compile(
1279    r'^\s+(?P<gateway>\d{1,3}(?:\.\d{1,3}){3})\s+(?P<bad_packets>\d+)\s+(?P<bad_routes>\d+)\s+'
1280    r'(?P<distance>\d+)\s+(?P<last_update>\S+)\s*$')
1281
1282
1283def parse_rip_routes(output: str) -> Dict[str, List[Dict[str, Any]]]:
1284    """Parse ``show ip rip`` (the RIP table of ripd, not the kernel one).
1285
1286    Returns ``{prefix: [path, ...]}`` — one path per row, an ECMP route has several — with
1287    ``code`` (``'R'`` RIP, ``'C'`` connected, ``'S'`` static, ``'K'`` kernel...), ``sub_code``
1288    (:data:`RIP_SUB_CODES`: ``'n'`` learned from a neighbour, ``'i'`` interface, ``'r'``
1289    redistributed, ``'d'`` default, ``'s'`` static RIP route), ``nexthop`` (``None`` for a local
1290    route), ``metric``, ``source`` (the neighbour the route was learned from, ``None`` for the
1291    router's own routes), ``external`` (``'static:20'`` of ``self (static:20)``), ``tag`` and
1292    ``time`` (the timeout or garbage-collection timer, ``None`` when the route has none).
1293    ``{}`` when ripd has no instance (``% RIP instance not found``) or on the placeholder.
1294    """
1295    result: Dict[str, List[Dict[str, Any]]] = {}
1296    for line in (output or '').splitlines():
1297        m = _RIP_ROUTE_RE.match(line)
1298        if not m:
1299            continue
1300        nexthop = m.group('nexthop')
1301        source = m.group('source')
1302        result.setdefault(m.group('prefix'), []).append({
1303            'code': m.group('code'),
1304            'sub_code': m.group('sub'),
1305            'nexthop': None if nexthop == '0.0.0.0' else nexthop,
1306            'metric': int(m.group('metric')),
1307            'source': None if source.startswith('self') else source,
1308            'external': m.group('external'),
1309            'tag': int(m.group('tag')),
1310            'time': m.group('time'),
1311        })
1312    return result
1313
1314
1315def get_rip_routes(grade: Grade0, machine_name: str, step: int = 1,
1316                   allow_error: bool = False) -> Dict[str, List[Dict[str, Any]]]:
1317    """Run 'vtysh -c "show ip rip"' and parse it (see :func:`parse_rip_routes`). ``{}`` on error."""
1318    output, code = _test(grade, machine_name, 'vtysh -c "show ip rip"', step, allow_error)
1319    if code != 0 or not output:
1320        return {}
1321    return parse_rip_routes(output)
1322
1323
1324def rip_route_metric(routes: Dict[str, List[Dict[str, Any]]], prefix, learned: bool = None) -> Optional[int]:
1325    """Smallest metric of *prefix* in a :func:`parse_rip_routes` result, ``None`` when the prefix
1326    is absent.  ``learned=True`` only looks at the paths learned from a neighbour (code ``R``,
1327    sub-code ``n``), ``learned=False`` at the router's own ones (interface, redistributed...)."""
1328    metrics = [p['metric'] for p in routes.get(str(prefix), [])
1329               if learned is None or learned == (p['source'] is not None)]
1330    return min(metrics) if metrics else None
1331
1332
1333def rip_route_sources(routes: Dict[str, List[Dict[str, Any]]], prefix) -> List[str]:
1334    """Neighbours *prefix* was learned from in a :func:`parse_rip_routes` result (reachable
1335    paths only: a route in garbage collection, metric 16, is left out)."""
1336    return [p['source'] for p in routes.get(str(prefix), [])
1337            if p['source'] is not None and p['metric'] < RIP_INFINITY]
1338
1339
1340def parse_rip_status(output: str) -> Dict[str, Any]:
1341    """Parse ``show ip rip status``.
1342
1343    Returns ``{'update', 'timeout', 'garbage'}`` (the three timers of ``timers basic``, seconds),
1344    ``default_metric``, ``redistribute`` (``['connected', 'static']``), ``send_version`` /
1345    ``receive_version`` (``'1'``, ``'2'``, ``'1 2'``), ``interfaces`` (``{name: {'send',
1346    'receive', 'key_chain'}}``: the interfaces RIP runs on, passive ones included), ``networks``
1347    (the ``network`` statements: prefixes and interface names, then the ``neighbor`` addresses),
1348    ``passive_interfaces``, ``sources`` (``{gateway: {'bad_packets', 'bad_routes', 'distance',
1349    'last_update'}}``: every neighbour a packet was received from) and ``distance``.
1350    ``{}`` when ripd has no instance or on the placeholder.
1351    """
1352    text = output or ''
1353    if 'Routing Protocol is "rip"' not in text:
1354        return {}
1355    status: Dict[str, Any] = {
1356        'update': None, 'timeout': None, 'garbage': None, 'default_metric': None, 'redistribute': [],
1357        'send_version': None, 'receive_version': None, 'interfaces': {}, 'networks': [],
1358        'passive_interfaces': [], 'sources': {}, 'distance': None,
1359    }
1360    section = None
1361    for line in text.splitlines():
1362        stripped = line.strip()
1363        if not stripped:
1364            continue
1365        m = re.match(r'Sending updates every (\d+) seconds', stripped)
1366        if m:
1367            status['update'] = int(m.group(1))
1368            continue
1369        m = re.match(r'Timeout after (\d+) seconds, garbage collect after (\d+) seconds', stripped)
1370        if m:
1371            status['timeout'], status['garbage'] = int(m.group(1)), int(m.group(2))
1372            continue
1373        m = re.match(r'Default redistribution metric is (\d+)', stripped)
1374        if m:
1375            status['default_metric'] = int(m.group(1))
1376            continue
1377        if stripped.startswith('Redistributing:'):
1378            status['redistribute'] = stripped[len('Redistributing:'):].split()
1379            continue
1380        m = re.match(r'Default version control: send version (.+?), receive (any version|version (.+?))\s*$',
1381                     stripped)
1382        if m:
1383            status['send_version'] = m.group(1).strip()
1384            status['receive_version'] = '1 2' if m.group(2) == 'any version' else m.group(3).strip()
1385            continue
1386        m = re.match(r'Distance: \(default is (\d+)\)', stripped)
1387        if m:
1388            status['distance'] = int(m.group(1))
1389            section = None
1390            continue
1391        if re.match(r'Interface\s+Send\s+Recv', stripped):
1392            section = 'interfaces'
1393            continue
1394        if stripped == 'Routing for Networks:':
1395            section = 'networks'
1396            continue
1397        if stripped == 'Passive Interface(s):':
1398            section = 'passive'
1399            continue
1400        if stripped == 'Routing Information Sources:':
1401            section = 'sources'
1402            continue
1403        if section == 'interfaces':
1404            m = _RIP_INTERFACE_RE.match(line)
1405            if m:
1406                status['interfaces'][m.group('name')] = {
1407                    'send': m.group('send'), 'receive': m.group('receive'), 'key_chain': m.group('key_chain')}
1408        elif section == 'networks':
1409            status['networks'].append(stripped)
1410        elif section == 'passive':
1411            status['passive_interfaces'].append(stripped)
1412        elif section == 'sources':
1413            m = _RIP_SOURCE_RE.match(line)
1414            if m:
1415                status['sources'][m.group('gateway')] = {
1416                    'bad_packets': int(m.group('bad_packets')), 'bad_routes': int(m.group('bad_routes')),
1417                    'distance': int(m.group('distance')), 'last_update': m.group('last_update')}
1418    return status
1419
1420
1421def get_rip_status(grade: Grade0, machine_name: str, step: int = 1, allow_error: bool = False) -> Dict[str, Any]:
1422    """Run 'vtysh -c "show ip rip status"' and parse it (see :func:`parse_rip_status`).
1423    ``{}`` on error or when RIP is not configured."""
1424    output, code = _test(grade, machine_name, 'vtysh -c "show ip rip status"', step, allow_error)
1425    if code != 0 or not output:
1426        return {}
1427    return parse_rip_status(output)