slack_import: Make check_token_access more flexible.

Previously, the `check_token_access` function had a hardcoded
`required_parameters` variable because it was only used in the Slack
data importer. This commit refactors `required_parameters` into a
function parameter, enabling the function to check a Slack token’s scope
for other purposes, such as Slack webhook integration.

Additionally, this commit changes the Slack API call in
`check_token_access` from `teams.info` to `api.test`. The endpoint is
better suited for this purpose since we're only checking a token’s scope
using the response header here.

zerver/data_import/slack.py

@@ -1481,6 +1481,9 @@ def do_convert_zipfile(
         rm_tree(slack_data_dir)
 
 
+SLACK_IMPORT_TOKEN_SCOPES = {"emoji:read", "users:read", "users:read.email", "team:read"}
+
+
 def do_convert_directory(
     slack_data_dir: str,
     output_dir: str,
@@ -1488,7 +1491,7 @@ def do_convert_directory(
     threads: int = 6,
     convert_slack_threads: bool = False,
 ) -> None:
-    check_token_access(token)
+    check_token_access(token, SLACK_IMPORT_TOKEN_SCOPES)
 
     os.makedirs(output_dir, exist_ok=True)
     if os.listdir(output_dir):
@@ -1583,12 +1586,12 @@ def get_data_file(path: str) -> Any:
         return data
 
 
-def check_token_access(token: str) -> None:
+def check_token_access(token: str, required_scopes: set[str]) -> None:
     if token.startswith("xoxp-"):
         logging.info("This is a Slack user token, which grants all rights the user has!")
     elif token.startswith("xoxb-"):
         data = requests.get(
-            "https://slack.com/api/team.info", headers={"Authorization": f"Bearer {token}"}
+            "https://slack.com/api/api.test", headers={"Authorization": f"Bearer {token}"}
         )
         if data.status_code != 200:
             raise ValueError(
@@ -1599,7 +1602,6 @@ def check_token_access(token: str) -> None:
             if error != "missing_scope":
                 raise ValueError(f"Invalid Slack token: {token}, {error}")
         has_scopes = set(data.headers.get("x-oauth-scopes", "").split(","))
-        required_scopes = {"emoji:read", "users:read", "users:read.email", "team:read"}
         missing_scopes = required_scopes - has_scopes
         if missing_scopes:
             raise ValueError(

zerver/tests/test_slack_importer.py

@@ -23,6 +23,7 @@ from zerver.data_import.import_util import (
 )
 from zerver.data_import.sequencer import NEXT_ID
 from zerver.data_import.slack import (
+    SLACK_IMPORT_TOKEN_SCOPES,
     AddedChannelsT,
     AddedMPIMsT,
     DMMembersT,
@@ -66,6 +67,7 @@ def request_callback(request: PreparedRequest) -> tuple[int, dict[str, str], byt
         "https://slack.com/api/users.info",
         "https://slack.com/api/team.info",
         "https://slack.com/api/bots.info",
+        "https://slack.com/api/api.test",
     ]
     for endpoint in endpoints:
         if request.url and endpoint in request.url:
@@ -120,6 +122,8 @@ def request_callback(request: PreparedRequest) -> tuple[int, dict[str, str], byt
             return (200, {}, orjson.dumps({"ok": False, "error": "bot_not_found"}))
         return (200, {}, orjson.dumps({"ok": True, "bot": bot_info}))
 
+    if request.url and "https://slack.com/api/api.test" in request.url:
+        return (200, {}, orjson.dumps({"ok": True}))
     # Else, https://slack.com/api/team.info
     team_not_found: tuple[int, dict[str, str], bytes] = (
         200,
@@ -178,6 +182,11 @@ class SlackImporter(ZulipTestCase):
             get_slack_api_data(slack_bots_info_url, "XXXYYYZZZ", token=token)
         self.assertEqual(invalid.exception.args, ("Error accessing Slack API: bot_not_found",))
 
+        # Api test
+        api_test_info_url = "https://slack.com/api/api.test"
+        responses.add_callback(responses.GET, api_test_info_url, callback=request_callback)
+        self.assertTrue(get_slack_api_data(api_test_info_url, "ok", token=token))
+
         # Team info
         slack_team_info_url = "https://slack.com/api/team.info"
         responses.add_callback(responses.GET, slack_team_info_url, callback=request_callback)
@@ -260,12 +269,12 @@ class SlackImporter(ZulipTestCase):
                 raise Exception("Unknown token mock")
 
         responses.add_callback(
-            responses.GET, "https://slack.com/api/team.info", callback=token_request_callback
+            responses.GET, "https://slack.com/api/api.test", callback=token_request_callback
         )
 
-        def exception_for(token: str) -> str:
+        def exception_for(token: str, required_scopes: set[str] = SLACK_IMPORT_TOKEN_SCOPES) -> str:
             with self.assertRaises(Exception) as invalid:
-                check_token_access(token)
+                check_token_access(token, required_scopes)
             return invalid.exception.args[0]
 
         self.assertEqual(
@@ -292,7 +301,7 @@ class SlackImporter(ZulipTestCase):
             "Slack token is missing the following required scopes: ['team:read', 'users:read', 'users:read.email']",
         )
 
-        check_token_access("xoxb-valid-token")
+        check_token_access("xoxb-valid-token", required_scopes=SLACK_IMPORT_TOKEN_SCOPES)
 
     def test_get_owner(self) -> None:
         user_data = [

zerver/webhooks/slack/view.py

@@ -211,6 +211,7 @@ def api_slack_webhook(
     # Handle initial URL verification handshake for Slack Events API.
     if is_challenge_handshake(payload):
         challenge = payload.get("challenge").tame(check_string)
+        check_token_access(slack_app_token)
         check_send_webhook_message(
             request,
             user_profile,
@@ -239,8 +240,6 @@ def api_slack_webhook(
     if event_type != "message":
         raise UnsupportedWebhookEventTypeError(event_type)
 
-    check_token_access(slack_app_token)
-
     raw_files = event_dict.get("files")
     files = convert_raw_file_data(raw_files) if raw_files else []
     raw_text = event_dict.get("text", "").tame(check_string)