Harden encryption failure handling

src/pulsenet/security/encryption.py

@@ -1,11 +1,18 @@
 # pyright: reportGeneralTypeIssues=false
 """
-AES-256 Fernet encryption with key rotation and secure key management.
+Authenticated Fernet encryption with key rotation support.
 
-Loads encryption key from:
-  1. Environment variable  PULSENET_ENCRYPTION_KEY
-  2. Local key file  .runtime/pulsenet-fernet.key
-  3. Auto-generates a new key if neither exists
+Important terminology:
+- Fernet provides authenticated symmetric encryption.
+- It must not be described as AES-256-GCM.
+
+Key loading order:
+  1. Environment variable PULSENET_ENCRYPTION_KEY
+  2. Local key file .runtime/pulsenet-fernet.key
+  3. Generate a new local development key only when neither exists
+
+Invalid configured keys fail closed. Decryption failures are never converted into
+valid-looking sensor values.
 """
 
 from __future__ import annotations
@@ -17,15 +24,15 @@
 from typing import Any
 
 import pandas as pd
-from cryptography.fernet import Fernet
+from cryptography.fernet import Fernet, InvalidToken
 
 from pulsenet.logger import get_logger
 
 log = get_logger(__name__)
 
 
 class EncryptionManager:
-    """AES-256 Fernet encryption with key rotation support."""
+    """Authenticated Fernet encryption with explicit key handling."""
 
     def __init__(
         self,
@@ -43,146 +50,143 @@
             "EncryptionManager initialized", extra={"key_source": self._key_source}
         )
 
-    # ------------------------------------------------------------------
-    # Key management
-    # ------------------------------------------------------------------
     def _load_or_generate_key(self) -> bytes:
-        """Load key from env → file → generate new."""
+        """Load a valid key from environment or file, otherwise create a dev key."""
         env_val = os.environ.get(self.key_env_var)
         if env_val:
-            self._key_source = "environment"
-            # Validate the key format - must be 32 url-safe base64-encoded bytes
+            key_bytes = env_val.encode()
             try:
-                key_bytes = env_val.encode()
-                Fernet(key_bytes)  # Validate
-                return key_bytes
-            except Exception:
-                log.warning(
-                    "PULSENET_ENCRYPTION_KEY is not a valid Fernet key, "
-                    "generating new one"
-                )
-                # Fall through to generate
+                Fernet(key_bytes)
+            except (ValueError, TypeError) as exc:
+                raise RuntimeError(
+                    f"{self.key_env_var} is set but is not a valid Fernet key"
+                ) from exc
+            self._key_source = "environment"
+            return key_bytes
 
         if self.key_file.exists():
-            self._key_source = "file"
             key = self.key_file.read_bytes().strip()
+            try:
+                Fernet(key)
+            except (ValueError, TypeError) as exc:
+                raise RuntimeError(
+                    f"Encryption key file is invalid: {self.key_file}"
+                ) from exc
+            self._key_source = "file"
             if self._should_rotate(self.key_file):
                 log.warning(
                     "Encryption key is due for rotation",
                     extra={"age_days": self._key_age_days(self.key_file)},
                 )
             return key
 
-        # Generate new key
         self._key_source = "generated"
         key = Fernet.generate_key()
         try:
             self.key_file.parent.mkdir(parents=True, exist_ok=True)
             self.key_file.write_bytes(key)
             try:
                 os.chmod(self.key_file, 0o600)
-            except Exception:
-                log.warning("Could not set key file permissions (non-fatal on Windows)")
-        except OSError as e:
-            log.error(f"Failed to save generated key to {self.key_file}: {e}")
-
+            except OSError:
+                log.warning("Could not set key file permissions")
+        except OSError as exc:
+            raise RuntimeError(
+                f"Failed to persist generated encryption key to {self.key_file}"
+            ) from exc
         return key
 
     def rotate_key(self) -> bytes:
-        """Generate a new key, back up old one, and save."""
+        """Rotate the active key and preserve the previous key as a backup.
+
+        Existing ciphertext must be re-encrypted before the backup is removed.
+        """
         old_backup = self.key_file.with_suffix(".key.bak")
         if self.key_file.exists():
-            try:
-                self.key_file.rename(old_backup)
-            except OSError as e:
-                log.warning(f"Failed to create key backup: {e}")
+            if old_backup.exists():
+                raise RuntimeError(
+                    f"Refusing rotation because backup already exists: {old_backup}"
+                )
+            self.key_file.rename(old_backup)
 
         new_key = Fernet.generate_key()
-        self.key_file.write_bytes(new_key)
         try:
+            self.key_file.write_bytes(new_key)
             os.chmod(self.key_file, 0o600)
-        except Exception:
-            log.warning(
-                "Could not set key file permissions during rotation (non-fatal)"
-            )
+        except OSError as exc:
+            if old_backup.exists() and not self.key_file.exists():
+                old_backup.rename(self.key_file)
+            raise RuntimeError("Failed to persist rotated encryption key") from exc
+
         self._key = new_key
         self._cipher = Fernet(new_key)
         log.info("Key rotated successfully", extra={"backup": str(old_backup)})
         return new_key
 
     def _should_rotate(self, path: Path) -> bool:
-        """Check if the key file is older than the rotation period."""
         if not path.exists():
             return False
