You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
201 lines
5.8 KiB
201 lines
5.8 KiB
from datetime import timedelta as td
|
|
import json
|
|
from urllib.parse import quote, urlencode
|
|
|
|
from django import forms
|
|
from django.forms import URLField
|
|
from django.conf import settings
|
|
from django.core.exceptions import ValidationError
|
|
from django.core.validators import RegexValidator
|
|
from hc.front.validators import (
|
|
CronExpressionValidator,
|
|
TimezoneValidator,
|
|
WebhookValidator,
|
|
)
|
|
import requests
|
|
|
|
|
|
class HeadersField(forms.Field):
|
|
message = """Use "Header-Name: value" pairs, one per line."""
|
|
|
|
def to_python(self, value):
|
|
if not value:
|
|
return {}
|
|
|
|
headers = {}
|
|
for line in value.split("\n"):
|
|
if not line.strip():
|
|
continue
|
|
|
|
if ":" not in line:
|
|
raise ValidationError(self.message)
|
|
|
|
n, v = line.split(":", maxsplit=1)
|
|
n, v = n.strip(), v.strip()
|
|
if not n or not v:
|
|
raise ValidationError(message=self.message)
|
|
|
|
headers[n] = v
|
|
|
|
return headers
|
|
|
|
def validate(self, value):
|
|
super().validate(value)
|
|
for k, v in value.items():
|
|
if len(k) > 1000 or len(v) > 1000:
|
|
raise ValidationError("Value too long")
|
|
|
|
|
|
class NameTagsForm(forms.Form):
|
|
name = forms.CharField(max_length=100, required=False)
|
|
tags = forms.CharField(max_length=500, required=False)
|
|
desc = forms.CharField(required=False)
|
|
|
|
def clean_tags(self):
|
|
result = []
|
|
|
|
for part in self.cleaned_data["tags"].split(" "):
|
|
part = part.strip()
|
|
if part != "":
|
|
result.append(part)
|
|
|
|
return " ".join(result)
|
|
|
|
|
|
class FilteringRulesForm(forms.Form):
|
|
subject = forms.CharField(max_length=100)
|
|
methods = forms.ChoiceField(required=False, choices=(("", "Any"), ("POST", "POST")))
|
|
|
|
|
|
class TimeoutForm(forms.Form):
|
|
timeout = forms.IntegerField(min_value=60, max_value=2592000)
|
|
grace = forms.IntegerField(min_value=60, max_value=2592000)
|
|
|
|
def clean_timeout(self):
|
|
return td(seconds=self.cleaned_data["timeout"])
|
|
|
|
def clean_grace(self):
|
|
return td(seconds=self.cleaned_data["grace"])
|
|
|
|
|
|
class CronForm(forms.Form):
|
|
schedule = forms.CharField(max_length=100, validators=[CronExpressionValidator()])
|
|
tz = forms.CharField(max_length=36, validators=[TimezoneValidator()])
|
|
grace = forms.IntegerField(min_value=1, max_value=43200)
|
|
|
|
|
|
class AddOpsGenieForm(forms.Form):
|
|
error_css_class = "has-error"
|
|
region = forms.ChoiceField(initial="us", choices=(("us", "US"), ("eu", "EU")))
|
|
key = forms.CharField(max_length=40)
|
|
|
|
|
|
class AddEmailForm(forms.Form):
|
|
error_css_class = "has-error"
|
|
value = forms.EmailField(max_length=100)
|
|
down = forms.BooleanField(required=False, initial=True)
|
|
up = forms.BooleanField(required=False, initial=True)
|
|
|
|
def clean(self):
|
|
super().clean()
|
|
|
|
down = self.cleaned_data.get("down")
|
|
up = self.cleaned_data.get("up")
|
|
|
|
if not down and not up:
|
|
self.add_error("down", "Please select at least one.")
|
|
|
|
|
|
class AddUrlForm(forms.Form):
|
|
error_css_class = "has-error"
|
|
value = forms.URLField(max_length=1000, validators=[WebhookValidator()])
|
|
|
|
|
|
METHODS = ("GET", "POST", "PUT")
|
|
|
|
|
|
class AddWebhookForm(forms.Form):
|
|
error_css_class = "has-error"
|
|
|
|
method_down = forms.ChoiceField(initial="GET", choices=zip(METHODS, METHODS))
|
|
body_down = forms.CharField(max_length=1000, required=False)
|
|
headers_down = HeadersField(required=False)
|
|
url_down = URLField(
|
|
max_length=1000, required=False, validators=[WebhookValidator()]
|
|
)
|
|
|
|
method_up = forms.ChoiceField(initial="GET", choices=zip(METHODS, METHODS))
|
|
body_up = forms.CharField(max_length=1000, required=False)
|
|
headers_up = HeadersField(required=False)
|
|
url_up = forms.URLField(
|
|
max_length=1000, required=False, validators=[WebhookValidator()]
|
|
)
|
|
|
|
def clean(self):
|
|
super().clean()
|
|
|
|
url_down = self.cleaned_data.get("url_down")
|
|
url_up = self.cleaned_data.get("url_up")
|
|
|
|
if not url_down and not url_up:
|
|
self.add_error("url_down", "Enter a valid URL.")
|
|
|
|
def get_value(self):
|
|
return json.dumps(dict(self.cleaned_data), sort_keys=True)
|
|
|
|
|
|
class AddShellForm(forms.Form):
|
|
error_css_class = "has-error"
|
|
|
|
cmd_down = forms.CharField(max_length=1000, required=False)
|
|
cmd_up = forms.CharField(max_length=1000, required=False)
|
|
|
|
def get_value(self):
|
|
return json.dumps(dict(self.cleaned_data), sort_keys=True)
|
|
|
|
|
|
phone_validator = RegexValidator(
|
|
regex="^\+\d{5,15}$", message="Invalid phone number format."
|
|
)
|
|
|
|
|
|
class AddSmsForm(forms.Form):
|
|
error_css_class = "has-error"
|
|
label = forms.CharField(max_length=100, required=False)
|
|
value = forms.CharField(max_length=16, validators=[phone_validator])
|
|
down = forms.BooleanField(required=False, initial=True)
|
|
up = forms.BooleanField(required=False, initial=True)
|
|
|
|
|
|
class ChannelNameForm(forms.Form):
|
|
name = forms.CharField(max_length=100, required=False)
|
|
|
|
|
|
class AddMatrixForm(forms.Form):
|
|
error_css_class = "has-error"
|
|
alias = forms.CharField(max_length=40)
|
|
|
|
def clean_alias(self):
|
|
v = self.cleaned_data["alias"]
|
|
|
|
# validate it by trying to join
|
|
url = settings.MATRIX_HOMESERVER
|
|
url += "/_matrix/client/r0/join/%s?" % quote(v)
|
|
url += urlencode({"access_token": settings.MATRIX_ACCESS_TOKEN})
|
|
doc = requests.post(url, {}).json()
|
|
if "error" in doc:
|
|
raise forms.ValidationError("Response from Matrix: %s" % doc["error"])
|
|
|
|
self.cleaned_data["room_id"] = doc["room_id"]
|
|
|
|
return v
|
|
|
|
|
|
class AddAppriseForm(forms.Form):
|
|
error_css_class = "has-error"
|
|
url = forms.CharField(max_length=512)
|
|
|
|
|
|
class AddPdForm(forms.Form):
|
|
error_css_class = "has-error"
|
|
value = forms.CharField(max_length=32)
|