Coverage for vpn/forms/bulk_import.py: 66%

1from django.core.exceptions import ValidationError 

2from django.utils.translation import gettext_lazy as _ 

3 

4from dcim.models import Device, Interface 

5from ipam.models import VLAN, IPAddress 

6from netbox.forms import NetBoxModelImportForm, OrganizationalModelImportForm, PrimaryModelImportForm 

7from tenancy.models import Tenant 

8from utilities.forms.fields import CSVChoiceField, CSVModelChoiceField, CSVModelMultipleChoiceField 

9from virtualization.models import VirtualMachine, VMInterface 

10from vpn.choices import * 

11from vpn.models import * 

12 

13__all__ = ( 

14 'IKEPolicyImportForm', 

15 'IKEProposalImportForm', 

16 'IPSecPolicyImportForm', 

17 'IPSecProfileImportForm', 

18 'IPSecProposalImportForm', 

19 'L2VPNImportForm', 

20 'L2VPNTerminationImportForm', 

21 'TunnelGroupImportForm', 

22 'TunnelImportForm', 

23 'TunnelTerminationImportForm', 

24) 

25 

26 

27class TunnelGroupImportForm(OrganizationalModelImportForm): 

28 

29 class Meta: 

30 model = TunnelGroup 

31 fields = ('name', 'slug', 'description', 'owner', 'comments', 'tags') 

32 

33 

34class TunnelImportForm(PrimaryModelImportForm): 

35 status = CSVChoiceField( 

36 label=_('Status'), 

37 choices=TunnelStatusChoices, 

38 help_text=_('Operational status') 

39 ) 

40 group = CSVModelChoiceField( 

41 label=_('Tunnel group'), 

42 queryset=TunnelGroup.objects.all(), 

43 required=False, 

44 to_field_name='name' 

45 ) 

46 encapsulation = CSVChoiceField( 

47 label=_('Encapsulation'), 

48 choices=TunnelEncapsulationChoices, 

49 help_text=_('Tunnel encapsulation') 

50 ) 

51 ipsec_profile = CSVModelChoiceField( 

52 label=_('IPSec profile'), 

53 queryset=IPSecProfile.objects.all(), 

54 required=False, 

55 to_field_name='name' 

56 ) 

57 tenant = CSVModelChoiceField( 

58 label=_('Tenant'), 

59 queryset=Tenant.objects.all(), 

60 required=False, 

61 to_field_name='name', 

62 help_text=_('Assigned tenant') 

63 ) 

64 

65 class Meta: 

66 model = Tunnel 

67 fields = ( 

68 'name', 'status', 'group', 'encapsulation', 'ipsec_profile', 'tenant', 'tunnel_id', 'description', 

69 'owner', 'comments', 'tags', 

70 ) 

71 

72 

73class TunnelTerminationImportForm(NetBoxModelImportForm): 

74 tunnel = CSVModelChoiceField( 

75 label=_('Tunnel'), 

76 queryset=Tunnel.objects.all(), 

77 to_field_name='name' 

78 ) 

79 role = CSVChoiceField( 

80 label=_('Role'), 

81 choices=TunnelTerminationRoleChoices, 

82 help_text=_('Operational role') 

83 ) 

84 device = CSVModelChoiceField( 

85 label=_('Device'), 

86 queryset=Device.objects.all(), 

87 required=False, 

88 to_field_name='name', 

89 help_text=_('Parent device of assigned interface') 

90 ) 

91 virtual_machine = CSVModelChoiceField( 

92 label=_('Virtual machine'), 

93 queryset=VirtualMachine.objects.all(), 

94 required=False, 

95 to_field_name='name', 

96 help_text=_('Parent VM of assigned interface') 

97 ) 

98 termination = CSVModelChoiceField( 

99 label=_('Termination'), 

100 queryset=Interface.objects.none(), # Can also refer to VMInterface 

101 required=False, 

102 to_field_name='name', 

103 help_text=_('Device or virtual machine interface') 

104 ) 

105 outside_ip = CSVModelChoiceField( 

106 label=_('Outside IP'), 

107 queryset=IPAddress.objects.all(), 

108 required=False, 

109 to_field_name='address' 

110 ) 

111 

112 class Meta: 

113 model = TunnelTermination 

