Coverage for vpn/forms/model_forms.py: 54%

1from django import forms 

2from django.core.exceptions import ValidationError 

3from django.utils.translation import gettext_lazy as _ 

4 

5from dcim.models import Device, Interface 

6from ipam.models import VLAN, IPAddress, RouteTarget 

7from netbox.forms import NetBoxModelForm, OrganizationalModelForm, PrimaryModelForm 

8from tenancy.forms import TenancyForm 

9from utilities.forms.fields import ( 

10 ChoiceField, 

11 DynamicModelChoiceField, 

12 DynamicModelMultipleChoiceField, 

13 SlugField, 

14 TypedChoiceField, 

15) 

16from utilities.forms.rendering import FieldSet, TabbedGroups 

17from utilities.forms.utils import add_blank_choice, get_field_value 

18from utilities.forms.widgets import HTMXSelect 

19from virtualization.models import VirtualMachine, VMInterface 

20from vpn.choices import * 

21from vpn.models import * 

22 

23__all__ = ( 

24 'IKEPolicyForm', 

25 'IKEProposalForm', 

26 'IPSecPolicyForm', 

27 'IPSecProfileForm', 

28 'IPSecProposalForm', 

29 'L2VPNForm', 

30 'L2VPNTerminationForm', 

31 'TunnelCreateForm', 

32 'TunnelForm', 

33 'TunnelGroupForm', 

34 'TunnelTerminationForm', 

35) 

36 

37 

38class TunnelGroupForm(OrganizationalModelForm): 

39 fieldsets = ( 

40 FieldSet('name', 'slug', 'description', 'tags', name=_('Tunnel Group')), 

41 ) 

42 

43 class Meta: 

44 model = TunnelGroup 

45 fields = [ 

46 'name', 'slug', 'description', 'owner', 'comments', 'tags', 

47 ] 

48 

49 

50class TunnelForm(TenancyForm, PrimaryModelForm): 

51 status = ChoiceField( 

52 label=_('Status'), 

53 choices=TunnelStatusChoices, 

54 initial=TunnelStatusChoices.STATUS_ACTIVE, 

55 ) 

56 encapsulation = ChoiceField( 

57 label=_('Encapsulation'), 

58 choices=TunnelEncapsulationChoices, 

59 ) 

60 group = DynamicModelChoiceField( 

61 queryset=TunnelGroup.objects.all(), 

62 label=_('Tunnel Group'), 

63 required=False, 

64 quick_add=True 

65 ) 

66 ipsec_profile = DynamicModelChoiceField( 

67 queryset=IPSecProfile.objects.all(), 

68 label=_('IPSec Profile'), 

69 required=False 

70 ) 

71 

72 fieldsets = ( 

73 FieldSet('name', 'status', 'group', 'encapsulation', 'description', 'tunnel_id', 'tags', name=_('Tunnel')), 

74 FieldSet('ipsec_profile', name=_('Security')), 

75 FieldSet('tenant_group', 'tenant', name=_('Tenancy')), 

76 ) 

77 

78 class Meta: 

79 model = Tunnel 

80 fields = [ 

81 'name', 'status', 'group', 'encapsulation', 'description', 'tunnel_id', 'ipsec_profile', 'tenant_group', 

82 'tenant', 'owner', 'comments', 'tags', 

83 ] 

84 

85 

86class TunnelCreateForm(TunnelForm): 

87 # First termination 

88 termination1_role = ChoiceField( 

89 choices=add_blank_choice(TunnelTerminationRoleChoices), 

90 required=False, 

91 label=_('Role') 

92 ) 

93 termination1_type = ChoiceField( 

94 choices=TunnelTerminationTypeChoices, 

95 required=False, 

96 widget=HTMXSelect(hx_target_id='tunnel-termination1'), 

97 label=_('Type') 

98 ) 

99 termination1_parent = DynamicModelChoiceField( 

100 queryset=Device.objects.all(), 

101 required=False, 

102 selector=True, 

103 label=_('Device') 

104 ) 

105 termination1_termination = DynamicModelChoiceField( 

106 queryset=Interface.objects.all(), 

107 required=False, 

108 label=_('Tunnel interface'), 

109 query_params={ 

110 'device_id': '$termination1_parent', 

111 } 

112 ) 

