/usr/lib/python2.7/dist-packages/swift/common/middleware/cname_lookup.py is in python-swift 2.7.0-0ubuntu2.
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 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 | # Copyright (c) 2010-2012 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
CNAME Lookup Middleware
Middleware that translates an unknown domain in the host header to
something that ends with the configured storage_domain by looking up
the given domain's CNAME record in DNS.
This middleware will continue to follow a CNAME chain in DNS until it finds
a record ending in the configured storage domain or it reaches the configured
maximum lookup depth. If a match is found, the environment's Host header is
rewritten and the request is passed further down the WSGI chain.
"""
from six.moves import range
import socket
from swift import gettext_ as _
try:
import dns.resolver
from dns.exception import DNSException
from dns.resolver import NXDOMAIN, NoAnswer
except ImportError:
# catch this to allow docs to be built without the dependency
MODULE_DEPENDENCY_MET = False
else: # executed if the try block finishes with no errors
MODULE_DEPENDENCY_MET = True
from swift.common.swob import Request, HTTPBadRequest
from swift.common.utils import cache_from_env, get_logger, list_from_csv
def lookup_cname(domain): # pragma: no cover
"""
Given a domain, returns its DNS CNAME mapping and DNS ttl.
:param domain: domain to query on
:returns: (ttl, result)
"""
try:
answer = dns.resolver.query(domain, 'CNAME').rrset
ttl = answer.ttl
result = answer.items[0].to_text()
result = result.rstrip('.')
return ttl, result
except (DNSException, NXDOMAIN, NoAnswer):
return 0, None
def is_ip(domain):
try:
socket.inet_pton(socket.AF_INET, domain)
return True
except socket.error:
try:
socket.inet_pton(socket.AF_INET6, domain)
return True
except socket.error:
return False
class CNAMELookupMiddleware(object):
"""
CNAME Lookup Middleware
See above for a full description.
:param app: The next WSGI filter or app in the paste.deploy
chain.
:param conf: The configuration dict for the middleware.
"""
def __init__(self, app, conf):
if not MODULE_DEPENDENCY_MET:
# reraise the exception if the dependency wasn't met
raise ImportError('dnspython is required for this module')
self.app = app
storage_domain = conf.get('storage_domain', 'example.com')
self.storage_domain = ['.' + s for s in
list_from_csv(storage_domain)
if not s.startswith('.')]
self.storage_domain += [s for s in list_from_csv(storage_domain)
if s.startswith('.')]
self.lookup_depth = int(conf.get('lookup_depth', '1'))
self.memcache = None
self.logger = get_logger(conf, log_route='cname-lookup')
def _domain_endswith_in_storage_domain(self, a_domain):
for domain in self.storage_domain:
if a_domain.endswith(domain):
return True
return False
def __call__(self, env, start_response):
if not self.storage_domain:
return self.app(env, start_response)
if 'HTTP_HOST' in env:
given_domain = env['HTTP_HOST']
else:
given_domain = env['SERVER_NAME']
port = ''
if ':' in given_domain:
given_domain, port = given_domain.rsplit(':', 1)
if is_ip(given_domain):
return self.app(env, start_response)
a_domain = given_domain
if not self._domain_endswith_in_storage_domain(a_domain):
if self.memcache is None:
self.memcache = cache_from_env(env)
error = True
for tries in range(self.lookup_depth):
found_domain = None
if self.memcache:
memcache_key = ''.join(['cname-', a_domain])
found_domain = self.memcache.get(memcache_key)
if not found_domain:
ttl, found_domain = lookup_cname(a_domain)
if self.memcache:
memcache_key = ''.join(['cname-', given_domain])
self.memcache.set(memcache_key, found_domain,
time=ttl)
if found_domain is None or found_domain == a_domain:
# no CNAME records or we're at the last lookup
error = True
found_domain = None
break
elif self._domain_endswith_in_storage_domain(found_domain):
# Found it!
self.logger.info(
_('Mapped %(given_domain)s to %(found_domain)s') %
{'given_domain': given_domain,
'found_domain': found_domain})
if port:
env['HTTP_HOST'] = ':'.join([found_domain, port])
else:
env['HTTP_HOST'] = found_domain
error = False
break
else:
# try one more deep in the chain
self.logger.debug(
_('Following CNAME chain for '
'%(given_domain)s to %(found_domain)s') %
{'given_domain': given_domain,
'found_domain': found_domain})
a_domain = found_domain
if error:
if found_domain:
msg = 'CNAME lookup failed after %d tries' % \
self.lookup_depth
else:
msg = 'CNAME lookup failed to resolve to a valid domain'
resp = HTTPBadRequest(request=Request(env), body=msg,
content_type='text/plain')
return resp(env, start_response)
return self.app(env, start_response)
def filter_factory(global_conf, **local_conf): # pragma: no cover
conf = global_conf.copy()
conf.update(local_conf)
def cname_filter(app):
return CNAMELookupMiddleware(app, conf)
return cname_filter
|