Coverage for ipam/api/views.py: 86%

1from copy import deepcopy 

2 

3from django.contrib.contenttypes.prefetch import GenericPrefetch 

4from django.core.exceptions import ObjectDoesNotExist, PermissionDenied 

5from django.db import router, transaction 

6from django.shortcuts import get_object_or_404 

7from django.utils.translation import gettext as _ 

8from django_pg_utils import advisory_lock 

9from drf_spectacular.utils import extend_schema 

10from netaddr import IPSet 

11from rest_framework import status 

12from rest_framework.exceptions import ValidationError 

13from rest_framework.response import Response 

14from rest_framework.routers import APIRootView 

15from rest_framework.views import APIView 

16 

17from dcim.models import Interface 

18from ipam import filtersets 

19from ipam.models import * 

20from ipam.utils import get_next_available_prefix 

21from netbox.api.viewsets import NetBoxModelViewSet 

22from netbox.api.viewsets.mixins import ObjectValidationMixin, discard_events_on_rollback 

23from netbox.config import get_config 

24from netbox.constants import ADVISORY_LOCK_KEYS 

25from utilities.api import get_positional_errors, get_serializer_for_model 

26from virtualization.models import VMInterface 

27 

28from . import serializers 

29 

30 

31class IPAMRootView(APIRootView): 

32 """ 

33 IPAM API root view 

34 """ 

35 def get_view_name(self): 

36 return 'IPAM' 

37 

38 

39# 

40# Viewsets 

41# 

42 

43class ASNRangeViewSet(NetBoxModelViewSet): 

44 queryset = ASNRange.objects.all() 

45 serializer_class = serializers.ASNRangeSerializer 

46 filterset_class = filtersets.ASNRangeFilterSet 

47 

48 

49class ASNViewSet(NetBoxModelViewSet): 

50 queryset = ASN.objects.all() 

51 serializer_class = serializers.ASNSerializer 

52 filterset_class = filtersets.ASNFilterSet 

53 

54 

55class VRFViewSet(NetBoxModelViewSet): 

56 queryset = VRF.objects.all() 

57 serializer_class = serializers.VRFSerializer 

58 filterset_class = filtersets.VRFFilterSet 

59 

60 

61class RouteTargetViewSet(NetBoxModelViewSet): 

62 queryset = RouteTarget.objects.all() 

63 serializer_class = serializers.RouteTargetSerializer 

64 filterset_class = filtersets.RouteTargetFilterSet 

65 

66 

67class RIRViewSet(NetBoxModelViewSet): 

68 queryset = RIR.objects.all() 

69 serializer_class = serializers.RIRSerializer 

70 filterset_class = filtersets.RIRFilterSet 

71 

72 

73class AggregateViewSet(NetBoxModelViewSet): 

74 queryset = Aggregate.objects.all() 

75 serializer_class = serializers.AggregateSerializer 

76 filterset_class = filtersets.AggregateFilterSet 

77 

78 

79class RoleViewSet(NetBoxModelViewSet): 

80 queryset = Role.objects.all() 

81 serializer_class = serializers.RoleSerializer 

82 filterset_class = filtersets.RoleFilterSet 

83 

84 

85class PrefixViewSet(NetBoxModelViewSet): 

86 queryset = Prefix.objects.prefetch_related("scope") 

87 serializer_class = serializers.PrefixSerializer 

88 filterset_class = filtersets.PrefixFilterSet 

89 

90 parent_model = Prefix # AvailableIPsMixin 

91 

92 def get_serializer_class(self): 

93 if self.action == "available_prefixes" and self.request.method == "POST": 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true

94 return serializers.PrefixLengthSerializer 

95 return super().get_serializer_class() 

96 

97 

98class IPRangeViewSet(NetBoxModelViewSet): 

99 queryset = IPRange.objects.all() 

100 serializer_class = serializers.IPRangeSerializer 

101 filterset_class = filtersets.IPRangeFilterSet 

102 

103 parent_model = IPRange # AvailableIPsMixin 

104 

105 

