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

   1 #
   2 # -*- coding: utf-8 -*- vim:fileencoding=utf-8:
   3 #Copyright © 2011-2013 Greek Research and Technology Network (GRNET S.A.)
   4
   5 #Developed by Leonidas Poulopoulos (leopoul-at-noc-dot-grnet-dot-gr),
   6 #GRNET NOC
   7 #
   8 #Permission to use, copy, modify, and/or distribute this software for any
   9 #purpose with or without fee is hereby granted, provided that the above
  10 #copyright notice and this permission notice appear in all copies.
  11 #
  12 #THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH REGARD
  13 #TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND
  14 #FITNESS. IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR
  15 #CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE,
  16 #DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  17 #ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS
  18 #SOFTWARE.
  19 #
  20 from django import forms
  21 from django.utils.safestring import mark_safe
  22 from django.utils.translation import ugettext as _
  23 from django.utils.translation import ugettext_lazy
  24 from django.template.defaultfilters import filesizeformat
  25 from flowspy.flowspec.models import *
  26 from flowspy.peers.models import *
  27 from flowspy.accounts.models import *
  28 from ipaddr import *
  29 from django.core.urlresolvers import reverse
  30 from django.contrib.auth.models import User
  31 from django.conf import settings
  32 import datetime
  33 from django.core.mail import mail_admins, mail_managers, send_mail
  34
  35 class UserProfileForm(forms.ModelForm):
  36     class Meta:
  37         model = UserProfile
  38
  39 class RouteForm(forms.ModelForm):
  40 #    name = forms.CharField(help_text=ugettext_lazy("A unique route name,"
  41 #                                         " e.g. uoa_block_p80"), label=ugettext_lazy("Route Name"), required=False)
  42 #    source = forms.CharField(help_text=ugettext_lazy("A qualified IP Network address. CIDR notation,"
  43 #                                         " e.g.10.10.0.1/32"), label=ugettext_lazy("Source Address"), required=False)
  44 #    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)
  45 #    destination = forms.CharField(help_text=ugettext_lazy("A qualified IP Network address. CIDR notation,"
  46 #                                         " e.g.10.10.0.1/32"), label=ugettext_lazy("Destination Address"), required=False)
  47 #    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)
  48 #    ports = forms.ModelMultipleChoiceField(queryset=MatchPort.objects.all(), help_text=ugettext_lazy("A set of ports to block"), label=ugettext_lazy("Ports"), required=False)
  49
  50     class Meta:
  51         model = Route
  52
  53     def clean_applier(self):
  54         applier = self.cleaned_data['applier']
  55         if applier:
  56             return self.cleaned_data["applier"]
  57         else:
  58             raise forms.ValidationError('This field is required.')
  59
  60     def clean_source(self):
  61         user = User.objects.get(pk=self.data['applier'])
  62         peer = user.get_profile().peer
  63         data = self.cleaned_data['source']
  64         private_error = False
  65         protected_error = False
  66         if data:
  67             try:
  68                 address = IPNetwork(data)
  69                 for net in settings.PROTECTED_SUBNETS:
  70                     if address in IPNetwork(net):
  71                         protected_error = True
  72                         mail_body = "User %s %s (%s) attempted to set %s as the source address in a firewall rule" %(user.username, user.email, peer.peer_name, data)
  73                         send_mail(settings.EMAIL_SUBJECT_PREFIX + "Caught an attempt to set a protected IP/network as a source address",
  74                               mail_body, settings.SERVER_EMAIL,
  75                               settings.NOTIFY_ADMIN_MAILS, fail_silently=True)
  76                         raise Exception
  77                 if address.is_private:
  78                     private_error = True
  79                     raise Exception
  80                 else:
  81                     return self.cleaned_data["source"]
  82             except Exception:
  83                 error_text = _('Invalid network address format')
  84                 if private_error:
  85                     error_text = _('Private addresses not allowed')
  86                 if protected_error:
  87                     error_text = _('You have no authority on this subnet')
  88                 raise forms.ValidationError(error_text)
  89
  90     def clean_destination(self):
  91         user = User.objects.get(pk=self.data['applier'])
  92         peer = user.get_profile().peer
  93         data = self.cleaned_data['destination']
  94         error = None
  95         protected_error = False
  96         if data:
  97             try:
  98                 address = IPNetwork(data)
  99                 for net in settings.PROTECTED_SUBNETS:
 100                     if address in IPNetwork(net):
 101                         protected_error = True
 102                         mail_body = "User %s %s (%s) attempted to set %s as the destination address in a firewall rule" %(user.username, user.email, peer.peer_name, data)
 103                         send_mail(settings.EMAIL_SUBJECT_PREFIX + "Caught an attempt to set a protected IP/network as the destination address",
 104                               mail_body, settings.SERVER_EMAIL,
 105                               settings.NOTIFY_ADMIN_MAILS, fail_silently=True)
 106                         raise Exception
 107                 if address.prefixlen < settings.PREFIX_LENGTH:
 108                     error = _("Currently no prefix lengths < %s are allowed") %settings.PREFIX_LENGTH
 109                     raise Exception
 110                 return self.cleaned_data["destination"]
 111             except Exception:
 112                 error_text = _('Invalid network address format')
 113                 if error:
 114                     error_text = error
 115                 if protected_error:
 116                     error_text = _('You have no authority on this subnet')
 117                 raise forms.ValidationError(error_text)
 118
 119     def clean_expires(self):
 120         date = self.cleaned_data['expires']
 121         if date:
 122             range_days = (date - datetime.date.today()).days
 123             if range_days > 0 and range_days < 11:
 124                 return self.cleaned_data["expires"]
 125             else:
 126                 raise forms.ValidationError('Invalid date range')
 127
 128     def clean(self):
 129         if self.errors:
 130              raise forms.ValidationError(_('Errors in form. Please review and fix them'))
 131         name = self.cleaned_data.get('name', None)
 132         source = self.cleaned_data.get('source', None)
 133         sourceports = self.cleaned_data.get('sourceport', None)
 134         ports = self.cleaned_data.get('port', None)
 135         then = self.cleaned_data.get('then', None)
 136         destination = self.cleaned_data.get('destination', None)
 137         destinationports = self.cleaned_data.get('destinationport', None)
 138         protocols = self.cleaned_data.get('protocol', None)
 139         user = self.cleaned_data.get('applier', None)
 140         try:
 141             issuperuser = self.data['issuperuser']
 142             su = User.objects.get(username=issuperuser)
 143         except:
 144             issuperuser = None
 145         peer = user.get_profile().peer
 146         networks = peer.networks.all()
 147         if issuperuser:
 148             networks = PeerRange.objects.filter(peer__in=Peer.objects.all()).distinct()
 149         mynetwork = False
 150         route_pk_list = []
 151         if destination:
 152             for network in networks:
 153                 net = IPNetwork(network.network)
 154                 if IPNetwork(destination) in net:
 155                     mynetwork = True
 156             if not mynetwork:
 157                 raise forms.ValidationError(_('Destination address/network should belong to your administrative address space. Check My Profile to review your networks'))
 158         if (sourceports and ports):
 159             raise forms.ValidationError(_('Cannot create rule for source ports and ports at the same time. Select either ports or source ports'))
 160         if (destinationports and ports):
 161             raise forms.ValidationError(_('Cannot create rule for destination ports and ports at the same time. Select either ports or destination ports'))
 162         if sourceports and not source:
 163             raise forms.ValidationError(_('Once source port is matched, source has to be filled as well. Either deselect source port or fill source address'))
 164         if destinationports and not destination:
 165             raise forms.ValidationError(_('Once destination port is matched, destination has to be filled as well. Either deselect destination port or fill destination address'))
 166         if not (source or sourceports or ports or destination or destinationports):
 167             raise forms.ValidationError(_('Fill at least a Rule Match Condition'))
 168         if not user.is_superuser and then[0].action not in settings.UI_USER_THEN_ACTIONS:
 169             raise forms.ValidationError(_('This action "%s" is not permitted') %(then[0].action))
 170         existing_routes = Route.objects.all()
 171         existing_routes = existing_routes.filter(applier__userprofile__peer=peer)
 172         if source:
 173             source = IPNetwork(source).compressed
 174             existing_routes = existing_routes.filter(source=source)
 175         else:
 176             existing_routes = existing_routes.filter(source=None)
 177         if protocols:
 178             route_pk_list=get_matchingprotocol_route_pks(protocols, existing_routes)
 179             if route_pk_list:
 180                 existing_routes = existing_routes.filter(pk__in=route_pk_list)
 181             else:
 182                 existing_routes = existing_routes.filter(protocol=None)
 183         else:
 184             existing_routes = existing_routes.filter(protocol=None)
 185         if sourceports:
 186             route_pk_list=get_matchingport_route_pks(sourceports, existing_routes)
 187             if route_pk_list:
 188                 existing_routes = existing_routes.filter(pk__in=route_pk_list)
 189         else:
 190             existing_routes = existing_routes.filter(sourceport=None)
 191         if destinationports:
 192             route_pk_list=get_matchingport_route_pks(destinationports, existing_routes)
 193             if route_pk_list:
 194                 existing_routes = existing_routes.filter(pk__in=route_pk_list)
 195         else:
 196             existing_routes = existing_routes.filter(destinationport=None)
 197         if ports:
 198             route_pk_list=get_matchingport_route_pks(ports, existing_routes)
 199             if route_pk_list:
 200                 existing_routes = existing_routes.filter(pk__in=route_pk_list)
 201         else:
 202             existing_routes = existing_routes.filter(port=None)
 203         for route in existing_routes:
 204             if name != route.name:
 205                 existing_url = reverse('edit-route', args=[route.name])
 206                 if IPNetwork(destination) in IPNetwork(route.destination) or IPNetwork(route.destination) in IPNetwork(destination):
 207                     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))
 208         return self.cleaned_data
 209
 210 class ThenPlainForm(forms.ModelForm):
 211 #    action = forms.CharField(initial='rate-limit')
 212     class Meta:
 213         model = ThenAction
 214
 215     def clean_action_value(self):
 216         action_value = self.cleaned_data['action_value']
 217         if action_value:
 218             try:
 219                 assert(int(action_value))
 220                 if int(action_value) < 50:
 221                     raise forms.ValidationError(_('Rate-limiting cannot be < 50kbps'))
 222                 return "%s" %self.cleaned_data["action_value"]
 223             except:
 224                 raise forms.ValidationError(_('Rate-limiting should be an integer < 50'))
 225         else:
 226             raise forms.ValidationError(_('Cannot be empty'))
 227
 228     def clean_action(self):
 229         action = self.cleaned_data['action']
 230         if action != 'rate-limit':
 231             raise forms.ValidationError(_('Cannot select something other than rate-limit'))
 232         else:
 233             return self.cleaned_data["action"]
 234
 235
 236 class PortPlainForm(forms.ModelForm):
 237 #    action = forms.CharField(initial='rate-limit')
 238     class Meta:
 239         model = MatchPort
 240
 241     def clean_port(self):
 242         port = self.cleaned_data['port']
 243         if port:
 244             try:
 245                 assert(int(port))
 246                 return "%s" %self.cleaned_data["port"]
 247             except:
 248                 raise forms.ValidationError(_('Port should be an integer'))
 249         else:
 250             raise forms.ValidationError(_('Cannot be empty'))
 251
 252 def value_list_to_list(valuelist):
 253     vl = []
 254     for val in valuelist:
 255         vl.append(val[0])
 256     return vl
 257
 258 def get_matchingport_route_pks(portlist, routes):
 259     route_pk_list = []
 260     ports_value_list = value_list_to_list(portlist.values_list('port').order_by('port'))
 261     for route in routes:
 262         rsp = value_list_to_list(route.destinationport.all().values_list('port').order_by('port'))
 263         if rsp and rsp == ports_value_list:
 264             route_pk_list.append(route.pk)
 265     return route_pk_list
 266
 267 def get_matchingprotocol_route_pks(protocolist, routes):
 268     route_pk_list = []
 269     protocols_value_list = value_list_to_list(protocolist.values_list('protocol').order_by('protocol'))
 270     for route in routes:
 271         rsp = value_list_to_list(route.protocol.all().values_list('protocol').order_by('protocol'))
 272         if rsp and rsp == protocols_value_list:
 273             route_pk_list.append(route.pk)
 274     return route_pk_list