fix: TA tokenless uploads

Author: joseph-sentry

codecov_auth/authentication/repo_auth.py

@@ -338,6 +338,49 @@ def authenticate(
             raise exceptions.AuthenticationFailed(self.auth_failed_message)
 
 
+class TestAnalyticsTokenlessAuthentication(TokenlessAuthentication):
+    def _get_info_from_request_path(
+        self, request: HttpRequest
+    ) -> tuple[Repository, str | None]:
+        try:
+            body = json.loads(str(request.body, "utf8"))
+
+            # Validate provider
+            service_enum = Service(body.get("service"))
+
+            # Validate that next group exists and decode slug
+            repo = get_repository_from_string(service_enum, body.get("slug"))
+            if repo is None:
+                # Purposefully using the generic message so that we don't tell that
+                # we don't have a certain repo
+                raise exceptions.AuthenticationFailed(self.auth_failed_message)
+
+            return repo, body.get("commit")
+        except json.JSONDecodeError:
+            # Validate request body format
+            raise exceptions.AuthenticationFailed(self.auth_failed_message)
+        except ValueError:
+            # Validate provider
+            raise exceptions.AuthenticationFailed(self.auth_failed_message)
+
+    def get_branch(
+        self,
+        request: HttpRequest,
+        repoid: Optional[int] = None,
+        commitid: Optional[str] = None,
+    ) -> str:
+        body = json.loads(str(request.body, "utf8"))
+
+        # If commit is not created yet (ie first upload for this commit), we just validate branch format.
+        # However, if a commit exists already (ie not the first upload for this commit), we must additionally
+        # validate the saved commit branch matches what is requested in this upload call.
+        commit = Commit.objects.filter(repository_id=repoid, commitid=commitid).first()
+        if commit and commit.branch != body.get("branch"):
+            raise exceptions.AuthenticationFailed(self.auth_failed_message)
+
+        return body.get("branch")
+
+
 class BundleAnalysisTokenlessAuthentication(TokenlessAuthentication):
     def _get_info_from_request_path(
         self, request: HttpRequest

upload/tests/views/test_test_results.py

@@ -447,3 +447,157 @@ def test_upload_test_results_file_not_found(db, client, mocker, mock_redis):
         arguments=args,
         countdown=4,
     )
+
+
+def test_upload_test_results_tokenless_authentication(db, client, mocker, mock_redis):
+    upload = mocker.patch.object(TaskService, "upload")
+    _ = mocker.patch(
+        "shared.storage.MinioStorageService.create_presigned_put",
+        return_value="test-presigned-put",
+    )
+
+    owner = OwnerFactory(service="github", username="codecov")
+    repository = RepositoryFactory.create(author=owner, private=False)
+    commit_sha = "6fd5b89357fc8cdf34d6197549ac7c6d7e5977ef"
+
+    client = APIClient()
+
+    res = client.post(
+        reverse("upload-test-results"),
+        {
+            "commit": commit_sha,
+            "slug": f"{repository.author.username}::::{repository.name}",
+            "build": "test-build",
+            "buildURL": "test-build-url",
+            "job": "test-job",
+            "service": "github",
+            "branch": "fork:branch",
+        },
+        format="json",
+        headers={"User-Agent": "codecov-cli/0.4.7"},
+    )
+    assert res.status_code == 201
+
+    assert res.data == {"raw_upload_location": "test-presigned-put"}
+
+    commit = Commit.objects.get(commitid=commit_sha)
+    assert commit
+    assert commit.branch is not None
+
+    # triggers upload task
+    upload.assert_called_with(
+        commitid=commit_sha,
+        repoid=repository.repoid,
+        report_code=None,
+        report_type="test_results",
+        arguments=mocker.ANY,
+        countdown=4,
+    )
+
+
+def test_upload_test_results_tokenless_authentication_private_repo(
+    db, client, mocker, mock_redis
+):
+    _ = mocker.patch.object(TaskService, "upload")
+    _ = mocker.patch(
+        "shared.storage.MinioStorageService.create_presigned_put",
+        return_value="test-presigned-put",
+    )
+
+    owner = OwnerFactory(service="github", username="codecov")
+    repository = RepositoryFactory.create(author=owner, private=True)
+    commit_sha = "6fd5b89357fc8cdf34d6197549ac7c6d7e5977ef"
+
+    client = APIClient()
+
+    res = client.post(
+        reverse("upload-test-results"),
+        {
+            "commit": commit_sha,
+            "slug": f"{repository.author.username}::::{repository.name}",
+            "build": "test-build",
+            "buildURL": "test-build-url",
+            "job": "test-job",
+            "service": "github",
+            "branch": "fork:branch",
+        },
+        format="json",
+        headers={"User-Agent": "codecov-cli/0.4.7"},
+    )
+    assert res.status_code == 401
+    assert res.json().get("detail") == "Not valid tokenless upload"
+
+
+def test_upload_test_results_tokenless_authentication_invalid_branch(
+    db, client, mocker, mock_redis
+):
+    _ = mocker.patch.object(TaskService, "upload")
+    _ = mocker.patch(
+        "shared.storage.MinioStorageService.create_presigned_put",
+        return_value="test-presigned-put",
+    )
+
+    owner = OwnerFactory(service="github", username="codecov")
+    owner.upload_token_required_for_public_repos = True
+    owner.save()
+
+    repository = RepositoryFactory.create(author=owner, private=False)
+    commit_sha = "6fd5b89357fc8cdf34d6197549ac7c6d7e5977ef"
+
+    client = APIClient()
+
+    res = client.post(
+        reverse("upload-test-results"),
+        {
+            "commit": commit_sha,
+            "slug": f"{repository.author.username}::::{repository.name}",
+            "build": "test-build",
+            "buildURL": "test-build-url",
+            "job": "test-job",
+            "service": "github",
+            "branch": "branch",
+        },
+        format="json",
+        headers={"User-Agent": "codecov-cli/0.4.7"},
+    )
+    assert res.status_code == 401
+    assert res.json().get("detail") == "Not valid tokenless upload"
+
+
+def test_upload_test_results_tokenless_authentication_invalid_branch_existing_commits(
+    db, client, mocker, mock_redis
+):
+    _ = mocker.patch.object(TaskService, "upload")
+    _ = mocker.patch(
+        "shared.storage.MinioStorageService.create_presigned_put",
+        return_value="test-presigned-put",
+    )
+
+    owner = OwnerFactory(service="github", username="codecov")
+    owner.upload_token_required_for_public_repos = True
+    owner.save()
+
+    repository = RepositoryFactory.create(author=owner, private=False)
+    commit_sha = "6fd5b89357fc8cdf34d6197549ac7c6d7e5977ef"
+    CommitFactory.create(repository=repository, commitid=commit_sha, branch="branch")
+    client = APIClient()
+
+    res = client.post(
+        reverse("upload-test-results"),
+        {
+            "commit": commit_sha,
+            "slug": f"{repository.author.username}::::{repository.name}",
+            "build": "test-build",
+            "buildURL": "test-build-url",
+            "job": "test-job",
+            "service": "github",
+            "branch": "branch",
+        },
+        format="json",
+        headers={"User-Agent": "codecov-cli/0.4.7"},
+    )
+    assert res.status_code == 401
+    assert res.json().get("detail") == "Not valid tokenless upload"
+    commit = Commit.objects.get(commitid=commit_sha)
+    assert commit
+    assert commit.branch == "branch"

upload/views/test_results.py

@@ -14,7 +14,7 @@
     GitHubOIDCTokenAuthentication,
     OrgLevelTokenAuthentication,
     RepositoryLegacyTokenAuthentication,
-    TokenlessAuthentication,
+    TestAnalyticsTokenlessAuthentication,
     UploadTokenRequiredGetFromBodyAuthenticationCheck,
     repo_auth_custom_exception_handler,
 )
@@ -64,7 +64,7 @@ class TestResultsView(
         OrgLevelTokenAuthentication,
         GitHubOIDCTokenAuthentication,
         RepositoryLegacyTokenAuthentication,
-        TokenlessAuthentication,
+        TestAnalyticsTokenlessAuthentication,
     ]
 
     def get_exception_handler(self):

upload/views/uploads.py

@@ -207,6 +207,7 @@ class UploadViews(ListCreateAPIView, GetterMixin):
         RepositoryLegacyTokenAuthentication,
         TokenlessAuthentication,
     ]
+
     throttle_classes = [UploadsPerCommitThrottle, UploadsPerWindowThrottle]
 
     def get_exception_handler(self) -> Callable[[Exception, Dict[str, Any]], Response]: