coretasks: implement scram-sha-256

docs/source/configuration.rst

@@ -565,7 +565,7 @@ When :attr:`~CoreSection.server_auth_method` is defined the settings used are:
 * :attr:`~CoreSection.server_auth_username`: account's username
 * :attr:`~CoreSection.server_auth_password`: account's password
 * :attr:`~CoreSection.server_auth_sasl_mech`: the SASL mechanism to use
-  (defaults to ``PLAIN``; ``EXTERNAL`` is also available)
+  (default is ``PLAIN``; ``EXTERNAL`` and ``SCRAM-SHA-256`` are also available)
 
 For example, this will use NickServ ``IDENTIFY`` command and SASL mechanism::
 

pyproject.toml

@@ -43,6 +43,7 @@ dependencies = [
     "importlib_metadata>=3.6",
     "packaging",
     "sopel-help>=0.4.0",
+    "scramp>=1.4.2,<2",
 ]
 
 [project.urls]

sopel/config/core_section.py

@@ -1181,12 +1181,13 @@ def homedir(self):
 
     :default: ``PLAIN``
 
-    ``EXTERNAL`` is also supported, e.g. for using :attr:`client_cert_file` to
-    authenticate via CertFP.
+    ``EXTERNAL`` is supported, e.g. for using :attr:`client_cert_file` to
+    authenticate via CertFP, and ``SCRAM-SHA-256`` for challenge-response
+    authentication.
 
     .. versionadded:: 7.0
     .. versionchanged:: 8.0
-        Added support for SASL EXTERNAL mechanism.
+        Added support for SASL EXTERNAL and SCRAM-SHA-256 mechanisms.
     """
 
     server_auth_username = ValidatedAttribute('server_auth_username')

sopel/coretasks.py

@@ -31,6 +31,9 @@
 import re
 import time
 
+from scramp import ScramClient, ScramException
+from scramp.core import ClientStage
+
 from sopel import config, plugin
 from sopel.irc import isupport, utils
 from sopel.tools import events, jobs, SopelMemory, target
@@ -1155,7 +1158,38 @@ def auth_proceed(bot, trigger):
             bot.write(('AUTHENTICATE', '*'))
             return
 
-    # TODO: Implement SCRAM challenges
+    elif mech == "SCRAM-SHA-256":
+        if trigger.args[0] == "+":
+            bot._scram_client = ScramClient([mech], sasl_username, sasl_password)
+            client_first = bot._scram_client.get_client_first()
+            LOGGER.info("Sending SASL SCRAM client first")
+            send_authenticate(bot, client_first)
+        elif bot._scram_client.stage == ClientStage.get_client_first:
+            server_first = base64.b64decode(trigger.args[0]).decode("utf-8")
+            bot._scram_client.set_server_first(server_first)
+            if bot._scram_client.iterations < 4096:
+                LOGGER.warning(
+                    "SASL SCRAM iteration count is insecure, continuing anyway"
+                )
+            elif bot._scram_client.iterations >= 4_000_000:
+                LOGGER.warning(
+                    "SASL SCRAM iteration count is very high, this will be slow..."
+                )
+            client_final = bot._scram_client.get_client_final()
+            LOGGER.info("Sending SASL SCRAM client final")
+            send_authenticate(bot, client_final)
+        elif bot._scram_client.stage == ClientStage.get_client_final:
+            server_final = base64.b64decode(trigger.args[0]).decode("utf-8")
+            try:
+                bot._scram_client.set_server_final(server_final)
+            except ScramException as e:
+                LOGGER.error("SASL SCRAM failed: %r", e)
+                bot.write(("AUTHENTICATE", "*"))
+                raise e
+            LOGGER.info("SASL SCRAM succeeded")
+            bot.write(("AUTHENTICATE", "+"))
+            bot._scram_client = None
+        return
 
 
 def _make_sasl_plain_token(account, password):

test/test_coretasks.py

@@ -1,10 +1,12 @@
 """coretasks.py tests"""
 from __future__ import annotations
 
+from base64 import b64decode, b64encode
 from datetime import datetime, timezone
 import logging
 
 import pytest
+from scramp import ScramException, ScramMechanism
 
 from sopel import coretasks
 from sopel.irc import isupport
@@ -17,6 +19,7 @@
 [core]
 owner = Uowner
 nick = TestBot
+auth_password = hunter2
 enable = coretasks
 """
 
