Slack token rotation

changes/421.added

@@ -0,0 +1 @@
+Add Slack token rotation support.

development/creds.example.env

@@ -39,6 +39,8 @@ MATTERMOST_API_TOKEN="nsutx44ibbd69r5hjjmd3hx4sw"
 # SLACK_API_TOKEN="xoxb-changeme"
 # SLACK_APP_TOKEN="changeme"
 # SLACK_SIGNING_SECRET="changeme"
+# SLACK_CLIENT_ID="changeme"
+# SLACK_CLIENT_SECRET="changeme"
 
 # - Cisco Webex ----------------------
 # WEBEX_ACCESS_TOKEN="changeme"

development/development.env

@@ -54,6 +54,7 @@ NAUTOBOT_CHATOPS_ENABLE_MS_TEAMS="False"
 # - Slack ----------------------------
 NAUTOBOT_CHATOPS_ENABLE_SLACK="False"
 # SLACK_SLASH_COMMAND_PREFIX="/"
+# SLACK_ENABLE_TOKEN_ROTATION="False"
 
 # - Cisco Webex ----------------------
 NAUTOBOT_CHATOPS_ENABLE_WEBEX="False"

development/nautobot_config.py

@@ -148,6 +148,9 @@
         "slack_app_token": os.environ.get("SLACK_APP_TOKEN"),
         "slack_signing_secret": os.environ.get("SLACK_SIGNING_SECRET"),
         "slack_slash_command_prefix": os.environ.get("SLACK_SLASH_COMMAND_PREFIX", "/"),
+        "slack_enable_token_rotation": is_truthy(os.getenv("SLACK_ENABLE_TOKEN_ROTATION", "false")),
+        "slack_client_id": os.environ.get("SLACK_CLIENT_ID"),
+        "slack_client_secret": os.environ.get("SLACK_CLIENT_SECRET"),
         # - Cisco Webex ----------------------
         "webex_msg_char_limit": int(os.getenv("WEBEX_MSG_CHAR_LIMIT", "7439")),
         "webex_signing_secret": os.environ.get("WEBEX_SIGNING_SECRET"),

docs/admin/platforms/slack.md

@@ -2,14 +2,17 @@
 
 These are the distinct configuration values you will need to configure in `nautobot_config.py`.
 
-| Configuration Setting        | Mandatory? | Default | Available on Admin Config |
-| ---------------------------- | ---------- | ------- | ------------------------- |
-| `enable_slack`               | **Yes**    | False   | Yes                       |
-| `slack_api_token`            | **Yes**    | --      | No                        |
-| `slack_app_token`            | Socket Mode| --      | No                        |
-| `slack_signing_secret`       | **Yes**    | --      | No                        |
-| `slack_slash_command_prefix` | No         | `"/"`   | No                        |
-| `slack_socket_static_host`   | No         | --      | No                        |
+| Configuration Setting        | Mandatory?    | Default | Available on Admin Config |
+| ---------------------------- | ------------- | ------- | ------------------------- |
+| `enable_slack`               | **Yes**       | False   | Yes                       |
+| `slack_api_token`            | **Yes**       | --      | No                        |
+| `slack_app_token`            | Socket Mode   | --      | No                        |
+| `slack_signing_secret`       | **Yes**       | --      | No                        |
+| `slack_slash_command_prefix` | No            | `"/"`   | No                        |
+| `slack_socket_static_host`   | No            | --      | No                        |
+| `slack_enable_token_rotation`| No            | --      | No                        |
+| `slack_client_id`            | Token Rotation| --      | No                        |
+| `slack_client_secret`        | Token Rotation| --      | No                        |
 
 These values will be used in the `nautobot_config.py` file, once we get to the section where we cover server configuration.
 For now, take a mental note that in this section where we are configuring the Slack application, we will need to explicitly note the
