Coverage for ipam/lookups.py: 61%

1import netaddr 

2from django.db.models import IntegerField, Lookup, Transform, lookups 

3 

4 

5class NetFieldDecoratorMixin: 

6 

7 def process_lhs(self, qn, connection, lhs=None): 

8 lhs = lhs or self.lhs 

9 lhs_string, lhs_params = qn.compile(lhs) 

10 lhs_string = f'TEXT({lhs_string})' 

11 return lhs_string, lhs_params 

12 

13 

14class IExact(NetFieldDecoratorMixin, lookups.IExact): 

15 

16 def get_rhs_op(self, connection, rhs): 

17 return f'= LOWER({rhs})' 

18 

19 

20class EndsWith(NetFieldDecoratorMixin, lookups.EndsWith): 

21 pass 

22 

23 

24class IEndsWith(NetFieldDecoratorMixin, lookups.IEndsWith): 

25 pass 

26 

27 def get_rhs_op(self, connection, rhs): 

28 return f'LIKE LOWER({rhs})' 

29 

30 

31class StartsWith(NetFieldDecoratorMixin, lookups.StartsWith): 

32 lookup_name = 'startswith' 

33 

34 

35class IStartsWith(NetFieldDecoratorMixin, lookups.IStartsWith): 

36 pass 

37 

38 def get_rhs_op(self, connection, rhs): 

39 return f'LIKE LOWER({rhs})' 

40 

41 

42class Regex(NetFieldDecoratorMixin, lookups.Regex): 

43 pass 

44 

45 

46class IRegex(NetFieldDecoratorMixin, lookups.IRegex): 

47 pass 

48 

49 

50class NetContainsOrEquals(Lookup): 

51 lookup_name = 'net_contains_or_equals' 

52 

53 def as_sql(self, qn, connection): 

54 lhs, lhs_params = self.process_lhs(qn, connection) 

55 rhs, rhs_params = self.process_rhs(qn, connection) 

56 params = lhs_params + rhs_params 

57 return f'{lhs} >>= {rhs}', params 

58 

59 

60class NetContains(Lookup): 

61 lookup_name = 'net_contains' 

62 

63 def as_sql(self, qn, connection): 

64 lhs, lhs_params = self.process_lhs(qn, connection) 

65 rhs, rhs_params = self.process_rhs(qn, connection) 

66 params = lhs_params + rhs_params 

67 return f'{lhs} >> {rhs}', params 

68 

69 

70class NetContained(Lookup): 

71 lookup_name = 'net_contained' 

72 

73 def as_sql(self, qn, connection): 

74 lhs, lhs_params = self.process_lhs(qn, connection) 

75 rhs, rhs_params = self.process_rhs(qn, connection) 

76 params = lhs_params + rhs_params 

77 return f'{lhs} << {rhs}', params 

78 

79 

80class NetContainedOrEqual(Lookup): 

81 lookup_name = 'net_contained_or_equal' 

82 

83 def as_sql(self, qn, connection): 

84 lhs, lhs_params = self.process_lhs(qn, connection) 

85 rhs, rhs_params = self.process_rhs(qn, connection) 

86 params = lhs_params + rhs_params 

87 return f'{lhs} <<= {rhs}', params 

88 

89 

90class NetHost(Lookup): 

91 lookup_name = 'net_host' 

92 

93 def as_sql(self, qn, connection): 

94 lhs, lhs_params = self.process_lhs(qn, connection) 

95 rhs, rhs_params = self.process_rhs(qn, connection) 

96 # Query parameters are automatically converted to IPNetwork objects, which are then turned to strings. We need 

97 # to omit the mask portion of the object's string representation to match PostgreSQL's HOST() function. 

98 # Note: params may be tuples (Django 6.0+) or lists (older Django), so convert before mutating. 

99 rhs_params = list(rhs_params) 

100 if rhs_params: 100 ↛ 102line 100 didn't jump to line 102 because the condition on line 100 was always true

101 rhs_params[0] = rhs_params[0].split('/')[0] 

102 params = list(lhs_params) + rhs_params 

103 # Cast to INET so the predicate matches the inet ipam_ipaddress_host index. 

104 return f'CAST(HOST({lhs}) AS INET) = {rhs}', params 

105 

106 

107class NetIn(Lookup): 

108 lookup_name = 'net_in' 

109 

