feat: activations with webhooks

pyproject.toml

@@ -52,6 +52,10 @@ django-split-settings = "^1.2.0"
 pexpect = "^4.9.0"
 ansible-runner = ">=2.3"
 python-gnupg = "^0.5.2"
+psycopg = "^3.1.17"
+xxhash = "*"
+pyjwt = { version="*", extras = ["cryptography"] }
+ecdsa = "*"
 
 [tool.poetry.group.test.dependencies]
 pytest = "*"

src/aap_eda/api/exceptions.py

@@ -125,3 +125,10 @@ class InvalidEventStreamRulebook(APIException):
  default_detail = (
  "Configuration Error: Event stream template rulebook is invalid"
  )
+
+
+class InvalidWebhookSource(APIException):
+ status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+ default_detail = (
+ "Configuration Error: Webhook source could not be upated in ruleset"
+ )

src/aap_eda/api/filters/webhook.py

@@ -0,0 +1,29 @@
+# Copyright 2024 Red Hat, Inc.
+#
+# 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 django_filters
+
+from aap_eda.core import models
+
+
+class WebhookFilter(django_filters.FilterSet):
+ name = django_filters.CharFilter(
+ field_name="name",
+ lookup_expr="istartswith",
+ label="Filter by webhook name.",
+ )
+
+ class Meta:
+ model = models.Webhook
+ fields = ["name"]

src/aap_eda/api/serializers/__init__.py

@@ -78,6 +78,7 @@
  UserListSerializer,
  UserSerializer,
 )
+from .webhook import WebhookInSerializer, WebhookOutSerializer
 
 __all__ = (
  # auth
@@ -131,4 +132,7 @@
  "TeamCreateSerializer",
  "TeamUpdateSerializer",
  "TeamDetailSerializer",
+ # webhooks
+ "WebhookInSerializer",
+ "WebhookOutSerializer",
 )

src/aap_eda/api/serializers/activation.py

@@ -26,7 +26,10 @@
  EDA_SERVER_VAULT_LABEL,
  PG_NOTIFY_TEMPLATE_RULEBOOK_DATA,
 )
-from aap_eda.api.exceptions import InvalidEventStreamRulebook
+from aap_eda.api.exceptions import (
+ InvalidEventStreamRulebook,
+ InvalidWebhookSource,
+)
 from aap_eda.api.serializers.decision_environment import (
  DecisionEnvironmentRefSerializer,
 )
@@ -39,6 +42,7 @@
  ProjectRefSerializer,
 )
 from aap_eda.api.serializers.rulebook import RulebookRefSerializer
+from aap_eda.api.serializers.webhook import WebhookOutSerializer
 from aap_eda.api.vault import encrypt_string
 from aap_eda.core import models, validators
 from aap_eda.core.enums import DefaultCredentialType, ProcessParentType
@@ -63,6 +67,36 @@ class VaultData:
  password_used: bool = False
 
 
+def _update_webhook_source(validated_data: dict, vault_data: VaultData):
+ try:
+ vault_data.password_used = True
+ encrypted_dsn = encrypt_string(
+ password=vault_data.password,
+ plaintext=settings.PG_NOTIFY_DSN,
+ vault_id=EDA_SERVER_VAULT_LABEL,
+ )
+
+ channels = []
+ for webhook_id in validated_data.get("webhooks"):
+ obj = models.Webhook.objects.get(id=webhook_id)
+ channels.append(obj.channel_name)
+
+ sources_info = [
+ {
+ "name": "webhook_event_stream",
+ "type": "ansible.eda.pg_listener",
+ "args": {
+ "dsn": encrypted_dsn,
+ "channels": channels,
+ },
+ }
+ ]
+ return swap_sources(validated_data["rulebook_rulesets"], sources_info)
+ except Exception as e:
+ logger.error(f"Failed to update webhook source in rulesets: {e}")
+ raise InvalidWebhookSource(e)
+
+
 def _updated_ruleset(validated_data: dict, vault_data: VaultData):
  try:
  sources_info = []
@@ -215,6 +249,12 @@ class ActivationSerializer(serializers.ModelSerializer):
  child=EdaCredentialSerializer(),
  )
 
