Browse Source

fix(jira): validate domain of `jira_url` parameter in autocomplete (#73107)

* domain of `jira_url` has to match the domain of the jira instance
configured in the plugin
* also added validation for `jira_url` presence and denested `if
jira_url:` branch
Alexander Tarasov 8 months ago
parent
commit
59433fde5a
2 changed files with 180 additions and 76 deletions
  1. 94 76
      src/sentry_plugins/jira/plugin.py
  2. 86 0
      tests/sentry_plugins/jira/test_plugin.py

+ 94 - 76
src/sentry_plugins/jira/plugin.py

@@ -303,83 +303,101 @@ class JiraPlugin(CorePluginMixin, IssuePlugin2):
                 return Response({field: issues})
 
         jira_url = request.GET.get("jira_url")
-        if jira_url:
-            jira_url = unquote_plus(jira_url)
-            parsed = list(urlsplit(jira_url))
-            jira_query = parse_qs(parsed[3])
-
-            jira_client = self.get_jira_client(group.project)
-
-            is_user_api = re.search("/rest/api/(latest|[0-9])/user/", jira_url)
-
-            is_user_picker = "/rest/api/1.0/users/picker" in jira_url
-
-            if is_user_api:  # its the JSON version of the autocompleter
-                is_xml = False
-                jira_query["username"] = query
-                jira_query.pop(
-                    "issueKey", False
-                )  # some reason JIRA complains if this key is in the URL.
-                jira_query["project"] = project
-            elif is_user_picker:
-                is_xml = False
-                # for whatever reason, the create meta api returns an
-                # invalid path, so let's just use the correct, documented one here:
-                # https://docs.atlassian.com/jira/REST/cloud/#api/2/user
-                # also, only pass path so saved instance url will be used
-                parsed[0] = ""
-                parsed[1] = ""
-                parsed[2] = "/rest/api/2/user/picker"
-                jira_query["query"] = query
-            else:  # its the stupid XML version of the API.
-                is_xml = True
-                jira_query["query"] = query
-                if jira_query.get("fieldName"):
-                    # for some reason its a list.
-                    jira_query["fieldName"] = jira_query["fieldName"][0]
-
-            parsed[3] = urlencode(jira_query)
-            final_url = urlunsplit(parsed)
-
-            autocomplete_response = jira_client.get_cached(final_url)
-
-            if is_user_picker:
-                autocomplete_response = autocomplete_response["users"]
-
-            users = []
-
-            if is_xml:
-                for userxml in autocomplete_response.xml.findAll("users"):
-                    users.append(
-                        {"id": userxml.find("name").text, "text": userxml.find("html").text}
-                    )
-            else:
-                for user in autocomplete_response:
-                    if user.get("name"):
-                        users.append(self._get_formatted_user(user))
+        if not jira_url:
+            return Response(
+                {
+                    "error_type": "validation",
+                    "errors": [{"jira_url": "missing required parameter"}],
+                },
+                status=400,
+            )
 
-            # if JIRA user doesn't have proper permission for user api,
-            # try the assignee api instead
-            if not users and is_user_api:
-                try:
-                    autocomplete_response = jira_client.search_users_for_project(
-                        jira_query.get("project"), jira_query.get("username")
-                    )
-                except (ApiUnauthorized, ApiError) as e:
-
-                    return Response(
-                        {
-                            "error_type": "validation",
-                            "errors": [{"__all__": self.message_from_error(e)}],
-                        },
-                        status=400,
-                    )
-
-                for user in autocomplete_response:
-                    if user.get("name"):
-                        users.append(self._get_formatted_user(user))
-
-            return Response({field: users})
+        jira_url = unquote_plus(jira_url)
+        parsed = urlsplit(jira_url)
+        instance_url = self.get_option("instance_url", group.project)
+
+        if parsed.netloc != urlsplit(instance_url).netloc:
+            return Response(
+                {
+                    "error_type": "validation",
+                    "errors": [{"jira_url": "domain must match"}],
+                },
+                status=400,
+            )
+
+        parsed = list(parsed)
+        jira_query = parse_qs(parsed[3])
+
+        jira_client = self.get_jira_client(group.project)
+
+        is_user_api = re.search("/rest/api/(latest|[0-9])/user/", jira_url)
+
+        is_user_picker = "/rest/api/1.0/users/picker" in jira_url
+
+        if is_user_api:  # its the JSON version of the autocompleter
+            is_xml = False
+            jira_query["username"] = query
+            jira_query.pop(
+                "issueKey", False
+            )  # some reason JIRA complains if this key is in the URL.
+            jira_query["project"] = project
+        elif is_user_picker:
+            is_xml = False
+            # for whatever reason, the create meta api returns an
+            # invalid path, so let's just use the correct, documented one here:
+            # https://docs.atlassian.com/jira/REST/cloud/#api/2/user
+            # also, only pass path so saved instance url will be used
+            parsed[0] = ""
+            parsed[1] = ""
+            parsed[2] = "/rest/api/2/user/picker"
+            jira_query["query"] = query
+        else:  # its the stupid XML version of the API.
+            is_xml = True
+            jira_query["query"] = query
+            if jira_query.get("fieldName"):
+                # for some reason its a list.
+                jira_query["fieldName"] = jira_query["fieldName"][0]
+
+        parsed[3] = urlencode(jira_query)
+        final_url = urlunsplit(parsed)
+
+        autocomplete_response = jira_client.get_cached(final_url)
+
+        if is_user_picker:
+            autocomplete_response = autocomplete_response["users"]
+
+        users = []
+
+        if is_xml:
+            for userxml in autocomplete_response.xml.findAll("users"):
+                users.append({"id": userxml.find("name").text, "text": userxml.find("html").text})
+        else:
+            for user in autocomplete_response:
+                if user.get("name"):
+                    users.append(self._get_formatted_user(user))
+
+        # if JIRA user doesn't have proper permission for user api,
+        # try the assignee api instead
+        if not users and is_user_api:
+            try:
+                autocomplete_response = jira_client.search_users_for_project(
+                    jira_query.get("project"), jira_query.get("username")
+                )
+            except (ApiUnauthorized, ApiError) as e:
+
+                return Response(
+                    {
+                        "error_type": "validation",
+                        "errors": [{"__all__": self.message_from_error(e)}],
+                    },
+                    status=400,
+                )
+
+            for user in autocomplete_response:
+                if user.get("name"):
+                    users.append(self._get_formatted_user(user))
+
+        return Response({field: users})
 
     def message_from_error(self, exc):
         if isinstance(exc, ApiUnauthorized):

+ 86 - 0
tests/sentry_plugins/jira/test_plugin.py

@@ -189,6 +189,27 @@ issue_response: dict[str, Any] = {
     "fields": {"summary": "TypeError: 'set' object has no attribute '__getitem__'"},
 }
 
+user_search_response: list[dict[str, Any]] = [
+    {
+        "self": "https://getsentry.atlassian.net/rest/api/2/user?username=userexample",
+        "key": "JIRAUSER10100",
+        "name": "userexample",
+        "emailAddress": "user@example.com",
+        "avatarUrls": {
+            "48x48": "http://www.example.com/jira/secure/useravatar?size=large&ownerId=andrew",
+            "24x24": "http://www.example.com/jira/secure/useravatar?size=small&ownerId=andrew",
+            "16x16": "http://www.example.com/jira/secure/useravatar?size=xsmall&ownerId=andrew",
+            "32x32": "http://www.example.com/jira/secure/useravatar?size=medium&ownerId=andrew",
+        },
+        "displayName": "User Example",
+        "active": True,
+        "deleted": False,
+        "timeZone": "Europe/Vienna",
+        "locale": "en_US",
+        "lastLoginTime": "2024-06-24T12:15:00+0200",
+    }
+]
+
 
 class JiraPluginTest(TestCase):
     @cached_property
@@ -296,3 +317,68 @@ class JiraPluginTest(TestCase):
                 "self": "https://something.atlassian.net/rest/api/2/user?username=someaddon",
             }
         ) == {"id": "robot", "text": "robot (robot)"}
