From 5534123af6d1e36732c8f818e6149c554c34deb3 Mon Sep 17 00:00:00 2001 From: Rhutvik Pachghare Date: Wed, 5 Aug 2026 19:15:59 -0700 Subject: [PATCH 1/2] Harden encryption failure handling --- src/pulsenet/security/encryption.py | 142 ++++++++++++++-------------- 1 file changed, 73 insertions(+), 69 deletions(-) diff --git a/src/pulsenet/security/encryption.py b/src/pulsenet/security/encryption.py index 586a271..4079fca 100644 --- a/src/pulsenet/security/encryption.py +++ b/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,7 +24,7 @@ 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 @@ -25,7 +32,7 @@ class EncryptionManager: - """AES-256 Fernet encryption with key rotation support.""" + """Authenticated Fernet encryption with explicit key handling.""" def __init__( self, @@ -43,29 +50,29 @@ def __init__( "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", @@ -73,7 +80,6 @@ def _load_or_generate_key(self) -> bytes: ) return key - # Generate new key self._key_source = "generated" key = Fernet.generate_key() try: @@ -81,108 +87,106 @@ def _load_or_generate_key(self) -> bytes: 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") From 800d78b07f5d5f12aa6b936d24f650838982c642 Mon Sep 17 00:00:00 2001 From: Rhutvik Pachghare Date: Wed, 5 Aug 2026 19:16:11 -0700 Subject: [PATCH 2/2] Test fail-closed encryption behavior --- tests/test_encryption_fail_closed.py | 37 ++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) create mode 100644 tests/test_encryption_fail_closed.py diff --git a/tests/test_encryption_fail_closed.py b/tests/test_encryption_fail_closed.py new file mode 100644 index 0000000..9bdcf12 --- /dev/null +++ b/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"))