mirror of https://github.com/scrapy/scrapy.git
Randomly generate telnet credentials by default
This commit is contained in:
parent
eb64214c8a
commit
37cfb49805
|
|
@ -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:
|
||||
|
|
@ -50,8 +52,21 @@ class TelnetConsole(protocol.ServerFactory):
|
|||
self.noisy = False
|
||||
self.portrange = [int(x) for x in crawler.settings.getlist('TELNETCONSOLE_PORT')]
|
||||
self.host = crawler.settings['TELNETCONSOLE_HOST']
|
||||
self.username = crawler.settings['TELNETCONSOLE_USERNAME']
|
||||
self.password = crawler.settings['TELNETCONSOLE_PASSWORD']
|
||||
|
||||
username = crawler.settings.get('TELNETCONSOLE_USERNAME', None)
|
||||
if username:
|
||||
self.username = username.encode('utf8')
|
||||
else:
|
||||
self.username = binascii.hexlify(os.urandom(8))
|
||||
|
||||
password = crawler.settings.get('TELNETCONSOLE_PASSWORD', None)
|
||||
if password:
|
||||
self.password = password.encode('utf8')
|
||||
else:
|
||||
self.password = binascii.hexlify(os.urandom(8))
|
||||
|
||||
logger.info('Telnet Username: %s' % self.username)
|
||||
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 +89,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 and
|
||||
credentials.checkPassword(self.password)):
|
||||
raise ValueError("Invalid credentials")
|
||||
|
||||
protocol = telnet.TelnetBootstrapProtocol(
|
||||
|
|
@ -104,7 +119,7 @@ class TelnetConsole(protocol.ServerFactory):
|
|||
'p': pprint.pprint,
|
||||
'prefs': print_live_refs,
|
||||
'hpy': hpy,
|
||||
'help': "This is Scrapy telnet console. For more info see: " \
|
||||
'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)
|
||||
|
|
|
|||
|
|
@ -277,8 +277,6 @@ USER_AGENT = 'Scrapy/%s (+https://scrapy.org)' % import_module('scrapy').__versi
|
|||
TELNETCONSOLE_ENABLED = 1
|
||||
TELNETCONSOLE_PORT = [6023, 6073]
|
||||
TELNETCONSOLE_HOST = '127.0.0.1'
|
||||
TELNETCONSOLE_USERNAME = 'scrapy'
|
||||
TELNETCONSOLE_PASSWORD = 'scrapy'
|
||||
|
||||
SPIDER_CONTRACTS = {}
|
||||
SPIDER_CONTRACTS_BASE = {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,59 @@
|
|||
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
|
||||
|
||||
def _get_telnet_vars():
|
||||
# This function has some side effects we don't need for this test
|
||||
return {}
|
||||
console._get_telnet_vars = _get_telnet_vars
|
||||
|
||||
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, console.password)
|
||||
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()
|
||||
Loading…
Reference in New Issue