fix(RHINENG-12444) Fix HTTP 409 for get_host_exists with granular RBAC (

#1931)

* Use find_all_hosts in get_host_id_by_insights_id; add test

* Remove redundant org_id filter; add function to __all__

api/host_query_db.py

@@ -35,6 +35,7 @@
     "get_all_hosts",
     "get_host_list",
     "get_host_list_by_id_list",
+    "get_host_id_by_insights_id",
     "get_host_tags_list_by_id_list",
     "params_to_order_by",
 )
@@ -164,14 +165,8 @@ def get_host_list_by_id_list(
 
 def get_host_id_by_insights_id(insights_id: str, rbac_filter=None) -> str:
     identity = get_current_identity()
-    all_filters = (
-        [Host.org_id == identity.org_id]
-        + canonical_fact_filter("insights_id", insights_id)
-        + rbac_permissions_filter(rbac_filter)
-    )
-
-    query = db.session.query(Host).filter(*all_filters)
-    query = update_query_for_owner_id(identity, query)
+    all_filters = canonical_fact_filter("insights_id", insights_id) + rbac_permissions_filter(rbac_filter)
+    query = _find_all_hosts(columns=[Host.id], identity=identity).filter(*all_filters)
 
     try:
         found_id = query.with_entities(Host.id).order_by(Host.modified_on.desc()).scalar()

tests/test_api_hosts_get.py

@@ -1854,3 +1854,47 @@ def test_get_host_exists_error_multiple_found(db_create_host, api_get):
     response_status, _ = api_get(url)
 
     assert response_status == 409
+
+
+def test_get_host_exists_granular_rbac(
+    db_create_host, db_create_group, db_create_host_group_assoc, api_get, mocker, enable_rbac
+):
+    # Create 3 hosts with unique insights IDs that the user has access to
+    accessible_group_id = db_create_group("accessible_group").id
+    accessible_insights_id_list = [generate_uuid() for _ in range(3)]
+    accessible_host_id_list = [
+        db_create_host(extra_data={"canonical_facts": {"insights_id": insights_id}}).id
+        for insights_id in accessible_insights_id_list
+    ]
+    for host_id in accessible_host_id_list:
+        db_create_host_group_assoc(host_id, accessible_group_id)
+
+    # Create 2 hosts with unique insights IDs that the user does not have access to
+    inaccessible_group_id = db_create_group("inaccessible_group").id
+    inaccessible_insights_id_list = [generate_uuid() for _ in range(2)]
+    inaccessible_host_id_list = [
+        db_create_host(extra_data={"canonical_facts": {"insights_id": insights_id}}).id
+        for insights_id in inaccessible_insights_id_list
+    ]
+    for host_id in inaccessible_host_id_list:
+        db_create_host_group_assoc(host_id, inaccessible_group_id)
+
+    # Grant access to first group
+    get_rbac_permissions_mock = mocker.patch("lib.middleware.get_rbac_permissions")
+    mock_rbac_response = create_mock_rbac_response(
+        "tests/helpers/rbac-mock-data/inv-hosts-read-resource-defs-template.json"
+    )
+    mock_rbac_response[0]["resourceDefinitions"][0]["attributeFilter"]["value"] = [str(accessible_group_id)]
+    get_rbac_permissions_mock.return_value = mock_rbac_response
+
+    # Verify that the user can see each of the hosts in the "accessible" group
+    for insights_id in accessible_insights_id_list:
+        url = build_host_exists_url(insights_id)
+        response_status, _ = api_get(url)
+        assert_response_status(response_status, 200)
+
+    # Verify that the user can NOT see the hosts in the "inaccessible" group
+    for insights_id in inaccessible_insights_id_list:
+        url = build_host_exists_url(insights_id)
+        response_status, _ = api_get(url)
+        assert_response_status(response_status, 404)