Coverage for vpn/models/tunnels.py: 79%

1from django.contrib.contenttypes.fields import GenericForeignKey 

2from django.core.exceptions import ValidationError 

3from django.db import models 

4from django.urls import reverse 

5from django.utils.translation import gettext_lazy as _ 

6 

7from netbox.models import ChangeLoggedModel, OrganizationalModel, PrimaryModel 

8from netbox.models.features import ContactsMixin, CustomFieldsMixin, CustomLinksMixin, TagsMixin 

9from vpn.choices import * 

10 

11__all__ = ( 

12 'Tunnel', 

13 'TunnelGroup', 

14 'TunnelTermination', 

15) 

16 

17 

18class TunnelGroup(ContactsMixin, OrganizationalModel): 

19 """ 

20 An administrative grouping of Tunnels. This can be used to correlate peer-to-peer tunnels which form a mesh, 

21 for example. 

22 """ 

23 class Meta: 

24 ordering = ('name',) 

25 verbose_name = _('tunnel group') 

26 verbose_name_plural = _('tunnel groups') 

27 

28 

29class Tunnel(ContactsMixin, PrimaryModel): 

30 name = models.CharField( 

31 verbose_name=_('name'), 

32 max_length=100, 

33 unique=True, 

34 db_collation="natural_sort" 

35 ) 

36 status = models.CharField( 

37 verbose_name=_('status'), 

38 max_length=50, 

39 choices=TunnelStatusChoices, 

40 default=TunnelStatusChoices.STATUS_ACTIVE 

41 ) 

42 group = models.ForeignKey( 

43 to='vpn.TunnelGroup', 

44 on_delete=models.PROTECT, 

45 related_name='tunnels', 

46 blank=True, 

47 null=True 

48 ) 

49 encapsulation = models.CharField( 

50 verbose_name=_('encapsulation'), 

51 max_length=50, 

52 choices=TunnelEncapsulationChoices 

53 ) 

54 ipsec_profile = models.ForeignKey( 

55 to='vpn.IPSecProfile', 

56 on_delete=models.PROTECT, 

57 related_name='tunnels', 

58 blank=True, 

59 null=True 

60 ) 

61 tenant = models.ForeignKey( 

62 to='tenancy.Tenant', 

63 on_delete=models.PROTECT, 

64 related_name='tunnels', 

65 blank=True, 

66 null=True 

67 ) 

68 tunnel_id = models.PositiveBigIntegerField( 

69 verbose_name=_('tunnel ID'), 

70 blank=True, 

71 null=True 

72 ) 

73 

74 clone_fields = ( 

75 'status', 'encapsulation', 'ipsec_profile', 'tenant', 

76 ) 

77 

78 class Meta: 

79 ordering = ('name',) 

80 verbose_name = _('tunnel') 

81 verbose_name_plural = _('tunnels') 

82 

83 def __str__(self): 

84 return self.name 

85 

86 def get_status_color(self): 

87 return TunnelStatusChoices.colors.get(self.status) 

88 

89 

90class TunnelTermination(CustomFieldsMixin, CustomLinksMixin, TagsMixin, ChangeLoggedModel): 

91 tunnel = models.ForeignKey( 

92 to='vpn.Tunnel', 

93 on_delete=models.CASCADE, 

94 related_name='terminations' 

95 ) 

96 role = models.CharField( 

97 verbose_name=_('role'), 

98 max_length=50, 

99 choices=TunnelTerminationRoleChoices, 

100 default=TunnelTerminationRoleChoices.ROLE_PEER 

101 ) 

102 termination_type = models.ForeignKey( 

103 to='contenttypes.ContentType', 

104 on_delete=models.PROTECT, 

105 related_name='+' 

106 ) 

107 termination_id = models.PositiveBigIntegerField( 

108 blank=True, 

109 null=True 

110 ) 

111 termination = GenericForeignKey( 

112 ct_field='termination_type', 

113 fk_field='termination_id' 

114 ) 

115 outside_ip = models.ForeignKey( 

116 to='ipam.IPAddress', 

117 on_delete=models.PROTECT, 

118 related_name='tunnel_terminations', 

119 blank=True, 

120 null=True 

121 ) 

122 

123 prerequisite_models = ( 

124 'vpn.Tunnel', 

125 ) 

126 

127 class Meta: 

128 ordering = ('tunnel', 'role', 'pk') 

129 constraints = ( 

130 models.UniqueConstraint( 

131 fields=('termination_type', 'termination_id'), 

132 name='%(app_label)s_%(class)s_termination', 

133 violation_error_message=_("An object may be terminated to only one tunnel at a time.") 

134 ), 

135 ) 

136 indexes = ( 

137 models.Index(fields=('tunnel', 'role', 'id')), # Default ordering 

138 ) 

139 verbose_name = _('tunnel termination') 

140 verbose_name_plural = _('tunnel terminations') 

141 

142 def __str__(self): 

143 return f'{self.tunnel}: Termination {self.pk}' 

144 

145 def get_absolute_url(self): 

146 return reverse('vpn:tunneltermination', args=[self.pk]) 

147 

148 def get_role_color(self): 

149 return TunnelTerminationRoleChoices.colors.get(self.role) 

150 

151 def clean(self): 

152 super().clean() 

153 

154 # Check that the selected termination object is not already attached to a Tunnel 

155 if getattr(self.termination, 'tunnel_termination', None) and self.termination.tunnel_termination.pk != self.pk: 

156 raise ValidationError({ 

157 'termination': _("{name} is already attached to a tunnel ({tunnel}).").format( 

158 name=self.termination.name, 

159 tunnel=self.termination.tunnel_termination.tunnel 

160 ) 

161 }) 

162 

163 def to_objectchange(self, action): 

164 objectchange = super().to_objectchange(action) 

165 objectchange.related_object = self.tunnel 

166 return objectchange