diff --git a/app/pt_credentials.py b/app/pt_credentials.py index e0cab487..137f9ef1 100644 --- a/app/pt_credentials.py +++ b/app/pt_credentials.py @@ -1,211 +1,924 @@ """ Secure credential management for PowerTraderAI+. -Handles encryption/decryption of API keys and private keys. +Handles encryption/decryption, rotation scheduling, and API permission validation. """ import base64 +import getpass import hashlib +import json +import logging import os +import shutil +import socket import stat -from typing import Optional, Tuple +import tempfile +import threading +import time +from dataclasses import dataclass, asdict +from datetime import datetime +from typing import Any, Callable, Dict, List, Optional, Set, Tuple from cryptography.fernet import Fernet from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- +DEFAULT_ROTATION_DAYS = 90 +ROTATION_WARNING_DAYS = 7 +REQUIRED_PERMISSIONS: Set[str] = {"read_account", "read_positions"} +TRADING_PERMISSIONS: Set[str] = {"buy", "sell"} + + +# --------------------------------------------------------------------------- +# Data classes +# --------------------------------------------------------------------------- +@dataclass +class CredentialMetadata: + """Metadata stored alongside encrypted credentials.""" + + created_at: float + last_rotated_at: float + rotation_due_at: float + rotation_interval_days: int = DEFAULT_ROTATION_DAYS + + def is_rotation_due(self) -> bool: + return time.time() >= self.rotation_due_at + + def days_until_rotation(self) -> int: + return max(0, int((self.rotation_due_at - time.time()) / 86400)) + + def to_dict(self) -> dict: + return asdict(self) + + @classmethod + def from_dict(cls, d: dict) -> "CredentialMetadata": + """Build from dict. Filters unknown keys; raises ValueError when + required fields are missing (instead of an opaque TypeError).""" + required = {"created_at", "last_rotated_at", "rotation_due_at"} + missing = required - d.keys() + if missing: + raise ValueError( + f"CredentialMetadata missing required fields: {sorted(missing)}" + ) + return cls(**{k: v for k, v in d.items() if k in cls.__dataclass_fields__}) + + @classmethod + def new(cls, interval_days: int = DEFAULT_ROTATION_DAYS) -> "CredentialMetadata": + now = time.time() + return cls( + created_at=now, + last_rotated_at=now, + rotation_due_at=now + interval_days * 86400, + rotation_interval_days=interval_days, + ) + + +@dataclass +class PermissionAuditResult: + """Result of an API permission validation check.""" + + timestamp: float + has_required: bool + has_trading: bool + granted_permissions: List[str] + missing_required: List[str] + missing_trading: List[str] + audit_passed: bool + message: str + def to_dict(self) -> dict: + return asdict(self) + + +# --------------------------------------------------------------------------- +# SecureCredentialManager +# --------------------------------------------------------------------------- class SecureCredentialManager: - """Manages encrypted storage of API credentials.""" + """Manages encrypted storage and rotation of API credentials.""" def __init__(self, base_dir: str = None): self.base_dir = base_dir or os.path.dirname(os.path.abspath(__file__)) self.salt_file = os.path.join(self.base_dir, ".pt_salt") self.encrypted_key_file = os.path.join(self.base_dir, "r_key.enc") self.encrypted_secret_file = os.path.join(self.base_dir, "r_secret.enc") + self.metadata_file = os.path.join(self.base_dir, ".pt_cred_meta") + self._lock = threading.RLock() + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ def _get_or_create_salt(self) -> bytes: - """Get existing salt or create a new one.""" if os.path.exists(self.salt_file): with open(self.salt_file, "rb") as f: return f.read() - else: - salt = os.urandom(16) - self._secure_write_binary(self.salt_file, salt) - return salt + salt = os.urandom(16) + self._atomic_write_binary(self.salt_file, salt) + return salt def _derive_key(self, password: str, salt: bytes) -> bytes: - """Derive encryption key from password and salt.""" kdf = PBKDF2HMAC( - algorithm=hashes.SHA256(), - length=32, - salt=salt, - iterations=100000, + algorithm=hashes.SHA256(), length=32, salt=salt, iterations=100_000 ) return base64.urlsafe_b64encode(kdf.derive(password.encode())) def _get_machine_password(self) -> str: - """Generate a machine-specific password for encryption.""" - # Use machine-specific identifiers for password generation - machine_info = ( - f"{os.environ.get('COMPUTERNAME', '')}{os.environ.get('USERNAME', '')}" + """ + Cross-platform machine-specific password using hostname + username. + Avoids Windows-only COMPUTERNAME/USERNAME env vars. + """ + try: + host = socket.gethostname() + except OSError: + host = "" + try: + user = getpass.getuser() + except OSError: + user = os.environ.get("USER", os.environ.get("USERNAME", "")) + return hashlib.sha256(f"{host}{user}".encode()).hexdigest()[:32] + + def _get_legacy_machine_password(self) -> Optional[str]: + """ + Legacy derivation (Windows-only, pre-cross-platform). Returned only + when COMPUTERNAME/USERNAME env vars are present so decrypt_credentials + can fall back transparently for vaults encrypted by older versions. + """ + host = os.environ.get("COMPUTERNAME") + user = os.environ.get("USERNAME") + if not host or not user: + return None + return hashlib.sha256(f"{host}{user}".encode()).hexdigest()[:32] + + def _atomic_write_text(self, filepath: str, content: str) -> None: + """Write text file atomically via unique temp → rename. Uses + tempfile.mkstemp so concurrent writers cannot collide on the same + temp name.""" + directory = os.path.dirname(filepath) or "." + fd, tmp = tempfile.mkstemp( + prefix=os.path.basename(filepath) + ".", suffix=".tmp", dir=directory ) - return hashlib.sha256(machine_info.encode()).hexdigest()[:32] - - def _secure_write_text(self, filepath: str, content: str) -> None: - """Write text file with secure permissions.""" - with open(filepath, "w", encoding="utf-8") as f: - f.write(content) - self._set_secure_permissions(filepath) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(content) + self._set_secure_permissions(tmp) + os.replace(tmp, filepath) + self._set_secure_permissions(filepath) + except Exception: + try: + os.remove(tmp) + except OSError: + pass + raise + + def _stage_temp_binary(self, filepath: str, content: bytes) -> str: + """Write ``content`` to a sibling temp file and return its path. + + The temp file lives next to ``filepath`` (same filesystem) so the + subsequent ``os.replace`` is atomic. Caller is responsible for the + rename or for unlinking on failure. + """ + directory = os.path.dirname(filepath) or "." + fd, tmp = tempfile.mkstemp( + prefix=os.path.basename(filepath) + ".", suffix=".tmp", dir=directory + ) + try: + with os.fdopen(fd, "wb") as f: + f.write(content) + self._set_secure_permissions(tmp) + return tmp + except Exception: + try: + os.remove(tmp) + except OSError: + pass + raise - def _secure_write_binary(self, filepath: str, content: bytes) -> None: - """Write binary file with secure permissions.""" - with open(filepath, "wb") as f: - f.write(content) - self._set_secure_permissions(filepath) + def _atomic_write_binary(self, filepath: str, content: bytes) -> None: + """Write binary file atomically via unique temp → rename.""" + tmp = self._stage_temp_binary(filepath, content) + try: + os.replace(tmp, filepath) + self._set_secure_permissions(filepath) + except Exception: + try: + os.remove(tmp) + except OSError: + pass + raise def _set_secure_permissions(self, filepath: str) -> None: - """Set file permissions to owner read/write only.""" try: - # On Windows, this sets the file to be accessible only by the owner os.chmod(filepath, stat.S_IRUSR | stat.S_IWUSR) except (OSError, AttributeError): - # Fallback for systems that don't support chmod pass - def encrypt_credentials(self, api_key: str, private_key_b64: str) -> bool: - """Encrypt and save credentials.""" + # ------------------------------------------------------------------ + # Metadata management + # ------------------------------------------------------------------ + def _load_metadata(self) -> Optional[CredentialMetadata]: + if not os.path.exists(self.metadata_file): + return None try: - salt = self._get_or_create_salt() - password = self._get_machine_password() - key = self._derive_key(password, salt) - cipher = Fernet(key) + with open(self.metadata_file, "r", encoding="utf-8") as f: + return CredentialMetadata.from_dict(json.load(f)) + except (OSError, json.JSONDecodeError, ValueError, TypeError): + return None - # Encrypt API key - encrypted_api_key = cipher.encrypt(api_key.encode("utf-8")) - self._secure_write_binary(self.encrypted_key_file, encrypted_api_key) + def _save_metadata(self, meta: CredentialMetadata) -> None: + self._atomic_write_text( + self.metadata_file, json.dumps(meta.to_dict(), indent=2) + ) - # Encrypt private key - encrypted_private_key = cipher.encrypt(private_key_b64.encode("utf-8")) - self._secure_write_binary(self.encrypted_secret_file, encrypted_private_key) + # ------------------------------------------------------------------ + # Core encrypt / decrypt + # ------------------------------------------------------------------ + def encrypt_credentials( + self, + api_key: str, + private_key_b64: str, + rotation_interval_days: int = DEFAULT_ROTATION_DAYS, + ) -> bool: + """ + Encrypt and persist credentials atomically. + + Two-phase commit: both ciphertexts are written to temp files first and + only then renamed into place. If the second os.replace fails after the + first has already swapped the key file, the previously-saved key file + is restored from an in-process backup so the vault never ends up with + a new key ciphertext paired with the old secret ciphertext. + """ + with self._lock: + try: + salt = self._get_or_create_salt() + key = self._derive_key(self._get_machine_password(), salt) + cipher = Fernet(key) - return True - except Exception: - return False + key_ct = cipher.encrypt(api_key.encode("utf-8")) + secret_ct = cipher.encrypt(private_key_b64.encode("utf-8")) - def decrypt_credentials(self) -> Optional[Tuple[str, str]]: - """Decrypt and return credentials.""" - try: - if not ( - os.path.exists(self.encrypted_key_file) - and os.path.exists(self.encrypted_secret_file) - and os.path.exists(self.salt_file) - ): - return None + # Phase 1: stage both ciphertexts on disk before committing + # either. If staging fails for one, neither gets renamed in. + key_tmp = self._stage_temp_binary(self.encrypted_key_file, key_ct) + try: + secret_tmp = self._stage_temp_binary( + self.encrypted_secret_file, secret_ct + ) + except Exception: + try: + os.remove(key_tmp) + except OSError: + pass + raise + + # Phase 2: commit both. If the second commit fails after the + # first has swapped the key file, restore the previous key + # ciphertext from a snapshot kept in memory. + prev_key_blob: Optional[bytes] = None + if os.path.exists(self.encrypted_key_file): + try: + with open(self.encrypted_key_file, "rb") as _f: + prev_key_blob = _f.read() + except OSError: + prev_key_blob = None - # Load salt - with open(self.salt_file, "rb") as f: - salt = f.read() + try: + os.replace(key_tmp, self.encrypted_key_file) + self._set_secure_permissions(self.encrypted_key_file) + except Exception: + try: + os.remove(key_tmp) + except OSError: + pass + try: + os.remove(secret_tmp) + except OSError: + pass + raise - password = self._get_machine_password() - key = self._derive_key(password, salt) - cipher = Fernet(key) + try: + os.replace(secret_tmp, self.encrypted_secret_file) + self._set_secure_permissions(self.encrypted_secret_file) + except Exception: + # Roll the key file back so callers don't see a mismatched + # ciphertext pair after we return False. + if prev_key_blob is not None: + try: + with open(self.encrypted_key_file, "wb") as _f: + _f.write(prev_key_blob) + self._set_secure_permissions(self.encrypted_key_file) + except OSError: + logger.error( + "Rollback of key ciphertext failed after secret " + "write error; vault may be inconsistent" + ) + try: + os.remove(secret_tmp) + except OSError: + pass + raise + + # Update metadata — keep interval consistent + existing = self._load_metadata() + now = time.time() + if existing: + existing.last_rotated_at = now + existing.rotation_due_at = now + rotation_interval_days * 86400 + existing.rotation_interval_days = ( + rotation_interval_days # keep consistent + ) + meta = existing + else: + meta = CredentialMetadata.new(rotation_interval_days) + self._save_metadata(meta) + + logger.info("Credentials encrypted and saved successfully") + return True + except Exception as exc: + logger.error("Failed to encrypt credentials: %s", exc) + return False + + def decrypt_credentials(self) -> Optional[Tuple[str, str]]: + """Decrypt and return (api_key, private_key_b64), or None on failure. + + Falls back to the legacy COMPUTERNAME/USERNAME derivation if the new + gethostname()/getuser() derivation fails — this allows vaults encrypted + by pre-cross-platform versions to decrypt without user intervention. + On successful legacy decrypt, re-encrypts with the new key so the + fallback is one-shot per vault. + """ + with self._lock: + if not self.has_encrypted_credentials(): + return None + try: + with open(self.salt_file, "rb") as f: + salt = f.read() + with open(self.encrypted_key_file, "rb") as f: + key_blob = f.read() + with open(self.encrypted_secret_file, "rb") as f: + secret_blob = f.read() + except OSError as exc: + logger.error("Failed to read vault files: %s", exc) + return None - # Decrypt API key - with open(self.encrypted_key_file, "rb") as f: - encrypted_api_key = f.read() - api_key = cipher.decrypt(encrypted_api_key).decode("utf-8") + # Try current derivation first + try: + cipher = Fernet(self._derive_key(self._get_machine_password(), salt)) + api_key = cipher.decrypt(key_blob).decode("utf-8").strip() + private_key = cipher.decrypt(secret_blob).decode("utf-8").strip() + return api_key, private_key + except Exception as exc: + logger.debug( + "Primary decrypt failed, trying legacy derivation: %s", exc + ) + + # Fallback: legacy Windows derivation + legacy_pw = self._get_legacy_machine_password() + if legacy_pw is None: + logger.error("Failed to decrypt credentials with current derivation") + return None + try: + cipher = Fernet(self._derive_key(legacy_pw, salt)) + api_key = cipher.decrypt(key_blob).decode("utf-8").strip() + private_key = cipher.decrypt(secret_blob).decode("utf-8").strip() + except Exception as exc: + logger.error("Failed to decrypt credentials (legacy fallback): %s", exc) + return None - # Decrypt private key - with open(self.encrypted_secret_file, "rb") as f: - encrypted_private_key = f.read() - private_key_b64 = cipher.decrypt(encrypted_private_key).decode("utf-8") + logger.warning( + "Decrypted vault using legacy machine-password derivation. " + "Re-encrypting with new derivation." + ) + # Snapshot rotation metadata so the derivation migration does not + # masquerade as a real credential rotation. encrypt_credentials() + # otherwise resets last_rotated_at / rotation_due_at and would + # silently push the next rotation warning out by a full interval. + prior_meta = self._load_metadata() + try: + if self.encrypt_credentials(api_key, private_key) and prior_meta: + refreshed = self._load_metadata() + if refreshed is not None: + refreshed.last_rotated_at = prior_meta.last_rotated_at + refreshed.rotation_due_at = prior_meta.rotation_due_at + refreshed.rotation_interval_days = ( + prior_meta.rotation_interval_days + ) + refreshed.created_at = prior_meta.created_at + self._save_metadata(refreshed) + except Exception as exc: + logger.warning("Re-encrypt after legacy decrypt failed: %s", exc) + return api_key, private_key + + # ------------------------------------------------------------------ + # Rotation + # ------------------------------------------------------------------ + def get_rotation_status(self) -> Dict[str, Any]: + """Return a dict with a consistent shape regardless of metadata + presence (all keys always present; None when not applicable).""" + meta = self._load_metadata() + if not meta: + return { + "has_metadata": False, + "rotation_due": False, + "days_until_rotation": None, + "last_rotated_at": None, + "rotation_due_at": None, + } + return { + "has_metadata": True, + "rotation_due": meta.is_rotation_due(), + "days_until_rotation": meta.days_until_rotation(), + "last_rotated_at": datetime.fromtimestamp(meta.last_rotated_at).isoformat(), + "rotation_due_at": datetime.fromtimestamp(meta.rotation_due_at).isoformat(), + } + + def rotate_credentials( + self, + new_api_key: str, + new_private_key_b64: str, + rotation_interval_days: int = DEFAULT_ROTATION_DAYS, + ) -> bool: + """ + Gracefully rotate credentials with full rollback on failure. + Backs up key, secret, AND metadata so all three are restored together. + """ + with self._lock: + backup_key = self.encrypted_key_file + ".bak" + backup_secret = self.encrypted_secret_file + ".bak" + backup_meta = self.metadata_file + ".bak" + backed_up = False + + # Snapshot created_at before rotation so it is never silently reset. + # encrypt_credentials() constructs new metadata (resetting created_at) + # when the metadata file is missing or corrupt at rotation time. + prior_created_at: Optional[float] = None + prior_meta = self._load_metadata() + if prior_meta is not None: + prior_created_at = prior_meta.created_at - return api_key.strip(), private_key_b64.strip() - except Exception: + try: + if self.has_encrypted_credentials(): + shutil.copy2(self.encrypted_key_file, backup_key) + shutil.copy2(self.encrypted_secret_file, backup_secret) + if os.path.exists(self.metadata_file): + shutil.copy2(self.metadata_file, backup_meta) + backed_up = True + + if self.encrypt_credentials( + new_api_key, new_private_key_b64, rotation_interval_days + ): + # Restore original created_at if encrypt_credentials reset it + if prior_created_at is not None: + meta = self._load_metadata() + if meta is not None and meta.created_at != prior_created_at: + meta.created_at = prior_created_at + self._save_metadata(meta) + + for f in (backup_key, backup_secret, backup_meta): + try: + os.remove(f) + except OSError: + pass + logger.info("Credentials rotated successfully") + return True + + raise RuntimeError("encrypt_credentials returned False") + + except Exception as exc: + logger.error("Credential rotation failed: %s", exc) + if backed_up: + try: + # os.replace is atomic (POSIX rename): no partial-restore window + os.replace(backup_key, self.encrypted_key_file) + os.replace(backup_secret, self.encrypted_secret_file) + if os.path.exists(backup_meta): + os.replace(backup_meta, self.metadata_file) + logger.info("Rolled back to previous credentials") + except OSError as restore_exc: + logger.critical( + "CRITICAL: Failed to restore credentials after rotation failure: %s", + restore_exc, + ) + return False + + def check_rotation_warning(self) -> Optional[str]: + """Return a warning string if rotation due soon/overdue, else None.""" + status = self.get_rotation_status() + if not status["has_metadata"]: return None - + days = status["days_until_rotation"] + if status["rotation_due"]: + return ( + f"SECURITY WARNING: API credentials rotation is OVERDUE. " + f"Last rotated: {status['last_rotated_at']}. Please rotate immediately." + ) + if days is not None and days <= ROTATION_WARNING_DAYS: + return ( + f"SECURITY NOTICE: API credentials rotation due in {days} day(s). " + f"Due at: {status['rotation_due_at']}." + ) + return None + + # ------------------------------------------------------------------ + # Migration + # ------------------------------------------------------------------ def migrate_from_plaintext(self) -> bool: - """Migrate existing plaintext credentials to encrypted format.""" + """Migrate existing plaintext r_key.txt / r_secret.txt to encrypted.""" key_file = os.path.join(self.base_dir, "r_key.txt") secret_file = os.path.join(self.base_dir, "r_secret.txt") - if not (os.path.exists(key_file) and os.path.exists(secret_file)): return False - try: - # Read plaintext credentials with open(key_file, "r", encoding="utf-8") as f: api_key = f.read().strip() with open(secret_file, "r", encoding="utf-8") as f: - private_key_b64 = f.read().strip() - - # Encrypt them - if self.encrypt_credentials(api_key, private_key_b64): - # Securely delete plaintext files - try: - os.remove(key_file) - os.remove(secret_file) - except OSError: - pass + private_key = f.read().strip() + if self.encrypt_credentials(api_key, private_key): + for path in (key_file, secret_file): + try: + os.remove(path) + except OSError: + pass + logger.info("Migrated plaintext credentials to encrypted storage") return True - except Exception: - pass + except Exception as exc: + logger.error("Plaintext migration failed: %s", exc) return False + # ------------------------------------------------------------------ + # State checks + # ------------------------------------------------------------------ def has_encrypted_credentials(self) -> bool: - """Check if encrypted credentials exist.""" - return ( - os.path.exists(self.encrypted_key_file) - and os.path.exists(self.encrypted_secret_file) - and os.path.exists(self.salt_file) + return all( + os.path.exists(p) + for p in ( + self.encrypted_key_file, + self.encrypted_secret_file, + self.salt_file, + ) ) def has_plaintext_credentials(self) -> bool: - """Check if plaintext credentials exist.""" - key_file = os.path.join(self.base_dir, "r_key.txt") - secret_file = os.path.join(self.base_dir, "r_secret.txt") - return os.path.exists(key_file) and os.path.exists(secret_file) + return all( + os.path.exists(os.path.join(self.base_dir, f)) + for f in ("r_key.txt", "r_secret.txt") + ) + + +# --------------------------------------------------------------------------- +# PermissionValidator +# --------------------------------------------------------------------------- +class PermissionValidator: + """Validates API key permissions on startup against required permission sets. + + Note on default base_dir: resolves to the source directory of this module + when not supplied. For production installs to read-only locations, pass a + writable base_dir explicitly (e.g. user data dir). + """ + + AUDIT_LOG_FILE = "credential_audit.jsonl" + MAX_AUDIT_LINES = 10_000 # Soft cap; older entries rotated to .1 + AUDIT_ROTATION_KEEP = 1 # Number of rotated backups to keep + + def __init__(self, base_dir: str = None): + self.base_dir = base_dir or os.path.dirname(os.path.abspath(__file__)) + self._audit_log = os.path.join(self.base_dir, self.AUDIT_LOG_FILE) + + def validate( + self, + permission_fetcher: Optional[Callable[[], List[str]]] = None, + require_trading: bool = False, + ) -> PermissionAuditResult: + """ + Validate API permissions. + + Args: + permission_fetcher: Callable → list of permission strings. + If None, returns failed audit (offline/CI use). + require_trading: Also check for buy/sell permissions. + """ + now = time.time() + + if permission_fetcher is None: + result = PermissionAuditResult( + timestamp=now, + has_required=False, + has_trading=False, + granted_permissions=[], + missing_required=sorted(REQUIRED_PERMISSIONS), + missing_trading=( + sorted(TRADING_PERMISSIONS) if require_trading else [] + ), + audit_passed=False, + message=( + "No permission fetcher provided — unable to validate API permissions. " + "Provide a permission_fetcher callable to enable validation." + ), + ) + self._log_audit(result) + return result + + try: + granted = set(permission_fetcher()) + except Exception as exc: + result = PermissionAuditResult( + timestamp=now, + has_required=False, + has_trading=False, + granted_permissions=[], + missing_required=sorted(REQUIRED_PERMISSIONS), + missing_trading=( + sorted(TRADING_PERMISSIONS) if require_trading else [] + ), + audit_passed=False, + message=f"Permission fetch failed: {exc}", + ) + self._log_audit(result) + logger.error("API permission validation failed: %s", exc) + return result + + missing_required = sorted(REQUIRED_PERMISSIONS - granted) + missing_trading = ( + sorted(TRADING_PERMISSIONS - granted) if require_trading else [] + ) + has_required = len(missing_required) == 0 + has_trading = len(missing_trading) == 0 + audit_passed = has_required and (has_trading if require_trading else True) + + if audit_passed: + message = "API permission validation passed." + elif not has_required: + message = ( + f"SECURITY ALERT: API key is missing required permissions: " + f"{missing_required}. Trading is disabled." + ) + logger.critical(message) + else: + message = ( + f"WARNING: API key is missing trading permissions: {missing_trading}. " + f"Live trading will be unavailable." + ) + logger.warning(message) + + result = PermissionAuditResult( + timestamp=now, + has_required=has_required, + has_trading=has_trading, + granted_permissions=sorted(granted), + missing_required=missing_required, + missing_trading=missing_trading, + audit_passed=audit_passed, + message=message, + ) + self._log_audit(result) + return result + + def _log_audit(self, result: PermissionAuditResult) -> None: + """Append audit result to JSONL log. Rotates when MAX_AUDIT_LINES is + reached (renames active log to ``*.1``, drops older rotations). This + avoids the per-call O(n) read/rewrite of the previous trim strategy.""" + try: + self._rotate_audit_if_needed() + with open(self._audit_log, "a", encoding="utf-8") as f: + f.write(json.dumps(result.to_dict()) + "\n") + try: + os.chmod(self._audit_log, stat.S_IRUSR | stat.S_IWUSR) + except (OSError, AttributeError): + pass + except OSError as exc: + logger.warning("Could not write permission audit log: %s", exc) + + def _rotate_audit_if_needed(self) -> None: + """Rotate audit log when line count reaches MAX_AUDIT_LINES. + + Uses size-based heuristic to avoid reading the file every call: only + counts lines if the file is large enough to plausibly hit the cap. + Approximate entry size is ~200 bytes; we trigger the precise check + once the file exceeds MAX_AUDIT_LINES * 100 bytes. + """ + try: + size = os.path.getsize(self._audit_log) + except OSError: + return + if size < self.MAX_AUDIT_LINES * 100: + return # cheap path: nowhere near the cap + try: + with open(self._audit_log, "rb") as f: + line_count = sum(1 for _ in f) + except OSError: + return + if line_count < self.MAX_AUDIT_LINES: + return + # Rotate: shift active → .1 → .2 → ... keep at most AUDIT_ROTATION_KEEP + # backups. Older generations are discarded. + keep = max(1, int(self.AUDIT_ROTATION_KEEP)) + try: + # Drop the oldest generation beyond the retention window. + oldest = f"{self._audit_log}.{keep}" + if os.path.exists(oldest): + os.remove(oldest) + # Shift down: .N-1 → .N, ..., .1 → .2 + for i in range(keep - 1, 0, -1): + src = f"{self._audit_log}.{i}" + dst = f"{self._audit_log}.{i + 1}" + if os.path.exists(src): + os.replace(src, dst) + # Active log → .1 + rotated = f"{self._audit_log}.1" + os.replace(self._audit_log, rotated) + try: + os.chmod(rotated, stat.S_IRUSR | stat.S_IWUSR) + except (OSError, AttributeError): + pass + except OSError as exc: + logger.warning("Audit log rotation failed: %s", exc) + + def get_audit_history(self, limit: int = 50) -> List[dict]: + if not os.path.exists(self._audit_log): + return [] + try: + with open(self._audit_log, "r", encoding="utf-8") as f: + lines = f.readlines() + return [json.loads(line) for line in lines[-limit:] if line.strip()] + except (OSError, json.JSONDecodeError): + return [] + + +# --------------------------------------------------------------------------- +# CredentialRotationScheduler +# --------------------------------------------------------------------------- +class CredentialRotationScheduler: + """ + Background scheduler: checks rotation status periodically and fires a + notification callback when action is needed. + De-duplicates notifications — only calls the callback when the warning + message changes, preventing repeated identical alerts on every tick while + credentials remain overdue. + """ + + MIN_INTERVAL_SECONDS = 60 # Lower bound on tick interval + + def __init__( + self, + notification_callback: Callable[[str], None], + check_interval_hours: float = 24.0, + base_dir: str = None, + ): + self._callback = notification_callback + raw_interval = check_interval_hours * 3600 + self._interval = max(raw_interval, self.MIN_INTERVAL_SECONDS) + if raw_interval < self.MIN_INTERVAL_SECONDS: + logger.warning( + "check_interval_hours=%.4f clamped to %ds minimum", + check_interval_hours, + self.MIN_INTERVAL_SECONDS, + ) + self._manager = SecureCredentialManager(base_dir) + self._stop_event = threading.Event() + self._thread: Optional[threading.Thread] = None + self._last_warning: Optional[str] = None # de-dup tracker + + def start(self) -> None: + if self._thread and self._thread.is_alive(): + return + self._stop_event.clear() + self._thread = threading.Thread( + target=self._run, name="CredentialRotationScheduler", daemon=True + ) + self._thread.start() + logger.info( + "Credential rotation scheduler started (interval: %.1fh)", + self._interval / 3600, + ) + def stop(self) -> None: + self._stop_event.set() + if self._thread: + self._thread.join(timeout=5) + if self._thread.is_alive(): + logger.warning( + "Credential rotation scheduler thread did not exit within 5s" + ) + return + logger.info("Credential rotation scheduler stopped") + + def _tick(self) -> None: + """Single check + dedup-callback cycle. Extracted so tests can exercise + the real dedup path without simulating it. + + ``_last_warning`` is updated *before* the callback fires so a failing + callback does not cause the scheduler to re-fire the same warning on + every subsequent tick. Callback exceptions are logged and isolated.""" + try: + warning = self._manager.check_rotation_warning() + except Exception as exc: + logger.error("Rotation scheduler check failed: %s", exc) + return + if warning and warning != self._last_warning: + logger.warning(warning) + # Update dedup state first so a callback exception cannot cause + # the same warning to be re-fired on every subsequent tick. + self._last_warning = warning + try: + self._callback(warning) + except Exception as cb_exc: + logger.error( + "Rotation notification callback raised: %s", cb_exc, exc_info=True + ) + return + self._last_warning = warning + + def _run(self) -> None: + while not self._stop_event.is_set(): + self._tick() + self._stop_event.wait(timeout=self._interval) + + def check_now(self) -> Optional[str]: + """Immediate one-shot check. Returns warning string or None.""" + return self._manager.check_rotation_warning() + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- def get_credentials() -> Optional[Tuple[str, str]]: """ - Get API credentials with priority order: - 1. Encrypted credentials (for desktop use) - 2. Environment variables (for CI/CD) - 3. Plaintext files (legacy support) - Returns (api_key, private_key_b64) or None if not found. + Get API credentials with priority: + 1. Encrypted vault + 2. Environment variables (CI/CD) + 3. Auto-migrate from plaintext (last resort — also preserves plaintext + fallback if vault write fails, to prevent user lockout) + + Returns (api_key, private_key_b64) or None. """ manager = SecureCredentialManager() - # Try encrypted credentials first (preferred for desktop) if manager.has_encrypted_credentials(): return manager.decrypt_credentials() - # Try environment variables (for CI/CD pipelines) - env_api_key = os.environ.get("POWERTRADER_ROBINHOOD_API_KEY") - env_private_key = os.environ.get("POWERTRADER_ROBINHOOD_PRIVATE_KEY") + env_key = os.environ.get("POWERTRADER_ROBINHOOD_API_KEY") + env_secret = os.environ.get("POWERTRADER_ROBINHOOD_PRIVATE_KEY") + if env_key and env_secret: + return env_key.strip(), env_secret.strip() - if env_api_key and env_private_key: - return env_api_key.strip(), env_private_key.strip() - - # Try to migrate from plaintext if manager.has_plaintext_credentials(): if manager.migrate_from_plaintext(): return manager.decrypt_credentials() - else: - # Fallback to reading plaintext if migration fails - try: - base_dir = os.path.dirname(os.path.abspath(__file__)) - with open( - os.path.join(base_dir, "r_key.txt"), "r", encoding="utf-8" - ) as f: - api_key = f.read().strip() - with open( - os.path.join(base_dir, "r_secret.txt"), "r", encoding="utf-8" - ) as f: - private_key_b64 = f.read().strip() - return api_key, private_key_b64 - except Exception: - pass + # Plaintext fallback: migration failed (e.g. vault write permission + # denied). Return plaintext creds rather than locking the user out. + # Logged at error level so the degraded security posture is visible. + logger.error( + "SECURITY DEGRADATION: encrypted vault write failed — returning " + "PLAINTEXT credentials. Callers cannot distinguish vault-backed " + "from plaintext via this API. Fix vault permissions and re-run " + "to migrate." + ) + try: + base_dir = os.path.dirname(os.path.abspath(__file__)) + with open(os.path.join(base_dir, "r_key.txt"), "r", encoding="utf-8") as f: + api_key = f.read().strip() + with open( + os.path.join(base_dir, "r_secret.txt"), "r", encoding="utf-8" + ) as f: + private_key = f.read().strip() + return api_key, private_key + except OSError: + pass return None + + +def validate_credentials_on_startup( + permission_fetcher: Optional[Callable[[], List[str]]] = None, + require_trading: bool = True, + notify_rotation: Optional[Callable[[str], None]] = None, + base_dir: Optional[str] = None, +) -> Tuple[bool, str]: + """ + Startup validation: checks permission audit AND rotation status. + + Args: + base_dir: Override the default base directory for the credential + vault and audit log. Useful for tests and non-default installs. + + Returns: + (audit_passed: bool, message: str) + """ + manager = SecureCredentialManager(base_dir) + validator = PermissionValidator(base_dir) + messages = [] + + warning = manager.check_rotation_warning() + if warning: + messages.append(warning) + if notify_rotation: + notify_rotation(warning) + + audit = validator.validate(permission_fetcher, require_trading) + messages.append(audit.message) + + return audit.audit_passed, " | ".join(messages) diff --git a/app/test_credentials_rotation.py b/app/test_credentials_rotation.py new file mode 100644 index 00000000..d7b716c1 --- /dev/null +++ b/app/test_credentials_rotation.py @@ -0,0 +1,470 @@ +"""Tests for credential rotation and permission validation (issues #58, #59).""" + +import json +import os +import shutil +import stat +import tempfile +import time +import unittest +from unittest.mock import MagicMock, patch + +from pt_credentials import ( + CredentialMetadata, + CredentialRotationScheduler, + PermissionValidator, + SecureCredentialManager, +) + + +class TestCredentialMetadata(unittest.TestCase): + def test_new_sets_rotation_due_future(self): + meta = CredentialMetadata.new(90) + self.assertFalse(meta.is_rotation_due()) + self.assertGreater(meta.days_until_rotation(), 0) + + def test_overdue_when_past_due(self): + meta = CredentialMetadata( + created_at=time.time() - 200 * 86400, + last_rotated_at=time.time() - 200 * 86400, + rotation_due_at=time.time() - 1, + ) + self.assertTrue(meta.is_rotation_due()) + self.assertEqual(meta.days_until_rotation(), 0) + + def test_roundtrip_dict(self): + meta = CredentialMetadata.new(30) + meta2 = CredentialMetadata.from_dict(meta.to_dict()) + self.assertAlmostEqual(meta.created_at, meta2.created_at, places=3) + self.assertEqual(meta.rotation_interval_days, meta2.rotation_interval_days) + + def test_from_dict_handles_corrupt_metadata(self): + """Missing required fields should be caught by _load_metadata.""" + tmpdir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, tmpdir, ignore_errors=True) + mgr = SecureCredentialManager(tmpdir) + # Write partial/corrupt metadata + with open(mgr.metadata_file, "w") as f: + json.dump({"created_at": 0}, f) # missing required fields + self.assertIsNone(mgr._load_metadata()) + + def test_from_dict_raises_value_error_on_missing_fields(self): + """from_dict must raise ValueError (not opaque TypeError) for missing fields.""" + with self.assertRaises(ValueError): + CredentialMetadata.from_dict({"created_at": 0}) + + +class TestSecureCredentialManager(unittest.TestCase): + def setUp(self): + self.tmpdir = tempfile.mkdtemp() + self.mgr = SecureCredentialManager(self.tmpdir) + + def tearDown(self): + shutil.rmtree(self.tmpdir, ignore_errors=True) + + def test_encrypt_decrypt_roundtrip(self): + self.assertTrue(self.mgr.encrypt_credentials("KEY123", "SECRET456")) + creds = self.mgr.decrypt_credentials() + self.assertIsNotNone(creds) + self.assertEqual(creds[0], "KEY123") + self.assertEqual(creds[1], "SECRET456") + + def test_has_encrypted_after_save(self): + self.assertFalse(self.mgr.has_encrypted_credentials()) + self.mgr.encrypt_credentials("K", "S") + self.assertTrue(self.mgr.has_encrypted_credentials()) + + def test_metadata_written_on_encrypt(self): + self.mgr.encrypt_credentials("K", "S", rotation_interval_days=30) + meta = self.mgr._load_metadata() + self.assertIsNotNone(meta) + self.assertEqual(meta.rotation_interval_days, 30) + + def test_metadata_interval_updated_on_reencrypt(self): + """rotation_interval_days must stay consistent with rotation_due_at.""" + self.mgr.encrypt_credentials("K", "S", rotation_interval_days=30) + self.mgr.encrypt_credentials("K", "S", rotation_interval_days=60) + meta = self.mgr._load_metadata() + self.assertEqual(meta.rotation_interval_days, 60) + + def test_rotation_status_no_metadata(self): + status = self.mgr.get_rotation_status() + self.assertFalse(status["has_metadata"]) + + def test_rotation_status_with_metadata(self): + self.mgr.encrypt_credentials("K", "S", rotation_interval_days=90) + status = self.mgr.get_rotation_status() + self.assertTrue(status["has_metadata"]) + self.assertFalse(status["rotation_due"]) + self.assertGreater(status["days_until_rotation"], 0) + + def test_rotate_credentials(self): + self.mgr.encrypt_credentials("OLD_KEY", "OLD_SECRET") + result = self.mgr.rotate_credentials("NEW_KEY", "NEW_SECRET") + self.assertTrue(result) + creds = self.mgr.decrypt_credentials() + self.assertEqual(creds[0], "NEW_KEY") + self.assertEqual(creds[1], "NEW_SECRET") + + def test_rotate_cleans_up_backups(self): + self.mgr.encrypt_credentials("OLD", "OLD") + self.mgr.rotate_credentials("NEW", "NEW") + self.assertFalse(os.path.exists(self.mgr.encrypted_key_file + ".bak")) + self.assertFalse(os.path.exists(self.mgr.encrypted_secret_file + ".bak")) + self.assertFalse(os.path.exists(self.mgr.metadata_file + ".bak")) + + def test_rotate_restores_metadata_on_failure(self): + """Rotation rollback must restore metadata alongside ciphertext files.""" + self.mgr.encrypt_credentials("OLD_KEY", "OLD_SECRET", rotation_interval_days=90) + pre_meta = self.mgr._load_metadata() + self.assertIsNotNone(pre_meta) + + # Corrupt the manager to force failure during encrypt. Both ciphertexts + # are staged via _stage_temp_binary; fail the second so the key rename + # has already committed and rollback must restore the previous key + # ciphertext. + original = self.mgr._stage_temp_binary + call_count = [0] + + def failing_stage(path, data): + call_count[0] += 1 + if call_count[0] >= 2: # succeed on key, fail on secret + raise OSError("disk full") + return original(path, data) + + self.mgr._stage_temp_binary = failing_stage + result = self.mgr.rotate_credentials( + "NEW_KEY", "NEW_SECRET", rotation_interval_days=30 + ) + self.assertFalse(result) + # Ciphertext: old credentials still decryptable + creds = self.mgr.decrypt_credentials() + self.assertEqual(creds[0], "OLD_KEY") + # Metadata: rotation_interval_days + rotation_due_at unchanged + post_meta = self.mgr._load_metadata() + self.assertEqual( + post_meta.rotation_interval_days, pre_meta.rotation_interval_days + ) + self.assertEqual(post_meta.rotation_due_at, pre_meta.rotation_due_at) + self.assertEqual(post_meta.created_at, pre_meta.created_at) + + def test_no_rotation_warning_when_fresh(self): + self.mgr.encrypt_credentials("K", "S", rotation_interval_days=90) + self.assertIsNone(self.mgr.check_rotation_warning()) + + def test_rotation_warning_when_overdue(self): + meta = CredentialMetadata( + created_at=time.time() - 100 * 86400, + last_rotated_at=time.time() - 100 * 86400, + rotation_due_at=time.time() - 1, + ) + self.mgr._save_metadata(meta) + warning = self.mgr.check_rotation_warning() + self.assertIsNotNone(warning) + self.assertIn("OVERDUE", warning) + + def test_rotation_warning_when_near_due(self): + meta = CredentialMetadata( + created_at=time.time(), + last_rotated_at=time.time(), + rotation_due_at=time.time() + 3 * 86400, + ) + self.mgr._save_metadata(meta) + warning = self.mgr.check_rotation_warning() + self.assertIsNotNone(warning) + self.assertIn("day", warning) + + def test_legacy_vault_auto_migrates_to_new_derivation(self): + """ + Vaults encrypted under the legacy COMPUTERNAME/USERNAME derivation + must decrypt transparently AND be rewritten under the new + gethostname()/getuser() derivation on the same call, so the legacy + fallback path is one-shot per vault. + """ + legacy_pw = "legacy_pw_fixed_for_test_0000000000000000" + new_pw = "new_pw_fixed_for_test_xxxxxxxxxxxxxxxxxxx" + + # Step 1: encrypt as if the running version were still the legacy build. + with patch.object(self.mgr, "_get_machine_password", return_value=legacy_pw): + self.assertTrue(self.mgr.encrypt_credentials("KEY_LEG", "SECRET_LEG")) + + # Sanity: cannot decrypt under new derivation alone (no legacy fallback). + with patch.object( + self.mgr, "_get_machine_password", return_value=new_pw + ), patch.object(self.mgr, "_get_legacy_machine_password", return_value=None): + self.assertIsNone(self.mgr.decrypt_credentials()) + + # Step 2: simulate the upgraded build — primary derivation is new_pw, + # legacy fallback exposes the same legacy_pw used in step 1. + with patch.object( + self.mgr, "_get_machine_password", return_value=new_pw + ), patch.object( + self.mgr, "_get_legacy_machine_password", return_value=legacy_pw + ): + creds = self.mgr.decrypt_credentials() + self.assertEqual(creds, ("KEY_LEG", "SECRET_LEG")) + + # Step 3: vault was auto-rewritten under new_pw. Now decrypt with + # ONLY the new derivation available — legacy fallback returns None — + # and it must still succeed. Proves the rewrite happened. + with patch.object( + self.mgr, "_get_machine_password", return_value=new_pw + ), patch.object(self.mgr, "_get_legacy_machine_password", return_value=None): + creds = self.mgr.decrypt_credentials() + self.assertEqual(creds, ("KEY_LEG", "SECRET_LEG")) + + def test_legacy_migration_preserves_rotation_metadata(self): + """Derivation migration must NOT reset last_rotated_at / + rotation_due_at. Without this, a one-shot derivation upgrade would + masquerade as a real rotation and silently push the next rotation + warning out by a full interval — defeating the rotation scheduler + for any vault that triggers the legacy fallback.""" + legacy_pw = "legacy_pw_fixed_for_test_0000000000000000" + new_pw = "new_pw_fixed_for_test_xxxxxxxxxxxxxxxxxxx" + + # Encrypt under legacy derivation. + with patch.object(self.mgr, "_get_machine_password", return_value=legacy_pw): + self.assertTrue(self.mgr.encrypt_credentials("KEY_LEG", "SECRET_LEG")) + + # Backdate metadata so we can detect any reset. + original = self.mgr._load_metadata() + self.assertIsNotNone(original) + backdated_last = time.time() - 30 * 86400 # rotated 30d ago + backdated_due = time.time() + 60 * 86400 # next due in 60d + original.last_rotated_at = backdated_last + original.rotation_due_at = backdated_due + self.mgr._save_metadata(original) + + # Trigger derivation migration via decrypt path. + with patch.object( + self.mgr, "_get_machine_password", return_value=new_pw + ), patch.object( + self.mgr, "_get_legacy_machine_password", return_value=legacy_pw + ): + self.assertEqual(self.mgr.decrypt_credentials(), ("KEY_LEG", "SECRET_LEG")) + + # Rotation timestamps must be unchanged after the migration. + migrated = self.mgr._load_metadata() + self.assertIsNotNone(migrated) + self.assertAlmostEqual(migrated.last_rotated_at, backdated_last, places=1) + self.assertAlmostEqual(migrated.rotation_due_at, backdated_due, places=1) + + def test_migrate_from_plaintext(self): + with open(os.path.join(self.tmpdir, "r_key.txt"), "w") as f: + f.write("PLAIN_KEY\n") + with open(os.path.join(self.tmpdir, "r_secret.txt"), "w") as f: + f.write("PLAIN_SECRET\n") + result = self.mgr.migrate_from_plaintext() + self.assertTrue(result) + self.assertFalse(os.path.exists(os.path.join(self.tmpdir, "r_key.txt"))) + creds = self.mgr.decrypt_credentials() + self.assertEqual(creds[0], "PLAIN_KEY") + + def test_cross_platform_machine_password(self): + """machine password should be non-empty on all platforms.""" + pwd = self.mgr._get_machine_password() + self.assertIsInstance(pwd, str) + self.assertGreater(len(pwd), 0) + + +class TestPermissionValidator(unittest.TestCase): + def setUp(self): + self.tmpdir = tempfile.mkdtemp() + self.validator = PermissionValidator(self.tmpdir) + + def tearDown(self): + shutil.rmtree(self.tmpdir, ignore_errors=True) + + def test_no_fetcher_returns_failed_audit(self): + result = self.validator.validate(None) + self.assertFalse(result.audit_passed) + self.assertFalse(result.has_required) + + def test_all_permissions_granted(self): + result = self.validator.validate( + lambda: ["read_account", "read_positions", "buy", "sell"], + require_trading=True, + ) + self.assertTrue(result.audit_passed) + self.assertTrue(result.has_required) + self.assertTrue(result.has_trading) + + def test_missing_required_permissions(self): + result = self.validator.validate( + lambda: ["read_account"], # missing read_positions + require_trading=False, + ) + self.assertFalse(result.audit_passed) + self.assertIn("read_positions", result.missing_required) + + def test_missing_trading_permissions(self): + result = self.validator.validate( + lambda: ["read_account", "read_positions"], + require_trading=True, + ) + self.assertFalse(result.audit_passed) + self.assertFalse(result.has_trading) + + def test_fetcher_exception_handled(self): + def fetcher(): + raise ConnectionError("API unreachable") + + result = self.validator.validate(fetcher) + self.assertFalse(result.audit_passed) + self.assertIn("failed", result.message.lower()) + + def test_audit_log_written_and_secured(self): + self.validator.validate(None) + log_path = os.path.join(self.tmpdir, PermissionValidator.AUDIT_LOG_FILE) + self.assertTrue(os.path.exists(log_path)) + with open(log_path) as f: + entry = json.loads(f.readline()) + self.assertIn("audit_passed", entry) + # Permission bits: 0600 (user rw only). POSIX-only — chmod is a no-op + # on Windows so st_mode reflects ACLs, not unix bits. + if os.name == "posix": + mode = stat.S_IMODE(os.stat(log_path).st_mode) + self.assertEqual(mode, stat.S_IRUSR | stat.S_IWUSR) + + def test_audit_history_returned(self): + self.validator.validate(None) + history = self.validator.get_audit_history() + self.assertGreater(len(history), 0) + + def test_audit_log_size_cap(self): + """Log should not grow past MAX_AUDIT_LINES.""" + # Write many entries manually to reach cap + log_path = os.path.join(self.tmpdir, PermissionValidator.AUDIT_LOG_FILE) + entry = json.dumps( + { + "audit_passed": False, + "timestamp": 0, + "has_required": False, + "has_trading": False, + "granted_permissions": [], + "missing_required": [], + "missing_trading": [], + "message": "x", + } + ) + with open(log_path, "w") as f: + for _ in range(PermissionValidator.MAX_AUDIT_LINES + 5): + f.write(entry + "\n") + # Trigger a new write, which should trim the file + self.validator.validate(None) + with open(log_path) as f: + lines = f.readlines() + self.assertLessEqual(len(lines), PermissionValidator.MAX_AUDIT_LINES) + + +class TestCredentialRotationScheduler(unittest.TestCase): + def setUp(self): + self.tmpdir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.tmpdir, ignore_errors=True) + + def test_start_stop(self): + cb = MagicMock() + sched = CredentialRotationScheduler( + cb, check_interval_hours=24, base_dir=self.tmpdir + ) + sched.start() + self.assertTrue(sched._thread.is_alive()) + sched.stop() + self.assertFalse(sched._thread.is_alive()) + + def test_no_callback_without_metadata(self): + cb = MagicMock() + sched = CredentialRotationScheduler( + cb, check_interval_hours=24, base_dir=self.tmpdir + ) + result = sched.check_now() + self.assertIsNone(result) + cb.assert_not_called() + + def test_callback_fires_when_overdue(self): + """Scheduler must invoke the user callback via _tick() when overdue + metadata is seeded. Exercises the real callback dispatch path + (_tick → _run → cb) rather than only the synchronous check_now() + helper, so a regression that broke the tick→callback wiring would + be caught here.""" + cb = MagicMock() + mgr = SecureCredentialManager(self.tmpdir) + # Seed overdue metadata + meta = CredentialMetadata( + created_at=time.time() - 200 * 86400, + last_rotated_at=time.time() - 200 * 86400, + rotation_due_at=time.time() - 1, + ) + mgr._save_metadata(meta) + + sched = CredentialRotationScheduler( + cb, check_interval_hours=24, base_dir=self.tmpdir + ) + + # check_now() returns the warning string for direct callers. + warning = sched.check_now() + self.assertIsNotNone(warning) + self.assertIn("OVERDUE", warning) + + # _tick() is the scheduler's real dispatch path; it must actually + # call the callback with the same overdue warning. + sched._tick() + cb.assert_called_once() + (delivered,), _ = cb.call_args + self.assertIn("OVERDUE", delivered) + + def test_dedup_callback_not_repeated(self): + """Same warning should not trigger callback twice. Exercises the + real _tick() path so a regression in dedup logic would fail this test.""" + cb = MagicMock() + mgr = SecureCredentialManager(self.tmpdir) + meta = CredentialMetadata( + created_at=time.time() - 200 * 86400, + last_rotated_at=time.time() - 200 * 86400, + rotation_due_at=time.time() - 1, + ) + mgr._save_metadata(meta) + sched = CredentialRotationScheduler( + cb, check_interval_hours=24, base_dir=self.tmpdir + ) + + sched._tick() # first tick — should fire + sched._tick() # second tick — same warning, dedup'd + + cb.assert_called_once() + + def test_tick_fires_on_warning_change(self): + """When warning text changes, callback fires again.""" + cb = MagicMock() + mgr = SecureCredentialManager(self.tmpdir) + sched = CredentialRotationScheduler( + cb, check_interval_hours=24, base_dir=self.tmpdir + ) + + # Seed overdue → tick should fire + mgr._save_metadata( + CredentialMetadata( + created_at=time.time() - 200 * 86400, + last_rotated_at=time.time() - 200 * 86400, + rotation_due_at=time.time() - 1, + ) + ) + sched._tick() + self.assertEqual(cb.call_count, 1) + + # Replace with near-due → different message, callback fires again + mgr._save_metadata( + CredentialMetadata( + created_at=time.time(), + last_rotated_at=time.time(), + rotation_due_at=time.time() + 3 * 86400, + ) + ) + sched._tick() + self.assertEqual(cb.call_count, 2) + + +if __name__ == "__main__": + unittest.main()