feat(otp): add ARP OTP login command

Author: progresshans

src/config/env_config.py

@@ -53,4 +53,9 @@ class Settings(BaseSettings):
     QA_SERVER_BASE_URL: str = ""
     QA_API_KEY: str = ""
 
+    # ARP OTP login
+    ARP_BASE_URL: str = ""
+    ARP_OTP_SECRET: str = ""
+
+
 envs = Settings()  # Singleton

src/env

@@ -15,6 +15,7 @@ def handle_mention(self):
             msg=strip_multiline(
                 """
                 나를 멘션했을 때, 사용할 수 있는 명령어야.
+                - `otp`: ARP 로그인 코드를 발급해줘!
                 - `shuffle` 또는 `섞어줘`: 멘션된 유저들을 섞어줘!
                 - `새로운 빅챗`: 새로운 빅챗 시트를 만들어줘!
                 - `help` 또는 `도움`: 도움말을 보여줘!

src/handler/bigchat/help_response.py

@@ -9,10 +9,12 @@
 from handler.bigchat.help_response import HelpResponse
 from handler.bigchat.question_response import QuestionResponse
 from handler.bigchat.subin_like_response import SubinLikeResponse
+from handler.otp.arp_response import ArpOtpResponse
 from handler.decorator import catch_global_error, loading_emoji_while_processing
 from implementation.google_spreadsheet_client import GoogleSpreadsheetClient
 from implementation.qa_client import QAClient
 from implementation.member_finder import MemberManager
+from implementation.arp_otp import ArpOtp
 from implementation.slack_client import SlackClient
 
 MEMBER_MANAGER = None
@@ -100,8 +102,22 @@ def mention_response(event, say, client):
     question_response = QuestionResponse(
         event, SlackClient(say, client), _get_qa_client()
     )
+    arp_otp_response = ArpOtpResponse(
+        event,
+        SlackClient(say, client),
+        ArpOtp(
+            base_url=envs.ARP_BASE_URL,
+            secret=envs.ARP_OTP_SECRET,
+        ),
+    )
     MentionResponse(
-        [question_response, shuffle_response, create_bigchat_sheet, help_response],
+        [
+            arp_otp_response,
+            question_response,
+            shuffle_response,
+            create_bigchat_sheet,
+            help_response,
+        ],
         simple_response,
     ).run()
 

src/handler/controller.py

@@ -0,0 +1 @@
+

src/handler/otp/__init__.py

@@ -0,0 +1,48 @@
+import re
+
+from handler.bigchat.mention_handler import MentionHandler
+from util.utils import strip_multiline
+
+
+class ArpOtpResponse(MentionHandler):
+    ANNA_MENTION_PATTERN = re.compile(r"<@[A-Z0-9]+>")
+
+    def __init__(self, event, slack_client, arp_otp):
+        self.text = event["text"]
+        self.ts = event["ts"]
+        self.channel = event["channel"]
+        self.user_id = event["user"]
+        self.slack_client = slack_client
+        self.arp_otp = arp_otp
+
+    def handle_mention(self):
+        if not self.can_handle():
+            return False
+
+        try:
+            login_url = self.arp_otp.issue_login_url(self.user_id)
+        except ValueError:
+            self.slack_client.send_message(
+                msg="ARP OTP 설정이 아직 안 되어 있어. 운영진에게 알려줘!",
+                ts=self.ts,
+            )
+            return False
+
+        self.slack_client.send_message_only_visible_to_user(
+            msg=strip_multiline(
+                """
+                ARP 로그인 링크야.
+                <{}|10분 안에 이 링크로 접속해줘!>
+
+                링크는 본인에게만 보이고, 만료되면 `@ANNA otp`로 다시 발급받으면 돼.""",
+                login_url,
+            ),
+            user_id=self.user_id,
+            channel=self.channel,
+            ts=self.ts,
+        )
+        return True
+
+    def can_handle(self):
+        text = self.ANNA_MENTION_PATTERN.sub("", self.text).strip().lower()
+        return text == "otp"

src/handler/otp/arp_response.py

@@ -0,0 +1,49 @@
+import base64
+import hashlib
+import hmac
+import json
+import time
+from urllib.parse import urlencode
+
+
+class ArpOtp:
+    PURPOSE = "arp-login-v1"
+    TTL_SECONDS = 10 * 60
+
+    def __init__(self, base_url: str, secret: str, now=None):
+        self.base_url = (base_url or "").rstrip("/")
+        self.secret = secret or ""
+        self.now = now or time.time
+
+    def issue_login_url(self, slack_user_id: str) -> str:
+        code = self.issue_code(slack_user_id)
+        return f"{self.base_url}/auth/otp?{urlencode({'code': code})}"
+
+    def issue_code(self, slack_user_id: str) -> str:
+        if not self.base_url:
+            raise ValueError("ARP_BASE_URL is required")
+        if not self.secret:
+            raise ValueError("ARP_OTP_SECRET is required")
+        if not slack_user_id:
+            raise ValueError("slack_user_id is required")
+
+        payload = {
+            "exp": int(self.now()) + self.TTL_SECONDS,
+            "purpose": self.PURPOSE,
+            "sub": slack_user_id,
+        }
+        payload_part = self._base64url(
+            json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")
+        )
+        signature = self._base64url(
+            hmac.new(
+                self.secret.encode("utf-8"),
+                payload_part.encode("utf-8"),
+                hashlib.sha256,
+            ).digest()
+        )
+        return f"{payload_part}.{signature}"
+
+    @staticmethod
+    def _base64url(value: bytes) -> str:
+        return base64.urlsafe_b64encode(value).decode("utf-8").rstrip("=")

src/implementation/arp_otp.py

@@ -0,0 +1 @@
+

test/handler/otp/__init__.py

@@ -0,0 +1,61 @@
+import unittest
+from unittest.mock import MagicMock
+
+from handler.otp.arp_response import ArpOtpResponse
+from test.handler.bigchat.sample_data import create_sample_app_mention_event
+
+
+class TestArpOtpResponse(unittest.TestCase):
+    def test_run(self):
+        event = create_sample_app_mention_event("<@U01BN035Y6L> otp")
+        mock_slack_client = MagicMock()
+        mock_otp = MagicMock()
+        mock_otp.issue_login_url.return_value = "https://arp.ausg.me/auth/otp?code=test"
+        sut = ArpOtpResponse(event, mock_slack_client, mock_otp)
+
+        result = sut.handle_mention()
+
+        mock_otp.issue_login_url.assert_called_once_with("UQJ8HQJG5")
+        mock_slack_client.send_message_only_visible_to_user.assert_called_once()
+        kwargs = mock_slack_client.send_message_only_visible_to_user.call_args.kwargs
+        assert kwargs["user_id"] == "UQJ8HQJG5"
+        assert kwargs["channel"] == "C03SZTDEDK3"
+        assert kwargs["ts"] == "1689403771.805849"
+        assert "10분" in kwargs["msg"]
+        assert result is True
+
+    def test_not_run_by_command_notfound(self):
+        event = create_sample_app_mention_event("<@U01BN035Y6L> help")
+        mock_slack_client = MagicMock()
+        mock_otp = MagicMock()
+        sut = ArpOtpResponse(event, mock_slack_client, mock_otp)
+
+        result = sut.handle_mention()
+
+        mock_otp.assert_not_called()
+        mock_slack_client.assert_not_called()
+        assert result is False
+
+    def test_not_run_when_otp_is_not_exact_command(self):
+        event = create_sample_app_mention_event("<@U01BN035Y6L> help otp")
+        mock_slack_client = MagicMock()
+        mock_otp = MagicMock()
+        sut = ArpOtpResponse(event, mock_slack_client, mock_otp)
+
+        result = sut.handle_mention()
+
+        mock_otp.assert_not_called()
+        mock_slack_client.assert_not_called()
+        assert result is False
+
+    def test_run_by_config_error(self):
+        event = create_sample_app_mention_event("<@U01BN035Y6L> otp")
+        mock_slack_client = MagicMock()
+        mock_otp = MagicMock()
+        mock_otp.issue_login_url.side_effect = ValueError()
+        sut = ArpOtpResponse(event, mock_slack_client, mock_otp)
+
+        result = sut.handle_mention()
+
+        mock_slack_client.send_message.assert_called_once()
+        assert result is False

test/handler/otp/test_arp_response.py

@@ -0,0 +1,48 @@
+import base64
+import json
+import unittest
+
+from implementation.arp_otp import ArpOtp
+
+
+def decode_base64url(value):
+    padding = "=" * ((4 - len(value) % 4) % 4)
+    return base64.urlsafe_b64decode(f"{value}{padding}")
+
+
+class TestArpOtp(unittest.TestCase):
+    def test_issue_code(self):
+        sut = ArpOtp(
+            base_url="https://arp.ausg.me/",
+            secret="test-secret",
+            now=lambda: 1000,
+        )
+
+        code = sut.issue_code("U123")
+        payload_part, signature = code.split(".")
+
+        payload = json.loads(decode_base64url(payload_part))
+        assert payload == {
+            "exp": 1600,
+            "purpose": "arp-login-v1",
+            "sub": "U123",
+        }
+        assert len(signature) > 0
+
+    def test_issue_login_url(self):
+        sut = ArpOtp(
+            base_url="https://arp.ausg.me/",
+            secret="test-secret",
+            now=lambda: 1000,
+        )
+
+        login_url = sut.issue_login_url("U123")
+
+        assert login_url.startswith("https://arp.ausg.me/auth/otp?code=")
+
+    def test_require_base_url_and_secret(self):
+        with self.assertRaises(ValueError):
+            ArpOtp(base_url="", secret="test-secret").issue_code("U123")
+
+        with self.assertRaises(ValueError):
+            ArpOtp(base_url="https://arp.ausg.me", secret="").issue_code("U123")