Coverage for src/birdplan/bird_config/sections/protocols/ospf/__init__.py: 97%

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/>. 

18 

19"""BIRD OSPF protocol configuration.""" 

20 

21from .....exceptions import BirdPlanError 

22from ....globals import BirdConfigGlobals 

23from ...bird_attributes import SectionBirdAttributes 

24from ...constants import SectionConstants 

25from ...functions import SectionFunctions 

26from ...tables import SectionTables 

27from ..base import SectionProtocolBase 

28from ..direct import ProtocolDirect 

29from ..pipe import ProtocolPipe, ProtocolPipeFilterType 

30from .area import ProtocolOSPFArea 

31from .area.ospf_area_types import OSPFAreaConfig 

32from .ospf_attributes import OSPFAttributes, OSPFRoutePolicyAccept, OSPFRoutePolicyRedistribute 

33from .ospf_functions import OSPFFunctions 

34 

35__all__ = ["ProtocolOSPF"] 

36 

37 

38OSPFAreas = dict[str, ProtocolOSPFArea] 

39 

40 

41class ProtocolOSPF(SectionProtocolBase): 

42 """BIRD OSPF protocol configuration.""" 

43 

44 _areas: OSPFAreas 

45 

46 _v4version: str 

47 

48 _ospf_attributes: OSPFAttributes 

49 # OSPF functions 

50 _ospf_functions: OSPFFunctions 

51 

52 def __init__( # pylint: disable=too-many-arguments,too-many-positional-arguments 

53 self, 

54 birdconfig_globals: BirdConfigGlobals, 

55 birdattributes: SectionBirdAttributes, 

56 constants: SectionConstants, 

57 functions: SectionFunctions, 

58 tables: SectionTables, 

59 ) -> None: 

60 """Initialize the object.""" 

61 super().__init__(birdconfig_globals, birdattributes, constants, functions, tables) 

62 

63 # Set section name 

64 self._section = "OSPF Protocol" 

65 

66 # OSPF areas 

67 self._areas = {} 

68 

69 # OSPF version to use for IPv4 

70 self._v4version = "2" 

71 

72 self._ospf_attributes = OSPFAttributes() 

73 # Setup OSPF functions 

74 self._ospf_functions = OSPFFunctions(self.birdconfig_globals, self.functions) 

75 

76 def configure(self) -> None: 

77 """Configure the OSPF protocol.""" 

78 super().configure() 

79 

80 # If we don't have any configuration, just abort 

81 if not self.areas: 

82 return 

83 

84 self.functions.conf.append(self.ospf_functions, deferred=True) 

85 

86 self.tables.conf.append("# OSPF Tables") 

87 self.tables.conf.append("ipv4 table t_ospf4;") 

88 self.tables.conf.append("ipv6 table t_ospf6;") 

89 self.tables.conf.append("") 

90 

91 self._ospf_export_filter() 

92 self._ospf_import_filter() 

93 self._ospf_to_master_export_filter() 

94 self._ospf_to_master_import_filter() 

95 

96 # OSPF protocol configuration 

97 self._setup_protocol("4") 

98 self._setup_protocol("6") 

99 

100 # Configure pipe from OSPF to the master routing table 

101 ospf_master_pipe = ProtocolPipe( 

102 birdconfig_globals=self.birdconfig_globals, 

103 table_from="ospf", 

104 table_to="master", 

105 export_filter_type=ProtocolPipeFilterType.UNVERSIONED, 

106 import_filter_type=ProtocolPipeFilterType.UNVERSIONED, 

107 ) 

108 self.conf.add(ospf_master_pipe) 

109 

110 # Check if we're redistributing connected routes, if we are, create the protocol and pipe 

111 if self.route_policy_redistribute.connected: 

112 # Create an interface list to feed to our routing table 

113 interfaces: list[str] = [] 

114 if isinstance(self.route_policy_redistribute.connected, list): 

115 interfaces = self.route_policy_redistribute.connected 

116 # Add direct protocol for redistribution of connected routes 

117 ospf_direct_protocol = ProtocolDirect( 

118 self.birdconfig_globals, 

119 self.birdattributes, 

120 self.constants, 

121 self.functions, 

122 self.tables, 

123 name="ospf", 

124 interfaces=interfaces, 

125 ) 

126 self.conf.add(ospf_direct_protocol) 

127 # Add pipe 

