def configure_http_pool():
global gl_http_pool
if gl_args.mode == 'auto-scan' or gl_args.mode == 'file-scan':
timeout = Timeout(connect=1.0, read=3.0)
else:
timeout = Timeout(connect=gl_args.timeout, read=6.0)
if gl_args.proxy:
# when using proxy, protocol should be informed
if (gl_args.host is not None and 'http' not in gl_args.host) or 'http' not in gl_args.proxy:
print_and_flush(RED + " * When using proxy, you must specify the http or https protocol"
" (eg. http://%s).\n\n" %(gl_args.host if 'http' not in gl_args.host else gl_args.proxy) +ENDC)
logging.critical('Protocol not specified')
exit(1)
try:
if gl_args.proxy_cred:
headers = make_headers(proxy_basic_auth=gl_args.proxy_cred)
gl_http_pool = ProxyManager(proxy_url=gl_args.proxy, proxy_headers=headers, timeout=timeout, cert_reqs='CERT_NONE')
else:
gl_http_pool = ProxyManager(proxy_url=gl_args.proxy, timeout=timeout, cert_reqs='CERT_NONE')
except:
print_and_flush(RED + " * An error occurred while setting the proxy. Please see log for details..\n\n" +ENDC)
logging.critical('Error while setting the proxy', exc_info=traceback)
exit(1)
else:
gl_http_pool = PoolManager(timeout=timeout, cert_reqs='CERT_NONE')
python类make_headers()的实例源码
def __init__(self, host='localhost', port=9200, http_auth=None,
use_ssl=False, verify_certs=False, ca_certs=None, client_cert=None,
ssl_version=None, ssl_assert_hostname=None, ssl_assert_fingerprint=None,
maxsize=10, **kwargs):
super(Urllib3HttpConnection, self).__init__(host=host, port=port, **kwargs)
self.headers = urllib3.make_headers(keep_alive=True)
if http_auth is not None:
if isinstance(http_auth, (tuple, list)):
http_auth = ':'.join(http_auth)
self.headers.update(urllib3.make_headers(basic_auth=http_auth))
pool_class = urllib3.HTTPConnectionPool
kw = {}
if use_ssl:
pool_class = urllib3.HTTPSConnectionPool
kw.update({
'ssl_version': ssl_version,
'assert_hostname': ssl_assert_hostname,
'assert_fingerprint': ssl_assert_fingerprint,
})
if verify_certs:
kw.update({
'cert_reqs': 'CERT_REQUIRED',
'ca_certs': ca_certs,
'cert_file': client_cert,
})
elif ca_certs:
raise ImproperlyConfigured("You cannot pass CA certificates when verify SSL is off.")
else:
warnings.warn(
'Connecting to %s using SSL with verify_certs=False is insecure.' % host)
self.pool = pool_class(host, port=port, timeout=self.timeout, maxsize=maxsize, **kw)
def _get_headers(self):
if self.username and self.password:
credentials = ':'.join((self.username, self.password))
return urllib3.make_headers(basic_auth=credentials)
return {}
def __init__(self,
con_pool_size=1,
proxy_url=None,
urllib3_proxy_kwargs=None,
connect_timeout=5.,
read_timeout=5.):
if urllib3_proxy_kwargs is None:
urllib3_proxy_kwargs = dict()
self._connect_timeout = connect_timeout
kwargs = dict(
maxsize=con_pool_size,
cert_reqs='CERT_REQUIRED',
ca_certs=certifi.where(),
socket_options=HTTPConnection.default_socket_options + [
(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
],
timeout=urllib3.Timeout(
connect=self._connect_timeout, read=read_timeout, total=None))
# Set a proxy according to the following order:
# * proxy defined in proxy_url (+ urllib3_proxy_kwargs)
# * proxy set in `HTTPS_PROXY` env. var.
# * proxy set in `https_proxy` env. var.
# * None (if no proxy is configured)
if not proxy_url:
proxy_url = os.environ.get('HTTPS_PROXY') or os.environ.get('https_proxy')
if not proxy_url:
if urllib3.contrib.appengine.is_appengine_sandbox():
# Use URLFetch service if running in App Engine
mgr = urllib3.contrib.appengine.AppEngineManager()
else:
mgr = urllib3.PoolManager(**kwargs)
else:
kwargs.update(urllib3_proxy_kwargs)
if proxy_url.startswith('socks'):
try:
from urllib3.contrib.socks import SOCKSProxyManager
except ImportError:
raise RuntimeError('PySocks is missing')
mgr = SOCKSProxyManager(proxy_url, **kwargs)
else:
mgr = urllib3.proxy_from_url(proxy_url, **kwargs)
if mgr.proxy.auth:
# TODO: what about other auth types?
auth_hdrs = urllib3.make_headers(proxy_basic_auth=mgr.proxy.auth)
mgr.proxy_headers.update(auth_hdrs)
self._con_pool = mgr
def clear(self):
'''
couchbase-cli
To clear all data in bucket. use command couchbase-cli on Couchbase Server,
/opt/couchbase/bin/couchbase-cli bucket-flush -u admin -p password -c 127.0.0.1:8091 -b bucket --force
Couchbase Command Line tools
folder /opt/couchbase/lib/python could be copied to other host
or get from GitHub
> git clone https://github.com/couchbase/couchbase-cli.git
execute:
> python couchbase-cli bucket-flush -u admin -p password -c 192.168.12.13:8091 -b bucket --force
import os,os.path
#if self._couchbase_cli == '':
if True:
log.error( 'CouchbaseError: please using couchbase-cli instead\n'
'couchbase-cli on GitHub\n'
'https://github.com/couchbase/couchbase-cli.git' )
return False
else:
# Always 'Permission denied' ?
return os.system( 'python %s bucket-flush -u %s -p %s -c %s -b %s --force'%
( path.normpath( self._couchbase_cli + '/couchbaase-cli' ),
self._options.get( 'admin','' ),
self._options.get( 'admin-pwd', '' ),
self._server[0], self._bucket ) )== 0
'''
import urllib3
conn = urllib3.connection_from_url(self._server[0], block=True, maxsize=100)
endpoint = '/pools/default/buckets/%s/controller/doFlush' % self._bucket
res = conn.urlopen(url=endpoint, method='POST', headers=urllib3.make_headers(basic_auth = self._options.get( 'admin:pwd', '' ),))
if len(res.data) == 0:
return True
else:
# '{"_":"Flush is disabled for the bucket"}'
# 'Requested resource not found.\r\n' REST error
try:
j = json.loads( res.data )
msg = j.get( '-' )
except Exception as e:
msg = res.data
log.error( "CouchbaseError: clear fail..: %s" % ( msg ) )
return False
