Coverage for birdplan/peeringdb.py: 96%

1# 

2# SPDX-License-Identifier: GPL-3.0-or-later 

3# 

4# Copyright (c) 2019-2024, 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"""PeeringDB support class.""" 

20 

21import time 

22from typing import Any, Dict, Optional 

23 

24import requests 

25 

26from .exceptions import BirdPlanError 

27 

28__all__ = ["PeeringDB"] 

29 

30 

31PeeringDBInfo = Dict[str, Any] 

32 

33 

34# Keep a cache for results returned while loaded into memory 

35# 

36# Example: 

37# peeringdb_cache = { 

38# 'objects': { 

39# 'asn:174': { # ASN 

40# '_timestamp': 0000000000, 

41# 'value': xxxxxx, 

42# } 

43# } 

44# } 

45peeringdb_cache: Dict[str, Dict[str, Any]] = {} 

46 

47# Keep track of the timestamp of our last request 

48peeringdb_last_request: float = 0 

49 

50 

51PEERINGDB_16BIT_LOWER = 64512 

52PEERINGDB_16BIT_UPPER = 65534 

53PEERINGDB_32BIT_LOWER = 4200000000 

54PEERINGDB_32BIT_UPPER = 4294967294 

55 

56 

57class PeeringDB: # pylint: disable=too-few-public-methods 

58 """PeeringDB support class.""" 

59 

60 def __init__(self) -> None: 

61 """Initialize object.""" 

62 

63 def get_prefix_limits(self, asn: int) -> PeeringDBInfo: 

64 """Return our peeringdb info entry, if there is one.""" 

65 global peeringdb_last_request # pylint: disable=global-statement 

66 

67 # We cannot do lookups on private ASN's 

68 if (PEERINGDB_16BIT_LOWER <= asn <= PEERINGDB_16BIT_UPPER) or (PEERINGDB_32BIT_LOWER <= asn <= PEERINGDB_32BIT_UPPER): 

69 return {"info_prefixes4": None, "info_prefixes6": None} 

70 

71 # Try pull result from our cache 

72 result = self._cache(f"asn:{asn}") 

73 # If we can't, grab the result from PeeringDB live 

74 if not result: 

75 # Sleep if last request was made within the past 5s 

76 time_delta = time.time() - peeringdb_last_request + 5 

77 if time_delta < 0: 

78 time.sleep(abs(time_delta)) 

79 # Update the last request 

80 peeringdb_last_request = time.time() 

81 # Request the PeeringDB info for this AS 

82 try: 

83 response = requests.get(f"https://www.peeringdb.com/api/net?asn__in={asn}", timeout=10) 

84 except requests.exceptions.Timeout as e: # pragma: no cover 

85 raise BirdPlanError(f"PeeringDB request timed out: {e}") from None 

86 # Update the last request 

87 peeringdb_last_request = time.time() 

88 # Check the result is not empty 

89 if not response: # pragma: no cover 

90 raise BirdPlanError("PeeringDB returned and empty result") 

91 # Decode response 

92 result = response.json()["data"][0] 

93 # Cache the result we got 

94 self._cache(f"asn:{asn}", result) 

95 

96 # Total cluster .... just to get typing happy 

97 peeringdb_info = {"info_prefixes4": 1, "info_prefixes6": 1} 

98 if result and "info_prefixes4" in result: 

99 peeringdb_info["info_prefixes4"] = result["info_prefixes4"] 

100 if result and "info_prefixes6" in result: 

101 peeringdb_info["info_prefixes6"] = result["info_prefixes6"] 

102 

103 # Lastly return it 

104 return peeringdb_info 

105 

106 def _cache(self, obj: str, value: Optional[PeeringDBInfo] = None) -> Optional[Any]: # noqa: CFQ004 

107 """Retrieve or store value in cache.""" 

108 

109 if "objects" not in peeringdb_cache: 

110 peeringdb_cache["objects"] = {} 

111 

112 if not value: 

113 # If the cached obj does not exist, return None 

114 if obj not in peeringdb_cache["objects"]: 

115 return None 

116 # Grab the cached object 

117 cached = peeringdb_cache["objects"][obj] 

118 # Make sure its timestamp is within 60s of being retrieved, if not, return None 

119 if cached["_timestamp"] + 60 < time.time(): # pragma: no cover 

120 return None 

121 # Else its valid, return the cached value 

122 return cached["value"] 

123 

124 # Set the cached value 

125 peeringdb_cache["objects"][obj] = { 

126 "_timestamp": time.time(), 

127 "value": value, 

128 } 

129 

130 return value