106class IPAddressViewSet(NetBoxModelViewSet): 

107 queryset = IPAddress.objects.prefetch_related( 

108 GenericPrefetch( 

109 "assigned_object", 

110 [ 

111 # serializers are taken according to IPADDRESS_ASSIGNMENT_MODELS 

112 FHRPGroup.objects.all(), 

113 Interface.objects.select_related("cable", "device"), 

114 VMInterface.objects.select_related("virtual_machine"), 

115 ], 

116 ), 

117 ) 

118 serializer_class = serializers.IPAddressSerializer 

119 filterset_class = filtersets.IPAddressFilterSet 

120 

121 @advisory_lock(ADVISORY_LOCK_KEYS['available-ips']) 

122 def create(self, request, *args, **kwargs): 

123 return super().create(request, *args, **kwargs) 

124 

125 @advisory_lock(ADVISORY_LOCK_KEYS['available-ips']) 

126 def update(self, request, *args, **kwargs): 

127 return super().update(request, *args, **kwargs) 

128 

129 @advisory_lock(ADVISORY_LOCK_KEYS['available-ips']) 

130 def destroy(self, request, *args, **kwargs): 

131 return super().destroy(request, *args, **kwargs) 

132 

133 

134class FHRPGroupViewSet(NetBoxModelViewSet): 

135 queryset = FHRPGroup.objects.all() 

136 serializer_class = serializers.FHRPGroupSerializer 

137 filterset_class = filtersets.FHRPGroupFilterSet 

138 

139 

140class FHRPGroupAssignmentViewSet(NetBoxModelViewSet): 

141 queryset = FHRPGroupAssignment.objects.all() 

142 serializer_class = serializers.FHRPGroupAssignmentSerializer 

143 filterset_class = filtersets.FHRPGroupAssignmentFilterSet 

144 

145 

146class VLANGroupViewSet(NetBoxModelViewSet): 

147 queryset = VLANGroup.objects.annotate_utilization() 

148 serializer_class = serializers.VLANGroupSerializer 

149 filterset_class = filtersets.VLANGroupFilterSet 

150 

151 

152class VLANViewSet(NetBoxModelViewSet): 

153 queryset = VLAN.objects.prefetch_related( 

154 'l2vpn_terminations', # Referenced by VLANSerializer.l2vpn_termination 

155 ) 

156 serializer_class = serializers.VLANSerializer 

157 filterset_class = filtersets.VLANFilterSet 

158 

159 

160class VLANTranslationPolicyViewSet(NetBoxModelViewSet): 

161 queryset = VLANTranslationPolicy.objects.all() 

162 serializer_class = serializers.VLANTranslationPolicySerializer 

163 filterset_class = filtersets.VLANTranslationPolicyFilterSet 

164 

165 

166class VLANTranslationRuleViewSet(NetBoxModelViewSet): 

167 queryset = VLANTranslationRule.objects.all() 

168 serializer_class = serializers.VLANTranslationRuleSerializer 

169 filterset_class = filtersets.VLANTranslationRuleFilterSet 

170 

171 

172class ServiceTemplateViewSet(NetBoxModelViewSet): 

173 queryset = ServiceTemplate.objects.all() 

174 serializer_class = serializers.ServiceTemplateSerializer 

175 filterset_class = filtersets.ServiceTemplateFilterSet 

176 

177 

178class ServiceViewSet(NetBoxModelViewSet): 

179 queryset = Service.objects.all() 

180 serializer_class = serializers.ServiceSerializer 

181 filterset_class = filtersets.ServiceFilterSet 

182 

183 

184# 

185# Views 

186# 

187 

188def get_results_limit(request): 

189 """ 

190 Return the lesser of the specified limit (if any) and the configured MAX_PAGE_SIZE. 

191 """ 

192 config = get_config() 

193 try: 

194 limit = int(request.query_params.get('limit', config.PAGINATE_COUNT)) or config.MAX_PAGE_SIZE 

195 except ValueError: 

196 limit = config.PAGINATE_COUNT 

