Merge pull request #686 from seratch/signature-verifier

Add SignatureVerifier for request verification

slack/signature/__init__.py

@@ -0,0 +1 @@
+from .verifier import SignatureVerifier  # noqa

slack/signature/verifier.py

@@ -0,0 +1,60 @@
+import hashlib
+import hmac
+from time import time
+from typing import Dict, Optional
+
+
+class Clock:
+    def now(self) -> float:
+        return time()
+
+
+class SignatureVerifier:
+    def __init__(self, signing_secret: str, clock: Clock = Clock()):
+        """Slack request signature verifier
+
+        Slack signs its requests using a secret that's unique to your app.
+        With the help of signing secrets, your app can more confidently verify
+        whether requests from us are authentic.
+        https://api.slack.com/authentication/verifying-requests-from-slack
+        """
+        self.signing_secret = signing_secret
+        self.clock = clock
+
+    def is_valid_request(self, body: str, headers: Dict[str, str],) -> bool:
+        """Verifies if the given signature is valid"""
+        if headers is None:
+            return False
+        normalized_headers = {k.lower(): v for k, v in headers.items()}
+        return self.is_valid(
+            body=body,
+            timestamp=normalized_headers.get("x-slack-request-timestamp", None),
+            signature=normalized_headers.get("x-slack-signature", None),
+        )
+
+    def is_valid(self, body: str, timestamp: str, signature: str,) -> bool:
+        """Verifies if the given signature is valid"""
+        if timestamp is None or signature is None:
+            return False
+
+        if abs(self.clock.now() - int(timestamp)) > 60 * 5:
+            return False
+
+        if body is None:
+            body = ""
+
+        calculated_signature = self.generate_signature(timestamp=timestamp, body=body)
+        if calculated_signature is None:
+            return False
+        return hmac.compare_digest(calculated_signature, signature)
+
+    def generate_signature(self, *, timestamp: str, body: str) -> Optional[str]:
+        """Generates a signature"""
+        if timestamp is None:
+            return None
+
+        format_req = str.encode(f"v0:{timestamp}:{body}")
+        encoded_secret = str.encode(self.signing_secret)
+        request_hash = hmac.new(encoded_secret, format_req, hashlib.sha256).hexdigest()
+        calculated_signature = f"v0={request_hash}"
+        return calculated_signature

slack/web/base_client.py

@@ -6,7 +6,6 @@
 import hmac
 import io
 import json
-import json as json_module
 import logging
 import mimetypes
 import os
@@ -15,8 +14,8 @@
 import uuid
 import warnings
 from http.client import HTTPResponse
-from typing import BinaryIO, Dict, List, Union
-from typing import Optional
+from typing import BinaryIO, Dict, List
+from typing import Optional, Union
 from urllib.error import HTTPError
 from urllib.parse import urlencode
 from urllib.parse import urljoin
@@ -333,7 +332,7 @@ def _request_for_pagination(self, api_url, req_args) -> Dict[str, any]:
         return {
             "status_code": int(response["status"]),
             "headers": dict(response["headers"]),
-            "data": json_module.loads(response["body"]),
+            "data": json.loads(response["body"]),
         }
 
     def _urllib_api_call(
@@ -613,6 +612,11 @@ def validate_slack_signature(
         Returns:
             True if signatures matches
         """
+        warnings.warn(
+            "As this method is deprecated since slackclient 2.6.0, "
+            "use `from slack.signature import SignatureVerifier` instead",
+            DeprecationWarning,
+        )
         format_req = str.encode(f"v0:{timestamp}:{data}")
         encoded_secret = str.encode(signing_secret)
         request_hash = hmac.new(encoded_secret, format_req, hashlib.sha256).hexdigest()

tests/signature/test_signature_verifier.py

@@ -0,0 +1,86 @@
+import unittest
+
+from slack.signature import SignatureVerifier
+
+
+class MockClock:
+    def now(self) -> float:
+        return 1531420618
+
+
+class TestSignatureVerifier(unittest.TestCase):
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    # https://api.slack.com/authentication/verifying-requests-from-slack
+    signing_secret = "8f742231b10e8888abcd99yyyzzz85a5"
+
+    body = "token=xyzz0WbapA4vBCDEFasx0q6G&team_id=T1DC2JH3J&team_domain=testteamnow&channel_id=G8PSS9T3V&channel_name=foobar&user_id=U2CERLKJA&user_name=roadrunner&command=%2Fwebhook-collect&text=&response_url=https%3A%2F%2Fhooks.slack.com%2Fcommands%2FT1DC2JH3J%2F397700885554%2F96rGlfmibIGlgcZRskXaIFfN&trigger_id=398738663015.47445629121.803a0bc887a14d10d2c447fce8b6703c"
+
+    timestamp = "1531420618"
+    valid_signature = "v0=a2114d57b48eac39b9ad189dd8316235a7b4a8d21a10bd27519666489c69b503"
+
+    headers = {
+        "X-Slack-Request-Timestamp": timestamp,
+        "X-Slack-Signature": valid_signature,
+    }
+
+    def test_generate_signature(self):
+        verifier = SignatureVerifier("8f742231b10e8888abcd99yyyzzz85a5")
+        timestamp = "1531420618"
+        signature = verifier.generate_signature(timestamp=timestamp, body=self.body)
+        self.assertEqual(self.valid_signature, signature)
+
+    def test_is_valid_request(self):
+        verifier = SignatureVerifier(
+            signing_secret=self.signing_secret,
+            clock=MockClock()
+        )
+        self.assertTrue(verifier.is_valid_request(self.body, self.headers))
+
+    def test_is_valid_request_invalid_body(self):
+        verifier = SignatureVerifier(
+            signing_secret=self.signing_secret,
+            clock=MockClock(),
+        )
+        modified_body = self.body + "------"
+        self.assertFalse(verifier.is_valid_request(modified_body, self.headers))
+
+    def test_is_valid_request_expiration(self):
+        verifier = SignatureVerifier(
+            signing_secret=self.signing_secret,
+        )
+        self.assertFalse(verifier.is_valid_request(self.body, self.headers))
+
+    def test_is_valid_request_none(self):
+        verifier = SignatureVerifier(
+            signing_secret=self.signing_secret,
+            clock=MockClock(),
+        )
+        self.assertFalse(verifier.is_valid_request(None, self.headers))
+        self.assertFalse(verifier.is_valid_request(self.body, None))
+        self.assertFalse(verifier.is_valid_request(None, None))
+
+    def test_is_valid(self):
+        verifier = SignatureVerifier(
+            signing_secret=self.signing_secret,
+            clock=MockClock(),
+        )
+        self.assertTrue(verifier.is_valid(self.body, self.timestamp, self.valid_signature))
+        self.assertTrue(verifier.is_valid(self.body, 1531420618, self.valid_signature))
+
+    def test_is_valid_none(self):
+        verifier = SignatureVerifier(
+            signing_secret=self.signing_secret,
+            clock=MockClock(),
+        )
+        self.assertFalse(verifier.is_valid(None, self.timestamp, self.valid_signature))
+        self.assertFalse(verifier.is_valid(self.body, None, self.valid_signature))
+        self.assertFalse(verifier.is_valid(self.body, self.timestamp, None))
+        self.assertFalse(verifier.is_valid(None, None, self.valid_signature))
+        self.assertFalse(verifier.is_valid(None, self.timestamp, None))
+        self.assertFalse(verifier.is_valid(self.body, None, None))
+        self.assertFalse(verifier.is_valid(None, None, None))