Merge pull request #50 from cybrota/feat/add-logging

feat: add access logging to secret fetches

Author: narenaryan

.github/workflows/release.yml

@@ -17,10 +17,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@f43a0e5ff2bd294095638e18286ca9a3d1956744 # v3
 
       - name: Set up Python
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@7f4fc3e22c37d6ff65e88745f38bd3157c663f7c # v4
         with:
           python-version: '3.10'
 

.github/workflows/security.yml

@@ -14,10 +14,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@f43a0e5ff2bd294095638e18286ca9a3d1956744 # v3
 
       - name: Set up Python
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@7f4fc3e22c37d6ff65e88745f38bd3157c663f7c # v4
         with:
           python-version: "3.10"
 

.github/workflows/test.yml

@@ -14,10 +14,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@f43a0e5ff2bd294095638e18286ca9a3d1956744 # v3
 
       - name: Set up Python
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@7f4fc3e22c37d6ff65e88745f38bd3157c663f7c # v4
         with:
           python-version: "3.10"
 

src/whispr/__about__.py

@@ -1 +1 @@
-version = "0.7.0"
+version = "0.8.0"

src/whispr/aws.py

@@ -4,8 +4,8 @@
 import botocore.exceptions
 import structlog
 
+from whispr.logging import log_secret_fetch
 from whispr.vault import SimpleVault
-from whispr.enums import AWSVaultSubType
 
 
 class AWSVault(SimpleVault):