197 if config.MAX_PAGE_SIZE: 197 ↛ 200line 197 didn't jump to line 200 because the condition on line 197 was always true

198 limit = min(limit, config.MAX_PAGE_SIZE) 

199 

200 return limit 

201 

202 

203class AvailableObjectsView(ObjectValidationMixin, APIView): 

204 """ 

205 Return a list of dicts representing child objects that have not yet been created for a parent object. 

206 """ 

207 read_serializer_class = None 

208 write_serializer_class = None 

209 advisory_lock_key = None 

210 

211 def get_parent(self, request, pk): 

212 """ 

213 Return the parent object. 

214 """ 

215 raise NotImplementedError() 

216 

217 def get_available_objects(self, parent, limit=None): 

218 """ 

219 Return all available objects for the parent. 

220 """ 

221 raise NotImplementedError() 

222 

223 def get_extra_context(self, parent): 

224 """ 

225 Return any extra context data for the serializer. 

226 """ 

227 return {} 

228 

229 def check_sufficient_available(self, requested_objects, available_objects): 

230 """ 

231 Check if there exist a sufficient number of available objects to satisfy the request. 

232 """ 

233 return len(requested_objects) <= len(available_objects) 

234 

235 def prep_object_data(self, requested_objects, available_objects, parent): 

236 """ 

237 Prepare data by setting any programmatically determined object attributes (e.g. next available VLAN ID) 

238 on the request data. 

239 """ 

240 return requested_objects 

241 

242 def get(self, request, pk): 

243 parent = self.get_parent(request, pk) 

244 limit = get_results_limit(request) 

245 available_objects = self.get_available_objects(parent, limit) 

246 

247 serializer = self.read_serializer_class(available_objects, many=True, context={ 

248 'request': request, 

249 **self.get_extra_context(parent), 

250 }) 

251 

252 return Response(serializer.data) 

253 

254 def post(self, request, pk): 

255 self.queryset = self.queryset.restrict(request.user, 'add') 

256 parent = self.get_parent(request, pk) 

257 

258 # Normalize request data to a list of objects 

259 requested_objects = request.data if isinstance(request.data, list) else [request.data] 

260 limit = len(requested_objects) 

261 

262 # Serialize and validate the request data 

263 serializer = self.write_serializer_class(data=requested_objects, many=True, context={ 

264 'request': request, 

265 **self.get_extra_context(parent), 

266 }) 

267 if not serializer.is_valid(): 

268 # Report the errors by the position of each entry in the request, as the serializer is 

269 # always bound to a list (a single object having been wrapped in one above) 

270 return Response( 

271 get_positional_errors(serializer.errors, len(requested_objects)), 

272 status=status.HTTP_400_BAD_REQUEST 

273 ) 

274 

275 with advisory_lock(ADVISORY_LOCK_KEYS[self.advisory_lock_key]): 

276 available_objects = self.get_available_objects(parent, limit) 

277 

278 # Determine if the requested number of objects is available 

279 if not self.check_sufficient_available(serializer.validated_data, available_objects): 279 ↛ 280line 279 didn't jump to line 280 because the condition on line 279 was never true

280 return Response( 

281 {"detail": "Insufficient resources are available to satisfy the request"}, 

282 status=status.HTTP_409_CONFLICT 

283 ) 

284 

285 # Prepare object data for deserialization 

286 requested_objects = self.prep_object_data(deepcopy(requested_objects), available_objects, parent) 

287 

288 # Initialize the serializer with a list or a single object depending on what was requested 

289 serializer_class = get_serializer_for_model(self.queryset.model) 

290 context = {'request': request} 

291 if isinstance(request.data, list): 291 ↛ 294line 291 didn't jump to line 294 because the condition on line 291 was always true

292 serializer = serializer_class(data=requested_objects, many=True, context=context) 

293 else: 

294 serializer = serializer_class(data=requested_objects[0], context=context) 

295 

296 if not serializer.is_valid(): 296 ↛ 299line 296 didn't jump to line 299 because the condition on line 296 was never true

297 # A list request is reported by position; a single object carries no position, and 

298 # its errors pass through unchanged 

