245 lines
9.1 KiB
Diff
245 lines
9.1 KiB
Diff
--- src/firefighter/raid/serializers.py
|
|
@@ -1,7 +1,10 @@
|
|
from __future__ import annotations
|
|
|
|
+import ipaddress
|
|
import logging
|
|
+import socket
|
|
from typing import TYPE_CHECKING, Any
|
|
+from urllib.parse import urlparse
|
|
|
|
from django.conf import settings
|
|
from django.core.cache import cache
|
|
@@ -56,6 +59,58 @@
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
+ATTACHMENT_MAX_COUNT = 10
|
|
+ATTACHMENT_URL_MAX_LENGTH = 2048
|
|
+ATTACHMENT_ALLOWED_SCHEMES = frozenset({"http", "https"})
|
|
+
|
|
+
|
|
+def parse_attachment_urls(raw: str | None) -> list[str]:
|
|
+ """Normalise the attachments payload sent by Landbot into a list of URLs.
|
|
+
|
|
+ Landbot historically sends a Python-stringified list (e.g. ``"['https://a', 'https://b']"``)
|
|
+ rather than a JSON array. This helper tolerates that legacy format along with
|
|
+ a plain comma-separated string or a single URL.
|
|
+ """
|
|
+ if not raw:
|
|
+ return []
|
|
+ stripped = raw.replace("[", "").replace("]", "").replace("'", "").replace('"', "")
|
|
+ return [item.strip() for item in stripped.split(",") if item.strip()]
|
|
+
|
|
+
|
|
+def _validate_attachment_url(url: str) -> None:
|
|
+ if len(url) > ATTACHMENT_URL_MAX_LENGTH:
|
|
+ msg = f"Attachment URL exceeds {ATTACHMENT_URL_MAX_LENGTH} characters."
|
|
+ raise serializers.ValidationError(msg)
|
|
+ parsed = urlparse(url)
|
|
+ if parsed.scheme not in ATTACHMENT_ALLOWED_SCHEMES:
|
|
+ msg = f"Attachment URL scheme '{parsed.scheme}' is not allowed."
|
|
+ raise serializers.ValidationError(msg)
|
|
+ host = parsed.hostname
|
|
+ if not host:
|
|
+ raise serializers.ValidationError("Attachment URL is missing a host.")
|
|
+ try:
|
|
+ addr_infos = socket.getaddrinfo(host, None)
|
|
+ except socket.gaierror as err:
|
|
+ msg = f"Attachment URL host '{host}' could not be resolved."
|
|
+ raise serializers.ValidationError(msg) from err
|
|
+ # SSRF guard: reject any host resolving to a non-routable address so the
|
|
+ # fetch in add_attachments_to_issue can never reach internal services
|
|
+ # (cloud metadata endpoint, RFC1918 networks, loopback).
|
|
+ for info in addr_infos:
|
|
+ ip = ipaddress.ip_address(info[4][0])
|
|
+ if (
|
|
+ ip.is_private
|
|
+ or ip.is_loopback
|
|
+ or ip.is_link_local
|
|
+ or ip.is_reserved
|
|
+ or ip.is_multicast
|
|
+ or ip.is_unspecified
|
|
+ ):
|
|
+ raise serializers.ValidationError(
|
|
+ "Attachment URL host resolves to a private, loopback or link-local address."
|
|
+ )
|
|
+
|
|
+
|
|
class IgnoreEmptyStringListField(serializers.ListField):
|
|
def to_internal_value(self, data: list[Any] | Any) -> list[str]:
|
|
# Check if data is a list
|
|
@@ -186,6 +241,8 @@ class LandbotIssueRequestSerializer(serializers.ModelSerializer[JiraTicket]):
|
|
required=False,
|
|
allow_blank=True,
|
|
allow_null=True,
|
|
+ default="",
|
|
+ max_length=ATTACHMENT_URL_MAX_LENGTH * ATTACHMENT_MAX_COUNT,
|
|
)
|
|
issue_type = serializers.ChoiceField(
|
|
required=False,
|
|
@@ -211,6 +268,15 @@ def validate_environments(self, value: list[str] | None) -> list[str] | Any:
|
|
return self.fields["environments"].default
|
|
return value
|
|
|
|
+ def validate_attachments(self, value: str | None) -> str:
|
|
+ urls = parse_attachment_urls(value)
|
|
+ if len(urls) > ATTACHMENT_MAX_COUNT:
|
|
+ msg = f"Too many attachments (max {ATTACHMENT_MAX_COUNT})."
|
|
+ raise serializers.ValidationError(msg)
|
|
+ for url in urls:
|
|
+ _validate_attachment_url(url)
|
|
+ return value or ""
|
|
+
|
|
def create(self, validated_data: dict[str, Any]) -> JiraTicket:
|
|
reporter_email: str = validated_data["reporter_email"]
|
|
|
|
@@ -251,17 +317,8 @@ def create(self, validated_data: dict[str, Any]) -> JiraTicket:
|
|
if issue_id is None:
|
|
logger.error("Could not create Jira ticket")
|
|
raise JiraAPIError("Could not create Jira ticket")
|
|
- if validated_data["attachments"] is not None:
|
|
- # Prepare attachments and double check we don't have any empty strings
|
|
- attachments: list[str] = [
|
|
- a
|
|
- for a in validated_data["attachments"]
|
|
- .replace("[", "")
|
|
- .replace("]", "")
|
|
- .replace("'", "")
|
|
- .split(", ")
|
|
- if a
|
|
- ]
|
|
+ attachments = parse_attachment_urls(validated_data.get("attachments"))
|
|
+ if attachments:
|
|
jira_client.add_attachments_to_issue(issue_id, attachments)
|
|
|
|
jira_ticket = JiraTicket.objects.create(**issue)
|
|
|
|
--- src/firefighter/raid/views/__init__.py
|
|
@@ -9,6 +9,7 @@
|
|
from rest_framework.renderers import JSONRenderer
|
|
from rest_framework.response import Response
|
|
|
|
+from firefighter.api.authentication import BearerTokenAuthentication
|
|
from firefighter.raid.models import JiraTicket
|
|
from firefighter.raid.serializers import (
|
|
JiraWebhookCommentSerializer,
|
|
@@ -73,7 +74,8 @@ class CreateJiraBotView(
|
|
"jiraissue_ptr",
|
|
)
|
|
serializer_class = LandbotIssueRequestSerializer
|
|
- permission_classes = [permissions.AllowAny]
|
|
+ authentication_classes = [BearerTokenAuthentication]
|
|
+ permission_classes = [permissions.IsAuthenticated]
|
|
renderer_classes = [JSONRenderer]
|
|
|
|
def post(self, request: Request, *args: Never, **kwargs: Never) -> Response:
|
|
|
|
--- tests/test_raid/test_raid_security.py
|
|
@@ -0,0 +1,106 @@
|
|
+"""Security regression tests for the raid endpoints.
|
|
+
|
|
+Covers the two vulnerabilities that motivated this module:
|
|
+ - `POST /api/v2/firefighter/raid/jira_bot` must require authentication.
|
|
+ - The `attachments` field must reject URLs pointing at private/link-local
|
|
+ networks so the server cannot be coerced into fetching cloud metadata
|
|
+ or internal services on behalf of the caller.
|
|
+"""
|
|
+
|
|
+from __future__ import annotations
|
|
+
|
|
+import pytest
|
|
+from rest_framework import status
|
|
+from rest_framework.test import APIClient
|
|
+
|
|
+from firefighter.incidents.factories import UserFactory
|
|
+from firefighter.raid.serializers import parse_attachment_urls
|
|
+
|
|
+JIRA_BOT_URL = "/api/v2/firefighter/raid/jira_bot"
|
|
+
|
|
+BASE_PAYLOAD: dict[str, object] = {
|
|
+ "summary": "sec-test",
|
|
+ "description": "sec-test",
|
|
+ "seller_contract_id": "",
|
|
+ "zoho": "",
|
|
+ "platform": "FR",
|
|
+ "reporter_email": "sec-test@manomano.com",
|
|
+ "incident_category": "",
|
|
+ "suggested_team_routing": "x",
|
|
+ "priority": 4,
|
|
+ "business_impact": "Low",
|
|
+ "issue_type": "Incident",
|
|
+}
|
|
+
|
|
+
|
|
+@pytest.mark.django_db
|
|
+class TestJiraBotAuthentication:
|
|
+ def test_anonymous_post_is_rejected(self) -> None:
|
|
+ client = APIClient()
|
|
+ response = client.post(JIRA_BOT_URL, data=BASE_PAYLOAD, format="json")
|
|
+ assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
|
+
|
|
+
|
|
+@pytest.mark.django_db
|
|
+class TestJiraBotAttachmentsSSRF:
|
|
+ @pytest.mark.parametrize(
|
|
+ "attachments",
|
|
+ [
|
|
+ "http://127.0.0.1/leak.png",
|
|
+ "http://169.254.169.254/latest/meta-data/iam/security-credentials/",
|
|
+ "http://10.0.0.1/leak.png",
|
|
+ "http://192.168.1.1/leak.png",
|
|
+ "http://[::1]/leak.png",
|
|
+ "['http://169.254.169.254/leak']",
|
|
+ ],
|
|
+ )
|
|
+ def test_private_or_link_local_hosts_are_rejected(
|
|
+ self, attachments: str
|
|
+ ) -> None:
|
|
+ client = APIClient()
|
|
+ client.force_authenticate(user=UserFactory.create())
|
|
+ payload = {**BASE_PAYLOAD, "attachments": attachments}
|
|
+ response = client.post(JIRA_BOT_URL, data=payload, format="json")
|
|
+ assert response.status_code == status.HTTP_400_BAD_REQUEST
|
|
+ body = response.json()
|
|
+ assert any(
|
|
+ err.get("attr") == "attachments" for err in body.get("errors", [])
|
|
+ ), body
|
|
+
|
|
+ @pytest.mark.parametrize(
|
|
+ "attachments",
|
|
+ [
|
|
+ "file:///etc/passwd",
|
|
+ "gopher://attacker.example/exploit",
|
|
+ ],
|
|
+ )
|
|
+ def test_non_http_schemes_are_rejected(self, attachments: str) -> None:
|
|
+ client = APIClient()
|
|
+ client.force_authenticate(user=UserFactory.create())
|
|
+ payload = {**BASE_PAYLOAD, "attachments": attachments}
|
|
+ response = client.post(JIRA_BOT_URL, data=payload, format="json")
|
|
+ assert response.status_code == status.HTTP_400_BAD_REQUEST
|
|
+ body = response.json()
|
|
+ assert any(
|
|
+ err.get("attr") == "attachments" for err in body.get("errors", [])
|
|
+ ), body
|
|
+
|
|
+
|
|
+class TestParseAttachmentUrls:
|
|
+ """Parser must tolerate absent, empty and Landbot's legacy list-as-string payloads."""
|
|
+
|
|
+ @pytest.mark.parametrize("raw", [None, "", " "])
|
|
+ def test_empty_returns_empty_list(self, raw: str | None) -> None:
|
|
+ assert parse_attachment_urls(raw) == []
|
|
+
|
|
+ def test_single_url(self) -> None:
|
|
+ assert parse_attachment_urls("https://example.com/a.png") == [
|
|
+ "https://example.com/a.png"
|
|
+ ]
|
|
+
|
|
+ def test_legacy_stringified_list_from_landbot(self) -> None:
|
|
+ raw = "['https://example.com/a.png', 'https://example.com/b.png']"
|
|
+ assert parse_attachment_urls(raw) == [
|
|
+ "https://example.com/a.png",
|
|
+ "https://example.com/b.png",
|
|
+ ]
|