diff --git a/Makefile b/Makefile index 35decee8c..a2c10f3c4 100644 --- a/Makefile +++ b/Makefile @@ -8,7 +8,7 @@ lint-style: flake8 lint-type: - mypy --check-untyped-defs sopel + mypy sopel .PHONY: test test_norecord test_novcr vcr_rerecord test: diff --git a/docs/source/configuration.rst b/docs/source/configuration.rst index 07b873297..42de7254b 100644 --- a/docs/source/configuration.rst +++ b/docs/source/configuration.rst @@ -576,7 +576,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:: diff --git a/pyproject.toml b/pyproject.toml index 5746b069a..39296191d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,6 +52,7 @@ dependencies = [ "importlib_metadata>=3.6", "packaging>=23.2", "sopel-help>=0.4.0", + "scramp>=1.4.4,<2", ] [project.urls] @@ -72,6 +73,16 @@ sopel-plugins = "sopel.cli.plugins:main" [project.entry-points.pytest11] pytest-sopel = "sopel.tests.pytest_plugin" +[tool.mypy] +check_untyped_defs = true +plugins = "sqlalchemy.ext.mypy.plugin" +show_error_codes = true + +# Remove once scramp has type stubs or annotations +[[tool.mypy.overrides]] +module = 'scramp.*' +ignore_missing_imports = true + [tool.pytest.ini_options] # NOTE: sopel/ is included here to include dynamically-generated tests testpaths = ["test", "sopel"] diff --git a/setup.cfg b/setup.cfg index 9d5eb9c20..bf53c04a3 100644 --- a/setup.cfg +++ b/setup.cfg @@ -26,7 +26,3 @@ exclude = contrib/*, conftest.py no-accept-encodings = True - -[mypy] -plugins = sqlalchemy.ext.mypy.plugin -show_error_codes = True diff --git a/sopel/config/core_section.py b/sopel/config/core_section.py index 15778927c..70afcde2d 100644 --- a/sopel/config/core_section.py +++ b/sopel/config/core_section.py @@ -1191,12 +1191,16 @@ def homedir(self): :default: ``PLAIN`` - ``EXTERNAL`` is also supported, e.g. for using :attr:`client_cert_file` to - authenticate via CertFP. + Supported mechanisms are: + + * ``PLAIN``, to authenticate by sending a plaintext password + * ``EXTERNAL``, to authenticate using a TLS client certificate + (see :attr:`client_cert_file`) + * ``SCRAM-SHA-256``, for password-based 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') diff --git a/sopel/coretasks.py b/sopel/coretasks.py index 45e935154..22f7fe1fd 100644 --- a/sopel/coretasks.py +++ b/sopel/coretasks.py @@ -23,6 +23,7 @@ from __future__ import annotations import base64 +from binascii import Error as BinasciiError import collections import copy from datetime import datetime, timedelta, timezone @@ -32,6 +33,9 @@ import time from typing import Callable, Optional, TYPE_CHECKING +from scramp import ScramClient, ScramException +from scramp.core import ClientStage as ScramClientStage + from sopel import config, plugin from sopel.irc import isupport, utils from sopel.tools import events, jobs, SopelMemory, target @@ -106,7 +110,6 @@ def _handle_sasl_capability( 'cannot authenticate with SASL.', ) return plugin.CapabilityNegotiation.ERROR - # Check SASL configuration (password is required) password, mech = _get_sasl_pass_and_mech(bot) if not password: @@ -118,9 +121,18 @@ def _handle_sasl_capability( cap_info = bot.capabilities.get_capability_info('sasl') cap_params = cap_info.params - available_mechs = cap_params.split(',') if cap_params else [] + server_mechs = cap_params.split(',') if cap_params else [] - if available_mechs and mech not in available_mechs: + sopel_mechs = ["PLAIN", "EXTERNAL", "SCRAM-SHA-256"] + if mech not in sopel_mechs: + raise config.ConfigurationError( + 'SASL mechanism "{mech}" is not supported by Sopel; ' + 'available mechanisms are: {available}.'.format( + mech=mech, + available=', '.join(sopel_mechs), + ) + ) + if server_mechs and mech not in server_mechs: # Raise an error if configured to use an unsupported SASL mechanism, # but only if the server actually advertised supported mechanisms, # i.e. this network supports SASL 3.2 @@ -129,11 +141,13 @@ def _handle_sasl_capability( # by the sasl_mechs() function # See https://github.com/sopel-irc/sopel/issues/1780 for background + + common_mechs = set(sopel_mechs) & set(server_mechs) raise config.ConfigurationError( 'SASL mechanism "{mech}" is not advertised by this server; ' 'available mechanisms are: {available}.'.format( mech=mech, - available=', '.join(available_mechs), + available=', '.join(common_mechs), ) ) @@ -1250,7 +1264,43 @@ 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 == ScramClientStage.get_client_first: + try: + server_first = base64.b64decode(trigger.args[0]).decode("utf-8") + bot._scram_client.set_server_first(server_first) + except (BinasciiError, KeyError, ScramException) as e: + LOGGER.error("SASL SCRAM server_first failed: %r", e) + bot.write(("AUTHENTICATE", "*")) + return + 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 == ScramClientStage.get_client_final: + try: + server_final = base64.b64decode(trigger.args[0]).decode("utf-8") + bot._scram_client.set_server_final(server_final) + except (BinasciiError, KeyError, ScramException) as e: + LOGGER.error("SASL SCRAM server_final failed: %r", e) + bot.write(("AUTHENTICATE", "*")) + return + LOGGER.info("SASL SCRAM succeeded") + bot.write(("AUTHENTICATE", "+")) + bot._scram_client = None + return def _make_sasl_plain_token(account, password): diff --git a/test/coretasks/test_coretasks_sasl.py b/test/coretasks/test_coretasks_sasl.py index 39da87b22..9819a50e1 100644 --- a/test/coretasks/test_coretasks_sasl.py +++ b/test/coretasks/test_coretasks_sasl.py @@ -1,14 +1,18 @@ """Test behavior of SASL by ``sopel.coretasks``""" from __future__ import annotations +from base64 import b64decode, b64encode +from logging import ERROR from typing import TYPE_CHECKING import pytest +from scramp import ScramMechanism from sopel import coretasks from sopel.tests import rawlist if TYPE_CHECKING: + from sopel.bot import Sopel from sopel.config import Config from sopel.tests.factories import BotFactory, ConfigFactory @@ -65,6 +69,11 @@ def tmpconfig(configfactory: ConfigFactory) -> Config: return configfactory('conf.ini', TMP_CONFIG_SASL_DEFAULT) +@pytest.fixture +def mockbot(tmpconfig, botfactory): + return botfactory.preloaded(tmpconfig) + + def test_sasl_plain_token_generation() -> None: """Make sure SASL PLAIN tokens match the expected format.""" assert ( @@ -344,3 +353,200 @@ def test_sasl_nak(botfactory: BotFactory, tmpconfig) -> None: 'CAP END', 'QUIT :Error negotiating capabilities.', ) + + +def test_sasl_bad_method(mockbot: Sopel, caplog: pytest.LogCaptureFixture): + """Verify the bot behaves when configured with an unsupported SASL method.""" + mockbot.settings.core.auth_method = "sasl" + mockbot.settings.core.auth_target = "SCRAM-MD4" + mockbot.on_message("CAP * LS :sasl") + mockbot.on_message("CAP TestBot ACK :sasl") + assert mockbot.backend.message_sent == rawlist( + "CAP REQ :sasl", + "CAP END", + ) + with caplog.at_level(ERROR): + mockbot.on_message("AUTHENTICATE +") + assert '"SCRAM-MD4" is not supported' in caplog.text + + +def test_sasl_plain_auth(mockbot: Sopel): + """Verify the bot performs SASL PLAIN auth correctly.""" + mockbot.settings.core.auth_method = "sasl" + mockbot.settings.core.auth_target = "PLAIN" + mockbot.on_message("CAP * LS :sasl") + mockbot.on_message("CAP TestBot ACK :sasl") + assert mockbot.backend.message_sent == rawlist( + "CAP REQ :sasl", + "AUTHENTICATE PLAIN", + ) + mockbot.on_message("AUTHENTICATE +") + assert ( + len(mockbot.backend.message_sent) == 3 + and mockbot.backend.message_sent[-1] + == rawlist("AUTHENTICATE VGVzdEJvdABUZXN0Qm90AHNlY3JldA==")[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) == 4 + and mockbot.backend.message_sent[-1] == rawlist("CAP END")[0] + ) + + +def test_sasl_scram_sha_256_auth(mockbot: Sopel): + """Verify the bot performs SASL SCRAM-SHA-256 auth correctly.""" + mech = ScramMechanism() + salt, stored_key, server_key, iter_count = mech.make_auth_info( + "secret", 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 * LS :sasl") + mockbot.on_message("CAP TestBot ACK :sasl") + assert mockbot.backend.message_sent == rawlist( + "CAP REQ :sasl", + "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) == 5 + 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) == 6 + and mockbot.backend.message_sent[-1] == rawlist("CAP END")[0] + ) + + +def test_sasl_scram_sha_256_invalid_server_first(mockbot: Sopel): + """Verify the bot handles an invalid SCRAM-SHA-256 server_first correctly.""" + mech = ScramMechanism() + salt, stored_key, server_key, iter_count = mech.make_auth_info( + "secret", 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 * LS :sasl") + mockbot.on_message("CAP TestBot ACK :sasl") + 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(b"junk").decode("utf-8")) + assert ( + len(mockbot.backend.message_sent) == 4 + and mockbot.backend.message_sent[-1] == rawlist("AUTHENTICATE *")[0] + ) + + +def test_sasl_scram_sha_256_invalid_server_final(mockbot: Sopel): + """Verify the bot handles an invalid SCRAM-SHA-256 server_final correctly.""" + mech = ScramMechanism() + salt, stored_key, server_key, iter_count = mech.make_auth_info( + "secret", 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 * LS :sasl") + mockbot.on_message("CAP TestBot ACK :sasl") + 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(b"junk").decode("utf-8")) + assert ( + len(mockbot.backend.message_sent) == 5 + and mockbot.backend.message_sent[-1] == rawlist("AUTHENTICATE *")[0] + ) + + +def test_sasl_scram_sha_256_error_server_first(mockbot: Sopel): + """Verify the bot handles an error SCRAM-SHA-256 server_first correctly.""" + mockbot.settings.core.auth_method = "sasl" + mockbot.settings.core.auth_target = "SCRAM-SHA-256" + mockbot.on_message("CAP * LS :sasl") + mockbot.on_message("CAP TestBot ACK :sasl") + mockbot.on_message("AUTHENTICATE +") + + mockbot.on_message("AUTHENTICATE " + b64encode(b"e=some-error").decode("utf-8")) + assert ( + len(mockbot.backend.message_sent) == 4 + and mockbot.backend.message_sent[-1] == rawlist("AUTHENTICATE *")[0] + ) + + +def test_sasl_scram_sha_256_error_server_final(mockbot: Sopel): + """Verify the bot handles an error SCRAM-SHA-256 server_final correctly.""" + mech = ScramMechanism() + salt, stored_key, server_key, iter_count = mech.make_auth_info( + "secret", 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 * LS :sasl") + mockbot.on_message("CAP TestBot ACK :sasl") + 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(b"e=some-error").decode("utf-8")) + assert ( + len(mockbot.backend.message_sent) == 5 + and mockbot.backend.message_sent[-1] == rawlist("AUTHENTICATE *")[0] + )