Merge pull request #3429 from hcoura/telnet-auth

Randomly generate telnet credentials by default

scrapy/extensions/telnet.py

@@ -7,6 +7,8 @@ See documentation in docs/topics/telnetconsole.rst
 import pprint
 import logging
 import traceback
+import binascii
+import os
 
 from twisted.internet import protocol
 try:
@@ -52,6 +54,11 @@ class TelnetConsole(protocol.ServerFactory):
         self.host = crawler.settings['TELNETCONSOLE_HOST']
         self.username = crawler.settings['TELNETCONSOLE_USERNAME']
         self.password = crawler.settings['TELNETCONSOLE_PASSWORD']
+
+        if not self.password:
+            self.password = binascii.hexlify(os.urandom(8)).decode('utf8')
+            logger.info('Telnet Password: %s', self.password)
+
         self.crawler.signals.connect(self.start_listening, signals.engine_started)
         self.crawler.signals.connect(self.stop_listening, signals.engine_stopped)
 
@@ -74,8 +81,8 @@ class TelnetConsole(protocol.ServerFactory):
             """An implementation of IPortal"""
             @defers
             def login(self_, credentials, mind, *interfaces):
-                if not (credentials.username == self.username
-                        and credentials.checkPassword(self.password)):
+                if not (credentials.username == self.username.encode('utf8') and
+                        credentials.checkPassword(self.password.encode('utf8'))):
                     raise ValueError("Invalid credentials")
 
                 protocol = telnet.TelnetBootstrapProtocol(
@@ -104,8 +111,8 @@ class TelnetConsole(protocol.ServerFactory):
             'p': pprint.pprint,
             'prefs': print_live_refs,
             'hpy': hpy,
-            'help': "This is Scrapy telnet console. For more info see: " \
-                "https://doc.scrapy.org/en/latest/topics/telnetconsole.html",
+            'help': "This is Scrapy telnet console. For more info see: "
+                    "https://doc.scrapy.org/en/latest/topics/telnetconsole.html",
         }
         self.crawler.signals.send_catch_log(update_telnet_vars, telnet_vars=telnet_vars)
         return telnet_vars

scrapy/settings/default_settings.py

@@ -278,7 +278,7 @@ TELNETCONSOLE_ENABLED = 1
 TELNETCONSOLE_PORT = [6023, 6073]
 TELNETCONSOLE_HOST = '127.0.0.1'
 TELNETCONSOLE_USERNAME = 'scrapy'
-TELNETCONSOLE_PASSWORD = 'scrapy'
+TELNETCONSOLE_PASSWORD = None
 
 SPIDER_CONTRACTS = {}
 SPIDER_CONTRACTS_BASE = {

tests/test_extension_telnet.py

@@ -0,0 +1,60 @@
+try:
+    import unittest.mock as mock
+except ImportError:
+    import mock
+
+from twisted.trial import unittest
+from twisted.conch.telnet import ITelnetProtocol
+from twisted.cred import credentials
+from twisted.internet import defer
+
+from scrapy.extensions.telnet import TelnetConsole, logger
+from scrapy.utils.test import get_crawler
+
+
+class TelnetExtensionTest(unittest.TestCase):
+    def _get_console_and_portal(self, settings=None):
+        crawler = get_crawler(settings_dict=settings)
+        console = TelnetConsole(crawler)
+        username = console.username
+        password = console.password
+
+        # This function has some side effects we don't need for this test
+        console._get_telnet_vars = lambda: {}
+
+        console.start_listening()
+        protocol = console.protocol()
+        portal = protocol.protocolArgs[0]
+
+        return console, portal
+
+    @defer.inlineCallbacks
+    def test_bad_credentials(self):
+        console, portal = self._get_console_and_portal()
+        creds = credentials.UsernamePassword(b'username', b'password')
+        d = portal.login(creds, None, ITelnetProtocol)
+        yield self.assertFailure(d, ValueError)
+        console.stop_listening()
+
+    @defer.inlineCallbacks
+    def test_good_credentials(self):
+        console, portal = self._get_console_and_portal()
+        creds = credentials.UsernamePassword(
+            console.username.encode('utf8'),
+            console.password.encode('utf8')
+        )
+        d = portal.login(creds, None, ITelnetProtocol)
+        yield d
+        console.stop_listening()
+
+    @defer.inlineCallbacks
+    def test_custom_credentials(self):
+        settings = {
+            'TELNETCONSOLE_USERNAME': 'user',
+            'TELNETCONSOLE_PASSWORD': 'pass',
+        }
+        console, portal = self._get_console_and_portal(settings=settings)
+        creds = credentials.UsernamePassword(b'user', b'pass')
+        d = portal.login(creds, None, ITelnetProtocol)
+        yield d
+        console.stop_listening()