+
+    def _setup_autocomplete_jira(self):
+        self.plugin.set_option("instance_url", "https://getsentry.atlassian.net", self.project)
+        self.plugin.set_option("default_project", "SEN", self.project)
+        self.login_as(user=self.user)
+        self.group = self.create_group(message="Hello world", culprit="foo.bar")
+
+    @responses.activate
+    def test_autocomplete_issue_id(self):
+        self._setup_autocomplete_jira()
+        responses.add(
+            responses.GET,
+            "https://getsentry.atlassian.net/rest/api/2/search/",
+            json={"issues": [issue_response]},
+        )
+
+        url = f"/api/0/issues/{self.group.id}/plugins/jira/autocomplete/?autocomplete_query=SEN&autocomplete_field=issue_id"
+        response = self.client.get(url)
+
+        assert response.json() == {
+            "issue_id": [
+                {
+                    "text": "(SEN-19) TypeError: 'set' object has no attribute '__getitem__'",
+                    "id": "SEN-19",
+                }
+            ]
+        }
+
+    @responses.activate
+    def test_autocomplete_jira_url_reporter(self):
+        self._setup_autocomplete_jira()
+
+        responses.add(
+            responses.GET,
+            "https://getsentry.atlassian.net/rest/api/2/user/search/?username=user&project=SEN",
+            json=user_search_response,
+        )
+
+        url = f"/api/0/issues/{self.group.id}/plugins/jira/autocomplete/?autocomplete_query=user&autocomplete_field=reporter&jira_url=https://getsentry.atlassian.net/rest/api/2/user/search/"
+        response = self.client.get(url)
+        assert response.json() == {
+            "reporter": [
+                {"id": "userexample", "text": "User Example - user@example.com (userexample)"}
+            ]
+        }
+
+    def test_autocomplete_jira_url_missing(self):
+        self._setup_autocomplete_jira()
+
+        url = f"/api/0/issues/{self.group.id}/plugins/jira/autocomplete/?autocomplete_query=SEN&autocomplete_field=reporter"
+        response = self.client.get(url)
+        assert response.json() == {
+            "error_type": "validation",
+            "errors": [{"jira_url": "missing required parameter"}],
+        }
+
+    def test_autocomplete_jira_url_mismatch(self):
+        self._setup_autocomplete_jira()
+
+        url = f"/api/0/issues/{self.group.id}/plugins/jira/autocomplete/?autocomplete_query=SEN&autocomplete_field=reporter&jira_url=https://eviljira.com/"
+        response = self.client.get(url)
+        assert response.json() == {
+            "error_type": "validation",
+            "errors": [{"jira_url": "domain must match"}],
+        }