|
|
# -*- coding: utf-8 -*-
|
|
|
|
|
|
# Copyright 2010 - 2017 RhodeCode GmbH and the AppEnlight project authors
|
|
|
#
|
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
# you may not use this file except in compliance with the License.
|
|
|
# You may obtain a copy of the License at
|
|
|
#
|
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
|
#
|
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
# See the License for the specific language governing permissions and
|
|
|
# limitations under the License.
|
|
|
|
|
|
import wtforms
|
|
|
import formencode
|
|
|
import re
|
|
|
import pyramid.threadlocal
|
|
|
import datetime
|
|
|
import appenlight.lib.helpers as h
|
|
|
|
|
|
from ziggurat_foundations.models.services.user import UserService
|
|
|
from ziggurat_foundations.models.services.group import GroupService
|
|
|
from appenlight.models import DBSession
|
|
|
from appenlight.models.alert_channel import AlertChannel
|
|
|
from appenlight.models.integrations import IntegrationException
|
|
|
from appenlight.models.integrations.campfire import CampfireIntegration
|
|
|
from appenlight.models.integrations.bitbucket import BitbucketIntegration
|
|
|
from appenlight.models.integrations.github import GithubIntegration
|
|
|
from appenlight.models.integrations.flowdock import FlowdockIntegration
|
|
|
from appenlight.models.integrations.hipchat import HipchatIntegration
|
|
|
from appenlight.models.integrations.jira import JiraClient
|
|
|
from appenlight.models.integrations.slack import SlackIntegration
|
|
|
from appenlight.lib.ext_json import json
|
|
|
from wtforms.ext.csrf.form import SecureForm
|
|
|
from wtforms.compat import iteritems
|
|
|
from collections import defaultdict
|
|
|
|
|
|
_ = str
|
|
|
|
|
|
strip_filter = lambda x: x.strip() if x else None
|
|
|
uppercase_filter = lambda x: x.upper() if x else None
|
|
|
|
|
|
FALSE_VALUES = ("false", "", False, None)
|
|
|
|
|
|
|
|
|
class CSRFException(Exception):
|
|
|
pass
|
|
|
|
|
|
|
|
|
class ReactorForm(SecureForm):
|
|
|
def __init__(self, formdata=None, obj=None, prefix="", csrf_context=None, **kwargs):
|
|
|
super(ReactorForm, self).__init__(
|
|
|
formdata=formdata,
|
|
|
obj=obj,
|
|
|
prefix=prefix,
|
|
|
csrf_context=csrf_context,
|
|
|
**kwargs
|
|
|
)
|
|
|
self._csrf_context = csrf_context
|
|
|
|
|
|
def generate_csrf_token(self, csrf_context):
|
|
|
return csrf_context.session.get_csrf_token()
|
|
|
|
|
|
def validate_csrf_token(self, field):
|
|
|
request = self._csrf_context or pyramid.threadlocal.get_current_request()
|
|
|
is_from_auth_token = "auth:auth_token" in request.effective_principals
|
|
|
if is_from_auth_token:
|
|
|
return True
|
|
|
|
|
|
if field.data != field.current_token:
|
|
|
# try to save the day by using token from angular
|
|
|
if request.headers.get("X-XSRF-TOKEN") != field.current_token:
|
|
|
raise CSRFException("Invalid CSRF token")
|
|
|
|
|
|
@property
|
|
|
def errors_dict(self):
|
|
|
r_dict = defaultdict(list)
|
|
|
for k, errors in self.errors.items():
|
|
|
r_dict[k].extend([str(e) for e in errors])
|
|
|
return r_dict
|
|
|
|
|
|
@property
|
|
|
def errors_json(self):
|
|
|
return json.dumps(self.errors_dict)
|
|
|
|
|
|
def populate_obj(self, obj, ignore_none=False):
|
|
|
"""
|
|
|
Populates the attributes of the passed `obj` with data from the form's
|
|
|
fields.
|
|
|
|
|
|
:note: This is a destructive operation; Any attribute with the same name
|
|
|
as a field will be overridden. Use with caution.
|
|
|
"""
|
|
|
if ignore_none:
|
|
|
for name, field in iteritems(self._fields):
|
|
|
if field.data is not None:
|
|
|
field.populate_obj(obj, name)
|
|
|
else:
|
|
|
for name, field in iteritems(self._fields):
|
|
|
field.populate_obj(obj, name)
|
|
|
|
|
|
css_classes = {}
|
|
|
ignore_labels = {}
|
|
|
|
|
|
|
|
|
class SignInForm(ReactorForm):
|
|
|
came_from = wtforms.HiddenField()
|
|
|
sign_in_user_name = wtforms.StringField(_("User Name"))
|
|
|
sign_in_user_password = wtforms.PasswordField(_("Password"))
|
|
|
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
html_attrs = {
|
|
|
"sign_in_user_name": {"placeholder": "Your login"},
|
|
|
"sign_in_user_password": {"placeholder": "Your password"},
|
|
|
}
|
|
|
|
|
|
|
|
|
from wtforms.widgets import html_params, HTMLString
|
|
|
|
|
|
|
|
|
def select_multi_checkbox(field, ul_class="set", **kwargs):
|
|
|
"""Render a multi-checkbox widget"""
|
|
|
kwargs.setdefault("type", "checkbox")
|
|
|
field_id = kwargs.pop("id", field.id)
|
|
|
html = ["<ul %s>" % html_params(id=field_id, class_=ul_class)]
|
|
|
for value, label, checked in field.iter_choices():
|
|
|
choice_id = "%s-%s" % (field_id, value)
|
|
|
options = dict(kwargs, name=field.name, value=value, id=choice_id)
|
|
|
if checked:
|
|
|
options["checked"] = "checked"
|
|
|
html.append("<li><input %s /> " % html_params(**options))
|
|
|
html.append('<label for="%s">%s</label></li>' % (choice_id, label))
|
|
|
html.append("</ul>")
|
|
|
return HTMLString("".join(html))
|
|
|
|
|
|
|
|
|
def button_widget(field, button_cls="ButtonField btn btn-default", **kwargs):
|
|
|
"""Render a button widget"""
|
|
|
kwargs.setdefault("type", "button")
|
|
|
field_id = kwargs.pop("id", field.id)
|
|
|
kwargs.setdefault("value", field.label.text)
|
|
|
html = [
|
|
|
"<button %s>%s</button>"
|
|
|
% (html_params(id=field_id, class_=button_cls), kwargs["value"])
|
|
|
]
|
|
|
return HTMLString("".join(html))
|
|
|
|
|
|
|
|
|
def clean_whitespace(value):
|
|
|
if value:
|
|
|
return value.strip()
|
|
|
return value
|
|
|
|
|
|
|
|
|
def found_username_validator(form, field):
|
|
|
user = UserService.by_user_name(field.data)
|
|
|
# sets user to recover in email validator
|
|
|
form.field_user = user
|
|
|
if not user:
|
|
|
raise wtforms.ValidationError("This username does not exist")
|
|
|
|
|
|
|
|
|
def found_username_email_validator(form, field):
|
|
|
user = UserService.by_email(field.data)
|
|
|
if not user:
|
|
|
raise wtforms.ValidationError("Email is incorrect")
|
|
|
|
|
|
|
|
|
def unique_username_validator(form, field):
|
|
|
user = UserService.by_user_name(field.data)
|
|
|
if user:
|
|
|
raise wtforms.ValidationError("This username already exists in system")
|
|
|
|
|
|
|
|
|
def unique_groupname_validator(form, field):
|
|
|
group = GroupService.by_group_name(field.data)
|
|
|
mod_group = getattr(form, "_modified_group", None)
|
|
|
if group and (not mod_group or mod_group.id != group.id):
|
|
|
raise wtforms.ValidationError("This group name already exists in system")
|
|
|
|
|
|
|
|
|
def unique_email_validator(form, field):
|
|
|
user = UserService.by_email(field.data)
|
|
|
if user:
|
|
|
raise wtforms.ValidationError("This email already exists in system")
|
|
|
|
|
|
|
|
|
def email_validator(form, field):
|
|
|
validator = formencode.validators.Email()
|
|
|
try:
|
|
|
validator.to_python(field.data)
|
|
|
except formencode.Invalid as e:
|
|
|
raise wtforms.ValidationError(e)
|
|
|
|
|
|
|
|
|
def unique_alert_email_validator(form, field):
|
|
|
q = DBSession.query(AlertChannel)
|
|
|
q = q.filter(AlertChannel.channel_name == "email")
|
|
|
q = q.filter(AlertChannel.channel_value == field.data)
|
|
|
email = q.first()
|
|
|
if email:
|
|
|
raise wtforms.ValidationError("This email already exists in alert system")
|
|
|
|
|
|
|
|
|
def blocked_email_validator(form, field):
|
|
|
blocked_emails = [
|
|
|
"goood-mail.org",
|
|
|
"shoeonlineblog.com",
|
|
|
"louboutinemart.com",
|
|
|
"guccibagshere.com",
|
|
|
"nikeshoesoutletforsale.com",
|
|
|
]
|
|
|
data = field.data or ""
|
|
|
domain = data.split("@")[-1]
|
|
|
if domain in blocked_emails:
|
|
|
raise wtforms.ValidationError("Don't spam")
|
|
|
|
|
|
|
|
|
def old_password_validator(form, field):
|
|
|
if not UserService.check_password(field.user, field.data or ""):
|
|
|
raise wtforms.ValidationError("You need to enter correct password")
|
|
|
|
|
|
|
|
|
class UserRegisterForm(ReactorForm):
|
|
|
user_name = wtforms.StringField(
|
|
|
_("User Name"),
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
wtforms.validators.Length(min=2, max=30),
|
|
|
wtforms.validators.Regexp(
|
|
|
re.compile(r"^[\.\w-]+$", re.UNICODE), message="Invalid characters used"
|
|
|
),
|
|
|
unique_username_validator,
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
|
|
|
user_password = wtforms.PasswordField(
|
|
|
_("User Password"),
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
wtforms.validators.Length(min=4),
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
|
|
|
email = wtforms.StringField(
|
|
|
_("Email Address"),
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
email_validator,
|
|
|
unique_email_validator,
|
|
|
blocked_email_validator,
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
first_name = wtforms.HiddenField(_("First Name"))
|
|
|
last_name = wtforms.HiddenField(_("Last Name"))
|
|
|
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
html_attrs = {
|
|
|
"user_name": {"placeholder": "Your login"},
|
|
|
"user_password": {"placeholder": "Your password"},
|
|
|
"email": {"placeholder": "Your email"},
|
|
|
}
|
|
|
|
|
|
|
|
|
class UserCreateForm(UserRegisterForm):
|
|
|
status = wtforms.BooleanField("User status", false_values=FALSE_VALUES)
|
|
|
|
|
|
|
|
|
class UserUpdateForm(UserCreateForm):
|
|
|
user_name = None
|
|
|
user_password = wtforms.PasswordField(
|
|
|
_("User Password"),
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.Length(min=4), wtforms.validators.Optional()],
|
|
|
)
|
|
|
email = wtforms.StringField(
|
|
|
_("Email Address"),
|
|
|
filters=[strip_filter],
|
|
|
validators=[email_validator, wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
|
|
|
|
|
|
class LostPasswordForm(ReactorForm):
|
|
|
email = wtforms.StringField(
|
|
|
_("Email Address"),
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
email_validator,
|
|
|
found_username_email_validator,
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
|
|
|
submit = wtforms.SubmitField(_("Reset password"))
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
|
|
|
class ChangePasswordForm(ReactorForm):
|
|
|
old_password = wtforms.PasswordField(
|
|
|
"Old Password",
|
|
|
filters=[strip_filter],
|
|
|
validators=[old_password_validator, wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
|
|
|
new_password = wtforms.PasswordField(
|
|
|
"New Password",
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
wtforms.validators.Length(min=4),
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
new_password_confirm = wtforms.PasswordField(
|
|
|
"Confirm Password",
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
wtforms.validators.EqualTo("new_password"),
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
submit = wtforms.SubmitField("Change Password")
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
|
|
|
class CheckPasswordForm(ReactorForm):
|
|
|
password = wtforms.PasswordField(
|
|
|
"Password",
|
|
|
filters=[strip_filter],
|
|
|
validators=[old_password_validator, wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
|
|
|
|
|
|
class NewPasswordForm(ReactorForm):
|
|
|
new_password = wtforms.PasswordField(
|
|
|
"New Password",
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
wtforms.validators.Length(min=4),
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
new_password_confirm = wtforms.PasswordField(
|
|
|
"Confirm Password",
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
wtforms.validators.EqualTo("new_password"),
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
submit = wtforms.SubmitField("Set Password")
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
|
|
|
class CORSTextAreaField(wtforms.StringField):
|
|
|
"""
|
|
|
This field represents an HTML ``<textarea>`` and can be used to take
|
|
|
multi-line input.
|
|
|
"""
|
|
|
|
|
|
widget = wtforms.widgets.TextArea()
|
|
|
|
|
|
def process_formdata(self, valuelist):
|
|
|
self.data = []
|
|
|
if valuelist:
|
|
|
data = [x.strip() for x in valuelist[0].split("\n")]
|
|
|
for d in data:
|
|
|
if not d:
|
|
|
continue
|
|
|
if d.startswith("www."):
|
|
|
d = d[4:]
|
|
|
if data:
|
|
|
self.data.append(d)
|
|
|
else:
|
|
|
self.data = []
|
|
|
self.data = "\n".join(self.data)
|
|
|
|
|
|
|
|
|
class ApplicationCreateForm(ReactorForm):
|
|
|
resource_name = wtforms.StringField(
|
|
|
_("Application name"),
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
wtforms.validators.Length(min=1),
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
|
|
|
domains = CORSTextAreaField(
|
|
|
_("Domain names for CORS headers "),
|
|
|
validators=[wtforms.validators.Length(min=1), wtforms.validators.Optional()],
|
|
|
description="Required for Javascript error "
|
|
|
"tracking (one line one domain, skip http:// part)",
|
|
|
)
|
|
|
|
|
|
submit = wtforms.SubmitField(_("Create Application"))
|
|
|
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
html_attrs = {
|
|
|
"resource_name": {"placeholder": "Application Name"},
|
|
|
"uptime_url": {"placeholder": "http://somedomain.com"},
|
|
|
}
|
|
|
|
|
|
|
|
|
class ApplicationUpdateForm(ApplicationCreateForm):
|
|
|
default_grouping = wtforms.SelectField(
|
|
|
_("Default grouping for errors"),
|
|
|
choices=[
|
|
|
("url_type", "Error Type + location"),
|
|
|
("url_traceback", "Traceback + location"),
|
|
|
("traceback_server", "Traceback + Server"),
|
|
|
],
|
|
|
default="url_traceback",
|
|
|
)
|
|
|
|
|
|
error_report_threshold = wtforms.IntegerField(
|
|
|
_("Alert on error reports"),
|
|
|
validators=[
|
|
|
wtforms.validators.NumberRange(min=1),
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
description="Application requires to send at least this amount of "
|
|
|
"error reports per minute to open alert",
|
|
|
)
|
|
|
|
|
|
slow_report_threshold = wtforms.IntegerField(
|
|
|
_("Alert on slow reports"),
|
|
|
validators=[
|
|
|
wtforms.validators.NumberRange(min=1),
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
description="Application requires to send at least this amount of "
|
|
|
"slow reports per minute to open alert",
|
|
|
)
|
|
|
|
|
|
allow_permanent_storage = wtforms.BooleanField(
|
|
|
_("Permanent logs"),
|
|
|
false_values=FALSE_VALUES,
|
|
|
description=_("Allow permanent storage of logs in separate DB partitions"),
|
|
|
)
|
|
|
|
|
|
submit = wtforms.SubmitField(_("Create Application"))
|
|
|
|
|
|
|
|
|
class UserSearchSchemaForm(ReactorForm):
|
|
|
user_name = wtforms.StringField("User Name", filters=[strip_filter])
|
|
|
|
|
|
submit = wtforms.SubmitField(_("Search User"))
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
'<li class="user_exists"><span></span></li>'
|
|
|
|
|
|
|
|
|
class YesNoForm(ReactorForm):
|
|
|
no = wtforms.SubmitField("No", default="")
|
|
|
yes = wtforms.SubmitField("Yes", default="")
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
|
|
|
status_codes = [("", "All"), ("500", "500"), ("404", "404")]
|
|
|
|
|
|
priorities = [("", "All")]
|
|
|
for i in range(1, 11):
|
|
|
priorities.append((str(i), str(i)))
|
|
|
|
|
|
report_status_choices = [
|
|
|
("", "All"),
|
|
|
("never_reviewed", "Never revieved"),
|
|
|
("reviewed", "Revieved"),
|
|
|
("public", "Public"),
|
|
|
("fixed", "Fixed"),
|
|
|
]
|
|
|
|
|
|
|
|
|
class ReportBrowserForm(ReactorForm):
|
|
|
applications = wtforms.SelectMultipleField(
|
|
|
"Applications", widget=select_multi_checkbox
|
|
|
)
|
|
|
http_status = wtforms.SelectField("HTTP Status", choices=status_codes)
|
|
|
priority = wtforms.SelectField("Priority", choices=priorities, default="")
|
|
|
start_date = wtforms.DateField("Start Date")
|
|
|
end_date = wtforms.DateField("End Date")
|
|
|
error = wtforms.StringField("Error")
|
|
|
url_path = wtforms.StringField("URL Path")
|
|
|
url_domain = wtforms.StringField("URL Domain")
|
|
|
report_status = wtforms.SelectField(
|
|
|
"Report status", choices=report_status_choices, default=""
|
|
|
)
|
|
|
submit = wtforms.SubmitField(
|
|
|
'<span class="glyphicon glyphicon-search">' "</span> Filter results",
|
|
|
widget=button_widget,
|
|
|
)
|
|
|
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
|
|
|
slow_report_status_choices = [
|
|
|
("", "All"),
|
|
|
("never_reviewed", "Never revieved"),
|
|
|
("reviewed", "Revieved"),
|
|
|
("public", "Public"),
|
|
|
]
|
|
|
|
|
|
|
|
|
class BulkOperationForm(ReactorForm):
|
|
|
applications = wtforms.SelectField("Applications")
|
|
|
start_date = wtforms.DateField(
|
|
|
"Start Date",
|
|
|
default=lambda: datetime.datetime.utcnow() - datetime.timedelta(days=90),
|
|
|
)
|
|
|
end_date = wtforms.DateField("End Date")
|
|
|
confirm = wtforms.BooleanField(
|
|
|
"Confirm operation", validators=[wtforms.validators.DataRequired()]
|
|
|
)
|
|
|
|
|
|
|
|
|
class LogBrowserForm(ReactorForm):
|
|
|
applications = wtforms.SelectMultipleField(
|
|
|
"Applications", widget=select_multi_checkbox
|
|
|
)
|
|
|
start_date = wtforms.DateField("Start Date")
|
|
|
log_level = wtforms.StringField("Log level")
|
|
|
message = wtforms.StringField("Message")
|
|
|
namespace = wtforms.StringField("Namespace")
|
|
|
submit = wtforms.SubmitField(
|
|
|
'<span class="glyphicon glyphicon-search"></span> Filter results',
|
|
|
widget=button_widget,
|
|
|
)
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
|
|
|
class CommentForm(ReactorForm):
|
|
|
body = wtforms.TextAreaField(
|
|
|
"Comment",
|
|
|
validators=[
|
|
|
wtforms.validators.Length(min=1),
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
submit = wtforms.SubmitField("Comment")
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
|
|
|
class EmailChannelCreateForm(ReactorForm):
|
|
|
email = wtforms.StringField(
|
|
|
_("Email Address"),
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
email_validator,
|
|
|
unique_alert_email_validator,
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
submit = wtforms.SubmitField("Add email channel")
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
|
|
|
def gen_user_profile_form():
|
|
|
class UserProfileForm(ReactorForm):
|
|
|
email = wtforms.StringField(
|
|
|
_("Email Address"),
|
|
|
validators=[email_validator, wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
first_name = wtforms.StringField(_("First Name"))
|
|
|
last_name = wtforms.StringField(_("Last Name"))
|
|
|
company_name = wtforms.StringField(_("Company Name"))
|
|
|
company_address = wtforms.TextAreaField(_("Company Address"))
|
|
|
zip_code = wtforms.StringField(_("ZIP code"))
|
|
|
city = wtforms.StringField(_("City"))
|
|
|
notifications = wtforms.BooleanField(
|
|
|
"Account notifications", false_values=FALSE_VALUES
|
|
|
)
|
|
|
submit = wtforms.SubmitField(_("Update Account"))
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
return UserProfileForm
|
|
|
|
|
|
|
|
|
class PurgeAppForm(ReactorForm):
|
|
|
resource_id = wtforms.HiddenField(
|
|
|
"App Id", validators=[wtforms.validators.DataRequired()]
|
|
|
)
|
|
|
days = wtforms.IntegerField("Days", validators=[wtforms.validators.DataRequired()])
|
|
|
password = wtforms.PasswordField(
|
|
|
"Admin Password",
|
|
|
validators=[old_password_validator, wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
submit = wtforms.SubmitField(_("Purge Data"))
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
|
|
|
class IntegrationRepoForm(ReactorForm):
|
|
|
host_name = wtforms.StringField("Service Host", default="")
|
|
|
user_name = wtforms.StringField(
|
|
|
"User Name",
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
wtforms.validators.DataRequired(),
|
|
|
wtforms.validators.Length(min=1),
|
|
|
],
|
|
|
)
|
|
|
repo_name = wtforms.StringField(
|
|
|
"Repo Name",
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
wtforms.validators.DataRequired(),
|
|
|
wtforms.validators.Length(min=1),
|
|
|
],
|
|
|
)
|
|
|
|
|
|
|
|
|
class IntegrationBitbucketForm(IntegrationRepoForm):
|
|
|
host_name = wtforms.StringField("Service Host", default="https://bitbucket.org")
|
|
|
|
|
|
def validate_user_name(self, field):
|
|
|
try:
|
|
|
request = pyramid.threadlocal.get_current_request()
|
|
|
client = BitbucketIntegration.create_client(
|
|
|
request, self.user_name.data, self.repo_name.data
|
|
|
)
|
|
|
client.get_assignees()
|
|
|
except IntegrationException as e:
|
|
|
raise wtforms.validators.ValidationError(str(e))
|
|
|
|
|
|
|
|
|
class IntegrationGithubForm(IntegrationRepoForm):
|
|
|
host_name = wtforms.StringField("Service Host", default="https://github.com")
|
|
|
|
|
|
def validate_user_name(self, field):
|
|
|
try:
|
|
|
request = pyramid.threadlocal.get_current_request()
|
|
|
client = GithubIntegration.create_client(
|
|
|
request, self.user_name.data, self.repo_name.data
|
|
|
)
|
|
|
client.get_assignees()
|
|
|
except IntegrationException as e:
|
|
|
raise wtforms.validators.ValidationError(str(e))
|
|
|
raise wtforms.validators.ValidationError(str(e))
|
|
|
|
|
|
|
|
|
def filter_rooms(data):
|
|
|
if data is not None:
|
|
|
rooms = data.split(",")
|
|
|
return ",".join([r.strip() for r in rooms])
|
|
|
|
|
|
|
|
|
class IntegrationCampfireForm(ReactorForm):
|
|
|
account = wtforms.StringField(
|
|
|
"Account",
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
api_token = wtforms.StringField(
|
|
|
"Api Token",
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
rooms = wtforms.StringField("Room ID list", filters=[filter_rooms])
|
|
|
|
|
|
def validate_api_token(self, field):
|
|
|
try:
|
|
|
client = CampfireIntegration.create_client(
|
|
|
self.api_token.data, self.account.data
|
|
|
)
|
|
|
client.get_account()
|
|
|
except IntegrationException as e:
|
|
|
raise wtforms.validators.ValidationError(str(e))
|
|
|
|
|
|
def validate_rooms(self, field):
|
|
|
if not field.data:
|
|
|
return
|
|
|
client = CampfireIntegration.create_client(
|
|
|
self.api_token.data, self.account.data
|
|
|
)
|
|
|
|
|
|
try:
|
|
|
room_list = [r["id"] for r in client.get_rooms()]
|
|
|
except IntegrationException as e:
|
|
|
raise wtforms.validators.ValidationError(str(e))
|
|
|
|
|
|
rooms = field.data.split(",")
|
|
|
if len(rooms) > 3:
|
|
|
msg = "You can use up to 3 room ids"
|
|
|
raise wtforms.validators.ValidationError(msg)
|
|
|
if rooms:
|
|
|
for room_id in rooms:
|
|
|
if int(room_id) not in room_list:
|
|
|
msg = "Room %s doesn't exist"
|
|
|
raise wtforms.validators.ValidationError(msg % room_id)
|
|
|
if not room_id.strip().isdigit():
|
|
|
msg = "You must use only integers for room ids"
|
|
|
raise wtforms.validators.ValidationError(msg)
|
|
|
|
|
|
submit = wtforms.SubmitField(_("Connect to Campfire"))
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
|
|
|
def filter_rooms(data):
|
|
|
if data is not None:
|
|
|
rooms = data.split(",")
|
|
|
return ",".join([r.strip() for r in rooms])
|
|
|
|
|
|
|
|
|
class IntegrationHipchatForm(ReactorForm):
|
|
|
api_token = wtforms.StringField(
|
|
|
"Api Token",
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
rooms = wtforms.StringField(
|
|
|
"Room ID list",
|
|
|
filters=[filter_rooms],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
|
|
|
def validate_rooms(self, field):
|
|
|
if not field.data:
|
|
|
return
|
|
|
client = HipchatIntegration.create_client(self.api_token.data)
|
|
|
rooms = field.data.split(",")
|
|
|
if len(rooms) > 3:
|
|
|
msg = "You can use up to 3 room ids"
|
|
|
raise wtforms.validators.ValidationError(msg)
|
|
|
if rooms:
|
|
|
for room_id in rooms:
|
|
|
if not room_id.strip().isdigit():
|
|
|
msg = "You must use only integers for room ids"
|
|
|
raise wtforms.validators.ValidationError(msg)
|
|
|
try:
|
|
|
client.send(
|
|
|
{
|
|
|
"message_format": "text",
|
|
|
"message": "testing for room existence",
|
|
|
"from": "AppEnlight",
|
|
|
"room_id": room_id,
|
|
|
"color": "green",
|
|
|
}
|
|
|
)
|
|
|
except IntegrationException as exc:
|
|
|
msg = "Room id: %s exception: %s"
|
|
|
raise wtforms.validators.ValidationError(msg % (room_id, exc))
|
|
|
|
|
|
|
|
|
class IntegrationFlowdockForm(ReactorForm):
|
|
|
api_token = wtforms.StringField(
|
|
|
"API Token",
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
|
|
|
def validate_api_token(self, field):
|
|
|
try:
|
|
|
client = FlowdockIntegration.create_client(self.api_token.data)
|
|
|
registry = pyramid.threadlocal.get_current_registry()
|
|
|
payload = {
|
|
|
"source": registry.settings["mailing.from_name"],
|
|
|
"from_address": registry.settings["mailing.from_email"],
|
|
|
"subject": "Integration test",
|
|
|
"content": "If you can see this it was successful",
|
|
|
"tags": ["appenlight"],
|
|
|
"link": registry.settings["mailing.app_url"],
|
|
|
}
|
|
|
client.send_to_inbox(payload)
|
|
|
except IntegrationException as e:
|
|
|
raise wtforms.validators.ValidationError(str(e))
|
|
|
|
|
|
|
|
|
class IntegrationSlackForm(ReactorForm):
|
|
|
webhook_url = wtforms.StringField(
|
|
|
"Reports webhook",
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
|
|
|
def validate_webhook_url(self, field):
|
|
|
registry = pyramid.threadlocal.get_current_registry()
|
|
|
client = SlackIntegration.create_client(field.data)
|
|
|
link = "<%s|%s>" % (
|
|
|
registry.settings["mailing.app_url"],
|
|
|
registry.settings["mailing.from_name"],
|
|
|
)
|
|
|
test_data = {
|
|
|
"username": "AppEnlight",
|
|
|
"icon_emoji": ":fire:",
|
|
|
"attachments": [
|
|
|
{
|
|
|
"fallback": "Testing integration channel: %s" % link,
|
|
|
"pretext": "Testing integration channel: %s" % link,
|
|
|
"color": "good",
|
|
|
"fields": [
|
|
|
{
|
|
|
"title": "Status",
|
|
|
"value": "Integration is working fine",
|
|
|
"short": False,
|
|
|
}
|
|
|
],
|
|
|
}
|
|
|
],
|
|
|
}
|
|
|
try:
|
|
|
client.make_request(data=test_data)
|
|
|
except IntegrationException as exc:
|
|
|
raise wtforms.validators.ValidationError(str(exc))
|
|
|
|
|
|
|
|
|
class IntegrationWebhooksForm(ReactorForm):
|
|
|
reports_webhook = wtforms.StringField(
|
|
|
"Reports webhook",
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
alerts_webhook = wtforms.StringField(
|
|
|
"Alerts webhook",
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
submit = wtforms.SubmitField(_("Setup webhooks"))
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-primary"}
|
|
|
|
|
|
|
|
|
class IntegrationJiraForm(ReactorForm):
|
|
|
host_name = wtforms.StringField(
|
|
|
"Server URL",
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
user_name = wtforms.StringField(
|
|
|
"Username",
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
password = wtforms.PasswordField(
|
|
|
"Password",
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
project = wtforms.StringField(
|
|
|
"Project key",
|
|
|
filters=[uppercase_filter, strip_filter],
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
|
|
|
def validate_project(self, field):
|
|
|
if not field.data:
|
|
|
return
|
|
|
try:
|
|
|
client = JiraClient(
|
|
|
self.user_name.data,
|
|
|
self.password.data,
|
|
|
self.host_name.data,
|
|
|
self.project.data,
|
|
|
)
|
|
|
except Exception as exc:
|
|
|
raise wtforms.validators.ValidationError(str(exc))
|
|
|
|
|
|
room_list = [r.key.upper() for r in client.get_projects()]
|
|
|
if field.data.upper() not in room_list:
|
|
|
msg = "Project %s doesn\t exist in your Jira Instance"
|
|
|
raise wtforms.validators.ValidationError(msg % field.data)
|
|
|
|
|
|
|
|
|
def get_deletion_form(resource):
|
|
|
class F(ReactorForm):
|
|
|
application_name = wtforms.StringField(
|
|
|
"Application Name",
|
|
|
filters=[strip_filter],
|
|
|
validators=[wtforms.validators.AnyOf([resource.resource_name])],
|
|
|
)
|
|
|
resource_id = wtforms.HiddenField(default=resource.resource_id)
|
|
|
submit = wtforms.SubmitField(_("Delete my application"))
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-danger"}
|
|
|
|
|
|
return F
|
|
|
|
|
|
|
|
|
class ChangeApplicationOwnerForm(ReactorForm):
|
|
|
password = wtforms.PasswordField(
|
|
|
"Password",
|
|
|
filters=[strip_filter],
|
|
|
validators=[old_password_validator, wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
|
|
|
user_name = wtforms.StringField(
|
|
|
"New owners username",
|
|
|
filters=[strip_filter],
|
|
|
validators=[found_username_validator, wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
submit = wtforms.SubmitField(_("Transfer ownership of application"))
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-danger"}
|
|
|
|
|
|
|
|
|
def default_filename():
|
|
|
return "Invoice %s" % datetime.datetime.utcnow().strftime("%Y/%m")
|
|
|
|
|
|
|
|
|
class FileUploadForm(ReactorForm):
|
|
|
title = wtforms.StringField(
|
|
|
"File Title",
|
|
|
default=default_filename,
|
|
|
validators=[wtforms.validators.DataRequired()],
|
|
|
)
|
|
|
file = wtforms.FileField("File")
|
|
|
|
|
|
def validate_file(self, field):
|
|
|
if not hasattr(field.data, "file"):
|
|
|
raise wtforms.ValidationError("File is missing")
|
|
|
|
|
|
submit = wtforms.SubmitField(_("Upload"))
|
|
|
|
|
|
|
|
|
def get_partition_deletion_form(es_indices, pg_indices):
|
|
|
class F(ReactorForm):
|
|
|
es_index = wtforms.SelectMultipleField(
|
|
|
"Elasticsearch", choices=[(ix, "") for ix in es_indices]
|
|
|
)
|
|
|
pg_index = wtforms.SelectMultipleField(
|
|
|
"pg", choices=[(ix, "") for ix in pg_indices]
|
|
|
)
|
|
|
confirm = wtforms.TextField(
|
|
|
"Confirm",
|
|
|
filters=[uppercase_filter, strip_filter],
|
|
|
validators=[
|
|
|
wtforms.validators.AnyOf(["CONFIRM"]),
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
ignore_labels = ["submit"]
|
|
|
css_classes = {"submit": "btn btn-danger"}
|
|
|
|
|
|
return F
|
|
|
|
|
|
|
|
|
class GroupCreateForm(ReactorForm):
|
|
|
group_name = wtforms.StringField(
|
|
|
_("Group Name"),
|
|
|
filters=[strip_filter],
|
|
|
validators=[
|
|
|
wtforms.validators.Length(min=2, max=50),
|
|
|
unique_groupname_validator,
|
|
|
wtforms.validators.DataRequired(),
|
|
|
],
|
|
|
)
|
|
|
description = wtforms.StringField(_("Group description"))
|
|
|
|
|
|
|
|
|
time_choices = [(k, v["label"]) for k, v in h.time_deltas.items()]
|
|
|
|
|
|
|
|
|
class AuthTokenCreateForm(ReactorForm):
|
|
|
description = wtforms.StringField(_("Token description"))
|
|
|
expires = wtforms.SelectField(
|
|
|
"Expires",
|
|
|
coerce=lambda x: x,
|
|
|
choices=time_choices,
|
|
|
validators=[wtforms.validators.Optional()],
|
|
|
)
|
|
|
|
