Coverage for src/birdplan/bgpq3.py: 92%
114 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"""BGPQ3/4 support class."""
21import functools
22import ipaddress
23import json
24import shutil
25import subprocess # nosec
26import time
27from typing import Any
29from .exceptions import BirdPlanError
31__all__ = ["BGPQ3"]
34# Keep a cache for results returned while loaded into memory
35#
36# Example:
37# > bgpq3_cache = {
38# > 'whois.radb.net:43': {
39# > 'objects': {
40# > 'AS174': {
41# > '_timestamp': 0000000000,
42# > 'value': xxxxxx,
43# > }
44# > }
45# > }
46# > }
47bgpq3_cache: dict[str, dict[str, Any]] = {}
50class BGPQ3:
51 """BGPQ3 support class."""
53 _host: str
54 _port: int
55 _sources: str
57 def __init__(self, host: str = "whois.radb.net", port: int = 43, sources: str = "RADB") -> None:
58 """Initialize object."""
60 # Grab items we can set and associated defaults
61 self._host = host
62 self._port = port
63 self._sources = sources
65 @functools.lru_cache(maxsize=1) # noqa: B019
66 def _exe(self) -> str:
67 """Return the bgpq3 executable."""
69 for exe in ("bgpq3", "bgpq4"):
70 if shutil.which(exe):
71 return exe
73 raise BirdPlanError("bgpq3/bgpq4 executable not found in PATH")
75 def get_asns(self, as_sets: str | list[str]) -> list[str]: # noqa: C901,PLR0912
76 """Get prefixes."""
78 # Build an object list depending on the type of "objects" above
79 objects: list[str] = []
80 if isinstance(as_sets, str):
81 objects.append(as_sets)
82 else:
83 objects.extend(as_sets)
85 # Grab ASNs
86 is_birdplan_internal = False
87 asns_bgpq3: dict[str, list[str]] = {}
88 for obj in objects:
89 # Try pull result from our cache
90 result: Any = self._cache(f"asns:{obj}")
91 # If we can't, grab the result from BGPQ3 live
92 if not result:
93 # Try query object
94 try:
95 result = self._bgpq3(["-l", "asns", "-t", "-3", obj])
96 except subprocess.CalledProcessError as err:
97 raise BirdPlanError(f"Failed to query IRR ASNs from object '{obj}':\n%s" % err.output.decode("UTF-8")) from None
98 except BirdPlanError as err:
99 raise BirdPlanError(f"Failed to query IRR ASNs from object '{obj}':\n{err}") from None
100 # Cache the result we got
101 self._cache(f"asns:{obj}", result)
102 # Check if this is a birdplan internal object
103 if obj.startswith("_BIRDPLAN:"):
104 is_birdplan_internal = True
105 # Update return value with result
106 asns_bgpq3.update(result)
108 # If we don't have "asns" returned in the JSON structure, raise an exception
109 if "asns" not in asns_bgpq3: # pragma: no cover
110 raise BirdPlanError(f"BGPQ3 output error, expecting 'asns': {asns_bgpq3}")
112 filtered_asns = []
113 for asn in asns_bgpq3["asns"]:
114 # Convert to int for below
115 asn_i = int(asn)
116 # 0 Reserved by [RFC7607] [RFC7607]
117 # 112 Used by the AS112 project to sink misdirected DNS queries; see [RFC7534] [RFC7534]
118 # 23456 AS_TRANS; reserved by [RFC6793] [RFC6793]
119 # 65535 Reserved by [RFC7300] [RFC7300]
120 # 4294967295 Reserved by [RFC7300] [RFC7300]
121 if asn_i in [0, 112, 23456, 65535, 4294967295]:
122 continue
124 #
125 # NK: So objects that start with _BIRDPLAN are used for the tests, so we need to treat them a little differently below
126 # if that's the case.
127 #
129 # 64496-64511 For documentation and sample code; reserved by [RFC5398] [RFC5398]
130 if (64496 <= asn_i <= 64511) and not is_birdplan_internal: # noqa: PLR2004
131 continue
132 # 64512-65534 For private use; reserved by [RFC6996] [RFC6996]
133 if (64512 <= asn_i <= 65534) and not is_birdplan_internal: # noqa: PLR2004
134 continue
135 # 65536-65551 For documentation and sample code; reserved by [RFC5398] [RFC5398]
136 if (65536 <= asn_i <= 65551) and not is_birdplan_internal: # noqa: PLR2004
137 continue
139 # 4200000000-4294967294 For private use; reserved by [RFC6996] [RFC6996]
140 if 4200000000 <= asn_i <= 4294967294: # noqa: PLR2004
141 continue
142 # We passed all the checks, lets add to the filtered list
143 filtered_asns.append(asn)
145 return filtered_asns
147 def get_prefixes(self, as_sets: str | list[str]) -> dict[str, list[str]]:
148 """Get prefixes."""
150 # Build an object list depending on the type of "objects" above
151 objects: list[str] = []
152 if isinstance(as_sets, str):
153 objects.append(as_sets)
154 else:
155 objects.extend(as_sets)
157 # Grab IPv4 and IPv6 prefixes
158 prefixes_bgpq3: dict[str, list[dict[str, Any]]] = {}
159 for obj in objects:
160 # Try pull result from our cache
161 result: Any = self._cache(f"prefixes:{obj}")
162 # If we can't, grab the result from BGPQ3 live
163 if not result:
164 result = {}
165 # Lets see if we get results back from our IRR queries
166 try:
167 result.update(self._bgpq3(["-l", "ipv4", "-m", "24", "-4", "-A", obj]))
168 except subprocess.CalledProcessError as err:
169 raise BirdPlanError(
170 f"Failed to query IRR IPv4 prefixes from object '{obj}':\n%s" % err.output.decode("UTF-8")
171 ) from None
172 try:
173 result.update(self._bgpq3(["-l", "ipv6", "-m", "48", "-6", "-A", obj]))
174 except subprocess.CalledProcessError as err:
175 raise BirdPlanError(
176 f"Failed to query IRR IPv6 prefixes from object '{obj}':\n%s" % err.output.decode("UTF-8")
177 ) from None
178 # Cache the result we got
179 self._cache(f"prefixes:{obj}", result)
180 # Update return value with result
181 prefixes_bgpq3.update(result)
183 # Start out with no prefixes
184 prefixes: dict[str, list[str]] = {"ipv4": [], "ipv6": []}
186 for family in ("ipv4", "ipv6"):
187 for prefix in prefixes_bgpq3[family]:
188 # If it is exact, its easy to add
189 if prefix["exact"]:
190 prefixes[family].append(prefix["prefix"])
191 else:
192 # Work out greater_equal component
193 if "greater-equal" in prefix:
194 greater_equal = prefix["greater-equal"]
195 else:
196 greater_equal = ipaddress.ip_network(prefix["prefix"]).prefixlen
197 # Add prefix, format is %s{%s,%s}
198 prefixes[family].append(f"{prefix['prefix']}{{{greater_equal},{prefix['less-equal']}}}")
200 return prefixes
202 def _bgpq3(self, args: list[str]) -> Any: # noqa: ANN401
203 """Run bgpq3."""
205 # Run the IP tool with JSON output
206 cmd_args = [self._exe(), "-h", self.server, "-j"]
207 # Add our args
208 cmd_args.extend(args)
210 # Grab result from process execution
211 result = subprocess.check_output(cmd_args, stderr=subprocess.STDOUT) # noqa: S603
212 try:
213 decoded = json.loads(result)
214 except json.JSONDecodeError as err:
215 raise BirdPlanError(f"Failed to decode JSON output from {self._exe()}: {err}") from None
216 # Return the decoded json output
217 return decoded
219 def _cache(self, obj: str, value: Any | None = None) -> Any | None: # noqa: ANN401
220 """Retrieve or store value in cache."""
222 if self.server not in bgpq3_cache:
223 bgpq3_cache[self.server] = {"objects": {}}
225 if not value:
226 # If the cached obj does not exist, return None
227 if obj not in bgpq3_cache[self.server]["objects"]:
228 return None
229 # Grab the cached object
230 cached = bgpq3_cache[self.server]["objects"][obj]
231 # Make sure its timestamp is within 60s of being retrieved, if not, return None
232 if cached["_timestamp"] + 60 < time.time(): # pragma: no cover
233 return None
234 # Else its valid, return the cached value
235 return cached["value"]
237 # Set the cached value
238 bgpq3_cache[self.server]["objects"][obj] = {
239 "_timestamp": time.time(),
240 "value": value,
241 }
243 return value
245 @property
246 def server(self) -> str:
247 """Return the server we're using."""
248 return f"{self.host}:{self.port}"
250 @property
251 def host(self) -> str:
252 """Return the host we're using."""
253 return self._host
255 @property
256 def port(self) -> int:
257 """Return the port we're using."""
258 return self._port