code.grnet.gr Git - flowspy/blob - flowspec/forms.py

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