feat(codecov): initial client for requests to codecov api

src/sentry/codecov/client.py

@@ -0,0 +1,129 @@
+import datetime
+import logging
+from enum import StrEnum
+from typing import TypeAlias
+
+import requests
+from rest_framework import status
+
+from sentry import options
+from sentry.api.exceptions import SentryAPIException
+from sentry.utils import jwt
+
+GitProviderId: TypeAlias = str
+
+
+class GitProvider(StrEnum):
+    """
+    Enum representing the Git provider that hosts the user/org that a
+    `CodecovApiClient` instance is acting on behalf of.
+
+    Codecov doesn't require this to be GitHub, but that's all that's implemented
+    for now.
+    """
+
+    GitHub = "github"
+
+
+logger = logging.getLogger(__name__)
+
+TIMEOUT_SECONDS = 10
+
+
+class ConfigurationError(SentryAPIException):
+    status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
+    code = "configuration-error"
+
+
+class CodecovApiClient:
+    """
+    Thin client for making JWT-authenticated requests to the Codecov API.
+
+    For each request, Sentry creates and signs (HS256) a JWT with a key shared
+    with Codecov. This JWT contains information that Codecov needs to service
+    the request.
+    """
+
+    def _create_jwt(self):
+        now = int(datetime.datetime.now(datetime.UTC).timestamp())
+        exp = now + 300  # 5 minutes
+        claims = {
+            "iss": "https://sentry.io",
+            "iat": now,
+            "exp": exp,
+        }
+        claims.update(self.custom_claims)
+
+        return jwt.encode(claims, self.signing_secret, algorithm="HS256")
+
+    def __init__(
+        self,
+        git_provider_user: GitProviderId,
+        git_provider: GitProvider = GitProvider.GitHub,
+    ):
+        """
+        Creates a `CodecovApiClient`.
+
+        :param git_provider_user: The ID of the current Sentry user's linked git
+           provider account, according to the git provider.
+        :param git_provider: The git provider that the above user's account is
+           hosted on.
+        """
+
+        if not (base_url := options.get("codecov.base-url")):
+            raise ConfigurationError()
+
+        if not (signing_secret := options.get("codecov.api-bridge-signing-secret")):
+            raise ConfigurationError()
+
+        self.base_url = base_url
+        self.signing_secret = signing_secret
+        self.custom_claims = {
+            "g_u": git_provider_user,
+            "g_p": git_provider,
+        }
+
+    def get(self, endpoint: str, params=None, headers=None) -> requests.Response | None:
+        """
+        Makes a GET request to the specified endpoint of the configured Codecov
+        API host with the provided params and headers.
+
+        :param endpoint: The endpoint to request, without the host portion. For
+           examples: `/api/v2/gh/getsentry/users` or `/graphql`
+        :param params: Dictionary of query params.
+        :param headers: Dictionary of request headers.
+        """
+        headers = headers or {}
+        token = self._create_jwt()
+        headers.update(jwt.authorization_header(token))
+
+        url = f"{self.base_url}{endpoint}"
+        try:
+            response = requests.get(url, params=params, headers=headers, timeout=TIMEOUT_SECONDS)
+        except Exception:
+            logger.exception("Error when making GET request")
+            return None
+
+        return response
+
+    def post(self, endpoint: str, data=None, headers=None) -> requests.Response | None:
+        """
+        Makes a POST request to the specified endpoint of the configured Codecov
+        API host with the provided data and headers.
+
+        :param endpoint: The endpoint to request, without the host portion. For
+           examples: `/api/v2/gh/getsentry/users` or `/graphql`
+        :param data: Dictionary of form data.
+        :param headers: Dictionary of request headers.
+        """
+        headers = headers or {}
+        token = self._create_jwt()
+        headers.update(jwt.authorization_header(token))
+        url = f"{self.base_url}{endpoint}"
+        try:
+            response = requests.post(url, data=data, headers=headers, timeout=TIMEOUT_SECONDS)
+        except Exception:
+            logger.exception("Error when making POST request")
+            return None
+
+        return response

src/sentry/options/defaults.py

@@ -602,6 +602,8 @@
 
 # Codecov Integration
 register("codecov.client-secret", flags=FLAG_CREDENTIAL | FLAG_PRIORITIZE_DISK)