@@ -140,6 +143,37 @@ PLUGINS_CONFIG = {
 
 Once these steps are completed, you can proceed to the [Install Guide](../install.md#install-guide) section.
 
+### Automatic Token Rotation
+If your slack app has [token rotation](https://docs.slack.dev/authentication/using-token-rotation/) enabled, you'll need to configure Nautobot ChatOps as follows:
+
+```python
+PLUGINS_CONFIG = {
+    "nautobot_chatops": {
+        # ...
+        "slack_enable_token_rotation": True,
+        "slack_api_token": "<slack-refresh-token>",
+        "slack_client_id": "<slack-client-id>",
+        "slack_client_secret": "<slack-client-secret>",
+        # ...
+    }
+}
+```
+
+Note that `slack_api_token` now contains the refresh token, and not a Slack bot token.  
+You can get the first refresh token after enabling token rotation with a call to the oauth.v2.exchange endpoint:  
+
+```bash
+BOT_TOKEN="<NEW_BOT_USER_TOKEN_HERE>"  # from OAuth & Permission tab
+CLIENT_ID="<CLIENT_ID_HERE>"           # from Basic Information tab
+CLIENT_SECRET="<CLIENT_SECRET_HERE>"   # from Basic Information tab
+curl -k -X POST -H "Content-type: application/x-www-form-urlencoded" "https://slack.com/api/oauth.v2.exchange" -d "client_id=$CLIENT_ID&client_secret=$CLIENT_SECRET&token=$BOT_TOKEN"
+```
+
+After the bot token is exchanged for the first time, the current refresh token can be retrieved from the "Oauth & Permission" tab of your slack app.  
+Slack refresh tokens don't expire, but are one-time use. Nautobot Chatops will consume the refresh token you specify in `PLUGINS_CONFIG` to acquire a new access and refresh token pair.  
+The new access token will be used to authenticate with Slack, and the new refresh token will be persisted in the Nautobot database and used when it's time to get a new access token.  
+To avoid token rotation delays, you should enable the `Rotate Slack Access Token` job to run periodically (like every hour).  
+
 ## Configuring Multiple Chatbots in a Workspace
 
 Chatbots from multiple Nautobot implementations can exist in a single Slack workspace and even channel.

nautobot_chatops/__init__.py

@@ -65,6 +65,10 @@ class NautobotChatOpsConfig(NautobotAppConfig):
         # this can be ignored.
         # If neither option is provided, then no static images (like Nautobot Logo) will be shown.
         "slack_socket_static_host": "",
+        # Enable Slack token rotation (use refresh token to rotate access token)
+        "slack_enable_token_rotation": False,
+        "slack_client_id": "",
+        "slack_client_secret": "",
         # - Cisco Webex ----------------------
         "webex_token": "",
         "webex_signing_secret": "",
@@ -142,6 +146,14 @@ class NautobotChatOpsConfig(NautobotAppConfig):
         ),
         "enable_nso": ConstanceConfigItem(default=False, help_text="Enable NSO Integration.", field_type=bool),
         "enable_slurpit": ConstanceConfigItem(default=False, help_text="Enable Slurpit Integration.", field_type=bool),
+        # Slack token rotation support
+        "slack_refresh_token": ConstanceConfigItem(default="", help_text="Current Slack refresh token. One-time use."),
+        "slack_access_token": ConstanceConfigItem(
+            default="", help_text="Current Slack access token. Expires after 12 hours."
+        ),
+        "slack_access_token_timestamp": ConstanceConfigItem(
+            default="", help_text="Creation timestamp of the current Slack access token (ISO8601 string)."
+        ),
     }
 
     home_view_name = "plugins:nautobot_chatops:commandlog_list"

nautobot_chatops/api/views/slack.py

@@ -11,9 +11,9 @@
 from django.utils.decorators import method_decorator
 from django.views import View
 from django.views.decorators.csrf import csrf_exempt
-from slack_sdk import WebClient
 
 from nautobot_chatops.dispatchers.slack import SlackDispatcher
+from nautobot_chatops.helpers.slack import RotationAwareWebClient, get_slack_api_token
 from nautobot_chatops.metrics import signature_error_cntr
 from nautobot_chatops.utils import check_and_enqueue_command
 from nautobot_chatops.views import SettingsControlledViewMixin
@@ -153,7 +153,7 @@ def post(self, request, *args, **kwargs):
         # Check for channel_name if channel_id is present
         if context["channel_name"] is None and context["channel_id"] is not None:
             # Build a Slack Client Object
-            slack_client = WebClient(token=settings.PLUGINS_CONFIG["nautobot_chatops"]["slack_api_token"])
+            slack_client = RotationAwareWebClient(token=get_slack_api_token())
 
             # Get the channel information from Slack API
             channel_info = slack_client.conversations_info(channel=context["channel_id"])

nautobot_chatops/dispatchers/slack.py

@@ -8,10 +8,10 @@
 
 from django.conf import settings
 from django.templatetags.static import static
-from slack_sdk import WebClient
 from slack_sdk.errors import SlackApiError, SlackClientError
 from slack_sdk.webhook.client import WebhookClient
 
+from nautobot_chatops.helpers.slack import RotationAwareWebClient, get_slack_api_token
 from nautobot_chatops.metrics import backend_action_sum
 
 from .base import Dispatcher
@@ -45,7 +45,7 @@ class SlackDispatcher(Dispatcher):
     def __init__(self, *args, **kwargs):
         """Init a SlackDispatcher."""
         super().__init__(*args, **kwargs)
-        self.slack_client = WebClient(token=settings.PLUGINS_CONFIG["nautobot_chatops"]["slack_api_token"])
+        self.slack_client = RotationAwareWebClient(token=get_slack_api_token())
         self.slack_menu_limit = int(os.getenv("SLACK_MENU_LIMIT", "100"))
 
     # pylint: disable=too-many-branches

nautobot_chatops/helpers/slack.py

@@ -0,0 +1,108 @@
+"""Helper functions and classes for Slack integration."""
+
+import logging
+from datetime import datetime, timedelta, timezone
+
+from constance import config
+from django.conf import settings
+from nautobot.apps.config import get_app_settings_or_config
+from slack_sdk import WebClient
+from slack_sdk.web.async_client import AsyncWebClient
+
+from nautobot_chatops.utils import database_sync_to_async
+
+logger = logging.getLogger(__name__)
+
+
+def get_slack_api_token() -> str:
+    """Return the current Slack token.
+
+    Returns:
+        str: A valid access token if token rotation is enabled, otherwise the bot token.
+    """
+    if not get_app_settings_or_config("nautobot_chatops", "slack_enable_token_rotation"):
+        return get_app_settings_or_config("nautobot_chatops", "slack_api_token")
+
+    token_time_str = config.nautobot_chatops__slack_access_token_timestamp
+    now = datetime.now(timezone.utc)
+    token_time = None
+    if token_time_str:
+        try:
+            token_time = datetime.fromisoformat(token_time_str)
+        except Exception as e:
+            logger.warning(f"Could not parse slack_access_token_timestamp: {e}")
+            token_time = None
+
+    is_access_token_expired = token_time is None or (now - token_time) > timedelta(hours=12)
+    if is_access_token_expired:
+        logger.info("Slack access token is expired or missing, attempting rotation. "
+                    "Consider enabling the renewal job to avoid delays.")
+        return rotate_slack_access_token()
+
+    logger.debug("Using existing Slack access token.")
+    return config.nautobot_chatops__slack_access_token
+
+
+def rotate_slack_access_token() -> str | None:
+    """Rotate the Slack access token using the refresh token.
+
+    Args:
+        refresh_token (str): Current Slack refresh token.
+
+    Returns:
+        str: access_token or None on failure.
+    """
+    slack_client_id = get_app_settings_or_config("nautobot_chatops", "slack_client_id")
+    if not slack_client_id:
+        logger.error("No Slack client ID found.")
+        return None
+
+    slack_client_secret = get_app_settings_or_config("nautobot_chatops", "slack_client_secret")
+    if not slack_client_secret:
+        logger.error("No Slack client secret found.")
+        return None
+
+    # not using get_app_settings_or_config here because we want to prioritize Constance for the refresh token
+    refresh_token = config.nautobot_chatops__slack_refresh_token or \
+        settings.PLUGINS_CONFIG["nautobot_chatops"].get("slack_api_token", "")
+    if not refresh_token:
+        logger.error("No Slack refresh token found.")
+        return None
+
+    new_timestamp = datetime.now(timezone.utc).isoformat()
+    try:
+        oauth_client = WebClient()
+        response = oauth_client.oauth_v2_access(client_id=slack_client_id, client_secret=slack_client_secret,
+                                                grant_type="refresh_token", refresh_token=refresh_token)
+
+        new_access_token = response["access_token"]
+        new_refresh_token = response["refresh_token"]
+
+        config.nautobot_chatops__slack_access_token = new_access_token
+        config.nautobot_chatops__slack_refresh_token = new_refresh_token
+        config.nautobot_chatops__slack_access_token_timestamp = new_timestamp
+
+        logger.info("Slack access token rotated successfully.")
+
+        return new_access_token
+    except Exception:
+        logger.exception("Slack token rotation error")
+        return None
+
+
+class RotationAwareWebClient(WebClient):
+    """A WebClient that refreshes its token on each request if token rotation is enabled."""
+
+    def api_call(self, api_method: str, **kwargs):
+        """Override api_call to refresh token if needed before making the call."""
+        self.token = get_slack_api_token()
+        return super().api_call(api_method, **kwargs)
+
+
+class RotationAwareAsyncWebClient(AsyncWebClient):
+    """An AsyncWebClient that refreshes its token on each request if token rotation is enabled."""
+
+    async def api_call(self, api_method: str, **kwargs):
+        """Override api_call to refresh token if needed before making the call."""
+        self.token = await database_sync_to_async(get_slack_api_token)()
+        return await super().api_call(api_method, **kwargs)

nautobot_chatops/jobs.py

@@ -0,0 +1,44 @@
+""""Job to rotate Slack access token using refresh token. Schedule to run every hour or so."""
+from datetime import datetime, timedelta, timezone
+
+from constance import config
+from nautobot.apps.config import get_app_settings_or_config
+from nautobot.apps.jobs import register_jobs
+from nautobot.extras.jobs import BooleanVar, Job
+
+from nautobot_chatops.helpers.slack import rotate_slack_access_token
+
+
+class RotateSlackTokenJob(Job):
+    """Rotate the Slack access token using the refresh token."""
+    force_rotate = BooleanVar(
+        description="Rotate the Slack access token now, regardless of expiration.",
+        default=False,
+    )
+
+    class Meta:
+        """Meta attributes for the RotateSlackTokenJob."""
+        name = "Rotate Slack Access Token"
+        description = "Rotate the Slack access token if neeeded, using the refresh token."
+        has_sensitive_variables = False
+
+    def run(self, force_rotate):
+        """__Run the job to rotate the Slack access token."""
+        if not get_app_settings_or_config("nautobot_chatops", "slack_enable_token_rotation"):
+            self.logger.error("Slack token rotation is not enabled.")
+            return
+
+        token_time = datetime.fromisoformat(config.nautobot_chatops__slack_access_token_timestamp)
+        if not force_rotate and datetime.now(timezone.utc) - token_time < timedelta(hours=3):
+            self.logger.info("Slack access token is still valid; no rotation needed.")
+            return
+
+        self.logger.info("Attempting Slack access token rotation...")
+        new_token = rotate_slack_access_token()
+        if new_token:
+            self.logger.success("Slack access token rotated successfully.")
+        else:
+            self.logger.failure("Slack token rotation failed; will retry on next scheduled run.")
+
+
+register_jobs(RotateSlackTokenJob)

nautobot_chatops/sockets/slack.py

@@ -8,9 +8,9 @@
 from slack_sdk.socket_mode.aiohttp import SocketModeClient
 from slack_sdk.socket_mode.request import SocketModeRequest
 from slack_sdk.socket_mode.response import SocketModeResponse
-from slack_sdk.web.async_client import AsyncWebClient
 
 from nautobot_chatops.dispatchers.slack import SlackDispatcher
+from nautobot_chatops.helpers.slack import RotationAwareAsyncWebClient, get_slack_api_token
 from nautobot_chatops.utils import database_sync_to_async, socket_check_and_enqueue_command
 from nautobot_chatops.workers import commands_help, get_commands_registry, parse_command_string
 
@@ -22,7 +22,7 @@ async def main():
     SLASH_PREFIX = settings.PLUGINS_CONFIG["nautobot_chatops"].get("slack_slash_command_prefix")
     client = SocketModeClient(
         app_token=settings.PLUGINS_CONFIG["nautobot_chatops"].get("slack_app_token"),
-        web_client=AsyncWebClient(token=settings.PLUGINS_CONFIG["nautobot_chatops"]["slack_api_token"]),
+        web_client=RotationAwareAsyncWebClient(token=get_slack_api_token()),
     )
 
     async def process(client: SocketModeClient, req: SocketModeRequest):
@@ -49,7 +49,7 @@ async def process(client: SocketModeClient, req: SocketModeRequest):
             await client.send_socket_mode_response(response)
             await process_mention(client, req)
 
-    async def process_slash_command(client, req):
+    async def process_slash_command(client: SocketModeClient, req: SocketModeRequest):
         client.logger.debug("Processing slash command.")
         command = req.payload.get("command")
         command = command.replace(SLASH_PREFIX, "")
@@ -79,7 +79,7 @@ async def process_slash_command(client, req):
         return await socket_check_and_enqueue_command(registry, command, subcommand, params, context, SlackDispatcher)
 
     # pylint: disable-next=too-many-locals,too-many-return-statements,too-many-branches,too-many-statements
-    async def process_interactive(client, req):
+    async def process_interactive(client: SocketModeClient, req: SocketModeRequest):
         client.logger.debug("Processing interactive.")
         payload = req.payload
         selected_value = ""
@@ -224,7 +224,7 @@ async def process_interactive(client, req):
 
         return await socket_check_and_enqueue_command(registry, command, subcommand, params, context, SlackDispatcher)
 
-    async def process_mention(client, req):
+    async def process_mention(client: SocketModeClient, req: SocketModeRequest):
         context = {
             "org_id": req.payload.get("team_id"),
             "org_name": req.payload.get("team_domain"),