def auth():
# Basic Authentication
requests.get('https://api.github.com/user', auth=HTTPBasicAuth('user', 'pass'))
requests.get('https://api.github.com/user', auth=('user', 'pass'))
# Digest Authentication
url = 'http://httpbin.org/digest-auth/auth/user/pass'
requests.get(url, auth=HTTPDigestAuth('user', 'pass'))
# OAuth2 Authentication????requests-oauthlib
url = 'https://api.twitter.com/1.1/account/verify_credentials.json'
auth = OAuth2('YOUR_APP_KEY', 'YOUR_APP_SECRET', 'USER_OAUTH_TOKEN')
requests.get(url, auth=auth)
pass
python类HTTPBasicAuth()的实例源码
def __init__(self, base_url, *args, **kwargs):
requests.Session.__init__(self, *args, **kwargs)
self.base_url = base_url
# Check for basic authentication
parsed_url = urlparse(self.base_url)
if parsed_url.username and parsed_url.password:
netloc = parsed_url.hostname
if parsed_url.port:
netloc += ":%d" % parsed_url.port
# remove username and password from the base_url
self.base_url = urlunparse((parsed_url.scheme, netloc, parsed_url.path, parsed_url.params, parsed_url.query, parsed_url.fragment))
# configure requests to use basic auth
self.auth = HTTPBasicAuth(parsed_url.username, parsed_url.password)
def __init__(self, api_url, auth_custom=None, auth_user=None, auth_pass=None, login=None):
"""
:param api_url: str Moira API URL
:param auth_custom: dict auth custom headers
:param auth_user: str auth user
:param auth_pass: str auth password
:param login: str auth login
"""
if not api_url.endswith('/'):
self.api_url = api_url + '/'
else:
self.api_url = api_url
self.auth = None
self.auth_custom = {'X-Webauth-User': login}
if auth_user and auth_pass:
self.auth = HTTPBasicAuth(auth_user, auth_pass)
if auth_custom:
self.auth_custom.update(auth_custom)
def __init__(
self,
*,
broker_url,
pact_dir='.',
user=None,
password=None,
authentication=False
):
self.broker_url = broker_url.rstrip('/')
self.username = user
self.password = password
self.pact_dir = pact_dir
self.authentication = authentication
self._auth = None
if self.authentication:
if not (self.username and self.password):
raise ValueError(
'When authentication is True, username and password '
'must be provided.'
)
else:
self._auth = HTTPBasicAuth(self.username, self.password)
def test_pull_pact_authentication(mock_get, pull_pact_response):
mock_get.return_value = pull_pact_response
broker_client = BrokerClient(
broker_url=settings.PACT_BROKER_URL,
user='user',
password='password',
authentication=True
)
broker_client.pull_pact(
provider=PROVIDER,
consumer=CONSUMER,
)
mock_get.assert_called_once_with(
EXPECTED_PULL_PACT_URL,
auth=HTTPBasicAuth('user', 'password')
)
def __init__(self, server_url, project="DefaultCollection", user=None, password=None, verify=False,
auth_type=HTTPBasicAuth,
connect_timeout=20, read_timeout=180, ):
"""
:param server_url: url to TFS server, e.g. https://tfs.example.com/
:param project: Collection or Collection\\Project
:param user: username, or DOMAIN\\username
:param password: password
:param verify: True|False - verify HTTPS cert
:param connect_timeout: Requests CONNECTION timeout, sec or None
:param read_timeout: Requests READ timeout, sec or None
"""
if user is None or password is None:
raise ValueError('User name and api-key must be specified!')
self.rest_client = TFSHTTPClient(server_url,
project=project,
user=user, password=password,
verify=verify,
timeout=(connect_timeout, read_timeout),
auth_type=auth_type,
)
def _compose_args(self, method, data=None):
args = {}
if data is None:
data = {}
query = 'params' if method == 'get' else 'data'
args[query] = self._structure.get('common-params', {})
args[query].update(data)
args['headers'] = self._structure.get('common-headers', {})
auth = self._structure.get('auth')
if auth == 'params':
args[query].update({'username': self.billing_username,
'password': self.billing_password})
elif auth == 'headers':
args['auth'] = HTTPBasicAuth(
self.billing_username, self.billing_password)
if self._structure.get('verify') == False:
args['verify'] = False
return args
def grant_access_token(self, code):
res = self._session.post(
'https://bitbucket.org/site/oauth2/access_token',
data={
'grant_type': 'authorization_code',
'code': code,
},
auth=HTTPBasicAuth(self._oauth_key, self._oauth_secret)
)
try:
res.raise_for_status()
except requests.RequestException as reqe:
error_info = res.json()
raise BitbucketAPIError(
res.status_code,
error_info.get('error', ''),
error_info.get('error_description', ''),
request=reqe.request,
response=reqe.response
)
data = res.json()
self._access_token = data['access_token']
self._refresh_token = data['refresh_token']
self._token_type = data['token_type']
return data
def refresh_token(self):
"""
Refresh token regardless of when it expires.
Raises HTTPError on any non 2xx response code
"""
refresh_token = self._token['refresh_token']
params = {
'refresh_token': refresh_token,
'grant_type': 'refresh_token'
}
auth = HTTPBasicAuth(self.client_id, self.client_secret)
response = self.session.post(self.TOKEN_URL, data=params, auth=auth)
response.raise_for_status()
token = response.json()
token['expires_at'] = int(time.time()) + token['expires_in']
if 'refresh_token' not in token:
token['refresh_token'] = refresh_token
self.token = token
def jsonrpc_call(url, method, params={}, verify_ssl_cert=True,
username=None, password=None):
payload = {"method": method, "params": params, "jsonrpc": "2.0", "id": 0}
kwargs = {
"url": url,
"headers": {'content-type': 'application/json'},
"data": json.dumps(payload),
"verify": verify_ssl_cert,
}
if username and password:
kwargs["auth"] = HTTPBasicAuth(username, password)
global method_cumulative_calltime
begin = time.time()
response = requests.post(**kwargs).json()
method_cumulative_calltime[method] += (time.time() - begin)
if "result" not in response:
raise JsonRpcCallFailed(payload, response)
return response["result"]
def _handle_basic_auth_401(self, r, kwargs):
if self.pos is not None:
r.request.body.seek(self.pos)
# Consume content and release the original connection
# to allow our new request to reuse the same one.
r.content
r.raw.release_conn()
prep = r.request.copy()
if not hasattr(prep, '_cookies'):
prep._cookies = cookies.RequestsCookieJar()
cookies.extract_cookies_to_jar(prep._cookies, r.request, r.raw)
prep.prepare_cookies(prep._cookies)
self.auth = auth.HTTPBasicAuth(self.username, self.password)
prep = self.auth(prep)
_r = r.connection.send(prep, **kwargs)
_r.history.append(r)
_r.request = prep
return _r
def add_strategy(self, domain, strategy):
"""Add a new domain and authentication strategy.
:param str domain: The domain you wish to match against. For example:
``'https://api.github.com'``
:param str strategy: The authentication strategy you wish to use for
that domain. For example: ``('username', 'password')`` or
``requests.HTTPDigestAuth('username', 'password')``
.. code-block:: python
a = AuthHandler({})
a.add_strategy('https://api.github.com', ('username', 'password'))
"""
# Turn tuples into Basic Authentication objects
if isinstance(strategy, tuple):
strategy = HTTPBasicAuth(*strategy)
key = self._key_from_url(domain)
self.strategies[key] = strategy
def get_strategy_for(self, url):
"""Retrieve the authentication strategy for a specified URL.
:param str url: The full URL you will be making a request against. For
example, ``'https://api.github.com/user'``
:returns: Callable that adds authentication to a request.
.. code-block:: python
import requests
a = AuthHandler({'example.com', ('foo', 'bar')})
strategy = a.get_strategy_for('http://example.com/example')
assert isinstance(strategy, requests.auth.HTTPBasicAuth)
"""
key = self._key_from_url(url)
return self.strategies.get(key, NullAuthStrategy())
def _auth(http_auth):
if not http_auth:
return None
if http_auth['auth_type'] == 'basic':
return HTTPBasicAuth(
http_auth['username'],
http_auth['password']
)
if http_auth['auth_type'] == 'digest':
return HTTPDigestAuth(
http_auth['username'],
http_auth['password']
)
raise Exception('Authorization information is not valid.')
def _handle_basic_auth_401(self, r, kwargs):
if self.pos is not None:
r.request.body.seek(self.pos)
# Consume content and release the original connection
# to allow our new request to reuse the same one.
r.content
r.raw.release_conn()
prep = r.request.copy()
if not hasattr(prep, '_cookies'):
prep._cookies = cookies.RequestsCookieJar()
cookies.extract_cookies_to_jar(prep._cookies, r.request, r.raw)
prep.prepare_cookies(prep._cookies)
self.auth = auth.HTTPBasicAuth(self.username, self.password)
prep = self.auth(prep)
_r = r.connection.send(prep, **kwargs)
_r.history.append(r)
_r.request = prep
return _r
def add_strategy(self, domain, strategy):
"""Add a new domain and authentication strategy.
:param str domain: The domain you wish to match against. For example:
``'https://api.github.com'``
:param str strategy: The authentication strategy you wish to use for
that domain. For example: ``('username', 'password')`` or
``requests.HTTPDigestAuth('username', 'password')``
.. code-block:: python
a = AuthHandler({})
a.add_strategy('https://api.github.com', ('username', 'password'))
"""
# Turn tuples into Basic Authentication objects
if isinstance(strategy, tuple):
strategy = HTTPBasicAuth(*strategy)
key = self._key_from_url(domain)
self.strategies[key] = strategy
def get_strategy_for(self, url):
"""Retrieve the authentication strategy for a specified URL.
:param str url: The full URL you will be making a request against. For
example, ``'https://api.github.com/user'``
:returns: Callable that adds authentication to a request.
.. code-block:: python
import requests
a = AuthHandler({'example.com', ('foo', 'bar')})
strategy = a.get_strategy_for('http://example.com/example')
assert isinstance(strategy, requests.auth.HTTPBasicAuth)
"""
key = self._key_from_url(url)
return self.strategies.get(key, NullAuthStrategy())
def test_get_podm_status_Offline_by_http_exception(self, mock_get):
mock_get.side_effect = requests.ConnectionError
self.assertEqual(redfish.pod_status('url', 'username', 'password'),
constants.PODM_STATUS_OFFLINE)
mock_get.asset_called_once_with('url',
auth=auth.HTTPBasicAuth('username',
'password'))
# SSL Error
mock_get.side_effect = requests.exceptions.SSLError
self.assertEqual(redfish.pod_status('url', 'username', 'password'),
constants.PODM_STATUS_OFFLINE)
self.assertEqual(mock_get.call_count, 2)
# Timeout
mock_get.side_effect = requests.Timeout
self.assertEqual(redfish.pod_status('url', 'username', 'password'),
constants.PODM_STATUS_OFFLINE)
self.assertEqual(mock_get.call_count, 3)
def _upload(url, data_file, username, password):
"""
Upload will submit a results file via polarion ReST interface.
"""
url_match = '(http(s)?)\:\/\/localhost'
if re.search(url_match, url):
print("Please configure url settings.")
exit(1)
polarion_request = post(url,
data=data_file,
auth=auth.HTTPBasicAuth(username,
password))
status_code = polarion_request.status_code
if status_code == codes.ok:
return status_code
else:
print("Results upload failed with the follow: {}".format(
polarion_request.status_code))
raise exceptions.RequestException
def division_standings():
url = base_url + 'division_team_standings.json'
response = requests.get((url),
auth=HTTPBasicAuth(secret.msf_username, secret.msf_pw))
data = response.json()
for division in data['divisionteamstandings']['division']:
for team in division['teamentry']:
team_id = team['team']['ID']
rank = team['rank']
conn = sqlite3.connect('nflpool.sqlite')
cur = conn.cursor()
cur.execute('''INSERT INTO division_standings(team_id, rank)
VALUES(?,?)''', (team_id, rank))
conn.commit()
conn.close()
# Get Playoff Standings for each team (need number 5 & 6 in each conference for Wild Card picks)
def playoff_standings():
url = base_url + 'playoff_team_standings.json'
response = requests.get((url),
auth=HTTPBasicAuth(secret.msf_username, secret.msf_pw))
data = response.json()
for conference in data['playoffteamstandings']['conference']:
for team in conference['teamentry']:
team_id = team['team']['ID']
rank = team['rank']
conn = sqlite3.connect('nflpool.sqlite')
cur = conn.cursor()
cur.execute('''INSERT INTO playoff_rankings(team_id, rank)
VALUES(?,?)''', (team_id, rank))
conn.commit()
conn.close()
# Get individual statistics for each category
def parse_auth_argument(args):
auth = args.auth
if auth == 'basic':
if not (args.username and args.password):
print("For basic authentication, 'username' and 'password' "
"parameter are needed")
exit(3)
auth = HTTPBasicAuth(args.username, args.password)
elif auth == 'oauth':
if not (args.consumer_key and args.private_key):
print("For oauth authentication, 'consumer-key' "
"and 'private-key' parameter are needed")
exit(3)
auth = get_oauth1session(args.consumer_key, args.consumer_secret,
args.private_key, args.passphrase)
return auth
def parse_auth_argument(args):
auth = args.auth
if auth == 'basic':
if not (args.username and args.password):
print("For basic authentication, 'username' and 'password' "
"parameter are needed")
exit(3)
auth = HTTPBasicAuth(args.username, args.password)
elif auth == 'oauth':
if not (args.consumer_key and args.private_key):
print("For oauth authentication, 'consumer-key' "
"and 'private-key' parameter are needed")
exit(3)
auth = get_oauth1session(args.consumer_key, args.consumer_secret,
args.private_key, args.passphrase)
return auth
def parse_auth_argument(args):
auth = args.auth
if auth == 'basic':
if not (args.username and args.password):
print("For basic authentication, 'username' and 'password' "
"parameter are needed")
exit(3)
auth = HTTPBasicAuth(args.username, args.password)
elif auth == 'oauth':
if not (args.consumer_key and args.private_key):
print("For oauth authentication, 'consumer-key' "
"and 'private-key' parameter are needed")
exit(3)
auth = get_oauth1session(args.consumer_key, args.consumer_secret,
args.private_key, args.passphrase)
return auth
def post_result(url, user, password, verify, data, debug):
r = requests.post(
url,
auth=HTTPBasicAuth(user, password),
verify=verify,
json=data
)
if debug:
err("Request result: " + str(r))
if r.status_code == 403:
err("HTTP 403 Forbidden - Does your bitbucket user have rights to the repo?")
elif r.status_code == 401:
err("HTTP 401 Unauthorized - Are your bitbucket credentials correct?")
# All other errors, just dump the JSON
if r.status_code != 204: # 204 is a success per Bitbucket docs
err(json_pp(r.json()))
return r
# Stop all this from executing if we were imported, say, for testing.
def refresh(self, session=None, auth=None):
"""
Refreshes the token.
:param session: :class:`requests_oauthlib.OAuth2Session` for refreshing token with.
:param auth: :class:`requests.auth.HTTPBasicAuth`
"""
if self.can_refresh:
if not session:
session = OAuth2Session(app_settings.ESI_SSO_CLIENT_ID)
if not auth:
auth = HTTPBasicAuth(app_settings.ESI_SSO_CLIENT_ID, app_settings.ESI_SSO_CLIENT_SECRET)
try:
self.access_token = \
session.refresh_token(app_settings.ESI_TOKEN_URL, refresh_token=self.refresh_token, auth=auth)[
'access_token']
self.created = timezone.now()
self.save()
except (InvalidGrantError, MissingTokenError):
raise TokenInvalidError()
except InvalidClientError:
raise ImproperlyConfigured('Verify ESI_SSO_CLIENT_ID and ESI_SSO_CLIENT_SECRET settings.')
else:
raise NotRefreshableTokenError()
def whois_search(self, query='', field=''):
"""
Performs a WHOIS search with the given parameters
:param query: the search term
:type str:
:param field: WHOIS field to execute the search on: domain, email,
name, organization, address, phone, nameserver
:type str:
:returns dict using json.loads()
"""
log.debug('Permforming WHOIS search with query: {0} and field: '
'{1}'.format(query, field))
params = {'query': query, 'field': field}
r = requests.get('https://api.passivetotal.org/v2/whois/search',
auth=HTTPBasicAuth(self.username, self.apikey),
proxies=self.proxies, params=params)
if r.status_code == 200:
results = json.loads(r.text)
return results
else:
log.error('HTTP code {0} returned during WHOIS search '
'request.'.format(r.status_code))
return None
def whois_search(self, query='', field=''):
"""
Performs a WHOIS search with the given parameters
:param query: the search term
:type str:
:param field: WHOIS field to execute the search on: domain, email,
name, organization, address, phone, nameserver
:type str:
:returns dict using json.loads()
"""
log.debug('Permforming WHOIS search with query: {0} and field: '
'{1}'.format(query, field))
params = {'query': query, 'field': field}
r = requests.get('https://api.passivetotal.org/v2/whois/search',
auth=HTTPBasicAuth(self.username, self.apikey),
proxies=self.proxies, params=params)
if r.status_code == 200:
results = json.loads(r.text)
return results
else:
log.error('HTTP code {0} returned during WHOIS search '
'request.'.format(r.status_code))
return None
def cuckoo_check_if_dup(self, sha256):
"""
Check if file already was analyzed by cuckoo
"""
try:
print("Looking for tasks for: {}".format(sha256))
res = requests.get(urljoin(self.url_base, "/files/view/sha256/{}".format(sha256)),
verify=False,
auth=HTTPBasicAuth(self.api_user,self.api_passwd),
timeout=60)
if res and res.ok and res.status_code == 200:
print("Sample found in Sandbox, with ID: {}".format(res.json().get("sample", {}).get("id", 0)))
return True
else:
return False
except Exception as e:
print(e)
return False
def postfile(self, artifact, fileName):
"""
Send a file to Cuckoo
"""
files = {"file": (fileName, open(artifact, "rb").read())}
try:
res = requests.post(urljoin(self.url_base, "tasks/create/file").encode("utf-8"), files=files, auth=HTTPBasicAuth(
self.api_user,
self.api_passwd
),
verify=False)
if res and res.ok:
print("Cuckoo Request: {}, Task created with ID: {}".format(res.status_code, res.json()["task_id"]))
else:
print("Cuckoo Request failed: {}".format(res.status_code))
except Exception as e:
print("Cuckoo Request failed: {}".format(e))
return
