feat(jira): Add a Forge endpoint for the Jira configure link

src/sentry/integrations/jira/forge/client.py

@@ -1,37 +1,43 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from typing import Any
 
 from requests import PreparedRequest
 
 from sentry.integrations.client import ApiClient
 from sentry.integrations.utils.jira import get_jira_key
+from sentry.shared_integrations.exceptions import ApiError
 
 # Atlassian's reserved, read-only app property holding the site's Connect clientKey.
 # It stops being served at Connect end of support.
 # https://developer.atlassian.com/platform/adopting-forge-from-connect/migrate-connect-clientkey/
 CONNECT_CLIENT_KEY_PROPERTY = "connect_client_key_019cdff3-8bfb-71fe-9628-875b700aebb8"
 
 
-class ForgeInstallationClient(ApiClient):
-    """
-    Calls a site's Jira API with its Forge system token while its integration is still
-    being found or created. Once it exists, JiraCloudClient makes the calls.
-    """
+class ForgeTokenClient(ApiClient):
+    """Calls a site's Jira API through Atlassian's API gateway with a Forge OAuth token."""
 
     integration_name = "jira"
     timeout = 10
     allow_redirects = False
 
-    def __init__(self, api_base_url: str, system_token: str) -> None:
+    def __init__(self, api_base_url: str, token: str) -> None:
         self.base_url = api_base_url
-        self.system_token = system_token
+        self.token = token
         super().__init__()
 
     def finalize_request(self, prepared_request: PreparedRequest) -> PreparedRequest:
-        prepared_request.headers["Authorization"] = f"Bearer {self.system_token}"
+        prepared_request.headers["Authorization"] = f"Bearer {self.token}"
         return prepared_request
 