128 ospf_direct_pipe = ProtocolPipe( 

129 self.birdconfig_globals, 

130 name="ospf", 

131 table_from="ospf", 

132 table_to="direct", 

133 table_export="none", 

134 table_import="all", 

135 ) 

136 self.conf.add(ospf_direct_pipe) 

137 

138 def add_area(self, area_name: str, area_config: OSPFAreaConfig) -> ProtocolOSPFArea: 

139 """Add area to OSPF.""" 

140 

141 # Make sure area doesn't exist 

142 if area_name in self.areas: 

143 raise BirdPlanError(f"OSPF area '{area_name}' already exists") 

144 

145 # Create OSPF area object 

146 area = ProtocolOSPFArea( 

147 self.birdconfig_globals, 

148 self.birdattributes, 

149 self.constants, 

150 self.functions, 

151 self.tables, 

152 self.ospf_attributes, 

153 area_name, 

154 area_config, 

155 ) 

156 

157 # Add area to OSPF 

158 self.areas[area.name] = area # Use the sanitized area name from the area object 

159 

160 return area 

161 

162 def _setup_protocol(self, ipv: str) -> None: 

163 # Work out which OSPF protocol version to use 

164 protocol_version = "3" 

165 if ipv == "4": 

166 protocol_version = self.v4version 

167 

168 self.conf.add(f"protocol ospf v{protocol_version} ospf{ipv} {{") 

169 self.conf.add(f' description "OSPF protocol for IPv{ipv}";') 

170 self.conf.add("") 

171 self.conf.add(f" vrf {self.birdconfig_globals.vrf};") 

172 self.conf.add("") 

173 self.conf.add(f" ipv{ipv} {{") 

174 self.conf.add(f" table t_ospf{ipv};") 

175 self.conf.add("") 

176 self.conf.add(" export filter f_ospf_export;") 

177 self.conf.add(" import filter f_ospf_import;") 

178 self.conf.add("") 

179 self.conf.add(" };") 

180 self.conf.add("") 

181 # Add areas 

182 for _, area in sorted(self.areas.items()): 

183 self.conf.add(area) 

184 # Close off block 

185 self.conf.add("};") 

186 self.conf.add("") 

187 

188 def _ospf_export_filter(self) -> None: 

189 """OSPF export filter setup.""" 

190 # Set our filter name 

191 filter_name = "f_ospf_export" 

192 

193 # Configure OSPF export filter 

194 self.conf.add("# OSPF export filter") 

195 self.conf.add(f"filter {filter_name}") 

196 self.conf.add("string filter_name;") 

197 self.conf.add("{") 

198 self.conf.add(f' filter_name = "{filter_name}";') 

199 # Redistribute connected 

200 if self.route_policy_redistribute.connected: 

201 self.conf.add(f" {self.ospf_functions.redistribute_connected()};") 

202 # Redistribute kernel routes 

203 if self.route_policy_redistribute.kernel: 

204 self.conf.add(f" {self.functions.redistribute_kernel()};") 

205 # Redistribute kernel routes 

206 if self.route_policy_redistribute.kernel_default: 

207 self.conf.add(f" {self.functions.redistribute_kernel_default()};") 

208 # Redistribute static routes 

209 if self.route_policy_redistribute.static: 

210 self.conf.add(f" {self.functions.redistribute_static()};") 

211 # Redistribute static default routes 

212 if self.route_policy_redistribute.static_default: 

213 self.conf.add(f" {self.functions.redistribute_static_default()};") 

214 # Else reject 

215 self.conf.add(" if DEBUG then") 

216 self.conf.add(f' print "[{filter_name}] Rejecting ", net, " from t_ospf export (fallthrough)";') 

217 self.conf.add(" reject;") 

218 self.conf.add("};") 

219 self.conf.add("") 

220 

221 def _ospf_import_filter(self) -> None: 

222 """OSPF import filter setup.""" 

223 # Set our filter name 

224 filter_name = "f_ospf_import" 

225 

226 # Configure OSPF import filter 

227 self.conf.add("# OSPF import filter") 

228 self.conf.add(f"filter {filter_name}") 

229 self.conf.add("string filter_name;") 

230 self.conf.add("{") 

231 # Accept all inbound routes into the table 

232 self.conf.add(" # Import all OSPF routes by default") 

233 self.conf.add(" if DEBUG then") 

234 self.conf.add(f' print "[{filter_name}] Accepting ", net, " from t_ospf import (fallthrough)";') 

235 self.conf.add(" accept;") 

236 self.conf.add("};") 

237 self.conf.add("") 

238 

