openstack/oslo.config
Code Issues Proposed changes

Bump bandit and make oslo.config compatible with latest rules

Browse Source 
Apply a timeout to requests calls to avoid uncontrolled
resource consumption (CWE400) [1].

[1] https://cwe.mitre.org/data/definitions/400.html

Change-Id: I9e3c1e5f98e2ecfb5564d8dbd608d19d4a66cfce
This commit is contained in:
Hervé Beraud 2023-03-10 10:21:35 +01:00
parent 1654dcc91c
commit 3dff974eea
2 changed files with 17 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
oslo_config/sources/_uri.py
View File

@ -107,6 +107,13 @@ class URIConfigurationSourceDriver(sources.ConfigurationSourceDriver):
help=('Client side private key, in case client_cert is '
'specified but does not includes the private key.'),
),
cfg.StrOpt(
'timeout',
default=60,
help=('Timeout is the number of seconds the request will wait '
'for your client to establish a connection to a remote '
'machine call on the socket.'),
),
]
def list_options_for_discovery(self):
@ -119,7 +126,8 @@ class URIConfigurationSourceDriver(sources.ConfigurationSourceDriver):
conf[group_name].uri,
conf[group_name].ca_path,
conf[group_name].client_cert,
conf[group_name].client_key)
conf[group_name].client_key,
conf[group_name].timeout)
class URIConfigurationSource(sources.ConfigurationSource):
@ -139,11 +147,12 @@ class URIConfigurationSource(sources.ConfigurationSource):
specified but does not includes the private key.
"""
def __init__(self, uri, ca_path=None, client_cert=None, client_key=None):
def __init__(self, uri, ca_path=None, client_cert=None, client_key=None,
timeout=60):
self._uri = uri
self._namespace = cfg._Namespace(cfg.ConfigOpts())
data = self._fetch_uri(uri, ca_path, client_cert, client_key)
data = self._fetch_uri(uri, ca_path, client_cert, client_key, timeout)
with tempfile.NamedTemporaryFile() as tmpfile:
tmpfile.write(data.encode("utf-8"))
@ -151,12 +160,14 @@ class URIConfigurationSource(sources.ConfigurationSource):
cfg.ConfigParser._parse_file(tmpfile.name, self._namespace)
def _fetch_uri(self, uri, ca_path, client_cert, client_key):
def _fetch_uri(self, uri, ca_path, client_cert, client_key,
timeout):
verify = ca_path if ca_path else True
cert = (client_cert, client_key) if client_cert and client_key else \
client_cert
with requests.get(uri, verify=verify, cert=cert) as response:
with requests.get(uri, verify=verify, cert=cert,
timeout=timeout) as response:
response.raise_for_status() # raises only in case of HTTPError
return response.text

2
test-requirements.txt
View File

@ -24,6 +24,6 @@ coverage!=4.4,>=4.0 # Apache-2.0
requests_mock>=1.5.0 # Apache-2.0
# Bandit security code scanner
bandit>=1.6.0,<1.7.0 # Apache-2.0
bandit>=1.7.0,<1.8.0 # Apache-2.0
pre-commit>=2.6.0 # MIT