+#
+# -*- coding: utf-8 -*- vim:fileencoding=utf-8:
+#Copyright © 2011-2013 Greek Research and Technology Network (GRNET S.A.)
+
+#Developed by Leonidas Poulopoulos (leopoul-at-noc-dot-grnet-dot-gr),
+#GRNET NOC
+#
+#Permission to use, copy, modify, and/or distribute this software for any
+#purpose with or without fee is hereby granted, provided that the above
+#copyright notice and this permission notice appear in all copies.
+#
+#THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH REGARD
+#TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND
+#FITNESS. IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR
+#CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE,
+#DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
+#ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS
+#SOFTWARE.
+#
from django import forms
from django.utils.safestring import mark_safe
from django.utils.translation import ugettext as _
from django.utils.translation import ugettext_lazy
from django.template.defaultfilters import filesizeformat
from flowspy.flowspec.models import *
+from flowspy.peers.models import *
+from flowspy.accounts.models import *
from ipaddr import *
from django.core.urlresolvers import reverse
from django.contrib.auth.models import User
import datetime
from django.core.mail import mail_admins, mail_managers, send_mail
+class UserProfileForm(forms.ModelForm):
+ class Meta:
+ model = UserProfile
class RouteForm(forms.ModelForm):
# name = forms.CharField(help_text=ugettext_lazy("A unique route name,"
class Meta:
model = Route
-
+
+ def clean_applier(self):
+ applier = self.cleaned_data['applier']
+ if applier:
+ return self.cleaned_data["applier"]
+ else:
+ raise forms.ValidationError('This field is required.')
+
def clean_source(self):
user = User.objects.get(pk=self.data['applier'])
peer = user.get_profile().peer
data = self.cleaned_data['source']
private_error = False
protected_error = False
+ networkaddr_error = False
+ broadcast_error = False
if data:
try:
address = IPNetwork(data)
send_mail(settings.EMAIL_SUBJECT_PREFIX + "Caught an attempt to set a protected IP/network as a source address",
mail_body, settings.SERVER_EMAIL,
settings.NOTIFY_ADMIN_MAILS, fail_silently=True)
- raise forms.ValidationError("Not allowed")
+ raise Exception
if address.is_private:
private_error = True
- raise forms.ValidationError('Private addresses not allowed')
- else:
- return self.cleaned_data["source"]
+ raise Exception
+ if address.version == 4 and int(address.prefixlen) == 32:
+ if int(address.network.compressed.split('.')[-1]) == 0:
+ broadcast_error = True
+ raise Exception
+ elif int(address.network.compressed.split('.')[-1]) == 255:
+ networkaddr_error = True
+ raise Exception
+ return self.cleaned_data["source"]
except Exception:
- error_text = 'Invalid network address format'
+ error_text = _('Invalid network address format')
if private_error:
- error_text = 'Private addresses not allowed'
+ error_text = _('Private addresses not allowed')
+ if networkaddr_error:
+ error_text = _('Malformed address format. Cannot be ...255/32')
+ if broadcast_error:
+ error_text = _('Malformed address format. Cannot be ...0/32')
if protected_error:
- error_text = 'You have no authority on this subnet'
+ error_text = _('You have no authority on this subnet')
raise forms.ValidationError(error_text)
def clean_destination(self):
user = User.objects.get(pk=self.data['applier'])
- peer = user.get_profile().peer
+ peer = user.get_profile().peer
data = self.cleaned_data['destination']
error = None
protected_error = False
+ networkaddr_error = False
+ broadcast_error = False
if data:
try:
address = IPNetwork(data)
send_mail(settings.EMAIL_SUBJECT_PREFIX + "Caught an attempt to set a protected IP/network as the destination address",
mail_body, settings.SERVER_EMAIL,
settings.NOTIFY_ADMIN_MAILS, fail_silently=True)
- raise forms.ValidationError("Not allowed")
+ raise Exception
if address.prefixlen < settings.PREFIX_LENGTH:
- error = "Currently no prefix lengths < %s are allowed" %settings.PREFIX_LENGTH
- raise forms.ValidationError('error')
+ error = _("Currently no prefix lengths < %s are allowed") %settings.PREFIX_LENGTH
+ raise Exception
+ if address.version == 4 and int(address.prefixlen) == 32:
+ if int(address.network.compressed.split('.')[-1]) == 0:
+ broadcast_error = True
+ raise Exception
+ elif int(address.network.compressed.split('.')[-1]) == 255:
+ networkaddr_error = True
+ raise Exception
return self.cleaned_data["destination"]
except Exception:
- error_text = 'Invalid network address format'
+ error_text = _('Invalid network address format')
if error:
error_text = error
if protected_error:
- error_text = 'You have no authority on this subnet'
+ error_text = _('You have no authority on this subnet')
+ if networkaddr_error:
+ error_text = _('Malformed address format. Cannot be ...255/32')
+ if broadcast_error:
+ error_text = _('Malformed address format. Cannot be ...0/32')
raise forms.ValidationError(error_text)
def clean_expires(self):
raise forms.ValidationError('Invalid date range')
def clean(self):
+ if self.errors:
+ raise forms.ValidationError(_('Errors in form. Please review and fix them'))
name = self.cleaned_data.get('name', None)
source = self.cleaned_data.get('source', None)
sourceports = self.cleaned_data.get('sourceport', None)
ports = self.cleaned_data.get('port', None)
+ fragmenttypes = self.cleaned_data.get('fragmenttype', None)
then = self.cleaned_data.get('then', None)
destination = self.cleaned_data.get('destination', None)
destinationports = self.cleaned_data.get('destinationport', None)
protocols = self.cleaned_data.get('protocol', None)
user = self.cleaned_data.get('applier', None)
+ try:
+ issuperuser = self.data['issuperuser']
+ su = User.objects.get(username=issuperuser)
+ except:
+ issuperuser = None
peer = user.get_profile().peer
networks = peer.networks.all()
+ if issuperuser:
+ networks = PeerRange.objects.filter(peer__in=Peer.objects.all()).distinct()
mynetwork = False
route_pk_list = []
if destination:
if IPNetwork(destination) in net:
mynetwork = True
if not mynetwork:
- raise forms.ValidationError('Destination address/network should belong to your administrative address space. Check My Profile to review your networks')
+ raise forms.ValidationError(_('Destination address/network should belong to your administrative address space. Check My Profile to review your networks'))
if (sourceports and ports):
- raise forms.ValidationError('Cannot create rule for source ports and ports at the same time. Select either ports or source ports')
+ raise forms.ValidationError(_('Cannot create rule for source ports and ports at the same time. Select either ports or source ports'))
if (destinationports and ports):
- raise forms.ValidationError('Cannot create rule for destination ports and ports at the same time. Select either ports or destination ports')
+ raise forms.ValidationError(_('Cannot create rule for destination ports and ports at the same time. Select either ports or destination ports'))
if sourceports and not source:
- raise forms.ValidationError('Once source port is matched, source has to be filled as well. Either deselect source port or fill source address')
+ raise forms.ValidationError(_('Once source port is matched, source has to be filled as well. Either deselect source port or fill source address'))
if destinationports and not destination:
- raise forms.ValidationError('Once destination port is matched, destination has to be filled as well. Either deselect destination port or fill destination address')
+ raise forms.ValidationError(_('Once destination port is matched, destination has to be filled as well. Either deselect destination port or fill destination address'))
if not (source or sourceports or ports or destination or destinationports):
- raise forms.ValidationError('Fill at least a Rule Match Condition')
+ raise forms.ValidationError(_('Fill at least a Rule Match Condition'))
if not user.is_superuser and then[0].action not in settings.UI_USER_THEN_ACTIONS:
- raise forms.ValidationError('This action "%s" is not permitted' %(then[0].action))
- existing_routes = Route.objects.exclude(status='EXPIRED').exclude(status='ERROR').exclude(status='ADMININACTIVE')
+ raise forms.ValidationError(_('This action "%s" is not permitted') %(then[0].action))
+ existing_routes = Route.objects.all()
existing_routes = existing_routes.filter(applier__userprofile__peer=peer)
if source:
source = IPNetwork(source).compressed
try:
assert(int(action_value))
if int(action_value) < 50:
- raise forms.ValidationError('Rate-limiting cannot be < 50kbps')
+ raise forms.ValidationError(_('Rate-limiting cannot be < 50kbps'))
return "%s" %self.cleaned_data["action_value"]
except:
- raise forms.ValidationError('Rate-limiting should be an integer < 50')
+ raise forms.ValidationError(_('Rate-limiting should be an integer < 50'))
else:
- raise forms.ValidationError('Cannot be empty')
+ raise forms.ValidationError(_('Cannot be empty'))
def clean_action(self):
action = self.cleaned_data['action']
if action != 'rate-limit':
- raise forms.ValidationError('Cannot select something other than rate-limit')
+ raise forms.ValidationError(_('Cannot select something other than rate-limit'))
else:
return self.cleaned_data["action"]
port = self.cleaned_data['port']
if port:
try:
- assert(int(port))
+ p = int(port)
+ if int(port) > 65535 or int(port) < 0:
+ raise forms.ValidationError(_(''))
return "%s" %self.cleaned_data["port"]
+ except forms.ValidationError:
+ raise forms.ValidationError(_('Port should be < 65535 and >= 0'))
except:
- raise forms.ValidationError('Port should be an integer')
+ raise forms.ValidationError(_('Port should be an integer'))
else:
- raise forms.ValidationError('Cannot be empty')
+ raise forms.ValidationError(_('Cannot be empty'))
def value_list_to_list(valuelist):
vl = []
projects / flowspy / blobdiff