+ webhooks = serializers.ListField(
+ required=False,
+ allow_null=True,
+ child=WebhookOutSerializer(),
+ )
+
  class Meta:
  model = models.Activation
  fields = [
@@ -241,6 +281,7 @@ class Meta:
  "event_streams",
  "eda_credentials",
  "log_level",
+ "webhooks",
  ]
  read_only_fields = [
  "id",
@@ -271,6 +312,11 @@ class ActivationListSerializer(serializers.ModelSerializer):
  allow_null=True,
  help_text="Service name of the activation",
  )
+ webhooks = serializers.ListField(
+ required=False,
+ allow_null=True,
+ child=WebhookOutSerializer(),
+ )
 
  class Meta:
  model = models.Activation
@@ -299,6 +345,7 @@ class Meta:
  "log_level",
  "eda_credentials",
  "k8s_service_name",
+ "webhooks",
  ]
  read_only_fields = ["id", "created_at", "modified_at"]
 
@@ -319,6 +366,10 @@ def to_representation(self, activation):
  if activation.extra_var
  else None
  )
+ webhooks = [
+ WebhookOutSerializer(webhook).data
+ for webhook in activation.webhooks.all()
+ ]
 
  return {
  "id": activation.id,
@@ -345,6 +396,7 @@ def to_representation(self, activation):
  "log_level": activation.log_level,
  "eda_credentials": eda_credentials,
  "k8s_service_name": activation.k8s_service_name,
+ "webhooks": webhooks,
  }
 
 
@@ -368,6 +420,7 @@ class Meta:
  "log_level",
  "eda_credentials",
  "k8s_service_name",
+ "webhooks",
  ]
 
  organization_id = serializers.IntegerField(
@@ -410,6 +463,12 @@ class Meta:
  allow_null=True,
  validators=[validators.check_if_rfc_1035_compliant],
  )
+ webhooks = serializers.ListField(
+ required=False,
+ allow_null=True,
+ child=serializers.IntegerField(),
+ validators=[validators.check_if_webhooks_exists],
+ )
 
  def validate(self, data):
  _validate_credentials_and_token_and_rulebook(data=data, creating=True)
@@ -437,6 +496,11 @@ def create(self, validated_data):
  validated_data, vault_data
  )
 
+ if validated_data.get("webhooks", []):
+ validated_data["rulebook_rulesets"] = _update_webhook_source(
+ validated_data, vault_data
+ )
+
  vault = _get_vault_credential_type()
 
  if validated_data.get("eda_credentials"):
@@ -525,6 +589,11 @@ class ActivationReadSerializer(serializers.ModelSerializer):
  allow_null=True,
  help_text="Service name of the activation",
  )
+ webhooks = serializers.ListField(
+ required=False,
+ allow_null=True,
+ child=WebhookOutSerializer(),
+ )
 
  class Meta:
  model = models.Activation
@@ -556,6 +625,7 @@ class Meta:
  "event_streams",
  "log_level",
  "k8s_service_name",
+ "webhooks",
  ]
  read_only_fields = ["id", "created_at", "modified_at", "restarted_at"]
 
@@ -610,6 +680,10 @@ def to_representation(self, activation):
  if activation.extra_var
  else None
  )
+ webhooks = [
+ WebhookOutSerializer(webhook).data
+ for webhook in activation.webhooks.all()
+ ]
 
  return {
  "id": activation.id,
@@ -641,6 +715,7 @@ def to_representation(self, activation):
  "log_level": activation.log_level,
  "eda_credentials": eda_credentials,
  "k8s_service_name": activation.k8s_service_name,
+ "webhooks": webhooks,
  }
 
 
@@ -672,6 +747,12 @@ class PostActivationSerializer(serializers.ModelSerializer):
  allow_null=True,
  validators=[validators.check_if_rfc_1035_compliant],
  )
+ webhooks = serializers.ListField(
+ required=False,
+ allow_null=True,
+ child=serializers.IntegerField(),
+ validators=[validators.check_if_webhooks_exists],
+ )
 
  def validate(self, data):
  _validate_credentials_and_token_and_rulebook(data=data, creating=False)
@@ -694,6 +775,7 @@ class Meta:
  "rulebook_id",
  "eda_credentials",
  "k8s_service_name",
+ "webhooks",
  ]
  read_only_fields = [
  "id",
@@ -719,6 +801,7 @@ def is_activation_valid(activation: models.Activation) -> tuple[bool, str]:
  data["eda_credentials"] = [
  obj.id for obj in activation.eda_credentials.all()
  ]
+ data["webhooks"] = [obj.id for obj in activation.webhooks.all()]
  serializer = PostActivationSerializer(data=data)
 
  valid = serializer.is_valid()