chore(RHINENG-14059): Add Ruff ARG rule (WIP)

api/assignment_rule.py

@@ -72,7 +72,7 @@ def create_assignment_rule(body, rbac_filter=None):
         return json_error_response("Validation Error", str(e.messages), HTTPStatus.BAD_REQUEST)
 
     try:
-        created_assignment_rule = add_assignment_rule(validated_create_assignment_rule)
+        created_assignment_rule = add_assignment_rule(validated_create_assignment_rule, rbac_filter)
         created_assignment_rule = serialize_assignment_rule(created_assignment_rule)
     except IntegrityError as error:
         group_id = validated_create_assignment_rule.get("group_id")

api/host_query_db.py

@@ -594,7 +594,6 @@ def get_host_ids_list(
 def get_hosts_to_export(
     identity: object,
     filters: dict | None = None,
-    export_format: str = "json",
     rbac_filter: dict | None = None,
     batch_size: int = 0,
 ) -> Iterator[dict]:

api/staleness.py

@@ -49,6 +49,7 @@ def _validate_input_data(body):
 @rbac(RbacResourceType.HOSTS, RbacPermission.READ)
 @metrics.api_request_time.time()
 def get_staleness(rbac_filter=None):
+    _ = (rbac_filter,)  # Unused: group-level RBAC does not apply to account-level staleness
     try:
         staleness = get_staleness_obj()
         staleness = serialize_staleness_response(staleness)
@@ -63,6 +64,7 @@ def get_staleness(rbac_filter=None):
 @rbac(RbacResourceType.HOSTS, RbacPermission.READ)
 @metrics.api_request_time.time()
 def get_default_staleness(rbac_filter=None):
+    _ = (rbac_filter,)  # Unused: group-level RBAC does not apply to account-level staleness
     try:
         identity = get_current_identity()
         staleness = get_sys_default_staleness_api(identity)

api/system_profile.py

@@ -109,6 +109,7 @@ def get_operating_system(
 @rbac(RbacResourceType.HOSTS, RbacPermission.READ)
 @metrics.schema_validation_time.time()
 def validate_schema(repo_fork="RedHatInsights", repo_branch="master", days=1, max_messages=10000, rbac_filter=None):
+    _ = (rbac_filter,)  # Unused. No RBAC needed because we check for specific users.
     # Use the identity header to make sure the user is someone from our team.
     config = Config(RuntimeEnvironment.SERVICE)
     identity = get_current_identity()

app/__init__.py

@@ -127,7 +127,7 @@ def flush_segmentio():
         analytics.flush()
 
 
-def initialize_segmentio(config):
+def initialize_segmentio():
     logger.info("Initializing Segmentio")
     analytics.write_key = os.getenv("SEGMENTIO_WRITE_KEY", None)
     logger.info("Registering Segmentio flush on shutdown")
@@ -356,6 +356,6 @@ def after_request_org_check(response):
     # initialize metrics to zero
     initialize_metrics(app_config)
 
-    initialize_segmentio(app_config)
+    initialize_segmentio()
 
     return app

app/auth/__init__.py

@@ -12,6 +12,7 @@
 
 
 def authentication_header_handler(apikey, required_scopes=None):
+    _ = (required_scopes,)  # Unused, but needed for the security scheme (apikeyInfoFunc)
     try:
         identity = from_auth_header(apikey)
     except Exception as exc:

app/queue/export_service.py

@@ -53,13 +53,10 @@ def build_headers(
     return rbac_request_headers, request_headers
 
 
-def get_host_list(
-    identity: Identity, exportFormat: str, rbac_filter: dict | None, inventory_config: Config
-) -> list[dict]:
+def get_host_list(identity: Identity, rbac_filter: dict | None, inventory_config: Config) -> list[dict]:
     host_data = list(
         get_hosts_to_export(
             identity,
-            export_format=exportFormat,
             rbac_filter=rbac_filter,
             batch_size=inventory_config.export_svc_batch_size,
         )
@@ -117,7 +114,7 @@ def create_export(
 
     try:
         # create a generator with serialized host data
-        host_data = get_host_list(identity, exportFormat, rbac_filter, inventory_config)
+        host_data = get_host_list(identity, rbac_filter, inventory_config)
 
         request_url = _build_export_request_url(
             export_service_endpoint, exportUUID, applicationName, resourceUUID, "upload"

lib/assignment_rule_repository.py

@@ -1,4 +1,5 @@
 from app.auth import get_current_identity
+from app.exceptions import InventoryException
 from app.logging import get_logger
 from app.models import AssignmentRule
 from app.models import db
@@ -7,8 +8,18 @@
 logger = get_logger(__name__)
 
 
-def add_assignment_rule(assign_rule_data) -> AssignmentRule:
+def add_assignment_rule(assign_rule_data, rbac_filter) -> AssignmentRule:
     logger.debug(f"Creating assignment rule: {assign_rule_data}")
+
+    # Do not allow the user to create an assignment rule for a group they don't have access to
+    if (
+        rbac_filter is not None
+        and "groups" in rbac_filter
+        and len(rbac_groups := rbac_filter["groups"]) > 0
+        and assign_rule_data.get("group_id") not in rbac_groups
+    ):
+        raise InventoryException(title="Forbidden", detail="User does not have permission for specified group.")
+
     org_id = get_current_identity().org_id
     account = get_current_identity().account_number
     assign_rule_name = assign_rule_data.get("name")

pyproject.toml

@@ -17,6 +17,8 @@ lint.select = [
     "SIM",
     # isort
     "I",
+    # flake8-unused-arguments
+    # "ARG",
 ]
 
 lint.ignore = [
@@ -33,6 +35,8 @@ lint.extend-safe-fixes = [
 
 lint.isort.force-single-line = true
 
+# lint.flake8-unused-arguments.ignore-variadic-names = true
+
 [tool.mypy]
 python_version = "3.9"
 

tests/fixtures/api_fixtures.py

@@ -172,7 +172,7 @@ def enable_org_id_translation(inventory_config):
 
 
 @pytest.fixture(scope="function")
-def enable_unleash(inventory_config):
+def _enable_unleash(inventory_config):
     inventory_config.unleash_token = "mockUnleashTokenValue"
 
 

tests/test_api_assignment_rules_create.py

@@ -103,7 +103,8 @@ def test_create_assignemnt_rule_same_group(api_create_assign_rule, db_create_gro
     assert group in response_data["detail"]
 
 
-def test_create_assignment_rule_RBAC_denied(subtests, mocker, api_create_assign_rule, db_create_group, enable_rbac):
+@pytest.mark.usefixtures("enable_rbac")
+def test_create_assignment_rule_RBAC_denied(subtests, mocker, api_create_assign_rule, db_create_group):
     get_rbac_permissions_mock = mocker.patch("lib.middleware.get_rbac_permissions")
 
     group = db_create_group("my_group")

tests/test_api_assignment_rules_get.py

@@ -189,7 +189,8 @@ def test_assignment_rule_id_list_bad_id(num_rules, db_create_assignment_rule, db
     assert len(response_data["results"]) == 0
 
 
-def test_get_assignment_rule_id_list_RBAC_denied(subtests, mocker, api_get, enable_rbac):
+@pytest.mark.usefixtures("enable_rbac")
+def test_get_assignment_rule_id_list_RBAC_denied(subtests, mocker, api_get):
     get_rbac_permissions_mock = mocker.patch("lib.middleware.get_rbac_permissions")
 
     # Assignment rules get, requires group read.
@@ -205,9 +206,8 @@ def test_get_assignment_rule_id_list_RBAC_denied(subtests, mocker, api_get, enab
             assert_response_status(response_status, 403)
 
 
-def test_get_assignment_rules_RBAC_denied_specific_groups(
-    mocker, db_create_group, db_create_assignment_rule, api_get, enable_rbac
-):
+@pytest.mark.usefixtures("enable_rbac")
+def test_get_assignment_rules_RBAC_denied_specific_groups(mocker, db_create_group, db_create_assignment_rule, api_get):
     get_rbac_permissions_mock = mocker.patch("lib.middleware.get_rbac_permissions")
 
     group_id_list = [str(db_create_group(f"test_group{g_index}").id) for g_index in range(2)]

tests/test_api_facts.py

@@ -20,15 +20,14 @@
 from tests.helpers.test_utils import get_staleness_timestamps
 
 
-def test_replace_facts_to_multiple_hosts_with_branch_id(
-    db_create_multiple_hosts, db_get_hosts, api_put, event_producer_mock
-):
+@pytest.mark.usefixtures("event_producer_mock")
+def test_replace_facts_to_multiple_hosts_with_branch_id(db_create_multiple_hosts, db_get_hosts, api_put):
     created_hosts = db_create_multiple_hosts(how_many=2, extra_data={"facts": DB_FACTS})
 
     host_id_list = get_id_list_from_hosts(created_hosts)
     facts_url = build_facts_url(host_list_or_id=created_hosts, namespace=DB_FACTS_NAMESPACE, query="?branch_id=1234")
 
-    response_status, response_data = api_put(facts_url, DB_NEW_FACTS)
+    response_status, _ = api_put(facts_url, DB_NEW_FACTS)
     assert_response_status(response_status, expected_status=200)
 
     expected_facts = get_expected_facts_after_update("replace", DB_FACTS_NAMESPACE, DB_FACTS, DB_NEW_FACTS)
@@ -37,20 +36,19 @@ def test_replace_facts_to_multiple_hosts_with_branch_id(
         assert host.facts == expected_facts
 
 
-def test_replace_facts_to_multiple_hosts_including_nonexistent_host(db_create_multiple_hosts, db_get_hosts, api_put):
+def test_replace_facts_to_multiple_hosts_including_nonexistent_host(db_create_multiple_hosts, api_put):
     created_hosts = db_create_multiple_hosts(how_many=2, extra_data={"facts": DB_FACTS})
 
     url_host_id_list = f"{build_id_list_for_url(created_hosts)},{generate_uuid()},{generate_uuid()}"
     facts_url = build_facts_url(host_list_or_id=url_host_id_list, namespace=DB_FACTS_NAMESPACE)
 
-    response_status, response_data = api_put(facts_url, DB_NEW_FACTS)
+    response_status, _ = api_put(facts_url, DB_NEW_FACTS)
 
     assert_response_status(response_status, expected_status=400)
 
 
-def test_replace_facts_to_multiple_hosts_with_empty_key_value_pair(
-    db_create_multiple_hosts, db_get_hosts, api_put, event_producer_mock
-):
+@pytest.mark.usefixtures("event_producer_mock")
+def test_replace_facts_to_multiple_hosts_with_empty_key_value_pair(db_create_multiple_hosts, db_get_hosts, api_put):
     new_facts = {}
 
     created_hosts = db_create_multiple_hosts(how_many=2, extra_data={"facts": DB_FACTS})
@@ -59,7 +57,7 @@ def test_replace_facts_to_multiple_hosts_with_empty_key_value_pair(
     facts_url = build_facts_url(host_list_or_id=created_hosts, namespace=DB_FACTS_NAMESPACE)
 
     # Set the value in the namespace to an empty fact set
-    response_status, response_data = api_put(facts_url, new_facts)
+    response_status, _ = api_put(facts_url, new_facts)
     assert_response_status(response_status, expected_status=200)
 
     expected_facts = get_expected_facts_after_update("replace", DB_FACTS_NAMESPACE, DB_FACTS, new_facts)
@@ -74,7 +72,7 @@ def test_replace_facts_to_namespace_that_does_not_exist(db_create_multiple_hosts
 
     facts_url = build_facts_url(host_list_or_id=created_hosts, namespace="imanonexistentnamespace")
 
-    response_status, response_data = api_patch(facts_url, new_facts)
+    response_status, _ = api_patch(facts_url, new_facts)
     assert_response_status(response_status, expected_status=400)
 
 
@@ -85,36 +83,38 @@ def test_replace_facts_without_fact_dict(api_put):
     assert_error_response(response_data, expected_status=400, expected_detail="Request body must not be empty")
 
 
-def test_replace_facts_on_multiple_hosts(db_create_multiple_hosts, db_get_hosts, api_put, event_producer_mock):
+@pytest.mark.usefixtures("event_producer_mock")
+def test_replace_facts_on_multiple_hosts(db_create_multiple_hosts, db_get_hosts, api_put):
     created_hosts = db_create_multiple_hosts(how_many=2, extra_data={"facts": DB_FACTS})
 
     host_id_list = get_id_list_from_hosts(created_hosts)
     facts_url = build_facts_url(host_list_or_id=created_hosts, namespace=DB_FACTS_NAMESPACE)
 
-    response_status, response_data = api_put(facts_url, DB_NEW_FACTS)
+    response_status, _ = api_put(facts_url, DB_NEW_FACTS)
     assert_response_status(response_status, expected_status=200)
 
     expected_facts = get_expected_facts_after_update("replace", DB_FACTS_NAMESPACE, DB_FACTS, DB_NEW_FACTS)
 
     assert all(host.facts == expected_facts for host in db_get_hosts(host_id_list))
 
 
-def test_replace_empty_facts_on_multiple_hosts(db_create_multiple_hosts, db_get_hosts, api_put, event_producer_mock):
+@pytest.mark.usefixtures("event_producer_mock")
+def test_replace_empty_facts_on_multiple_hosts(db_create_multiple_hosts, db_get_hosts, api_put):
     new_facts = {}
 
     created_hosts = db_create_multiple_hosts(how_many=2, extra_data={"facts": DB_FACTS})
 
     host_id_list = get_id_list_from_hosts(created_hosts)
     facts_url = build_facts_url(host_list_or_id=created_hosts, namespace=DB_FACTS_NAMESPACE)
 
-    response_status, response_data = api_put(facts_url, new_facts)
+    response_status, _ = api_put(facts_url, new_facts)
     assert_response_status(response_status, expected_status=200)
 
     expected_facts = get_expected_facts_after_update("replace", DB_FACTS_NAMESPACE, DB_FACTS, new_facts)
 
     assert all(host.facts == expected_facts for host in db_get_hosts(host_id_list))
 
-    response_status, response_data = api_put(facts_url, DB_NEW_FACTS)
+    response_status, _ = api_put(facts_url, DB_NEW_FACTS)
     assert_response_status(response_status, expected_status=200)
 
     expected_facts = get_expected_facts_after_update("replace", DB_FACTS_NAMESPACE, DB_FACTS, DB_NEW_FACTS)
@@ -141,7 +141,8 @@ def test_replace_facts_on_multiple_culled_hosts(db_create_multiple_hosts, api_pu
         assert_response_status(response_status, expected_status=400)
 
 
-def test_put_facts_with_RBAC_allowed(subtests, mocker, api_put, db_create_host, enable_rbac, event_producer_mock):
+@pytest.mark.usefixtures("enable_rbac", "event_producer_mock")
+def test_put_facts_with_RBAC_allowed(subtests, mocker, api_put, db_create_host):
     get_rbac_permissions_mock = mocker.patch("lib.middleware.get_rbac_permissions")
 
     for response_file in HOST_WRITE_ALLOWED_RBAC_RESPONSE_FILES:
@@ -157,8 +158,9 @@ def test_put_facts_with_RBAC_allowed(subtests, mocker, api_put, db_create_host,
             assert_response_status(response_status, 200)
 
 
+@pytest.mark.usefixtures("enable_rbac", "event_producer_mock")
 def test_put_facts_with_RBAC_allowed_specific_groups(
-    mocker, api_put, db_create_host, enable_rbac, event_producer_mock, db_create_group, db_create_host_group_assoc
+    mocker, api_put, db_create_host, db_create_group, db_create_host_group_assoc
 ):
     get_rbac_permissions_mock = mocker.patch("lib.middleware.get_rbac_permissions")
 
@@ -182,9 +184,8 @@ def test_put_facts_with_RBAC_allowed_specific_groups(
     assert_response_status(response_status, 200)
 
 
-def test_put_facts_with_RBAC_denied(
-    subtests, mocker, api_put, db_create_host, db_get_host, enable_rbac, event_producer_mock
-):
+@pytest.mark.usefixtures("enable_rbac", "event_producer_mock")
+def test_put_facts_with_RBAC_denied(subtests, mocker, api_put, db_create_host, db_get_host):
     get_rbac_permissions_mock = mocker.patch("lib.middleware.get_rbac_permissions")
 
     updated_facts = {"updatedfact1": "updatedvalue1", "updatedfact2": "updatedvalue2"}
@@ -204,13 +205,12 @@ def test_put_facts_with_RBAC_denied(
             assert db_get_host(host.id).facts[DB_FACTS_NAMESPACE] != updated_facts
 
 
+@pytest.mark.usefixtures("enable_rbac", "event_producer_mock")
 def test_put_facts_with_RBAC_denied_specific_groups(
     mocker,
     api_put,
     db_create_host,
     db_get_host,
-    enable_rbac,
-    event_producer_mock,
     db_create_group,
     db_create_host_group_assoc,
 ):
@@ -241,14 +241,15 @@ def test_put_facts_with_RBAC_denied_specific_groups(
     assert db_get_host(host.id).facts[DB_FACTS_NAMESPACE] != updated_facts
 
 
-def test_put_facts_with_RBAC_bypassed_as_system(api_put, db_create_host, enable_rbac, event_producer_mock):
+@pytest.mark.usefixtures("enable_rbac", "event_producer_mock")
+def test_put_facts_with_RBAC_bypassed_as_system(api_put, db_create_host):
     host = db_create_host(
         SYSTEM_IDENTITY,
         extra_data={"facts": DB_FACTS, "system_profile_facts": {"owner_id": SYSTEM_IDENTITY["system"].get("cn")}},
     )
 
     url = build_facts_url(host_list_or_id=host.id, namespace=DB_FACTS_NAMESPACE)
 
-    response_status, response_data = api_put(url, DB_NEW_FACTS, SYSTEM_IDENTITY)
+    response_status, _ = api_put(url, DB_NEW_FACTS, SYSTEM_IDENTITY)
 
     assert_response_status(response_status, 200)

tests/test_api_groups_create.py

@@ -177,7 +177,8 @@ def test_create_group_with_host_from_another_org(db_create_host, api_create_grou
     assert event_producer.write_event.call_count == 0
 
 
-def test_create_group_RBAC_denied(subtests, mocker, api_create_group, db_get_group_by_name, enable_rbac):
+@pytest.mark.usefixtures("enable_rbac")
+def test_create_group_RBAC_denied(subtests, mocker, api_create_group, db_get_group_by_name):
     get_rbac_permissions_mock = mocker.patch("lib.middleware.get_rbac_permissions")
     group_data = {"name": "my_awesome_group", "host_ids": []}
 
@@ -194,7 +195,8 @@ def test_create_group_RBAC_denied(subtests, mocker, api_create_group, db_get_gro
             assert not db_get_group_by_name("my_awesome_group")
 
 
-def test_create_group_RBAC_denied_attribute_filter(mocker, api_create_group, enable_rbac):
+@pytest.mark.usefixtures("enable_rbac")
+def test_create_group_RBAC_denied_attribute_filter(mocker, api_create_group):
     get_rbac_permissions_mock = mocker.patch("lib.middleware.get_rbac_permissions")
 
     # Mock the RBAC response with a groups-write permission that has an attributeFilter