Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- # File: a (Python 3.4)
- '''
- requests.auth
- ~~~~~~~~~~~~~
- This module contains the authentication handlers for Requests.
- '''
- import os
- import re
- import time
- import hashlib
- from base64 import b64encode
- from compat import urlparse, str
- from cookies import extract_cookies_to_jar
- from utils import parse_dict_header, to_native_string
- from status_codes import codes
- CONTENT_TYPE_FORM_URLENCODED = 'application/x-www-form-urlencoded'
- CONTENT_TYPE_MULTI_PART = 'multipart/form-data'
- def _basic_auth_str(username, password):
- '''Returns a Basic Auth string.'''
- authstr = 'Basic ' + to_native_string(b64encode(('%s:%s' % (username, password)).encode('latin1')).strip())
- return authstr
- class AuthBase(object):
- __qualname__ = 'AuthBase'
- __doc__ = 'Base class that all auth implementations derive from'
- def __call__(self, r):
- raise NotImplementedError('Auth hooks must be callable.')
- class HTTPBasicAuth(AuthBase):
- __qualname__ = 'HTTPBasicAuth'
- __doc__ = 'Attaches HTTP Basic Authentication to the given Request object.'
- def __init__(self, username, password):
- self.username = username
- self.password = password
- def __call__(self, r):
- r.headers['Authorization'] = _basic_auth_str(self.username, self.password)
- return r
- class HTTPProxyAuth(HTTPBasicAuth):
- __qualname__ = 'HTTPProxyAuth'
- __doc__ = 'Attaches HTTP Proxy Authentication to a given Request object.'
- def __call__(self, r):
- r.headers['Proxy-Authorization'] = _basic_auth_str(self.username, self.password)
- return r
- class HTTPDigestAuth(AuthBase):
- __qualname__ = 'HTTPDigestAuth'
- __doc__ = 'Attaches HTTP Digest Authentication to the given Request object.'
- def __init__(self, username, password):
- self.username = username
- self.password = password
- self.last_nonce = ''
- self.nonce_count = 0
- self.chal = {}
- self.pos = None
- def build_digest_header(self, method, url):
- realm = self.chal['realm']
- nonce = self.chal['nonce']
- qop = self.chal.get('qop')
- algorithm = self.chal.get('algorithm')
- opaque = self.chal.get('opaque')
- if algorithm is None:
- _algorithm = 'MD5'
- else:
- _algorithm = algorithm.upper()
- if _algorithm == 'MD5' or _algorithm == 'MD5-SESS':
- def md5_utf8(x):
- if isinstance(x, str):
- x = x.encode('utf-8')
- return hashlib.md5(x).hexdigest()
- hash_utf8 = md5_utf8
- elif _algorithm == 'SHA':
- def sha_utf8(x):
- if isinstance(x, str):
- x = x.encode('utf-8')
- return hashlib.sha1(x).hexdigest()
- hash_utf8 = sha_utf8
- KD = lambda s, d: hash_utf8('%s:%s' % (s, d))
- if hash_utf8 is None:
- return None
- entdig = (None,)
- p_parsed = urlparse(url)
- path = p_parsed.path
- if p_parsed.query:
- path += '?' + p_parsed.query
- A1 = '%s:%s:%s' % (self.username, realm, self.password)
- A2 = '%s:%s' % (method, path)
- HA1 = hash_utf8(A1)
- HA2 = hash_utf8(A2)
- if nonce == self.last_nonce:
- self.nonce_count += 1
- else:
- self.nonce_count = 1
- ncvalue = '%08x' % self.nonce_count
- s = str(self.nonce_count).encode('utf-8')
- s += nonce.encode('utf-8')
- s += time.ctime().encode('utf-8')
- s += os.urandom(8)
- cnonce = hashlib.sha1(s).hexdigest()[:16]
- noncebit = '%s:%s:%s:%s:%s' % (nonce, ncvalue, cnonce, qop, HA2)
- if _algorithm == 'MD5-SESS':
- HA1 = hash_utf8('%s:%s:%s' % (HA1, nonce, cnonce))
- if qop is None:
- respdig = KD(HA1, '%s:%s' % (nonce, HA2))
- elif qop == 'auth' or 'auth' in qop.split(','):
- respdig = KD(HA1, noncebit)
- else:
- return None
- self.last_nonce = None
- base = 'username="%s", realm="%s", nonce="%s", uri="%s", response="%s"' % (
- self.username, realm, nonce, path, respdig)
- if opaque:
- base += ', opaque="%s"' % opaque
- if algorithm:
- base += ', algorithm="%s"' % algorithm
- if entdig:
- base += ', digest="%s"' % entdig
- if qop:
- base += ', qop="auth", nc=%s, cnonce="%s"' % (ncvalue, cnonce)
- return 'Digest %s' % base
- def handle_redirect(self, r, **kwargs):
- '''Reset num_401_calls counter on redirects.'''
- if r.is_redirect:
- setattr(self, 'num_401_calls', 1)
- def handle_401(self, r, **kwargs):
- '''Takes the given response and tries digest-auth, if needed.'''
- if self.pos is not None:
- r.request.body.seek(self.pos)
- num_401_calls = getattr(self, 'num_401_calls', 1)
- s_auth = r.headers.get('www-authenticate', '')
- if 'digest' in s_auth.lower() and num_401_calls < 2:
- setattr(self, 'num_401_calls', num_401_calls + 1)
- pat = re.compile('digest ', flags=re.IGNORECASE)
- self.chal = parse_dict_header(pat.sub('', s_auth, count=1))
- r.raw.release_conn()
- prep = r.request.copy()
- extract_cookies_to_jar(prep._cookies, r.request, r.raw)
- prep.prepare_cookies(prep._cookies)
- prep.headers['Authorization'] = self.build_digest_header(prep.method, prep.url)
- _r = r.connection.send(prep, **None)
- _r.history.append(r)
- _r.request = prep
- return _r
- None(self, 'num_401_calls', num_401_calls + 1)
- return r
- def __call__(self, r):
- if self.last_nonce:
- r.headers['Authorization'] = self.build_digest_header(r.method, r.url)
- try:
- self.pos = r.body.tell()
- except AttributeError:
- self.pos = None
- r.register_hook('response', self.handle_401)
- r.register_hook('response', self.handle_redirect)
- return r
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement