From 479867fca742e89f086a77f707e402a99f508ff5 Mon Sep 17 00:00:00 2001 From: Josema Camacho Date: Thu, 23 Oct 2025 16:50:48 +0200 Subject: [PATCH] feat(attack-paths): prowler to cartography data model - WIP - IAM WIP --- api/poetry.lock | 6 +- api/pyproject.toml | 2 +- .../backend/tasks/jobs/cartography/aws/ecs.py | 86 +-- .../backend/tasks/jobs/cartography/aws/iam.py | 600 ++++++++++++++---- .../providers/aws/services/iam/iam_service.py | 2 + 5 files changed, 543 insertions(+), 153 deletions(-) diff --git a/api/poetry.lock b/api/poetry.lock index c9bd9c6be3..8239b2db00 100644 --- a/api/poetry.lock +++ b/api/poetry.lock @@ -4146,8 +4146,8 @@ tzlocal = "5.3.1" [package.source] type = "git" url = "https://github.com/prowler-cloud/prowler.git" -reference = "master" -resolved_reference = "a52697bfdfee83d14a49c11dcbe96888b5cd767e" +reference = "PROWLER-253-extract-aws-data-from-prowler-database-and-transform-it-to-be-ingested-by-cartography" +resolved_reference = "66008e8e9ee39265c273b409b7ff6be214255d9a" [[package]] name = "psutil" @@ -6259,4 +6259,4 @@ type = ["pytest-mypy"] [metadata] lock-version = "2.1" python-versions = ">=3.11,<3.13" -content-hash = "03442fd4673006c5a74374f90f53621fd1c9d117279fe6cc0355ef833eb7f9bb" +content-hash = "533a5e0e733df594dd591b8af7910562e8d4cd032ed1850aa454c5eb298f5748" diff --git a/api/pyproject.toml b/api/pyproject.toml index d7ef074e36..2c7c2691e8 100644 --- a/api/pyproject.toml +++ b/api/pyproject.toml @@ -24,7 +24,7 @@ dependencies = [ "drf-spectacular-jsonapi==0.5.1", "gunicorn==23.0.0", "lxml==5.3.2", - "prowler @ git+https://github.com/prowler-cloud/prowler.git@master", + "prowler @ git+https://github.com/prowler-cloud/prowler.git@PROWLER-253-extract-aws-data-from-prowler-database-and-transform-it-to-be-ingested-by-cartography", "psycopg2-binary==2.9.9", "pytest-celery[redis] (>=1.0.1,<2.0.0)", "sentry-sdk[django] (>=2.20.0,<3.0.0)", diff --git a/api/src/backend/tasks/jobs/cartography/aws/ecs.py b/api/src/backend/tasks/jobs/cartography/aws/ecs.py index 10bd2d4018..e5d5092c20 100644 --- a/api/src/backend/tasks/jobs/cartography/aws/ecs.py +++ b/api/src/backend/tasks/jobs/cartography/aws/ecs.py @@ -1,10 +1,8 @@ -from asyncio import tasks import json from collections import defaultdict from typing import Any -from httpx import get import neo4j from cartography.intel.aws import ecs as cartography_ecs @@ -91,8 +89,7 @@ def _get_ecs_task_definitions_region_metadata( regions: list[str], ) -> dict[str, list[dict[str, Any]]]: """ - Getting ECS tasks metadata from Prowler DB. - # TODO: We can't filter the tasks by cluster ARN using Prowler data + Getting ECS task definitions metadata from Prowler DB. """ with rls_transaction(tenant_id): @@ -110,9 +107,7 @@ def _get_ecs_task_definitions_region_metadata( for task_definition in task_definitions_qs: task_metadata = json.loads(task_definition.metadata) task_metadata["inserted_at"] = task_definition.inserted_at - task_definitions_region_metadata[ - task_metadata.get("region") - ].append(task_metadata) + task_definitions_region_metadata[task_metadata.get("region")].append(task_metadata) return task_definitions_region_metadata @@ -203,8 +198,24 @@ def _sync_ecs_clusters( update_tag: int, ) -> None: """ - Code based on `cartography.intel.aws.ecs._sync_ecs_cluster_arns` and - `cartography.intel.aws.ecs.get_ecs_clusters`. + Code based on `cartography.intel.aws.ecs._sync_ecs_cluster_arns`. + """ + + clusters = _get_ecs_clusters(clusters_metadata) + cartography_ecs.load_ecs_clusters( + neo4j_session, + clusters, + region, + account_id, + update_tag, + ) + + return clusters + + +def _get_ecs_clusters(clusters_metadata: list[dict[str, Any]]) -> list[dict[str, Any]]: + """ + Code based on `cartography.intel.aws.ecs.get_ecs_clusters`. # TODO: There are missing fields to implement """ @@ -229,14 +240,6 @@ def _sync_ecs_clusters( # "defaultCapacityProviderStrategy" # TODO }) - cartography_ecs.load_ecs_clusters( - neo4j_session, - clusters, - region, - account_id, - update_tag, - ) - return clusters @@ -250,7 +253,7 @@ def _sync_ecs_container_instances( """ Code based on `cartography.intel.aws.ecs._sync_ecs_container_instances` and `cartography.intel.aws.ecs.get_ecs_container_instances`. - # TODO: Container instances data is missing from Prowler DB + # TODO: AWS ECS Container instances data is missing from Prowler DB """ cluster_arn = cluster_metadata.get("arn") @@ -284,10 +287,10 @@ def _sync_ecs_task_and_container_defns( """ cluster_arn = cluster_metadata.get("arn") - tasks = [] # TODO: Prowler doesn't save ECS tasks data - containers = [] # TODO: Prowler doesn't save ECS tasks' containers data + tasks = [] # TODO: Prowler doesn't save AWS ECS tasks data + containers = [] # TODO: Prowler doesn't save AWS ECS tasks' containers data - task_definitions = get_ecs_task_definitions(task_definitions_metadata) + task_definitions = _get_ecs_task_definitions(task_definitions_metadata) container_defs = cartography_ecs._get_container_defs_from_task_definitions(task_definitions) cartography_ecs.load_ecs_tasks( @@ -326,7 +329,7 @@ def _sync_ecs_task_and_container_defns( return tasks, containers, task_definitions, container_defs -def get_ecs_task_definitions(task_definitions_metadata: list[dict[str, Any]]) -> list[dict[str, Any]]: +def _get_ecs_task_definitions(task_definitions_metadata: list[dict[str, Any]]) -> list[dict[str, Any]]: """ Getting ECS task definitions from Prowler DB. """ @@ -365,17 +368,35 @@ def _sync_ecs_services( update_tag: int, ) -> None: """ - Code based on `cartography.intel.aws.ecs._sync_ecs_services` and - `cartography.intel.aws.ecs.get_ecs_services`. - # TODO: A lot of fields are missing + Code based on `cartography.intel.aws.ecs._sync_ecs_services`. """ cluster_arn = cluster_metadata.get("arn") - services = [ + services = _get_ecs_services(cluster_metadata) + + cartography_ecs.load_ecs_services( + neo4j_session, + cluster_arn, + services, + region, + account_id, + update_tag, + ) + + return services + + +def _get_ecs_services(cluster_metadata: dict[str, Any]) -> list[dict[str, Any]]: + """ + Code based on `cartography.intel.aws.ecs.get_ecs_services`. + # TODO: A lot of fields are missing + """ + + return [ { "serviceArn": service.get("arn"), "serviceName": service.get("name"), - "clusterArn": cluster_arn, + "clusterArn": cluster_metadata.get("arn"), "loadBalancers": service.get("load_balancers"), "serviceRegistries": service.get("service_registries"), "status": service.get("status"), @@ -404,14 +425,3 @@ def _sync_ecs_services( } for service in cluster_metadata.get("services").values() ] - - cartography_ecs.load_ecs_services( - neo4j_session, - cluster_arn, - services, - region, - account_id, - update_tag, - ) - - return services diff --git a/api/src/backend/tasks/jobs/cartography/aws/iam.py b/api/src/backend/tasks/jobs/cartography/aws/iam.py index 6501c05d86..bdd62088a9 100644 --- a/api/src/backend/tasks/jobs/cartography/aws/iam.py +++ b/api/src/backend/tasks/jobs/cartography/aws/iam.py @@ -1,135 +1,513 @@ -from __future__ import annotations - import json -from datetime import datetime, timezone -from typing import Any, Dict, List, Optional -from cartography.intel.aws import iam as carto_iam +from collections import defaultdict +from typing import Any + +import neo4j + +from cartography.intel.aws import iam as cartography_iam from celery.utils.log import get_task_logger -from neo4j import GraphDatabase +from openai import containers from api.db_utils import rls_transaction -from api.models import Provider, Resource, ResourceScanSummary +from api.models import Resource, ResourceScanSummary -logger = get_task_logger(__name__) +# TODO: Do the rigth logging setup +# logger = get_task_logger(__name__) +import logging +from config.custom_logging import BackendLogger +logger = logging.getLogger(BackendLogger.API) def sync_aws_iam( tenant_id: str, provider_id: str, - scan_id: Optional[str], - neo4j_conf: Dict[str, Any], -) -> Dict[str, Any]: - try: - from neo4j import GraphDatabase as _ # ensure import present - except Exception as e: - logger.error(f"Neo4j not available: {e}") - return {"error": str(e)} + account_id: str, + scan_id: str, + regions: list[str], + neo4j_session: neo4j.Session, + update_tag: int, + common_job_parameters: dict[str, Any], +) -> dict[str, Any]: + """ + Entry point for syncing AWS IAM data into Cartography. + """ + + # Calling our version of cartography AWS IAM sync + return _sync( + tenant_id, + provider_id, + account_id, + scan_id, + regions, + neo4j_session, + update_tag, + common_job_parameters, + ) + + +def _sync( + tenant_id: str, + provider_id: str, + account_id: str, + scan_id: str, + regions: list[str], + neo4j_session: neo4j.Session, + update_tag: int, + common_job_parameters: dict[str, Any], +) -> dict[str, Any]: + """ + Code based on `cartography.intel.aws.iam.sync`. + """ + + cartography_iam.sync_root_principal( + neo4j_session, + account_id, + update_tag, + ) + + _sync_users( + tenant_id, + provider_id, + account_id, + scan_id, + regions, + neo4j_session, + update_tag, + ) + + _sync_groups( + tenant_id, + provider_id, + account_id, + scan_id, + regions, + neo4j_session, + update_tag, + ) + + _sync_roles( + tenant_id, + provider_id, + account_id, + scan_id, + regions, + neo4j_session, + update_tag, + ) + + cartography_iam.sync_assumerole_relationships( + neo4j_session, + account_id, + update_tag, + common_job_parameters, + ) + + _sync_user_access_keys( + tenant_id, + provider_id, + account_id, + scan_id, + regions, + neo4j_session, + update_tag, + common_job_parameters, + ) + + cartography_iam.cleanup_iam(neo4j_session, common_job_parameters) + + cartography_iam.merge_module_sync_metadata( + neo4j_session, + group_type="AWSAccount", + group_id=account_id, + synced_type="AWSPrincipal", + update_tag=update_tag, + stat_handler=cartography_iam.stat_handler, + ) + + return {} + +def _sync_users( + tenant_id: str, + provider_id: str, + account_id: str, + scan_id: str, + regions: list[str], + neo4j_session: neo4j.Session, + update_tag: int, +) -> None: + + user_data = _get_user_list_data(tenant_id, provider_id, scan_id, regions) + transformed_user_data = cartography_iam.transform_users(user_data["Users"]) + + cartography_iam.load_users(neo4j_session, transformed_user_data, account_id, update_tag) + _sync_inline_policies(user_data["Users"], neo4j_session, update_tag, account_id) + _sync_managed_policies(user_data["Users"], neo4j_session, update_tag, account_id) + + +def _get_user_list_data( + tenant_id: str, + provider_id: str, + scan_id: str, + regions: list[str], +) -> dict[str, list[dict[str, Any]]]: + """ + Code based on `cartography.intel.aws.iam.get_user_list_data`. + # TODO: There are missing fields to implement + """ + + users = [] + users_metadata = _get_users_metadata(tenant_id, provider_id, scan_id, regions) + + for user_metadata in users_metadata: + user = { + "Arn": user_metadata.get("arn"), + "UserId": None, # TODO + "UserName": user_metadata.get("name"), + "Path": None, # TODO + "CreateDate": user_metadata.get("inserted_at"), + "PasswordLastUsed": user_metadata.get("password_last_used"), + "InlinePolicies": user_metadata.get("inline_policies", []), + "AttachedPolicies": user_metadata.get("attached_policies", []), + "AccessKeyMetadata": user_metadata.get("access_keys_metadata", []), + } + users.append(user) + + return {"Users": users} + + +def _get_users_metadata( + tenant_id: str, + provider_id: str, + scan_id: str, + regions: list[str], +) -> list[dict[str, Any]]: + """ + Getting IAM users data from Prowler DB. + """ with rls_transaction(tenant_id): - provider = Provider.objects.get(pk=provider_id) - account_id = provider.uid + users_qs = Resource.objects.filter( + provider_id=provider_id, + id__in=ResourceScanSummary.objects.filter( + scan_id=scan_id, + service="iam", + resource_type="AwsIamUser", + ).values_list("resource_id", flat=True), + region__in=regions, + ).only("metadata", "inserted_at") - base_qs = Resource.objects.filter( - provider_id=provider_id, service="iam", type__in=["role", "iam_role"] - ) - if scan_id: - rss_ids = ResourceScanSummary.objects.filter( - tenant_id=tenant_id, scan_id=scan_id, service="iam" - ).values_list("resource_id", flat=True) - base_qs = base_qs.filter(id__in=list(rss_ids)) - role_resources = list( - base_qs.only("uid", "name", "metadata", "details", "inserted_at") - ) + users_metadata = [] + for user in users_qs: + user_metadata = json.loads(user.metadata) + user_metadata["inserted_at"] = user.inserted_at + users_metadata.append(user_metadata) - roles: List[Dict[str, Any]] = [] - for r in role_resources: - role_obj: Dict[str, Any] = { - "Arn": r.uid, - "RoleName": r.name or r.uid.split("/")[-1], - "RoleId": r.uid.split("/")[-1], - "Path": "/", - "CreateDate": (r.inserted_at or datetime.now(tz=timezone.utc)).isoformat(), - "AssumeRolePolicyDocument": {"Statement": []}, + return users_metadata + + +def _sync_inline_policies( + resource_data: list[dict[str, Any]], + neo4j_session: neo4j.Session, + update_tag: int, + account_id: str, +) -> None: + """ + Code based on `cartography.intel.aws.iam.sync_[user|group|role|]_inline_policies`. + """ + + inline_policy_data = _get_inline_resource_policy_data(resource_data) + transformed_inline_policy_data = cartography_iam.transform_policy_data( + inline_policy_data, + cartography_iam.PolicyType.inline.value, + ) + + cartography_iam.load_policy_data( + neo4j_session, + transformed_inline_policy_data, + update_tag, + account_id, + ) + + +def _get_inline_resource_policy_data(resource_data: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: + """ + Code based on `cartography.intel.aws.iam.get_[user|group|role]_policy_data`. + # TODO: It looks like Prowler does not store AWS IAM [User|Group|Role] policies document statement + """ + + inline_policies = {} + + for resource in resource_data: + inline_policies[resource.get("Arn")] = { + policy_name: None # TODO: The policy document statement is missing + for policy_name in resource.get("InlinePolicies", []) } - for raw in (getattr(r, "metadata", None), getattr(r, "details", None)): - if not raw: - continue - try: - data = json.loads(raw) if isinstance(raw, str) else raw - except Exception: - continue - if not isinstance(data, dict): - continue - pol = ( - data.get("AssumeRolePolicyDocument") - or data.get("AssumeRolePolicy") - or data.get("assume_role_policy_document") - ) - if pol and isinstance(pol, dict) and pol.get("Statement"): - role_obj["AssumeRolePolicyDocument"] = pol - break - roles.append(role_obj) - uri = neo4j_conf.get("uri") - user = neo4j_conf.get("user") or neo4j_conf.get("username") - password = neo4j_conf.get("password") - database = neo4j_conf.get("database") - if not all([uri, user, password]): - logger.error("Neo4j configuration incomplete: require uri, user, password") - return {"error": "missing_neo4j_config"} + return inline_policies - update_tag = int(datetime.now(tz=timezone.utc).timestamp() * 1000) - common_job_parameters = {"UPDATE_TAG": update_tag, "AWS_ID": account_id} - driver = GraphDatabase.driver(uri, auth=(user, password)) - # Provide minimal patches for IAM getters - def _patched_get_iam_roles(_boto3_session): - return roles +def _sync_managed_policies( + resource_data: list[dict[str, Any]], + neo4j_session: neo4j.Session, + update_tag: int, + account_id: str, +) -> None: + """ + Code based on `cartography.intel.aws.iam.sync_[user|group|role|]_managed_policies`. + """ - def _empty_list(*args, **kwargs): - return [] + managed_policy_data = _get_resource_managed_policy_data(resource_data) + transformed_policy_data = cartography_iam.transform_policy_data( + managed_policy_data, + cartography_iam.PolicyType.managed.value, + ) - setattr(carto_iam, "get_iam_roles", _patched_get_iam_roles) - # Some versions may use list_roles instead - if hasattr(carto_iam, "list_roles"): - setattr(carto_iam, "list_roles", _patched_get_iam_roles) - for fname in [ - "get_iam_users", - "get_iam_groups", - "get_iam_policies", - "get_iam_role_inline_policies", - "get_iam_role_attached_policies", - "get_iam_instance_profiles", - ]: - if hasattr(carto_iam, fname): - setattr(carto_iam, fname, _empty_list) + cartography_iam.load_policy_data( + neo4j_session, + transformed_policy_data, + update_tag, + account_id, + ) - try: - with driver.session(database=database) if database else driver.session() as neo4j_session: - class _Boto3SessionStub: - pass - boto3_session = _Boto3SessionStub() - try: - carto_iam.sync( - neo4j_session, - boto3_session, - account_id, - update_tag, - common_job_parameters, - ) - except TypeError: - carto_iam.sync( - neo4j_session, - boto3_session, - account_id, - update_tag, - ) - finally: - try: - driver.close() - except Exception: - pass +def _get_resource_managed_policy_data(resource_data: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: + """ + Code based on `cartography.intel.aws.iam.get_[user|group|role]_managed_policy_data`. + # TODO: It looks like Prowler does not store AWS IAM [User|Group|Role] attached policy default + # version document statement + """ - return {"roles": len(roles)} + attached_policies = {} + + for resource in resource_data: + attached_policies[resource.get("Arn")] = { + policy.get("PolicyArn"): None # TODO: The policy default version document statement is missing + for policy in resource.get("AttachedPolicies", []) + } + + return attached_policies + + +def _sync_groups( + tenant_id: str, + provider_id: str, + account_id: str, + scan_id: str, + regions: list[str], + neo4j_session: neo4j.Session, + update_tag: int, +): + """ + Code based on `cartography.intel.aws.iam.sync_groups`. + """ + + group_data = _get_group_list_data(tenant_id, provider_id, scan_id, regions) + group_memberships = _get_group_memberships(group_data["Groups"]) + transformed_group_data = cartography_iam.transform_groups(group_data["Groups"], group_memberships) + + cartography_iam.load_groups(neo4j_session, transformed_group_data, account_id, update_tag) + + _sync_inline_policies(group_data["Groups"], neo4j_session, update_tag, account_id) + + _sync_managed_policies(group_data["Groups"], neo4j_session, update_tag, account_id) + + +def _get_group_list_data( + tenant_id: str, + provider_id: str, + scan_id: str, + regions: list[str], +) -> dict[str, list[dict[str, Any]]]: + """ + Code based on `cartography.intel.aws.iam.get_group_list_data`. + # TODO: There are missing fields to implement + """ + + groups = [] + groups_metadata = _get_groups_metadata(tenant_id, provider_id, scan_id, regions) + + for group_metadata in groups_metadata: + group = { + "Arn": group_metadata.get("arn"), + "GroupId": None, # TODO + "GroupName": group_metadata.get("name"), + "Path": None, # TODO + "CreateDate": group_metadata.get("inserted_at"), + "Users": group_metadata.get("users", []), + "InlinePolicies": group_metadata.get("inline_policies", []), + "AttachedPolicies": group_metadata.get("attached_policies", []), + } + groups.append(group) + + return {"Groups": groups} + + +def _get_groups_metadata( + tenant_id: str, + provider_id: str, + scan_id: str, + regions: list[str], +) -> list[dict[str, Any]]: + """ + Getting IAM groups data from Prowler DB. + """ + + with rls_transaction(tenant_id): + groups_qs = Resource.objects.filter( + provider_id=provider_id, + id__in=ResourceScanSummary.objects.filter( + scan_id=scan_id, + service="iam", + resource_type="AwsIamGroup", + ).values_list("resource_id", flat=True), + region__in=regions, + ).only("metadata", "inserted_at") + + groups_metadata = [] + for group in groups_qs: + group_metadata = json.loads(group.metadata) + group_metadata["inserted_at"] = group.inserted_at + groups_metadata.append(group_metadata) + + return groups_metadata + + +def _get_group_memberships(group_data: list[dict[str, Any]]) -> dict[str, list[str]]: + """ + Code based on `cartography.intel.aws.iam.get_group_memberships`. + """ + + group_memberships = {} + + for group in group_data: + group_memberships[group.get("Arn")] = [ + user.get("arn") for user in group.get("Users", []) + ] + + return group_memberships + + +def _sync_roles( + tenant_id: str, + provider_id: str, + account_id: str, + scan_id: str, + regions: list[str], + neo4j_session: neo4j.Session, + update_tag: int, +): + """ + Code based on `cartography.intel.aws.iam.sync_roles`. + """ + + roles_data = _get_role_list_data(tenant_id, provider_id, scan_id, regions) + + cartography_iam.sync_role_assumptions(neo4j_session, roles_data, account_id, update_tag) + + _sync_inline_policies(roles_data["Roles"], neo4j_session, update_tag, account_id) + + _sync_managed_policies(roles_data["Roles"], neo4j_session, update_tag, account_id) + + +def _get_role_list_data( + tenant_id: str, + provider_id: str, + scan_id: str, + regions: list[str], +) -> dict[str, list[dict[str, Any]]]: + """ + Code based on `cartography.intel.aws.iam.get_role_list_data`. + # TODO: There are missing fields to implement + """ + + roles = [] + roles_metadata = _get_roles_metadata(tenant_id, provider_id, scan_id, regions) + + for role_metadata in roles_metadata: + role = { + "Arn": role_metadata.get("arn"), + "RoleId": None, # TODO + "RoleName": role_metadata.get("name"), + "Path": None, # TODO + "CreateDate": role_metadata.get("inserted_at"), + "AssumeRolePolicyDocument": role_metadata.get("assume_role_policy", {}), + "Tags": role_metadata.get("tags", []), + "InlinePolicies": role_metadata.get("inline_policies", []), + "AttachedPolicies": role_metadata.get("attached_policies", []), + } + roles.append(role) + + return {"Roles": roles} + + +def _get_roles_metadata( + tenant_id: str, + provider_id: str, + scan_id: str, + regions: list[str], +) -> list[dict[str, Any]]: + """ + Getting IAM roles data from Prowler DB. + """ + + with rls_transaction(tenant_id): + roles_qs = Resource.objects.filter( + provider_id=provider_id, + id__in=ResourceScanSummary.objects.filter( + scan_id=scan_id, + service="iam", + resource_type="AwsIamRole", + ).values_list("resource_id", flat=True), + region__in=regions, + ).only("metadata", "inserted_at") + + roles_metadata = [] + for role in roles_qs: + role_metadata = json.loads(role.metadata) + role_metadata["inserted_at"] = role.inserted_at + roles_metadata.append(role_metadata) + + return roles_metadata + + +def _sync_user_access_keys( + tenant_id: str, + provider_id: str, + account_id: str, + scan_id: str, + regions: list[str], + neo4j_session: neo4j.Session, + update_tag: int, + common_job_parameters: dict[str, Any], +): + """ + Code based on `cartography.intel.aws.iam.sync_user_access_keys`. + """ + + user_data = _get_user_list_data(tenant_id, provider_id, scan_id, regions) + user_access_keys = _pretransform_access_keys(user_data["Users"]) + access_key_data = cartography_iam.transform_access_keys(user_access_keys) + cartography_iam.load_access_keys( + neo4j_session, access_key_data, update_tag, account_id + ) + cartography_iam.GraphJob.from_node_schema( + cartography_iam.AccountAccessKeySchema(), + common_job_parameters, + ).run( + neo4j_session, + ) + + +def _pretransform_access_keys(users: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]: + """ + Code based on `cartography.intel.aws.iam.get_user_access_keys_data`. + # TODO: Some AWS IAM Access Key `last_used_info` data is missing from Prowler DB + """ + + user_access_keys = {} + + for user in users: + user_access_keys[user.get("Arn")] = user.get("AccessKeyMetadata", []) + + return user_access_keys diff --git a/prowler/providers/aws/services/iam/iam_service.py b/prowler/providers/aws/services/iam/iam_service.py index 1ff3dbd07a..c6209a7593 100644 --- a/prowler/providers/aws/services/iam/iam_service.py +++ b/prowler/providers/aws/services/iam/iam_service.py @@ -926,6 +926,7 @@ class IAM(AWSService): self.access_keys_metadata[(user.name, user.arn)] = response[ "AccessKeyMetadata" ] + user.access_keys_metadata = response["AccessKeyMetadata"] except ClientError as error: if error.response["Error"]["Code"] == "NoSuchEntity": logger.warning( @@ -1068,6 +1069,7 @@ class User(BaseModel): console_access: Optional[bool] attached_policies: list[dict] = [] inline_policies: list[str] = [] + access_keys_metadata: Optional[list[dict]] = [] tags: Optional[list]