113 termination1_outside_ip = DynamicModelChoiceField( 

114 queryset=IPAddress.objects.all(), 

115 label=_('Outside IP'), 

116 required=False, 

117 query_params={ 

118 'device_id': '$termination1_parent', 

119 } 

120 ) 

121 

122 # Second termination 

123 termination2_role = ChoiceField( 

124 choices=add_blank_choice(TunnelTerminationRoleChoices), 

125 required=False, 

126 label=_('Role') 

127 ) 

128 termination2_type = ChoiceField( 

129 choices=TunnelTerminationTypeChoices, 

130 required=False, 

131 widget=HTMXSelect(hx_target_id='tunnel-termination2'), 

132 label=_('Type') 

133 ) 

134 termination2_parent = DynamicModelChoiceField( 

135 queryset=Device.objects.all(), 

136 required=False, 

137 selector=True, 

138 label=_('Device') 

139 ) 

140 termination2_termination = DynamicModelChoiceField( 

141 queryset=Interface.objects.all(), 

142 required=False, 

143 label=_('Tunnel interface'), 

144 query_params={ 

145 'device_id': '$termination2_parent', 

146 } 

147 ) 

148 termination2_outside_ip = DynamicModelChoiceField( 

149 queryset=IPAddress.objects.all(), 

150 required=False, 

151 label=_('Outside IP'), 

152 query_params={ 

153 'device_id': '$termination2_parent', 

154 } 

155 ) 

156 

157 fieldsets = ( 

158 FieldSet('name', 'status', 'group', 'encapsulation', 'description', 'tunnel_id', 'tags', name=_('Tunnel')), 

159 FieldSet('ipsec_profile', name=_('Security')), 

160 FieldSet('tenant_group', 'tenant', name=_('Tenancy')), 

161 FieldSet( 

162 'termination1_role', 'termination1_type', 'termination1_parent', 'termination1_termination', 

163 'termination1_outside_ip', name=_('First Termination'), html_id='tunnel-termination1'), 

164 FieldSet( 

165 'termination2_role', 'termination2_type', 'termination2_parent', 'termination2_termination', 

166 'termination2_outside_ip', name=_('Second Termination'), html_id='tunnel-termination2'), 

167 ) 

168 

169 def __init__(self, *args, initial=None, **kwargs): 

170 super().__init__(*args, initial=initial, **kwargs) 

171 

172 if get_field_value(self, 'termination1_type') == TunnelTerminationTypeChoices.TYPE_VIRTUALMACHINE: 

173 self.fields['termination1_parent'].label = _('Virtual Machine') 

174 self.fields['termination1_parent'].queryset = VirtualMachine.objects.all() 

175 self.fields['termination1_termination'].queryset = VMInterface.objects.all() 

176 self.fields['termination1_termination'].widget.add_query_params({ 

177 'virtual_machine_id': '$termination1_parent', 

178 }) 

179 self.fields['termination1_outside_ip'].widget.add_query_params({ 

180 'virtual_machine_id': '$termination1_parent', 

181 }) 

182 

183 if get_field_value(self, 'termination2_type') == TunnelTerminationTypeChoices.TYPE_VIRTUALMACHINE: 

184 self.fields['termination2_parent'].label = _('Virtual Machine') 

185 self.fields['termination2_parent'].queryset = VirtualMachine.objects.all() 

186 self.fields['termination2_termination'].queryset = VMInterface.objects.all() 

187 self.fields['termination2_termination'].widget.add_query_params({ 

188 'virtual_machine_id': '$termination2_parent', 

189 }) 

190 self.fields['termination2_outside_ip'].widget.add_query_params({ 

191 'virtual_machine_id': '$termination2_parent', 

192 }) 

193 

194 def clean(self): 

195 super().clean() 

196 

197 # Validate attributes for each termination (if any) 

198 for term in ('termination1', 'termination2'): 

199 required_parameters = ( 

200 f'{term}_role', f'{term}_parent', f'{term}_termination', 

201 ) 

202 parameters = ( 

203 *required_parameters, 

204 f'{term}_outside_ip', 

205 ) 

206 if any([self.cleaned_data[param] for param in parameters]): 

207 for param in required_parameters: 

208 if not self.cleaned_data[param]: 

209 raise forms.ValidationError({ 

210 param: _("This parameter is required when defining a termination.") 

211 }) 

212 

