gavindiaz/mt5-ai-xauusd-trader
1
0
Fork 0
mirror of https://github.com/triqbit/mt5-ai-xauusd-trader.git synced 2026-11-02 06:56:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: 🔐 Jules02: Security hardening — dynamic secret masking (#969)

Browse Source 
* feat: 🔐 Jules02: Security hardening — dynamic secret masking

- Implement dynamic secret discovery in SecretMaskingProcessor to automatically mask all SecretStr/SecretBytes fields.
- Enhance URL credential extraction to protect embedded passwords in connection strings (database, redis, etc).
- Refactor main.py to dynamically exclude sensitive fields from CLI output and audit snapshots.
- Add comprehensive tests for the new masking logic.

Co-authored-by: xnessom <255288032+xnessom@users.noreply.github.com>

* feat: 🔐 Jules02: Security hardening — dynamic secret masking

- Implement dynamic secret discovery in SecretMaskingProcessor to automatically mask all SecretStr/SecretBytes fields.
- Enhance URL credential extraction to protect embedded passwords in connection strings (database, redis, etc).
- Refactor main.py to dynamically exclude sensitive fields from CLI output and audit snapshots.
- Update SECURITY.md to reflect implemented hardening measures.
- Harmonize numpy version to 2.2.7 across all requirements to resolve CI dependency conflicts.
- Fix Ruff lint errors in log_config.py.

Co-authored-by: xnessom <255288032+xnessom@users.noreply.github.com>

