Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion web/opencve/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,10 @@ def __str__(self):

# Update the update_at field at each change
def _pre_save(instance, **kwargs):
instance.updated_at = timezone.now()
if getattr(instance, "_skip_auto_updated_at", False):
return
if hasattr(instance, "updated_at"):
instance.updated_at = timezone.now()


signals.pre_save.connect(_pre_save)
86 changes: 85 additions & 1 deletion web/projects/admin.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,87 @@
from django.contrib import admin

# Register your models here.
from projects.forms import AutomationAdminForm
from projects.models import Automation


@admin.register(Automation)
class AutomationAdmin(admin.ModelAdmin):
form = AutomationAdminForm
view_on_site = False
ordering = ("-created_at",)
list_display = (
"name",
"project",
"organization",
"trigger_type",
"is_enabled",
"frequency",
"last_execution_at",
"created_at",
)
list_filter = ("trigger_type", "is_enabled", "frequency")
search_fields = (
"name",
"project__name",
"project__organization__name",
)
raw_id_fields = ("project",)
readonly_fields = ("conditions_count_display",)
fieldsets = (
(
None,
{
"fields": (
"name",
"project",
"is_enabled",
"trigger_type",
)
},
),
(
"Schedule",
{
"fields": (
"frequency",
"schedule_timezone",
"schedule_time",
"schedule_weekday",
),
},
),
(
"Configuration",
{
"fields": (
"configuration",
"conditions_count_display",
),
},
),
(
"Metadata",
{
"fields": (
"last_execution_at",
"created_at",
"updated_at",
),
},
),
)

def get_queryset(self, request):
return (
super()
.get_queryset(request)
.select_related("project", "project__organization")
)

@admin.display(description="Organization")
def organization(self, obj):
return obj.project.organization.name

@admin.display(description="Conditions")
def conditions_count_display(self, obj):
return obj.conditions_count
114 changes: 114 additions & 0 deletions web/projects/forms.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from django import forms
from django.conf import settings
from django.db.models import Q
from django.utils import timezone

from projects.models import Automation, Notification, Project, CveTracker
from users.models import User
Expand Down Expand Up @@ -435,3 +436,116 @@ def save(self, commit=True):
if commit:
instance.save()
return instance


class AutomationAdminForm(AutomationForm):
"""Admin form for Automation with project and configuration fields."""

class Meta(AutomationForm.Meta):
fields = [
"project",
"name",
"is_enabled",
"trigger_type",
"frequency",
"schedule_timezone",
"schedule_time",
"schedule_weekday",
"configuration",
"last_execution_at",
"created_at",
"updated_at",
]

def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.fields.pop("configuration_json", None)
self.fields["last_execution_at"].required = False
self.fields["created_at"].required = False
self.fields["updated_at"].required = False
self.fields["trigger_type"].widget = forms.Select(
choices=Automation.TRIGGER_CHOICES
)
self.fields["frequency"].widget = forms.Select(
choices=[("", "---------"), *Automation.FREQUENCY_CHOICES]
)
self.fields["schedule_timezone"] = forms.CharField(
required=False,
max_length=64,
widget=forms.TextInput(attrs={"class": "vTextField"}),
)
if not self.instance._state.adding and self.instance.project_id:
self.project = self.instance.project

def clean_name(self):
name = self.cleaned_data["name"]
if name in ("add",):
raise forms.ValidationError("This name is reserved.")
return name

def clean(self):
cleaned_data = super().clean()

project = cleaned_data.get("project")
if not project and not self.instance._state.adding and self.instance.project_id:
project = self.instance.project

if project:
self.project = project

name = cleaned_data.get("name")
if project and name and (not self.instance.pk or self.instance.name != name):
if Automation.objects.filter(project=project, name=name).exists():
self.add_error("name", "This name already exists.")

return cleaned_data

def clean_configuration(self):
config = self.cleaned_data.get("configuration")

if not isinstance(config, dict):
raise forms.ValidationError("Invalid configuration format.")
if "conditions" not in config or "actions" not in config:
raise forms.ValidationError(
"Configuration must contain 'conditions' and 'actions'."
)

self._validate_conditions_tree(config["conditions"])

if not isinstance(config["actions"], list):
raise forms.ValidationError("Actions must be a list.")

if "triggers" in config:
if not isinstance(config["triggers"], list):
raise forms.ValidationError("Triggers must be a list.")
for trigger in config["triggers"]:
if not isinstance(trigger, str):
raise forms.ValidationError("Each trigger must be a string.")

if self._get_trigger_type() == Automation.TRIGGER_ALERT:
triggers = config.get("triggers") or []
if not triggers:
raise forms.ValidationError(
"At least one event is required for alert automations."
)
if not config["actions"]:
raise forms.ValidationError(
"At least one action is required for alert automations."
)

return config

def save(self, commit=True):
instance = forms.ModelForm.save(self, commit=False)
instance._skip_auto_updated_at = True

if instance._state.adding:
if not self.cleaned_data.get("created_at"):
instance.created_at = timezone.now()
if not self.cleaned_data.get("updated_at"):
instance.updated_at = timezone.now()

if commit:
instance.save()

return instance
Loading