如何通过扭曲框架发送具有基本 http 身份验证的 http 请求答案

【问题标题】：How to send http request with basic http authentication via twisted framework如何通过扭曲框架发送具有基本 http 身份验证的 http 请求
【发布时间】：2014-05-27 09:37:06
【问题描述】：

当我从http://twistedmatrix.com/documents/13.2.0/web/howto/client.html#auto4 运行代码时，出现<HTML><HEAD><TITLE>401 Unauthorized</TITLE></HEAD> 错误，但我不知道如何在请求中添加身份验证。

更新

我将我的代码更新为：

from base64 import b64encode
authorization = b64encode(b"admin:admin")

d = agent.request(
    'GET',
    'http://172.19.1.76/',
    Headers(
        {
            'User-Agent': ['Twisted Web Client Example'],
            b"authorization": b"Basic " + authorization
        }        
    ),
    None)

但出现以下错误，但我不知道我应该提供列表中的哪些内容。

packages/twisted/web/http_headers.py", line 199, in setRawHeaders
    "instance of %r instead" % (name, type(values)))
TypeError: Header entry 'authorization' should be list but found instance of <type 'str'> instead

更新

请求的内容应该在一个列表中，像这样：

"authorization": ["Basic " + authorization]

【问题讨论】：

仔细阅读错误信息:)

标签： python twisted twisted.web twisted.client

【解决方案1】：

您可以为使用Agent 发送的请求添加标头（请注意您链接到的示例的第 29 行）。

例如，要执行基本的身份验证/授权，您可以尝试以下操作：

from base64 import b64encode
authorization = b64encode(b"username:password")
getting = agent.request(..., Headers({b"authorization": [b"Basic " + authorization]}))

【讨论】：

【解决方案2】：

你可以使用treq的authentication support，像这样：

d = treq.get(
    'http://...',
    auth=('username', 'password')
)

【讨论】：

【解决方案3】：

作为一个完整的例子：

from twisted.trial import unittest   
from urllib import urlencode
from base64 import b64encode
from twisted.python.log import err
from twisted.web.client import Agent, readBody
from twisted.internet import reactor
from twisted.internet.ssl import ClientContextFactory
from twisted.web.http_headers import Headers
from zope.interface import implements
from twisted.internet.defer import succeed
from twisted.web.iweb import IBodyProducer


class StringProducer(object):
    implements(IBodyProducer)

    def __init__(self, body):
        self.body = body
        self.length = len(body)

    def startProducing(self, consumer):
        consumer.write(self.body)
        return succeed(None)

    def pauseProducing(self):
        pass

    def stopProducing(self):
        pass

class WebClientContextFactory(ClientContextFactory):
    def getContext(self, hostname, port):
        return ClientContextFactory.getContext(self)

class HttpsClientTestCases(unittest.TestCase):
    def test_https_client(self):
        def cbRequest(response):
            print 'Response version:', response.version
            print 'Response code:', response.code
            print 'Response phrase:', response.phrase
            print 'Response headers:[{}]'.format(list(response.headers.getAllRawHeaders()))
            d = readBody(response)
            d.addCallback(cbBody)
            return d

        def cbBody(body):
            print 'Response body:'
            print body

        contextFactory = WebClientContextFactory()
        agent = Agent(reactor, contextFactory)

        authorization = b64encode(b"username:password")
        data = StringProducer({'hello': 'world'})
        d = agent.request(
            'POST',
            'https://....',
            headers = Headers(
                {
                    'Content-Type': ['application/x-www-form-urlencoded'],
                    'Authorization': ['Basic ' + authorization]
                }
            ),
            bodyProducer = data
        )

        d.addCallbacks(cbRequest, err)
        d.addCallback(lambda ignored: reactor.stop())
        return d

【讨论】：