Yijia-Xiao
2c97bad45c
The ticker symbol reaches three filesystem-path construction sites (load_ohlcv cache filename, checkpointer DB path, _log_state results directory) without validation. A value containing path separators or "../" escapes the configured cache / checkpoints / results directory. Two attack vectors: - Programmatic callers passing arbitrary ticker to propagate() - Prompt injection via fetched news content steering the LLM into tool calls with attacker-chosen ticker Fix: new safe_ticker_component() validator in tradingagents/dataflows/ utils.py applied at all three sites. Allows the standard ticker character set ([A-Za-z0-9._\-\^], up to 32 chars) and explicitly rejects dot-only values like "." and ".." which would otherwise pass the regex but traverse parent directories. Seven test cases cover the accepted formats (BRK-B, 7203.T, ^GSPC, etc.) and the rejected inputs (path separators, null bytes, whitespace, empty values, overlong strings, dot-only values). Closes #618.
75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
import os
|
|
import re
|
|
import json
|
|
import pandas as pd
|
|
from datetime import date, timedelta, datetime
|
|
from typing import Annotated
|
|
|
|
SavePathType = Annotated[str, "File path to save data. If None, data is not saved."]
|
|
|
|
# Tickers can contain letters, digits, dot, dash, underscore, and caret
|
|
# (for index symbols like ^GSPC). Anything else is rejected so the value
|
|
# never escapes a containing directory when interpolated into a path.
|
|
_TICKER_PATH_RE = re.compile(r"^[A-Za-z0-9._\-\^]+$")
|
|
|
|
|
|
def safe_ticker_component(value: str, *, max_len: int = 32) -> str:
|
|
"""Validate ``value`` is safe to interpolate into a filesystem path.
|
|
|
|
Tickers come from user CLI input or from LLM tool calls, both of which
|
|
can be influenced by attacker-controlled content (e.g. prompt injection
|
|
embedded in fetched news). Without validation, a value like
|
|
``"../../../etc/foo"`` flows into ``os.path.join`` / ``Path /`` and
|
|
escapes the configured cache, checkpoint, or results directory.
|
|
|
|
Returns ``value`` unchanged when it matches the allowed pattern; raises
|
|
``ValueError`` otherwise.
|
|
"""
|
|
if not isinstance(value, str) or not value:
|
|
raise ValueError(f"ticker must be a non-empty string, got {value!r}")
|
|
if len(value) > max_len:
|
|
raise ValueError(f"ticker exceeds {max_len} chars: {value!r}")
|
|
if not _TICKER_PATH_RE.fullmatch(value):
|
|
raise ValueError(
|
|
f"ticker contains characters not allowed in a filesystem path: {value!r}"
|
|
)
|
|
# The regex above allows '.', so values like '.', '..', '...' would pass,
|
|
# and as a path component they traverse the parent directory. Reject any
|
|
# value that's only dots.
|
|
if set(value) == {"."}:
|
|
raise ValueError(f"ticker cannot consist solely of dots: {value!r}")
|
|
return value
|
|
|
|
|
|
def save_output(data: pd.DataFrame, tag: str, save_path: SavePathType = None) -> None:
|
|
if save_path:
|
|
data.to_csv(save_path, encoding="utf-8")
|
|
print(f"{tag} saved to {save_path}")
|
|
|
|
|
|
def get_current_date():
|
|
return date.today().strftime("%Y-%m-%d")
|
|
|
|
|
|
def decorate_all_methods(decorator):
|
|
def class_decorator(cls):
|
|
for attr_name, attr_value in cls.__dict__.items():
|
|
if callable(attr_value):
|
|
setattr(cls, attr_name, decorator(attr_value))
|
|
return cls
|
|
|
|
return class_decorator
|
|
|
|
|
|
def get_next_weekday(date):
|
|
|
|
if not isinstance(date, datetime):
|
|
date = datetime.strptime(date, "%Y-%m-%d")
|
|
|
|
if date.weekday() >= 5:
|
|
days_to_add = 7 - date.weekday()
|
|
next_weekday = date + timedelta(days=days_to_add)
|
|
return next_weekday
|
|
else:
|
|
return date
|