-        age = self._key_age_days(path)
-        return age > self.rotation_days
+        return self._key_age_days(path) > self.rotation_days
 
     @staticmethod
     def _key_age_days(path: Path) -> float:
-        """Return age of a file in days."""
         try:
             return (time.time() - path.stat().st_mtime) / 86400
         except OSError:
             return 0.0
 
-    # ------------------------------------------------------------------
-    # Encrypt / Decrypt primitives
-    # ------------------------------------------------------------------
     def encrypt(self, plaintext: str) -> str:
-        """Encrypt a string to a base64-encoded ciphertext."""
         return self._cipher.encrypt(plaintext.encode()).decode()
 
     def decrypt(self, ciphertext: str) -> str:
-        """Decrypt a base64-encoded ciphertext back to a string."""
-        return self._cipher.decrypt(ciphertext.encode()).decode()
+        try:
+            return self._cipher.decrypt(ciphertext.encode()).decode()
+        except (InvalidToken, UnicodeDecodeError) as exc:
+            raise ValueError("Ciphertext authentication or decoding failed") from exc
 
     def encrypt_bytes(self, data: bytes) -> bytes:
-        """Encrypt raw bytes."""
         return self._cipher.encrypt(data)
 
     def decrypt_bytes(self, data: bytes) -> bytes:
-        """Decrypt raw bytes."""
-        return self._cipher.decrypt(data)
+        try:
+            return self._cipher.decrypt(data)
+        except InvalidToken as exc:
+            raise ValueError("Ciphertext authentication failed") from exc
 
-    # ------------------------------------------------------------------
-    # DataFrame helpers
-    # ------------------------------------------------------------------
     def encrypt_dataframe(self, df: pd.DataFrame) -> pd.DataFrame:
-        """Encrypt every cell in a DataFrame (string representation)."""
         log.info(
             "Encrypting DataFrame", extra={"rows": len(df), "cols": len(df.columns)}
         )
         return pd.DataFrame(
-            df.apply(lambda col: col.astype(str).apply(lambda v: self.encrypt(v)))
+            df.apply(lambda col: col.astype(str).apply(lambda value: self.encrypt(value)))
         )
 
     def decrypt_dataframe(self, df_enc: pd.DataFrame) -> pd.DataFrame:
-        """Decrypt every cell back to string."""
         log.info(
             "Decrypting DataFrame",
             extra={"rows": len(df_enc), "cols": len(df_enc.columns)},
         )
         return pd.DataFrame(
-            df_enc.apply(lambda col: col.astype(str).apply(lambda v: self.decrypt(v)))
+            df_enc.apply(
+                lambda col: col.astype(str).apply(lambda value: self.decrypt(value))
+            )
         )
 
     def decrypt_cell(self, val: str) -> float:
-        """Decrypt a single cell to float (streaming use-case)."""
+        """Decrypt one numeric cell.
+
+        Raises ValueError when authentication fails or plaintext is not numeric.
+        Returning a default sensor value would hide corruption and is prohibited.
+        """
+        plaintext = self.decrypt(val)
         try:
-            return float(self.decrypt(val))
-        except (ValueError, TypeError, Exception) as e:
-            log.debug(f"Cell decryption failed: {e}")
-            return 0.0
+            return float(plaintext)
+        except (TypeError, ValueError) as exc:
+            raise ValueError("Decrypted cell is not a valid numeric value") from exc
 
-    # ------------------------------------------------------------------
-    # API payload helpers
-    # ------------------------------------------------------------------
     def encrypt_payload(self, payload: dict[str, Any]) -> str:
-        """Encrypt a JSON-serializable dict."""
         return self.encrypt(json.dumps(payload, default=str))
 
     def decrypt_payload(self, ciphertext: str) -> dict[str, Any]:
-        """Decrypt back to dict."""
         result = json.loads(self.decrypt(ciphertext))
         if not isinstance(result, dict):
             raise ValueError("Decrypted payload is not a dictionary")

tests/test_encryption_fail_closed.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from cryptography.fernet import Fernet
+
+from pulsenet.security.encryption import EncryptionManager
+
+
+def test_invalid_configured_key_fails_closed(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    monkeypatch.setenv("PULSENET_ENCRYPTION_KEY", "not-a-valid-fernet-key")
+
+    with pytest.raises(RuntimeError, match="not a valid Fernet key"):
+        EncryptionManager(key_file=tmp_path / "unused.key")
+
+
+def test_tampered_ciphertext_never_becomes_sensor_zero(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setenv("PULSENET_ENCRYPTION_KEY", Fernet.generate_key().decode())
+    manager = EncryptionManager(key_file=tmp_path / "unused.key")
+    ciphertext = manager.encrypt("42.5")
+    tampered = ciphertext[:-1] + ("A" if ciphertext[-1] != "A" else "B")
+
+    with pytest.raises(ValueError, match="authentication"):
+        manager.decrypt_cell(tampered)
+
+
+def test_non_numeric_plaintext_fails_instead_of_defaulting_to_zero(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setenv("PULSENET_ENCRYPTION_KEY", Fernet.generate_key().decode())
+    manager = EncryptionManager(key_file=tmp_path / "unused.key")
+
+    with pytest.raises(ValueError, match="not a valid numeric value"):
+        manager.decrypt_cell(manager.encrypt("sensor-error"))