213 def save(self, *args, **kwargs): 

214 instance = super().save(*args, **kwargs) 

215 

216 # Create first termination 

217 if self.cleaned_data['termination1_termination']: 

218 TunnelTermination.objects.create( 

219 tunnel=instance, 

220 role=self.cleaned_data['termination1_role'], 

221 termination=self.cleaned_data['termination1_termination'], 

222 outside_ip=self.cleaned_data['termination1_outside_ip'], 

223 ) 

224 

225 # Create second termination, if defined 

226 if self.cleaned_data['termination2_termination']: 

227 TunnelTermination.objects.create( 

228 tunnel=instance, 

229 role=self.cleaned_data['termination2_role'], 

230 termination=self.cleaned_data['termination2_termination'], 

231 outside_ip=self.cleaned_data.get('termination2_outside_ip'), 

232 ) 

233 

234 return instance 

235 

236 

237class TunnelTerminationForm(NetBoxModelForm): 

238 role = ChoiceField( 

239 label=_('Role'), 

240 choices=TunnelTerminationRoleChoices, 

241 initial=TunnelTerminationRoleChoices.ROLE_PEER, 

242 ) 

243 tunnel = DynamicModelChoiceField( 

244 queryset=Tunnel.objects.all() 

245 ) 

246 type = ChoiceField( 

247 choices=TunnelTerminationTypeChoices, 

248 widget=HTMXSelect(hx_target_id='tunnel-termination'), 

249 label=_('Type') 

250 ) 

251 parent = DynamicModelChoiceField( 

252 queryset=Device.objects.all(), 

253 selector=True, 

254 label=_('Device') 

255 ) 

256 termination = DynamicModelChoiceField( 

257 queryset=Interface.objects.all(), 

258 label=_('Tunnel interface'), 

259 query_params={ 

260 'device_id': '$parent', 

261 } 

262 ) 

263 outside_ip = DynamicModelChoiceField( 

264 queryset=IPAddress.objects.all(), 

265 label=_('Outside IP'), 

266 required=False, 

267 query_params={ 

268 'device_id': '$parent', 

269 } 

270 ) 

271 

272 fieldsets = ( 

273 FieldSet( 

274 'tunnel', 'role', 'type', 'parent', 'termination', 'outside_ip', 'tags', 

275 html_id='tunnel-termination', 

276 ), 

277 ) 

278 

279 class Meta: 

280 model = TunnelTermination 

281 fields = [ 

282 'tunnel', 'role', 'outside_ip', 'tags', 

283 ] 

284 

285 def __init__(self, *args, initial=None, **kwargs): 

286 super().__init__(*args, initial=initial, **kwargs) 

287 

288 if (get_field_value(self, 'type') is None and 

289 self.instance.pk and isinstance(self.instance.termination.parent_object, VirtualMachine)): 

290 self.fields['type'].initial = TunnelTerminationTypeChoices.TYPE_VIRTUALMACHINE 

291 

292 # If initial or self.data is set and the type is a VIRTUALMACHINE type, swap the field querysets. 

293 if get_field_value(self, 'type') == TunnelTerminationTypeChoices.TYPE_VIRTUALMACHINE: 

294 self.fields['parent'].label = _('Virtual Machine') 

295 self.fields['parent'].queryset = VirtualMachine.objects.all() 

296 self.fields['parent'].widget.attrs['selector'] = 'virtualization.virtualmachine' 

297 self.fields['termination'].queryset = VMInterface.objects.all() 

298 self.fields['termination'].widget.add_query_params({ 

299 'virtual_machine_id': '$parent', 

300 }) 

301 self.fields['outside_ip'].widget.add_query_params({ 

302 'virtual_machine_id': '$parent', 

303 }) 

304 

305 if self.instance.pk: 

306 self.fields['parent'].initial = self.instance.termination.parent_object 

307 self.fields['termination'].initial = self.instance.termination 

308 

309 def clean(self): 

310 super().clean() 

311 

312 # Set the terminated object 

313 self.instance.termination = self.cleaned_data.get('termination') 

314 

315 

316class IKEProposalForm(PrimaryModelForm): 

317 authentication_method = ChoiceField( 

318 label=_('Authentication method'), 

319 choices=AuthenticationMethodChoices, 

320 ) 