299 return Response( 

300 get_positional_errors(serializer.errors, len(requested_objects)), 

301 status=status.HTTP_400_BAD_REQUEST 

302 ) 

303 

304 # Create the new IP address(es) 

305 using = router.db_for_write(self.queryset.model) 

306 try: 

307 with transaction.atomic(using=using), discard_events_on_rollback(self, using=using): 

308 created = serializer.save() 

309 self._validate_objects(created) 

310 except ObjectDoesNotExist: 

311 raise PermissionDenied() 

312 

313 return Response(serializer.data, status=status.HTTP_201_CREATED) 

314 

315 

316class AvailableASNsView(AvailableObjectsView): 

317 queryset = ASN.objects.all() 

318 read_serializer_class = serializers.AvailableASNSerializer 

319 write_serializer_class = serializers.AvailableASNSerializer 

320 advisory_lock_key = 'available-asns' 

321 

322 def get_parent(self, request, pk): 

323 return get_object_or_404(ASNRange.objects.restrict(request.user), pk=pk) 

324 

325 def get_available_objects(self, parent, limit=None): 

326 return parent.get_available_asns()[:limit] 

327 

328 def get_extra_context(self, parent): 

329 return { 

330 'range': parent, 

331 } 

332 

333 def prep_object_data(self, requested_objects, available_objects, parent): 

334 for i, request_data in enumerate(requested_objects): 

335 request_data.update({ 

336 'rir': parent.rir.pk, 

337 'range': parent.pk, 

338 'asn': available_objects[i], 

339 }) 

340 

341 return requested_objects 

342 

343 @extend_schema(methods=["get"], responses={200: serializers.AvailableASNSerializer(many=True)}) 

344 def get(self, request, pk): 

345 return super().get(request, pk) 

346 

347 @extend_schema( 

348 methods=["post"], 

349 responses={201: serializers.ASNSerializer(many=True)}, 

350 request=serializers.ASNSerializer(many=True), 

351 ) 

352 def post(self, request, pk): 

353 return super().post(request, pk) 

354 

355 

356class AvailablePrefixesView(AvailableObjectsView): 

357 queryset = Prefix.objects.all() 

358 read_serializer_class = serializers.AvailablePrefixSerializer 

359 write_serializer_class = serializers.PrefixLengthSerializer 

360 advisory_lock_key = 'available-prefixes' 

361 

362 def get_parent(self, request, pk): 

363 return get_object_or_404(Prefix.objects.restrict(request.user), pk=pk) 

364 

365 def get_available_objects(self, parent, limit=None): 

366 return parent.get_available_prefixes().iter_cidrs() 

367 

368 def check_sufficient_available(self, requested_objects, available_objects): 

369 available_prefixes = IPSet(available_objects) 

370 for requested_object in requested_objects: 370 ↛ 371line 370 didn't jump to line 371 because the loop on line 370 never started

371 if not get_next_available_prefix(available_prefixes, requested_object['prefix_length']): 

372 return False 

373 return True 

374 

375 def get_extra_context(self, parent): 

376 return { 

377 'prefix': parent, 

378 'vrf': parent.vrf, 

379 } 

380 

381 def prep_object_data(self, requested_objects, available_objects, parent): 

382 available_prefixes = IPSet(available_objects) 

383 for i, request_data in enumerate(requested_objects): 383 ↛ 386line 383 didn't jump to line 386 because the loop on line 383 never started

384 

385 # Find the first available prefix equal to or larger than the requested size 

386 if allocated_prefix := get_next_available_prefix(available_prefixes, request_data['prefix_length']): 

387 request_data.update({ 

388 'prefix': allocated_prefix, 

389 'vrf': parent.vrf.pk if parent.vrf else None, 

390 }) 

391 else: 

392 raise ValidationError(_("Insufficient space is available to accommodate the requested prefix size(s)")) 

393 

394 return requested_objects 

395 

396 @extend_schema(methods=["get"], responses={200: serializers.AvailablePrefixSerializer(many=True)}) 

397 def get(self, request, pk): 