+register("codecov.base-url", default="https://api.codecov.io")
+register("codecov.api-bridge-signing-secret", flags=FLAG_CREDENTIAL | FLAG_PRIORITIZE_DISK)
 
 # GitHub Integration
 register("github-app.id", default=0, flags=FLAG_AUTOMATOR_MODIFIABLE)

tests/sentry/codecov/test_client.py

@@ -0,0 +1,105 @@
+import datetime
+from unittest.mock import patch
+
+import pytest
+
+from sentry.codecov.client import CodecovApiClient, ConfigurationError, GitProvider
+from sentry.testutils.cases import TestCase
+from sentry.testutils.silo import all_silo_test
+from sentry.utils import jwt
+
+
+@all_silo_test
+class TestCodecovApiClient(TestCase):
+    def setUp(self):
+        self.test_git_provider_user = "12345"
+        self.test_base_url = "http://example.com"
+        self.test_secret = "test-secret-" + "a" * 20
+
+        self.test_timestamp = datetime.datetime.now(datetime.UTC)
+        self._mock_now = patch("datetime.datetime.now", return_value=self.test_timestamp)
+
+        with self.options(
+            {
+                "codecov.base-url": self.test_base_url,
+                "codecov.api-bridge-signing-secret": self.test_secret,
+            }
+        ):
+            self.codecov_client = CodecovApiClient(self.test_git_provider_user)
+
+    def test_raises_configuration_error_without_base_url(self):
+        with self.options(
+            {
+                "codecov.base-url": None,
+                "codecov.api-bridge-signing-secret": self.test_secret,
+            }
+        ):
+            with pytest.raises(ConfigurationError):
+                CodecovApiClient(self.test_git_provider_user)
+
+    def test_raises_configuration_error_without_signing_secret(self):
+        with self.options(
+            {
+                "codecov.base-url": self.test_base_url,
+                "codecov.api-bridge-signing-secret": None,
+            }
+        ):
+            with pytest.raises(ConfigurationError):
+                CodecovApiClient(self.test_git_provider_user)
+
+    def test_creates_valid_jwt(self):
+        encoded_jwt = self.codecov_client._create_jwt()
+
+        header = jwt.peek_header(encoded_jwt)
+        assert header == {
+            "typ": "JWT",
+            "alg": "HS256",
+        }
+
+        # Ensure the claims are what we expect, separate from verifying the
+        # signature and standard claims
+        claims = jwt.peek_claims(encoded_jwt)
+        expected_iat = int(self.test_timestamp.timestamp())
+        expected_exp = expected_iat + 300
+        assert claims == {
+            "g_u": self.test_git_provider_user,
+            "g_p": GitProvider.GitHub.value,
+            "iss": "https://sentry.io",
+            "iat": expected_iat,
+            "exp": expected_exp,
+        }
+
+        # Ensure we can verify the signature and whatall
+        jwt.decode(encoded_jwt, self.test_secret)
+
+    @patch("requests.get")
+    def test_sends_get_request_with_jwt_auth_header(self, mock_get):
+        with patch.object(self.codecov_client, "_create_jwt", return_value="test"):
+            self.codecov_client.get(
+                "/example/endpoint", {"example-param": "foo"}, {"X_TEST_HEADER": "bar"}
+            )
+            mock_get.assert_called_once_with(
+                "http://example.com/example/endpoint",
+                params={"example-param": "foo"},
+                headers={
+                    "Authorization": "Bearer test",
+                    "X_TEST_HEADER": "bar",
+                },
+                timeout=10,
+            )
+
+    @patch("requests.post")
+    def test_sends_post_request_with_jwt_auth_header(self, mock_post):
+        with patch.object(self.codecov_client, "_create_jwt", return_value="test"):
+            self.codecov_client.post(
+                "/example/endpoint", {"example-param": "foo"}, {"X_TEST_HEADER": "bar"}
+            )
+            mock_post.assert_called_once_with(
+                "http://example.com/example/endpoint",
+                data={"example-param": "foo"},
+                headers={
+                    "Authorization": "Bearer test",
+                    "X_TEST_HEADER": "bar",
+                },
+                timeout=10,
+            )