321 encryption_algorithm = ChoiceField( 

322 label=_('Encryption algorithm'), 

323 choices=EncryptionAlgorithmChoices, 

324 ) 

325 authentication_algorithm = TypedChoiceField( 

326 label=_('Authentication algorithm'), 

327 choices=add_blank_choice(AuthenticationAlgorithmChoices), 

328 required=False, 

329 ) 

330 

331 fieldsets = ( 

332 FieldSet('name', 'description', 'tags', name=_('Proposal')), 

333 FieldSet( 

334 'authentication_method', 'encryption_algorithm', 'authentication_algorithm', 'group', 'sa_lifetime', 

335 name=_('Parameters') 

336 ), 

337 ) 

338 

339 class Meta: 

340 model = IKEProposal 

341 fields = [ 

342 'name', 'description', 'authentication_method', 'encryption_algorithm', 'authentication_algorithm', 'group', 

343 'sa_lifetime', 'owner', 'comments', 'tags', 

344 ] 

345 

346 

347class IKEPolicyForm(PrimaryModelForm): 

348 mode = TypedChoiceField( 

349 label=_('Mode'), 

350 choices=add_blank_choice(IKEModeChoices), 

351 required=False, 

352 ) 

353 proposals = DynamicModelMultipleChoiceField( 

354 queryset=IKEProposal.objects.all(), 

355 label=_('Proposals'), 

356 quick_add=True 

357 ) 

358 

359 fieldsets = ( 

360 FieldSet('name', 'description', 'tags', name=_('Policy')), 

361 FieldSet('version', 'mode', 'proposals', 'preshared_key', name=_('Parameters')), 

362 ) 

363 

364 class Meta: 

365 model = IKEPolicy 

366 fields = [ 

367 'name', 'description', 'version', 'mode', 'proposals', 'preshared_key', 'owner', 'comments', 'tags', 

368 ] 

369 

370 

371class IPSecProposalForm(PrimaryModelForm): 

372 encryption_algorithm = TypedChoiceField( 

373 label=_('Encryption'), 

374 choices=add_blank_choice(EncryptionAlgorithmChoices), 

375 required=False, 

376 ) 

377 authentication_algorithm = TypedChoiceField( 

378 label=_('Authentication'), 

379 choices=add_blank_choice(AuthenticationAlgorithmChoices), 

380 required=False, 

381 ) 

382 

383 fieldsets = ( 

384 FieldSet('name', 'description', 'tags', name=_('Proposal')), 

385 FieldSet( 

386 'encryption_algorithm', 'authentication_algorithm', 'sa_lifetime_seconds', 'sa_lifetime_data', 

387 name=_('Parameters') 

388 ), 

389 ) 

390 

391 class Meta: 

392 model = IPSecProposal 

393 fields = [ 

394 'name', 'description', 'encryption_algorithm', 'authentication_algorithm', 'sa_lifetime_seconds', 

395 'sa_lifetime_data', 'owner', 'comments', 'tags', 

396 ] 

397 

398 

399class IPSecPolicyForm(PrimaryModelForm): 

400 proposals = DynamicModelMultipleChoiceField( 

401 queryset=IPSecProposal.objects.all(), 

402 label=_('Proposals'), 

403 quick_add=True 

404 ) 

405 

406 fieldsets = ( 

407 FieldSet('name', 'description', 'tags', name=_('Policy')), 

408 FieldSet('proposals', 'pfs_group', name=_('Parameters')), 

409 ) 

410 

411 class Meta: 

412 model = IPSecPolicy 

413 fields = [ 

414 'name', 'description', 'proposals', 'pfs_group', 'owner', 'comments', 'tags', 

415 ] 

416 

417 

418class IPSecProfileForm(PrimaryModelForm): 

419 mode = ChoiceField( 

420 label=_('Mode'), 

421 choices=IPSecModeChoices, 

422 ) 

423 ike_policy = DynamicModelChoiceField( 

424 queryset=IKEPolicy.objects.all(), 

425 label=_('IKE policy') 

426 ) 

427 ipsec_policy = DynamicModelChoiceField( 

428 queryset=IPSecPolicy.objects.all(), 

429 label=_('IPSec policy') 

430 ) 

431 

432 fieldsets = ( 

433 FieldSet('name', 'description', 'tags', name=_('Profile')), 

434 FieldSet('mode', 'ike_policy', 'ipsec_policy', name=_('Parameters')), 

435 ) 

