/usr/lib/python3/dist-packages/kombu/utils/url.py is in python3-kombu 3.0.33-1ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 62 63 64 | from __future__ import absolute_import
from functools import partial
try:
from urllib.parse import parse_qsl, quote, unquote, urlparse
except ImportError:
from urllib import quote, unquote # noqa
from urlparse import urlparse, parse_qsl # noqa
from . import kwdict
from kombu.five import string_t
safequote = partial(quote, safe='')
def _parse_url(url):
scheme = urlparse(url).scheme
schemeless = url[len(scheme) + 3:]
# parse with HTTP URL semantics
parts = urlparse('http://' + schemeless)
path = parts.path or ''
path = path[1:] if path and path[0] == '/' else path
return (scheme, unquote(parts.hostname or '') or None, parts.port,
unquote(parts.username or '') or None,
unquote(parts.password or '') or None,
unquote(path or '') or None,
kwdict(dict(parse_qsl(parts.query))))
def parse_url(url):
scheme, host, port, user, password, path, query = _parse_url(url)
return dict(transport=scheme, hostname=host,
port=port, userid=user,
password=password, virtual_host=path, **query)
def as_url(scheme, host=None, port=None, user=None, password=None,
path=None, query=None, sanitize=False, mask='**'):
parts = ['{0}://'.format(scheme)]
if user or password:
if user:
parts.append(safequote(user))
if password:
if sanitize:
parts.extend([':', mask] if mask else [':'])
else:
parts.extend([':', safequote(password)])
parts.append('@')
parts.append(safequote(host) if host else '')
if port:
parts.extend([':', port])
parts.extend(['/', path])
return ''.join(str(part) for part in parts if part)
def sanitize_url(url, mask='**'):
return as_url(*_parse_url(url), sanitize=True, mask=mask)
def maybe_sanitize_url(url, mask='**'):
if isinstance(url, string_t) and '://' in url:
return sanitize_url(url, mask)
return url
|