How to send http request with basic http authentication via twisted framework

2.3k views Asked by At

I got <HTML><HEAD><TITLE>401 Unauthorized</TITLE></HEAD> error, when I run the code from http://twistedmatrix.com/documents/13.2.0/web/howto/client.html#auto4, but I don't know how to add authentication in the request.

Update

I updated my code to this:

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)

But got the following error, but I don't know what content in a list should I provide it.

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

Update

The content of the request should be in a single list, like this:

"authorization": ["Basic " + authorization]
3

There are 3 answers

0
Jean-Paul Calderone On BEST ANSWER

You can add headers to requests sent using Agent (notice line 29 of the example you linked to).

For example, to perform basic authentication/authorization, you could try something like this:

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

You can use treq's authentication support, like this:

d = treq.get(
    'http://...',
    auth=('username', 'password')
)
0
Steve D On

As a complete example:

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