436 

437 class Meta: 

438 model = IPSecProfile 

439 fields = [ 

440 'name', 'description', 'mode', 'ike_policy', 'ipsec_policy', 'description', 'owner', 'comments', 'tags', 

441 ] 

442 

443 

444# 

445# L2VPN 

446# 

447 

448class L2VPNForm(TenancyForm, PrimaryModelForm): 

449 type = ChoiceField( 

450 label=_('Type'), 

451 choices=L2VPNTypeChoices, 

452 ) 

453 status = ChoiceField( 

454 label=_('Status'), 

455 choices=L2VPNStatusChoices, 

456 initial=L2VPNStatusChoices.STATUS_ACTIVE, 

457 ) 

458 slug = SlugField() 

459 import_targets = DynamicModelMultipleChoiceField( 

460 label=_('Import targets'), 

461 queryset=RouteTarget.objects.all(), 

462 required=False 

463 ) 

464 export_targets = DynamicModelMultipleChoiceField( 

465 label=_('Export targets'), 

466 queryset=RouteTarget.objects.all(), 

467 required=False 

468 ) 

469 

470 fieldsets = ( 

471 FieldSet('name', 'slug', 'type', 'status', 'identifier', 'description', 'tags', name=_('L2VPN')), 

472 FieldSet('import_targets', 'export_targets', name=_('Route Targets')), 

473 FieldSet('tenant_group', 'tenant', name=_('Tenancy')), 

474 ) 

475 

476 class Meta: 

477 model = L2VPN 

478 fields = ( 

479 'name', 'slug', 'type', 'status', 'identifier', 'import_targets', 'export_targets', 'tenant', 'description', 

480 'owner', 'comments', 'tags' 

481 ) 

482 

483 

484class L2VPNTerminationForm(NetBoxModelForm): 

485 l2vpn = DynamicModelChoiceField( 

486 queryset=L2VPN.objects.all(), 

487 required=True, 

488 query_params={}, 

489 label=_('L2VPN') 

490 ) 

491 vlan = DynamicModelChoiceField( 

492 queryset=VLAN.objects.all(), 

493 required=False, 

494 selector=True, 

495 label=_('VLAN') 

496 ) 

497 interface = DynamicModelChoiceField( 

498 label=_('Interface'), 

499 queryset=Interface.objects.all(), 

500 required=False, 

501 selector=True 

502 ) 

503 vminterface = DynamicModelChoiceField( 

504 queryset=VMInterface.objects.all(), 

505 required=False, 

506 selector=True, 

507 label=_('Interface') 

508 ) 

509 

510 fieldsets = ( 

511 FieldSet( 

512 'l2vpn', 

513 TabbedGroups( 

514 FieldSet('vlan', name=_('VLAN')), 

515 FieldSet('interface', name=_('Device')), 

516 FieldSet('vminterface', name=_('Virtual Machine')), 

517 ), 

518 'tags', 

519 ), 

520 ) 

521 

522 class Meta: 

523 model = L2VPNTermination 

524 fields = ('l2vpn', 'tags') 

525 

526 def __init__(self, *args, **kwargs): 

527 instance = kwargs.get('instance') 

528 initial = kwargs.get('initial', {}).copy() 

529 

530 if instance: 

531 if type(instance.assigned_object) is Interface: 

532 initial['interface'] = instance.assigned_object 

533 elif type(instance.assigned_object) is VLAN: 

534 initial['vlan'] = instance.assigned_object 

535 elif type(instance.assigned_object) is VMInterface: 

536 initial['vminterface'] = instance.assigned_object 

537 kwargs['initial'] = initial 

538 

539 super().__init__(*args, **kwargs) 

540 

541 def clean(self): 

542 super().clean() 

543 

544 interface = self.cleaned_data.get('interface') 

545 vminterface = self.cleaned_data.get('vminterface') 

546 vlan = self.cleaned_data.get('vlan') 

547 

548 if not (interface or vminterface or vlan): 

549 raise ValidationError(_('A termination must specify an interface or VLAN.')) 

550 if len([x for x in (interface, vminterface, vlan) if x]) > 1: 

551 raise ValidationError(_('A termination can only have one terminating object (an interface or VLAN).')) 

552 

553 self.instance.assigned_object = interface or vminterface or vlan