114 fields = ( 

115 'tunnel', 'role', 'outside_ip', 'tags', 

116 ) 

117 

118 def __init__(self, data=None, *args, **kwargs): 

119 super().__init__(data, *args, **kwargs) 

120 

121 if data: 

122 

123 # Limit termination queryset by assigned device/VM 

124 if data.get('device'): 

125 self.fields['termination'].queryset = Interface.objects.filter( 

126 **{f"device__{self.fields['device'].to_field_name}": data['device']} 

127 ) 

128 elif data.get('virtual_machine'): 

129 self.fields['termination'].queryset = VMInterface.objects.filter( 

130 **{f"virtual_machine__{self.fields['virtual_machine'].to_field_name}": data['virtual_machine']} 

131 ) 

132 

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

134 

135 # Assign termination object 

136 if self.cleaned_data.get('termination'): 

137 self.instance.termination = self.cleaned_data['termination'] 

138 

139 return super().save(*args, **kwargs) 

140 

141 

142class IKEProposalImportForm(PrimaryModelImportForm): 

143 authentication_method = CSVChoiceField( 

144 label=_('Authentication method'), 

145 choices=AuthenticationMethodChoices 

146 ) 

147 encryption_algorithm = CSVChoiceField( 

148 label=_('Encryption algorithm'), 

149 choices=EncryptionAlgorithmChoices 

150 ) 

151 authentication_algorithm = CSVChoiceField( 

152 label=_('Authentication algorithm'), 

153 choices=AuthenticationAlgorithmChoices, 

154 required=False 

155 ) 

156 group = CSVChoiceField( 

157 label=_('Group'), 

158 choices=DHGroupChoices 

159 ) 

160 

161 class Meta: 

162 model = IKEProposal 

163 fields = ( 

164 'name', 'description', 'authentication_method', 'encryption_algorithm', 'authentication_algorithm', 

165 'group', 'sa_lifetime', 'owner', 'comments', 'tags', 

166 ) 

167 

168 

169class IKEPolicyImportForm(PrimaryModelImportForm): 

170 version = CSVChoiceField( 

171 label=_('Version'), 

172 choices=IKEVersionChoices 

173 ) 

174 mode = CSVChoiceField( 

175 label=_('Mode'), 

176 choices=IKEModeChoices, 

177 required=False 

178 ) 

179 proposals = CSVModelMultipleChoiceField( 

180 queryset=IKEProposal.objects.all(), 

181 to_field_name='name', 

182 help_text=_('IKE proposal(s)'), 

183 ) 

184 

185 class Meta: 

186 model = IKEPolicy 

187 fields = ( 

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

189 ) 

190 

191 

192class IPSecProposalImportForm(PrimaryModelImportForm): 

193 encryption_algorithm = CSVChoiceField( 

194 label=_('Encryption algorithm'), 

195 choices=EncryptionAlgorithmChoices, 

196 required=False 

197 ) 

198 authentication_algorithm = CSVChoiceField( 

199 label=_('Authentication algorithm'), 

200 choices=AuthenticationAlgorithmChoices, 

201 required=False 

202 ) 

203 

204 class Meta: 

205 model = IPSecProposal 

206 fields = ( 

207 'name', 'description', 'encryption_algorithm', 'authentication_algorithm', 'sa_lifetime_seconds', 

208 'sa_lifetime_data', 'owner', 'comments', 'tags', 

209 ) 

210 

211 

212class IPSecPolicyImportForm(PrimaryModelImportForm): 

213 pfs_group = CSVChoiceField( 

214 label=_('Diffie-Hellman group for Perfect Forward Secrecy'), 

215 choices=DHGroupChoices, 

216 required=False 

217 ) 

218 proposals = CSVModelMultipleChoiceField( 

219 queryset=IPSecProposal.objects.all(), 

220 to_field_name='name', 

221 help_text=_('IPSec proposal(s)'), 

222 ) 

223 

224 class Meta: 

225 model = IPSecPolicy 

226 fields = ( 

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

228 ) 

229 

230 

231class IPSecProfileImportForm(PrimaryModelImportForm): 

232 mode = CSVChoiceField( 

233 label=_('Mode'), 

234 choices=IPSecModeChoices, 

235 help_text=_('IPSec protocol') 

236 ) 