+
+class ForgeInstallationClient(ForgeTokenClient):
+    """
+    Calls a site's Jira API with its Forge system token while its integration is still
+    being found or created. Once it exists, JiraCloudClient makes the calls.
+    """
+
     def get_connect_client_key(self) -> str:
         result = self.get(
             f"/rest/atlassian-connect/1/addons/{get_jira_key()}/properties/{CONNECT_CLIENT_KEY_PROPERTY}"
@@ -43,3 +49,19 @@ def get_connect_client_key(self) -> str:
 
     def get_server_info(self) -> dict[str, Any]:
         return self.get("/rest/api/3/serverInfo")
+
+
+class ForgeUserClient(ForgeTokenClient):
+    """Calls a site's Jira API as the user in a Forge invocation, with their user token."""
+
+    def is_jira_admin(self) -> bool:
+        try:
+            result = self.get("/rest/api/3/mypermissions", params={"permissions": "ADMINISTER"})
+        except ValueError as e:
+            # A body that isn't JSON, such as a gateway error page.
+            raise ApiError("Unexpected Jira permissions response") from e
+        permissions = result.get("permissions") if isinstance(result, Mapping) else None
+        permission = permissions.get("ADMINISTER") if isinstance(permissions, Mapping) else None
+        if not isinstance(permission, Mapping):
+            raise ApiError("Unexpected Jira permissions response")
+        return permission.get("havePermission") is True

src/sentry/integrations/jira/urls.py

@@ -8,6 +8,7 @@
     JiraSentryIssueDetailsView,
 )
 from .webhooks import (
+    JiraForgeConfigureLinkWebhook,
     JiraForgeInstallationWebhook,
     JiraForgeIssueUpdatedWebhook,
     JiraIssueUpdatedWebhook,
@@ -16,6 +17,11 @@
 )
 
 urlpatterns = [
+    re_path(
+        r"^forge/configure-link/$",
+        JiraForgeConfigureLinkWebhook.as_view(),
+        name="sentry-extensions-jira-forge-configure-link",
+    ),
     re_path(
         r"^forge/installation/$",
         JiraForgeInstallationWebhook.as_view(),

src/sentry/integrations/jira/views/sentry_installation.py

@@ -15,6 +15,18 @@
 from . import SALT, UNABLE_TO_VERIFY_INSTALLATION, JiraSentryUIBaseView
 
 
+def build_configure_link(external_id: str) -> str:
+    """The link that connects a Jira site's integration to a Sentry org.
+
+    The pipeline loads everything else from the integration record, so the link only
+    needs to identify the installation.
+    """
+    return "{}.?signed_params={}".format(
+        absolute_uri("/extensions/jira/configure/"),
+        sign(salt=SALT, external_id=external_id),
+    )
+
+
 @control_silo_view
 class JiraSentryInstallationView(JiraSentryUIBaseView):
     """
@@ -32,12 +44,7 @@ def get(self, request: Request, *args: object, **kwargs: object) -> HttpResponse
         except AtlassianConnectValidationError:
             return self.get_response({"error_message": UNABLE_TO_VERIFY_INSTALLATION})
 
-        # The pipeline loads everything else from the integration record, so the
-        # link only needs to identify the installation.
-        finish_link = "{}.?signed_params={}".format(
-            absolute_uri("/extensions/jira/configure/"),
-            sign(salt=SALT, external_id=integration.external_id),
-        )
+        finish_link = build_configure_link(integration.external_id)
 
         image_path = absolute_uri(get_asset_url("sentry", "images/sentry-glyph-black.png"))
         return self.get_response({"finish_link": finish_link, "image_path": image_path})

src/sentry/integrations/jira/webhooks/__init__.py

@@ -1,10 +1,12 @@
+from .forge_configure_link import JiraForgeConfigureLinkWebhook
 from .forge_installation import JiraForgeInstallationWebhook
 from .forge_issue_updated import JiraForgeIssueUpdatedWebhook
 from .installed import JiraSentryInstalledWebhook
 from .issue_updated import JiraIssueUpdatedWebhook
 from .uninstalled import JiraSentryUninstalledWebhook
 
 __all__ = (
+    "JiraForgeConfigureLinkWebhook",
     "JiraForgeInstallationWebhook",
     "JiraForgeIssueUpdatedWebhook",
     "JiraSentryInstalledWebhook",

src/sentry/integrations/jira/webhooks/forge_configure_link.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import logging
+
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from sentry.api.api_owners import ApiOwner
+from sentry.api.api_publish_status import ApiPublishStatus
+from sentry.api.base import control_silo_endpoint
+from sentry.constants import ObjectStatus
+from sentry.integrations.jira.forge.client import ForgeUserClient
+from sentry.integrations.jira.forge.invocation_token import (
+    ForgeInvocationTokenError,
+    ForgeSigningKeysUnavailableError,
+    jira_cloud_id,
+    verify_forge_request,
+)
+from sentry.integrations.jira.views.sentry_installation import build_configure_link
+from sentry.integrations.services.integration import integration_service
+from sentry.integrations.types import IntegrationProviderSlug
+from sentry.shared_integrations.exceptions import ApiError
+
+from .base import JiraWebhookBase
+
+# Sent for invocations with a user in session when the endpoint enables `appUserToken`.
+FORGE_USER_TOKEN_HEADER = "X-Forge-OAuth-User"
+
+logger = logging.getLogger(__name__)
+
+
+@control_silo_endpoint
+class JiraForgeConfigureLinkWebhook(JiraWebhookBase):
+    owner = ApiOwner.PROJECT_MANAGEMENT_INTEGRATIONS
+    publish_status = {
+        "POST": ApiPublishStatus.PRIVATE,
+    }
+    """
+    Called by the Forge config page to get the link that connects the Jira site to a
+    Sentry organization. Only Forge-only integrations, keyed by cloud ID, are found here;
+    Connect installs keep using the Connect config page until they're rekeyed.
+    """
+
+    def post(self, request: Request, *args: object, **kwargs: object) -> Response:
+        try:
+            invocation = verify_forge_request(request)
+        except ForgeSigningKeysUnavailableError:
+            return self.respond({"detail": "Unable to verify Forge invocation token"}, status=503)
+        except ForgeInvocationTokenError:
+            return self.respond({"detail": "Invalid Forge invocation token"}, status=401)
+
+        # Only a Jira admin may link the site, whichever page they call from. Atlassian
+        # doesn't document that its admin pages enforce that, so it's checked below.
+        if invocation.principal is None:
+            return self.respond({"detail": "No user in session"}, status=403)
+
+        cloud_id = jira_cloud_id(invocation)
+        if cloud_id is None:
+            return self.respond({"detail": "Invalid Forge installation context"}, status=400)
+
+        # Asked as the caller, with their own token: checking another user's permissions
+        # as the app needs the app itself to be a Jira admin.
+        user_token = request.headers.get(FORGE_USER_TOKEN_HEADER)
+        if not user_token:
+            return self.respond({"detail": "Missing Forge user token"}, status=403)
+        try:
+            is_admin = ForgeUserClient(invocation.api_base_url, user_token).is_jira_admin()
+        except ApiError as e:
+            logger.warning(
+                "jira.forge.configure_link.permission_check_failed",
+                extra={"cloud_id": cloud_id, "status": e.code},
+            )
+            return self.respond({"detail": "Unable to check Jira permissions"}, status=502)
+        if not is_admin:
+            return self.respond({"detail": "Only Jira admins can connect Sentry"}, status=403)
+
+        integration = integration_service.get_integration(
+            provider=IntegrationProviderSlug.JIRA.value,
+            external_id=cloud_id,
+            status=ObjectStatus.ACTIVE,
+        )
+        # The installation task may not have finished yet. The integration must also belong
+        # to the calling installation, so another trusted app on the same site can't mint
+        # links for it.
+        forge = (integration.metadata.get("forge") or {}) if integration else {}
+        if integration is None or forge.get("installation_id") != invocation.installation_id:
+            return self.respond({"detail": "Installation not found"}, status=404)
+
+        return self.respond({"link": build_configure_link(integration.external_id)})

src/sentry/middleware/integrations/parsers/jira.py

@@ -24,6 +24,7 @@
 )
 from sentry.integrations.jira.views.sentry_issue_details import JiraSentryIssueDetailsControlView
 from sentry.integrations.jira.webhooks import (
+    JiraForgeConfigureLinkWebhook,
     JiraForgeInstallationWebhook,
     JiraForgeIssueUpdatedWebhook,
     JiraIssueUpdatedWebhook,
@@ -50,6 +51,7 @@ class JiraRequestParser(BaseRequestParser):
 
     control_classes = [
         JiraDescriptorEndpoint,
+        JiraForgeConfigureLinkWebhook,
         JiraForgeInstallationWebhook,
         JiraSentryInstallationView,
         JiraSentryInstalledWebhook,

static/app/data/controlsiloUrlPatterns.ts

@@ -255,6 +255,7 @@ export const controlsiloUrlPatterns: RegExp[] = [
   new RegExp('^extensions/jira/descriptor/$'),
   new RegExp('^extensions/jira/installed/$'),
   new RegExp('^extensions/jira/forge/installation/$'),
+  new RegExp('^extensions/jira/forge/configure-link/$'),
   new RegExp('^extensions/jira/uninstalled/$'),
   new RegExp('^extensions/jira/search/[^/]+/[^/]+/$'),
   new RegExp('^extensions/jira/configure/$'),

tests/sentry/integrations/jira/test_forge_configure_link.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+from urllib.parse import parse_qs, urlparse
+
+import responses
+from django.urls import reverse
+
+from sentry.constants import ObjectStatus
+from sentry.integrations.jira.forge.invocation_token import (
+    ForgeInvocation,
+    ForgeInvocationTokenError,
+    ForgeSigningKeysUnavailableError,
+)
+from sentry.integrations.jira.views import SALT
+from sentry.testutils.cases import APITestCase
+from sentry.testutils.silo import control_silo_test
+from sentry.utils.signing import unsign
+
+CLOUD_ID = "11111111-2222-3333-4444-555555555555"
+INSTALLATION_ID = "ari:cloud:ecosystem::installation/test"
+MY_PERMISSIONS_URL = f"https://api.atlassian.com/ex/jira/{CLOUD_ID}/rest/api/3/mypermissions"
+
+
+def _mock_admin(have_permission: bool) -> None:
+    responses.add(
+        responses.GET,
+        MY_PERMISSIONS_URL,
+        match=[responses.matchers.query_param_matcher({"permissions": "ADMINISTER"})],
+        json={
+            "permissions": {"ADMINISTER": {"key": "ADMINISTER", "havePermission": have_permission}}
+        },
+    )
+
+
+def _invocation(**overrides: str | None) -> ForgeInvocation:
+    fields: dict[str, str | None] = {
+        "app_id": "ari:cloud:ecosystem::app/test",
+        "installation_id": INSTALLATION_ID,
+        "api_base_url": f"https://api.atlassian.com/ex/jira/{CLOUD_ID}",
+        "cloud_id": CLOUD_ID,
+        "environment_type": "DEVELOPMENT",
+        "module_key": "sentry-configure",
+        "principal": "admin-account-id",
+    }
+    fields.update(overrides)
+    return ForgeInvocation(**fields)  # type: ignore[arg-type]
+
+
+@control_silo_test
+@patch("sentry.integrations.jira.forge.invocation_token.verify_forge_invocation_token")
+class JiraForgeConfigureLinkWebhookTest(APITestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.url = reverse("sentry-extensions-jira-forge-configure-link")
+        self.headers = {"Authorization": "Bearer fit", "X-Forge-OAuth-User": "user-token"}
+
+    def create_forge_integration(self) -> None:
+        self.create_provider_integration(
+            provider="jira",
+            external_id=CLOUD_ID,
+            metadata={
+                "base_url": "https://example.net",
+                "forge": {"installation_id": INSTALLATION_ID},
+            },
+        )
+
+    @responses.activate
+    def test_returns_signed_configure_link(self, mock_verify: MagicMock) -> None:
+        self.create_forge_integration()
+        mock_verify.return_value = _invocation()
+        _mock_admin(True)
+
+        response = self.client.post(self.url, headers=self.headers)
+
+        assert response.status_code == 200
+        # Asked as the caller, not as the app.
+        assert responses.calls[0].request.headers["Authorization"] == "Bearer user-token"
+        link = urlparse(response.data["link"])
+        assert link.path == "/extensions/jira/configure/."
+        signed_params = parse_qs(link.query)["signed_params"][0]
+        assert unsign(signed_params, salt=SALT) == {"external_id": CLOUD_ID}
+
+    @responses.activate
+    def test_requires_a_jira_admin(self, mock_verify: MagicMock) -> None:
+        self.create_forge_integration()
+        mock_verify.return_value = _invocation()
+
+        _mock_admin(False)
+        assert self.client.post(self.url, headers=self.headers).status_code == 403
+
+        headers = {"Authorization": "Bearer fit"}
+        assert self.client.post(self.url, headers=headers).status_code == 403
+
+        responses.replace(responses.GET, MY_PERMISSIONS_URL, status=500)
+        assert self.client.post(self.url, headers=self.headers).status_code == 502
+        responses.replace(responses.GET, MY_PERMISSIONS_URL, body="<html>", status=200)
+        assert self.client.post(self.url, headers=self.headers).status_code == 502
+
+    @responses.activate
+    def test_rejects_bad_requests(self, mock_verify: MagicMock) -> None:
+        mock_verify.side_effect = ForgeInvocationTokenError("bad")
+        assert self.client.post(self.url, headers=self.headers).status_code == 401
+        mock_verify.side_effect = ForgeSigningKeysUnavailableError("down")
+        assert self.client.post(self.url, headers=self.headers).status_code == 503
+        mock_verify.side_effect = None
+
+        # Only a user can be checked for Jira admin.
+        mock_verify.return_value = _invocation(principal=None)
+        assert self.client.post(self.url, headers=self.headers).status_code == 403
+
+        mock_verify.return_value = _invocation(api_base_url="https://attacker.example.com")
+        assert self.client.post(self.url, headers=self.headers).status_code == 400
+
+        # Not installed yet, or disabled.
+        mock_verify.return_value = _invocation()
+        _mock_admin(True)
+        assert self.client.post(self.url, headers=self.headers).status_code == 404
+        integration = self.create_provider_integration(
+            provider="jira",
+            external_id=CLOUD_ID,
+            status=ObjectStatus.DISABLED,
+            metadata={"forge": {"installation_id": INSTALLATION_ID}},
+        )
+        assert self.client.post(self.url, headers=self.headers).status_code == 404
+
+        # Installed by a different Forge app or installation.
+        integration.update(status=ObjectStatus.ACTIVE)
+        mock_verify.return_value = _invocation(
+            installation_id="ari:cloud:ecosystem::installation/other"
+        )
+        assert self.client.post(self.url, headers=self.headers).status_code == 404