Fix CodeQL Alert for SSRF

videos/constants.py

@@ -14,6 +14,9 @@
 PDF_FORMAT_ID = 46
 WEBVTT_FORMAT_ID = 51
 
+TRANSCODE_JOB_SUBSCRIPTION_URL = "https://sns.{AWS_REGION}.amazonaws.com/?Action=ConfirmSubscription&TopicArn=arn:aws:sns:{AWS_REGION}:{AWS_ACCOUNT_ID}:MediaConvertJobAlert&Token={TOKEN}"
+EMPTY_TOKEN_MSG = "Token Cannot be empty!"  # noqa: S105
+
 
 class VideoStatus:
     """Simple class for possible VideoFile statuses"""

videos/exceptions.py

@@ -0,0 +1,12 @@
+"""Videos Exceptions"""
+
+from rest_framework import status
+from rest_framework.exceptions import APIException
+
+
+class BadRequest(APIException):
+    """Exception for Invalid request data"""
+
+    status_code = status.HTTP_400_BAD_REQUEST
+    default_detail = "Bad request."
+    default_code = "bad_request"

videos/utils.py

@@ -11,6 +11,7 @@
 
 from main.s3_utils import get_boto3_resource
 from main.utils import get_dirpath_and_filename, get_file_extension, uuid_string
+from videos.constants import TRANSCODE_JOB_SUBSCRIPTION_URL
 from videos.models import WebsiteContent
 from websites.models import Website, WebsiteStarter
 
@@ -111,3 +112,13 @@ def update_metadata(source_obj, new_uid, new_s3_path):
     new_metadata["uid"] = new_uid
     new_metadata["file"] = str(new_s3_path).lstrip("/")
     return new_metadata
+
+
+def get_subscribe_url(token: str) -> str:
+    """Get a sns subscribe url"""
+
+    return TRANSCODE_JOB_SUBSCRIPTION_URL.format(
+        AWS_REGION=settings.AWS_REGION,
+        AWS_ACCOUNT_ID=settings.AWS_ACCOUNT_ID,
+        TOKEN=token,
+    )

videos/utils_test.py

@@ -10,6 +10,7 @@
     create_new_content,
     generate_s3_path,
     get_content_dirpath,
+    get_subscribe_url,
     update_metadata,
 )
 from websites.factories import (
@@ -99,3 +100,15 @@ def test_create_new_content(mocker):
     mock_copy_obj_s3.assert_called_once_with(source_content, destination_course)
     mock_get_dirpath_and_filename.assert_called_once_with("new_s3_loc")
     mock_uuid_string.assert_called_once()
+
+
+def test_get_subscribe_url(mocker):
+    """Test for get_subscribe_url method"""
+
+    mocker.patch("django.conf.settings.AWS_REGION", "us-east-1")
+    mocker.patch("django.conf.settings.AWS_ACCOUNT_ID", "1234567890")
+
+    assert (
+        get_subscribe_url("fake-token")
+        == "https://sns.us-east-1.amazonaws.com/?Action=ConfirmSubscription&TopicArn=arn:aws:sns:us-east-1:1234567890:MediaConvertJobAlert&Token=fake-token"
+    )

videos/views.py

@@ -15,9 +15,11 @@
 from rest_framework.response import Response
 
 from videos.api import update_video_job
-from videos.constants import VideoStatus
+from videos.constants import EMPTY_TOKEN_MSG, VideoStatus
+from videos.exceptions import BadRequest
 from videos.models import Video, VideoJob
 from videos.tasks import update_transcripts_for_video
+from videos.utils import get_subscribe_url
 
 log = logging.getLogger()
 
@@ -39,7 +41,13 @@ def post(
             # Confirm the subscription
             if settings.AWS_ACCOUNT_ID not in message.get("TopicArn", ""):
                 raise PermissionDenied
-            requests.get(message.get("SubscribeURL"), timeout=60)
+
+            token = message.get("Token", "")
+            if not token:
+                raise BadRequest(EMPTY_TOKEN_MSG)
+
+            subscribe_url = get_subscribe_url(token)
+            requests.get(subscribe_url, timeout=60)
         else:
             if message.get("account", "") != settings.AWS_ACCOUNT_ID:
                 raise PermissionDenied

videos/views_test.py

@@ -132,6 +132,26 @@ def test_transcode_jobs_subscribe_denied(settings, mocker, drf_client):
     mock_get.assert_not_called()
 
 
+def test_transcode_jobs_subscribe_bad_request(settings, mocker, drf_client):
+    """TranscodeJobView should deny a subscription request if token is invalid"""
+    mock_get = mocker.patch("videos.views.requests.get")
+    with open(  # noqa: PTH123
+        f"{TEST_VIDEOS_WEBHOOK_PATH}/subscribe.json", encoding="utf-8"
+    ) as infile:
+        data = json.loads(
+            infile.read()
+            .replace("AWS_ACCOUNT_ID", settings.AWS_ACCOUNT_ID)
+            .replace("AWS_REGION", settings.AWS_REGION)
+        )
+
+    # mock token
+    data["Token"] = ""
+
+    response = drf_client.post(reverse("transcode_jobs"), data=data)
+    assert response.status_code == 400
+    mock_get.assert_not_called()
+
+
 @pytest.mark.parametrize("callback_key", [None, "callback_key", "different_key"])
 @pytest.mark.parametrize("video_status", ["submitted_for_transcription", "complete"])
 def test_transcript_job(mocker, video_status, callback_key, drf_client, settings):