Log scanner progress messages

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
 
 
-

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

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."""