From 720eb9bdd89f20900366e005e62555fe9009bde9 Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Sun, 17 May 2026 23:49:16 +0300 Subject: [PATCH 1/7] feat: Add production security and audit logging with correlation ID tracking pt_security_logger.py: - SecurityLogger with RotatingFileHandler (10MB/10 backups), secure permissions, JSONL structured events - SecurityEventType enum: auth, credential, suspicious, trade, system events - correlation_context() thread-local context manager propagates request IDs - CorrelationLogFilter injects correlation_id into all LogRecords - Module-level security_logger singleton for application-wide use - 25 unit tests, all passing --- app/pt_security_logger.py | 369 ++++++++++++++++++++++++++++++++++++ app/test_security_logger.py | 227 ++++++++++++++++++++++ 2 files changed, 596 insertions(+) create mode 100644 app/pt_security_logger.py create mode 100644 app/test_security_logger.py diff --git a/app/pt_security_logger.py b/app/pt_security_logger.py new file mode 100644 index 00000000..8fe6267e --- /dev/null +++ b/app/pt_security_logger.py @@ -0,0 +1,369 @@ +""" +PowerTraderAI+ Security & Audit Logging +Extends pt_logging_system with: +- Correlation ID context (thread-local, propagated through request chains) +- Security event logging: API auth attempts, credential usage, suspicious activity +- Dedicated audit log (separate file, rotation, secure permissions) +- Structured JSON security events for SIEM integration + +Usage: + from pt_security_logger import security_logger, correlation_context + + # Propagate correlation ID through a trading workflow + with correlation_context("trade-abc-123"): + security_logger.log_auth_attempt("robinhood", success=True) + security_logger.log_trade_event("BTC-USD", "buy", 0.1, 45000.0) + + # Standalone + security_logger.log_suspicious_activity( + "rate_limit_exceeded", source_ip="10.0.0.1", details={"endpoint": "/orders"} + ) +""" + +import json +import logging +import logging.handlers +import os +import stat +import threading +import time +import uuid +from dataclasses import dataclass, asdict +from datetime import datetime +from enum import Enum +from typing import Any, Dict, Generator, Optional + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Security event types +# --------------------------------------------------------------------------- +class SecurityEventType(Enum): + AUTH_ATTEMPT = "auth_attempt" # API key authentication attempt + AUTH_SUCCESS = "auth_success" # Successful authentication + AUTH_FAILURE = "auth_failure" # Failed authentication + CREDENTIAL_USE = "credential_use" # Credential accessed/used + CREDENTIAL_ROTATION = "credential_rotation" # Credential rotated + SUSPICIOUS_ACTIVITY = "suspicious_activity" # Anomalous behavior detected + PERMISSION_DENIED = "permission_denied" # Insufficient API permissions + RATE_LIMIT = "rate_limit" # Rate limit hit + TRADE_EXECUTED = "trade_executed" # Order placed + TRADE_REJECTED = "trade_rejected" # Order rejected + CONFIG_CHANGE = "config_change" # Configuration modified + SYSTEM_START = "system_start" # Application started + SYSTEM_STOP = "system_stop" # Application stopped + + +@dataclass +class SecurityEvent: + """Structured security event for audit trail.""" + event_id: str + event_type: str + timestamp: float + message: str + correlation_id: Optional[str] + source: Optional[str] = None + user_id: Optional[str] = None + source_ip: Optional[str] = None + success: Optional[bool] = None + details: Optional[Dict[str, Any]] = None + + def to_dict(self) -> dict: + return asdict(self) + + def to_json(self) -> str: + return json.dumps(self.to_dict(), default=str) + + @staticmethod + def new_id() -> str: + return f"SEC_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}" + + +# --------------------------------------------------------------------------- +# Correlation ID context (thread-local) +# --------------------------------------------------------------------------- +_correlation_local = threading.local() + + +def get_correlation_id() -> Optional[str]: + """Return the current thread's correlation ID, or None if not set.""" + return getattr(_correlation_local, "correlation_id", None) + + +def set_correlation_id(cid: str) -> None: + """Set the correlation ID for the current thread.""" + _correlation_local.correlation_id = cid + + +def clear_correlation_id() -> None: + """Clear the correlation ID for the current thread.""" + _correlation_local.correlation_id = None + + +class correlation_context: + """ + Context manager that sets a correlation ID for the current thread, + then restores the previous value on exit. + + Usage: + with correlation_context("trade-workflow-xyz"): + process_order() # all logs within will carry this correlation ID + """ + + def __init__(self, correlation_id: Optional[str] = None): + self._cid = correlation_id or f"CID_{uuid.uuid4().hex[:12]}" + self._previous: Optional[str] = None + + def __enter__(self) -> str: + self._previous = get_correlation_id() + set_correlation_id(self._cid) + return self._cid + + def __exit__(self, *_) -> None: + if self._previous is not None: + set_correlation_id(self._previous) + else: + clear_correlation_id() + + +# --------------------------------------------------------------------------- +# CorrelationLogFilter — injects correlation_id into log records +# --------------------------------------------------------------------------- +class CorrelationLogFilter(logging.Filter): + """ + Logging filter that injects the current thread's correlation ID + into every LogRecord. Works with any handler/formatter. + """ + + def filter(self, record: logging.LogRecord) -> bool: + record.correlation_id = get_correlation_id() or "" + return True + + +# --------------------------------------------------------------------------- +# SecurityLogger +# --------------------------------------------------------------------------- +class SecurityLogger: + """ + Application-wide security and audit logger. + + Writes structured JSON security events to a dedicated audit log file + (separate from the main application log). The audit log uses a + RotatingFileHandler with secure (owner-only) file permissions. + """ + + AUDIT_LOG_FILENAME = "security_audit.jsonl" + MAX_BYTES = 10 * 1024 * 1024 # 10 MB per file + BACKUP_COUNT = 10 # Keep 10 rotated files + + def __init__(self, log_dir: Optional[str] = None): + self._log_dir = log_dir or os.path.dirname(os.path.abspath(__file__)) + self._audit_path = os.path.join(self._log_dir, self.AUDIT_LOG_FILENAME) + self._lock = threading.Lock() + self._handler: Optional[logging.handlers.RotatingFileHandler] = None + self._logger = self._setup_audit_logger() + + def _setup_audit_logger(self) -> logging.Logger: + """Configure dedicated audit logger with rotation and secure permissions.""" + import hashlib + path_hash = hashlib.md5(self._audit_path.encode()).hexdigest()[:8] + audit_logger = logging.getLogger(f"pt.security.audit.{path_hash}") + audit_logger.setLevel(logging.DEBUG) + audit_logger.propagate = False # Don't propagate to root logger + + if not audit_logger.handlers: + handler = logging.handlers.RotatingFileHandler( + self._audit_path, + maxBytes=self.MAX_BYTES, + backupCount=self.BACKUP_COUNT, + encoding="utf-8", + ) + handler.setFormatter(logging.Formatter("%(message)s")) # Raw JSON + handler.addFilter(CorrelationLogFilter()) + audit_logger.addHandler(handler) + self._handler = handler + self._secure_audit_file() + + return audit_logger + + def _secure_audit_file(self) -> None: + """Set owner-only permissions on audit log file.""" + try: + if os.path.exists(self._audit_path): + os.chmod(self._audit_path, stat.S_IRUSR | stat.S_IWUSR) + except (OSError, AttributeError): + pass + + def _emit(self, event: SecurityEvent) -> None: + """Write security event to audit log (thread-safe).""" + with self._lock: + try: + self._logger.info(event.to_json()) + self._secure_audit_file() + except Exception as exc: + logger.error("Failed to write security audit event: %s", exc) + + def _make_event( + self, + event_type: SecurityEventType, + message: str, + source: Optional[str] = None, + success: Optional[bool] = None, + details: Optional[Dict[str, Any]] = None, + source_ip: Optional[str] = None, + user_id: Optional[str] = None, + ) -> SecurityEvent: + return SecurityEvent( + event_id=SecurityEvent.new_id(), + event_type=event_type.value, + timestamp=time.time(), + message=message, + correlation_id=get_correlation_id(), + source=source, + success=success, + details=details, + source_ip=source_ip, + user_id=user_id, + ) + + # ------------------------------------------------------------------ + # Public logging methods + # ------------------------------------------------------------------ + def log_auth_attempt( + self, + api_name: str, + success: bool, + user_id: Optional[str] = None, + source_ip: Optional[str] = None, + details: Optional[Dict[str, Any]] = None, + ) -> None: + """Log an API authentication attempt (success or failure).""" + event_type = SecurityEventType.AUTH_SUCCESS if success else SecurityEventType.AUTH_FAILURE + msg = f"API auth {'succeeded' if success else 'FAILED'} for {api_name}" + if not success: + logger.warning("SECURITY: %s", msg) + event = self._make_event(event_type, msg, source=api_name, + success=success, details=details, + source_ip=source_ip, user_id=user_id) + self._emit(event) + + def log_credential_use( + self, + api_name: str, + operation: str, + details: Optional[Dict[str, Any]] = None, + ) -> None: + """Log that credentials were accessed/used for an operation.""" + msg = f"Credential used: {api_name} for {operation}" + event = self._make_event( + SecurityEventType.CREDENTIAL_USE, msg, + source=api_name, success=True, + details={**(details or {}), "operation": operation}, + ) + self._emit(event) + + def log_credential_rotation( + self, + api_name: str, + success: bool, + details: Optional[Dict[str, Any]] = None, + ) -> None: + """Log a credential rotation event.""" + msg = f"Credential rotation {'succeeded' if success else 'FAILED'} for {api_name}" + if not success: + logger.critical("SECURITY: %s", msg) + event = self._make_event( + SecurityEventType.CREDENTIAL_ROTATION, msg, + source=api_name, success=success, details=details, + ) + self._emit(event) + + def log_suspicious_activity( + self, + activity_type: str, + source_ip: Optional[str] = None, + user_id: Optional[str] = None, + details: Optional[Dict[str, Any]] = None, + ) -> None: + """Log detected suspicious activity.""" + msg = f"Suspicious activity detected: {activity_type}" + logger.warning("SECURITY ALERT: %s", msg) + event = self._make_event( + SecurityEventType.SUSPICIOUS_ACTIVITY, msg, + source=activity_type, success=False, + source_ip=source_ip, user_id=user_id, details=details, + ) + self._emit(event) + + def log_permission_denied( + self, + api_name: str, + required_permission: str, + details: Optional[Dict[str, Any]] = None, + ) -> None: + """Log an API permission denial.""" + msg = f"Permission denied: {required_permission} on {api_name}" + logger.error("SECURITY: %s", msg) + event = self._make_event( + SecurityEventType.PERMISSION_DENIED, msg, + source=api_name, success=False, + details={**(details or {}), "required_permission": required_permission}, + ) + self._emit(event) + + def log_trade_event( + self, + symbol: str, + side: str, + quantity: float, + price: float, + success: bool = True, + order_id: Optional[str] = None, + details: Optional[Dict[str, Any]] = None, + ) -> None: + """Log a trade execution event to the audit trail.""" + event_type = SecurityEventType.TRADE_EXECUTED if success else SecurityEventType.TRADE_REJECTED + msg = f"Trade {'executed' if success else 'REJECTED'}: {side} {quantity} {symbol} @ {price}" + event = self._make_event( + event_type, msg, success=success, + details={ + "symbol": symbol, "side": side, + "quantity": quantity, "price": price, + "order_id": order_id, **(details or {}), + }, + ) + self._emit(event) + + def log_system_event( + self, event_type: SecurityEventType, message: str, + details: Optional[Dict[str, Any]] = None, + ) -> None: + """Log a system lifecycle event (start, stop, config change).""" + event = self._make_event(event_type, message, details=details) + self._emit(event) + + def get_recent_events(self, limit: int = 100) -> list: + """Read the last `limit` events from the audit log.""" + if not os.path.exists(self._audit_path): + return [] + try: + with open(self._audit_path, "r", encoding="utf-8") as f: + lines = f.readlines() + events = [] + for line in lines[-limit:]: + line = line.strip() + if line: + try: + events.append(json.loads(line)) + except json.JSONDecodeError: + pass + return events + except OSError: + return [] + + +# --------------------------------------------------------------------------- +# Module-level singleton +# --------------------------------------------------------------------------- +security_logger = SecurityLogger() diff --git a/app/test_security_logger.py b/app/test_security_logger.py new file mode 100644 index 00000000..84e57f5f --- /dev/null +++ b/app/test_security_logger.py @@ -0,0 +1,227 @@ +"""Tests for pt_security_logger - issue #55.""" + +import json +import os +import tempfile +import threading +import time +import unittest + +from pt_security_logger import ( + CorrelationLogFilter, + SecurityEvent, + SecurityEventType, + SecurityLogger, + clear_correlation_id, + correlation_context, + get_correlation_id, + set_correlation_id, +) + + +class TestCorrelationID(unittest.TestCase): + + def tearDown(self): + clear_correlation_id() + + def test_default_is_none(self): + clear_correlation_id() + self.assertIsNone(get_correlation_id()) + + def test_set_and_get(self): + set_correlation_id("test-cid-123") + self.assertEqual(get_correlation_id(), "test-cid-123") + + def test_context_manager_sets_id(self): + with correlation_context("ctx-abc") as cid: + self.assertEqual(cid, "ctx-abc") + self.assertEqual(get_correlation_id(), "ctx-abc") + + def test_context_manager_restores_previous(self): + set_correlation_id("outer") + with correlation_context("inner"): + self.assertEqual(get_correlation_id(), "inner") + self.assertEqual(get_correlation_id(), "outer") + + def test_context_manager_clears_when_no_previous(self): + clear_correlation_id() + with correlation_context("temp"): + pass + self.assertIsNone(get_correlation_id()) + + def test_context_manager_auto_generates_id(self): + with correlation_context() as cid: + self.assertIsNotNone(cid) + self.assertTrue(cid.startswith("CID_")) + + def test_thread_local_isolation(self): + """Each thread has its own correlation ID.""" + results = {} + + def worker(name, cid): + set_correlation_id(cid) + time.sleep(0.01) + results[name] = get_correlation_id() + + t1 = threading.Thread(target=worker, args=("t1", "cid-thread-1")) + t2 = threading.Thread(target=worker, args=("t2", "cid-thread-2")) + t1.start(); t2.start() + t1.join(); t2.join() + self.assertEqual(results["t1"], "cid-thread-1") + self.assertEqual(results["t2"], "cid-thread-2") + + +class TestSecurityEvent(unittest.TestCase): + + def test_new_id_format(self): + eid = SecurityEvent.new_id() + self.assertTrue(eid.startswith("SEC_")) + self.assertGreater(len(eid), 10) + + def test_to_dict_has_required_fields(self): + event = SecurityEvent( + event_id="SEC_001", + event_type=SecurityEventType.AUTH_ATTEMPT.value, + timestamp=time.time(), + message="test", + correlation_id=None, + ) + d = event.to_dict() + self.assertIn("event_id", d) + self.assertIn("event_type", d) + self.assertIn("timestamp", d) + self.assertIn("message", d) + + def test_to_json_valid(self): + event = SecurityEvent( + event_id="SEC_002", + event_type=SecurityEventType.AUTH_SUCCESS.value, + timestamp=time.time(), + message="auth ok", + correlation_id="cid-test", + ) + parsed = json.loads(event.to_json()) + self.assertEqual(parsed["correlation_id"], "cid-test") + + +class TestSecurityLogger(unittest.TestCase): + + def setUp(self): + self.tmpdir = tempfile.mkdtemp() + self.sec_logger = SecurityLogger(log_dir=self.tmpdir) + clear_correlation_id() + + def tearDown(self): + import shutil + clear_correlation_id() + # Close handlers to release file locks (Windows) + for handler in self.sec_logger._logger.handlers[:]: + handler.close() + self.sec_logger._logger.removeHandler(handler) + shutil.rmtree(self.tmpdir, ignore_errors=True) + + def _get_events(self): + return self.sec_logger.get_recent_events() + + def test_log_auth_attempt_success(self): + self.sec_logger.log_auth_attempt("robinhood", success=True) + events = self._get_events() + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["event_type"], SecurityEventType.AUTH_SUCCESS.value) + self.assertTrue(events[0]["success"]) + + def test_log_auth_attempt_failure(self): + self.sec_logger.log_auth_attempt("robinhood", success=False) + events = self._get_events() + self.assertEqual(events[0]["event_type"], SecurityEventType.AUTH_FAILURE.value) + self.assertFalse(events[0]["success"]) + + def test_log_credential_use(self): + self.sec_logger.log_credential_use("robinhood", "place_order") + events = self._get_events() + self.assertEqual(events[0]["event_type"], SecurityEventType.CREDENTIAL_USE.value) + self.assertEqual(events[0]["details"]["operation"], "place_order") + + def test_log_credential_rotation(self): + self.sec_logger.log_credential_rotation("robinhood", success=True) + events = self._get_events() + self.assertEqual(events[0]["event_type"], SecurityEventType.CREDENTIAL_ROTATION.value) + + def test_log_suspicious_activity(self): + self.sec_logger.log_suspicious_activity( + "rate_limit_exceeded", source_ip="1.2.3.4", + details={"endpoint": "/orders", "count": 100} + ) + events = self._get_events() + self.assertEqual(events[0]["event_type"], SecurityEventType.SUSPICIOUS_ACTIVITY.value) + self.assertEqual(events[0]["source_ip"], "1.2.3.4") + + def test_log_permission_denied(self): + self.sec_logger.log_permission_denied("robinhood", "sell") + events = self._get_events() + self.assertEqual(events[0]["event_type"], SecurityEventType.PERMISSION_DENIED.value) + self.assertEqual(events[0]["details"]["required_permission"], "sell") + + def test_log_trade_event_success(self): + self.sec_logger.log_trade_event("BTC-USD", "buy", 0.1, 45000.0, order_id="ord-001") + events = self._get_events() + self.assertEqual(events[0]["event_type"], SecurityEventType.TRADE_EXECUTED.value) + self.assertEqual(events[0]["details"]["symbol"], "BTC-USD") + + def test_log_trade_event_rejected(self): + self.sec_logger.log_trade_event("ETH-USD", "sell", 1.0, 3000.0, success=False) + events = self._get_events() + self.assertEqual(events[0]["event_type"], SecurityEventType.TRADE_REJECTED.value) + + def test_correlation_id_captured_in_event(self): + with correlation_context("trade-workflow-xyz"): + self.sec_logger.log_auth_attempt("robinhood", success=True) + events = self._get_events() + self.assertEqual(events[0]["correlation_id"], "trade-workflow-xyz") + + def test_correlation_id_none_outside_context(self): + clear_correlation_id() + self.sec_logger.log_auth_attempt("robinhood", success=True) + events = self._get_events() + self.assertIsNone(events[0]["correlation_id"]) + + def test_multiple_events_ordered(self): + self.sec_logger.log_auth_attempt("robinhood", success=True) + self.sec_logger.log_credential_use("robinhood", "fetch_positions") + self.sec_logger.log_trade_event("BTC-USD", "buy", 0.5, 44000.0) + events = self._get_events() + self.assertEqual(len(events), 3) + timestamps = [e["timestamp"] for e in events] + self.assertEqual(timestamps, sorted(timestamps)) + + def test_audit_file_created(self): + self.sec_logger.log_auth_attempt("robinhood", success=True) + self.assertTrue(os.path.exists(self.sec_logger._audit_path)) + + def test_get_recent_events_empty_when_no_log(self): + # Logger on a path with no events yet + tmpdir2 = tempfile.mkdtemp() + sl = SecurityLogger(log_dir=tmpdir2) + for h in sl._logger.handlers[:]: + h.close(); sl._logger.removeHandler(h) + import shutil + shutil.rmtree(tmpdir2, ignore_errors=True) + + def test_get_recent_events_limit(self): + for i in range(10): + self.sec_logger.log_auth_attempt("api", success=True) + events = self.sec_logger.get_recent_events(limit=5) + self.assertEqual(len(events), 5) + + def test_log_system_event(self): + self.sec_logger.log_system_event( + SecurityEventType.SYSTEM_START, + "PowerTraderAI started", + details={"version": "1.0"}, + ) + events = self._get_events() + self.assertEqual(events[0]["event_type"], SecurityEventType.SYSTEM_START.value) + + +if __name__ == "__main__": + unittest.main() From 1d62f86fce6101cb756890d956074c5a1a10c830 Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Mon, 18 May 2026 04:00:31 +0300 Subject: [PATCH 2/7] fix: Black formatting and remove unused imports (flake8) --- app/pt_security_logger.py | 104 ++++++++++++++++++++++++------------ app/test_security_logger.py | 48 +++++++++++------ 2 files changed, 102 insertions(+), 50 deletions(-) diff --git a/app/pt_security_logger.py b/app/pt_security_logger.py index 8fe6267e..d4a91ff6 100644 --- a/app/pt_security_logger.py +++ b/app/pt_security_logger.py @@ -31,7 +31,7 @@ from dataclasses import dataclass, asdict from datetime import datetime from enum import Enum -from typing import Any, Dict, Generator, Optional +from typing import Any, Dict, Optional logger = logging.getLogger(__name__) @@ -40,24 +40,25 @@ # Security event types # --------------------------------------------------------------------------- class SecurityEventType(Enum): - AUTH_ATTEMPT = "auth_attempt" # API key authentication attempt - AUTH_SUCCESS = "auth_success" # Successful authentication - AUTH_FAILURE = "auth_failure" # Failed authentication - CREDENTIAL_USE = "credential_use" # Credential accessed/used + AUTH_ATTEMPT = "auth_attempt" # API key authentication attempt + AUTH_SUCCESS = "auth_success" # Successful authentication + AUTH_FAILURE = "auth_failure" # Failed authentication + CREDENTIAL_USE = "credential_use" # Credential accessed/used CREDENTIAL_ROTATION = "credential_rotation" # Credential rotated SUSPICIOUS_ACTIVITY = "suspicious_activity" # Anomalous behavior detected - PERMISSION_DENIED = "permission_denied" # Insufficient API permissions - RATE_LIMIT = "rate_limit" # Rate limit hit - TRADE_EXECUTED = "trade_executed" # Order placed - TRADE_REJECTED = "trade_rejected" # Order rejected - CONFIG_CHANGE = "config_change" # Configuration modified - SYSTEM_START = "system_start" # Application started - SYSTEM_STOP = "system_stop" # Application stopped + PERMISSION_DENIED = "permission_denied" # Insufficient API permissions + RATE_LIMIT = "rate_limit" # Rate limit hit + TRADE_EXECUTED = "trade_executed" # Order placed + TRADE_REJECTED = "trade_rejected" # Order rejected + CONFIG_CHANGE = "config_change" # Configuration modified + SYSTEM_START = "system_start" # Application started + SYSTEM_STOP = "system_stop" # Application stopped @dataclass class SecurityEvent: """Structured security event for audit trail.""" + event_id: str event_type: str timestamp: float @@ -154,8 +155,8 @@ class SecurityLogger: """ AUDIT_LOG_FILENAME = "security_audit.jsonl" - MAX_BYTES = 10 * 1024 * 1024 # 10 MB per file - BACKUP_COUNT = 10 # Keep 10 rotated files + MAX_BYTES = 10 * 1024 * 1024 # 10 MB per file + BACKUP_COUNT = 10 # Keep 10 rotated files def __init__(self, log_dir: Optional[str] = None): self._log_dir = log_dir or os.path.dirname(os.path.abspath(__file__)) @@ -167,6 +168,7 @@ def __init__(self, log_dir: Optional[str] = None): def _setup_audit_logger(self) -> logging.Logger: """Configure dedicated audit logger with rotation and secure permissions.""" import hashlib + path_hash = hashlib.md5(self._audit_path.encode()).hexdigest()[:8] audit_logger = logging.getLogger(f"pt.security.audit.{path_hash}") audit_logger.setLevel(logging.DEBUG) @@ -239,13 +241,23 @@ def log_auth_attempt( details: Optional[Dict[str, Any]] = None, ) -> None: """Log an API authentication attempt (success or failure).""" - event_type = SecurityEventType.AUTH_SUCCESS if success else SecurityEventType.AUTH_FAILURE + event_type = ( + SecurityEventType.AUTH_SUCCESS + if success + else SecurityEventType.AUTH_FAILURE + ) msg = f"API auth {'succeeded' if success else 'FAILED'} for {api_name}" if not success: logger.warning("SECURITY: %s", msg) - event = self._make_event(event_type, msg, source=api_name, - success=success, details=details, - source_ip=source_ip, user_id=user_id) + event = self._make_event( + event_type, + msg, + source=api_name, + success=success, + details=details, + source_ip=source_ip, + user_id=user_id, + ) self._emit(event) def log_credential_use( @@ -257,8 +269,10 @@ def log_credential_use( """Log that credentials were accessed/used for an operation.""" msg = f"Credential used: {api_name} for {operation}" event = self._make_event( - SecurityEventType.CREDENTIAL_USE, msg, - source=api_name, success=True, + SecurityEventType.CREDENTIAL_USE, + msg, + source=api_name, + success=True, details={**(details or {}), "operation": operation}, ) self._emit(event) @@ -270,12 +284,17 @@ def log_credential_rotation( details: Optional[Dict[str, Any]] = None, ) -> None: """Log a credential rotation event.""" - msg = f"Credential rotation {'succeeded' if success else 'FAILED'} for {api_name}" + msg = ( + f"Credential rotation {'succeeded' if success else 'FAILED'} for {api_name}" + ) if not success: logger.critical("SECURITY: %s", msg) event = self._make_event( - SecurityEventType.CREDENTIAL_ROTATION, msg, - source=api_name, success=success, details=details, + SecurityEventType.CREDENTIAL_ROTATION, + msg, + source=api_name, + success=success, + details=details, ) self._emit(event) @@ -290,9 +309,13 @@ def log_suspicious_activity( msg = f"Suspicious activity detected: {activity_type}" logger.warning("SECURITY ALERT: %s", msg) event = self._make_event( - SecurityEventType.SUSPICIOUS_ACTIVITY, msg, - source=activity_type, success=False, - source_ip=source_ip, user_id=user_id, details=details, + SecurityEventType.SUSPICIOUS_ACTIVITY, + msg, + source=activity_type, + success=False, + source_ip=source_ip, + user_id=user_id, + details=details, ) self._emit(event) @@ -306,8 +329,10 @@ def log_permission_denied( msg = f"Permission denied: {required_permission} on {api_name}" logger.error("SECURITY: %s", msg) event = self._make_event( - SecurityEventType.PERMISSION_DENIED, msg, - source=api_name, success=False, + SecurityEventType.PERMISSION_DENIED, + msg, + source=api_name, + success=False, details={**(details or {}), "required_permission": required_permission}, ) self._emit(event) @@ -323,20 +348,31 @@ def log_trade_event( details: Optional[Dict[str, Any]] = None, ) -> None: """Log a trade execution event to the audit trail.""" - event_type = SecurityEventType.TRADE_EXECUTED if success else SecurityEventType.TRADE_REJECTED + event_type = ( + SecurityEventType.TRADE_EXECUTED + if success + else SecurityEventType.TRADE_REJECTED + ) msg = f"Trade {'executed' if success else 'REJECTED'}: {side} {quantity} {symbol} @ {price}" event = self._make_event( - event_type, msg, success=success, + event_type, + msg, + success=success, details={ - "symbol": symbol, "side": side, - "quantity": quantity, "price": price, - "order_id": order_id, **(details or {}), + "symbol": symbol, + "side": side, + "quantity": quantity, + "price": price, + "order_id": order_id, + **(details or {}), }, ) self._emit(event) def log_system_event( - self, event_type: SecurityEventType, message: str, + self, + event_type: SecurityEventType, + message: str, details: Optional[Dict[str, Any]] = None, ) -> None: """Log a system lifecycle event (start, stop, config change).""" diff --git a/app/test_security_logger.py b/app/test_security_logger.py index 84e57f5f..f977d831 100644 --- a/app/test_security_logger.py +++ b/app/test_security_logger.py @@ -8,7 +8,6 @@ import unittest from pt_security_logger import ( - CorrelationLogFilter, SecurityEvent, SecurityEventType, SecurityLogger, @@ -20,7 +19,6 @@ class TestCorrelationID(unittest.TestCase): - def tearDown(self): clear_correlation_id() @@ -65,14 +63,15 @@ def worker(name, cid): t1 = threading.Thread(target=worker, args=("t1", "cid-thread-1")) t2 = threading.Thread(target=worker, args=("t2", "cid-thread-2")) - t1.start(); t2.start() - t1.join(); t2.join() + t1.start() + t2.start() + t1.join() + t2.join() self.assertEqual(results["t1"], "cid-thread-1") self.assertEqual(results["t2"], "cid-thread-2") class TestSecurityEvent(unittest.TestCase): - def test_new_id_format(self): eid = SecurityEvent.new_id() self.assertTrue(eid.startswith("SEC_")) @@ -105,7 +104,6 @@ def test_to_json_valid(self): class TestSecurityLogger(unittest.TestCase): - def setUp(self): self.tmpdir = tempfile.mkdtemp() self.sec_logger = SecurityLogger(log_dir=self.tmpdir) @@ -113,6 +111,7 @@ def setUp(self): def tearDown(self): import shutil + clear_correlation_id() # Close handlers to release file locks (Windows) for handler in self.sec_logger._logger.handlers[:]: @@ -139,39 +138,54 @@ def test_log_auth_attempt_failure(self): def test_log_credential_use(self): self.sec_logger.log_credential_use("robinhood", "place_order") events = self._get_events() - self.assertEqual(events[0]["event_type"], SecurityEventType.CREDENTIAL_USE.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.CREDENTIAL_USE.value + ) self.assertEqual(events[0]["details"]["operation"], "place_order") def test_log_credential_rotation(self): self.sec_logger.log_credential_rotation("robinhood", success=True) events = self._get_events() - self.assertEqual(events[0]["event_type"], SecurityEventType.CREDENTIAL_ROTATION.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.CREDENTIAL_ROTATION.value + ) def test_log_suspicious_activity(self): self.sec_logger.log_suspicious_activity( - "rate_limit_exceeded", source_ip="1.2.3.4", - details={"endpoint": "/orders", "count": 100} + "rate_limit_exceeded", + source_ip="1.2.3.4", + details={"endpoint": "/orders", "count": 100}, ) events = self._get_events() - self.assertEqual(events[0]["event_type"], SecurityEventType.SUSPICIOUS_ACTIVITY.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.SUSPICIOUS_ACTIVITY.value + ) self.assertEqual(events[0]["source_ip"], "1.2.3.4") def test_log_permission_denied(self): self.sec_logger.log_permission_denied("robinhood", "sell") events = self._get_events() - self.assertEqual(events[0]["event_type"], SecurityEventType.PERMISSION_DENIED.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.PERMISSION_DENIED.value + ) self.assertEqual(events[0]["details"]["required_permission"], "sell") def test_log_trade_event_success(self): - self.sec_logger.log_trade_event("BTC-USD", "buy", 0.1, 45000.0, order_id="ord-001") + self.sec_logger.log_trade_event( + "BTC-USD", "buy", 0.1, 45000.0, order_id="ord-001" + ) events = self._get_events() - self.assertEqual(events[0]["event_type"], SecurityEventType.TRADE_EXECUTED.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.TRADE_EXECUTED.value + ) self.assertEqual(events[0]["details"]["symbol"], "BTC-USD") def test_log_trade_event_rejected(self): self.sec_logger.log_trade_event("ETH-USD", "sell", 1.0, 3000.0, success=False) events = self._get_events() - self.assertEqual(events[0]["event_type"], SecurityEventType.TRADE_REJECTED.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.TRADE_REJECTED.value + ) def test_correlation_id_captured_in_event(self): with correlation_context("trade-workflow-xyz"): @@ -203,8 +217,10 @@ def test_get_recent_events_empty_when_no_log(self): tmpdir2 = tempfile.mkdtemp() sl = SecurityLogger(log_dir=tmpdir2) for h in sl._logger.handlers[:]: - h.close(); sl._logger.removeHandler(h) + h.close() + sl._logger.removeHandler(h) import shutil + shutil.rmtree(tmpdir2, ignore_errors=True) def test_get_recent_events_limit(self): From 76c882341207e82d118e41f78ec107ef43f2889f Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Mon, 18 May 2026 04:51:49 +0300 Subject: [PATCH 3/7] chore: Ignore audit log artifacts --- .gitignore | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.gitignore b/.gitignore index a5caa7d5..58d3904b 100644 --- a/.gitignore +++ b/.gitignore @@ -218,3 +218,5 @@ app/AVAX/ .DS_Store Thumbs.db desktop.ini +app/security_audit.jsonl +app/credential_audit.jsonl From bbfa7c17ffbf3e6b816887174cbab798a67ed44b Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Tue, 19 May 2026 11:53:40 +0300 Subject: [PATCH 4/7] fix: address all Copilot review comments on security logger MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pt_security_logger.py: - Module-level singleton is now lazy (get_security_logger()) — audit file not created at import time; default log_dir is ~/.powertraderai/logs, not the source tree; os.makedirs ensures dir exists before handler creation - Per-instance logging.Logger (not global registry) eliminates the shared-handler/_handler=None pitfall when two instances share a log_dir - _SecureRotatingFileHandler subclass overrides doRollover() to chmod active file and all backups after rotation - _chmod_secure() called once at handler creation, not on every _emit(); OSError logged as WARNING instead of silently swallowed - correlation_context renamed to CorrelationContext (PEP 8 CapWords); snake_case alias kept for backwards compatibility - CorrelationContext uses a _stack list so nested/reused instances are safe (no _previous corruption) - clear_correlation_id() uses del instead of setting to None - SecurityEvent.new_id() uses datetime.now(timezone.utc) for UTC IDs - get_recent_events() uses collections.deque(f, maxlen=limit) for O(limit) tail read instead of readlines() O(file size) - hashlib.md5 removed (per-instance logger needs no path hash) - Added log_rate_limit() and log_config_change() helpers for enum values that previously had no public method - Generator import removed (was unused) - Docstring updated: module does NOT extend pt_logging_system test_security_logger.py: - tearDown uses sec_logger.close() to release file handles - test_get_recent_events_empty_when_no_log: adds assertEqual([], ...) assertion - test_get_recent_events_limit: asserts returned events are the last N - Added test_new_id_uses_utc, test_context_manager_nested_safe, test_snake_case_alias_works, test_log_rate_limit, test_log_config_change --- app/pt_security_logger.py | 359 ++++++++++++++++++++++-------------- app/test_security_logger.py | 140 ++++++++------ 2 files changed, 299 insertions(+), 200 deletions(-) diff --git a/app/pt_security_logger.py b/app/pt_security_logger.py index d4a91ff6..87721ee5 100644 --- a/app/pt_security_logger.py +++ b/app/pt_security_logger.py @@ -1,25 +1,31 @@ """ PowerTraderAI+ Security & Audit Logging -Extends pt_logging_system with: +Standalone security event logging module providing: - Correlation ID context (thread-local, propagated through request chains) - Security event logging: API auth attempts, credential usage, suspicious activity - Dedicated audit log (separate file, rotation, secure permissions) - Structured JSON security events for SIEM integration +Note: This module does NOT depend on pt_logging_system. It manages its own +logging handler so security events are always written to a dedicated audit +file regardless of the application's root logger configuration. + Usage: - from pt_security_logger import security_logger, correlation_context + from pt_security_logger import get_security_logger, CorrelationContext # Propagate correlation ID through a trading workflow - with correlation_context("trade-abc-123"): - security_logger.log_auth_attempt("robinhood", success=True) - security_logger.log_trade_event("BTC-USD", "buy", 0.1, 45000.0) + with CorrelationContext("trade-abc-123"): + get_security_logger().log_auth_attempt("robinhood", success=True) + get_security_logger().log_trade_event("BTC-USD", "buy", 0.1, 45000.0) # Standalone - security_logger.log_suspicious_activity( + get_security_logger().log_suspicious_activity( "rate_limit_exceeded", source_ip="10.0.0.1", details={"endpoint": "/orders"} ) """ +import collections +import hashlib import json import logging import logging.handlers @@ -29,9 +35,9 @@ import time import uuid from dataclasses import dataclass, asdict -from datetime import datetime +from datetime import datetime, timezone from enum import Enum -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) @@ -40,19 +46,19 @@ # Security event types # --------------------------------------------------------------------------- class SecurityEventType(Enum): - AUTH_ATTEMPT = "auth_attempt" # API key authentication attempt - AUTH_SUCCESS = "auth_success" # Successful authentication - AUTH_FAILURE = "auth_failure" # Failed authentication - CREDENTIAL_USE = "credential_use" # Credential accessed/used - CREDENTIAL_ROTATION = "credential_rotation" # Credential rotated - SUSPICIOUS_ACTIVITY = "suspicious_activity" # Anomalous behavior detected - PERMISSION_DENIED = "permission_denied" # Insufficient API permissions - RATE_LIMIT = "rate_limit" # Rate limit hit - TRADE_EXECUTED = "trade_executed" # Order placed - TRADE_REJECTED = "trade_rejected" # Order rejected - CONFIG_CHANGE = "config_change" # Configuration modified - SYSTEM_START = "system_start" # Application started - SYSTEM_STOP = "system_stop" # Application stopped + AUTH_ATTEMPT = "auth_attempt" # API key authentication attempt + AUTH_SUCCESS = "auth_success" # Successful authentication + AUTH_FAILURE = "auth_failure" # Failed authentication + CREDENTIAL_USE = "credential_use" # Credential accessed/used + CREDENTIAL_ROTATION = "credential_rotation" # Credential rotated + SUSPICIOUS_ACTIVITY = "suspicious_activity" # Anomalous behavior detected + PERMISSION_DENIED = "permission_denied" # Insufficient API permissions + RATE_LIMIT = "rate_limit" # Rate limit hit + TRADE_EXECUTED = "trade_executed" # Order placed + TRADE_REJECTED = "trade_rejected" # Order rejected + CONFIG_CHANGE = "config_change" # Configuration modified + SYSTEM_START = "system_start" # Application started + SYSTEM_STOP = "system_stop" # Application stopped @dataclass @@ -78,7 +84,9 @@ def to_json(self) -> str: @staticmethod def new_id() -> str: - return f"SEC_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}" + # UTC timestamp in ID so audit records are timezone-unambiguous + ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") + return f"SEC_{ts}_{uuid.uuid4().hex[:8]}" # --------------------------------------------------------------------------- @@ -99,33 +107,45 @@ def set_correlation_id(cid: str) -> None: def clear_correlation_id() -> None: """Clear the correlation ID for the current thread.""" - _correlation_local.correlation_id = None + try: + del _correlation_local.correlation_id + except AttributeError: + pass -class correlation_context: +class CorrelationContext: """ Context manager that sets a correlation ID for the current thread, then restores the previous value on exit. + Each call to __enter__ saves the previous ID on a stack, so nested + and reused instances are safe. + Usage: - with correlation_context("trade-workflow-xyz"): + with CorrelationContext("trade-workflow-xyz"): process_order() # all logs within will carry this correlation ID """ def __init__(self, correlation_id: Optional[str] = None): self._cid = correlation_id or f"CID_{uuid.uuid4().hex[:12]}" - self._previous: Optional[str] = None + self._stack: List[Optional[str]] = [] def __enter__(self) -> str: - self._previous = get_correlation_id() + self._stack.append(get_correlation_id()) set_correlation_id(self._cid) return self._cid def __exit__(self, *_) -> None: - if self._previous is not None: - set_correlation_id(self._previous) - else: - clear_correlation_id() + if self._stack: + previous = self._stack.pop() + if previous is not None: + set_correlation_id(previous) + else: + clear_correlation_id() + + +# Keep snake_case alias for backwards compatibility +correlation_context = CorrelationContext # --------------------------------------------------------------------------- @@ -142,6 +162,40 @@ def filter(self, record: logging.LogRecord) -> bool: return True +# --------------------------------------------------------------------------- +# SecureRotatingFileHandler — chmod backup files after rotation +# --------------------------------------------------------------------------- +class _SecureRotatingFileHandler(logging.handlers.RotatingFileHandler): + """RotatingFileHandler subclass that applies owner-only permissions after rollover.""" + + def doRollover(self) -> None: + super().doRollover() + # Re-secure the new active file and all existing backups + for path in self._audit_paths(): + _chmod_secure(path) + + def _audit_paths(self): + yield self.baseFilename + for n in range(1, self.backupCount + 1): + candidate = f"{self.baseFilename}.{n}" + if os.path.exists(candidate): + yield candidate + + +def _chmod_secure(path: str) -> None: + """Set owner-only read/write on path; log warning on failure.""" + try: + if os.path.exists(path): + os.chmod(path, stat.S_IRUSR | stat.S_IWUSR) + except OSError as exc: + logger.warning( + "Could not secure audit log permissions on %s: %s — " + "file may have permissive permissions", + path, + exc, + ) + + # --------------------------------------------------------------------------- # SecurityLogger # --------------------------------------------------------------------------- @@ -152,57 +206,57 @@ class SecurityLogger: Writes structured JSON security events to a dedicated audit log file (separate from the main application log). The audit log uses a RotatingFileHandler with secure (owner-only) file permissions. + + Args: + log_dir: Directory for the audit log file. Defaults to + ~/.powertraderai/logs so the source tree is never polluted. """ AUDIT_LOG_FILENAME = "security_audit.jsonl" MAX_BYTES = 10 * 1024 * 1024 # 10 MB per file - BACKUP_COUNT = 10 # Keep 10 rotated files + BACKUP_COUNT = 10 # Keep 10 rotated files def __init__(self, log_dir: Optional[str] = None): - self._log_dir = log_dir or os.path.dirname(os.path.abspath(__file__)) + self._log_dir = log_dir or os.path.join( + os.path.expanduser("~"), ".powertraderai", "logs" + ) + os.makedirs(self._log_dir, exist_ok=True) self._audit_path = os.path.join(self._log_dir, self.AUDIT_LOG_FILENAME) self._lock = threading.Lock() - self._handler: Optional[logging.handlers.RotatingFileHandler] = None - self._logger = self._setup_audit_logger() - - def _setup_audit_logger(self) -> logging.Logger: - """Configure dedicated audit logger with rotation and secure permissions.""" - import hashlib - - path_hash = hashlib.md5(self._audit_path.encode()).hexdigest()[:8] - audit_logger = logging.getLogger(f"pt.security.audit.{path_hash}") - audit_logger.setLevel(logging.DEBUG) - audit_logger.propagate = False # Don't propagate to root logger - - if not audit_logger.handlers: - handler = logging.handlers.RotatingFileHandler( - self._audit_path, - maxBytes=self.MAX_BYTES, - backupCount=self.BACKUP_COUNT, - encoding="utf-8", - ) - handler.setFormatter(logging.Formatter("%(message)s")) # Raw JSON - handler.addFilter(CorrelationLogFilter()) - audit_logger.addHandler(handler) - self._handler = handler - self._secure_audit_file() - - return audit_logger - - def _secure_audit_file(self) -> None: - """Set owner-only permissions on audit log file.""" - try: - if os.path.exists(self._audit_path): - os.chmod(self._audit_path, stat.S_IRUSR | stat.S_IWUSR) - except (OSError, AttributeError): - pass + # Use a private logger instance (not the global registry) to avoid the + # shared-handler pitfall when two SecurityLogger instances use the same dir. + self._audit_logger = logging.Logger( + f"pt.security.audit.{id(self)}", level=logging.DEBUG + ) + self._audit_logger.propagate = False + self._handler = self._create_handler() + self._audit_logger.addHandler(self._handler) + + def _create_handler(self) -> _SecureRotatingFileHandler: + handler = _SecureRotatingFileHandler( + self._audit_path, + maxBytes=self.MAX_BYTES, + backupCount=self.BACKUP_COUNT, + encoding="utf-8", + ) + handler.setFormatter(logging.Formatter("%(message)s")) # Raw JSON lines + handler.addFilter(CorrelationLogFilter()) + # Secure permissions once at creation + _chmod_secure(self._audit_path) + return handler + + def close(self) -> None: + """Flush and close the underlying handler (call on shutdown).""" + if self._handler: + self._handler.flush() + self._handler.close() + self._audit_logger.removeHandler(self._handler) def _emit(self, event: SecurityEvent) -> None: """Write security event to audit log (thread-safe).""" with self._lock: try: - self._logger.info(event.to_json()) - self._secure_audit_file() + self._audit_logger.info(event.to_json()) except Exception as exc: logger.error("Failed to write security audit event: %s", exc) @@ -242,23 +296,15 @@ def log_auth_attempt( ) -> None: """Log an API authentication attempt (success or failure).""" event_type = ( - SecurityEventType.AUTH_SUCCESS - if success - else SecurityEventType.AUTH_FAILURE + SecurityEventType.AUTH_SUCCESS if success else SecurityEventType.AUTH_FAILURE ) msg = f"API auth {'succeeded' if success else 'FAILED'} for {api_name}" if not success: logger.warning("SECURITY: %s", msg) - event = self._make_event( - event_type, - msg, - source=api_name, - success=success, - details=details, - source_ip=source_ip, - user_id=user_id, - ) - self._emit(event) + self._emit(self._make_event( + event_type, msg, source=api_name, success=success, + details=details, source_ip=source_ip, user_id=user_id, + )) def log_credential_use( self, @@ -268,14 +314,10 @@ def log_credential_use( ) -> None: """Log that credentials were accessed/used for an operation.""" msg = f"Credential used: {api_name} for {operation}" - event = self._make_event( - SecurityEventType.CREDENTIAL_USE, - msg, - source=api_name, - success=True, + self._emit(self._make_event( + SecurityEventType.CREDENTIAL_USE, msg, source=api_name, success=True, details={**(details or {}), "operation": operation}, - ) - self._emit(event) + )) def log_credential_rotation( self, @@ -284,19 +326,13 @@ def log_credential_rotation( details: Optional[Dict[str, Any]] = None, ) -> None: """Log a credential rotation event.""" - msg = ( - f"Credential rotation {'succeeded' if success else 'FAILED'} for {api_name}" - ) + msg = f"Credential rotation {'succeeded' if success else 'FAILED'} for {api_name}" if not success: logger.critical("SECURITY: %s", msg) - event = self._make_event( - SecurityEventType.CREDENTIAL_ROTATION, - msg, - source=api_name, - success=success, - details=details, - ) - self._emit(event) + self._emit(self._make_event( + SecurityEventType.CREDENTIAL_ROTATION, msg, source=api_name, + success=success, details=details, + )) def log_suspicious_activity( self, @@ -308,16 +344,10 @@ def log_suspicious_activity( """Log detected suspicious activity.""" msg = f"Suspicious activity detected: {activity_type}" logger.warning("SECURITY ALERT: %s", msg) - event = self._make_event( - SecurityEventType.SUSPICIOUS_ACTIVITY, - msg, - source=activity_type, - success=False, - source_ip=source_ip, - user_id=user_id, - details=details, - ) - self._emit(event) + self._emit(self._make_event( + SecurityEventType.SUSPICIOUS_ACTIVITY, msg, source=activity_type, + success=False, source_ip=source_ip, user_id=user_id, details=details, + )) def log_permission_denied( self, @@ -328,14 +358,10 @@ def log_permission_denied( """Log an API permission denial.""" msg = f"Permission denied: {required_permission} on {api_name}" logger.error("SECURITY: %s", msg) - event = self._make_event( - SecurityEventType.PERMISSION_DENIED, - msg, - source=api_name, - success=False, + self._emit(self._make_event( + SecurityEventType.PERMISSION_DENIED, msg, source=api_name, success=False, details={**(details or {}), "required_permission": required_permission}, - ) - self._emit(event) + )) def log_trade_event( self, @@ -349,25 +375,39 @@ def log_trade_event( ) -> None: """Log a trade execution event to the audit trail.""" event_type = ( - SecurityEventType.TRADE_EXECUTED - if success - else SecurityEventType.TRADE_REJECTED + SecurityEventType.TRADE_EXECUTED if success else SecurityEventType.TRADE_REJECTED ) msg = f"Trade {'executed' if success else 'REJECTED'}: {side} {quantity} {symbol} @ {price}" - event = self._make_event( - event_type, - msg, - success=success, - details={ - "symbol": symbol, - "side": side, - "quantity": quantity, - "price": price, - "order_id": order_id, - **(details or {}), - }, - ) - self._emit(event) + self._emit(self._make_event( + event_type, msg, success=success, + details={"symbol": symbol, "side": side, "quantity": quantity, + "price": price, "order_id": order_id, **(details or {})}, + )) + + def log_rate_limit( + self, + api_name: str, + details: Optional[Dict[str, Any]] = None, + ) -> None: + """Log a rate limit event.""" + msg = f"Rate limit hit on {api_name}" + logger.warning("SECURITY: %s", msg) + self._emit(self._make_event( + SecurityEventType.RATE_LIMIT, msg, source=api_name, + success=False, details=details, + )) + + def log_config_change( + self, + component: str, + details: Optional[Dict[str, Any]] = None, + ) -> None: + """Log a configuration change event.""" + msg = f"Configuration changed: {component}" + logger.info("SECURITY: %s", msg) + self._emit(self._make_event( + SecurityEventType.CONFIG_CHANGE, msg, source=component, details=details, + )) def log_system_event( self, @@ -375,19 +415,24 @@ def log_system_event( message: str, details: Optional[Dict[str, Any]] = None, ) -> None: - """Log a system lifecycle event (start, stop, config change).""" - event = self._make_event(event_type, message, details=details) - self._emit(event) + """Log a system lifecycle event (start, stop, config change, rate limit).""" + self._emit(self._make_event(event_type, message, details=details)) + + def get_recent_events(self, limit: int = 100) -> List[dict]: + """ + Read the last `limit` events from the audit log. - def get_recent_events(self, limit: int = 100) -> list: - """Read the last `limit` events from the audit log.""" + Uses O(limit) tail reading via collections.deque so cost scales with + limit, not file size. Only reads the active log file; rotated backup + files are not included. + """ if not os.path.exists(self._audit_path): return [] try: with open(self._audit_path, "r", encoding="utf-8") as f: - lines = f.readlines() + tail = collections.deque(f, maxlen=limit) events = [] - for line in lines[-limit:]: + for line in tail: line = line.strip() if line: try: @@ -400,6 +445,34 @@ def get_recent_events(self, limit: int = 100) -> list: # --------------------------------------------------------------------------- -# Module-level singleton +# Module-level lazy singleton # --------------------------------------------------------------------------- -security_logger = SecurityLogger() +_security_logger_instance: Optional[SecurityLogger] = None +_security_logger_lock = threading.Lock() + + +def get_security_logger(log_dir: Optional[str] = None) -> SecurityLogger: + """ + Return the application-wide SecurityLogger singleton. + + Lazy: the logger (and its audit file) is only created on first call, + not at module import time. Pass log_dir only on the first call; + subsequent calls return the existing instance. + """ + global _security_logger_instance + if _security_logger_instance is None: + with _security_logger_lock: + if _security_logger_instance is None: + _security_logger_instance = SecurityLogger(log_dir=log_dir) + return _security_logger_instance + + +# Convenience alias — module-level name that lazily resolves on first attribute access +class _LazySecurityLogger: + """Proxy that forwards all attribute access to get_security_logger().""" + + def __getattr__(self, name: str): + return getattr(get_security_logger(), name) + + +security_logger = _LazySecurityLogger() diff --git a/app/test_security_logger.py b/app/test_security_logger.py index f977d831..70013235 100644 --- a/app/test_security_logger.py +++ b/app/test_security_logger.py @@ -2,12 +2,14 @@ import json import os +import shutil import tempfile import threading import time import unittest from pt_security_logger import ( + CorrelationContext, SecurityEvent, SecurityEventType, SecurityLogger, @@ -31,27 +33,43 @@ def test_set_and_get(self): self.assertEqual(get_correlation_id(), "test-cid-123") def test_context_manager_sets_id(self): - with correlation_context("ctx-abc") as cid: + with CorrelationContext("ctx-abc") as cid: self.assertEqual(cid, "ctx-abc") self.assertEqual(get_correlation_id(), "ctx-abc") def test_context_manager_restores_previous(self): set_correlation_id("outer") - with correlation_context("inner"): + with CorrelationContext("inner"): self.assertEqual(get_correlation_id(), "inner") self.assertEqual(get_correlation_id(), "outer") def test_context_manager_clears_when_no_previous(self): clear_correlation_id() - with correlation_context("temp"): + with CorrelationContext("temp"): pass self.assertIsNone(get_correlation_id()) def test_context_manager_auto_generates_id(self): - with correlation_context() as cid: + with CorrelationContext() as cid: self.assertIsNotNone(cid) self.assertTrue(cid.startswith("CID_")) + def test_context_manager_nested_safe(self): + """Reusing the same CorrelationContext instance in nested with-blocks is safe.""" + ctx = CorrelationContext("reused") + with ctx as cid1: + self.assertEqual(cid1, "reused") + with ctx as cid2: + self.assertEqual(cid2, "reused") + # Inner exit must restore the outer's saved state + self.assertEqual(get_correlation_id(), "reused") + self.assertIsNone(get_correlation_id()) + + def test_snake_case_alias_works(self): + """correlation_context alias must still function for backwards compatibility.""" + with correlation_context("alias-test") as cid: + self.assertEqual(cid, "alias-test") + def test_thread_local_isolation(self): """Each thread has its own correlation ID.""" results = {} @@ -77,6 +95,13 @@ def test_new_id_format(self): self.assertTrue(eid.startswith("SEC_")) self.assertGreater(len(eid), 10) + def test_new_id_uses_utc(self): + """Event ID must be UTC-based (no timezone-ambiguous local time).""" + from datetime import datetime, timezone + before = datetime.now(timezone.utc).strftime("%Y%m%d") + eid = SecurityEvent.new_id() + self.assertIn(before, eid) + def test_to_dict_has_required_fields(self): event = SecurityEvent( event_id="SEC_001", @@ -86,10 +111,8 @@ def test_to_dict_has_required_fields(self): correlation_id=None, ) d = event.to_dict() - self.assertIn("event_id", d) - self.assertIn("event_type", d) - self.assertIn("timestamp", d) - self.assertIn("message", d) + for field in ("event_id", "event_type", "timestamp", "message"): + self.assertIn(field, d) def test_to_json_valid(self): event = SecurityEvent( @@ -110,45 +133,37 @@ def setUp(self): clear_correlation_id() def tearDown(self): - import shutil - clear_correlation_id() - # Close handlers to release file locks (Windows) - for handler in self.sec_logger._logger.handlers[:]: - handler.close() - self.sec_logger._logger.removeHandler(handler) + # close() flushes and releases file handles (critical on Windows) + self.sec_logger.close() shutil.rmtree(self.tmpdir, ignore_errors=True) - def _get_events(self): + def _events(self): return self.sec_logger.get_recent_events() def test_log_auth_attempt_success(self): self.sec_logger.log_auth_attempt("robinhood", success=True) - events = self._get_events() + events = self._events() self.assertEqual(len(events), 1) self.assertEqual(events[0]["event_type"], SecurityEventType.AUTH_SUCCESS.value) self.assertTrue(events[0]["success"]) def test_log_auth_attempt_failure(self): self.sec_logger.log_auth_attempt("robinhood", success=False) - events = self._get_events() + events = self._events() self.assertEqual(events[0]["event_type"], SecurityEventType.AUTH_FAILURE.value) self.assertFalse(events[0]["success"]) def test_log_credential_use(self): self.sec_logger.log_credential_use("robinhood", "place_order") - events = self._get_events() - self.assertEqual( - events[0]["event_type"], SecurityEventType.CREDENTIAL_USE.value - ) + events = self._events() + self.assertEqual(events[0]["event_type"], SecurityEventType.CREDENTIAL_USE.value) self.assertEqual(events[0]["details"]["operation"], "place_order") def test_log_credential_rotation(self): self.sec_logger.log_credential_rotation("robinhood", success=True) - events = self._get_events() - self.assertEqual( - events[0]["event_type"], SecurityEventType.CREDENTIAL_ROTATION.value - ) + events = self._events() + self.assertEqual(events[0]["event_type"], SecurityEventType.CREDENTIAL_ROTATION.value) def test_log_suspicious_activity(self): self.sec_logger.log_suspicious_activity( @@ -156,54 +171,56 @@ def test_log_suspicious_activity(self): source_ip="1.2.3.4", details={"endpoint": "/orders", "count": 100}, ) - events = self._get_events() - self.assertEqual( - events[0]["event_type"], SecurityEventType.SUSPICIOUS_ACTIVITY.value - ) + events = self._events() + self.assertEqual(events[0]["event_type"], SecurityEventType.SUSPICIOUS_ACTIVITY.value) self.assertEqual(events[0]["source_ip"], "1.2.3.4") def test_log_permission_denied(self): self.sec_logger.log_permission_denied("robinhood", "sell") - events = self._get_events() - self.assertEqual( - events[0]["event_type"], SecurityEventType.PERMISSION_DENIED.value - ) + events = self._events() + self.assertEqual(events[0]["event_type"], SecurityEventType.PERMISSION_DENIED.value) self.assertEqual(events[0]["details"]["required_permission"], "sell") def test_log_trade_event_success(self): - self.sec_logger.log_trade_event( - "BTC-USD", "buy", 0.1, 45000.0, order_id="ord-001" - ) - events = self._get_events() - self.assertEqual( - events[0]["event_type"], SecurityEventType.TRADE_EXECUTED.value - ) + self.sec_logger.log_trade_event("BTC-USD", "buy", 0.1, 45000.0, order_id="ord-001") + events = self._events() + self.assertEqual(events[0]["event_type"], SecurityEventType.TRADE_EXECUTED.value) self.assertEqual(events[0]["details"]["symbol"], "BTC-USD") def test_log_trade_event_rejected(self): self.sec_logger.log_trade_event("ETH-USD", "sell", 1.0, 3000.0, success=False) - events = self._get_events() - self.assertEqual( - events[0]["event_type"], SecurityEventType.TRADE_REJECTED.value - ) + events = self._events() + self.assertEqual(events[0]["event_type"], SecurityEventType.TRADE_REJECTED.value) + + def test_log_rate_limit(self): + self.sec_logger.log_rate_limit("binance", details={"endpoint": "/api/v3/order"}) + events = self._events() + self.assertEqual(events[0]["event_type"], SecurityEventType.RATE_LIMIT.value) + self.assertEqual(events[0]["source"], "binance") + + def test_log_config_change(self): + self.sec_logger.log_config_change("risk_limits", details={"max_position": 0.1}) + events = self._events() + self.assertEqual(events[0]["event_type"], SecurityEventType.CONFIG_CHANGE.value) + self.assertEqual(events[0]["source"], "risk_limits") def test_correlation_id_captured_in_event(self): - with correlation_context("trade-workflow-xyz"): + with CorrelationContext("trade-workflow-xyz"): self.sec_logger.log_auth_attempt("robinhood", success=True) - events = self._get_events() + events = self._events() self.assertEqual(events[0]["correlation_id"], "trade-workflow-xyz") def test_correlation_id_none_outside_context(self): clear_correlation_id() self.sec_logger.log_auth_attempt("robinhood", success=True) - events = self._get_events() + events = self._events() self.assertIsNone(events[0]["correlation_id"]) def test_multiple_events_ordered(self): self.sec_logger.log_auth_attempt("robinhood", success=True) self.sec_logger.log_credential_use("robinhood", "fetch_positions") self.sec_logger.log_trade_event("BTC-USD", "buy", 0.5, 44000.0) - events = self._get_events() + events = self._events() self.assertEqual(len(events), 3) timestamps = [e["timestamp"] for e in events] self.assertEqual(timestamps, sorted(timestamps)) @@ -213,21 +230,30 @@ def test_audit_file_created(self): self.assertTrue(os.path.exists(self.sec_logger._audit_path)) def test_get_recent_events_empty_when_no_log(self): - # Logger on a path with no events yet + """get_recent_events() returns [] when no audit log file exists.""" tmpdir2 = tempfile.mkdtemp() - sl = SecurityLogger(log_dir=tmpdir2) - for h in sl._logger.handlers[:]: - h.close() - sl._logger.removeHandler(h) - import shutil - - shutil.rmtree(tmpdir2, ignore_errors=True) + try: + sl = SecurityLogger(log_dir=tmpdir2) + # Close immediately — no events written, so file may not exist yet + sl.close() + # Remove the file if it was created + audit_path = sl._audit_path + if os.path.exists(audit_path): + os.remove(audit_path) + # Now verify empty result + self.assertEqual(sl.get_recent_events(), []) + finally: + shutil.rmtree(tmpdir2, ignore_errors=True) def test_get_recent_events_limit(self): + """get_recent_events(limit=N) returns exactly the last N events.""" for i in range(10): self.sec_logger.log_auth_attempt("api", success=True) events = self.sec_logger.get_recent_events(limit=5) self.assertEqual(len(events), 5) + # Should be the last 5 — timestamps are non-decreasing + all_events = self.sec_logger.get_recent_events(limit=100) + self.assertEqual(events, all_events[-5:]) def test_log_system_event(self): self.sec_logger.log_system_event( @@ -235,7 +261,7 @@ def test_log_system_event(self): "PowerTraderAI started", details={"version": "1.0"}, ) - events = self._get_events() + events = self._events() self.assertEqual(events[0]["event_type"], SecurityEventType.SYSTEM_START.value) From 2561604639bcf8dba22b7ac494d064d08a2723c4 Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Tue, 19 May 2026 15:32:37 +0300 Subject: [PATCH 5/7] style: apply Black formatting --- app/pt_security_logger.py | 153 ++++++++++++++++++++++++------------ app/test_security_logger.py | 29 +++++-- 2 files changed, 126 insertions(+), 56 deletions(-) diff --git a/app/pt_security_logger.py b/app/pt_security_logger.py index 87721ee5..c6b414e9 100644 --- a/app/pt_security_logger.py +++ b/app/pt_security_logger.py @@ -46,19 +46,19 @@ # Security event types # --------------------------------------------------------------------------- class SecurityEventType(Enum): - AUTH_ATTEMPT = "auth_attempt" # API key authentication attempt - AUTH_SUCCESS = "auth_success" # Successful authentication - AUTH_FAILURE = "auth_failure" # Failed authentication - CREDENTIAL_USE = "credential_use" # Credential accessed/used - CREDENTIAL_ROTATION = "credential_rotation" # Credential rotated - SUSPICIOUS_ACTIVITY = "suspicious_activity" # Anomalous behavior detected - PERMISSION_DENIED = "permission_denied" # Insufficient API permissions - RATE_LIMIT = "rate_limit" # Rate limit hit - TRADE_EXECUTED = "trade_executed" # Order placed - TRADE_REJECTED = "trade_rejected" # Order rejected - CONFIG_CHANGE = "config_change" # Configuration modified - SYSTEM_START = "system_start" # Application started - SYSTEM_STOP = "system_stop" # Application stopped + AUTH_ATTEMPT = "auth_attempt" # API key authentication attempt + AUTH_SUCCESS = "auth_success" # Successful authentication + AUTH_FAILURE = "auth_failure" # Failed authentication + CREDENTIAL_USE = "credential_use" # Credential accessed/used + CREDENTIAL_ROTATION = "credential_rotation" # Credential rotated + SUSPICIOUS_ACTIVITY = "suspicious_activity" # Anomalous behavior detected + PERMISSION_DENIED = "permission_denied" # Insufficient API permissions + RATE_LIMIT = "rate_limit" # Rate limit hit + TRADE_EXECUTED = "trade_executed" # Order placed + TRADE_REJECTED = "trade_rejected" # Order rejected + CONFIG_CHANGE = "config_change" # Configuration modified + SYSTEM_START = "system_start" # Application started + SYSTEM_STOP = "system_stop" # Application stopped @dataclass @@ -214,7 +214,7 @@ class SecurityLogger: AUDIT_LOG_FILENAME = "security_audit.jsonl" MAX_BYTES = 10 * 1024 * 1024 # 10 MB per file - BACKUP_COUNT = 10 # Keep 10 rotated files + BACKUP_COUNT = 10 # Keep 10 rotated files def __init__(self, log_dir: Optional[str] = None): self._log_dir = log_dir or os.path.join( @@ -296,15 +296,24 @@ def log_auth_attempt( ) -> None: """Log an API authentication attempt (success or failure).""" event_type = ( - SecurityEventType.AUTH_SUCCESS if success else SecurityEventType.AUTH_FAILURE + SecurityEventType.AUTH_SUCCESS + if success + else SecurityEventType.AUTH_FAILURE ) msg = f"API auth {'succeeded' if success else 'FAILED'} for {api_name}" if not success: logger.warning("SECURITY: %s", msg) - self._emit(self._make_event( - event_type, msg, source=api_name, success=success, - details=details, source_ip=source_ip, user_id=user_id, - )) + self._emit( + self._make_event( + event_type, + msg, + source=api_name, + success=success, + details=details, + source_ip=source_ip, + user_id=user_id, + ) + ) def log_credential_use( self, @@ -314,10 +323,15 @@ def log_credential_use( ) -> None: """Log that credentials were accessed/used for an operation.""" msg = f"Credential used: {api_name} for {operation}" - self._emit(self._make_event( - SecurityEventType.CREDENTIAL_USE, msg, source=api_name, success=True, - details={**(details or {}), "operation": operation}, - )) + self._emit( + self._make_event( + SecurityEventType.CREDENTIAL_USE, + msg, + source=api_name, + success=True, + details={**(details or {}), "operation": operation}, + ) + ) def log_credential_rotation( self, @@ -326,13 +340,20 @@ def log_credential_rotation( details: Optional[Dict[str, Any]] = None, ) -> None: """Log a credential rotation event.""" - msg = f"Credential rotation {'succeeded' if success else 'FAILED'} for {api_name}" + msg = ( + f"Credential rotation {'succeeded' if success else 'FAILED'} for {api_name}" + ) if not success: logger.critical("SECURITY: %s", msg) - self._emit(self._make_event( - SecurityEventType.CREDENTIAL_ROTATION, msg, source=api_name, - success=success, details=details, - )) + self._emit( + self._make_event( + SecurityEventType.CREDENTIAL_ROTATION, + msg, + source=api_name, + success=success, + details=details, + ) + ) def log_suspicious_activity( self, @@ -344,10 +365,17 @@ def log_suspicious_activity( """Log detected suspicious activity.""" msg = f"Suspicious activity detected: {activity_type}" logger.warning("SECURITY ALERT: %s", msg) - self._emit(self._make_event( - SecurityEventType.SUSPICIOUS_ACTIVITY, msg, source=activity_type, - success=False, source_ip=source_ip, user_id=user_id, details=details, - )) + self._emit( + self._make_event( + SecurityEventType.SUSPICIOUS_ACTIVITY, + msg, + source=activity_type, + success=False, + source_ip=source_ip, + user_id=user_id, + details=details, + ) + ) def log_permission_denied( self, @@ -358,10 +386,15 @@ def log_permission_denied( """Log an API permission denial.""" msg = f"Permission denied: {required_permission} on {api_name}" logger.error("SECURITY: %s", msg) - self._emit(self._make_event( - SecurityEventType.PERMISSION_DENIED, msg, source=api_name, success=False, - details={**(details or {}), "required_permission": required_permission}, - )) + self._emit( + self._make_event( + SecurityEventType.PERMISSION_DENIED, + msg, + source=api_name, + success=False, + details={**(details or {}), "required_permission": required_permission}, + ) + ) def log_trade_event( self, @@ -375,14 +408,26 @@ def log_trade_event( ) -> None: """Log a trade execution event to the audit trail.""" event_type = ( - SecurityEventType.TRADE_EXECUTED if success else SecurityEventType.TRADE_REJECTED + SecurityEventType.TRADE_EXECUTED + if success + else SecurityEventType.TRADE_REJECTED ) msg = f"Trade {'executed' if success else 'REJECTED'}: {side} {quantity} {symbol} @ {price}" - self._emit(self._make_event( - event_type, msg, success=success, - details={"symbol": symbol, "side": side, "quantity": quantity, - "price": price, "order_id": order_id, **(details or {})}, - )) + self._emit( + self._make_event( + event_type, + msg, + success=success, + details={ + "symbol": symbol, + "side": side, + "quantity": quantity, + "price": price, + "order_id": order_id, + **(details or {}), + }, + ) + ) def log_rate_limit( self, @@ -392,10 +437,15 @@ def log_rate_limit( """Log a rate limit event.""" msg = f"Rate limit hit on {api_name}" logger.warning("SECURITY: %s", msg) - self._emit(self._make_event( - SecurityEventType.RATE_LIMIT, msg, source=api_name, - success=False, details=details, - )) + self._emit( + self._make_event( + SecurityEventType.RATE_LIMIT, + msg, + source=api_name, + success=False, + details=details, + ) + ) def log_config_change( self, @@ -405,9 +455,14 @@ def log_config_change( """Log a configuration change event.""" msg = f"Configuration changed: {component}" logger.info("SECURITY: %s", msg) - self._emit(self._make_event( - SecurityEventType.CONFIG_CHANGE, msg, source=component, details=details, - )) + self._emit( + self._make_event( + SecurityEventType.CONFIG_CHANGE, + msg, + source=component, + details=details, + ) + ) def log_system_event( self, diff --git a/app/test_security_logger.py b/app/test_security_logger.py index 70013235..45cd03c3 100644 --- a/app/test_security_logger.py +++ b/app/test_security_logger.py @@ -98,6 +98,7 @@ def test_new_id_format(self): def test_new_id_uses_utc(self): """Event ID must be UTC-based (no timezone-ambiguous local time).""" from datetime import datetime, timezone + before = datetime.now(timezone.utc).strftime("%Y%m%d") eid = SecurityEvent.new_id() self.assertIn(before, eid) @@ -157,13 +158,17 @@ def test_log_auth_attempt_failure(self): def test_log_credential_use(self): self.sec_logger.log_credential_use("robinhood", "place_order") events = self._events() - self.assertEqual(events[0]["event_type"], SecurityEventType.CREDENTIAL_USE.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.CREDENTIAL_USE.value + ) self.assertEqual(events[0]["details"]["operation"], "place_order") def test_log_credential_rotation(self): self.sec_logger.log_credential_rotation("robinhood", success=True) events = self._events() - self.assertEqual(events[0]["event_type"], SecurityEventType.CREDENTIAL_ROTATION.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.CREDENTIAL_ROTATION.value + ) def test_log_suspicious_activity(self): self.sec_logger.log_suspicious_activity( @@ -172,25 +177,35 @@ def test_log_suspicious_activity(self): details={"endpoint": "/orders", "count": 100}, ) events = self._events() - self.assertEqual(events[0]["event_type"], SecurityEventType.SUSPICIOUS_ACTIVITY.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.SUSPICIOUS_ACTIVITY.value + ) self.assertEqual(events[0]["source_ip"], "1.2.3.4") def test_log_permission_denied(self): self.sec_logger.log_permission_denied("robinhood", "sell") events = self._events() - self.assertEqual(events[0]["event_type"], SecurityEventType.PERMISSION_DENIED.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.PERMISSION_DENIED.value + ) self.assertEqual(events[0]["details"]["required_permission"], "sell") def test_log_trade_event_success(self): - self.sec_logger.log_trade_event("BTC-USD", "buy", 0.1, 45000.0, order_id="ord-001") + self.sec_logger.log_trade_event( + "BTC-USD", "buy", 0.1, 45000.0, order_id="ord-001" + ) events = self._events() - self.assertEqual(events[0]["event_type"], SecurityEventType.TRADE_EXECUTED.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.TRADE_EXECUTED.value + ) self.assertEqual(events[0]["details"]["symbol"], "BTC-USD") def test_log_trade_event_rejected(self): self.sec_logger.log_trade_event("ETH-USD", "sell", 1.0, 3000.0, success=False) events = self._events() - self.assertEqual(events[0]["event_type"], SecurityEventType.TRADE_REJECTED.value) + self.assertEqual( + events[0]["event_type"], SecurityEventType.TRADE_REJECTED.value + ) def test_log_rate_limit(self): self.sec_logger.log_rate_limit("binance", details={"endpoint": "/api/v3/order"}) From 10858f4fcd1b208a62d9016f6445396ed3118904 Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Wed, 20 May 2026 16:48:14 +0300 Subject: [PATCH 6/7] fix: address Copilot round-3 review on production security logging - Remove unused hashlib import - close(): make idempotent by clearing self._handler after shutdown - get_recent_events: guard limit <= 0 (deque maxlen rejects negatives); docstring corrected to reflect O(file_size) time / O(limit) memory - test_multiple_events_ordered: assert event_type append order (robust to clock adjustments) + non-decreasing timestamp sanity check --- app/pt_security_logger.py | 28 +++++++++++++++++++--------- app/test_security_logger.py | 15 ++++++++++++++- 2 files changed, 33 insertions(+), 10 deletions(-) diff --git a/app/pt_security_logger.py b/app/pt_security_logger.py index c6b414e9..1afa5911 100644 --- a/app/pt_security_logger.py +++ b/app/pt_security_logger.py @@ -25,7 +25,6 @@ """ import collections -import hashlib import json import logging import logging.handlers @@ -246,11 +245,17 @@ def _create_handler(self) -> _SecureRotatingFileHandler: return handler def close(self) -> None: - """Flush and close the underlying handler (call on shutdown).""" - if self._handler: - self._handler.flush() - self._handler.close() - self._audit_logger.removeHandler(self._handler) + """Flush and close the underlying handler (call on shutdown). + Idempotent: safe to call multiple times.""" + if self._handler is None: + return + handler = self._handler + self._handler = None + try: + handler.flush() + handler.close() + finally: + self._audit_logger.removeHandler(handler) def _emit(self, event: SecurityEvent) -> None: """Write security event to audit log (thread-safe).""" @@ -477,10 +482,15 @@ def get_recent_events(self, limit: int = 100) -> List[dict]: """ Read the last `limit` events from the audit log. - Uses O(limit) tail reading via collections.deque so cost scales with - limit, not file size. Only reads the active log file; rotated backup - files are not included. + Time complexity is O(file_size) — the iterator walks every line; + memory is bounded to O(limit) via ``collections.deque(maxlen=limit)``. + Only reads the active log file; rotated backups (`*.1`, `*.2`, …) + are not included. + + Returns [] when limit <= 0. """ + if limit <= 0: + return [] if not os.path.exists(self._audit_path): return [] try: diff --git a/app/test_security_logger.py b/app/test_security_logger.py index 45cd03c3..c180edd6 100644 --- a/app/test_security_logger.py +++ b/app/test_security_logger.py @@ -237,8 +237,21 @@ def test_multiple_events_ordered(self): self.sec_logger.log_trade_event("BTC-USD", "buy", 0.5, 44000.0) events = self._events() self.assertEqual(len(events), 3) + # Assert event_type append order — robust against clock adjustments + # (time.time() can move backwards on NTP correction or DST shifts). + self.assertEqual( + [e["event_type"] for e in events], + [ + SecurityEventType.AUTH_SUCCESS.value, + SecurityEventType.CREDENTIAL_USE.value, + SecurityEventType.TRADE_EXECUTED.value, + ], + ) + # Sanity check: timestamps are non-decreasing (cheap monotonicity + # check; sub-second jumps backwards from NTP do not break this). timestamps = [e["timestamp"] for e in events] - self.assertEqual(timestamps, sorted(timestamps)) + for prev, curr in zip(timestamps, timestamps[1:]): + self.assertLessEqual(prev, curr) def test_audit_file_created(self): self.sec_logger.log_auth_attempt("robinhood", success=True) From 98cfa3c5deb5d24e2feddbacdc9bf3aacfe9ca0b Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Thu, 21 May 2026 17:01:46 +0300 Subject: [PATCH 7/7] chore: gitignore contributor working notes (CLAUDE.md, LEARNINGS.md) --- .gitignore | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.gitignore b/.gitignore index 58d3904b..87908cf7 100644 --- a/.gitignore +++ b/.gitignore @@ -220,3 +220,7 @@ Thumbs.db desktop.ini app/security_audit.jsonl app/credential_audit.jsonl + +# Contributor working notes (not project docs) +CLAUDE.md +LEARNINGS.md