398 return super().get(request, pk) 

399 

400 @extend_schema( 

401 methods=["post"], 

402 responses={201: serializers.PrefixSerializer(many=True)}, 

403 request=serializers.PrefixLengthSerializer(many=True), 

404 ) 

405 def post(self, request, pk): 

406 return super().post(request, pk) 

407 

408 

409class AvailableIPAddressesView(AvailableObjectsView): 

410 queryset = IPAddress.objects.all() 

411 read_serializer_class = serializers.AvailableIPSerializer 

412 write_serializer_class = serializers.AvailableIPRequestSerializer 

413 advisory_lock_key = 'available-ips' 

414 

415 def get_available_objects(self, parent, limit=None): 

416 # Calculate available IPs within the parent 

417 ip_list = [] 

418 for index, ip in enumerate(parent.iter_available_ips(), start=1): 

419 ip_list.append(ip) 

420 if index == limit: 

421 break 

422 return ip_list 

423 

424 def get_extra_context(self, parent): 

425 return { 

426 'parent': parent, 

427 'vrf': parent.vrf, 

428 } 

429 

430 def prep_object_data(self, requested_objects, available_objects, parent): 

431 available_ips = iter(available_objects) 

432 for i, request_data in enumerate(requested_objects): 432 ↛ 433line 432 didn't jump to line 433 because the loop on line 432 never started

433 prefix_length = request_data.pop('prefix_length', None) or parent.mask_length 

434 request_data.update({ 

435 'address': f'{next(available_ips)}/{prefix_length}', 

436 'vrf': parent.vrf.pk if parent.vrf else None, 

437 }) 

438 

439 return requested_objects 

440 

441 @extend_schema(methods=["get"], responses={200: serializers.AvailableIPSerializer(many=True)}) 

442 def get(self, request, pk): 

443 return super().get(request, pk) 

444 

445 @extend_schema( 

446 methods=["post"], 

447 responses={201: serializers.IPAddressSerializer(many=True)}, 

448 request=serializers.AvailableIPRequestSerializer(many=True), 

449 ) 

450 def post(self, request, pk): 

451 return super().post(request, pk) 

452 

453 

454class PrefixAvailableIPAddressesView(AvailableIPAddressesView): 

455 

456 def get_parent(self, request, pk): 

457 return get_object_or_404(Prefix.objects.restrict(request.user), pk=pk) 

458 

459 

460class IPRangeAvailableIPAddressesView(AvailableIPAddressesView): 

461 

462 def get_parent(self, request, pk): 

463 return get_object_or_404(IPRange.objects.restrict(request.user), pk=pk) 

464 

465 

466class AvailableVLANsView(AvailableObjectsView): 

467 queryset = VLAN.objects.all() 

468 read_serializer_class = serializers.AvailableVLANSerializer 

469 write_serializer_class = serializers.CreateAvailableVLANSerializer 

470 advisory_lock_key = 'available-vlans' 

471 

472 def get_parent(self, request, pk): 

473 return get_object_or_404(VLANGroup.objects.restrict(request.user), pk=pk) 

474 

475 def get_available_objects(self, parent, limit=None): 

476 return parent.get_available_vids()[:limit] 

477 

478 def get_extra_context(self, parent): 

479 return { 

480 'group': parent, 

481 } 

482 

483 def prep_object_data(self, requested_objects, available_objects, parent): 

484 for i, request_data in enumerate(requested_objects): 

485 request_data.update({ 

486 'vid': available_objects.pop(0), 

487 'group': parent.pk, 

488 }) 

489 

490 return requested_objects 

491 

492 @extend_schema(methods=["get"], responses={200: serializers.AvailableVLANSerializer(many=True)}) 

493 def get(self, request, pk): 

494 return super().get(request, pk) 

495 

496 @extend_schema( 

497 methods=["post"], 

498 responses={201: serializers.VLANSerializer(many=True)}, 

499 request=serializers.CreateAvailableVLANSerializer(many=True), 

500 ) 

501 def post(self, request, pk): 

502 return super().post(request, pk)