Apply HTTP security policies consistently in HTTPD and WSGI.
This class centralizes the decision to accept or reject a request, based on: - temporary ban of the addresses that met a forbidden URL path - blacklist of user agents - blacklist of URL paths - signed URL verification (missing/invalid/expired)
It must be instantiated once and configured once. The temporary list of banned addresses is the only state: it is written in a file while the server runs, not in this instance.
Create the policy with default (disabled) configuration.
def __init__(self):
"""Create the policy with default (disabled) configuration.
"""
self.__blacklist_agent = BlackListAgent()
self.__blacklist_path = BlackListPath()
self.__banlist = BanList()
self.__signed_url = SignedURL()
self.__blacklist_agent_enabled = False
self.__blacklist_path_enabled = False
self.__banlist_enabled = False
self.__signed_url_cfg = {'ttl': None, 'protect': []}
Configure the policy from a configuration dict.
Expected keys: - "blacklist": optional, the user agents to reject - "blacklistpath": optional, the URL paths to reject - "banlist": optional, the file and the duration of the temporary ban - "signedurl": optional (filepath or json dict, depending on SignedURL.load)
If a key is missing, the corresponding policy is disabled.
def configure(self, config: dict) -> None:
"""Configure the policy from a configuration dict.
Expected keys:
- "blacklist": optional, the user agents to reject
- "blacklist_path": optional, the URL paths to reject
- "banlist": optional, the file and the duration of the temporary ban
- "signed_url": optional (filepath or json dict, depending on SignedURL.load)
If a key is missing, the corresponding policy is disabled.
:param config: (dict) The configuration data.
"""
if type(config) is not dict:
raise TypeError('HTTPDPolicy.configure: config must be a dict.')
self.__blacklist_agent_enabled = False
if 'blacklist' in config and config['blacklist'] is not None:
self.__blacklist_agent.configure(config, 'blacklist')
self.__blacklist_agent_enabled = True
self.__blacklist_path_enabled = False
if 'blacklist_path' in config and config['blacklist_path'] is not None:
self.__blacklist_path.configure(config, 'blacklist_path')
self.__blacklist_path_enabled = True
self.__banlist_enabled = False
if 'banlist' in config and config['banlist'] is not None:
self.__banlist.configure(config, 'banlist')
self.__banlist_enabled = True
self.__signed_url_cfg = {'ttl': None, 'protect': []}
if 'signed_url' in config and config['signed_url'] is not None:
self.__signed_url_cfg = self.__signed_url.configure(config, 'signed_url')
Check the request against policies and return decision and response.
Returned values are: - allowed: (bool) True if request is accepted - content: (bytes|None) Response body if rejected, otherwise None - status: (HTTPDStatus|None) Status if rejected, otherwise None - mime_type: (str|None) Mime of the body, None if the body is empty
def check(self, path: str, query_string: str, headers, client_address: str) -> tuple:
"""Check the request against policies and return decision and response.
Returned values are:
- allowed: (bool) True if request is accepted
- content: (bytes|None) Response body if rejected, otherwise None
- status: (HTTPDStatus|None) Status if rejected, otherwise None
- mime_type: (str|None) Mime of the body, None if the body is empty
:param path: (str) Normalized URL path (without query).
:param query_string: (str) Raw query string (without '?').
:param headers: (Any) Request headers or environ. Must support ".get(...)".
:param client_address: (str) The address of the client of the request.
:return: (tuple) (allowed, content, status, mime_type)
"""
if type(path) is not str:
raise TypeError('HTTPDPolicy.check: path must be a string.')
if type(query_string) is not str:
raise TypeError('HTTPDPolicy.check: query_string must be a string.')
if type(client_address) is not str:
raise TypeError('HTTPDPolicy.check: client_address must be a string.')
user_agent = self._get_user_agent(headers)
client = self._get_client_address(headers, client_address)
if self.__banlist_enabled is True:
if self.__banlist.match(client_address) is True:
logging.warning("Banned address rejected: '{:s}' for path '{:s}'".format(client_address, path))
content, status = HTTPDHandlerUtils.blacklisted_page_answer()
return (False, content, status, None)
if self.__blacklist_agent_enabled is True:
if self.__blacklist_agent.match(user_agent) is True:
logging.warning("Blacklisted user agent rejected: '{:s}' for path '{:s}' from {:s}".format(user_agent, path, client))
content, status = HTTPDHandlerUtils.blacklisted_page_answer()
return (False, content, status, None)
if self.__blacklist_path_enabled is True:
if self.__blacklist_path.match(path) is True:
logging.warning("Blacklisted path rejected: '{:s}' for user agent '{:s}' from {:s}".format(path, user_agent, client))
if self.__banlist_enabled is True:
self.__banlist.add(client_address)
content, status = HTTPDHandlerUtils.blacklisted_page_answer()
return (False, content, status, None)
ttl_seconds = self.__signed_url_cfg.get('ttl', None)
if ttl_seconds is not None:
protect = self.__signed_url_cfg.get('protect', [])
if self.__signed_url.match_protect(path, protect) is True:
if self.__signed_url.verify(path, query_string, ttl_seconds) is False:
content, status = HTTPDHandlerUtils.signed_url_page_answer()
return (False, content, status, 'text/html')
return (True, None, None, None)
Finalize an outgoing HTML page.
This method applies outbound policies that must not be implemented inside Response classes.
Current behavior: - If signed URLs are enabled, sign protected links found in: - href="..." - action="..."
def finalize_html(self, content: bytes) -> bytes:
"""Finalize an outgoing HTML page.
This method applies outbound policies that must not be implemented
inside Response classes.
Current behavior:
- If signed URLs are enabled, sign protected links found in:
- href="..."
- action="..."
:param content: (bytes) HTML content.
:return: (bytes) Updated HTML content.
"""
if isinstance(content, (bytes, bytearray)) is False:
raise TypeError('HTTPDPolicy.finalize_html: content must be bytes.')
ttl_seconds = self.__signed_url_cfg.get('ttl', None)
if ttl_seconds is None:
return content
protect = self.__signed_url_cfg.get('protect', [])
if isinstance(protect, list) is False or len(protect) == 0:
return content
html = content.decode('utf-8', errors='replace')
def _should_skip(value: str) -> bool:
if len(value) == 0:
return True
v = value.lower()
if v.startswith('#'):
return True
if v.startswith('http://') or v.startswith('https://'):
return True
if v.startswith('mailto:'):
return True
if v.startswith('javascript:'):
return True
if v.startswith('data:'):
return True
return False
def _is_already_signed(query: str) -> bool:
if len(query) == 0:
return False
q = ('&' + query).lower()
return '&ts=' in q and '&sig=' in q
def _sign_value(value: str) -> str:
if _should_skip(value) is True:
return value
parts = urlsplit(value)
path = parts.path
if self.__signed_url.match_protect(path, protect) is False:
return value
if _is_already_signed(parts.query) is True:
return value
extra_params = []
if len(parts.query) > 0:
for k, v in parse_qsl(parts.query, keep_blank_values=True):
if k in ('ts', 'sig'):
continue
extra_params.append((k, v))
signed = self.__signed_url.sign(path, ttl_seconds)
signed_parts = urlsplit(signed)
final_query = signed_parts.query
if len(extra_params) > 0:
extra = urlencode(extra_params, doseq=True)
if len(final_query) > 0:
final_query = final_query + '&' + extra
else:
final_query = extra
return urlunsplit((signed_parts.scheme or parts.scheme, signed_parts.netloc or parts.netloc, signed_parts.path, final_query, parts.fragment))
def _repl(m):
attr = m.group(1)
quote = m.group(2)
val = m.group(3)
return attr + '=' + quote + _sign_value(val) + quote
html = re.sub('(href|action)\\s*=\\s*([\\\'"])(.*?)\\2', _repl, html, flags=re.IGNORECASE)
return html.encode('utf-8')
Return the address of the client, and the relayed one if any.
A relay in front of the server announces the address of the visitor in the "X-Forwarded-For" header. Both are returned, because only the first one is the address the ban list records.
def _get_client_address(self, headers, client_address: str) -> str:
"""Return the address of the client, and the relayed one if any.
A relay in front of the server announces the address of the visitor
in the "X-Forwarded-For" header. Both are returned, because only
the first one is the address the ban list records.
:param headers: (Any) Object or dict supporting get().
:param client_address: (str) The address the request came from.
:return: (str) The address, and the relayed one between parentheses.
"""
forwarded = self._get_forwarded_for(headers)
if len(forwarded) == 0:
return client_address
return client_address + ' (X-Forwarded-For: ' + forwarded + ')'
Extract X-Forwarded-For from headers/environ.
def _get_forwarded_for(self, headers) -> str:
"""Extract X-Forwarded-For from headers/environ.
:param headers: (Any) Object or dict supporting get().
:return: (str) X-Forwarded-For or empty string.
"""
if headers is None:
return ''
for key in ('X-Forwarded-For', 'HTTP_X_FORWARDED_FOR'):
try:
value = headers.get(key, '')
except Exception:
return ''
if type(value) is str and len(value) > 0:
return value
return ''
Extract User-Agent from headers/environ.
def _get_user_agent(self, headers) -> str:
"""Extract User-Agent from headers/environ.
:param headers: (Any) Object or dict supporting get().
:return: (str) User-Agent or empty string.
"""
if headers is None:
return ''
try:
ua = headers.get('User-Agent', '')
if type(ua) is str and len(ua) > 0:
return ua
except:
pass
try:
ua = headers.get('HTTP_USER_AGENT', '')
if type(ua) is str and len(ua) > 0:
return ua
except:
pass
return ''
def _should_skip(value: str) -> bool:
if len(value) == 0:
return True
v = value.lower()
if v.startswith('#'):
return True
if v.startswith('http://') or v.startswith('https://'):
return True
if v.startswith('mailto:'):
return True
if v.startswith('javascript:'):
return True
if v.startswith('data:'):
return True
return False
def _is_already_signed(query: str) -> bool:
if len(query) == 0:
return False
q = ('&' + query).lower()
return '&ts=' in q and '&sig=' in q
def _sign_value(value: str) -> str:
if _should_skip(value) is True:
return value
parts = urlsplit(value)
path = parts.path
if self.__signed_url.match_protect(path, protect) is False:
return value
if _is_already_signed(parts.query) is True:
return value
extra_params = []
if len(parts.query) > 0:
for k, v in parse_qsl(parts.query, keep_blank_values=True):
if k in ('ts', 'sig'):
continue
extra_params.append((k, v))
signed = self.__signed_url.sign(path, ttl_seconds)
signed_parts = urlsplit(signed)
final_query = signed_parts.query
if len(extra_params) > 0:
extra = urlencode(extra_params, doseq=True)
if len(final_query) > 0:
final_query = final_query + '&' + extra
else:
final_query = extra
return urlunsplit((signed_parts.scheme or parts.scheme, signed_parts.netloc or parts.netloc, signed_parts.path, final_query, parts.fragment))
def _repl(m):
attr = m.group(1)
quote = m.group(2)
val = m.group(3)
return attr + '=' + quote + _sign_value(val) + quote