Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions astrbot/core/utils/trace_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -490,6 +490,8 @@ def list_traces(
errors: bool = False,
models: bool = False,
plugins: bool = False,
start_time: int | None = None,
end_time: int | None = None,
) -> dict:
"""Read one numbered or keyset page, never all stored span bodies.

Expand All @@ -501,6 +503,8 @@ def list_traces(
errors: Require an error.
models: Require an Agent call.
plugins: Require a plugin call.
start_time: Inclusive minimum trace start time, in Unix seconds.
end_time: Inclusive maximum trace start time, in Unix seconds.

Returns:
Summaries, filtered count, next cursor, and storage health.
Expand All @@ -515,6 +519,12 @@ def list_traces(
if page < 1 or (cursor and page != 1):
raise ValueError("Invalid trace page.")
conditions, values = [], []
if start_time is not None:
conditions.append("start_ns >= ?")
values.append(start_time * 1_000_000_000)
if end_time is not None:
conditions.append("start_ns <= ?")
values.append(end_time * 1_000_000_000)
for key, enabled in (
("has_errors", errors),
("has_models", models),
Expand Down
7 changes: 6 additions & 1 deletion astrbot/dashboard/api/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,12 @@ def create_dashboard_asgi_app(
files=FileService(),
knowledge_bases=KnowledgeBaseService(core_lifecycle),
live_chat=LiveChatService(db, core_lifecycle),
logs=LogService(log_broker, core_lifecycle.astrbot_config, core=core_lifecycle),
logs=LogService(
log_broker,
core_lifecycle.astrbot_config,
core=core_lifecycle,
dashboard_static_folder=static_folder,
),
bots=BotConfigService(core_lifecycle),
platforms=PlatformService(core_lifecycle),
providers=ProviderConfigService(core_lifecycle),
Expand Down
9 changes: 8 additions & 1 deletion astrbot/dashboard/api/logs.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,11 +87,18 @@ async def live_logs(
@router.get("/logs/export")
async def export_logs(
redact: bool = Query(default=True),
start_time: int | None = Query(default=None, ge=0),
end_time: int | None = Query(default=None, ge=0),
_username: str = Depends(require_dashboard_user),
service: LogService = Depends(get_service),
):
try:
archive_path = await asyncio.to_thread(service.export_logs, redact=redact)
archive_path = await asyncio.to_thread(
service.export_logs,
redact=redact,
start_time=start_time,
end_time=end_time,
)
except LogServiceError as exc:
_raise_log_error(exc)
return FileResponse(
Expand Down
172 changes: 148 additions & 24 deletions astrbot/dashboard/services/diagnostic_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import platform
import re
import subprocess
from datetime import datetime
from pathlib import Path
from typing import Any, BinaryIO
from urllib.parse import quote, quote_plus
Expand All @@ -17,7 +18,7 @@

REDACTED = "[REDACTED]"
_SECRET_KEY = re.compile(
r"password|passwd|secret|token$|api[_-]?key|access[_-]?key|private[_-]?key|"
r"password|passwd|secret|token$|api[_-]?key|access[_-]?key|private[_-]?key|aes[_-]?key|"
r"authorization|cookie|credential|(?:client|app|signing|encrypt(?:ion)?)[_-]?key|"
r"recovery[_-]?code|(?:^|\.)key$|^headers$|^env$",
re.IGNORECASE,
Expand All @@ -26,13 +27,14 @@
_AUTH = re.compile(r"\b(Bearer|Basic|ApiKey)\s+[^\s\"'<>]+", re.IGNORECASE)
_COOKIE = re.compile(r"(\b(?:set-cookie|cookie)\s*:\s*)[^\r\n]+", re.IGNORECASE)
_URL_AUTH = re.compile(r"(https?://)[^/\s@]+@", re.IGNORECASE)
_ASSIGNMENT = re.compile(
r"((?:[\w.-]*(?:token|secret|password|passwd|api[_-]?key|access[_-]?key|"
r"(?:client|app|signing|encrypt(?:ion)?)[_-]?key|recovery[_-]?code)"
r"[\w.-]*|authorization|cookie|\bkey)[\"']?\s*[:=]\s*)"
r"(\"[^\"]*\"|'[^']*'|[^\s&,;}]+)",
_ASSIGNMENT = re.compile(r"(?<![\w.-])([\w.-]+)[\"']?\s*[:=]\s*")
_ASSIGNMENT_KEY = re.compile(
r"token|secret|password|passwd|api[_-]?key|access[_-]?key|aes[_-]?key|"
r"(?:client|app|signing|encrypt(?:ion)?)[_-]?key|recovery[_-]?code|"
r"(?:authorization|cookie|\bkey)$",
re.IGNORECASE,
)
_ASSIGNMENT_VALUE = re.compile(r"\"[^\"]*\"|'[^']*'|[^\s&,;}]+")
_TOKEN = re.compile(
r"\b(?:sk-[A-Za-z0-9_-]{16,}|gh[pousr]_[A-Za-z0-9_]{20,}|"
r"eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,})"
Expand All @@ -49,6 +51,10 @@
r")(?!\w|\.\w)"
)
MAX_LOG_LINE_BYTES = 1024 * 1024
_LOG_TIMESTAMP = re.compile(
r"^\[?(\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}"
r"(?:[.,]\d+)?(?:Z|[+-]\d{2}:?\d{2})?)(?:\]|\s|$)"
)


class DiagnosticRedactor:
Expand All @@ -57,6 +63,7 @@ class DiagnosticRedactor:
def __init__(self, *, enabled: bool = True) -> None:
self.enabled = enabled
self.secrets: set[str] = set()
self._sorted_secrets: list[str] | None = None
self.public_ips: dict[str, str] = {}

def collect(self, value: Any, sensitive: bool = False) -> None:
Expand All @@ -79,6 +86,7 @@ def collect(self, value: Any, sensitive: bool = False) -> None:
self.secrets.update(
(text, quote(text, safe=""), quote_plus(text), json.dumps(text)[1:-1])
)
self._sorted_secrets = None

def redact(self, value: Any, sensitive: bool = False) -> Any:
"""Return a sanitized copy while retaining containers and boolean flags.
Expand Down Expand Up @@ -118,15 +126,59 @@ def text(self, value: str) -> str:
"""
if not self.enabled:
return value
value = _ANSI.sub("", value)
value = _PRIVATE_KEY.sub(REDACTED, value)
for secret in sorted(self.secrets, key=len, reverse=True):
value = value.replace(secret, REDACTED)
value = _AUTH.sub(lambda match: f"{match[1]} {REDACTED}", value)
value = _COOKIE.sub(lambda match: f"{match[1]}{REDACTED}", value)
value = _URL_AUTH.sub(lambda match: f"{match[1]}{REDACTED}@", value)
value = _ASSIGNMENT.sub(lambda match: f"{match[1]}{REDACTED}", value)
value = _TOKEN.sub(REDACTED, value)
if "\x1b[" in value:
value = _ANSI.sub("", value)
if "PRIVATE KEY-----" in value:
value = _PRIVATE_KEY.sub(REDACTED, value)
if self.secrets:
if self._sorted_secrets is None:
self._sorted_secrets = sorted(self.secrets, key=len, reverse=True)
for secret in self._sorted_secrets:
value = value.replace(secret, REDACTED)
# These cheap substring checks avoid running every regex on ordinary
# ASCII log lines. Unicode case-insensitive regex matches can include
# characters that lower() does not normalize to the ASCII hint.
lowered = value.lower()
ascii_only = value.isascii()
if not ascii_only or any(
prefix in lowered for prefix in ("bearer", "basic", "apikey")
):
value = _AUTH.sub(lambda match: f"{match[1]} {REDACTED}", value)
if not ascii_only or "cookie" in lowered:
value = _COOKIE.sub(lambda match: f"{match[1]}{REDACTED}", value)
if "://" in value and "@" in value:
value = _URL_AUTH.sub(lambda match: f"{match[1]}{REDACTED}@", value)
if not ascii_only or any(
hint in lowered
for hint in (
"token",
"secret",
"password",
"passwd",
"key",
"code",
"authorization",
"cookie",
)
):
# Scan assignment prefixes without consuming ordinary values, which
# may themselves contain a nested credential assignment.
parts = []
end = 0
for match in _ASSIGNMENT.finditer(value):
if match.start() < end or not _ASSIGNMENT_KEY.search(match[1]):
continue
assigned = _ASSIGNMENT_VALUE.match(value, match.end())
if assigned is not None:
parts.extend((value[end : match.end()], REDACTED))
end = assigned.end()
parts.append(value[end:])
value = "".join(parts)
if any(
prefix in value
for prefix in ("sk-", "ghp_", "gho_", "ghu_", "ghs_", "ghr_", "eyJ")
):
value = _TOKEN.sub(REDACTED, value)
parts = []
end = 0
for match in _IP_LITERAL.finditer(value):
Expand All @@ -136,6 +188,12 @@ def text(self, value: str) -> str:
literal[1:-1] if literal.startswith("[") else literal.rstrip(".")
)
host, separator, scope = candidate.partition("%")
if (
":" in host
and "::" not in host
and host.count(":") != (6 if "." in host else 7)
):
continue
try:
address = ipaddress.ip_address(host)
except ValueError:
Expand All @@ -156,18 +214,29 @@ def text(self, value: str) -> str:
parts.append(value[end:])
return "".join(parts)

def copy_log(self, source: BinaryIO, target: BinaryIO, limit: int) -> bool:
def copy_log(
self,
source: BinaryIO,
target: BinaryIO,
limit: int,
*,
start_time: int | None = None,
end_time: int | None = None,
) -> bool:
"""Stream complete text lines, skipping oversized lines and private keys.

