Python requests.auth 模块，AuthBase() 实例源码

def get_session(username, password, cookie_path=COOKIE_PATH):
    """Get session, existing or new."""
    class USPSAuth(AuthBase):  # pylint: disable=too-few-public-methods
        """USPS authorization storage."""

        def __init__(self, username, password, cookie_path):
            """Init."""
            self.username = username
            self.password = password
            self.cookie_path = cookie_path

        def __call__(self, r):
            """Call is no-op."""
            return r

    session = requests.session()
    session.auth = USPSAuth(username, password, cookie_path)
    session.headers.update({'User-Agent': USER_AGENT})
    if os.path.exists(cookie_path):
        _LOGGER.debug("cookie found at: %s", cookie_path)
        session.cookies = _load_cookies(cookie_path)
    else:
        _login(session)
    return session

def _get_http_auth(username: str, password: str, auth_type: str) -> AuthBase:
    if auth_type == 'basic':
        return HTTPBasicAuth(username, password)
    if auth_type == 'digest':
        return HTTPDigestAuth(username, password)
    raise RetsClientError('unknown auth type %s' % auth_type)

def _get_http_auth(response, url, auth_scheme):
    """Get authentication mechanism required by server

    :param response: requests.response
    :type response: requests.Response
    :param url: parsed request url
    :type url: str
    :param auth_scheme: str
    :type auth_scheme: str
    :returns: AuthBase
    :rtype: AuthBase
    """

    hostname = url.hostname
    username = url.username
    password = url.password

    if 'www-authenticate' in response.headers:
        if auth_scheme not in ['basic', 'acsjwt', 'oauthjwt']:
            msg = ("Server responded with an HTTP 'www-authenticate' field of "
                   "'{}', DCOS only supports 'Basic'".format(
                       response.headers['www-authenticate']))
            raise DCOSException(msg)

        if auth_scheme == 'basic':
            # for basic auth if username + password was present,
            # we'd already be authed by python requests module
            username, password = _get_auth_credentials(username, hostname)
            return HTTPBasicAuth(username, password)
        # dcos auth (acs or oauth)
        else:
            return _get_dcos_auth(auth_scheme, username, password, hostname)
    else:
        msg = ("Invalid HTTP response: server returned an HTTP 401 response "
               "with no 'www-authenticate' field")
        raise DCOSException(msg)

def _get_dcos_auth(auth_scheme, username, password, hostname):
    """Get authentication flow for dcos acs auth and dcos oauth

    :param auth_scheme: authentication_scheme
    :type auth_scheme: str
    :param username: username user for authentication
    :type username: str
    :param password: password for authentication
    :type password: str
    :param hostname: hostname for credentials
    :type hostname: str
    :returns: DCOSAcsAuth
    :rtype: AuthBase
    """

    toml_config = util.get_config()
    token = toml_config.get("core.dcos_acs_token")
    if token is None:
        dcos_url = toml_config.get("core.dcos_url")
        if auth_scheme == "acsjwt":
            creds = _get_dcos_acs_auth_creds(username, password, hostname)
        else:
            creds = _get_dcos_oauth_creds(dcos_url)

        verify = _verify_ssl()
        # Silence 'Unverified HTTPS request' and 'SecurityWarning' for bad cert
        if verify is not None:
            silence_requests_warnings()

        url = urllib.parse.urljoin(dcos_url, 'acs/api/v1/auth/login')
        # using private method here, so we don't retry on this request
        # error here will be bubbled up to _request_with_auth
        response = _request('post', url, json=creds, verify=verify)

        if response.status_code == 200:
            token = response.json()['token']
            config.set_val("core.dcos_acs_token", token)

    return DCOSAcsAuth(token)