Improve Handling of Time Data (#25)

* Prepped for adding tests

* Handle malformed timestamps, improve handling for missing timestamps

* Add tests for vault_time module

* Update poetry.lock with correct dependency setup

* Bump version: 0.3.8 → 0.3.9

* Correct UTC handling by swapping to timezone aware usage of datetime

* Apply formatting

* Update pytest execution to properly include coverage

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.3.8
+current_version = 0.3.9
 commit = True
 tag = False
 message = Bump version: {current_version} → {new_version}

.github/workflows/pr-checks.yml

@@ -6,7 +6,7 @@ on:
   workflow_dispatch:
 
 env:
-  REQUIRED_COVERAGE: 0
+  REQUIRED_COVERAGE: 30
 
 jobs:
   python:
@@ -35,7 +35,7 @@ jobs:
       run: poetry run mypy --config-file pyproject.toml .
 
     - name: Execute tests
-      run: poetry run pytest --cov=vem --cov-fail-under $REQUIRED_COVERAGE
+      run: poetry run pytest --cov-fail-under $REQUIRED_COVERAGE
 
   docker:
     name: docker checks

.github/workflows/release.yml

@@ -6,7 +6,7 @@ on:
   workflow_dispatch:
 
 env:
-  VERSION: 0.3.8
+  VERSION: 0.3.9
 
 jobs:
   release:

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "vault-assesment-prometheus-exporter"
-version = "0.3.8"
+version = "0.3.9"
 description = "Prometheus exporter to monitor custom metadata for KV2 secrets for (self-imposed) expiration."
 authors = ["Eugene Davis <[email protected]>"]
 readme = "README.md"
@@ -31,10 +31,9 @@ types-requests = "2.27.25"
 types-PyYAML = "6.0.7"
 bandit = "^1.7.4"
 bump2version = "^1.0.1"
-pytest-mock = "3.6.1"
+pytest-mock = "^3.7.0"
 mock = "4.0.3"
 
-
 [build-system]
 requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
@@ -161,3 +160,4 @@ ignore-imports = true
 [tool.pytest.ini_options]
 minversion = "6.2.5"
 testpaths = ["tests"]
+addopts = "--cov=vault_monitor --cov-report term-missing"

tests/test_vault_time.py

@@ -0,0 +1,90 @@
+import datetime
+
+import pytest
+from pytest_mock import mocker
+
+from vault_monitor.secret_expiration_monitor.vault_time import ExpirationMetadata
+
+
+@pytest.mark.parametrize("weeks, days, hours, minutes, seconds", [(1, 1, 1, 2, 4), (8, 10, 1105, 20, 4444)])
+def test_from_duration_get_seralized_expiration(weeks, days, hours, minutes, seconds):
+    """
+    Tests that creating from duration results in matching metadata
+    """
+    last_renewed_time = datetime.datetime.now(tz=datetime.timezone.utc)
+    expiration_delta = datetime.timedelta(weeks=weeks, days=days, hours=hours, minutes=minutes, seconds=seconds)
+    expiration = last_renewed_time + expiration_delta
+    expiration_metadata = ExpirationMetadata.from_duration(weeks, days, hours, minutes, seconds)
+
+    metadata_dict = expiration_metadata.get_serialized_expiration_metadata()
+
+    # Give 50 milliseconds grace period between calling utcnow here and the code calling it
+    assert datetime.datetime.fromisoformat(metadata_dict["last_renewed_timestamp"][:-1]) - last_renewed_time < datetime.timedelta(milliseconds=50)
+    assert datetime.datetime.fromisoformat(metadata_dict["expiration_timestamp"][:-1]) - expiration < datetime.timedelta(milliseconds=50)
+
+
+def test_from_metadata_get_seralized_expiration():
+    """
+    Tests that creating from metadata results in matching metadata being available
+    """
+    metadata = {"expiration_timestamp": "2022-08-08T09:49:41.415869Z", "last_renewed_timestamp": "2022-05-02T09:49:41.415869Z"}
+
+    expiration_metadata = ExpirationMetadata.from_metadata(metadata)
+    assert expiration_metadata.get_serialized_expiration_metadata() == metadata
+
+
+@pytest.mark.parametrize(
+    "input, output",
+    [
+        # Missing last_renewed_timestamp
+        (
+            {
+                "expiration_timestamp": "2022-08-08T09:49:41.415869Z",
+            },
+            {"expiration_timestamp": "2022-08-08T09:49:41.415869Z", "last_renewed_timestamp": "1970-01-01T00:00:00+00:00Z"},
+        ),
+        # Missing expiration_timestamp
+        (
+            {
+                "last_renewed_timestamp": "2022-08-08T09:49:41.415869Z",
+            },
+            {"last_renewed_timestamp": "2022-08-08T09:49:41.415869Z", "expiration_timestamp": "1970-01-01T00:00:00+00:00Z"},
+        ),
+        # Timezone dropped from expiration_timestamp
+        (
+            {"expiration_timestamp": "2022-08-08T09:49:41.415869", "last_renewed_timestamp": "2022-05-02T09:49:41.415869Z"},
+            {"expiration_timestamp": "1970-01-01T00:00:00+00:00Z", "last_renewed_timestamp": "2022-05-02T09:49:41.415869Z"},
+        ),
+        # Timezone dropped from last_renewed_timestamp
+        (
+            {"expiration_timestamp": "2022-08-08T09:49:41.415869Z", "last_renewed_timestamp": "2022-05-02T09:49:41.415869"},
+            {"expiration_timestamp": "2022-08-08T09:49:41.415869Z", "last_renewed_timestamp": "1970-01-01T00:00:00+00:00Z"},
+        ),
+        # Malformed expiration_timestamp
+        (
+            {"expiration_timestamp": "2022-08-008T09:49:41.415869Z", "last_renewed_timestamp": "2022-05-02T09:49:41.415869Z"},
+            {"expiration_timestamp": "1970-01-01T00:00:00+00:00Z", "last_renewed_timestamp": "2022-05-02T09:49:41.415869Z"},
+        ),
+        # Malformed last_renewed_timestamp
+        (
+            {"expiration_timestamp": "2022-08-08T09:49:41.415869Z", "last_renewed_timestamp": "2022-05-002T09:49:41.415869Z"},
+            {"expiration_timestamp": "2022-08-08T09:49:41.415869Z", "last_renewed_timestamp": "1970-01-01T00:00:00+00:00Z"},
+        ),
+        # Missing both
+        (
+            {},
+            {"expiration_timestamp": "1970-01-01T00:00:00+00:00Z", "last_renewed_timestamp": "1970-01-01T00:00:00+00:00Z"},
+        ),
+        # Malformed both
+        (
+            {"expiration_timestamp": "2022-008-08T09:49:41.415869Z", "last_renewed_timestamp": "2022-05-002T09:49:41.415869Z"},
+            {"expiration_timestamp": "1970-01-01T00:00:00+00:00Z", "last_renewed_timestamp": "1970-01-01T00:00:00+00:00Z"},
+        ),
+    ],
+)
+def test_missing_fieldname_and_malformed_timestamp_is_zero(input, output):
+    """
+    Tests that missing or malformed timestamps get set to zero (the beginning of the Unix epoch)
+    """
+    expiration_metadata_object = ExpirationMetadata.from_metadata(input)
+    assert expiration_metadata_object.get_serialized_expiration_metadata() == output

vault_monitor/secret_expiration_monitor/vault_time.py

@@ -1,9 +1,9 @@
 """
 Wraps time handling calls to ensure consistent formatting
 """
-
+import logging
 from typing import Dict, TypeVar, Type
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 ExpirationMetadataType = TypeVar("ExpirationMetadataType", bound="ExpirationMetadata")  # pylint: disable=invalid-name
 
@@ -28,13 +28,13 @@ def from_duration(
         expiration_hours: int,
         expiration_minutes: int,
         expiration_seconds: int,
-        last_renewed_timestamp_fieldname: str,
-        expiration_timestamp_fieldname: str,
+        last_renewed_timestamp_fieldname: str = "last_renewed_timestamp",
+        expiration_timestamp_fieldname: str = "expiration_timestamp",
     ) -> ExpirationMetadataType:
         """
         Creates an instance of ExpirationMetadata from the current time for last_renewed_time and gets the expiration from duration input
         """
-        last_renewed_time = datetime.utcnow()
+        last_renewed_time = datetime.now(timezone.utc)
         expiration_delta = timedelta(weeks=expiration_weeks, days=expiration_days, hours=expiration_hours, minutes=expiration_minutes, seconds=expiration_seconds)
 
         expiration_time = last_renewed_time + expiration_delta
@@ -43,23 +43,33 @@ def from_duration(
 
     # Used when reading from a secret
     @classmethod
-    def from_metadata(cls: Type[ExpirationMetadataType], metadata: dict, last_renewed_timestamp_fieldname: str, expiration_timestamp_fieldname: str) -> ExpirationMetadataType:
+    def from_metadata(
+        cls: Type[ExpirationMetadataType], metadata: dict, last_renewed_timestamp_fieldname: str = "last_renewed_timestamp", expiration_timestamp_fieldname: str = "expiration_timestamp"
+    ) -> ExpirationMetadataType:
         """
         Creates an instance of ExpirationMetadata based on custom_metadata from the secret.
         """
         last_renewed_timestamp = metadata.get(last_renewed_timestamp_fieldname, None)
         expiration_timestamp = metadata.get(expiration_timestamp_fieldname, None)
 
-        # Missing fields means we go back to the 70s (for both dates - so well expired)
-        if last_renewed_timestamp:
+        # Missing fields or malformed timestamps means we go back to the 70s, should be very obvious to the user
+        try:
             last_renewed_time = cls.__get_time_from_iso_utc(last_renewed_timestamp)
-        else:
-            last_renewed_time = datetime.fromtimestamp(0)
-
-        if expiration_timestamp:
+        except TypeError:
+            logging.error("Failed to get last_renewed_timestamp due to issues retrieving metadata for %s, setting to 1970.", last_renewed_timestamp_fieldname)
+            last_renewed_time = datetime.fromtimestamp(0, tz=timezone.utc)
+        except ValueError:
+            logging.error("Failed to parse last_renewed_timestamp for %s, setting to 1970.", last_renewed_timestamp_fieldname)
+            last_renewed_time = datetime.fromtimestamp(0, tz=timezone.utc)
+
+        try:
             expiration_time = cls.__get_time_from_iso_utc(expiration_timestamp)
-        else:
-            expiration_time = datetime.fromtimestamp(0)
+        except TypeError:
+            logging.error("Failed to get expiration_timestamp due to issues retrieving metadata for %s, setting to 1970.", expiration_timestamp_fieldname)
+            expiration_time = datetime.fromtimestamp(0, tz=timezone.utc)
+        except ValueError:
+            logging.error("Failed to parse expiration_timestamp_field for %s, setting to 1970.", expiration_timestamp_fieldname)
+            expiration_time = datetime.fromtimestamp(0, tz=timezone.utc)
 
         return cls(last_renewed_time, expiration_time, last_renewed_timestamp_fieldname, expiration_timestamp_fieldname)