Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ permissions:

env:
JUST_VERSION: '1'
PREK_VERSION: '0.2.12'
PREK_VERSION: '0.2.13'
TERRAFORM_VERSION: '1.13'

jobs:
Expand Down
6 changes: 6 additions & 0 deletions policy_sentry/command/query.py
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,12 @@ def query_action_table(
else:
# Otherwise, leverage the datastore inside the python package
logger.debug("Leveraging the bundled IAM Definition.")

if service == "catalog":
# make sure nothing is returned when querying for `catalog`,
# which has some special behaviour related to `servicecatalog`
service = ""

# Actions on all services
if service == "all":
all_services = get_all_service_prefixes()
Expand Down
7 changes: 6 additions & 1 deletion policy_sentry/querying/actions.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,9 +59,12 @@ def get_action_data(service: str, action_name: str) -> dict[str, list[dict[str,
Returns:
List: A dictionary containing metadata about an IAM Action.
"""
action_data_results = {}
action_data_results: dict[str, list[dict[str, Any]]] = {}
try:
service_prefix_data = get_service_prefix_data(service)
if not service_prefix_data:
return action_data_results

if action_name.endswith("*"):
stripped_action_name = action_name.removesuffix("*")
results = []
Expand Down Expand Up @@ -239,6 +242,8 @@ def get_actions_with_arn_type_and_access_level(
results.extend(actions)
else:
service_prefix_data = get_service_prefix_data(service_prefix)
# mainly needed for the use case of `catalog` -> `servicecatalog` mapping
service_prefix = service_prefix_data["prefix"]
for action_name, action_data in service_prefix_data["privileges"].items():
if (
action_data["access_level"] == access_level
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ Twitter = "https://twitter.com/kmcquade3"
dev = [
"coverage>=7.11.0,<8.0.0",
"mypy>=1.18.0,<2.0.0",
"prek>=0.2.10,<0.3.0",
"prek>=0.2.13,<0.3.0",
"pytest>=8.4.0,<9.0.0",
"rust-just>=1.43.0,<2.0.0",
"types-beautifulsoup4>=4.12.0,<5.0.0",
Expand Down
18 changes: 17 additions & 1 deletion test/querying/test_all.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ def test_GH_296_query_all_actions_with_wildcard_resources(self):
access_level=None,
condition=None,
)
self.assertTrue(len(result) > 3000)
self.assertGreater(len(result), 3000)

def test_get_service_authorization_url(self):
result = get_service_authorization_url("a4b")
Expand All @@ -59,3 +59,19 @@ def test_get_service_authorization_url(self):
"https://docs.aws.amazon.com/service-authorization/latest/reference/list_alexaforbusiness.html"
)
self.assertTrue(result == expected_result)

def test_query_catalog_service(self):
# given
service = "catalog"

# when
result = query_action_table(
service=service,
resource_type=None,
name=None,
access_level=None,
condition=None,
)

# then
self.assertFalse(result)
7 changes: 7 additions & 0 deletions test/querying/test_query_actions.py
Original file line number Diff line number Diff line change
Expand Up @@ -346,6 +346,13 @@ def test_get_actions_with_arn_type_and_access_level_case_5(self):
output = get_actions_with_arn_type_and_access_level("s3", "object", "List")
self.assertTrue("s3:ListMultipartUploadParts" in output)

def test_get_actions_with_arn_type_and_access_level_servicecatalog(self):
# querying with `catalog` as a service prefix should yield `servicecatalog` as an action prefix
# https://github.com/salesforce/policy_sentry/issues/563

output = get_actions_with_arn_type_and_access_level("catalog", "Portfolio", "List")
self.assertIn("servicecatalog:ListPortfolioAccess", output)

def test_get_actions_matching_arn_type_case_1(self):
"""querying.actions.get_actions_matching_arn_type"""
expected_results = [
Expand Down
Loading