diff --git a/src/gcp_scanner/credsdb.py b/src/gcp_scanner/credsdb.py index 3bbdc577..c77e3c53 100644 --- a/src/gcp_scanner/credsdb.py +++ b/src/gcp_scanner/credsdb.py @@ -85,7 +85,7 @@ def get_creds_from_metadata() -> Tuple[Optional[str], Optional[Credentials]]: google.auth.service_account.Credentials: The constructed credentials. """ - print("Retrieving access token from instance metadata") + logging.info("Retrieving access token from instance metadata") token_url = "http://metadata.google.internal/computeMetadata/v1/instance/\ service-accounts/default/token" @@ -121,7 +121,7 @@ def get_creds_from_metadata() -> Tuple[Optional[str], Optional[Credentials]]: logging.error(sys.exc_info()[1]) return None, None - print("Successfully retrieved instance metadata") + logging.info("Successfully retrieved instance metadata") logging.info("Access token length: %d", len(token)) logging.info("Instance email: %s", email) logging.info("Instance scopes: %s", instance_scopes) @@ -192,9 +192,9 @@ def find_creds(explicit_path: Optional[str] = None) -> List[str]: logging.info("Scanning %s for credentials.db", dir_path) full_path = os.path.join(dir_path, "credentials.db") if os.path.exists(full_path) and os.access(full_path, os.R_OK): - print(f"Identified accessible gcloud config profile {full_path}") + logging.info("Identified accessible gcloud config profile %s", full_path) list_of_creds_files.append(full_path) - print(f"Identified {len(list_of_creds_files)} credential DBs") + logging.info("Identified %d credential DBs", len(list_of_creds_files)) return list_of_creds_files @@ -263,7 +263,7 @@ def extract_creds(path_to_creds_db: str) -> List[Tuple[str, str, str]]: logging.info("Found valid access token for %s", row[0]) access_token = access_tokens[row[0]] res.append(SA(row[0], row[1], access_token)) - print(f"Identified {len(res)} credential entries") + logging.info("Identified %d credential entries", len(res)) return res @@ -422,4 +422,3 @@ def get_scopes_from_refresh_token(context) -> Union[List[str], None]: return None - diff --git a/src/gcp_scanner/scanner.py b/src/gcp_scanner/scanner.py index e0b76daf..50846a66 100644 --- a/src/gcp_scanner/scanner.py +++ b/src/gcp_scanner/scanner.py @@ -207,7 +207,7 @@ def get_resources(project: models.ProjectInfo): return project_id = project.project['projectId'] - print(f'Inspecting project {project_id}') + logging.info('Inspecting project %s', project_id) project_result = dict() project_result['project_info'] = project.project diff --git a/src/gcp_scanner/test_unit.py b/src/gcp_scanner/test_unit.py index eb707a67..cf36ce7c 100644 --- a/src/gcp_scanner/test_unit.py +++ b/src/gcp_scanner/test_unit.py @@ -272,6 +272,55 @@ def test_get_sa_details_from_key_files_with_invalid_and_valid_key_file( self.assertEqual(actual, expect) +class TestLoggingOutput(unittest.TestCase): + """Test scanner progress output uses logging.""" + + def test_get_resources_logs_project_progress(self): + project_id = 'test-project' + sa_results = scanner.infinite_defaultdict() + sa_results['service_account_chain'] = [] + sa_results['current_service_account'] = 'service-account@example.com' + sa_results['token_scopes'] = [] + sa_results[project_id]['service_account_edges'] = [] + + with tempfile.TemporaryDirectory() as output_dir: + project = scanner.models.ProjectInfo( + {'projectId': project_id}, + sa_results, + output_dir, + {}, + False, + None, + 'unit-test', + 'service-account@example.com', + Mock(), + [], + 1, + ) + + with patch('builtins.print') as mocked_print: + with patch('gcp_scanner.scanner.logging.info') as mocked_log: + scanner.get_resources(project) + + mocked_print.assert_not_called() + mocked_log.assert_any_call('Inspecting project %s', project_id) + + def test_find_creds_logs_discovered_profiles(self): + with tempfile.TemporaryDirectory() as profile_path: + creds_path = os.path.join(profile_path, 'credentials.db') + open(creds_path, 'w', encoding='utf-8').close() + + with patch('builtins.print') as mocked_print: + with patch('gcp_scanner.credsdb.logging.info') as mocked_log: + actual = credsdb.find_creds(profile_path) + + self.assertEqual(actual, [creds_path]) + mocked_print.assert_not_called() + mocked_log.assert_any_call( + 'Identified accessible gcloud config profile %s', creds_path) + mocked_log.assert_any_call('Identified %d credential DBs', 1) + + class TestScopes(unittest.TestCase): """Test fetching scopes from a refresh token."""