Coverage for extras/event_rules.py: 57%

1import logging 

2 

3from django.core.exceptions import ValidationError 

4from django.utils import timezone 

5from django.utils.translation import gettext_lazy as _ 

6from django_rq import get_queue 

7 

8from netbox.config import get_config 

9from netbox.constants import RQ_QUEUE_DEFAULT 

10from netbox.event_rules import EventRuleAction 

11from utilities.request import copy_safe_request 

12from utilities.rqworker import get_rq_retry 

13 

14from .choices import EventRuleActionChoices 

15from .models import NotificationGroup, Script, Webhook 

16 

17__all__ = ( 

18 'NotificationAction', 

19 'ScriptAction', 

20 'WebhookAction', 

21) 

22 

23logger = logging.getLogger('netbox.events_processor') 

24 

25 

26class WebhookAction(EventRuleAction): 

27 slug = EventRuleActionChoices.WEBHOOK 

28 label = _('Webhook') 

29 description = _('Send an outgoing HTTP request to a remote endpoint') 

30 object_model = Webhook 

31 object_required = True 

32 

33 def enqueue(self, *, event_rule, event_context, action_object, action_data): 

34 # Select the appropriate RQ queue 

35 queue_name = get_config().QUEUE_MAPPINGS.get('webhook', RQ_QUEUE_DEFAULT) 

36 rq_queue = get_queue(queue_name) 

37 

38 # Compile the task parameters 

39 params = { 

40 'event_rule': event_rule, 

41 'object_type': event_context['object_type'], 

42 'event_type': event_context['event_type'], 

43 'data': action_data, 

44 'snapshots': event_context.get('snapshots'), 

45 'timestamp': timezone.now().isoformat(), 

46 'retry': get_rq_retry(), 

47 } 

48 if 'request' in event_context: 

49 # Exclude FILES - webhooks don't need uploaded files, 

50 # which can cause pickle errors with Pillow. 

51 params['request'] = copy_safe_request(event_context['request'], include_files=False) 

52 

53 # Enqueue the task 

54 rq_queue.enqueue('extras.webhooks.send_webhook', **params) 

55 

56 def resolve_import_object(self, value): 

57 return Webhook.objects.get(name=value) 

58 

59 

60class ScriptAction(EventRuleAction): 

61 slug = EventRuleActionChoices.SCRIPT 

62 label = _('Script') 

63 description = _('Execute a custom script') 

64 object_model = Script 

65 object_required = True 

66 

67 def enqueue(self, *, event_rule, event_context, action_object, action_data): 

68 # Resolve the script from action parameters 

69 script = action_object.python_class() 

70 

71 # Enqueue a Job to record the script's execution 

72 from extras.jobs import ScriptJob 

73 

74 params = { 

75 'instance': action_object, 

76 'name': script.name, 

77 'user': event_context['user'], 

78 'data': action_data, 

79 'notifications': script.notifications_default, 

80 'job_timeout': script.job_timeout, 

81 } 

82 if 'snapshots' in event_context: 

83 params['snapshots'] = event_context['snapshots'] 

84 if 'request' in event_context: 

85 params['request'] = copy_safe_request(event_context['request'], include_files=False) 

86 

87 # Enqueue the job. If the script's Meta configuration is invalid (see #22872), log the error and skip this 

88 # action rather than allowing the exception to abort the event pipeline (and, since events are processed 

89 # in-request, the originating object change). Note this is intentionally asymmetric with the webhook 

90 # action, which lets enqueue failures propagate: script Meta is validated eagerly at enqueue and a 

91 # misconfigured script must not take down an unrelated object change. 

92 try: 

93 ScriptJob.enqueue(**params) 

94 except ValidationError as e: 

95 logger.error( 

96 "Skipping script action for event rule %s: invalid script configuration: %s", 

97 event_rule, '; '.join(e.messages) 

98 ) 

99 

100 def resolve_import_object(self, value): 

101 from extras.scripts import get_module_and_script 

102 module_name, script_name = value.split('.', 1) 

103 return get_module_and_script(module_name, script_name)[1] 

104 

105 

106class NotificationAction(EventRuleAction): 

107 slug = EventRuleActionChoices.NOTIFICATION 

108 label = _('Notification') 

109 description = _('Generate a notification for one or more users or groups') 

110 object_model = NotificationGroup 

111 object_required = True 

112 

113 def enqueue(self, *, event_rule, event_context, action_object, action_data): 

114 # Bulk-create notifications for all members of the notification group 

115 action_object.notify( 

116 object_type=event_context['object_type'], 

117 object_id=action_data['id'], 

118 object_repr=action_data.get('display'), 

119 event_type=event_context['event_type'], 

120 ) 

121 

122 def resolve_import_object(self, value): 

123 return NotificationGroup.objects.get(name=value)