Coverage for extras/webhooks.py: 0%

1import hashlib 

2import hmac 

3import logging 

4 

5import requests 

6from django.conf import settings 

7from django_rq import job 

8from jinja2.exceptions import TemplateError 

9 

10from netbox.registry import registry 

11from netbox.settings_utils import parse_job_timeout 

12from utilities.proxy import resolve_proxies 

13from utilities.request import get_safe_request_context 

14 

15from .constants import WEBHOOK_EVENT_TYPES 

16 

17__all__ = ( 

18 'generate_signature', 

19 'register_webhook_callback', 

20 'send_webhook', 

21) 

22 

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

24 

25 

26def register_webhook_callback(func): 

27 """ 

28 Register a function as a webhook callback. 

29 """ 

30 registry['webhook_callbacks'].append(func) 

31 logger.debug(f'Registered webhook callback {func.__module__}.{func.__name__}') 

32 return func 

33 

34 

35def generate_signature(request_body, secret): 

36 """ 

37 Return a cryptographic signature that can be used to verify the authenticity of webhook data. 

38 """ 

39 hmac_prep = hmac.new( 

40 key=secret.encode('utf8'), 

41 msg=request_body, 

42 digestmod=hashlib.sha512 

43 ) 

44 return hmac_prep.hexdigest() 

45 

46 

47@job('default') 

48def send_webhook(event_rule, object_type, event_type, data, timestamp, request=None, snapshots=None): 

49 """ 

50 Make a POST request to the defined Webhook 

51 """ 

52 webhook = event_rule.action_object 

53 

54 # Prepare context data for headers & body templates 

55 context = { 

56 'event': WEBHOOK_EVENT_TYPES.get(event_type, event_type), 

57 'timestamp': timestamp, 

58 'object_type': '.'.join(object_type.natural_key()), 

59 'data': data, 

60 } 

61 if request: 

62 context['request'] = get_safe_request_context(request) 

63 if snapshots: 

64 context['snapshots'] = snapshots 

65 

66 # Add any additional context from plugins 

67 callback_data = {} 

68 for callback in registry['webhook_callbacks']: 

69 try: 

70 if ret := callback(object_type, event_type, data, request): 

71 callback_data.update(**ret) 

72 except Exception as e: 

73 logger.warning(f"Caught exception when processing callback {callback}: {e}") 

74 pass 

75 if callback_data: 

76 context['context'] = callback_data 

77 

78 # Build the headers for the HTTP request 

79 headers = { 

80 'Content-Type': webhook.http_content_type, 

81 } 

82 try: 

83 headers.update(webhook.render_headers(context)) 

84 except (TemplateError, ValueError) as e: 

85 logger.error(f"Error parsing HTTP headers for webhook {webhook}: {e}") 

86 raise e 

87 

88 # Render the request body 

89 try: 

90 body = webhook.render_body(context) 

91 except TemplateError as e: 

92 logger.error(f"Error rendering request body for webhook {webhook}: {e}") 

93 raise e 

94 

95 # Prepare the HTTP request 

96 url = webhook.render_payload_url(context) 

97 params = { 

98 'method': webhook.http_method, 

99 'url': url, 

100 'headers': headers, 

101 'data': body.encode('utf8'), 

102 } 

103 logger.info( 

104 f"Sending {params['method']} request to {params['url']} ({context['object_type']} {context['event']})" 

105 ) 

106 logger.debug(params) 

107 try: 

108 prepared_request = requests.Request(**params).prepare() 

109 except requests.exceptions.RequestException as e: 

110 logger.error(f"Error forming HTTP request: {e}") 

111 raise e 

112 

113 # If a secret key is defined, sign the request with a hash of the key and its content 

114 if webhook.secret != '': 

115 prepared_request.headers['X-Hook-Signature'] = generate_signature(prepared_request.body, webhook.secret) 

116 

117 # Determine the request timeout, preferring the webhook-specific value over the global default 

118 timeout = webhook.timeout if webhook.timeout is not None else settings.WEBHOOK_DEFAULT_TIMEOUT 

119 

120 # Webhook.clean() enforces this when the webhook is saved, but RQ_DEFAULT_TIMEOUT may have been lowered since. 

121 job_timeout = parse_job_timeout(settings.RQ_DEFAULT_TIMEOUT) 

122 if job_timeout is not None and timeout >= job_timeout: 

123 logger.warning( 

124 f"Webhook timeout ({timeout} seconds) is not less than the background job timeout ({job_timeout} " 

125 f"seconds); the job may be terminated before the request can time out." 

126 ) 

127 

128 # Send the request 

129 with requests.Session() as session: 

130 session.verify = webhook.ssl_verification 

131 if webhook.ca_file_path: 

132 session.verify = webhook.ca_file_path 

133 proxies = resolve_proxies(url=url, context={'client': webhook}) 

134 try: 

135 response = session.send(prepared_request, proxies=proxies, timeout=timeout) 

136 except requests.exceptions.Timeout: 

137 logger.error(f"Request to {url} timed out after {timeout} seconds") 

138 raise 

139 

140 if 200 <= response.status_code <= 299: 

141 logger.info(f"Request succeeded; response status {response.status_code}") 

142 return f"Status {response.status_code} returned, webhook successfully processed." 

143 logger.warning(f"Request failed; response status {response.status_code}: {response.content}") 

144 raise requests.exceptions.RequestException( 

145 f"Status {response.status_code} returned with content '{response.content}', webhook FAILED to process." 

146 )