239 def _ospf_to_master_export_filter(self) -> None: 

240 """OSPF to master export filter setup.""" 

241 # Set our filter name 

242 filter_name = "f_ospf_master_export" 

243 

244 # Configure export filter to master table 

245 self.conf.add("# OSPF export filter to master table") 

246 self.conf.add(f"filter {filter_name}") 

247 self.conf.add("string filter_name;") 

248 self.conf.add("{") 

249 self.conf.add(f' filter_name = "{filter_name}";') 

250 # Accept only OSPF routes into the master table 

251 self.conf.add(" # Export OSPF routes to the master table by default") 

252 self.conf.add(f" {self.ospf_functions.accept_ospf()};") 

253 # Check if we accept the default route 

254 if self.route_policy_accept.default: 

255 self.conf.add(" # Export default route to master (accept:ospf_default is set)") 

256 self.conf.add(f" {self.ospf_functions.accept_ospf_default()};") 

257 # Default to reject 

258 self.conf.add(" # Reject everything else;") 

259 self.conf.add(" if DEBUG then") 

260 self.conf.add(f' print "[{filter_name}] Rejecting ", net, " from t_ospf to master (fallthrough)";') 

261 self.conf.add(" reject;") 

262 self.conf.add("};") 

263 self.conf.add("") 

264 

265 def _ospf_to_master_import_filter(self) -> None: 

266 """OSPF to master import filter setup.""" 

267 # Set our filter name 

268 filter_name = "f_ospf_master_import" 

269 

270 # Configure import filter from master table 

271 self.conf.add("# OSPF import filter from master table") 

272 self.conf.add(f"filter {filter_name}") 

273 self.conf.add("string filter_name;") 

274 self.conf.add("{") 

275 self.conf.add(f' filter_name = "{filter_name}";') 

276 # Redistribute connected 

277 if self.route_policy_redistribute.connected: 

278 self.conf.add(f" {self.ospf_functions.accept_connected()};") 

279 # Redistribute static routes 

280 if self.route_policy_redistribute.static: 

281 self.conf.add(f" {self.functions.accept_static()};") 

282 # Redistribute static default routes 

283 if self.route_policy_redistribute.static_default: 

284 self.conf.add(f" {self.functions.accept_static_default()};") 

285 # Redistribute kernel routes 

286 if self.route_policy_redistribute.kernel: 

287 self.conf.add(f" {self.functions.accept_kernel()};") 

288 # Redistribute kernel default routes 

289 if self.route_policy_redistribute.kernel_default: 

290 self.conf.add(f" {self.functions.accept_kernel_default()};") 

291 # Else accept 

292 self.conf.add(" # Reject by default") 

293 self.conf.add(" if DEBUG then") 

294 self.conf.add(f' print "[{filter_name}] Rejecting ", net, " from master to t_ospf (fallthrough)";') 

295 self.conf.add(" reject;") 

296 self.conf.add("};") 

297 self.conf.add("") 

298 

299 def area(self, name: str) -> ProtocolOSPFArea: 

300 """Return a OSPF area configuration object.""" 

301 if name not in self.areas: 

302 raise BirdPlanError(f"Area '{name}' not found") 

303 return self.areas[name] 

304 

305 @property 

306 def areas(self) -> OSPFAreas: 

307 """Return OSPF areas.""" 

308 return self._areas 

309 

310 @property 

311 def v4version(self) -> str: 

312 """Return OSPF IPv4 version to use.""" 

313 return self._v4version 

314 

315 @v4version.setter 

316 def v4version(self, v4version: str) -> None: 

317 """Set the OSPF IPv4 version to use.""" 

318 self._v4version = v4version 

319 

320 @property 

321 def ospf_attributes(self) -> OSPFAttributes: 

322 """Return our OSPF protocol attributes.""" 

323 return self._ospf_attributes 

324 

325 @property 

326 def ospf_functions(self) -> OSPFFunctions: 

327 """Return our OSPF protocol functions.""" 

328 return self._ospf_functions 

329 

330 @property 

331 def route_policy_accept(self) -> OSPFRoutePolicyAccept: 

332 """Return our route policy for accepting of routes from peers into the master table.""" 

333 return self.ospf_attributes.route_policy_accept 

334 

335 @property 

336 def route_policy_redistribute(self) -> OSPFRoutePolicyRedistribute: 

337 """Return our route policy for redistributing of routes to the main OSPF table.""" 

338 return self.ospf_attributes.route_policy_redistribute