saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Merge 3006.x into master

Browse source
This commit is contained in:
Pedro Algarvio 2023-11-29 18:31:52 +00:00
commit 53acabebdb
2 changed files with 401 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

138
tests/pytests/functional/utils/test_http.py
View file

@ -1,6 +1,9 @@
import shutil
import tarfile
import pytest
from pytestshellutils.utils import ports
from saltfactories.utils import random_string
import salt.utils.http
@ -14,3 +17,138 @@ def test_decode_body(webserver, integration_files_dir, backend):
webserver.url("test.tar.gz"), backend=backend, decode_body=False
)
assert isinstance(ret["body"], bytes)
pytestmark = [
pytest.mark.slow_test,
pytest.mark.skip_if_binaries_missing("docker", "dockerd", check_all=False),
]
@pytest.fixture(scope="module")
def tinyproxy_port():
return ports.get_unused_localhost_port()
@pytest.fixture(scope="module")
def tinyproxy_user():
return random_string("tinyproxy-user-")
@pytest.fixture(scope="module")
def tinyproxy_pass():
return random_string("tinyproxy-pass-")
@pytest.fixture(params=[True, False], ids=lambda x: "basic-auth" if x else "no-auth")
def tinyproxy_basic_auth(request):
return request.param
@pytest.fixture(params=[True, False], ids=lambda x: "no-proxy" if x else "with-proxy")
def no_proxy(request):
return request.param
@pytest.fixture(params=["POST", "GET"], ids=lambda x: x)
def http_method(request):
return request.param
@pytest.fixture(scope="module")
def tinyproxy_dir(tmp_path_factory):
try:
dirname = tmp_path_factory.mktemp("tinyproxy")
yield dirname
finally:
shutil.rmtree(dirname, ignore_errors=True)
@pytest.fixture
def tinyproxy_conf(
tinyproxy_dir, tinyproxy_port, tinyproxy_user, tinyproxy_pass, tinyproxy_basic_auth
):
basic_auth = (
f"\nBasicAuth {tinyproxy_user} {tinyproxy_pass}" if tinyproxy_basic_auth else ""
)
conf = """Port {port}
Listen 127.0.0.1
Timeout 600
Allow 127.0.0.1
AddHeader "X-Tinyproxy-Header" "Test custom tinyproxy header"{auth}
""".format(
port=tinyproxy_port, auth=basic_auth
)
(tinyproxy_dir / "tinyproxy.conf").write_text(conf)
@pytest.fixture
def tinyproxy_container(
salt_factories,
tinyproxy_conf,
tinyproxy_dir,
):
container = salt_factories.get_container(
"tinyproxy",
image_name="ghcr.io/saltstack/salt-ci-containers/tinyproxy:latest",
container_run_kwargs={
"network_mode": "host",
"volumes": {str(tinyproxy_dir): {"bind": "/etc/tinyproxy", "mode": "z"}},
},
pull_before_start=True,
skip_on_pull_failure=True,
skip_if_docker_client_not_connectable=True,
)
with container.started() as factory:
yield factory
@pytest.mark.parametrize("backend", ["requests", "tornado", "urllib2"])
def test_real_proxy(
tinyproxy_container,
httpserver,
tinyproxy_port,
tinyproxy_user,
tinyproxy_pass,
backend,
tinyproxy_basic_auth,
no_proxy,
http_method,
):
data = b"mydatahere"
opts = {
"proxy_host": "localhost",
"proxy_port": tinyproxy_port,
}
if tinyproxy_basic_auth:
opts.update(
{
"proxy_username": tinyproxy_user,
"proxy_password": tinyproxy_pass,
}
)
# Expecting the headers allows verification that it went through the proxy without looking at the logs
if no_proxy:
opts["no_proxy"] = ["random.hostname.io", httpserver.host]
httpserver.expect_request(
"/real_proxy_test",
).respond_with_data(data, content_type="application/octet-stream")
else:
httpserver.expect_request(
"/real_proxy_test",
headers={"X-Tinyproxy-Header": "Test custom tinyproxy header"},
).respond_with_data(data, content_type="application/octet-stream")
url = httpserver.url_for("/real_proxy_test").replace("localhost", "127.0.0.1")
# We just want to be sure that it's using the proxy
ret = salt.utils.http.query(
url,
method=http_method,
data=data,
backend=backend,
opts=opts,
decode_body=False,
)
body = ret.get("body", "")
assert body == data

269
tests/pytests/unit/utils/test_http.py
View file

