--- 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", + ]