Args:
source: An already verified regular file, positioned at its start.
target: Open ZIP entry for redacted text.
limit: Maximum source bytes to consume, fixed before reading.
start_time: Inclusive minimum record time, in Unix seconds.
end_time: Inclusive maximum record time, in Unix seconds.

Returns:
Whether content was omitted to respect line or file limits.
"""
if not self.enabled:
filtering = start_time is not None or end_time is not None
if not self.enabled and not filtering:
remaining = limit
while remaining:
chunk = source.read(min(remaining, MAX_LOG_LINE_BYTES))
Expand All @@ -179,42 +248,96 @@ def copy_log(self, source: BinaryIO, target: BinaryIO, limit: int) -> bool:
truncated = False
oversized = False
private_key = False
pending: list[str] = []
pending_bytes = 0
remaining = limit
include_record = not filtering
while remaining:
line = source.readline(min(remaining, MAX_LOG_LINE_BYTES + 1))
if not line:
break
remaining -= len(line)
if filtering and not oversized and not private_key:
# Timestamp-free continuation lines inherit their record's
# selection; content before the first dated record is excluded.
prefix = line[:160].decode("utf-8", errors="replace")
match = _LOG_TIMESTAMP.match(_ANSI.sub("", prefix))
timestamp = None
dated_record = match is not None
try:
if match is not None:
timestamp = datetime.fromisoformat(
match[1].replace("Z", "+00:00").replace(",", ".")
).timestamp()
elif prefix.startswith("{"):
record = json.loads(line)
if isinstance(record, dict):
dated_record = True
if "time_unix_nano" in record:
timestamp = int(record["time_unix_nano"]) / 1e9
elif isinstance(record.get("time"), (int, float)):
timestamp = record["time"]
elif "start_time_unix_nano" in record:
timestamp = int(record["start_time_unix_nano"]) / 1e9
except (ValueError, TypeError, OverflowError, OSError):
pass
if dated_record:
include_record = (
timestamp is not None
and (start_time is None or timestamp >= start_time)
and (end_time is None or timestamp <= end_time)
)
complete = line.endswith(b"\n") or len(line) <= MAX_LOG_LINE_BYTES
if oversized or not complete:
oversized = not line.endswith(b"\n")
truncated = True
continue
if self.enabled:
truncated = True
continue
# Never emit a fragment cut at the export boundary: it could be a token.
if not remaining and not line.endswith(b"\n") and source.read(1):
truncated = True
break
if not self.enabled:
if include_record:
target.write(line)
continue
text = line.decode("utf-8", errors="replace")
if "-----BEGIN " in text and "PRIVATE KEY-----" in text:
private_key = True
target.write((REDACTED + "\n").encode())
if private_key:
private_key = not ("-----END " in text and "PRIVATE KEY-----" in text)
text = REDACTED + "\n"
elif private_key:
if "-----END " in text and "PRIVATE KEY-----" in text:
private_key = False
continue
target.write(self.text(text).encode("utf-8"))
if not include_record:
continue
pending.append(text)
pending_bytes += len(line)
# Keep complete lines together while amortizing regex and ZIP writes
# across a bounded batch instead of repeating them for every line.
if pending_bytes >= 64 * 1024:
target.write(self.text("".join(pending)).encode("utf-8"))
pending.clear()
pending_bytes = 0
if pending:
target.write(self.text("".join(pending)).encode("utf-8"))
return truncated


def collect_diagnostics(
config: dict, core: Any = None, *, redact: bool = True
config: dict,
core: Any = None,
*,
redact: bool = True,
webui_version: str | None = None,
) -> tuple[dict, DiagnosticRedactor]:
"""Snapshot configuration, plugin status and environment for a log export.

Args:
config: Running global configuration.
core: Running lifecycle, when metadata is available.
redact: Whether to mask credentials throughout the bundle.
webui_version: Version declared by the active WebUI assets, if available.

Returns:
JSON sections and a redactor seeded from all available credentials.
Expand Down Expand Up @@ -281,6 +404,7 @@ def collect_diagnostics(
pass
sections["environment.json"] = {
"astrbot_version": __version__,
"webui_version": webui_version,
"commit": commit,
"python_version": platform.python_version(),
"os": platform.system(),
Expand Down
Loading
Loading