@ -1,7 +1,9 @@
import pytest
import requests
from pytestshellutils.utils import ports
from werkzeug.wrappers import Response # pylint: disable=3rd-party-module-not-gated
import salt.utils.http
import salt.utils.http as http
from tests.support.mock import MagicMock, patch
@ -16,10 +18,10 @@ def test_requests_session_verify_ssl_false(ssl_webserver, integration_files_dir)
if verify is True or verify is None:
with pytest.raises(requests.exceptions.SSLError) as excinfo:
session = salt.utils.http.session(**kwargs)
session = http.session(**kwargs)
ret = session.get(ssl_webserver.url("this.txt"))
else:
session = salt.utils.http.session(**kwargs)
session = http.session(**kwargs)
ret = session.get(ssl_webserver.url("this.txt"))
assert ret.status_code == 200
@ -29,7 +31,7 @@ def test_session_ca_bundle_verify_false():
test salt.utils.http.session when using
both ca_bunlde and verify_ssl false
"""
ret = salt.utils.http.session(ca_bundle="/tmp/test_bundle", verify_ssl=False)
ret = http.session(ca_bundle="/tmp/test_bundle", verify_ssl=False)
assert ret is False
@ -38,7 +40,7 @@ def test_session_headers():
test salt.utils.http.session when setting
headers
"""
ret = salt.utils.http.session(headers={"Content-Type": "application/json"})
ret = http.session(headers={"Content-Type": "application/json"})
assert ret.headers["Content-Type"] == "application/json"
@ -49,5 +51,260 @@ def test_session_ca_bundle():
fpath = "/tmp/test_bundle"
patch_os = patch("os.path.exists", MagicMock(return_value=True))
with patch_os:
ret = salt.utils.http.session(ca_bundle=fpath)
ret = http.session(ca_bundle=fpath)
assert ret.verify == fpath
def test_sanitize_url_hide_fields_none():
"""
Tests sanitizing a url when the hide_fields kwarg is None.
"""
mock_url = "https://api.testing.com/?&foo=bar&test=testing"
ret = http.sanitize_url(mock_url, hide_fields=None)
assert ret == mock_url
def test_sanitize_url_no_elements():
"""
Tests sanitizing a url when no elements should be sanitized.
"""
mock_url = "https://api.testing.com/?&foo=bar&test=testing"
ret = http.sanitize_url(mock_url, [""])
assert ret == mock_url
def test_sanitize_url_single_element():
"""
Tests sanitizing a url with only a single element to be sanitized.
"""
mock_url = (
"https://api.testing.com/?&keep_it_secret=abcdefghijklmn"
"&api_action=module.function"
)
mock_ret = (
"https://api.testing.com/?&keep_it_secret=XXXXXXXXXX&"
"api_action=module.function"
)
ret = http.sanitize_url(mock_url, ["keep_it_secret"])
assert ret == mock_ret
def test_sanitize_url_multiple_elements():
"""
Tests sanitizing a url with multiple elements to be sanitized.
"""
mock_url = (
"https://api.testing.com/?rootPass=badpassword%21"
"&skipChecks=True&api_key=abcdefghijklmn"
"&NodeID=12345&api_action=module.function"
)
mock_ret = (
"https://api.testing.com/?rootPass=XXXXXXXXXX"
"&skipChecks=True&api_key=XXXXXXXXXX"
"&NodeID=12345&api_action=module.function"
)
ret = http.sanitize_url(mock_url, ["api_key", "rootPass"])
assert ret == mock_ret
# _sanitize_components tests
def test_sanitize_components_no_elements():
"""
Tests when zero elements need to be sanitized.
"""
mock_component_list = ["foo=bar", "bar=baz", "hello=world"]
mock_ret = "foo=bar&bar=baz&hello=world&"
ret = http._sanitize_url_components(mock_component_list, "api_key")
assert ret == mock_ret
def test_sanitize_components_one_element():
"""
Tests a single component to be sanitized.
"""
mock_component_list = ["foo=bar", "api_key=abcdefghijklmnop"]
mock_ret = "foo=bar&api_key=XXXXXXXXXX&"
ret = http._sanitize_url_components(mock_component_list, "api_key")
assert ret == mock_ret
def test_sanitize_components_multiple_elements():
"""
Tests two componenets to be sanitized.
"""
mock_component_list = ["foo=bar", "foo=baz", "api_key=testing"]
mock_ret = "foo=XXXXXXXXXX&foo=XXXXXXXXXX&api_key=testing&"
ret = http._sanitize_url_components(mock_component_list, "foo")
assert ret == mock_ret
@pytest.mark.slow_test
def test_query_null_response():
"""
This tests that we get a null response when raise_error=False and the
host/port cannot be reached.
"""
host = "127.0.0.1"
port = ports.get_unused_localhost_port()
url = "http://{host}:{port}/".format(host=host, port=port)
result = http.query(url, raise_error=False)
assert result == {"body": None}, result
def test_query_error_handling():
ret = http.query("http://127.0.0.1:0")
assert isinstance(ret, dict)
assert isinstance(ret.get("error", None), str)
ret = http.query("http://myfoobardomainthatnotexist")
assert isinstance(ret, dict)
assert isinstance(ret.get("error", None), str)
def test_parse_cookie_header():
header = "; ".join(
[
"foo=bar",
"expires=Mon, 03-Aug-20 14:26:27 GMT",
"path=/",
"domain=.mydomain.tld",
"HttpOnly",
"SameSite=Lax",
"Secure",
]
)
ret = http.parse_cookie_header(header)
cookie = ret.pop(0)
assert cookie.name == "foo", cookie.name
assert cookie.value == "bar", cookie.value
assert cookie.expires == 1596464787, cookie.expires
assert cookie.path == "/", cookie.path
assert cookie.domain == ".mydomain.tld", cookie.domain
assert cookie.secure
# Only one cookie should have been returned, if anything is left in the
# parse_cookie_header return then something went wrong.
assert not ret
@pytest.mark.requires_network
def test_requests_multipart_formdata_post(httpserver):
"""
Test handling of a multipart/form-data POST using the requests backend
"""
match_this = (
"{0}\r\nContent-Disposition: form-data;"
' name="fieldname_here"\r\n\r\nmydatahere\r\n{0}--\r\n'
)
def mirror_post_handler(request):
return Response(request.data)
httpserver.expect_request(
"/multipart_form_data",
).respond_with_handler(mirror_post_handler)
url = httpserver.url_for("/multipart_form_data")
ret = http.query(
url,
method="POST",
data="mydatahere",
formdata=True,
formdata_fieldname="fieldname_here",
backend="requests",
)
body = ret.get("body", "")
boundary = body[: body.find("\r")]
assert body == match_this.format(boundary)
def test_query_proxy(httpserver):
"""
Test http.query with tornado and with proxy opts set
and then test with no_proxy set to ensure we dont
run into issue #55192 again.
"""
data = "mydatahere"
opts = {
"proxy_host": "127.0.0.1",
"proxy_port": 88,
"proxy_username": "salt_test",
"proxy_password": "super_secret",
}
with patch("requests.Session") as mock_session:
mock_session.return_value = MagicMock()
ret = http.query(
"https://fake_url",
method="POST",
data=data,
backend="tornado",
opts=opts,
)
assert mock_session.return_value.proxies == {
"http": "http://salt_test:super_secret@127.0.0.1:88"
}
opts["no_proxy"] = [httpserver.host]
httpserver.expect_request(
"/no_proxy_test",
).respond_with_data(data)
url = httpserver.url_for("/no_proxy_test")
with patch("requests.Session") as mock_session:
mock_session.return_value = MagicMock()
ret = http.query(
url,
method="POST",
data=data,
backend="tornado",
opts=opts,
)
assert not isinstance(mock_session.return_value.proxies, dict)
ret = http.query(url, method="POST", data=data, backend="tornado", opts=opts)
body = ret.get("body", "")
assert body == data
@pytest.mark.parametrize("backend", ["requests", "tornado", "urllib2"])
def test_backends_decode_body_false(httpserver, backend):
"""
test all backends when using
decode_body=False that it returns
bytes and does not try to decode
"""
url = "/test-bytes"
data = b"test-bytes"
httpserver.expect_request(
url,
).respond_with_data(data, content_type="application/octet-stream")
ret = http.query(
httpserver.url_for(url),
backend=backend,
decode_body=False,
)
body = ret.get("body", "")
assert isinstance(body, bytes)
@pytest.mark.parametrize("backend", ["requests", "tornado", "urllib2"])
def test_backends_decode_body_true(httpserver, backend):
"""
test all backends when using decode_body=True that it returns string and decodes it.
"""
url = "/test-decoded-bytes"
data = b"test-decoded-bytes"
httpserver.expect_request(
url,
).respond_with_data(data, content_type="application/octet-stream")
ret = http.query(
httpserver.url_for(url),
backend=backend,
)
body = ret.get("body", "")
assert isinstance(body, str)