---------

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
Co-authored-by: xnessom <255288032+xnessom@users.noreply.github.com>
This commit is contained in:
xnessom
2026-05-10 02:26:14 +00:00
committed by GitHub
co-authored by xnessom google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
parent e4f7252eac
commit 87df32b64b
9 changed files with 129 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SECURITY.md
+1 -1
View File
@@ -129,7 +129,7 @@ written justification before the deadline passes.
| Category | Specific Examples |
|---|---|
| **Credential Exposure** | MT5 password/token in logs, `.env` leakage, SecretStr bypass |
| **Credential Exposure** | MT5 password/token in logs, `.env` leakage, SecretStr bypass, dynamic discovery gaps, URL credential leaks |
| **Trade Execution** | RiskManager approval bypass, signal injection, order tampering |
| **Database Layer** | SQL injection in TradeLogger, Alembic migration tampering, constraint bypass |
| **Model Integrity** | Unsafe deserialization of PPO/LSTM/Dreamer weights, model poisoning |
main.py
+20 -24
View File
@@ -211,17 +211,15 @@ def run_live(
# 0. Periodic Audit of Configuration State
if loop_count % 100 == 0 and audit_logger:
# Dynamic exclusion of all SecretStr/SecretBytes fields
secret_fields = {
f for f, info in cfg.__class__.model_fields.items()
if "Secret" in str(info.annotation)
}
audit_logger.log_config_snapshot(
cfg.model_dump(
mode="json",
exclude={
"mt5_password",
"metaapi_token",
"metaapi_account_id",
"database_url",
"redis_url",
"telegram_token",
},
exclude=secret_fields,
),
reason=f"periodic_check_loop_{loop_count}",
)
@@ -933,17 +931,16 @@ def main() -> int:
config_table.add_column("Parameter", style="cyan")
config_table.add_column("Value", style="white")
# Dynamic exclusion of all SecretStr/SecretBytes fields
secret_fields = {
f for f, info in cfg.__class__.model_fields.items()
if "Secret" in str(info.annotation)
}
# Get sanitized dump
sanitized_cfg = cfg.model_dump(
mode="json",
exclude={
"mt5_password",
"metaapi_token",
"metaapi_account_id",
"database_url",
"redis_url",
"telegram_token",
},
exclude=secret_fields,
)
for key, value in sorted(sanitized_cfg.items()):
@@ -1056,18 +1053,17 @@ def main() -> int:
audit_db_url = database_url if "sqlite" in database_url else "sqlite:///audit.db"
audit_logger = AuditLogger(db_url=audit_db_url)
# Dynamic exclusion of all SecretStr/SecretBytes fields for audit snapshot
secret_fields = {
f for f, info in cfg.__class__.model_fields.items()
if "Secret" in str(info.annotation)
}
# Log sanitized configuration snapshot
audit_logger.log_config_snapshot(
cfg.model_dump(
mode="json",
exclude={
"mt5_password",
"metaapi_token",
"metaapi_account_id",
"database_url",
"redis_url",
"telegram_token",
},
exclude=secret_fields,
)
)
audit_logger.log("system", "startup_initiated", f"Mode: {cfg.mode}, Algo: {cfg.algorithm}")
requirements-ci-no-talib.txt
+1 -1
View File
@@ -4,7 +4,7 @@
# ============================================================
# -- Core numeric / data -------------------------------------------
numpy==1.26.4
numpy==2.2.7
scipy==1.15.2
scikit-learn==1.6.1
pandas==2.2.3
requirements-ci.txt
+1 -1
View File
@@ -4,7 +4,7 @@
# ============================================================
# -- Core numeric / data -------------------------------------------
numpy==1.26.4
numpy==2.2.7
scipy==1.15.2
scikit-learn==1.6.1
pandas==2.2.3
requirements-docker.txt
+1 -1
View File
@@ -9,7 +9,7 @@ torch==2.5.1
torchvision==0.20.1
stable-baselines3==2.4.0
gymnasium==1.0.0
numpy==1.26.4
numpy==2.2.7
scipy==1.15.2
scikit-learn==1.6.1
pandas==2.2.3
requirements-linux.txt
+1 -1
View File
@@ -9,7 +9,7 @@ torch==2.5.1
torchvision==0.20.1
stable-baselines3==2.4.0
gymnasium==1.0.0
numpy==1.26.4
numpy==2.2.7
scipy==1.15.2
scikit-learn==1.6.1
pandas==2.2.3
requirements.txt
+1 -1
View File
@@ -9,7 +9,7 @@ torch==2.5.1
torchvision==0.20.1
stable-baselines3==2.4.0
gymnasium==1.0.0
numpy==1.26.4
numpy==2.2.7
scipy==1.15.2
scikit-learn==1.6.1
pandas==2.2.3
src/core/log_config.py
+39 -26
View File
@@ -5,7 +5,7 @@ src/core/log_config.py
from __future__ import annotations
import re
import contextlib
from typing import Any
from src.core.config import TradingConfig
@@ -34,32 +34,45 @@ class SecretMaskingProcessor:
self.update_secrets(config)
def update_secrets(self, config: TradingConfig) -> None:
"""Extract all SecretStr values from the config."""
# Use the class's model_fields to avoid Pydantic 2.11+ instance attribute warning
for field_name, field_info in config.__class__.model_fields.items():
# Check if SecretStr is in the type annotation
annotation_str = str(field_info.annotation)
if "SecretStr" in annotation_str:
val = getattr(config, field_name)
if hasattr(val, "get_secret_value"):
secret_val = val.get_secret_value()
if secret_val and len(secret_val) > 3:
self.secrets.add(secret_val)
elif isinstance(val, str) and val and len(val) > 3:
self.secrets.add(val)
"""
Extract all SecretStr/SecretBytes values from the config.
Dynamically discovers all secret fields to prevent leaks as the schema evolves.
"""
# Use the class's model_fields to avoid Pydantic instance attribute warnings
for field_name in config.__class__.model_fields:
val = getattr(config, field_name)
# Also specifically check database_url for embedded passwords
db_url = (
config.database_url.get_secret_value()
if hasattr(config.database_url, "get_secret_value")
else str(config.database_url)
)
if db_url and "@" in db_url:
# Mask the password part of the URL specifically
# postgresql://user:password@host:port/db
match = re.search(r"://([^:]+):([^@]+)@", db_url)
if match:
self.secrets.add(match.group(2))
# Extract the raw value if it's a Pydantic Secret type
secret_val = None
if hasattr(val, "get_secret_value"):
secret_val = val.get_secret_value()
elif "Secret" in str(type(val)):
# Defensive check for other secret-like types that might not have get_secret_value
with contextlib.suppress(AttributeError, TypeError):
secret_val = val.get_secret_value()
if secret_val is not None:
if isinstance(secret_val, bytes):
secret_val = secret_val.decode("utf-8", errors="replace")
if isinstance(secret_val, str) and len(secret_val) > 3:
self.secrets.add(secret_val)
# Generic URL credential extraction: protocol://user:password@host
# This protects embedded passwords in DATABASE_URL, REDIS_URL, etc.
for secret in list(self.secrets):
if isinstance(secret, str) and "://" in secret and "@" in secret:
try:
# Extract auth part (user:password) between :// and @
# Using rsplit for @ to handle cases where @ might be in password (escaped)
auth_part = secret.split("://", 1)[1].rsplit("@", 1)[0]
if ":" in auth_part:
# Password is the part after the LAST colon in the auth section
password = auth_part.rsplit(":", 1)[1]
if password and len(password) > 3:
self.secrets.add(password)
except (IndexError, ValueError):
continue
def redact_any(self, data: Any) -> Any:
"""
tests/test_secret_masking_hardening.py
+64
View File
@@ -0,0 +1,64 @@
import pytest
from pydantic import SecretStr
from src.core.config import TradingConfig
from src.core.log_config import SecretMaskingProcessor
def test_dynamic_secret_discovery():
"""Verify that all SecretStr fields are automatically discovered and masked."""
# Create a config with various secrets
config = TradingConfig(
mt5_password=SecretStr("super_secret_mt5"),
redis_url=SecretStr("redis://:redis_pass_123@localhost:6379/0"),
database_url=SecretStr("postgresql://user:db_pass_456@localhost:5432/db"),
telegram_token=SecretStr("bot123:tele_secret_789")
)
processor = SecretMaskingProcessor()
processor.update_secrets(config)
# Test individual secret extraction
assert "super_secret_mt5" in processor.secrets
assert "redis_pass_123" in processor.secrets
assert "db_pass_456" in processor.secrets
assert "bot123:tele_secret_789" in processor.secrets
# Test masking in a log message
log_event = {
"msg": "Failed to connect to redis with pass redis_pass_123",
"details": "Database error for db_pass_456",
"meta": {"token": "bot123:tele_secret_789"}
}
redacted = processor.redact_any(log_event)
assert redacted["msg"] == "Failed to connect to redis with pass [MASKED]"
assert redacted["details"] == "Database error for [MASKED]"
assert redacted["meta"]["token"] == "[MASKED]"
def test_url_password_extraction_robustness():
"""Test various URL formats for password extraction."""
processor = SecretMaskingProcessor()
# Mock secrets set
processor.secrets = {
"redis://user:pass1@host",
"postgresql://u:pass2@h:p/db",
"mongodb+srv://u:pass3@cluster.mongodb.net",
"http://admin:pass4@127.0.0.1:8080"
}
# Trigger extraction logic (normally inside update_secrets)
# We replicate the logic here or just call it if we can mock TradingConfig
for secret in list(processor.secrets):
if "://" in secret and "@" in secret:
auth_part = secret.split("://", 1)[1].rsplit("@", 1)[0]
if ":" in auth_part:
password = auth_part.rsplit(":", 1)[1]
if password and len(password) > 3:
processor.secrets.add(password)
assert "pass1" in processor.secrets
assert "pass2" in processor.secrets
assert "pass3" in processor.secrets
assert "pass4" in processor.secrets