feat(seer-infra-telemetry): Add DatadogIdentityProvider for OAuth2

src/sentry/identity/__init__.py

@@ -11,6 +11,7 @@
 
 def _register_providers() -> None:
     from .bitbucket.provider import BitbucketIdentityProvider
+    from .datadog.provider import DatadogIdentityProvider
     from .discord.provider import DiscordIdentityProvider
     from .gcp.provider import GCPIdentityProvider
     from .github.provider import GitHubIdentityProvider
@@ -34,6 +35,7 @@ def _register_providers() -> None:
     register(GitlabIdentityProvider)
     register(GoogleIdentityProvider)
     register(DiscordIdentityProvider)
+    register(DatadogIdentityProvider)
     register(GCPIdentityProvider)
 
 

src/sentry/identity/datadog/provider.py

@@ -3,30 +3,75 @@
 import base64
 import hashlib
 import secrets
+from typing import Any
 
 import orjson
 from django.http.request import HttpRequest
 from django.http.response import HttpResponseBase
 from requests import ConnectionError, HTTPError, Response
 from requests.exceptions import SSLError
 
+from sentry.auth.exceptions import IdentityNotValid
 from sentry.http import safe_urlopen, safe_urlread
 from sentry.identity.oauth2 import (
     OAuth2CallbackView,
     OAuth2LoginView,
+    OAuth2Provider,
     _redirect_url,
     record_event,
 )
 from sentry.identity.pipeline import IdentityPipeline
+from sentry.identity.services.identity.model import RpcIdentity
+from sentry.integrations.types import IntegrationProviderSlug
 from sentry.integrations.utils.metrics import IntegrationPipelineViewType
+from sentry.pipeline.views.base import PipelineView
 from sentry.users.models.identity import Identity
 from sentry.utils.http import absolute_uri
 
+MCP_REGISTER_PATH = "/api/unstable/mcp-server/register"
+MCP_AUTHORIZE_PATH = "/api/unstable/mcp-server/authorize"
+MCP_TOKEN_PATH = "/api/unstable/mcp-server/token"
+MCP_ENDPOINT_PATH = "/api/unstable/mcp-server/mcp"
+
 
 def _basic_auth_header(client_id: str, client_secret: str) -> str:
     return "Basic " + base64.b64encode(f"{client_id}:{client_secret}".encode()).decode("ascii")
 
 
+def get_user_info(access_token: str, site: str) -> dict[str, Any]:
+    """Fetch the current Datadog user via the MCP ``datadog://mcp/whoami`` resource."""
+    url = f"https://mcp.{site}{MCP_ENDPOINT_PATH}"
+    headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
+
+    init_resp = safe_urlopen(
+        url,
+        method="POST",
+        headers=headers,
+        json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}},
+    )
+    init_resp.raise_for_status()
+    headers["Mcp-Session-Id"] = init_resp.headers["mcp-session-id"]
+
+    resp = safe_urlopen(
+        url,
+        method="POST",
+        headers=headers,
+        json={
+            "jsonrpc": "2.0",
+            "id": 2,
+            "method": "resources/read",
+            "params": {"uri": "datadog://mcp/whoami"},
+        },
+    )
+    resp.raise_for_status()
+
+    try:
+        body = orjson.loads(safe_urlread(resp))
+        return orjson.loads(body["result"]["contents"][0]["text"])
+    except (KeyError, IndexError, orjson.JSONDecodeError) as e:
+        raise IdentityNotValid("MCP whoami returned an unexpected response") from e
+
+
 def generate_pkce_code_verifier() -> str:
     return secrets.token_urlsafe(96)
 
@@ -196,3 +241,113 @@
         return safe_urlopen(
             self.access_token_url, data=data, headers=headers, verify_ssl=verify_ssl
         )
