Coverage for src/birdplan/bird_config/sections/protocols/rpki.py: 93%

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 RPKI protocol configuration.""" 

20 

21import urllib.parse 

22 

23from ...globals import BirdConfigGlobals 

24from ..base import SectionBase 

25from ..bird_attributes import SectionBirdAttributes 

26from ..tables import SectionTables 

27 

28__all__ = ["ProtocolRPKI"] 

29 

30 

31BIRDPLAN_RPKI_PRIVATE_KEY = "/etc/birdplan/rpki_id_rsa" 

32BIRDPLAN_RPKI_PUBLIC_KEY = "/etc/birdplan/rpki_known_hosts" 

33BIRDPLAN_RPKI_USERNAME = "rpki" 

34 

35 

36class RPKISource: # pylint: disable=too-many-instance-attributes 

37 """RPKI server configuration.""" 

38 

39 # List-based sources 

40 _rpki_data: list[str] | None 

41 # String-based sources, aka a URI 

42 _protocol: str | None 

43 _hostname: str | None 

44 _port: int | None 

45 

46 _private_key: str | None 

47 _public_key: str | None 

48 _username: str | None 

49 

50 _local_address: str | None 

51 _refresh: int | None 

52 _retry: int | None 

53 

54 def __init__(self, rpki_source: str | list[str]) -> None: # noqa: C901,PLR0912 

55 """Initialize object.""" 

56 

57 self._rpki_data = None 

58 self._protocol = None 

59 self._hostname = None 

60 

61 self._port = None 

62 self._private_key = None 

63 self._public_key = None 

64 self._username = None 

65 

66 self._local_address = None 

67 self._refresh = None 

68 self._retry = None 

69 

70 # Check if we have a list of RPKI data 

71 if isinstance(rpki_source, list): 

72 self._rpki_data = rpki_source 

73 

74 else: 

75 # Parse RPKI server URI to get protocol, hostname, port and parameters 

76 parsed_uri = urllib.parse.urlparse(rpki_source) 

77 

78 # Grab the protocol 

79 self._protocol = parsed_uri.scheme 

80 if self._protocol not in ["ssh", "tcp"]: 

81 raise ValueError(f"Invalid protocol '{self._protocol}' for RPKI server '{rpki_source}'") 

82 

83 # Check if we have a hostname we can use 

84 hostname = parsed_uri.hostname 

85 if not hostname: 

86 raise ValueError(f"Invalid hostname '{hostname}' for RPKI server '{rpki_source}'") 

87 self._hostname = hostname 

88 

89 # Work out which port we're using 

90 if parsed_uri.port: 

91 self._port = parsed_uri.port 

92 elif self._protocol == "ssh": 

93 self._port = 22 

94 elif self._protocol == "tcp": 

95 self._port = 323 

96 

97 # Grab parameters 

98 parameters = urllib.parse.parse_qs(parsed_uri.query) 

99 

100 # If we're dealing with SSH, check for private and public keys in the query parameters 

101 if self._protocol == "ssh": 

102 # Private key 

103 if "private_key" in parameters: 

104 self._private_key = parameters["private_key"][-1] 

105 else: 

106 self._private_key = BIRDPLAN_RPKI_PRIVATE_KEY 

107 # Public key 

108 if "public_key" in parameters: 

109 self._public_key = parameters["public_key"][-1] 

110 # Username 

111 if "username" in parameters: 

112 self._username = parameters["username"][-1] 

113 else: 

114 self._username = BIRDPLAN_RPKI_USERNAME 

115 

116 # Check for additional options 

117 if "local_address" in parameters: 

118 self._local_address = parameters["local_address"][-1] 

119 if "refresh" in parameters: 

120 self._refresh = int(parameters["refresh"][-1]) 

121 if "retry" in parameters: 

122 self._retry = int(parameters["retry"][-1]) 

123 

124 @property 

125 def protocol(self) -> str | None: 

126 """Return the protocol.""" 

127 return self._protocol 

128 

129 @property 

130 def hostname(self) -> str | None: 

131 """Return the hostname.""" 

132 return self._hostname 

133 

134 @property 

135 def port(self) -> int | None: 

136 """Return the port.""" 

137 return self._port 

138 

139 @property 

140 def private_key(self) -> str | None: 

141 """Return the private key.""" 

142 return self._private_key 

143 

144 @property 

145 def public_key(self) -> str | None: 

146 """Return the public key.""" 

147 return self._public_key 

148 

149 @property 

150 def username(self) -> str | None: 

151 """Return the username.""" 

152 return self._username 

153 

154 @property 

155 def rpki_data(self) -> list[str] | None: 

156 """Return the RPKI data.""" 

157 return self._rpki_data 

158 

159 @property 

160 def local_address(self) -> str | None: 

161 """Return the local address.""" 

162 return self._local_address 

163 

164 @property 

165 def refresh(self) -> int | None: 

166 """Return the refresh interval.""" 

167 return self._refresh 

168 

169 @property 

170 def retry(self) -> int | None: 

171 """Return the retry interval.""" 

172 return self._retry 

173 

174 @property 

175 def is_uri(self) -> bool: 

176 """Return True if the source is a URI.""" 

177 return self._protocol is not None 

178 

179 

180class ProtocolRPKI(SectionBase): 

181 """BIRD RPKI protocol configuration.""" 

182 

183 _server: RPKISource 

184 _birdattributes: SectionBirdAttributes 

185 _tables: SectionTables 

186 

187 def __init__( 

188 self, 

189 birdconfig_globals: BirdConfigGlobals, 

190 birdattributes: SectionBirdAttributes, 

191 tables: SectionTables, 

192 rpki_source: RPKISource, 

193 ) -> None: 

194 """Initialize the object.""" 

195 super().__init__(birdconfig_globals) 

196 self._server = rpki_source 

197 self._birdattributes = birdattributes 

198 self._tables = tables 

199 

200 def configure(self) -> None: 

201 """Configure the RPKI protocol.""" 

202 super().configure() 

203 

204 # Set section header 

205 self._section = "RPKI Protocol" 

206 

207 # Configure the RPKI protocol 

208 self._configure_tables_rpki() 

209 

210 if self.server.is_uri: 

211 self._configure_protocol_rpki_uri() 

212 else: 

213 self._configure_protocol_rpki_static() 

214 

215 def _configure_tables_rpki(self) -> None: 

216 """Tables configuration.""" 

217 self.tables.conf.append("# RPKI ROA Tables") 

218 self.tables.conf.append("roa4 table t_roa4;") 

219 self.tables.conf.append("roa6 table t_roa6;") 

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

221 

222 def _configure_protocol_rpki_static(self) -> None: 

223 """Configure RPKI static protocol.""" 

224 # Build the IPv4 static table 

225 self.conf.add("protocol static rpki4 {") 

226 self.conf.add("") 

227 self.conf.add(" roa4 { table t_roa4; };") 

228 self.conf.add("") 

229 if self.server.rpki_data: 

230 for route in self.server.rpki_data: 

231 if "." not in route: 

232 continue 

233 self.conf.add(f" route {route};") 

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

235 # Build the IPv6 static table 

236 self.conf.add("protocol static rpki6 {") 

237 self.conf.add("") 

238 self.conf.add(" roa6 { table t_roa6; };") 

239 self.conf.add("") 

240 if self.server.rpki_data: 

241 for route in self.server.rpki_data: 

242 if ":" not in route: 

243 continue 

244 self.conf.add(f" route {route};") 

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

246 

247 def _configure_protocol_rpki_uri(self) -> None: 

248 """Protocol configuration.""" 

249 self.conf.add("protocol rpki p_rpki {") 

250 self.conf.add("") 

251 self.conf.add(" roa4 { table t_roa4; };") 

252 self.conf.add(" roa6 { table t_roa6; };") 

253 

254 # SSH support 

255 if self.server.protocol == "ssh": 

256 self.conf.add(f" remote {self.server.hostname} port {self.server.port};") 

257 self.conf.add(" transport ssh {") 

258 self.conf.add(f' bird private key "{self.server.private_key}";') 

259 if self.server.public_key: 

260 self.conf.add(f' remote public key "{self.server.public_key}";') 

261 self.conf.add(f' user "{self.server.username}";') 

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

263 

264 # TCP support 

265 elif self.server.protocol == "tcp": 

266 self.conf.add(f" remote {self.server.hostname} port {self.server.port};") 

267 

268 # Check if we have additional options 

269 if self.server.local_address: 

270 self.conf.add(f" local address {self.server.local_address};") 

271 if self.server.refresh: 

272 self.conf.add(f" refresh {self.server.refresh};") 

273 if self.server.retry: 

274 self.conf.add(f" retry {self.server.retry};") 

275 

276 self.conf.add("") 

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

278 self.conf.add("") 

279 

280 @property 

281 def server(self) -> RPKISource: 

282 """Return the RPKI server string.""" 

283 return self._server 

284 

285 @property 

286 def birdattributes(self) -> SectionBirdAttributes: 

287 """Return the attributes section.""" 

288 return self._birdattributes 

289 

290 @property 

291 def tables(self) -> SectionTables: 

292 """Return the tables section.""" 

293 return self._tables