Prevent local webhook requests
parent
221211e716
commit
178ab9ec43
|
@ -30,6 +30,7 @@ from .serializers import (EpicSerializer, EpicRelatedUserStorySerializer,
|
||||||
UserStorySerializer, IssueSerializer, TaskSerializer,
|
UserStorySerializer, IssueSerializer, TaskSerializer,
|
||||||
WikiPageSerializer, MilestoneSerializer,
|
WikiPageSerializer, MilestoneSerializer,
|
||||||
HistoryEntrySerializer, UserSerializer)
|
HistoryEntrySerializer, UserSerializer)
|
||||||
|
from . import utils
|
||||||
from .models import WebhookLog
|
from .models import WebhookLog
|
||||||
|
|
||||||
|
|
||||||
|
@ -71,6 +72,21 @@ def _send_request(webhook_id, url, key, data):
|
||||||
"X-Hub-Signature": "sha1={}".format(signature),
|
"X-Hub-Signature": "sha1={}".format(signature),
|
||||||
"Content-Type": "application/json"
|
"Content-Type": "application/json"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
try:
|
||||||
|
utils.validate_destination_address(url)
|
||||||
|
except utils.IpaddresValueError as e:
|
||||||
|
# Error validating url
|
||||||
|
webhook_log = WebhookLog.objects.create(webhook_id=webhook_id, url=url,
|
||||||
|
status=0,
|
||||||
|
request_data=data,
|
||||||
|
request_headers=dict(),
|
||||||
|
response_data="error-in-request: {}".format(
|
||||||
|
str(e)),
|
||||||
|
response_headers={},
|
||||||
|
duration=0)
|
||||||
|
return webhook_log
|
||||||
|
|
||||||
request = requests.Request('POST', url, data=serialized_data, headers=headers)
|
request = requests.Request('POST', url, data=serialized_data, headers=headers)
|
||||||
prepared_request = request.prepare()
|
prepared_request = request.prepare()
|
||||||
|
|
||||||
|
|
|
@ -0,0 +1,21 @@
|
||||||
|
import ipaddress
|
||||||
|
import socket
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
|
|
||||||
|
class IpaddresValueError(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def validate_destination_address(url):
|
||||||
|
host = urlparse(url).hostname
|
||||||
|
port = urlparse(url).port
|
||||||
|
socket_args, _ = socket.getaddrinfo(host, port)
|
||||||
|
destination_address = socket_args[4][0]
|
||||||
|
|
||||||
|
try:
|
||||||
|
ipa = ipaddress.ip_address(destination_address)
|
||||||
|
except ValueError:
|
||||||
|
raise IpaddresValueError(_("IP Address error"))
|
||||||
|
if ipa.is_private:
|
||||||
|
raise IpaddresValueError("Not allowed IP Address")
|
Loading…
Reference in New Issue