237 ike_policy = CSVModelChoiceField( 

238 label=_('IKE policy'), 

239 queryset=IKEPolicy.objects.all(), 

240 to_field_name='name' 

241 ) 

242 ipsec_policy = CSVModelChoiceField( 

243 label=_('IPSec policy'), 

244 queryset=IPSecPolicy.objects.all(), 

245 to_field_name='name' 

246 ) 

247 

248 class Meta: 

249 model = IPSecProfile 

250 fields = ( 

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

252 ) 

253 

254 

255class L2VPNImportForm(PrimaryModelImportForm): 

256 tenant = CSVModelChoiceField( 

257 label=_('Tenant'), 

258 queryset=Tenant.objects.all(), 

259 required=False, 

260 to_field_name='name', 

261 ) 

262 status = CSVChoiceField( 

263 label=_('Status'), 

264 choices=L2VPNStatusChoices, 

265 help_text=_('Operational status') 

266 ) 

267 type = CSVChoiceField( 

268 label=_('Type'), 

269 choices=L2VPNTypeChoices, 

270 help_text=_('L2VPN type') 

271 ) 

272 

273 class Meta: 

274 model = L2VPN 

275 fields = ( 

276 'identifier', 'name', 'slug', 'tenant', 'type', 'description', 'owner', 'comments', 'tags', 

277 ) 

278 

279 

280class L2VPNTerminationImportForm(NetBoxModelImportForm): 

281 l2vpn = CSVModelChoiceField( 

282 queryset=L2VPN.objects.all(), 

283 required=True, 

284 to_field_name='name', 

285 label=_('L2VPN'), 

286 ) 

287 device = CSVModelChoiceField( 

288 label=_('Device'), 

289 queryset=Device.objects.all(), 

290 required=False, 

291 to_field_name='name', 

292 help_text=_('Parent device (for interface)') 

293 ) 

294 virtual_machine = CSVModelChoiceField( 

295 label=_('Virtual machine'), 

296 queryset=VirtualMachine.objects.all(), 

297 required=False, 

298 to_field_name='name', 

299 help_text=_('Parent virtual machine (for interface)') 

300 ) 

301 interface = CSVModelChoiceField( 

302 label=_('Interface'), 

303 queryset=Interface.objects.none(), # Can also refer to VMInterface 

304 required=False, 

305 to_field_name='name', 

306 help_text=_('Assigned interface (device or VM)') 

307 ) 

308 vlan = CSVModelChoiceField( 

309 label=_('VLAN'), 

310 queryset=VLAN.objects.all(), 

311 required=False, 

312 to_field_name='name', 

313 help_text=_('Assigned VLAN') 

314 ) 

315 

316 class Meta: 

317 model = L2VPNTermination 

318 fields = ('l2vpn', 'device', 'virtual_machine', 'interface', 'vlan', 'tags') 

319 

320 def __init__(self, data=None, *args, **kwargs): 

321 super().__init__(data, *args, **kwargs) 

322 

323 if data: 

324 

325 # Limit interface queryset by device or VM 

326 if data.get('device'): 

327 self.fields['interface'].queryset = Interface.objects.filter( 

328 **{f"device__{self.fields['device'].to_field_name}": data['device']} 

329 ) 

330 elif data.get('virtual_machine'): 

331 self.fields['interface'].queryset = VMInterface.objects.filter( 

332 **{f"virtual_machine__{self.fields['virtual_machine'].to_field_name}": data['virtual_machine']} 

333 ) 

334 

335 def clean(self): 

336 super().clean() 

337 

338 if self.cleaned_data.get('device') and self.cleaned_data.get('virtual_machine'): 

339 raise ValidationError(_('Cannot import device and VM interface terminations simultaneously.')) 

340 if not self.instance and not (self.cleaned_data.get('interface') or self.cleaned_data.get('vlan')): 

341 raise ValidationError(_('Each termination must specify either an interface or a VLAN.')) 

342 if self.cleaned_data.get('interface') and self.cleaned_data.get('vlan'): 

343 raise ValidationError(_('Cannot assign both an interface and a VLAN.')) 

344 

345 # if this is an update we might not have interface or vlan in the form data 

346 if self.cleaned_data.get('interface') or self.cleaned_data.get('vlan'): 

347 self.instance.assigned_object = self.cleaned_data.get('interface') or self.cleaned_data.get('vlan')