root / flowspec / forms.py @ 6d48c46c
History | View | Annotate | Download (9.2 kB)
1 |
from django import forms |
---|---|
2 |
from django.utils.safestring import mark_safe |
3 |
from django.utils.translation import ugettext as _ |
4 |
from django.utils.translation import ugettext_lazy |
5 |
from django.template.defaultfilters import filesizeformat |
6 |
from flowspy.flowspec.models import * |
7 |
from ipaddr import * |
8 |
from django.core.urlresolvers import reverse |
9 |
from django.contrib.auth.models import User |
10 |
from django.conf import settings |
11 |
import datetime |
12 |
|
13 |
|
14 |
|
15 |
class RouteForm(forms.ModelForm): |
16 |
# name = forms.CharField(help_text=ugettext_lazy("A unique route name,"
|
17 |
# " e.g. uoa_block_p80"), label=ugettext_lazy("Route Name"), required=False)
|
18 |
# source = forms.CharField(help_text=ugettext_lazy("A qualified IP Network address. CIDR notation,"
|
19 |
# " e.g.10.10.0.1/32"), label=ugettext_lazy("Source Address"), required=False)
|
20 |
# source_ports = forms.ModelMultipleChoiceField(queryset=MatchPort.objects.all(), help_text=ugettext_lazy("A set of source ports to block"), label=ugettext_lazy("Source Ports"), required=False)
|
21 |
# destination = forms.CharField(help_text=ugettext_lazy("A qualified IP Network address. CIDR notation,"
|
22 |
# " e.g.10.10.0.1/32"), label=ugettext_lazy("Destination Address"), required=False)
|
23 |
# destination_ports = forms.ModelMultipleChoiceField(queryset=MatchPort.objects.all(), help_text=ugettext_lazy("A set of destination ports to block"), label=ugettext_lazy("Destination Ports"), required=False)
|
24 |
# ports = forms.ModelMultipleChoiceField(queryset=MatchPort.objects.all(), help_text=ugettext_lazy("A set of ports to block"), label=ugettext_lazy("Ports"), required=False)
|
25 |
|
26 |
class Meta: |
27 |
model = Route |
28 |
|
29 |
def clean_source(self): |
30 |
data = self.cleaned_data['source'] |
31 |
private_error = False
|
32 |
if data:
|
33 |
try:
|
34 |
address = IPNetwork(data) |
35 |
if address.is_private:
|
36 |
private_error = True
|
37 |
raise forms.ValidationError('Private addresses not allowed') |
38 |
else:
|
39 |
return self.cleaned_data["source"] |
40 |
except Exception: |
41 |
error_text = 'Invalid network address format'
|
42 |
if private_error:
|
43 |
error_text = 'Private addresses not allowed'
|
44 |
raise forms.ValidationError(error_text)
|
45 |
|
46 |
def clean_destination(self): |
47 |
data = self.cleaned_data['destination'] |
48 |
error = None
|
49 |
if data:
|
50 |
try:
|
51 |
address = IPNetwork(data) |
52 |
if address.prefixlen < settings.PREFIX_LENGTH:
|
53 |
error = "Currently no prefix lengths < %s are allowed" %settings.PREFIX_LENGTH
|
54 |
raise forms.ValidationError('error') |
55 |
return self.cleaned_data["destination"] |
56 |
except Exception: |
57 |
if error:
|
58 |
error_text = error |
59 |
else:
|
60 |
error_text = 'Invalid network address format'
|
61 |
raise forms.ValidationError(error_text)
|
62 |
|
63 |
def clean_expires(self): |
64 |
date = self.cleaned_data['expires'] |
65 |
if date:
|
66 |
range_days = (date - datetime.date.today()).days |
67 |
if range_days > 0 and range_days < 11: |
68 |
return self.cleaned_data["expires"] |
69 |
else:
|
70 |
raise forms.ValidationError('Invalid date range') |
71 |
|
72 |
def clean(self): |
73 |
name = self.cleaned_data.get('name', None) |
74 |
source = self.cleaned_data.get('source', None) |
75 |
sourceports = self.cleaned_data.get('sourceport', None) |
76 |
ports = self.cleaned_data.get('port', None) |
77 |
then = self.cleaned_data.get('then', None) |
78 |
destination = self.cleaned_data.get('destination', None) |
79 |
destinationports = self.cleaned_data.get('destinationport', None) |
80 |
user = self.cleaned_data.get('applier', None) |
81 |
peer = user.get_profile().peer |
82 |
networks = peer.networks.all() |
83 |
mynetwork = False
|
84 |
route_pk_list = [] |
85 |
|
86 |
if destination:
|
87 |
for network in networks: |
88 |
net = IPNetwork(network.network) |
89 |
if IPNetwork(destination) in net: |
90 |
mynetwork = True
|
91 |
if not mynetwork: |
92 |
raise forms.ValidationError('Destination address/network should belong to your administrative address space. Check My Profile to review your networks') |
93 |
if (sourceports and ports): |
94 |
raise forms.ValidationError('Cannot create rule for source ports and ports at the same time. Select either ports or source ports') |
95 |
if (destinationports and ports): |
96 |
raise forms.ValidationError('Cannot create rule for destination ports and ports at the same time. Select either ports or destination ports') |
97 |
if sourceports and not source: |
98 |
raise forms.ValidationError('Once source port is matched, source has to be filled as well. Either deselect source port or fill source address') |
99 |
if destinationports and not destination: |
100 |
raise forms.ValidationError('Once destination port is matched, destination has to be filled as well. Either deselect destination port or fill destination address') |
101 |
if not (source or sourceports or ports or destination or destinationports): |
102 |
raise forms.ValidationError('Fill at least a Route Match Condition') |
103 |
if not user.is_superuser and then[0].action not in settings.UI_USER_THEN_ACTIONS: |
104 |
raise forms.ValidationError('This action "%s" is not permitted' %(then[0].action)) |
105 |
existing_routes = Route.objects.exclude(status='EXPIRED').exclude(status='PENDING').exclude(status='ERROR').exclude(status='ADMININACTIVE') |
106 |
existing_routes = existing_routes.filter(applier__userprofile__peer=peer) |
107 |
if source:
|
108 |
source = IPNetwork(source).compressed |
109 |
existing_routes = existing_routes.filter(source=source) |
110 |
else:
|
111 |
existing_routes = existing_routes.filter(source=None)
|
112 |
if sourceports:
|
113 |
route_pk_list=get_matchingport_route_pks(sourceports, existing_routes) |
114 |
if route_pk_list:
|
115 |
existing_routes = existing_routes.filter(pk__in=route_pk_list) |
116 |
else:
|
117 |
existing_routes = existing_routes.filter(sourceport=None)
|
118 |
if destinationports:
|
119 |
route_pk_list=get_matchingport_route_pks(destinationports, existing_routes) |
120 |
if route_pk_list:
|
121 |
existing_routes = existing_routes.filter(pk__in=route_pk_list) |
122 |
else:
|
123 |
existing_routes = existing_routes.filter(destinationport=None)
|
124 |
if ports:
|
125 |
route_pk_list=get_matchingport_route_pks(ports, existing_routes) |
126 |
if route_pk_list:
|
127 |
existing_routes = existing_routes.filter(pk__in=route_pk_list) |
128 |
else:
|
129 |
existing_routes = existing_routes.filter(port=None)
|
130 |
|
131 |
for route in existing_routes: |
132 |
if name != route.name:
|
133 |
existing_url = reverse('edit-route', args=[route.name])
|
134 |
if IPNetwork(destination) in IPNetwork(route.destination) or IPNetwork(route.destination) in IPNetwork(destination): |
135 |
raise forms.ValidationError('Found an exact %s rule, %s with destination prefix %s<br>To avoid overlapping try editing rule <a href=\'%s\'>%s</a>' %(route.status, route.name, route.destination, existing_url, route.name)) |
136 |
return self.cleaned_data |
137 |
|
138 |
class ThenPlainForm(forms.ModelForm): |
139 |
# action = forms.CharField(initial='rate-limit')
|
140 |
class Meta: |
141 |
model = ThenAction |
142 |
|
143 |
def clean_action_value(self): |
144 |
action_value = self.cleaned_data['action_value'] |
145 |
if action_value:
|
146 |
try:
|
147 |
assert(int(action_value)) |
148 |
return "%s" %self.cleaned_data["action_value"] |
149 |
except:
|
150 |
raise forms.ValidationError('Rate-limiting should be an integer') |
151 |
if int(action_value) < 50: |
152 |
raise forms.ValidationError('Rate-limiting cannot be < 50kbps') |
153 |
else:
|
154 |
raise forms.ValidationError('Cannot be empty') |
155 |
|
156 |
def clean_action(self): |
157 |
action = self.cleaned_data['action'] |
158 |
if action != 'rate-limit': |
159 |
raise forms.ValidationError('Cannot select something other than rate-limit') |
160 |
else:
|
161 |
return self.cleaned_data["action"] |
162 |
|
163 |
class PortPlainForm(forms.ModelForm): |
164 |
# action = forms.CharField(initial='rate-limit')
|
165 |
class Meta: |
166 |
model = MatchPort |
167 |
|
168 |
def clean_port(self): |
169 |
port = self.cleaned_data['port'] |
170 |
if port:
|
171 |
try:
|
172 |
assert(int(port)) |
173 |
return "%s" %self.cleaned_data["port"] |
174 |
except:
|
175 |
raise forms.ValidationError('Port should be an integer') |
176 |
else:
|
177 |
raise forms.ValidationError('Cannot be empty') |
178 |
|
179 |
def value_list_to_list(valuelist): |
180 |
vl = [] |
181 |
for val in valuelist: |
182 |
vl.append(val[0])
|
183 |
return vl
|
184 |
|
185 |
def get_matchingport_route_pks(portlist, routes): |
186 |
route_pk_list = [] |
187 |
ports_value_list = value_list_to_list(portlist.values_list('port').order_by('port')) |
188 |
for route in routes: |
189 |
rsp = value_list_to_list(route.destinationport.all().values_list('port').order_by('port')) |
190 |
if rsp and rsp == ports_value_list: |
191 |
route_pk_list.append(route.pk) |
192 |
return route_pk_list
|