curl.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC)
elif request.auth_mode == "digest":
curl.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_DIGEST)
+ else:
+ raise ValueError("Unsupported auth_mode %s" % request.auth_mode)
curl.setopt(pycurl.USERPWD, native_str(userpwd))
gen_log.debug("%s %s (username: %r)", request.method, request.url,
:type headers: `~tornado.httputil.HTTPHeaders` or `dict`
:arg string auth_username: Username for HTTP authentication
:arg string auth_password: Password for HTTP authentication
- :arg string auth_mode: Authentication mode (basic, digest)
+ :arg string auth_mode: Authentication mode; default is "basic".
+ Allowed values are implementation-defined; ``curl_httpclient``
+ supports "basic" and "digest"; ``simple_httpclient`` only supports
+ "basic"
:arg float connect_timeout: Timeout for initial connection in seconds
:arg float request_timeout: Timeout for entire request in seconds
:arg if_modified_since: Timestamp for ``If-Modified-Since`` header
username = self.request.auth_username
password = self.request.auth_password or ''
if username is not None:
+ if self.request.auth_mode not in (None, "basic"):
+ raise ValueError("unsupported auth_mode %s",
+ self.request.auth_mode)
auth = utf8(username) + b":" + utf8(password)
self.request.headers["Authorization"] = (b"Basic " +
base64.b64encode(auth))
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.iostream import IOStream
+from tornado.log import gen_log
from tornado import netutil
from tornado.stack_context import ExceptionStackContext, NullContext
-from tornado.testing import AsyncHTTPTestCase, bind_unused_port, gen_test
+from tornado.testing import AsyncHTTPTestCase, bind_unused_port, gen_test, ExpectLog
from tornado.test.util import unittest
from tornado.util import u, bytes_type
from tornado.web import Application, RequestHandler, url
auth_password="open sesame").body,
b"Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==")
+ def test_basic_auth_explicit_mode(self):
+ self.assertEqual(self.fetch("/auth", auth_username="Aladdin",
+ auth_password="open sesame",
+ auth_mode="basic").body,
+ b"Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==")
+
+ def test_unsupported_auth_mode(self):
+ # curl and simple clients handle errors a bit differently; the
+ # important thing is that they don't fall back to basic auth
+ # on an unknown mode.
+ with ExpectLog(gen_log, "uncaught exception", required=False):
+ with self.assertRaises((ValueError, HTTPError)):
+ response = self.fetch("/auth", auth_username="Aladdin",
+ auth_password="open sesame",
+ auth_mode="asdf")
+ response.rethrow()
+
def test_follow_redirect(self):
response = self.fetch("/countdown/2", follow_redirects=False)
self.assertEqual(302, response.code)