1362 lines
49 KiB
Python
Executable File
1362 lines
49 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Run a PII-safe live audit of every read-only Chat Shortcut.
|
||
|
||
The audit exercises the real Shortcut entry point and, when DWS_DUMP_RAW is
|
||
enabled by the locally built binary, captures the lower MCP text response from
|
||
the same invocation. Raw business responses are kept in memory only. The
|
||
written JSON/Markdown reports contain shapes, counts, tool names, and redacted
|
||
error facts, but no message text, user names, IDs, or invite/download URLs.
|
||
|
||
This is deliberately a live-account audit, not a mock/Help contract check.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import signal
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(SCRIPT_DIR))
|
||
|
||
from shortcut_real_result import count_projection_items, payload_indicates_error # noqa: E402
|
||
|
||
|
||
READ_SHORTCUTS = (
|
||
"+at-me",
|
||
"+bot-find",
|
||
"+bot-search",
|
||
"+category-list",
|
||
"+category-list-conversations",
|
||
"+chat-bots",
|
||
"+chat-get-by-id",
|
||
"+chat-invite-url",
|
||
"+chat-list-all",
|
||
"+chat-list-join-requests",
|
||
"+chat-list-mine",
|
||
"+chat-members-list",
|
||
"+chat-members-get",
|
||
"+chat-messages",
|
||
"+chat-role-list",
|
||
"+chat-role-query-user",
|
||
"+chat-search",
|
||
"+conversation-info",
|
||
"+conversation-list",
|
||
"+conversation-list-top",
|
||
"+group-members",
|
||
"+messages-list",
|
||
"+messages-list-direct",
|
||
"+messages-list-pin",
|
||
"+messages-list-unread-conversations",
|
||
"+messages-mget",
|
||
"+messages-query-send-status",
|
||
"+messages-read-status",
|
||
"+messages-resource-download",
|
||
"+messages-resource-url",
|
||
"+my-groups",
|
||
"+search-msg",
|
||
"+thread-replies",
|
||
"+unread-chats",
|
||
)
|
||
|
||
LIST_PATHS_BY_TOOL = {
|
||
"search_at_me_message": (
|
||
("result", "conversationMessagesList", "*", "messages"),
|
||
),
|
||
"search_bots": (("result", "bots"), ("result", "list")),
|
||
"search_my_robots": (("result", "robots"), ("result", "list")),
|
||
"list_user_define_conv_categories": (
|
||
("result", "categories"),
|
||
("result", "categoryList"),
|
||
),
|
||
"list_conversations_by_category": (
|
||
("result", "conversations"),
|
||
("result", "conversationList"),
|
||
),
|
||
"list_group_bots": (("result", "bots"), ("result", "robots")),
|
||
"list_my_groups_pagination": (("result", "groups"),),
|
||
"list_owned_or_admin_groups": (("result", "groups"),),
|
||
"list_group_member_by_ids": (
|
||
("result", "members"),
|
||
("result", "list"),
|
||
),
|
||
"list_custom_group_roles": (("result", "roles"), ("result", "list")),
|
||
"list_all_conversations": (
|
||
("result", "conversations"),
|
||
("result", "conversationList"),
|
||
),
|
||
"list_top_conversations": (
|
||
("result", "conversations"),
|
||
("result", "conversationList"),
|
||
),
|
||
"get_group_members": (("result", "list"), ("result", "members")),
|
||
"list_conversation_message_v2": (
|
||
("result", "messages"),
|
||
("result", "messageList"),
|
||
),
|
||
"list_individual_chat_message": (
|
||
("result", "messages"),
|
||
("result", "messageList"),
|
||
),
|
||
"list_pin_messages": (
|
||
("result", "messages"),
|
||
("result", "pinMessages"),
|
||
("result", "list"),
|
||
),
|
||
"unread_message_conversation_list": (
|
||
("result", "conversations"),
|
||
("result", "conversationList"),
|
||
),
|
||
"list_messages_by_ids": (("result", "messages"), ("result", "list")),
|
||
"search_messages_by_keyword": (
|
||
("result", "messages"),
|
||
("result", "messageList"),
|
||
("result", "list"),
|
||
),
|
||
"list_topic_replies": (
|
||
("result", "messages"),
|
||
("result", "replies"),
|
||
("result", "list"),
|
||
),
|
||
}
|
||
|
||
@dataclass
|
||
class Capture:
|
||
command: str
|
||
argv: list[str]
|
||
exit_code: int | None
|
||
stdout: str
|
||
stderr: str
|
||
raw_calls: list[tuple[str, str, Any]]
|
||
duration_ms: int
|
||
timed_out: bool = False
|
||
|
||
def upper(self) -> Any:
|
||
try:
|
||
return json.loads(self.stdout)
|
||
except (json.JSONDecodeError, TypeError):
|
||
return None
|
||
|
||
|
||
def parse_raw_calls(stderr: str) -> tuple[list[tuple[str, str, Any]], str]:
|
||
raw_calls: list[tuple[str, str, Any]] = []
|
||
clean_lines: list[str] = []
|
||
for line in (stderr or "").splitlines():
|
||
if not line.startswith("DWSRAW\t"):
|
||
clean_lines.append(line)
|
||
continue
|
||
fields = line.split("\t", 3)
|
||
if len(fields) != 4:
|
||
continue
|
||
try:
|
||
raw_calls.append((fields[1], fields[2], json.loads(fields[3])))
|
||
except json.JSONDecodeError:
|
||
raw_calls.append((fields[1], fields[2], fields[3]))
|
||
return raw_calls, "\n".join(clean_lines)
|
||
|
||
|
||
def run_capture(binary: Path, command: str, args: list[str], timeout: int) -> Capture:
|
||
argv = [str(binary), "chat", command, *args, "--format", "json"]
|
||
env = os.environ.copy()
|
||
env["DWS_DUMP_RAW"] = "1"
|
||
started = time.monotonic()
|
||
proc = subprocess.Popen(
|
||
argv,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
env=env,
|
||
start_new_session=True,
|
||
)
|
||
try:
|
||
stdout, stderr = proc.communicate(timeout=timeout)
|
||
raw_calls, clean_stderr = parse_raw_calls(stderr)
|
||
return Capture(
|
||
command=command,
|
||
argv=argv,
|
||
exit_code=proc.returncode,
|
||
stdout=stdout,
|
||
stderr=clean_stderr,
|
||
raw_calls=raw_calls,
|
||
duration_ms=round((time.monotonic() - started) * 1000),
|
||
)
|
||
except subprocess.TimeoutExpired as exc:
|
||
# dws may have transport children which inherit its output pipes.
|
||
# Killing only the immediate process leaves those descendants alive and
|
||
# communicate() can block forever despite the advertised timeout.
|
||
os.killpg(proc.pid, signal.SIGKILL)
|
||
stdout, stderr = proc.communicate()
|
||
raw_calls, clean_stderr = parse_raw_calls(stderr)
|
||
return Capture(
|
||
command=command,
|
||
argv=argv,
|
||
exit_code=None,
|
||
stdout=stdout,
|
||
stderr=clean_stderr,
|
||
raw_calls=raw_calls,
|
||
duration_ms=round((time.monotonic() - started) * 1000),
|
||
timed_out=True,
|
||
)
|
||
|
||
|
||
def at_path(value: Any, path: tuple[str, ...]) -> list[Any] | None:
|
||
current = value
|
||
for index, part in enumerate(path):
|
||
if part == "*":
|
||
if not isinstance(current, list):
|
||
return None
|
||
remainder = path[index + 1 :]
|
||
combined: list[Any] = []
|
||
for item in current:
|
||
found = at_path(item, remainder)
|
||
if isinstance(found, list):
|
||
combined.extend(found)
|
||
return combined
|
||
if not isinstance(current, dict) or part not in current:
|
||
return None
|
||
current = current[part]
|
||
return current if isinstance(current, list) else None
|
||
|
||
|
||
def lower_count(tool: str, raw: Any) -> int | None:
|
||
for path in LIST_PATHS_BY_TOOL.get(tool, ()):
|
||
found = at_path(raw, path)
|
||
if found is not None:
|
||
return len(found)
|
||
return None
|
||
|
||
|
||
def distinct_lower_message_count(
|
||
raw_calls: list[tuple[str, str, Any]], tool: str
|
||
) -> int | None:
|
||
"""Count unique messages across all pages without retaining their IDs.
|
||
|
||
Time-boundary pagination may repeat the boundary message on the following
|
||
page. Summing page lengths would therefore report a false projection
|
||
mismatch for an otherwise correct de-duplicating Shortcut.
|
||
"""
|
||
metrics = distinct_lower_message_metrics(raw_calls, tool)
|
||
return metrics[0] if metrics is not None else None
|
||
|
||
|
||
def distinct_lower_message_metrics(
|
||
raw_calls: list[tuple[str, str, Any]], tool: str
|
||
) -> tuple[int, int, int] | None:
|
||
"""Aggregate count/reaction/thread facts over one stable message-ID set."""
|
||
seen: set[str] = set()
|
||
reaction_messages: set[str] = set()
|
||
thread_messages: set[str] = set()
|
||
anonymous = 0
|
||
found_collection = False
|
||
identity_keys = ("openMessageId", "openMsgId", "messageId", "msgId", "id")
|
||
for _, call_tool, raw in raw_calls:
|
||
if call_tool != tool:
|
||
continue
|
||
rows: list[Any] | None = None
|
||
for path in LIST_PATHS_BY_TOOL.get(tool, ()):
|
||
rows = at_path(raw, path)
|
||
if rows is not None:
|
||
break
|
||
if rows is None:
|
||
continue
|
||
found_collection = True
|
||
for row in rows:
|
||
identity = ""
|
||
if isinstance(row, dict):
|
||
for key in identity_keys:
|
||
value = row.get(key)
|
||
if value not in (None, ""):
|
||
identity = f"id:{value}"
|
||
break
|
||
if not identity:
|
||
identity = f"anonymous:{anonymous}"
|
||
anonymous += 1
|
||
seen.add(identity)
|
||
if reaction_message_count(row, upper=False) > 0:
|
||
reaction_messages.add(identity)
|
||
if thread_message_count(row, upper=False) > 0:
|
||
thread_messages.add(identity)
|
||
if not found_collection:
|
||
return None
|
||
return len(seen), len(reaction_messages), len(thread_messages)
|
||
|
||
|
||
def distinct_lower_collection_count(
|
||
raw_calls: list[tuple[str, str, Any]],
|
||
tool: str,
|
||
identity_keys: tuple[str, ...],
|
||
) -> int | None:
|
||
"""Count unique rows across every lower page for one tool."""
|
||
seen: set[str] = set()
|
||
anonymous = 0
|
||
found_collection = False
|
||
for _, call_tool, raw in raw_calls:
|
||
if call_tool != tool:
|
||
continue
|
||
rows: list[Any] | None = None
|
||
for path in LIST_PATHS_BY_TOOL.get(tool, ()):
|
||
rows = at_path(raw, path)
|
||
if rows is not None:
|
||
break
|
||
if rows is None:
|
||
continue
|
||
found_collection = True
|
||
for row in rows:
|
||
if not isinstance(row, dict):
|
||
anonymous += 1
|
||
continue
|
||
identity = ""
|
||
for key in identity_keys:
|
||
value = row.get(key)
|
||
if value not in (None, ""):
|
||
identity = str(value)
|
||
break
|
||
if identity:
|
||
seen.add(identity)
|
||
else:
|
||
anonymous += 1
|
||
if not found_collection:
|
||
return None
|
||
return len(seen) + anonymous
|
||
|
||
|
||
def response_is_error(value: Any) -> bool:
|
||
if not isinstance(value, dict):
|
||
return False
|
||
if value.get("success") is False:
|
||
return True
|
||
if str(value.get("status") or "").lower() == "error":
|
||
return True
|
||
for key in ("error", "errorCode", "error_code", "code"):
|
||
item = value.get(key)
|
||
if item not in (None, False, "", 0, "0", "ok", "success", "succeed"):
|
||
return True
|
||
return False
|
||
|
||
|
||
def error_facts(stderr: str) -> dict[str, Any]:
|
||
facts: dict[str, Any] = {}
|
||
try:
|
||
value = json.loads(stderr)
|
||
error = value.get("error", value) if isinstance(value, dict) else {}
|
||
if isinstance(error, dict):
|
||
for key in ("category", "code", "reason", "server_error_code", "server_key"):
|
||
if key in error:
|
||
facts[key] = error[key]
|
||
message = str(error.get("message") or "")
|
||
else:
|
||
message = stderr
|
||
except json.JSONDecodeError:
|
||
message = stderr
|
||
lowered = message.lower()
|
||
for marker, label in (
|
||
("permission", "permission"),
|
||
("not found", "not_found"),
|
||
("不存在", "not_found"),
|
||
("invalid", "invalid_argument"),
|
||
("必填", "missing_argument"),
|
||
("超时", "timeout"),
|
||
):
|
||
if marker in lowered:
|
||
facts["message_class"] = label
|
||
break
|
||
return facts
|
||
|
||
|
||
def shape(value: Any) -> dict[str, Any]:
|
||
if isinstance(value, dict):
|
||
return {"type": "object", "keys": sorted(value)}
|
||
if isinstance(value, list):
|
||
return {"type": "array", "length": len(value)}
|
||
if value is None:
|
||
return {"type": "none"}
|
||
return {"type": type(value).__name__}
|
||
|
||
|
||
def reaction_message_count(value: Any, *, upper: bool) -> int:
|
||
"""Count reaction-bearing message objects without retaining business values."""
|
||
count = 0
|
||
if isinstance(value, dict):
|
||
keys = ("reactions", "emotionReplyList", "reactionList") if upper else (
|
||
"emotionReplyList",
|
||
"reactionList",
|
||
"reactions",
|
||
)
|
||
if any(value.get(key) not in (None, [], {}) for key in keys):
|
||
count += 1
|
||
for item in value.values():
|
||
count += reaction_message_count(item, upper=upper)
|
||
elif isinstance(value, list):
|
||
for item in value:
|
||
count += reaction_message_count(item, upper=upper)
|
||
return count
|
||
|
||
|
||
def primary_projection_for_message_metrics(value: Any) -> Any:
|
||
"""Select one canonical projected collection for message-level metrics.
|
||
|
||
Some stable Shortcut payloads publish compatibility aliases such as
|
||
``messages`` and ``items`` with the same rows. Walking the entire envelope
|
||
would count reactions and thread IDs once per alias and report a false
|
||
projection mismatch even though the lower and upper message sets agree.
|
||
"""
|
||
if not isinstance(value, dict):
|
||
return value
|
||
for key in ("messages", "replies", "results", "items"):
|
||
items = value.get(key)
|
||
if isinstance(items, list):
|
||
return items
|
||
return value
|
||
|
||
|
||
def thread_message_count(value: Any, *, upper: bool) -> int:
|
||
keys = {"threadId"} if upper else {
|
||
"openConvThreadId",
|
||
"openConversationThreadId",
|
||
"threadId",
|
||
"topicId",
|
||
}
|
||
count = 0
|
||
if isinstance(value, dict):
|
||
if any(value.get(key) not in (None, "") for key in keys):
|
||
count += 1
|
||
for item in value.values():
|
||
count += thread_message_count(item, upper=upper)
|
||
elif isinstance(value, list):
|
||
for item in value:
|
||
count += thread_message_count(item, upper=upper)
|
||
return count
|
||
|
||
|
||
def pagination_meta(value: Any) -> dict[str, Any]:
|
||
"""Extract only completeness facts; never retain cursor values."""
|
||
if isinstance(value, dict):
|
||
has_more = None
|
||
for key in ("hasMore", "has_more"):
|
||
if key in value and isinstance(value[key], bool):
|
||
has_more = value[key]
|
||
break
|
||
cursor_present = any(
|
||
value.get(key) not in (None, "", 0, "0")
|
||
for key in ("nextCursor", "next_cursor", "nextToken", "next_token", "pageToken", "page_token")
|
||
)
|
||
next_page = value.get("nextPage")
|
||
resume_present = cursor_present or (
|
||
isinstance(next_page, dict)
|
||
and next_page.get("time") not in (None, "")
|
||
)
|
||
if has_more is not None or resume_present:
|
||
return {
|
||
"has_more": has_more,
|
||
"next_cursor_present": cursor_present,
|
||
"resume_present": resume_present,
|
||
}
|
||
for item in value.values():
|
||
found = pagination_meta(item)
|
||
if found:
|
||
return found
|
||
elif isinstance(value, list):
|
||
for item in value:
|
||
found = pagination_meta(item)
|
||
if found:
|
||
return found
|
||
return {}
|
||
|
||
|
||
def summarize_capture(capture: Capture) -> dict[str, Any]:
|
||
upper = capture.upper()
|
||
raw = capture.raw_calls[-1][2] if capture.raw_calls else None
|
||
tool = capture.raw_calls[-1][1] if capture.raw_calls else None
|
||
upper_count = count_projection_items(capture.stdout)
|
||
expected_lower_count = lower_count(tool, raw) if tool else None
|
||
pagination_raw = raw
|
||
lower_message_projection = raw
|
||
lower_message_metrics: tuple[int, int, int] | None = None
|
||
if capture.command == "+thread-replies" and tool == "list_topic_replies":
|
||
lower_message_metrics = distinct_lower_message_metrics(
|
||
capture.raw_calls, "list_topic_replies"
|
||
)
|
||
expected_lower_count = lower_message_metrics[0] if lower_message_metrics else None
|
||
elif capture.command == "+at-me" and tool == "search_at_me_message":
|
||
lower_message_metrics = distinct_lower_message_metrics(
|
||
capture.raw_calls, "search_at_me_message"
|
||
)
|
||
expected_lower_count = lower_message_metrics[0] if lower_message_metrics else None
|
||
elif (
|
||
capture.command == "+messages-list-direct"
|
||
and tool == "list_individual_chat_message"
|
||
):
|
||
lower_message_metrics = distinct_lower_message_metrics(
|
||
capture.raw_calls, "list_individual_chat_message"
|
||
)
|
||
expected_lower_count = lower_message_metrics[0] if lower_message_metrics else None
|
||
elif capture.command in {"+chat-list-all", "+my-groups"} and tool == (
|
||
"list_my_groups_pagination"
|
||
):
|
||
expected_lower_count = distinct_lower_collection_count(
|
||
capture.raw_calls,
|
||
"list_my_groups_pagination",
|
||
(
|
||
"openConversationId",
|
||
"openconversation_id",
|
||
"conversationId",
|
||
"openCid",
|
||
"cid",
|
||
"id",
|
||
),
|
||
)
|
||
if capture.command == "+chat-members-list":
|
||
counts = upper.get("counts", {}) if isinstance(upper, dict) else {}
|
||
upper_count = counts.get("total") if isinstance(counts, dict) else None
|
||
expected_lower_count = 0
|
||
for _, call_tool, call_raw in capture.raw_calls:
|
||
if call_tool in {"get_group_members", "list_group_bots"}:
|
||
item_count = lower_count(call_tool, call_raw)
|
||
if item_count is not None:
|
||
expected_lower_count += item_count
|
||
if call_tool == "get_group_members":
|
||
pagination_raw = call_raw
|
||
elif capture.command == "+messages-resource-download":
|
||
upper_count = (
|
||
1
|
||
if isinstance(upper, dict)
|
||
and isinstance(upper.get("sizeBytes"), (int, float))
|
||
and upper.get("sizeBytes") >= 0
|
||
and bool(upper.get("localPath"))
|
||
else 0
|
||
)
|
||
expected_lower_count = (
|
||
1
|
||
if first_string(raw, "resourceUrl", "downloadUrl")
|
||
else 0
|
||
)
|
||
exact_passthrough = bool(capture.raw_calls and upper == raw)
|
||
upper_message_projection = primary_projection_for_message_metrics(upper)
|
||
upper_reactions = reaction_message_count(upper_message_projection, upper=True)
|
||
upper_threads = thread_message_count(upper_message_projection, upper=True)
|
||
if lower_message_metrics is not None:
|
||
_, lower_reactions, lower_threads = lower_message_metrics
|
||
else:
|
||
lower_reactions = reaction_message_count(lower_message_projection, upper=False)
|
||
lower_threads = thread_message_count(lower_message_projection, upper=False)
|
||
upper_pagination = pagination_meta(upper)
|
||
lower_pagination = pagination_meta(pagination_raw)
|
||
|
||
upper_explicitly_incomplete = (
|
||
isinstance(upper, dict)
|
||
and upper.get("complete") is False
|
||
and (
|
||
upper.get("hasMore") is True
|
||
or bool(upper.get("failures"))
|
||
or bool(upper.get("truncatedByPageLimit"))
|
||
or bool(upper.get("truncatedByResultLimit"))
|
||
)
|
||
)
|
||
|
||
if capture.timed_out:
|
||
status = "timeout"
|
||
elif capture.exit_code != 0:
|
||
status = "backend_error" if capture.stderr else "cli_error"
|
||
elif payload_indicates_error(capture.stdout) or response_is_error(raw):
|
||
status = "error_envelope"
|
||
elif not capture.raw_calls:
|
||
status = "no_lower_capture"
|
||
elif upper_explicitly_incomplete:
|
||
status = "incomplete"
|
||
elif expected_lower_count == 0 and upper_count in (None, 0):
|
||
# An empty collection is useful evidence, but it is not a successful
|
||
# non-empty business-data E2E. This must precede exact-passthrough so a
|
||
# raw empty list cannot be promoted to pass merely because projection
|
||
# preserved it byte-for-byte.
|
||
status = "pass_empty"
|
||
elif expected_lower_count is not None and upper_count != expected_lower_count:
|
||
status = "projection_mismatch"
|
||
elif upper_count == 0:
|
||
# The upper projection is empty but the lower response shape does not
|
||
# expose a countable collection, so correctness cannot be proven.
|
||
status = "projection_unverified"
|
||
elif exact_passthrough:
|
||
status = "pass"
|
||
elif lower_reactions > 0 and upper_reactions != lower_reactions:
|
||
status = "projection_mismatch"
|
||
elif lower_threads > 0 and upper_threads != lower_threads:
|
||
status = "projection_mismatch"
|
||
elif lower_pagination.get("has_more") is True and (
|
||
upper_pagination.get("has_more") is not True
|
||
or not upper_pagination.get("resume_present")
|
||
):
|
||
status = "pagination_mismatch"
|
||
else:
|
||
status = "pass"
|
||
|
||
return {
|
||
"command": capture.command,
|
||
"status": status,
|
||
"exit_code": capture.exit_code,
|
||
"duration_ms": capture.duration_ms,
|
||
"upper": {
|
||
**shape(upper),
|
||
"count": upper_count,
|
||
"reaction_message_count": upper_reactions,
|
||
"thread_message_count": upper_threads,
|
||
"pagination": upper_pagination,
|
||
},
|
||
"lower": {
|
||
"call_count": len(capture.raw_calls),
|
||
"tools": [f"{server}/{name}" for server, name, _ in capture.raw_calls],
|
||
"shape": shape(raw),
|
||
"count": expected_lower_count,
|
||
"reaction_message_count": lower_reactions,
|
||
"thread_message_count": lower_threads,
|
||
"pagination": lower_pagination,
|
||
},
|
||
"exact_passthrough": exact_passthrough,
|
||
"error": error_facts(capture.stderr) if capture.stderr else {},
|
||
}
|
||
|
||
|
||
def blocked(command: str, requirement: str) -> dict[str, Any]:
|
||
return {
|
||
"command": command,
|
||
"status": "fixture_blocked",
|
||
"requirement": requirement,
|
||
}
|
||
|
||
|
||
def dig_values(value: Any, keys: set[str]) -> list[Any]:
|
||
found: list[Any] = []
|
||
if isinstance(value, dict):
|
||
for key, item in value.items():
|
||
if key in keys and item not in (None, "", [], {}):
|
||
found.append(item)
|
||
found.extend(dig_values(item, keys))
|
||
elif isinstance(value, list):
|
||
for item in value:
|
||
found.extend(dig_values(item, keys))
|
||
elif isinstance(value, str):
|
||
stripped = value.strip()
|
||
if stripped.startswith(("{", "[")):
|
||
try:
|
||
embedded = json.loads(stripped)
|
||
except json.JSONDecodeError:
|
||
embedded = None
|
||
if isinstance(embedded, (dict, list)):
|
||
found.extend(dig_values(embedded, keys))
|
||
return found
|
||
|
||
|
||
def first_string(value: Any, *keys: str) -> str:
|
||
for item in dig_values(value, set(keys)):
|
||
if isinstance(item, str) and item.strip():
|
||
return item.strip()
|
||
if isinstance(item, (int, float)) and not isinstance(item, bool):
|
||
return str(int(item))
|
||
return ""
|
||
|
||
|
||
def first_list(value: Any, *keys: str) -> list[Any]:
|
||
for item in dig_values(value, set(keys)):
|
||
if isinstance(item, list):
|
||
return item
|
||
return []
|
||
|
||
|
||
def safe_keyword(messages: list[Any]) -> str:
|
||
for message in messages:
|
||
if not isinstance(message, dict):
|
||
continue
|
||
text = message.get("text")
|
||
if not isinstance(text, str):
|
||
continue
|
||
for token in re.findall(r"[\u4e00-\u9fff]{2,6}|[A-Za-z0-9]{3,12}", text):
|
||
if token:
|
||
return token
|
||
return "工作"
|
||
|
||
|
||
def group_candidates(capture: Capture) -> list[dict[str, Any]]:
|
||
raw = capture.raw_calls[-1][2] if capture.raw_calls else {}
|
||
groups = first_list(raw, "groups")
|
||
valid = [item for item in groups if isinstance(item, dict)]
|
||
priority = {"OWNER": 0, "ADMIN": 1}
|
||
return sorted(valid, key=lambda item: priority.get(str(item.get("myRole")), 2))
|
||
|
||
|
||
def merge_group_candidates(
|
||
groups_all: Capture, groups_mine: Capture
|
||
) -> list[dict[str, Any]]:
|
||
all_groups = group_candidates(groups_all)
|
||
mine_groups = group_candidates(groups_mine)
|
||
ordered: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for item in mine_groups:
|
||
group_id = str(item.get("openConversationId") or "")
|
||
if not group_id or group_id in seen:
|
||
continue
|
||
# The owned/admin endpoint carries myRole and ownerOpenDingtalkId,
|
||
# which are needed to locate a message sent by the current user. Prefer
|
||
# that richer record; the all-groups page is only a fallback.
|
||
ordered.append(item)
|
||
seen.add(group_id)
|
||
for item in all_groups:
|
||
group_id = str(item.get("openConversationId") or "")
|
||
if group_id and group_id not in seen:
|
||
ordered.append(item)
|
||
seen.add(group_id)
|
||
return ordered
|
||
|
||
|
||
def live_audit(
|
||
binary: Path, timeout: int, max_group_probes: int, max_member_probes: int
|
||
) -> list[dict[str, Any]]:
|
||
results: dict[str, dict[str, Any]] = {}
|
||
captures: dict[str, Capture] = {}
|
||
group_override = os.environ.get("DWS_AUDIT_GROUP_ID", "").strip()
|
||
numeric_group_override = os.environ.get("DWS_AUDIT_NUMERIC_GROUP_ID", "").strip()
|
||
open_task_override = os.environ.get("DWS_AUDIT_OPEN_TASK_ID", "").strip()
|
||
topic_group_override = os.environ.get("DWS_AUDIT_TOPIC_GROUP_ID", "").strip()
|
||
topic_id_override = os.environ.get("DWS_AUDIT_TOPIC_ID", "").strip()
|
||
topic_page_size = os.environ.get("DWS_AUDIT_TOPIC_PAGE_SIZE", "").strip()
|
||
media_group_override = os.environ.get("DWS_AUDIT_MEDIA_GROUP_ID", "").strip()
|
||
media_message_override = os.environ.get("DWS_AUDIT_MEDIA_MESSAGE_ID", "").strip()
|
||
media_resource_override = os.environ.get("DWS_AUDIT_MEDIA_RESOURCE_ID", "").strip()
|
||
|
||
def run(command: str, *args: str, retries: int = 0) -> Capture:
|
||
capture = run_capture(binary, command, list(args), timeout)
|
||
for _ in range(retries):
|
||
if summarize_capture(capture)["status"] not in {
|
||
"backend_error",
|
||
"cli_error",
|
||
"error_envelope",
|
||
"timeout",
|
||
"no_lower_capture",
|
||
}:
|
||
break
|
||
capture = run_capture(binary, command, list(args), timeout)
|
||
captures[command] = capture
|
||
results[command] = summarize_capture(capture)
|
||
return capture
|
||
|
||
category = run("+category-list")
|
||
groups_all = run(
|
||
"+chat-list-all", "--limit", "200", "--page-all", "--page-limit", "50"
|
||
)
|
||
groups_mine = run("+chat-list-mine", "--limit", "200")
|
||
run("+conversation-list", "--limit", "100")
|
||
run("+conversation-list-top", "--limit", "100")
|
||
run("+messages-list-unread-conversations", "--count", "100")
|
||
run("+unread-chats", "--count", "100")
|
||
run("+chat-list-join-requests", "--limit", "50")
|
||
run("+at-me", "--days", "30", "--page-all", "--page-limit", "50")
|
||
run("+bot-search", "--page", "1", "--size", "100")
|
||
run("+bot-find", "--query", "机器人", "--limit", "20")
|
||
run("+my-groups", "--page-all", "--page-limit", "50")
|
||
|
||
category_id = first_string(category.upper(), "categoryId", "category_id")
|
||
if category_id:
|
||
run("+category-list-conversations", "--category-id", category_id)
|
||
else:
|
||
results["+category-list-conversations"] = blocked(
|
||
"+category-list-conversations",
|
||
"账号下需要至少一个真实会话分组 categoryId",
|
||
)
|
||
|
||
candidates = merge_group_candidates(groups_all, groups_mine)
|
||
group = candidates[0] if candidates else {}
|
||
messages: Capture | None = None
|
||
best_group: dict[str, Any] = {}
|
||
best_score = -1
|
||
# A valid but quiet first group is not sufficient evidence for the
|
||
# message-dependent Shortcuts. Walk real joined groups until one produces a
|
||
# non-empty message page. Prefer a page that also supplies downstream
|
||
# fixtures (a message sent by the current user, media, or a thread ID).
|
||
# This prevents the audit from stopping at the first technically valid but
|
||
# fixture-poor group.
|
||
probe_candidates = candidates[:max_group_probes]
|
||
if group_override:
|
||
forced = next(
|
||
(
|
||
candidate
|
||
for candidate in candidates
|
||
if str(candidate.get("openConversationId") or "") == group_override
|
||
),
|
||
{"openConversationId": group_override},
|
||
)
|
||
probe_candidates = [forced]
|
||
for candidate in probe_candidates:
|
||
candidate_id = str(candidate.get("openConversationId") or "")
|
||
if not candidate_id:
|
||
continue
|
||
probe = run_capture(
|
||
binary,
|
||
"+messages-list",
|
||
[
|
||
"--group",
|
||
candidate_id,
|
||
"--time",
|
||
time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"--forward=false",
|
||
"--limit",
|
||
"50",
|
||
],
|
||
timeout,
|
||
)
|
||
probe_summary = summarize_capture(probe)
|
||
if probe_summary["status"] not in {"pass", "pass_empty"}:
|
||
continue
|
||
rows = first_list(probe.upper(), "messages")
|
||
own_open_id = (
|
||
str(candidate.get("ownerOpenDingtalkId") or "")
|
||
if str(candidate.get("myRole") or "") == "OWNER"
|
||
else ""
|
||
)
|
||
own_message = any(
|
||
isinstance(item, dict) and item.get("senderId") == own_open_id
|
||
for item in rows
|
||
)
|
||
probe_raw = probe.raw_calls[-1][2] if probe.raw_calls else {}
|
||
has_resource = bool(
|
||
first_string(probe_raw, "mediaId", "resourceId", "downloadCode")
|
||
)
|
||
has_thread = bool(
|
||
first_string(
|
||
probe_raw,
|
||
"openConvThreadId",
|
||
"openConversationThreadId",
|
||
"threadId",
|
||
"topicId",
|
||
)
|
||
)
|
||
score = (
|
||
int(probe_summary.get("upper", {}).get("count") or 0)
|
||
+ (1000 if own_message else 0)
|
||
+ (500 if has_resource else 0)
|
||
+ (500 if has_thread else 0)
|
||
)
|
||
if score > best_score:
|
||
best_score = score
|
||
best_group = candidate
|
||
messages = probe
|
||
if messages is not None:
|
||
group = best_group
|
||
|
||
group_id = str(group.get("openConversationId") or "")
|
||
group_name = str(group.get("title") or group.get("name") or "")
|
||
|
||
if group_id:
|
||
conversation = run("+conversation-info", "--group", group_id)
|
||
run("+chat-invite-url", "--group", group_id, "--expires-seconds", "60")
|
||
run("+chat-bots", "--group", group_id)
|
||
run("+chat-role-list", "--group", group_id)
|
||
run("+messages-list-pin", "--open-conversation-id", group_id, "--size", "100")
|
||
if messages is None:
|
||
messages = run(
|
||
"+messages-list",
|
||
"--group",
|
||
group_id,
|
||
"--time",
|
||
time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"--forward=false",
|
||
"--limit",
|
||
"50",
|
||
)
|
||
else:
|
||
captures["+messages-list"] = messages
|
||
results["+messages-list"] = summarize_capture(messages)
|
||
run(
|
||
"+chat-messages",
|
||
"--group",
|
||
group_id,
|
||
"--time",
|
||
time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"--direction",
|
||
"older",
|
||
"--limit",
|
||
"50",
|
||
retries=2,
|
||
)
|
||
if group_name:
|
||
run("+chat-search", "--query", group_name, "--limit", "20")
|
||
members = run("+group-members", "--group", group_name)
|
||
run("+chat-members-list", "--group", group_name)
|
||
else:
|
||
results["+chat-search"] = blocked("+chat-search", "真实群需要有可搜索标题")
|
||
results["+group-members"] = blocked("+group-members", "真实群需要有可搜索标题")
|
||
results["+chat-members-list"] = blocked(
|
||
"+chat-members-list", "真实群需要有可搜索标题"
|
||
)
|
||
members = None
|
||
|
||
numeric_group_id = numeric_group_override or first_string(
|
||
conversation.raw_calls[-1][2] if conversation.raw_calls else {},
|
||
"groupId",
|
||
"group_id",
|
||
)
|
||
if numeric_group_id and numeric_group_id.isdigit():
|
||
run("+chat-get-by-id", "--group-id", numeric_group_id)
|
||
else:
|
||
results["+chat-get-by-id"] = blocked(
|
||
"+chat-get-by-id",
|
||
"需要一个真实数字群号 groupId；openConversationId 不能替代",
|
||
)
|
||
|
||
member_rows = (
|
||
first_list(members.upper(), "members") if members is not None else []
|
||
)
|
||
member_open_id = first_string(
|
||
member_rows,
|
||
"openDingtalkId",
|
||
"openDingTalkId",
|
||
"memberDingtalkId",
|
||
)
|
||
if member_open_id:
|
||
run(
|
||
"+chat-members-get",
|
||
"--id",
|
||
group_id,
|
||
"--users",
|
||
member_open_id,
|
||
)
|
||
role_query_capture: Capture | None = None
|
||
for member in member_rows[:max_member_probes]:
|
||
candidate_open_id = first_string(
|
||
member,
|
||
"openDingtalkId",
|
||
"openDingTalkId",
|
||
"memberDingtalkId",
|
||
)
|
||
if not candidate_open_id:
|
||
continue
|
||
probe = run_capture(
|
||
binary,
|
||
"+chat-role-query-user",
|
||
["--group", group_id, "--user", candidate_open_id],
|
||
timeout,
|
||
)
|
||
role_query_capture = probe
|
||
if summarize_capture(probe)["status"] in {"pass", "pass_empty"}:
|
||
break
|
||
if role_query_capture is not None:
|
||
captures["+chat-role-query-user"] = role_query_capture
|
||
results["+chat-role-query-user"] = summarize_capture(
|
||
role_query_capture
|
||
)
|
||
direct_capture: Capture | None = None
|
||
cross_org_only = True
|
||
for member in member_rows[:max_member_probes]:
|
||
candidate_open_id = first_string(
|
||
member,
|
||
"openDingtalkId",
|
||
"openDingTalkId",
|
||
"memberDingtalkId",
|
||
)
|
||
if not candidate_open_id:
|
||
continue
|
||
probe = run_capture(
|
||
binary,
|
||
"+messages-list-direct",
|
||
[
|
||
"--open-dingtalk-id",
|
||
candidate_open_id,
|
||
"--time",
|
||
time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"--forward=false",
|
||
"--limit",
|
||
"20",
|
||
"--page-all",
|
||
"--page-limit",
|
||
"50",
|
||
],
|
||
timeout,
|
||
)
|
||
summary = summarize_capture(probe)
|
||
if summary["status"] in {"pass", "pass_empty"}:
|
||
direct_capture = probe
|
||
break
|
||
if summary.get("error", {}).get("server_error_code") != (
|
||
"CrossOrgPermissionDenied"
|
||
):
|
||
cross_org_only = False
|
||
direct_capture = probe
|
||
break
|
||
if direct_capture is not None:
|
||
captures["+messages-list-direct"] = direct_capture
|
||
results["+messages-list-direct"] = summarize_capture(direct_capture)
|
||
elif cross_org_only:
|
||
results["+messages-list-direct"] = blocked(
|
||
"+messages-list-direct",
|
||
"需要一个当前账号有权读取单聊记录的同组织成员",
|
||
)
|
||
else:
|
||
requirement = "需要一个可读取 openDingTalkId 的真实群成员"
|
||
for command in (
|
||
"+chat-members-get",
|
||
"+chat-role-query-user",
|
||
"+messages-list-direct",
|
||
):
|
||
results[command] = blocked(command, requirement)
|
||
|
||
message_rows = first_list(messages.upper(), "messages")
|
||
own_open_id = (
|
||
str(group.get("ownerOpenDingtalkId") or "")
|
||
if str(group.get("myRole") or "") == "OWNER"
|
||
else ""
|
||
)
|
||
own_message_rows = [
|
||
item
|
||
for item in message_rows
|
||
if isinstance(item, dict) and item.get("senderId") == own_open_id
|
||
]
|
||
message_id = first_string(
|
||
own_message_rows or message_rows,
|
||
"messageId",
|
||
"openMessageId",
|
||
"openMsgId",
|
||
)
|
||
if message_id:
|
||
run("+messages-mget", "--msg-ids", message_id)
|
||
if own_message_rows:
|
||
run(
|
||
"+messages-read-status",
|
||
"--conversation-id",
|
||
group_id,
|
||
"--message-id",
|
||
message_id,
|
||
)
|
||
else:
|
||
results["+messages-read-status"] = blocked(
|
||
"+messages-read-status",
|
||
"需要当前账号自己发送的一条真实消息 openMessageId",
|
||
)
|
||
else:
|
||
requirement = "需要该群内至少一条可读取的真实消息 openMessageId"
|
||
results["+messages-mget"] = blocked("+messages-mget", requirement)
|
||
results["+messages-read-status"] = blocked(
|
||
"+messages-read-status", requirement
|
||
)
|
||
|
||
keyword = safe_keyword(message_rows)
|
||
run(
|
||
"+search-msg",
|
||
"--group",
|
||
group_id,
|
||
"--query",
|
||
keyword,
|
||
"--days",
|
||
"30",
|
||
)
|
||
|
||
raw_messages = messages.raw_calls[-1][2] if messages.raw_calls else {}
|
||
topic_id = topic_id_override or first_string(
|
||
raw_messages,
|
||
"openConvThreadId",
|
||
"openConversationThreadId",
|
||
"threadId",
|
||
"topicId",
|
||
)
|
||
if topic_id:
|
||
topic_args = [
|
||
"--group",
|
||
topic_group_override or group_id,
|
||
"--topic-id",
|
||
topic_id,
|
||
]
|
||
if topic_page_size:
|
||
topic_args.extend(
|
||
[
|
||
"--limit",
|
||
topic_page_size,
|
||
"--page-all",
|
||
"--page-limit",
|
||
"50",
|
||
]
|
||
)
|
||
else:
|
||
topic_args.extend(
|
||
[
|
||
"--time",
|
||
time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"--limit",
|
||
"20",
|
||
]
|
||
)
|
||
run("+thread-replies", *topic_args)
|
||
else:
|
||
results["+thread-replies"] = blocked(
|
||
"+thread-replies",
|
||
"需要一个真实话题群及其 topic/thread ID",
|
||
)
|
||
|
||
resource_id = media_resource_override
|
||
resource_message_id = media_message_override
|
||
resource_group_id = media_group_override or group_id
|
||
for raw_message in first_list(raw_messages, "messages", "messageList"):
|
||
if resource_id and resource_message_id:
|
||
break
|
||
if not isinstance(raw_message, dict):
|
||
continue
|
||
candidate_resource_id = first_string(
|
||
raw_message,
|
||
"mediaId",
|
||
"resourceId",
|
||
"downloadCode",
|
||
)
|
||
candidate_message_id = first_string(
|
||
raw_message,
|
||
"openMessageId",
|
||
"openMsgId",
|
||
)
|
||
if candidate_resource_id and candidate_message_id:
|
||
resource_id = candidate_resource_id
|
||
resource_message_id = candidate_message_id
|
||
break
|
||
if resource_id and resource_message_id:
|
||
run(
|
||
"+messages-resource-url",
|
||
"--type",
|
||
"mediaId",
|
||
"--resource-id",
|
||
resource_id,
|
||
"--message-id",
|
||
resource_message_id,
|
||
"--open-conversation-id",
|
||
resource_group_id,
|
||
)
|
||
download_output = (
|
||
Path("tmp")
|
||
/ "im-shortcut-live-audit"
|
||
/ f".read-resource-{os.getpid()}-{time.time_ns()}.bin"
|
||
)
|
||
download_output.parent.mkdir(parents=True, exist_ok=True)
|
||
try:
|
||
download_capture = run(
|
||
"+messages-resource-download",
|
||
"--type",
|
||
"mediaId",
|
||
"--resource-id",
|
||
resource_id,
|
||
"--message-id",
|
||
resource_message_id,
|
||
"--open-conversation-id",
|
||
resource_group_id,
|
||
"--output",
|
||
str(download_output),
|
||
)
|
||
download_summary = results["+messages-resource-download"]
|
||
upper_download = download_capture.upper()
|
||
expected_size = (
|
||
upper_download.get("sizeBytes")
|
||
if isinstance(upper_download, dict)
|
||
else None
|
||
)
|
||
if (
|
||
download_summary["status"] == "pass"
|
||
and (
|
||
not download_output.is_file()
|
||
or not isinstance(expected_size, (int, float))
|
||
or download_output.stat().st_size != expected_size
|
||
)
|
||
):
|
||
download_summary["status"] = "local_artifact_mismatch"
|
||
finally:
|
||
download_output.unlink(missing_ok=True)
|
||
else:
|
||
results["+messages-resource-url"] = blocked(
|
||
"+messages-resource-url",
|
||
"需要一条含 mediaId 的真实图片/视频/语音消息",
|
||
)
|
||
results["+messages-resource-download"] = blocked(
|
||
"+messages-resource-download",
|
||
"需要一条含 mediaId 的真实图片/视频/语音消息",
|
||
)
|
||
else:
|
||
requirement = "当前账号需要至少加入一个可读真实群"
|
||
for command in (
|
||
"+conversation-info",
|
||
"+chat-invite-url",
|
||
"+chat-bots",
|
||
"+chat-role-list",
|
||
"+messages-list-pin",
|
||
"+messages-list",
|
||
"+chat-messages",
|
||
"+chat-search",
|
||
"+group-members",
|
||
"+chat-members-list",
|
||
"+chat-get-by-id",
|
||
"+chat-members-get",
|
||
"+chat-role-query-user",
|
||
"+messages-list-direct",
|
||
"+messages-mget",
|
||
"+messages-read-status",
|
||
"+search-msg",
|
||
"+thread-replies",
|
||
"+messages-resource-download",
|
||
"+messages-resource-url",
|
||
):
|
||
results[command] = blocked(command, requirement)
|
||
|
||
open_task_id = open_task_override
|
||
for capture in captures.values():
|
||
if open_task_id:
|
||
break
|
||
for _, _, raw in capture.raw_calls:
|
||
open_task_id = first_string(raw, "openTaskId", "taskId")
|
||
if open_task_id:
|
||
break
|
||
if open_task_id:
|
||
break
|
||
if open_task_id:
|
||
run("+messages-query-send-status", "--open-task-id", open_task_id)
|
||
else:
|
||
results["+messages-query-send-status"] = blocked(
|
||
"+messages-query-send-status",
|
||
"需要一次真实消息发送返回的 openTaskId",
|
||
)
|
||
|
||
return [results.get(command, blocked(command, "审计器未覆盖")) for command in READ_SHORTCUTS]
|
||
|
||
|
||
def report_markdown(results: list[dict[str, Any]]) -> str:
|
||
counts: dict[str, int] = {}
|
||
for result in results:
|
||
counts[result["status"]] = counts.get(result["status"], 0) + 1
|
||
lines = [
|
||
"# Chat Shortcut live read audit",
|
||
"",
|
||
"This report was produced through the real Shortcut entry point and the",
|
||
"lower MCP response captured from the same invocation. It intentionally",
|
||
"contains no business payload values.",
|
||
"",
|
||
f"- Total: {len(results)}",
|
||
]
|
||
for status in sorted(counts):
|
||
lines.append(f"- {status}: {counts[status]}")
|
||
lines.extend(
|
||
[
|
||
"",
|
||
"| Shortcut | Status | Upper | Lower | Reactions U/L | Threads U/L | MCP calls | Evidence / requirement |",
|
||
"|---|---|---:|---:|---:|---:|---:|---|",
|
||
]
|
||
)
|
||
for result in results:
|
||
upper = result.get("upper", {}).get("count")
|
||
lower = result.get("lower", {}).get("count")
|
||
calls = result.get("lower", {}).get("call_count")
|
||
upper_reactions = result.get("upper", {}).get("reaction_message_count")
|
||
lower_reactions = result.get("lower", {}).get("reaction_message_count")
|
||
upper_threads = result.get("upper", {}).get("thread_message_count")
|
||
lower_threads = result.get("lower", {}).get("thread_message_count")
|
||
tools = ", ".join(result.get("lower", {}).get("tools", []))
|
||
evidence = result.get("requirement") or tools
|
||
lines.append(
|
||
f"| `{result['command']}` | {result['status']} | "
|
||
f"{'' if upper is None else upper} | "
|
||
f"{'' if lower is None else lower} | "
|
||
f"{'' if upper_reactions is None else upper_reactions}/"
|
||
f"{'' if lower_reactions is None else lower_reactions} | "
|
||
f"{'' if upper_threads is None else upper_threads}/"
|
||
f"{'' if lower_threads is None else lower_threads} | "
|
||
f"{'' if calls is None else calls} | {evidence} |"
|
||
)
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def validate_report(results: list[dict[str, Any]]) -> None:
|
||
commands = [result["command"] for result in results]
|
||
if tuple(commands) != READ_SHORTCUTS:
|
||
raise RuntimeError("read Shortcut audit coverage drifted from the expected set")
|
||
encoded = json.dumps(results, ensure_ascii=False)
|
||
for forbidden in ("stdout", "stderr", "backend_raw", "raw_backend", "inviteUrl"):
|
||
if f'"{forbidden}"' in encoded:
|
||
raise RuntimeError(f"sanitized report contains forbidden field {forbidden!r}")
|
||
if "https://" in encoded or "http://" in encoded:
|
||
raise RuntimeError("sanitized report appears to contain a URL")
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--dws", required=True, type=Path, help="locally built dws binary")
|
||
parser.add_argument("--out-dir", required=True, type=Path)
|
||
parser.add_argument("--timeout", type=int, default=60)
|
||
parser.add_argument("--max-group-probes", type=int, default=8)
|
||
parser.add_argument("--max-member-probes", type=int, default=8)
|
||
args = parser.parse_args()
|
||
|
||
binary = args.dws.resolve()
|
||
if not binary.is_file():
|
||
parser.error(f"--dws is not a file: {binary}")
|
||
numeric_group_id = os.environ.get("DWS_AUDIT_NUMERIC_GROUP_ID", "").strip()
|
||
if numeric_group_id and not numeric_group_id.isdigit():
|
||
parser.error("DWS_AUDIT_NUMERIC_GROUP_ID must contain only digits")
|
||
topic_override = [
|
||
os.environ.get("DWS_AUDIT_TOPIC_GROUP_ID", "").strip(),
|
||
os.environ.get("DWS_AUDIT_TOPIC_ID", "").strip(),
|
||
]
|
||
if any(topic_override) and not all(topic_override):
|
||
parser.error("topic fixture override requires both group and topic IDs")
|
||
topic_page_size = os.environ.get("DWS_AUDIT_TOPIC_PAGE_SIZE", "").strip()
|
||
if topic_page_size and (
|
||
not topic_page_size.isdigit() or int(topic_page_size) < 1
|
||
):
|
||
parser.error("DWS_AUDIT_TOPIC_PAGE_SIZE must be a positive integer")
|
||
media_override = [
|
||
os.environ.get("DWS_AUDIT_MEDIA_GROUP_ID", "").strip(),
|
||
os.environ.get("DWS_AUDIT_MEDIA_MESSAGE_ID", "").strip(),
|
||
os.environ.get("DWS_AUDIT_MEDIA_RESOURCE_ID", "").strip(),
|
||
]
|
||
if any(media_override) and not all(media_override):
|
||
parser.error("media fixture override requires group, message, and resource IDs")
|
||
results = live_audit(
|
||
binary,
|
||
args.timeout,
|
||
args.max_group_probes,
|
||
args.max_member_probes,
|
||
)
|
||
validate_report(results)
|
||
|
||
args.out_dir.mkdir(parents=True, exist_ok=True)
|
||
json_path = args.out_dir / "chat-read-live-audit.json"
|
||
markdown_path = args.out_dir / "chat-read-live-audit.md"
|
||
json_path.write_text(
|
||
json.dumps(
|
||
{
|
||
"scope": "chat-read-shortcuts",
|
||
"count": len(results),
|
||
"results": results,
|
||
},
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
+ "\n",
|
||
encoding="utf-8",
|
||
)
|
||
markdown_path.write_text(report_markdown(results), encoding="utf-8")
|
||
print(markdown_path)
|
||
|
||
failures = {
|
||
"projection_mismatch",
|
||
"projection_unverified",
|
||
"pagination_mismatch",
|
||
"backend_error",
|
||
"cli_error",
|
||
"error_envelope",
|
||
"timeout",
|
||
"no_lower_capture",
|
||
"local_artifact_mismatch",
|
||
"incomplete",
|
||
}
|
||
return 1 if any(result["status"] in failures for result in results) else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|