Merge pull request #688 from seratch/pr-686-addition

Additional improvements for #686 and document updates

Author: seratch

docs-src/basic_usage.rst

@@ -178,19 +178,16 @@ Modals use the same blocks that compose messages with the addition of an `input`
 
 .. code-block:: python
 
+  # This module is available since v2.6.0rc1
+  from slack.signature import SignatureVerifier
+  signature_verifier = SignatureVerifier(os.environ["SLACK_SIGNING_SECRET"])
+
   from flask import Flask, request, make_response
   app = Flask(__name__)
-  signing_secret = os.environ["SLACK_SIGNING_SECRET"]
 
   @app.route("/slack/events", methods=["POST"])
   def slack_app():
-    # Refer to https://github.com/slackapi/python-slack-events-api
-    # (The Slack Team is going to provide a new package soon)
-    if not verify_request(
-      signing_secret=signing_secret,
-      request_body=request.get_data(),
-      timestamp=request.headers.get("X-Slack-Request-Timestamp"),
-      signature=request.headers.get("X-Slack-Signature")):
+    if not signature_verifier.is_valid_request(request.get_data(), request.headers):
       return make_response("invalid request", 403)
 
     if "command" in request.form \

integration_tests/samples/basic_usage/views.py

@@ -7,52 +7,6 @@
 
 sys.path.insert(1, f"{dirname(__file__)}/../../..")
 logging.basicConfig(level=logging.DEBUG)
-# ------------------
-
-# ---------------------
-# Slack Request Verification
-# https://github.com/slackapi/python-slack-events-api
-# ---------------------
-
-import hmac
-import hashlib
-from time import time
-
-
-def verify_request(
-    signing_secret: str,
-    request_body: str,
-    timestamp: str,
-    signature: str) -> bool:
-    if abs(time() - int(timestamp)) > 60 * 5:
-        return False
-
-    if hasattr(hmac, "compare_digest"):
-        req = str.encode('v0:' + str(timestamp) + ':') + request_body
-        request_hash = 'v0=' + hmac.new(
-            str.encode(signing_secret),
-            req, hashlib.sha256
-        ).hexdigest()
-        return hmac.compare_digest(request_hash, signature)
-    else:
-        # So, we'll compare the signatures explicitly
-        req = str.encode('v0:' + str(timestamp) + ':') + request_body
-        request_hash = 'v0=' + hmac.new(
-            str.encode(signing_secret),
-            req, hashlib.sha256
-        ).hexdigest()
-
-        if len(request_hash) != len(signature):
-            return False
-        result = 0
-        if isinstance(request_hash, bytes) and isinstance(signature, bytes):
-            for x, y in zip(request_hash, signature):
-                result |= x ^ y
-        else:
-            for x, y in zip(request_hash, signature):
-                result |= ord(x) ^ ord(y)
-        return result == 0
-
 
 # ---------------------
 # Slack WebClient
@@ -62,8 +16,10 @@ def verify_request(
 
 from slack import WebClient
 from slack.errors import SlackApiError
+from slack.signature import SignatureVerifier
 
 client = WebClient(token=os.environ["SLACK_API_TOKEN"])
+signature_verifier = SignatureVerifier(os.environ["SLACK_SIGNING_SECRET"])
 
 # ---------------------
 # Flask App
@@ -73,16 +29,11 @@ def verify_request(
 from flask import Flask, request, make_response
 
 app = Flask(__name__)
-signing_secret = os.environ["SLACK_SIGNING_SECRET"]
 
 
 @app.route("/slack/events", methods=["POST"])
 def slack_app():
-    if not verify_request(
-        signing_secret=signing_secret,
-        request_body=request.get_data(),
-        timestamp=request.headers.get("X-Slack-Request-Timestamp"),
-        signature=request.headers.get("X-Slack-Signature")):
+    if not signature_verifier.is_valid_request(request.get_data(), request.headers):
         return make_response("invalid request", 403)
 
     if "command" in request.form \

slack/signature/verifier.py

@@ -1,7 +1,7 @@
 import hashlib
 import hmac
 from time import time
-from typing import Dict, Optional
+from typing import Dict, Optional, Union
 
 
 class Clock:
@@ -21,7 +21,9 @@ def __init__(self, signing_secret: str, clock: Clock = Clock()):
         self.signing_secret = signing_secret
         self.clock = clock
 
-    def is_valid_request(self, body: str, headers: Dict[str, str],) -> bool:
+    def is_valid_request(
+        self, body: Union[str, bytes], headers: Dict[str, str],
+    ) -> bool:
         """Verifies if the given signature is valid"""
         if headers is None:
             return False
@@ -32,26 +34,31 @@ def is_valid_request(self, body: str, headers: Dict[str, str],) -> bool:
             signature=normalized_headers.get("x-slack-signature", None),
         )
 
-    def is_valid(self, body: str, timestamp: str, signature: str,) -> bool:
+    def is_valid(
+        self, body: Union[str, bytes], 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]:
+    def generate_signature(
+        self, *, timestamp: str, body: Union[str, bytes]
+    ) -> Optional[str]:
         """Generates a signature"""
         if timestamp is None:
             return None
+        if body is None:
+            body = ""
+        if isinstance(body, bytes):
+            body = body.decode("utf-8")
 
         format_req = str.encode(f"v0:{timestamp}:{body}")
         encoded_secret = str.encode(self.signing_secret)

tests/signature/test_signature_verifier.py

@@ -29,9 +29,13 @@ def tearDown(self):
     }
 
     def test_generate_signature(self):
-        verifier = SignatureVerifier("8f742231b10e8888abcd99yyyzzz85a5")
-        timestamp = "1531420618"
-        signature = verifier.generate_signature(timestamp=timestamp, body=self.body)
+        verifier = SignatureVerifier(self.signing_secret)
+        signature = verifier.generate_signature(timestamp=self.timestamp, body=self.body)
+        self.assertEqual(self.valid_signature, signature)
+
+    def test_generate_signature_body_as_bytes(self):
+        verifier = SignatureVerifier(self.signing_secret)
+        signature = verifier.generate_signature(timestamp=self.timestamp, body=self.body.encode("utf-8"))
         self.assertEqual(self.valid_signature, signature)
 
     def test_is_valid_request(self):
@@ -41,6 +45,13 @@ def test_is_valid_request(self):
         )
         self.assertTrue(verifier.is_valid_request(self.body, self.headers))
 
+    def test_is_valid_request_body_as_bytes(self):
+        verifier = SignatureVerifier(
+            signing_secret=self.signing_secret,
+            clock=MockClock()
+        )
+        self.assertTrue(verifier.is_valid_request(self.body.encode("utf-8"), self.headers))
+
     def test_is_valid_request_invalid_body(self):
         verifier = SignatureVerifier(
             signing_secret=self.signing_secret,
@@ -49,6 +60,14 @@ def test_is_valid_request_invalid_body(self):
         modified_body = self.body + "------"
         self.assertFalse(verifier.is_valid_request(modified_body, self.headers))
 
+    def test_is_valid_request_invalid_body_as_bytes(self):
+        verifier = SignatureVerifier(
+            signing_secret=self.signing_secret,
+            clock=MockClock(),
+        )
+        modified_body = self.body + "------"
+        self.assertFalse(verifier.is_valid_request(modified_body.encode("utf-8"), self.headers))
+
     def test_is_valid_request_expiration(self):
         verifier = SignatureVerifier(
             signing_secret=self.signing_secret,