diff --git a/app/api/network.py b/app/api/network.py
index 0e95f45420f5c63d5258c4b1ffdb98e2af182262..144d5c5bac1993de22c865ea1f2126f3725927ec 100644
--- a/app/api/network.py
+++ b/app/api/network.py
@@ -104,25 +104,19 @@ def get_interfaces():
.. :quickref: Network; Get interfaces
"""
+ query = models.Interface.query
+ query = query.join(models.Interface.network).order_by(models.Interface.ip)
+ if not current_user.is_admin:
+ query = query.filter(models.Network.sensitive.is_(False))
domain = request.args.get("domain", None)
if domain is not None:
- query = models.Interface.query
- query = (
- query.join(models.Interface.network)
- .join(models.Network.domain)
- .filter(models.Domain.name == domain)
- )
- query = query.order_by(models.Interface.ip)
+ query = query.join(models.Network.domain).filter(models.Domain.name == domain)
return utils.get_generic_model(model=None, query=query)
network = request.args.get("network", None)
if network is not None:
- query = models.Interface.query
- query = query.join(models.Interface.network).filter(
- models.Network.vlan_name == network
- )
- query = query.order_by(models.Interface.ip)
+ query = query.filter(models.Network.vlan_name == network)
return utils.get_generic_model(model=None, query=query)
- return utils.get_generic_model(models.Interface, order_by=models.Interface.ip)
+ return utils.get_generic_model(model=models.Interface, base_query=query)
@bp.route("/interfaces", methods=["POST"])
@@ -251,7 +245,18 @@ def get_hosts():
.. :quickref: Network; Get hosts
"""
- return utils.get_generic_model(models.Host, order_by=models.Host.name)
+ query = models.Host.query
+ if not current_user.is_admin:
+ # Note that hosts without interface will be filtered out
+ # by this query. This is not an issue as hosts should always
+ # have an interface. They are useless otherwise.
+ query = (
+ query.join(models.Host.interfaces)
+ .join(models.Interface.network)
+ .filter(models.Network.sensitive.is_(False))
+ )
+ query = query.order_by(models.Host.name)
+ return utils.get_generic_model(model=models.Host, base_query=query)
@bp.route("/hosts/search")
@@ -263,7 +268,7 @@ def search_hosts():
:query q: the search query
"""
- return utils.search_generic_model(models.Host)
+ return utils.search_generic_model(models.Host, filter_sensitive=True)
@bp.route("/hosts", methods=["POST"])
diff --git a/app/api/utils.py b/app/api/utils.py
index 15f389b6d9363d4dc66f9f389fa3f96ae231f4f8..8468eb844294e2dd30aae8554b730980fb3aa821 100644
--- a/app/api/utils.py
+++ b/app/api/utils.py
@@ -60,11 +60,12 @@ def build_pagination_header(pagination, base_url, **kwargs):
return header
-def get_generic_model(model, order_by=None, query=None):
+def get_generic_model(model, order_by=None, base_query=None, query=None):
"""Return data from model as json
:param model: model class
- :param order_by: column to order the result by
+ :param order_by: column to order the result by (not used if base_query or query is passed)
+ :param query: optional base_query to use [default: model.query]
:param query: optional query to use (for more complex queries)
:returns: data from model as json
"""
@@ -75,10 +76,12 @@ def get_generic_model(model, order_by=None, query=None):
# to query.filter_by in get_query
recursive = kwargs.pop("recursive", "false").lower() == "true"
if query is None:
- query = utils.get_query(model.query, **kwargs)
- if order_by is None:
- order_by = getattr(model, "name")
- query = query.order_by(order_by)
+ if base_query is None:
+ base_query = model.query
+ if order_by is None:
+ order_by = getattr(model, "name")
+ base_query = base_query.order_by(order_by)
+ query = utils.get_query(base_query, model, **kwargs)
pagination = query.paginate(
page, per_page, max_per_page=current_app.config["MAX_PER_PAGE"]
)
@@ -89,10 +92,11 @@ def get_generic_model(model, order_by=None, query=None):
return jsonify(data), 200, header
-def search_generic_model(model):
+def search_generic_model(model, filter_sensitive=False):
"""Return filtered data from model as json
:param model: model class
+ :param bool filter_sensitive: filter out sensitive data if set to True
:returns: filtered data from model as json
"""
kwargs = request.args.to_dict()
@@ -100,7 +104,9 @@ def search_generic_model(model):
per_page = int(kwargs.pop("per_page", 20))
per_page = min(per_page, current_app.config["MAX_PER_PAGE"])
search = kwargs.get("q", "*")
- instances, nb_filtered = model.search(search, page=page, per_page=per_page)
+ instances, nb_filtered = model.search(
+ search, page=page, per_page=per_page, filter_sensitive=filter_sensitive
+ )
current_app.logger.debug(
f'Found {nb_filtered} {model.__tablename__}(s) when searching "{search}"'
)
diff --git a/app/models.py b/app/models.py
index 9dffdd6bf57192576dbb592d772a75964c49d10d..54b0db77495ec61887d85b1f2d496be991cc14ab 100644
--- a/app/models.py
+++ b/app/models.py
@@ -433,7 +433,12 @@ class SearchableMixin(object):
"""Add search capability to a class"""
@classmethod
- def search(cls, query, page=1, per_page=20, sort=None):
+ def search(cls, query, page=1, per_page=20, sort=None, filter_sensitive=False):
+ if filter_sensitive and not current_user.is_admin:
+ if query == "*":
+ query = "sensitive:false"
+ else:
+ query = f"({query}) AND sensitive:false"
try:
ids, total = search.query_index(
cls.__tablename__ + current_app.config["ELASTICSEARCH_INDEX_SUFFIX"],
diff --git a/app/network/views.py b/app/network/views.py
index d4052180ebb2ec36d098c8e7b547f0475cf8e935..1e69490eaec0e3ce1bf4fc11320c788b2f16f51c 100644
--- a/app/network/views.py
+++ b/app/network/views.py
@@ -46,7 +46,9 @@ bp = Blueprint("network", __name__)
@bp.route("/_retrieve_hosts", methods=["POST"])
@login_required
def retrieve_hosts():
- return utils.retrieve_data_for_datatables(request.values, models.Host)
+ return utils.retrieve_data_for_datatables(
+ request.values, models.Host, filter_sensitive=True
+ )
@bp.route("/hosts")
@@ -150,6 +152,9 @@ def delete_host():
@login_required
def view_host(name):
host = models.Host.query.filter_by(name=name).first_or_404()
+ if not current_user.is_admin and host.sensitive:
+ # Only admin users can see hosts on sensitive networks
+ abort(403)
if host.main_interface is None:
flash(f"Host {host.name} has no interface! Add one or delete it.", "warning")
elif host.main_interface.name != host.name:
diff --git a/app/search.py b/app/search.py
index f2a6f0146a2cc5c3f4e949b6a6a08b20bdd8b48f..798446e451b9c01059231d3c045e9a0ddd08cb96 100644
--- a/app/search.py
+++ b/app/search.py
@@ -67,6 +67,7 @@ def query_index(index, query, page=1, per_page=20, sort=None):
}
if sort is not None:
kwargs["sort"] = sort
+ current_app.logger.debug(f"Search: {kwargs}")
search = current_app.elasticsearch.search(**kwargs)
ids = [int(hit["_id"]) for hit in search["hits"]["hits"]]
return ids, search["hits"]["total"]["value"]
diff --git a/app/utils.py b/app/utils.py
index 4ab053f5bd96ea860116baa40cb311bcf87448f3..ce7a9be0ee82f0993f66e46a2f031923a381b479 100644
--- a/app/utils.py
+++ b/app/utils.py
@@ -168,19 +168,27 @@ def get_model_choices(model, allow_none=False, attr="name", query=None, order_by
return choices
-def get_query(query, **kwargs):
+def get_query(query, model, **kwargs):
"""Retrieve the query from the arguments
:param query: sqlalchemy base query
+ :param model: model class
:param kwargs: kwargs from a request
:returns: query filtered by the arguments
"""
- if kwargs:
- try:
- query = query.filter_by(**kwargs)
- except (sa.exc.InvalidRequestError, AttributeError) as e:
- current_app.logger.warning(f"Invalid query arguments: {e}")
- raise CSEntryError("Invalid query arguments", status_code=422)
+ if not kwargs:
+ return query
+ try:
+ # With filter_by(**kwargs), the keyword expressions are extracted
+ # from the primary entity of the query, or the last entity that was
+ # the target of a call to Query.join()
+ # This might not be what we want when join() is used.
+ # Always apply filtering on the given model
+ for key, value in kwargs.items():
+ query = query.filter(getattr(model, key) == value)
+ except (sa.exc.InvalidRequestError, AttributeError) as e:
+ current_app.logger.warning(f"Invalid query arguments: {e}")
+ raise CSEntryError("Invalid query arguments", status_code=422)
return query
@@ -455,7 +463,7 @@ def unique_filename(filename):
return unique
-def retrieve_data_for_datatables(values, model):
+def retrieve_data_for_datatables(values, model, filter_sensitive=False):
"""Return the filtered data of model to datatables
This function is supposed to be called when using datatables
@@ -463,6 +471,7 @@ def retrieve_data_for_datatables(values, model):
:param values: a `~werkzeug.datastructures.MultiDict`, typically request.values
:param model: class of the model to query
+ :param bool filter_sensitive: filter out sensitive data if set to True
:return: json object required by datatables
"""
# Get the parameters from the post data sent by datatables
@@ -487,7 +496,11 @@ def retrieve_data_for_datatables(values, model):
else:
sort = f"{name}.keyword:{order_dir}"
instances, nb_filtered = model.search(
- search, page=page, per_page=per_page, sort=sort
+ search,
+ page=page,
+ per_page=per_page,
+ sort=sort,
+ filter_sensitive=filter_sensitive,
)
data = [instance.to_dict(recursive=True) for instance in instances]
# Total number of items before filtering
diff --git a/tests/functional/test_api.py b/tests/functional/test_api.py
index 5e0c580d6a2967a77e102846febcba313fda955d..ce9be1455d201c4463626889bed9422588f8aeef 100644
--- a/tests/functional/test_api.py
+++ b/tests/functional/test_api.py
@@ -673,7 +673,7 @@ def test_get_networks(client, network_scope_factory, network_factory, admin_toke
# test filtering by address
response = get(
- client, f"{API_URL}/network/networks?address=172.16.20.0/22", token=admin_token,
+ client, f"{API_URL}/network/networks?address=172.16.20.0/22", token=admin_token
)
assert response.status_code == 200
assert len(response.get_json()) == 1
@@ -774,7 +774,7 @@ def test_create_network(client, admin_token, network_scope_factory):
client, f"{API_URL}/network/networks", data=data_same_address, token=admin_token
)
check_response_message(
- response, "172.16.1.0/24 overlaps network1 (172.16.1.0/24)", 422,
+ response, "172.16.1.0/24 overlaps network1 (172.16.1.0/24)", 422
)
data_same_name = {
"vlan_name": "network1",
@@ -1067,6 +1067,63 @@ def test_get_interfaces_with_model(
assert response.get_json()[0]["model"] == "EX3400"
+def test_get_interfaces_with_sensitive_network(
+ client, network_scope_factory, network_factory, interface_factory, host_factory
+):
+ # Create some interfaces
+ scope = network_scope_factory(supernet="192.168.0.0/16")
+ network1 = network_factory(
+ address="192.168.1.0/24",
+ first_ip="192.168.1.10",
+ last_ip="192.168.1.250",
+ scope=scope,
+ )
+ network2 = network_factory(
+ address="192.168.2.0/24",
+ first_ip="192.168.2.10",
+ last_ip="192.168.2.250",
+ scope=scope,
+ sensitive=True,
+ )
+ host1 = host_factory(name="host1")
+ host2 = host_factory(name="host2")
+ host3 = host_factory(name="host3")
+ interface1 = interface_factory(
+ name=host1.name, network=network1, ip="192.168.1.10", host=host1
+ )
+ interface2 = interface_factory(
+ name=host2.name, network=network1, ip="192.168.1.11", host=host2
+ )
+ interface3 = interface_factory(
+ name=host3.name, network=network2, ip="192.168.2.10", host=host3
+ )
+ # Normal users can't get interfaces on sensitive networks
+ token = get_token(client, "user_ro", "userro")
+ response = get(client, f"{API_URL}/network/interfaces", token=token)
+ assert response.status_code == 200
+ assert len(response.get_json()) == 2
+ check_input_is_subset_of_response(
+ response, (interface1.to_dict(), interface2.to_dict())
+ )
+ response = get(client, f"{API_URL}/network/interfaces?name=host1", token=token)
+ assert response.status_code == 200
+ assert len(response.get_json()) == 1
+ response = get(client, f"{API_URL}/network/interfaces?name=host3", token=token)
+ assert response.status_code == 200
+ assert len(response.get_json()) == 0
+ # Admin users can get all interfaces
+ token = get_token(client, "admin", "adminpasswd")
+ response = get(client, f"{API_URL}/network/interfaces", token=token)
+ assert response.status_code == 200
+ assert len(response.get_json()) == 3
+ check_input_is_subset_of_response(
+ response, (interface1.to_dict(), interface2.to_dict(), interface3.to_dict())
+ )
+ response = get(client, f"{API_URL}/network/interfaces?name=host3", token=token)
+ assert response.status_code == 200
+ assert len(response.get_json()) == 1
+
+
def test_create_interface_fails(
client, host, network_scope_factory, network_factory, no_login_check_token
):
@@ -1422,39 +1479,39 @@ def test_create_ansible_group_with_vars(client, admin_token):
assert group.vars == data["vars"]
-def test_get_hosts(client, host_factory, readonly_token):
+def test_get_hosts(client, host_factory, admin_token):
# Create some hosts
host1 = host_factory()
host2 = host_factory()
- response = get(client, f"{API_URL}/network/hosts", token=readonly_token)
+ response = get(client, f"{API_URL}/network/hosts", token=admin_token)
assert response.status_code == 200
assert len(response.get_json()) == 2
assert HOST_KEYS == set(response.get_json()[0].keys())
check_input_is_subset_of_response(response, (host1.to_dict(), host2.to_dict()))
-def test_get_hosts_with_ansible_vars(client, host_factory, readonly_token):
+def test_get_hosts_with_ansible_vars(client, host_factory, admin_token):
vars = {"foo": "hello", "mylist": [1, 2, 3]}
host_factory(ansible_vars=vars)
- response = get(client, f"{API_URL}/network/hosts", token=readonly_token)
+ response = get(client, f"{API_URL}/network/hosts", token=admin_token)
assert response.status_code == 200
assert response.get_json()[0]["ansible_vars"] == vars
def test_get_hosts_with_model(
- client, model_factory, item_factory, host_factory, readonly_token
+ client, model_factory, item_factory, host_factory, admin_token
):
host1 = host_factory()
model1 = model_factory(name="EX3400")
item_factory(model=model1, host_id=host1.id)
- response = get(client, f"{API_URL}/network/hosts", token=readonly_token)
+ response = get(client, f"{API_URL}/network/hosts", token=admin_token)
assert response.status_code == 200
assert response.get_json()[0]["model"] == "EX3400"
-def test_get_hosts_with_no_model(client, host_factory, readonly_token):
+def test_get_hosts_with_no_model(client, host_factory, admin_token):
host_factory()
- response = get(client, f"{API_URL}/network/hosts", token=readonly_token)
+ response = get(client, f"{API_URL}/network/hosts", token=admin_token)
assert response.status_code == 200
assert response.get_json()[0]["model"] is None
@@ -1493,7 +1550,7 @@ def test_get_hosts_recursive_interfaces(
assert rinterface21["network"] == interface21.network.vlan_name
-def test_get_hosts_recursive_items(client, item_factory, host_factory, readonly_token):
+def test_get_hosts_recursive_items(client, item_factory, host_factory, admin_token):
host1 = host_factory()
item11 = item_factory(ics_id="AAA001", host_id=host1.id, stack_member=1)
item12 = item_factory(ics_id="AAA002", host_id=host1.id, stack_member=0)
@@ -1501,7 +1558,7 @@ def test_get_hosts_recursive_items(client, item_factory, host_factory, readonly_
item21 = item_factory(ics_id="AAB001", host_id=host2.id)
item22 = item_factory(ics_id="AAB002", host_id=host2.id)
# Without recursive, we only get the ics_id of the items
- response = get(client, f"{API_URL}/network/hosts", token=readonly_token)
+ response = get(client, f"{API_URL}/network/hosts", token=admin_token)
assert response.status_code == 200
assert len(response.get_json()) == 2
rhost1, rhost2 = response.get_json()
@@ -1510,9 +1567,7 @@ def test_get_hosts_recursive_items(client, item_factory, host_factory, readonly_
# or by ics_id when stack_member is None
assert rhost2["items"] == ["AAB001", "AAB002"]
# With recursive, items are expanded
- response = get(
- client, f"{API_URL}/network/hosts?recursive=true", token=readonly_token
- )
+ response = get(client, f"{API_URL}/network/hosts?recursive=true", token=admin_token)
assert response.status_code == 200
assert len(response.get_json()) == 2
rhost1, rhost2 = response.get_json()
@@ -1544,6 +1599,58 @@ def test_get_hosts_recursive_items(client, item_factory, host_factory, readonly_
}
+def test_get_hosts_sensitive(
+ client, network_scope_factory, network_factory, interface_factory, host_factory
+):
+ # Create some hosts
+ scope = network_scope_factory(supernet="192.168.0.0/16")
+ network1 = network_factory(
+ address="192.168.1.0/24",
+ first_ip="192.168.1.10",
+ last_ip="192.168.1.250",
+ scope=scope,
+ )
+ network2 = network_factory(
+ address="192.168.2.0/24",
+ first_ip="192.168.2.10",
+ last_ip="192.168.2.250",
+ scope=scope,
+ sensitive=True,
+ )
+ host1 = host_factory(name="host1")
+ host2 = host_factory(name="host2")
+ host3 = host_factory(name="host3")
+ # Add a host without interface
+ host4 = host_factory(name="host4")
+ interface_factory(name=host1.name, network=network1, ip="192.168.1.10", host=host1)
+ interface_factory(name=host2.name, network=network1, ip="192.168.1.11", host=host2)
+ interface_factory(name=host3.name, network=network2, ip="192.168.2.10", host=host3)
+ # Normal users can't get hosts on sensitive networks
+ token = get_token(client, "user_ro", "userro")
+ response = get(client, f"{API_URL}/network/hosts", token=token)
+ assert response.status_code == 200
+ # Note that hosts without interfaces can't be retrieved either!
+ assert len(response.get_json()) == 2
+ check_input_is_subset_of_response(response, (host1.to_dict(), host2.to_dict()))
+ response = get(client, f"{API_URL}/network/hosts?name=host1", token=token)
+ assert response.status_code == 200
+ assert len(response.get_json()) == 1
+ response = get(client, f"{API_URL}/network/hosts?name=host3", token=token)
+ assert response.status_code == 200
+ assert len(response.get_json()) == 0
+ # Admin users can get all interfaces
+ token = get_token(client, "admin", "adminpasswd")
+ response = get(client, f"{API_URL}/network/hosts", token=token)
+ assert response.status_code == 200
+ assert len(response.get_json()) == 4
+ check_input_is_subset_of_response(
+ response, (host1.to_dict(), host2.to_dict(), host3.to_dict(), host4.to_dict())
+ )
+ response = get(client, f"{API_URL}/network/hosts?name=host3", token=token)
+ assert response.status_code == 200
+ assert len(response.get_json()) == 1
+
+
def test_create_host(client, device_type_factory, user_token):
device_type = device_type_factory(name="Virtual")
# check that name and device_type are mandatory
@@ -1831,8 +1938,58 @@ def test_search_hosts(client, host_factory, readonly_token):
assert r[0]["description"] == host1.description
+@pytest.mark.parametrize(
+ "username,password,query,expected",
+ [
+ ("admin", "adminpasswd", "", 3),
+ ("admin", "adminpasswd", "?q=beautiful", 2),
+ ("admin", "adminpasswd", "?q=description:beautiful", 1),
+ ("user_ro", "userro", "", 1),
+ ("user_ro", "userro", "?q=beautiful", 1),
+ ("user_ro", "userro", "?q=description:beautiful", 0),
+ ("user_ro", "userro", "?q=description:explicit", 1),
+ ],
+)
+def test_search_hosts_sensitive(
+ username,
+ password,
+ query,
+ expected,
+ client,
+ network_scope_factory,
+ network_factory,
+ interface_factory,
+ host_factory,
+):
+ scope = network_scope_factory(name="ProdNetworks", supernet="192.168.0.0/16")
+ network1 = network_factory(
+ address="192.168.1.0/24",
+ first_ip="192.168.1.10",
+ last_ip="192.168.1.250",
+ sensitive=False,
+ scope=scope,
+ )
+ network2 = network_factory(
+ address="192.168.2.0/24",
+ first_ip="192.168.2.10",
+ last_ip="192.168.2.250",
+ sensitive=True,
+ scope=scope,
+ )
+ host1 = host_factory(name="test-beautiful", description="search explicit")
+ interface_factory(name=host1.name, host=host1, network=network1, ip="192.168.1.10")
+ host2 = host_factory(name="test-explicit")
+ interface_factory(name=host2.name, host=host2, network=network2, ip="192.168.2.10")
+ host3 = host_factory(name="another-host", description="search beautiful")
+ interface_factory(name=host3.name, host=host3, network=network2, ip="192.168.2.11")
+ token = get_token(client, username, password)
+ response = get(client, f"{API_URL}/network/hosts/search{query}", token=token)
+ assert response.status_code == 200
+ assert len(response.get_json()) == expected
+
+
@pytest.mark.parametrize("endpoint", ["network/hosts", "network/hosts/search"])
-def test_pagination(endpoint, client, host_factory, readonly_token):
+def test_pagination(endpoint, client, host_factory, admin_token):
# MAX_PER_PAGE set to 25 for testing
# Create 30 hosts
for i in range(30):
@@ -1842,7 +1999,7 @@ def test_pagination(endpoint, client, host_factory, readonly_token):
else:
extra_args = ""
# By default 20 hosts per page shall be returned
- response = get(client, f"{API_URL}/{endpoint}", token=readonly_token)
+ response = get(client, f"{API_URL}/{endpoint}", token=admin_token)
assert response.status_code == 200
assert len(response.get_json()) == 20
assert response.headers["x-total-count"] == "30"
@@ -1856,7 +2013,7 @@ def test_pagination(endpoint, client, host_factory, readonly_token):
response = get(
client,
f"{API_URL}/{endpoint}?per_page=20&page=2{extra_args}",
- token=readonly_token,
+ token=admin_token,
)
assert response.status_code == 200
assert len(response.get_json()) == 10
@@ -1871,7 +2028,7 @@ def test_pagination(endpoint, client, host_factory, readonly_token):
assert 'rel="next"' not in response.headers["link"]
assert 'rel="last"' not in response.headers["link"]
# Request 10 elements per_page
- response = get(client, f"{API_URL}/{endpoint}?per_page=10", token=readonly_token)
+ response = get(client, f"{API_URL}/{endpoint}?per_page=10", token=admin_token)
assert response.status_code == 200
assert len(response.get_json()) == 10
assert response.headers["x-total-count"] == "30"
@@ -1884,7 +2041,7 @@ def test_pagination(endpoint, client, host_factory, readonly_token):
in response.headers["link"]
)
# You can't request more than MAX_PER_PAGE elements
- response = get(client, f"{API_URL}/{endpoint}?per_page=50", token=readonly_token)
+ response = get(client, f"{API_URL}/{endpoint}?per_page=50", token=admin_token)
assert response.status_code == 200
assert len(response.get_json()) == 25
assert response.headers["x-total-count"] == "30"
diff --git a/tests/functional/test_web.py b/tests/functional/test_web.py
index 93d38f6519c643828a492443ebace989f55a11a1..92410ea71ae5aeb3f509423021a9d6648cfb9241 100644
--- a/tests/functional/test_web.py
+++ b/tests/functional/test_web.py
@@ -263,6 +263,48 @@ def test_retrieve_hosts_by_ip(logged_client, interface_factory):
assert r["data"][0]["name"] == interface1.host.name
+def test_retrieve_sensitive_hosts(
+ client, network_scope_factory, network_factory, host_factory, interface_factory
+):
+ scope = network_scope_factory(name="ProdNetworks", supernet="192.168.0.0/16")
+ network1 = network_factory(
+ address="192.168.1.0/24",
+ first_ip="192.168.1.10",
+ last_ip="192.168.1.250",
+ sensitive=False,
+ scope=scope,
+ )
+ network2 = network_factory(
+ address="192.168.2.0/24",
+ first_ip="192.168.2.10",
+ last_ip="192.168.2.250",
+ sensitive=True,
+ scope=scope,
+ )
+ host1 = host_factory()
+ interface_factory(name=host1.name, host=host1, network=network1, ip="192.168.1.10")
+ host2 = host_factory()
+ interface_factory(name=host2.name, host=host2, network=network2, ip="192.168.2.10")
+ host3 = host_factory()
+ interface_factory(name=host3.name, host=host3, network=network2, ip="192.168.2.11")
+ # Normal users can't see hosts on sensitive networks
+ login(client, "user_lab", "userlab")
+ response = client.post("/network/_retrieve_hosts")
+ r = response.get_json()
+ assert r["recordsTotal"] == 3
+ assert r["recordsFiltered"] == 1
+ assert len(r["data"]) == 1
+ assert r["data"][0]["name"] == host1.name
+ logout(client)
+ # admin users can see all hosts
+ login(client, "admin", "adminpasswd")
+ response = client.post("/network/_retrieve_hosts",)
+ r = response.get_json()
+ assert r["recordsTotal"] == 3
+ assert r["recordsFiltered"] == 3
+ assert len(r["data"]) == 3
+
+
def test_delete_interface_from_index(
no_login_check_client, interface_factory, host_factory
):