From 578769d892fafff209225b194d3e24a345ed2615 Mon Sep 17 00:00:00 2001 From: Haihan Jiang Date: Fri, 29 May 2026 18:41:32 -0700 Subject: [PATCH] Add service account impersonation tests --- src/gcp_scanner/scanner.py | 5 +- src/gcp_scanner/test_unit.py | 154 +++++++++++++++++++++++++++++++++++ 2 files changed, 157 insertions(+), 2 deletions(-) diff --git a/src/gcp_scanner/scanner.py b/src/gcp_scanner/scanner.py index e0b76daf..1397e66f 100644 --- a/src/gcp_scanner/scanner.py +++ b/src/gcp_scanner/scanner.py @@ -315,7 +315,7 @@ def impersonate_service_accounts( # Enumerate projects accessible by SA project_id = project['projectId'] - project_result = sa_results['projects'][project_id] + project_result = sa_results[project_id] project_result['project_info'] = project # Iterate over discovered service accounts by attempting impersonation project_result['service_account_edges'] = [] @@ -330,7 +330,8 @@ def impersonate_service_accounts( if impers is not None and impers.get('impersonate', False) is True: logging.info('Looking for impersonation options in %s', project_id) iam_client = iam_client_for_credentials(credentials) - if is_set(scan_config, 'iam_policy') is False: + iam_policy = project_result.get('iam_policy') + if iam_policy is None: iam_policy = CrawlerFactory.create_crawler('iam_policy').crawl( project_id, ClientFactory.get_client('cloudresourcemanager').get_service( diff --git a/src/gcp_scanner/test_unit.py b/src/gcp_scanner/test_unit.py index eb707a67..02b10530 100644 --- a/src/gcp_scanner/test_unit.py +++ b/src/gcp_scanner/test_unit.py @@ -22,6 +22,7 @@ import json import logging import os +import queue import shutil import sqlite3 import tempfile @@ -272,6 +273,159 @@ def test_get_sa_details_from_key_files_with_invalid_and_valid_key_file( self.assertEqual(actual, expect) +class TestServiceAccountImpersonation(unittest.TestCase): + """Tests for service account impersonation.""" + + def test_get_sas_for_impersonation(self): + """Test extracting service accounts from IAM policy bindings.""" + iam_policy = [ + { + "role": "roles/iam.serviceAccountTokenCreator", + "members": [ + "user:owner@example.com", + "serviceAccount:first@test-project.iam.gserviceaccount.com", + "group:admins@example.com", + ], + }, + { + "role": "roles/viewer", + "members": [ + "serviceAccount:second@test-project.iam.gserviceaccount.com", + "serviceAccount:first@test-project.iam.gserviceaccount.com", + ], + }, + { + "role": "roles/editor", + }, + ] + + actual = scanner.get_sas_for_impersonation(iam_policy) + + self.assertEqual( + actual, + [ + "first@test-project.iam.gserviceaccount.com", + "second@test-project.iam.gserviceaccount.com", + ], + ) + + @patch("gcp_scanner.credsdb.credentials_from_token") + def test_impersonate_sa_requests_access_token(self, mocked_creds_from_token): + """Test impersonation asks IAM Credentials for a cloud-platform token.""" + scopes = ["https://www.googleapis.com/auth/cloud-platform"] + target_account = "target@test-project.iam.gserviceaccount.com" + iam_client = Mock() + iam_client.generate_access_token.return_value = Mock( + access_token="generated-token" + ) + mocked_creds_from_token.return_value = "impersonated-credentials" + + actual = credsdb.impersonate_sa(iam_client, target_account) + + self.assertEqual(actual, "impersonated-credentials") + iam_client.generate_access_token.assert_called_once_with( + name=target_account, + scope=scopes, + retry=None, + ) + mocked_creds_from_token.assert_called_once_with( + "generated-token", + None, + None, + None, + None, + scopes, + ) + + @patch("gcp_scanner.scanner.ClientFactory.get_client") + @patch("gcp_scanner.scanner.CrawlerFactory.create_crawler") + @patch("gcp_scanner.scanner.credsdb.impersonate_sa") + @patch("gcp_scanner.scanner.iam_client_for_credentials") + def test_impersonate_service_accounts_queues_discovered_accounts( + self, + mocked_iam_client_for_credentials, + mocked_impersonate_sa, + mocked_create_crawler, + mocked_get_client, + ): + """Test scanner queues impersonated service accounts.""" + project_id = "test-project" + source_credentials = Mock() + source_sa = "source@test-project.iam.gserviceaccount.com" + target_accounts = [ + "first@test-project.iam.gserviceaccount.com", + "second@test-project.iam.gserviceaccount.com", + ] + iam_policy = [{ + "members": [ + f"serviceAccount:{target_accounts[0]}", + f"serviceAccount:{target_accounts[1]}", + ], + }] + context = Mock() + context.service_account_queue = queue.Queue() + sa_results = scanner.infinite_defaultdict() + crawler = Mock() + crawler.crawl.return_value = iam_policy + client = Mock() + client.get_service.return_value = "cloudresourcemanager-service" + mocked_create_crawler.return_value = crawler + mocked_get_client.return_value = client + mocked_iam_client_for_credentials.return_value = "iam-client" + mocked_impersonate_sa.side_effect = [ + "first-credentials", + "second-credentials", + ] + + scanner.impersonate_service_accounts( + context, + {"projectId": project_id}, + { + "service_accounts": { + "impersonate": True, + }, + "iam_policy": { + "fetch": True, + }, + }, + sa_results, + ["root@test-project.iam.gserviceaccount.com"], + source_sa, + source_credentials, + ) + + self.assertEqual( + sa_results[project_id]["service_account_edges"], + target_accounts, + ) + self.assertEqual( + list(context.service_account_queue.queue), + [ + ( + target_accounts[0], + "first-credentials", + ["root@test-project.iam.gserviceaccount.com", source_sa], + ), + ( + target_accounts[1], + "second-credentials", + ["root@test-project.iam.gserviceaccount.com", source_sa], + ), + ], + ) + mocked_iam_client_for_credentials.assert_called_once_with( + source_credentials + ) + mocked_create_crawler.assert_called_once_with("iam_policy") + mocked_get_client.assert_called_once_with("cloudresourcemanager") + client.get_service.assert_called_once_with(source_credentials) + crawler.crawl.assert_called_once_with( + project_id, + "cloudresourcemanager-service", + ) + self.assertEqual(mocked_impersonate_sa.call_count, 2) + + class TestScopes(unittest.TestCase): """Test fetching scopes from a refresh token."""