-
Notifications
You must be signed in to change notification settings - Fork 0
/
kerb_monkey.py
61 lines (48 loc) · 2.26 KB
/
kerb_monkey.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
import urllib3
import os
from requests_kerberos import HTTPKerberosAuth
try:
import botocore
except ImportError:
botocore = None
pass
try:
import requests
except ImportError:
requests = None
pass
def patch_urllib3_kerb():
'''
This will patch urllib3 to inject a kerberos authentication header into the proxies.
If the auth details are left blank (IE: https://:@proxy_url/) then it will use kerberos
it does this at the ConnectionPool level, so will work for most libraries we can't directly
integrate with
'''
old_proxy_from_url = urllib3.poolmanager.proxy_from_url
def proxy_from_url(proxy_url, **kw):
if(not proxy_url.startswith("http://:@") and not proxy_url.startswith("https://:@")):
return old_proxy_from_url(proxy_url, **kw)
if(not 'proxy_headers' in kw):
kw['proxy_headers'] = {}
proxy_headers = kw['proxy_headers']
if(not 'Proxy-Authorization' in proxy_headers):
if isinstance(proxy_url, urllib3.connectionpool.HTTPConnectionPool):
proxy_url = '%s://%s:%i' % (proxy_url.scheme, proxy_url.host,
proxy_url.port)
proxy = urllib3.util.url.parse_url(proxy_url)
kerb_auth = HTTPKerberosAuth(force_preemptive=True)
auth_header = kerb_auth.generate_request_header(None, proxy.host, True)
proxy_headers['Proxy-Authorization'] = auth_header
return urllib3.poolmanager.ProxyManager(proxy_url=proxy_url, **kw)
urllib3.poolmanager.proxy_from_url = proxy_from_url
##Because requests statically import proxy_from_url, we're going to do this here
if(requests):
requests.adapters.proxy_from_url = proxy_from_url
##Because boto statically import proxy_from_url, we're going to do this here
if(botocore):
botocore.httpsession.proxy_from_url = proxy_from_url
def _create_auth_header(proxy_host, proxy_headers):
if(not 'Proxy-Authorization' in proxy_headers):
kerb_auth = HTTPKerberosAuth(force_preemptive=True)
auth_header = kerb_auth.generate_request_header(None, proxy_host, True)
proxy_headers['Proxy-Authorization'] = auth_header