@@ -505,6 +508,73 @@ def test_sasl_plain_token_generation():
         'sopel\x00sopel\x00sasliscool')
 
 
+def test_sasl_plain_auth(mockbot):
+    """Verify the bot performs SASL PLAIN auth correctly."""
+    mockbot.settings.core.auth_method = "sasl"
+    mockbot.settings.core.auth_target = "PLAIN"
+    mockbot.on_message("CAP TestBot ACK :sasl")
+    assert mockbot.backend.message_sent == rawlist("AUTHENTICATE PLAIN")
+    mockbot.on_message("AUTHENTICATE +")
+    assert mockbot.backend.message_sent == rawlist(
+        "AUTHENTICATE PLAIN",
+        "AUTHENTICATE VGVzdEJvdABUZXN0Qm90AGh1bnRlcjI=",
+    )
+    mockbot.on_message(
+        "900 TestBot test!test@test TestBot :You are now logged in as TestBot"
+    )
+    mockbot.on_message("903 TestBot :SASL authentication succeeded")
+    assert mockbot.backend.message_sent == rawlist(
+        "AUTHENTICATE PLAIN",
+        "AUTHENTICATE VGVzdEJvdABUZXN0Qm90AGh1bnRlcjI=",
+        "CAP END",
+    )
+
+
+def test_sasl_scram_sha_256_auth(mockbot):
+    """Verify the bot performs SASL SCRAM-SHA-256 auth correctly."""
+    mech = ScramMechanism()
+    salt, stored_key, server_key, iter_count = mech.make_auth_info(
+        "hunter2", iteration_count=5000
+    )
+    scram_server = mech.make_server(
+        lambda x: (salt, stored_key, server_key, iter_count)
+    )
+
+    mockbot.settings.core.auth_method = "sasl"
+    mockbot.settings.core.auth_target = "SCRAM-SHA-256"
+    mockbot.on_message("CAP TestBot ACK :sasl")
+    assert mockbot.backend.message_sent == rawlist("AUTHENTICATE SCRAM-SHA-256")
+    mockbot.on_message("AUTHENTICATE +")
+
+    scram_server.set_client_first(
+        b64decode(mockbot.backend.message_sent[-1].split(b" ")[-1]).decode("utf-8")
+    )
+    mockbot.on_message(
+        "AUTHENTICATE "
+        + b64encode(scram_server.get_server_first().encode("utf-8")).decode("utf-8")
+    )
+    scram_server.set_client_final(
+        b64decode(mockbot.backend.message_sent[-1].split(b" ")[-1]).decode("utf-8")
+    )
+    mockbot.on_message(
+        "AUTHENTICATE "
+        + b64encode(scram_server.get_server_final().encode("utf-8")).decode("utf-8")
+    )
+    assert (
+        len(mockbot.backend.message_sent) == 4
+        and mockbot.backend.message_sent[-1] == rawlist("AUTHENTICATE +")[0]
+    )
+
+    mockbot.on_message(
+        "900 TestBot test!test@test TestBot :You are now logged in as TestBot"
+    )
+    mockbot.on_message("903 TestBot :SASL authentication succeeded")
+    assert (
+        len(mockbot.backend.message_sent) == 5
+        and mockbot.backend.message_sent[-1] == rawlist("CAP END")[0]
+    )
+
+
 def test_recv_chghost(mockbot, ircfactory):
     """Ensure that CHGHOST messages are correctly handled."""
     irc = ircfactory(mockbot)