+
+
+class DatadogIdentityProvider(OAuth2Provider):
+    key = IntegrationProviderSlug.DATADOG
+    name = "Datadog"
+
+    oauth_scopes: tuple[str, ...] = (
+        "mcp_read",
+        "apm_read",
+        "error_tracking_read",
+        "events_read",
+        "hosts_read",
+        "incident_read",
+        "logs_read_data",
+        "metrics_read",
+        "monitors_read",
+    )
+
+    def _get_mcp_base_url(self) -> str:
+        return f"https://mcp.{self._get_oauth_parameter('site')}"
+
+    def get_oauth_authorize_url(self) -> str:
+        return self._get_mcp_base_url() + MCP_AUTHORIZE_PATH
+
+    def get_oauth_access_token_url(self) -> str:
+        return self._get_mcp_base_url() + MCP_TOKEN_PATH
+
+    def get_pipeline_views(self) -> list[PipelineView[IdentityPipeline]]:
+        return [
+            DatadogOAuth2DCRView(
+                register_url=self._get_mcp_base_url() + MCP_REGISTER_PATH,
+            ),
+            DatadogOAuth2LoginView(
+                authorize_url=self.get_oauth_authorize_url(),
+                scope=" ".join(self.get_oauth_scopes()),
+                resource=self._get_mcp_base_url(),
+            ),
+            DatadogOAuth2CallbackView(
+                access_token_url=self.get_oauth_access_token_url(),
+            ),
+        ]
+
+    def get_oauth_data(self, payload: dict[str, Any]) -> dict[str, Any]:
+        data = super().get_oauth_data(payload)
+        if "scope" in payload:
+            data["scope"] = payload["scope"]
+        return data
+
+    def build_identity(self, data: dict[str, Any]) -> dict[str, Any]:
+        token_data = data["data"]
+        access_token = token_data.get("access_token")
+        if not access_token:
+            raise ValueError("Datadog token exchange did not return an access_token")
+
+        user = get_user_info(access_token, self._get_oauth_parameter("site"))
+
+        oauth_data = self.get_oauth_data(token_data)
+
+        # Persist DCR credentials and site so refresh_identity can access them outside a pipeline context.
+        client_id = data.get("dcr_client_id")
+        client_secret = data.get("dcr_client_secret")
+        if not client_id or not client_secret:
+            raise IdentityNotValid("Missing DCR credentials")
+        oauth_data["client_id"] = client_id
+        oauth_data["client_secret"] = client_secret
+        oauth_data["site"] = self._get_oauth_parameter("site")
+
+        return {
+            "type": IntegrationProviderSlug.DATADOG,
+            "id": user["user_uuid"],
+            "email": user.get("user_email"),
+            "name": user.get("user_name"),
+            "scopes": [],
+            "data": oauth_data,
+        }
+
+    def get_refresh_token_url(self) -> str:
+        return self.get_oauth_access_token_url()
+
+    def get_refresh_token_headers(self) -> dict[str, str]:
+        client_id = self.config.get("client_id")
+        client_secret = self.config.get("client_secret")
+
+        if not client_id or not client_secret:
+            raise IdentityNotValid("Missing DCR credentials")
+
+        return {"Authorization": _basic_auth_header(client_id, client_secret)}
+
+    def get_refresh_token_params(
+        self, refresh_token: str, identity: Identity | RpcIdentity, **kwargs: Any
+    ) -> dict[str, str | None]:
+        return {"grant_type": "refresh_token", "refresh_token": refresh_token}
+
+    def refresh_identity(self, identity: Identity | RpcIdentity, **kwargs: Any) -> None:
+        # Add site and client credentials to config so get_refresh_token_headers
+        # and get_refresh_token_url can access them.
+
+        site = identity.data.get("site")
+        if not site:
+            raise IdentityNotValid("Missing Datadog site")
+        self.config["site"] = site
+
+        client_id = identity.data.get("client_id")
+        client_secret = identity.data.get("client_secret")
+        if not client_id or not client_secret:
+            raise IdentityNotValid("Missing DCR credentials")
+        self.config["client_id"] = client_id
+        self.config["client_secret"] = client_secret
+
+        super().refresh_identity(identity, **kwargs)

src/sentry/integrations/types.py

@@ -46,6 +46,7 @@ class IntegrationProviderSlug(StrEnum):
     PAGERDUTY = "pagerduty"
     OPSGENIE = "opsgenie"
     PERFORCE = "perforce"
+    DATADOG = "datadog"
     GCP = "gcp"