Coverage for src/birdplan/__init__.py: 75%
361 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-25 07:38 +0000
1#
2# SPDX-License-Identifier: GPL-3.0-or-later
3#
4# Copyright (c) 2019-2025, AllWorldIT
5#
6# This program is free software: you can redistribute it and/or modify
7# it under the terms of the GNU General Public License as published by
8# the Free Software Foundation, either version 3 of the License, or
9# (at your option) any later version.
10#
11# This program is distributed in the hope that it will be useful,
12# but WITHOUT ANY WARRANTY; without even the implied warranty of
13# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14# GNU General Public License for more details.
15#
16# You should have received a copy of the GNU General Public License
17# along with this program. If not, see <http://www.gnu.org/licenses/>.
19"""BirdPlan package."""
21# pylint: disable=too-many-lines
23import contextlib
24import grp
25import json
26import os
27import pathlib
28import pwd
29from typing import Any
31import birdclient
32import jinja2
33import packaging.version
35from .bird_config import BirdConfig
36from .bird_config.sections.protocols.bgp.bgp_config_parser import BGPConfigParser
37from .bird_config.sections.protocols.ospf.ospf_config_parser import OSPFConfigParser
38from .bird_config.sections.protocols.rip.rip_config_parser import RIPConfigParser
39from .exceptions import BirdPlanError
40from .version import __version__
41from .yaml import YAML, YAMLError
43__all__ = [
44 "BirdPlan",
45 "__version__",
46]
48# Some types we need
49BirdPlanBGPPeerSummary = dict[str, dict[str, Any]]
50BirdPlanBGPPeerShow = dict[str, Any]
51BirdPlanBGPPeerGracefulShutdownStatus = dict[str, dict[str, bool]]
52BirdPlanBGPPeerQuarantineStatus = dict[str, dict[str, bool]]
53BirdPlanOSPFInterfaceStatus = dict[str, dict[str, dict[str, Any]]]
54BirdPlanOSPFSummary = dict[str, dict[str, Any]]
56# Check we have a sufficiently new version of birdclient
57if packaging.version.parse(birdclient.__version__) < packaging.version.parse("0.0.11"):
58 raise BirdPlanError("BirdPlan requires birdclient version 0.0.11 or newer")
61class BirdPlan: # pylint: disable=too-many-public-methods
62 """Main BirdPlan class."""
64 _birdconf: BirdConfig
65 _config: dict[str, Any]
66 _state_file: str | None
67 _yaml: YAML
69 def __init__(self, test_mode: bool = False) -> None: # noqa: FBT001,FBT002
70 """Initialize object."""
72 self._birdconf = BirdConfig(test_mode=test_mode)
73 self._config = {}
74 self._state_file = None
75 self._yaml = YAML()
77 def load(self, **kwargs: Any) -> None: # noqa: ANN401,D417
78 """
79 Initialize object.
81 Parameters
82 ----------
83 plan_file : str
84 Source plan file to generate configuration from.
86 state_file : Optional[str]
87 Optional state file, used for commands like BGP graceful shutdown.
89 ignore_irr_changes : bool
90 Optional parameter to ignore IRR lookups during configuration load.
92 ignore_peeringdb_changes : bool
93 Optional parameter to ignore peering DB lookups during configuraiton load.
95 use_cached : bool
96 Optional parameter to use cached values from state during configuration load.
98 """
100 # Grab parameters
101 plan_file: str | None = kwargs.get("plan_file")
102 state_file: str | None = kwargs.get("state_file")
103 ignore_irr_changes: bool = kwargs.get("ignore_irr_changes", False)
104 ignore_peeringdb_changes: bool = kwargs.get("ignore_peeringdb_changes", False)
105 use_cached: bool = kwargs.get("use_cached", False)
107 # Make sure we have the parameters we need
108 if not plan_file:
109 raise BirdPlanError("Required parameter 'plan_file' not found")
111 plan_file_path = pathlib.Path(plan_file)
113 # Create search paths for Jinja2
114 search_paths = [plan_file_path.parent]
115 # We need to pass Jinja2 our filename, as it is in the search path
116 plan_file_fname = plan_file_path.name
118 # Render first with jinja
119 template_env = jinja2.Environment( # noqa: S701
120 loader=jinja2.FileSystemLoader(searchpath=search_paths),
121 trim_blocks=True,
122 lstrip_blocks=True,
123 )
125 # Check if we can load the configuration
126 try:
127 raw_config = template_env.get_template(plan_file_fname).render()
128 except jinja2.TemplateError as err:
129 raise BirdPlanError(f"Failed to template BirdPlan configuration file '{plan_file}': {err}") from None
131 # Load configuration using YAML
132 try:
133 self.config = self.yaml.load(raw_config)
134 except YAMLError as err: # pragma: no cover
135 raise BirdPlanError(f" Failed to parse BirdPlan configuration in '{plan_file}': {err}") from None
137 # Set our state file and load state
138 self.state_file = state_file
139 self.load_state()
141 # Make sure we have configuration...
142 if not self.config:
143 raise BirdPlanError("No configuration found")
145 # Check configuration options are supported
146 for config_item in self.config:
147 if config_item not in ("router_id", "kernel", "log_file", "debug", "static", "export_kernel", "bgp", "rip", "ospf"):
148 raise BirdPlanError(f"The config item '{config_item}' is not supported")
150 # Setup globals we need
151 self.birdconf.birdconfig_globals.ignore_irr_changes = ignore_irr_changes
152 self.birdconf.birdconfig_globals.ignore_peeringdb_changes = ignore_peeringdb_changes
153 self.birdconf.birdconfig_globals.use_cached = use_cached
155 # Configure sections
156 self._config_global()
157 self._config_kernel()
158 self._config_static()
159 self._config_export_kernel()
161 rip_parser = RIPConfigParser(self.birdconf)
162 rip_parser.parse(self.config)
164 ospf_parser = OSPFConfigParser(self.birdconf)
165 ospf_parser.parse(self.config)
167 bgp_parser = BGPConfigParser(self.birdconf)
168 bgp_parser.parse(self.config)
170 def configure(self) -> str:
171 """
172 Create BIRD configuration.
174 Returns
175 -------
176 str : Bird configuration as a string.
178 """
179 return "\n".join(self.birdconf.get_config())
181 def commit_state(self) -> None:
182 """Commit our current state."""
184 # Raise an exception if we don't have a state file loaded
185 if self.state_file is None:
186 raise BirdPlanError("Commit of BirdPlan state requires a state file, none loaded")
188 # Try get user and group ID's
189 try:
190 birdplan_uid = pwd.getpwnam("birdplan").pw_uid
191 except KeyError:
192 birdplan_uid = None
193 try:
194 birdplan_gid = grp.getgrnam("birdplan").gr_gid
195 except KeyError:
196 birdplan_gid = None
198 # Write out state file
199 try:
200 fd = os.open(self.state_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o640)
201 # Chown the file if we have the user and group ID's
202 if birdplan_uid and birdplan_gid:
203 with contextlib.suppress(PermissionError):
204 os.fchown(fd, birdplan_uid, birdplan_gid)
205 # Open for writing
206 with os.fdopen(fd, "w") as file:
207 file.write(json.dumps(self.state))
208 except OSError as err: # pragma: no cover
209 raise BirdPlanError(f"Failed to open '{self.state_file}' for writing: {err}") from None
211 def load_state(self) -> None:
212 """Load our state."""
214 # Clear state
215 self.state = {}
217 # Skip if we don't have a state file
218 if not self.state_file:
219 return
221 # Check if the state file exists...
222 state_file = pathlib.Path(self.state_file)
223 if state_file.is_file():
224 # Read in state file
225 try:
226 self.state = json.loads(state_file.read_text(encoding="UTF-8"))
227 except OSError as err:
228 raise BirdPlanError(f"Failed to read BirdPlan state file '{state_file}': {err}") from None
229 except json.JSONDecodeError as err: # pragma: no cover
230 # We use the state_file here because the size of raw_state may be larger than 100MiB
231 raise BirdPlanError(f" Failed to parse BirdPlan state file '{state_file}': {err}") from None
233 def state_ospf_summary(self, bird_socket: str | None = None) -> BirdPlanOSPFSummary:
234 """
235 Return OSPF summary.
237 Returns
238 -------
239 BirdPlanOSPFSummary
240 Dictionary containing the OSPF summary.
242 eg.
243 {
244 'name1': {
245 'channel': ...,
246 'info': ...,
247 'input_filter': ...,
248 'name': ...,
249 'output_filter': ...,
250 'preference': ...,
251 'proto': ...,
252 'routes_exported': ...,
253 'routes_imported': ...,
254 'since': ...,
255 'state': ...
256 'table': ...,
257 }
258 'name2': {
259 ...,
260 }
261 }
263 """
265 # Raise an exception if we don't have a state file loaded
266 if self.state_file is None:
267 raise BirdPlanError("The use of OSPF summary requires a state file, none loaded")
269 # Initialize our return structure
270 ret: BirdPlanOSPFSummary = {}
272 # Return if we don't have any BGP state
273 if "ospf" not in self.state:
274 return ret
276 # Query bird client for the current protocols
277 birdc = birdclient.BirdClient(control_socket=bird_socket)
278 bird_protocols = birdc.show_protocols()
280 for name, data in bird_protocols.items():
281 if data["proto"] != "OSPF":
282 continue
283 ret[name] = data
285 return ret
287 def state_bgp_peer_summary(self, bird_socket: str | None = None) -> BirdPlanBGPPeerSummary:
288 """
289 Return BGP peer summary.
291 Returns
292 -------
293 BirdPlanBGPPeerStatus
294 Dictionary containing the BGP peer summary.
296 eg.
297 {
298 'peer1': {
299 'name': ...,
300 'asn': ...,
301 'description': ...,
302 'protocols': {
303 'ipv4': ...,
304 'ipv6': ...,
305 }
306 }
307 'peer2': {
308 ...,
309 }
310 }
312 """
314 # Raise an exception if we don't have a state file loaded
315 if self.state_file is None:
316 raise BirdPlanError("The use of BGP peer summary requires a state file, none loaded")
318 # Initialize our return structure
319 ret: BirdPlanBGPPeerSummary = {}
321 # Return if we don't have any BGP state
322 if "bgp" not in self.state:
323 return ret
325 # Query bird client for the current protocols
326 birdc = birdclient.BirdClient(control_socket=bird_socket)
327 bird_protocols = birdc.show_protocols()
329 # Check if we have any peers in our state
330 if "peers" in self.state["bgp"]:
331 # If we do loop with them
332 for peer, peer_state in self.state["bgp"]["peers"].items():
333 # Start with a clear status
334 ret[peer] = {
335 "name": peer,
336 "asn": peer_state["asn"],
337 "description": peer_state["description"],
338 "protocols": peer_state["protocols"],
339 }
341 # Next loop through each protocol
342 for ipv, peer_state_protocol in peer_state["protocols"].items():
343 # If we don't have a live session, skip adding it
344 if peer_state_protocol["name"] not in bird_protocols:
345 continue
346 # Set protocol name
347 ret[peer]["protocols"][ipv]["protocol"] = ipv
348 # But if we do, add it
349 ret[peer]["protocols"][ipv]["status"] = bird_protocols[peer_state_protocol["name"]]
351 return ret
353 def state_bgp_peer_show(self, peer: str, bird_socket: str | None = None) -> BirdPlanBGPPeerShow:
354 """
355 Return the status of a specific BGP peer.
357 Returns
358 -------
359 BirdPlanBGPPeerShow
360 Dictionary containing the status of a BGP peer.
362 eg.
363 {
364 'asn': ...,
365 'description': ...,
366 'protocols': {
367 'ipv4': {
368 ...,
369 'status': ...,
370 }
371 'ipv6': ...,
372 },
373 }
375 """
377 # Raise an exception if we don't have a state file loaded
378 if self.state_file is None:
379 raise BirdPlanError("The use of BGP peer show requires a state file, none loaded")
381 # Return if we don't have any BGP state
382 if "bgp" not in self.state:
383 raise BirdPlanError("No BGP state found")
384 # Check if the configured state has this peer, if not return
385 if peer not in self.state["bgp"]["peers"]:
386 raise BirdPlanError(f"BGP peer '{peer}' not found in configured state")
388 # Make things easier below
389 configured = self.state["bgp"]["peers"][peer]
391 # Set our peer info to the configured state
392 ret: BirdPlanBGPPeerShow = configured
394 # Add peer name
395 ret["name"] = peer
397 # Query bird client for the current protocols
398 birdc = birdclient.BirdClient(control_socket=bird_socket)
400 # Loop with protocols and grab live bird status
401 for ipv, protocol_info in configured["protocols"].items():
402 bird_state = birdc.show_protocol(protocol_info["name"])
403 # Skip if we have no bird state
404 if not bird_state:
405 continue
406 # Set the protocol status
407 ret["protocols"][ipv]["status"] = bird_state
409 return ret
411 def state_bgp_peer_graceful_shutdown_set(self, peer: str, value: bool) -> None: # noqa: FBT001
412 """
413 Set the BGP graceful shutdown override state for a peer.
415 Parameters
416 ----------
417 peer : str
418 Peer name to set to BGP graceful shutdown state for.
419 Pattern matches can be specified with '*'.
421 value : bool
422 State of the graceful shutdown option for this peer.
424 """
426 # Raise an exception if we don't have a state file loaded
427 if self.state_file is None:
428 raise BirdPlanError("The use of BGP graceful shutdown override requires a state file, none loaded")
430 # Prepare the state structure if its not got what we need
431 if "bgp" not in self.state:
432 self.state["bgp"] = {}
434 # Make sure we have the global setting
435 if "+graceful_shutdown" not in self.state["bgp"]:
436 self.state["bgp"]["+graceful_shutdown"] = {}
437 # Set the global setting for this pattern
438 self.state["bgp"]["+graceful_shutdown"][peer] = value
440 def state_bgp_peer_graceful_shutdown_remove(self, peer: str) -> None:
441 """
442 Remove a BGP graceful shutdown override flag from a peer or pattern.
444 Parameters
445 ----------
446 peer : str
447 Peer name or pattern to remove the BGP graceful shutdown override flag from.
449 """
451 # Raise an exception if we don't have a state file loaded
452 if self.state_file is None:
453 raise BirdPlanError("The use of BGP graceful shutdown override requires a state file, none loaded")
455 # Prepare the state structure if its not got what we need
456 if "bgp" not in self.state:
457 return
459 # Remove from the global settings
460 if "+graceful_shutdown" in self.state["bgp"]:
461 # Check it exists first, if not raise an exception
462 if peer not in self.state["bgp"]["+graceful_shutdown"]:
463 raise BirdPlanError(f"BGP peer '{peer}' graceful shutdown override not found")
464 # Remove peer from graceful shutdown list
465 del self.state["bgp"]["+graceful_shutdown"][peer]
466 # If the result is an empty dict, just delete it too
467 if not self.state["bgp"]["+graceful_shutdown"]:
468 del self.state["bgp"]["+graceful_shutdown"]
470 def state_bgp_peer_graceful_shutdown_status(self) -> BirdPlanBGPPeerGracefulShutdownStatus:
471 """
472 Return the status of BGP peer graceful shutdown.
474 Returns
475 -------
476 BirdPlanBGPPeerGracefulShutdownStatus
477 Dictionary containing the status of overrides and peers.
479 eg.
480 {
481 'overrides': {
482 'p*': True,
483 'peer1': False,
484 }
485 'current': {
486 'peer1': False,
487 }
488 'pending': {
489 'peer1': False,
490 }
491 }
493 """
495 # Raise an exception if we don't have a state file loaded
496 if self.state_file is None:
497 raise BirdPlanError("The use of BGP graceful shutdown override requires a state file, none loaded")
499 # Initialize our return structure
500 ret: BirdPlanBGPPeerGracefulShutdownStatus = {
501 "overrides": {},
502 "current": {},
503 "pending": {},
504 }
506 # Return if we don't have any BGP state
507 if "bgp" not in self.state:
508 return ret
510 # Pull in any overrides we may have
511 if "+graceful_shutdown" in self.state["bgp"]:
512 ret["overrides"] = self.state["bgp"]["+graceful_shutdown"]
514 # Check if we have any peers in our state
515 if "peers" in self.state["bgp"]:
516 # If we do loop with them
517 for peer, peer_state in self.state["bgp"]["peers"].items():
518 # And check if they have a graceful shutdown state or not
519 ret["current"][peer] = peer_state.get("graceful_shutdown", False)
521 # Generate the override status as if we were doing a configure
522 for peer in self.birdconf.protocols.bgp.peers:
523 ret["pending"][peer] = self.birdconf.protocols.bgp.peer(peer).graceful_shutdown
525 return ret
527 def state_bgp_peer_quarantine_set(self, peer: str, value: bool) -> None: # noqa: FBT001
528 """
529 Set the BGP quarantine override state for a peer.
531 Parameters
532 ----------
533 peer : str
534 Peer name to set to BGP quarantine state for.
535 Pattern matches can be specified with '*'.
537 value : bool
538 State of the quarantine option for this peer.
540 """
542 # Raise an exception if we don't have a state file loaded
543 if self.state_file is None:
544 raise BirdPlanError("The use of BGP quarantine override requires a state file, none loaded")
546 # Prepare the state structure if its not got what we need
547 if "bgp" not in self.state:
548 self.state["bgp"] = {}
550 # Make sure we have the global setting
551 if "+quarantine" not in self.state["bgp"]:
552 self.state["bgp"]["+quarantine"] = {}
553 # Set the global setting for this pattern
554 self.state["bgp"]["+quarantine"][peer] = value
556 def state_bgp_peer_quarantine_remove(self, peer: str) -> None:
557 """
558 Remove a BGP quarantine override flag from a peer or pattern.
560 Parameters
561 ----------
562 peer : str
563 Peer name or pattern to remove the BGP quarantine override flag from.
565 """
567 # Raise an exception if we don't have a state file loaded
568 if self.state_file is None:
569 raise BirdPlanError("The use of BGP quarantine override requires a state file, none loaded")
571 # Prepare the state structure if its not got what we need
572 if "bgp" not in self.state:
573 return
575 # Remove from the global settings
576 if ("+quarantine" not in self.state["bgp"]) or (peer not in self.state["bgp"]["+quarantine"]):
577 raise BirdPlanError(f"BGP peer '{peer}' quarantine override not found")
579 # Remove peer from quarantine list
580 del self.state["bgp"]["+quarantine"][peer]
581 # If the result is an empty dict, just delete it too
582 if not self.state["bgp"]["+quarantine"]:
583 del self.state["bgp"]["+quarantine"]
585 def state_bgp_peer_quarantine_status(self) -> BirdPlanBGPPeerQuarantineStatus:
586 """
587 Return the status of BGP peer quarantine.
589 Returns
590 -------
591 BirdPlanBGPPeerQuarantineStatus
592 Dictionary containing the status of overrides and peers.
594 eg.
595 {
596 'overrides': {
597 'p*': True,
598 'peer1': False,
599 }
600 'current': {
601 'peer1': False,
602 }
603 'pending': {
604 'peer1': False,
605 }
606 }
608 """
610 # Raise an exception if we don't have a state file loaded
611 if self.state_file is None:
612 raise BirdPlanError("The use of BGP quarantine override requires a state file, none loaded")
614 # Initialize our return structure
615 ret: BirdPlanBGPPeerQuarantineStatus = {
616 "overrides": {},
617 "current": {},
618 "pending": {},
619 }
621 # Return if we don't have any BGP state
622 if "bgp" not in self.state:
623 return ret
625 # Pull in any overrides we may have
626 if "+quarantine" in self.state["bgp"]:
627 ret["overrides"] = self.state["bgp"]["+quarantine"]
629 # Check if we have any peers in our state
630 if "peers" in self.state["bgp"]:
631 # If we do loop with them
632 for peer, peer_state in self.state["bgp"]["peers"].items():
633 # And check if they have a quarantine state or not
634 ret["current"][peer] = peer_state.get("quarantine", False)
636 # Generate the override status as if we were doing a configure
637 for peer in self.birdconf.protocols.bgp.peers:
638 ret["pending"][peer] = self.birdconf.protocols.bgp.peer(peer).quarantine
640 return ret
642 def state_ospf_set_interface_cost(self, area: str, interface: str, cost: int) -> None:
643 """
644 Set an OSPF interface cost override.
646 Parameters
647 ----------
648 area : str
649 Interface to set the OSPF cost for.
651 interface : str
652 Interface to set the OSPF cost for.
654 cost : int
655 OSPF interface cost.
657 """
659 # Raise an exception if we don't have a state file loaded
660 if self.state_file is None:
661 raise BirdPlanError("The use of OSPF interface cost override requires a state file, none loaded")
663 # Prepare the state structure if its not got what we need
664 if "ospf" not in self.state:
665 self.state["ospf"] = {}
666 if "areas" not in self.state["ospf"]:
667 self.state["ospf"]["areas"] = {}
668 if area not in self.state["ospf"]["areas"]:
669 self.state["ospf"]["areas"][area] = {}
670 if "+interfaces" not in self.state["ospf"]["areas"][area]:
671 self.state["ospf"]["areas"][area]["+interfaces"] = {}
672 if interface not in self.state["ospf"]["areas"][area]["+interfaces"]:
673 self.state["ospf"]["areas"][area]["+interfaces"][interface] = {}
675 # Set the interface cost value
676 self.state["ospf"]["areas"][area]["+interfaces"][interface]["cost"] = cost
678 def state_ospf_remove_interface_cost(self, area: str, interface: str) -> None:
679 """
680 Remove an OSPF interface cost override.
682 Parameters
683 ----------
684 area : str
685 OSPF area which contains the interface.
687 interface : str
688 Interface to remove the OSPF cost for.
690 """
692 # Raise an exception if we don't have a state file loaded
693 if self.state_file is None:
694 raise BirdPlanError("The use of OSPF interface cost override requires a state file, none loaded")
696 # Check if this cost override exists
697 if ( # pylint: disable=too-many-boolean-expressions
698 "ospf" not in self.state
699 or "areas" not in self.state["ospf"]
700 or area not in self.state["ospf"]["areas"]
701 or "+interfaces" not in self.state["ospf"]["areas"][area]
702 or interface not in self.state["ospf"]["areas"][area]["+interfaces"]
703 or "cost" not in self.state["ospf"]["areas"][area]["+interfaces"][interface]
704 ):
705 raise BirdPlanError(f"OSPF area '{area}' interface '{interface}' cost override not found")
707 # Remove OSPF interface cost from state
708 del self.state["ospf"]["areas"][area]["+interfaces"][interface]["cost"]
709 # Remove hanging data structure endpoint
710 if not self.state["ospf"]["areas"][area]["+interfaces"][interface]:
711 del self.state["ospf"]["areas"][area]["+interfaces"][interface]
712 if not self.state["ospf"]["areas"][area]["+interfaces"]:
713 del self.state["ospf"]["areas"][area]["+interfaces"]
715 def state_ospf_set_interface_ecmp_weight(self, area: str, interface: str, ecmp_weight: int) -> None:
716 """
717 Set an OSPF interface ECMP weight override.
719 Parameters
720 ----------
721 area : str
722 OSPF area which contains the interface.
724 interface : str
725 Interface to set the OSPF ECMP weight for.
727 ecmp_weight : int
728 OSPF interface ECMP weight.
730 """
732 # Raise an exception if we don't have a state file loaded
733 if self.state_file is None:
734 raise BirdPlanError("The use of OSPF interface ECMP weight override requires a state file, none loaded")
736 # Prepare the state structure if its not got what we need
737 if "ospf" not in self.state:
738 self.state["ospf"] = {}
739 if "areas" not in self.state["ospf"]:
740 self.state["ospf"]["areas"] = {}
741 if area not in self.state["ospf"]["areas"]:
742 self.state["ospf"]["areas"][area] = {}
743 if "+interfaces" not in self.state["ospf"]["areas"][area]:
744 self.state["ospf"]["areas"][area]["+interfaces"] = {}
745 if interface not in self.state["ospf"]["areas"][area]["+interfaces"]:
746 self.state["ospf"]["areas"][area]["+interfaces"][interface] = {}
748 # Set the interface ecmp_weight value
749 self.state["ospf"]["areas"][area]["+interfaces"][interface]["ecmp_weight"] = ecmp_weight
751 def state_ospf_remove_interface_ecmp_weight(self, area: str, interface: str) -> None:
752 """
753 Remove an OSPF interface ECMP weight override.
755 Parameters
756 ----------
757 area : str
758 OSPF area which contains the interface.
760 interface : str
761 Interface to remove the OSPF ECMP weight for.
763 """
765 # Raise an exception if we don't have a state file loaded
766 if self.state_file is None:
767 raise BirdPlanError("The use of OSPF interface ECMP weight override requires a state file, none loaded")
769 # Check if this ECMP weight override exists
770 if ( # pylint: disable=too-many-boolean-expressions
771 "ospf" not in self.state
772 or "areas" not in self.state["ospf"]
773 or area not in self.state["ospf"]["areas"]
774 or "+interfaces" not in self.state["ospf"]["areas"][area]
775 or interface not in self.state["ospf"]["areas"][area]["+interfaces"]
776 or "ecmp_weight" not in self.state["ospf"]["areas"][area]["+interfaces"][interface]
777 ):
778 raise BirdPlanError(f"OSPF area '{area}' interface '{interface}' ECMP weight override not found")
780 # Remove OSPF interface ECMP weight from state
781 del self.state["ospf"]["areas"][area]["+interfaces"][interface]["ecmp_weight"]
782 # Remove hanging data structure endpoint
783 if not self.state["ospf"]["areas"][area]["+interfaces"][interface]:
784 del self.state["ospf"]["areas"][area]["+interfaces"][interface]
785 if not self.state["ospf"]["areas"][area]["+interfaces"]:
786 del self.state["ospf"]["areas"][area]["+interfaces"]
788 def state_ospf_interface_status(self) -> BirdPlanOSPFInterfaceStatus: # noqa: C901,PLR0912
789 """
790 Return the status of OSPF interfaces.
792 Returns
793 -------
794 BirdPlanOSPFInterfaceStatus
795 Dictionary containing the status of overrides and peers.
797 eg.
798 {
799 'overrides': {
800 'areas': {
801 '0': {
802 'interfaces': {
803 'eth0': {
804 'cost': 10,
805 'ecmp_weight': 100,
806 }
807 }
808 }
809 }
810 },
811 'current': {
812 'areas': {
813 '0': {
814 'interfaces': {
815 'eth0': {
816 'cost': 10,
817 'ecmp_weight': 100,
818 }
819 }
820 }
821 }
822 },
823 'pending': {
824 'areas': {
825 '0': {
826 'interfaces': {
827 'eth0': {
828 'cost': 10,
829 'ecmp_weight': 100,
830 }
831 }
832 }
833 }
834 }
835 }
837 """
839 # Raise an exception if we don't have a state file loaded
840 if self.state_file is None:
841 raise BirdPlanError("The use of OSPF interface override requires a state file, none loaded")
843 # Initialize our return structure
844 ret: BirdPlanOSPFInterfaceStatus = {
845 "overrides": {},
846 "current": {},
847 "pending": {},
848 }
850 # Return if we don't have any OSPF state
851 if "ospf" not in self.state or "areas" not in self.state["ospf"]:
852 return ret
854 # Process overrides
855 for area_name, area in self.state["ospf"]["areas"].items():
856 # Make sure we have interfaces in the area
857 if "+interfaces" not in area:
858 continue
859 # Loop with interfaces
860 for interface_name, interface in area["+interfaces"].items():
861 # Check our structure is setup
862 if "areas" not in ret["overrides"]:
863 ret["overrides"]["areas"] = {}
864 if area_name not in ret["overrides"]["areas"]:
865 ret["overrides"]["areas"][area_name] = {}
866 if "interfaces" not in ret["overrides"]["areas"][area_name]:
867 ret["overrides"]["areas"][area_name]["interfaces"] = {}
868 # Link interface
869 ret["overrides"]["areas"][area_name]["interfaces"][interface_name] = interface
871 # Process current state
872 for area_name, area in self.state["ospf"]["areas"].items():
873 # Make sure we have interfaces in the area
874 if "interfaces" not in area:
875 continue
876 # Loop with interfaces
877 for interface_name, interface in area["interfaces"].items():
878 # Check our structure is setup
879 if "areas" not in ret["current"]:
880 ret["current"]["areas"] = {}
881 if area_name not in ret["current"]["areas"]:
882 ret["current"]["areas"][area_name] = {}
883 if "interfaces" not in ret["current"]["areas"][area_name]:
884 ret["current"]["areas"][area_name]["interfaces"] = {}
885 # Link interface
886 ret["current"]["areas"][area_name]["interfaces"][interface_name] = interface
888 # Generate the override status as if we were doing a configure
889 for area_name, area in self.birdconf.protocols.ospf.areas.items():
890 for interface_name, interface in area.interfaces.items():
891 # Check our structure is setup
892 if "areas" not in ret["pending"]:
893 ret["pending"]["areas"] = {}
894 if area_name not in ret["pending"]["areas"]:
895 ret["pending"]["areas"][area_name] = {}
896 if "interfaces" not in ret["pending"]["areas"][area_name]:
897 ret["pending"]["areas"][area_name]["interfaces"] = {}
898 if interface_name not in ret["pending"]["areas"][area_name]["interfaces"]:
899 ret["pending"]["areas"][area_name]["interfaces"][interface_name] = {}
900 # Add attributes we need
901 ret["pending"]["areas"][area_name]["interfaces"][interface_name]["cost"] = interface.cost
902 ret["pending"]["areas"][area_name]["interfaces"][interface_name]["ecmp_weight"] = interface.ecmp_weight
904 return ret
906 def _config_global(self) -> None:
907 """Configure global options."""
909 # Check that a router ID was specified
910 if "router_id" not in self.config:
911 raise BirdPlanError("The 'router_id' attribute must be specified")
912 self.birdconf.router_id = self.config["router_id"]
914 # Check if we have a log_file specified to use
915 if "log_file" in self.config:
916 self.birdconf.log_file = self.config["log_file"]
918 # Check if we're in debugging mode or not
919 if "debug" in self.config:
920 self.birdconf.debug = self.config["debug"]
922 def _config_kernel(self) -> None:
923 """Configure kernel section."""
925 # If we have no rip section, just return
926 if "kernel" not in self.config:
927 return
929 # Check configuration options are supported
930 for config_item in self.config["kernel"]:
931 if config_item not in ("vrf", "routing_table"):
932 raise BirdPlanError(f"The 'kernel' config item '{config_item}' is not supported")
934 # Check if we have a VRF to use
935 if "vrf" in self.config["kernel"]:
936 self.birdconf.vrf = '"' + self.config["kernel"]["vrf"] + '"'
937 # Make sure we also have a routing talbe
938 if "routing_table" not in self.config["kernel"]:
939 raise BirdPlanError("The 'kernel' config item 'vrf' requires that 'routing_table' is also specified")
941 if "routing_table" in self.config["kernel"]:
942 self.birdconf.routing_table = self.config["kernel"]["routing_table"]
944 def _config_static(self) -> None:
945 """Configure static section."""
946 # Static routes
947 if "static" in self.config:
948 for route in self.config["static"]:
949 self.birdconf.protocols.static.add_route(route)
951 def _config_export_kernel(self) -> None:
952 """Configure export_kernel section."""
954 # Check if we're exporting routes from the master tables to the kernel tables
955 if "export_kernel" in self.config:
956 # Loop with export_kernel items
957 for export, export_config in self.config["export_kernel"].items():
958 # Static routes
959 if export == "static":
960 self.birdconf.tables.master.route_policy_export.kernel.static = export_config
961 # RIP routes
962 elif export == "rip":
963 self.birdconf.tables.master.route_policy_export.kernel.rip = export_config
964 # OSPF routes
965 elif export == "ospf":
966 self.birdconf.tables.master.route_policy_export.kernel.ospf = export_config
967 # BGP routes
968 elif export == "bgp":
969 self.birdconf.tables.master.route_policy_export.kernel.bgp = export_config
970 # If we don't understand this 'accept' entry, throw an error
971 else:
972 raise BirdPlanError(f"Configuration item '{export}' not understood in 'export_kernel'")
974 @property
975 def birdconf(self) -> BirdConfig:
976 """Return the BirdConfig object."""
977 return self._birdconf
979 @property
980 def config(self) -> dict[str, Any]:
981 """Return our config."""
982 return self._config
984 @config.setter
985 def config(self, config: dict[str, Any]) -> None:
986 """Set our configuration."""
987 self._config = config
989 @property
990 def state(self) -> dict[str, Any]:
991 """Return our state."""
992 return self.birdconf.state
994 @state.setter
995 def state(self, state: dict[str, Any]) -> None:
996 """Set our state."""
997 self.birdconf.state = state
999 @property
1000 def state_file(self) -> str | None:
1001 """State file we're using."""
1002 return self._state_file
1004 @state_file.setter
1005 def state_file(self, state_file: str | None) -> None:
1006 """Set our state file."""
1007 self._state_file = state_file
1009 @property
1010 def yaml(self) -> YAML:
1011 """Return our YAML parser."""
1012 return self._yaml