110 def get_prep_lookup(self): 

111 # Don't cast the query value to a netaddr object, since it may or may not include a mask. 

112 return self.rhs 

113 

114 def as_sql(self, qn, connection): 

115 lhs = self.process_lhs(qn, connection)[0] 

116 rhs_params = self.process_rhs(qn, connection)[1] 

117 with_mask, without_mask = [], [] 

118 for address in rhs_params[0]: 

119 if '/' in address: 

120 with_mask.append(address) 

121 else: 

122 without_mask.append(address) 

123 

124 address_in_clause = self.create_in_clause('{} IN ('.format(lhs), len(with_mask)) 

125 # Cast to INET so the predicate matches the inet ipam_ipaddress_host index. 

126 host_in_clause = self.create_in_clause('CAST(HOST({}) AS INET) IN ('.format(lhs), len(without_mask)) 

127 

128 if with_mask and not without_mask: 

129 return address_in_clause, with_mask 

130 if not with_mask and without_mask: 

131 return host_in_clause, without_mask 

132 

133 in_clause = '({}) OR ({})'.format(address_in_clause, host_in_clause) 

134 with_mask.extend(without_mask) 

135 return in_clause, with_mask 

136 

137 @staticmethod 

138 def create_in_clause(clause_part, max_size): 

139 clause_elements = [clause_part] 

140 for offset in range(0, max_size): 

141 if offset > 0: 

142 clause_elements.append(', ') 

143 clause_elements.append('%s') 

144 clause_elements.append(')') 

145 return ''.join(clause_elements) 

146 

147 

148class NetHostContained(Lookup): 

149 """ 

150 Check for the host portion of an IP address without regard to its mask. This allows us to find e.g. 192.0.2.1/24 

151 when specifying a parent prefix of 192.0.2.0/26. 

152 """ 

153 lookup_name = 'net_host_contained' 

154 

155 def as_sql(self, qn, connection): 

156 lhs, lhs_params = self.process_lhs(qn, connection) 

157 rhs, rhs_params = self.process_rhs(qn, connection) 

158 params = lhs_params + rhs_params 

159 return f'CAST(HOST({lhs}) AS INET) <<= {rhs}', params 

160 

161 

162class NetHostBetween(Lookup): 

163 """ 

164 Match host addresses (mask ignored) falling inclusively between two bounds. The left-hand 

165 side is kept as an inet-typed host expression so PostgreSQL can use the host expression 

166 indexes on the IPAM address and range tables; the CAST(HOST(...) AS INET) spelling matches 

167 NetHost/NetIn for consistency (PostgreSQL canonicalizes the INET(HOST(...)) function form 

168 to the same expression). 

169 """ 

170 lookup_name = 'host_between' 

171 

172 def get_prep_lookup(self): 

173 if not isinstance(self.rhs, (list, tuple)) or len(self.rhs) != 2: 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true

174 raise ValueError('The host_between lookup requires a (lower, upper) pair of bounds') 

175 try: 

176 # Normalize to bare hosts; reject malformed values before they reach SQL. 

177 lower, upper = (netaddr.IPNetwork(str(bound)).ip for bound in self.rhs) 

178 except (netaddr.AddrFormatError, ValueError) as e: 

179 raise ValueError(f'Invalid host_between bound: {e}') from e 

180 if lower.version != upper.version: 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true

181 raise ValueError('host_between bounds must not mix address families') 

182 return lower, upper 

183 

184 def as_sql(self, qn, connection): 

185 lhs, lhs_params = self.process_lhs(qn, connection) 

186 params = list(lhs_params) + [str(bound) for bound in self.rhs] 

187 return f'CAST(HOST({lhs}) AS INET) BETWEEN %s AND %s', params 

188 

189 

190class NetFamily(Transform): 

191 lookup_name = 'family' 

192 function = 'FAMILY' 

193 

194 @property 

195 def output_field(self): 

196 return IntegerField() 

197 

198 

199class NetMaskLength(Transform): 

200 function = 'MASKLEN' 

201 lookup_name = 'net_mask_length' 

202 

203 @property 

204 def output_field(self): 

205 return IntegerField() 

206 

207 

208class Host(Transform): 

209 function = 'HOST' 

210 lookup_name = 'host' 

211 

212 

213class Inet(Transform): 

214 function = 'INET' 

215 lookup_name = 'inet'