@@ -29,8 +29,11 @@ def fetch_secrets(self, secret_name: str) -> str:
         """
         try:
             response = self.client.get_secret_value(SecretId=secret_name)
+            secret_value = response.get("SecretString") or ""
             self.logger.debug(f"Successfully fetched aws secret: {secret_name}")
-            return response.get("SecretString")
+            if secret_value:
+                log_secret_fetch(self.logger, secret_name, "aws")
+            return secret_value
         except botocore.exceptions.ClientError as error:
             if error.response["Error"]["Code"] == "ResourceNotFoundException":
                 self.logger.error(
@@ -74,9 +77,11 @@ def fetch_secrets(self, secret_name: str) -> str:
             response = self.client.get_parameter(Name=secret_name)
             param = response.get("Parameter")
             if param:
-                return param.get("Value")
-            else:
-                return ""
+                secret_value = param.get("Value") or ""
+                if secret_value:
+                    log_secret_fetch(self.logger, secret_name, "aws-ssm")
+                return secret_value
+            return ""
         except botocore.exceptions.ClientError as error:
             if error.response["Error"]["Code"] == "ParameterNotFound":
                 self.logger.error(

src/whispr/azure.py

@@ -3,6 +3,7 @@
 import structlog
 from azure.keyvault.secrets import SecretClient
 import azure.core.exceptions
+from whispr.logging import log_secret_fetch
 from whispr.vault import SimpleVault
 
 
@@ -31,6 +32,8 @@ def fetch_secrets(self, secret_name: str) -> str:
         try:
             secret = self.client.get_secret(secret_name)
             self.logger.info(f"Successfully fetched secret: {secret_name}")
+            if secret.value:
+                log_secret_fetch(self.logger, secret_name, "azure")
             return secret.value
         except azure.core.exceptions.ResourceNotFoundError:
             self.logger.error(

src/whispr/gcp.py

@@ -3,6 +3,7 @@
 import structlog
 from google.cloud import secretmanager
 import google.api_core
+from whispr.logging import log_secret_fetch
 from whispr.vault import SimpleVault
 
 
@@ -38,6 +39,8 @@ def fetch_secrets(self, secret_name: str) -> str:
             response = self.client.access_secret_version(name=secret_name)
             secret_data = response.payload.data.decode("UTF-8")
             self.logger.info(f"Successfully fetched gcp secret: {secret_name}")
+            if secret_data:
+                log_secret_fetch(self.logger, secret_name, "gcp")
             return secret_data
         except google.api_core.exceptions.NotFound:
             self.logger.error(

src/whispr/logging.py

@@ -1,14 +1,58 @@
 """Logger configuration"""
 
 import logging
-import sys
+import os
+import platform
+from pathlib import Path
+from datetime import datetime, timezone
 
 import structlog
 
 
 # Configure structlog with human-readable output
+def _default_log_path() -> str:
+    env_path = os.getenv("WHISPR_LOG_PATH")
+    if env_path:
+        return env_path
+
+    if os.name == "nt":
+        base_dir = os.getenv("PROGRAMDATA") or os.getenv("LOCALAPPDATA") or str(Path.home())
+    else:
+        base_dir = "/var/log"
+    return str(Path(base_dir) / "whispr" / "access.log")
+
+
+def _resolve_log_path() -> str:
+    log_path = _default_log_path()
+    if _ensure_writable_log_path(log_path):
+        return log_path
+
+    if platform.system() == "Darwin":
+        fallback_dir = Path.home() / "Library" / "Logs" / "whispr"
+    else:
+        fallback_dir = Path.home() / ".local" / "state" / "whispr"
+    fallback_path = str(fallback_dir / "access.log")
+    if _ensure_writable_log_path(fallback_path):
+        return fallback_path
+
+    return str(Path.cwd() / "whispr_access.log")
+
+
+def _ensure_writable_log_path(log_path: str) -> bool:
+    log_file = Path(log_path)
+    try:
+        log_file.parent.mkdir(parents=True, exist_ok=True)
+        with log_file.open("a", encoding="utf-8"):
+            pass
+        return True
+    except OSError:
+        return False
+
+
 def setup_structlog() -> structlog.BoundLogger:
     """Initializes a structured logger"""
+    log_path = _resolve_log_path()
+
     structlog.configure(
         processors=[
             structlog.stdlib.filter_by_level,
@@ -27,11 +71,25 @@ def setup_structlog() -> structlog.BoundLogger:
     )
 
     # Set up basic configuration for the standard library logging
-    logging.basicConfig(format="%(message)s", stream=sys.stdout, level=logging.ERROR)
+    logging.basicConfig(format="%(message)s", handlers=[logging.FileHandler(log_path)], level=logging.INFO)
 
     # Return the structlog logger instance
     return structlog.get_logger()
 
 
 # Initialize logger
 logger = setup_structlog()
+
+
+def log_secret_fetch(
+    logger_instance: structlog.BoundLogger,
+    secret_name: str,
+    vault_type: str,
+) -> None:
+    """Log a fetched secret with a timezone-aware timestamp."""
+    logger_instance.info(
+        "Secret fetched",
+        secret_name=secret_name,
+        vault_type=vault_type,
+        fetched_at=datetime.now(timezone.utc).isoformat(),
+    )

tests/test_azure.py

@@ -1,7 +1,7 @@
 """Tests for Azure module"""
 
 import unittest
-from unittest.mock import Mock, MagicMock
+from unittest.mock import Mock, MagicMock, ANY
 
 import structlog
 from azure.core.exceptions import ResourceNotFoundError
@@ -37,8 +37,12 @@ def test_fetch_secrets_success(self):
 
         result = self.vault.fetch_secrets("test_secret")
         self.assertEqual(result, '{"key": "value"}')
-        self.mock_logger.info.assert_called_with(
-            "Successfully fetched secret: test_secret"
+        self.mock_logger.info.assert_any_call("Successfully fetched secret: test_secret")
+        self.mock_logger.info.assert_any_call(
+            "Secret fetched",
+            secret_name="test_secret",
+            vault_type="azure",
+            fetched_at=ANY,
         )
         self.mock_client.get_secret.assert_called_with("test_secret")
 

tests/test_gcp.py

@@ -1,7 +1,7 @@
 """Tests for GCP module"""
 
 import unittest
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, ANY
 
 import google.api_core.exceptions
 
@@ -35,9 +35,15 @@ def test_fetch_secrets_success(self):
 
         result = self.vault.fetch_secrets("test_secret")
         self.assertEqual(result, '{"key": "value"}')
-        self.mock_logger.info.assert_called_with(
+        self.mock_logger.info.assert_any_call(
             "Successfully fetched gcp secret: projects/test_project_id/secrets/test_secret/versions/latest"
         )
+        self.mock_logger.info.assert_any_call(
+            "Secret fetched",
+            secret_name="projects/test_project_id/secrets/test_secret/versions/latest",
+            vault_type="gcp",
+            fetched_at=ANY,
+        )
         self.mock_client.access_secret_version.assert_called_with(
             name="projects/test_project_id/secrets/test_secret/versions/latest"
         )