From 132947ef6957c37e24a0fd736cf5079d8e7819d9 Mon Sep 17 00:00:00 2001 From: Rogee Date: Sat, 10 Oct 2026 09:26:37 +0800 Subject: [PATCH] feat: add safe structured transport logging --- AGENTS.md | 6 +- README.md | 4 + cmd/sip-call/README.md | 6 +- cmd/sip-go-agent/agent_command.go | 18 ++- cmd/sip-go-agent/agent_real.go | 4 +- cmd/sip-go-agent/agent_setup.go | 6 +- cmd/sip-go-agent/debug_test.go | 60 ++++++-- cmd/sip-go-agent/dispatcher_command.go | 11 +- cmd/sip-go-agent/dispatcher_http_test.go | 6 +- cmd/sip-go-agent/dispatcher_sip.go | 3 +- deploys/production/README.md | 6 +- deploys/production/production.py | 5 +- deploys/production/remote.py | 7 +- deploys/production/test_lifecycle.py | 14 ++ deploys/test/README.md | 6 +- docs/logging.md | 39 +++++ go.mod | 1 + go.sum | 2 + internal/agent/recording_scan.go | 7 + internal/agent/result_journal.go | 9 +- internal/agent/upload.go | 3 +- internal/ai/pipeline.go | 2 + internal/config/agent_debug_test.go | 17 ++- internal/config/agent_runtime.go | 9 +- internal/configread/client.go | 4 +- internal/dispatcher/execute.go | 18 ++- internal/logging/http.go | 166 ++++++++++++++++++++ internal/logging/http_regression_test.go | 94 ++++++++++++ internal/logging/http_test.go | 88 +++++++++++ internal/logging/logging.go | 114 ++++++++++++++ internal/logging/logging_test.go | 101 +++++++++++++ internal/logging/redact.go | 184 +++++++++++++++++++++++ internal/logging/rpc.go | 141 +++++++++++++++++ internal/logging/rpc_test.go | 65 ++++++++ internal/mq/broker.go | 24 ++- internal/mq/logging.go | 74 +++++++++ internal/mq/logging_test.go | 76 ++++++++++ internal/rpc/approved_debug_test.go | 4 +- internal/rpc/approved_recorded_real.go | 6 +- internal/rpc/approved_worker.go | 10 +- 40 files changed, 1352 insertions(+), 68 deletions(-) create mode 100644 docs/logging.md create mode 100644 internal/logging/http.go create mode 100644 internal/logging/http_regression_test.go create mode 100644 internal/logging/http_test.go create mode 100644 internal/logging/logging.go create mode 100644 internal/logging/logging_test.go create mode 100644 internal/logging/redact.go create mode 100644 internal/logging/rpc.go create mode 100644 internal/logging/rpc_test.go create mode 100644 internal/mq/logging.go create mode 100644 internal/mq/logging_test.go diff --git a/AGENTS.md b/AGENTS.md index 40da88e..ef08608 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -79,6 +79,8 @@ ## 当前范围与权威入口 +- **当前日志与抓包规则**：Dispatcher、Agent 的 `--debug/-D` 仅开启 Logrus 详细收发日志；普通模式记录自身关键工作流程、方向、身份、结果、耗时及错误。HTTP/MQ/RPC 正文只在 Debug 中记录安全副本，禁止密钥、Token、签名 URL、完整提示词/对话/音频及原始 SIP；不截断其余可读业务 JSON，不修改发送数据、不吞错误。Agent 原 `--debug` 抓包含义移除，逐通外部抓包凭证检查改为独立 `--capture`，不足时仍失败关闭。独立 `sip-call` 的 Debug 抓包参数不变。详细规则见 [`docs/logging.md`](docs/logging.md)。 + - **2026-10-09 17:24 最新参数规则与现场**：使用者确认 params 重名键忽略、任务/session 字段优先；共享合同 `f490afa`、实现 `ddd4de2` 已推送并安装。真实 SaaS 任务 `4` 的原重复 voice 和 revision `1` 未改，现已接纳；四个真实任务均暂停，Dispatcher 持久发现 ready、SIP 本地加载代次 `17`/pending=0。Agent/Dispatcher/Asterisk 均 enabled+active，D PID 稳定、零重启；未清数据、重启 Asterisk、独立拨号或主动投递。部署工具仍引用启动前独立 probe 的 tasks 阶段失败而显示 `saas_pending/business_ready=false`，具体 probe 原因未定位，不伪造成功，也不以标签否定实际业务启动事实；完整独立探测、真实通话/AI/OSS及生产验收仍不宣称完成。证据见 [`docs/evidence/ai-params-task-precedence-20261009.md`](docs/evidence/ai-params-task-precedence-20261009.md)。 - **2026-10-09 最新固定环境事实**：使用者确认修复正式 Dispatcher 的 SIP 加载遗漏并重部署；源码 `b37d9f3` 已推送、安装，Agent/Asterisk 的 revision `16`、三条数字 ID endpoint/AOR/主叫/contact/UDP/PCMA 已从原生运行态逐条核对，Asterisk 未重启，零活动通话，未清数据、未独立拨号。Dispatcher 仍因真实 SaaS 任务 `4`（revision `1`）同时包含 `agent.tts.voice` 与 `agent.tts.params.voice` 而退出重启；该失败是当时采用的冲突拒绝规则造成；随后使用者确认改为任务/session 字段优先、忽略 params 中重名键，共享合同已更新为 `f490afa`，SaaS 无需删除原有重复 voice。新规则部署及最新业务状态见 [`docs/evidence/ai-params-task-precedence-20261009.md`](docs/evidence/ai-params-task-precedence-20261009.md)；未完成现场核对前，不得把 SIP 加载成功或状态中的 `saas_pending` 标签当作全部配置/真实通话/生产签收。证据见 [`docs/evidence/dispatcher-sip-startup-load-20261009.md`](docs/evidence/dispatcher-sip-startup-load-20261009.md)。 @@ -129,7 +131,7 @@ | 百应 | `160.202.254.79:5060` | `KQ91526` | `mka755108` | -- 业务全线路的原始被叫号码仅由校验归属及任务后的 SaaS `call.execute.payload.callee` 确定；不在 Dispatcher、SaaS Mock 或抓证脚本设置固定号码/日期特判，不向任务快照增设号码列表。只接受 1–32 位 ASCII 数字；格式校验不等于真实拨号授权。不再另设本地固定的 `09:00`–`20:00` 窗口或每线路每号码每日 3 次上限；任务、线路时段和额度以 SaaS 配置快照校验为准，不等候/自动延迟/自动重试/静默换线。独立工具或非上述真实 SaaS 获批任务的真实试拨仍须使用者明确安排；本轮使用非正式配置的正式部署任务按当前授权执行。仅在 Agent 指定 `--debug/-D` 时，由专用主机脚本在拨号前启用抓包和 PJSIP logger、签发与该通 `event_id`/trunk/原始号码绑定的短时有效活跃抓包凭证，调试 Agent 拒绝缺失/失效/不匹配的凭证。默认 Agent 不依赖外部抓包或凭证；独立 sip-call 的号码与线路仅来自本次显式命令和 ENV，不冒充业务任务授权。SaaS Mock 投递和本地时段测试不构成真实拨号授权。 +- 业务全线路的原始被叫号码仅由校验归属及任务后的 SaaS `call.execute.payload.callee` 确定；不在 Dispatcher、SaaS Mock 或抓证脚本设置固定号码/日期特判，不向任务快照增设号码列表。只接受 1–32 位 ASCII 数字；格式校验不等于真实拨号授权。不再另设本地固定的 `09:00`–`20:00` 窗口或每线路每号码每日 3 次上限；任务、线路时段和额度以 SaaS 配置快照校验为准，不等候/自动延迟/自动重试/静默换线。独立工具或非上述真实 SaaS 获批任务的真实试拨仍须使用者明确安排；本轮使用非正式配置的正式部署任务按当前授权执行。仅在 Agent 指定 `--capture` 时，由专用主机脚本在拨号前启用抓包和 PJSIP logger、签发与该通 `event_id`/trunk/原始号码绑定的短时有效活跃抓包凭证，调试 Agent 拒绝缺失/失效/不匹配的凭证。默认 Agent 不依赖外部抓包或凭证；独立 sip-call 的号码与线路仅来自本次显式命令和 ENV，不冒充业务任务授权。SaaS Mock 投递和本地时段测试不构成真实拨号授权。 - 任务按周一至周日多个时段与指定排除日期配置，线路只有每周允许时段（**没有线路排除日期**），Asia/Shanghai 左闭右开、跨日拆分；缺失或不确定 fail-closed，不自动重拨。由 Dispatcher 在持久接纳与实际发出指令前判定，并取任务/获批 AI 较小通话时限；Agent 仅校验会话和签发期限，不重算外呼策略。本地 SaaS 快照策略测试与主机抓证/逐次授权须分别报告。 - 当前修复仅在本地核验，未更新测试机上的旧多主叫 Mock 快照，也未部署或授权新呼叫。`BD` 等主叫原值不得清洗或当作 Digest 用户名；业务原始被叫号码不变，仅被选定数企 trunk 按规则构造 `7089<原号>`（其它线路使用自己的前缀），不重复加前缀。三条 trunk 独立，不能把同地址伪造为备用线路或换线重拨；服务商反馈 PCMA，对应 Asterisk `allow=alaw`，传输/注册/鉴权/并发仍待真实签收。不以 sipgo/diago 另造 Asterisk 替代架构。 @@ -138,6 +140,6 @@ - 项目是独立 Go module，工具链 Go **1.27.1**；普通构建、测试、运行不读取父项目业务模块、数据库、env 或夹具。标准库和成熟官方 SDK 优先，Cobra 显式 `agent`/`dispatcher`，单制品分角色/权限/目录。禁止自行重写 SIP/ARI、RTP/RTCP/G.711、WebSocket、AMQP、SQLite 驱动、OSS 签名及 SDK 已覆盖的 AI 协议；核验现有依赖能力后再新增库。生产原生 Asterisk 仍由独立 Cell 的 systemd 统一管理，不声称当前 Mock 已完成真实媒体或 1000 路容量验收。 - 生产 ECS 优先 Debian 13（Trixie）minimal；只有阿里云北京无可用镜像时允许 Ubuntu 24.04 LTS。Debian 12 仅供明确标记的非生产测试：必须在 Debian 12 原生构建 Asterisk，不得部署 Debian 13 编译的制品，也不得据此批准生产发布。Asterisk 直接安装在承载 ECS 主机，以 `rogee` 的 `systemd --user` 服务管理，核对 `enabled+active`；生产环境还须启用 `loginctl enable-linger rogee` 并验证重启后持续运行。非生产若未启用 lingering，必须标记重启自启动未验收。不得以前台进程或容器入口代替。 - 开发、Mock、mixed、real 的**非生产主机**部署与诊断步骤默认强制，不因时间/旧环境/调用方参数跳过或静默降级；显式关闭即失败。每次新主机/版本/Cell 至少留存脱敏 ECS/EIP/网络只读核验、Debian/架构/磁盘/权限、`rogee` SSH 与加固、发布包/依赖 SHA-256、Asterisk/systemd `enabled+active`、ARI/PJSIP endpoint/contact、媒体 profile/端口及运行版本。 -- 非生产 mixed/real 呼叫必须通过 SaaS 下发的任务、线路时段与额度校验及逐次授权；ARI/PJSIP 状态必须可核验。仅指定 Agent `--debug/-D` 时，**拨号前**启动受限 SIP/RTP 抓包和 Asterisk PJSIP logger，结束后采集 SIP 响应/INVITE–BYE 时间线、SDP codec/媒体地址端口、RTP 包/字节、录音与 ASR/LLM/TTS 事实及 SHA-256，失败通话也保存状态和抓包；此调试模式下 tcpdump/CAP_NET_RAW、PJSIP logger 或证据凭证任一不可用须失败关闭。默认 Agent 不依赖上述外部抓包工具，不能将未抓包详情作为抓包证据。独立 sip-call 不测试 AI/录音链路，其调试 pcap 只覆盖指定服务端 IP，不能据缺失 RTP 断言无媒体。原始抓包/日志/录音只写受限证据目录，聊天、提交与长期证据只存脱敏摘要/计数/状态码/hash，不含完整用户音频/对话或凭据。统一入口见 [`deploys/test/nonprod-call-evidence.sh`](deploys/test/nonprod-call-evidence.sh)；本地 `make check` 与 `make release-check-local` **不能代签主机诊断或生产门禁**。 +- 非生产 mixed/real 呼叫必须通过 SaaS 下发的任务、线路时段与额度校验及逐次授权；ARI/PJSIP 状态必须可核验。仅指定 Agent `--capture` 时，**拨号前**启动受限 SIP/RTP 抓包和 Asterisk PJSIP logger，结束后采集 SIP 响应/INVITE–BYE 时间线、SDP codec/媒体地址端口、RTP 包/字节、录音与 ASR/LLM/TTS 事实及 SHA-256，失败通话也保存状态和抓包；此调试模式下 tcpdump/CAP_NET_RAW、PJSIP logger 或证据凭证任一不可用须失败关闭。默认 Agent 不依赖上述外部抓包工具，不能将未抓包详情作为抓包证据。独立 sip-call 不测试 AI/录音链路，其调试 pcap 只覆盖指定服务端 IP，不能据缺失 RTP 断言无媒体。原始抓包/日志/录音只写受限证据目录，聊天、提交与长期证据只存脱敏摘要/计数/状态码/hash，不含完整用户音频/对话或凭据。统一入口见 [`deploys/test/nonprod-call-evidence.sh`](deploys/test/nonprod-call-evidence.sh)；本地 `make check` 与 `make release-check-local` **不能代签主机诊断或生产门禁**。 - 当前完成前至少检查格式、当前合同和 Proto 来源/hash、`go vet ./...`、`go test -race ./...`、构建、确实运行的隔离 RabbitMQ/HTTPS/双向 TLS 端到端测试、业务单元覆盖率 ≥65% 及 A01–A12/K01–K16 对照。真实 SaaS/management/OSS/AI/Asterisk/ECS、第二节点/Cell/租户、多 D 额度、容量/N+1及生产切换必须另有事实与授权，任何本机 Mock 通过不得写成其签收。 - 不自动提交/暂存/清理使用者在父项目或本项目的无关修改；并行开发仍须有可追溯 Git/合同基线、一 lane 一工作区/测试资源、无交叠写集合、合并后回归。本目标明确禁用子 Agent，不能以模型、fast 环境或外部服务不可用阻塞本地 Mock 目标。问题根因不明时补可观测性并诚实报告，不能用静默兜底伪装修复。 diff --git a/README.md b/README.md index eea19ca..fd09245 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,10 @@ - 任务内不可变获批 AI 参数支持 ASR-only 与 ASR+LLM+TTS；录音经短期授权由 Agent **一次** HTTPS PUT 至 OSS，明确失败才按受控文件恢复，结果未知不自动再传，最终事实进入 MQ。批准 Mock 的 SIP revision 回报不证明 Asterisk 已实际加载。 - 历史来源和使用者修改的旧文档按原字节保存在 [`docs/archive/sources/`](docs/archive/sources/README.md)，上游 v1 保存在 [`docs/archive/upstream/`](docs/archive/upstream/README.md) 并可离线校验；旧阶段计划已归档。它们不是另一个当前版本或运行回退。 +## 日志与抓包 + +Dispatcher、Agent 的 `--debug/-D` 只开启 Logrus 详细收发日志；普通模式记录工作流程。Agent 的逐通抓包检查独立为 `--capture`，不与日志联动；独立 `sip-call` 的参数不变。详见 [日志说明](docs/logging.md)。 + ## 本地验证（不触碰真实线路） 首次获取：`git clone --recurse-submodules <本项目仓库地址>`；已有项目执行 `git submodule update --init --recursive`。 diff --git a/cmd/sip-call/README.md b/cmd/sip-call/README.md index 3477425..23bfa2c 100644 --- a/cmd/sip-call/README.md +++ b/cmd/sip-call/README.md @@ -70,12 +70,12 @@ SIP 服务端与媒体服务器使用不同 IP 时，这个窄过滤器**不覆 ## Agent 的调试参数 ```sh -sip-go-agent agent --mode nonprod-real -D +sip-go-agent agent --mode nonprod-real --capture ``` -普通 Agent 默认不依赖 tcpdump，不校验外部抓包凭证。`-D` 则复用 [`nonprod-call-evidence.sh`](../../deploys/test/nonprod-call-evidence.sh)：脚本提前开启抓包与 PJSIP logger，Agent 核对 `AGENT_EVIDENCE_ROOT` 下绑定本通 event/线路/原始号码的活跃凭证。调试条件缺失拒绝拨号；脚本不会因为新增参数而自动启动。 +普通 Agent 默认不依赖 tcpdump，不校验外部抓包凭证。`--capture` 则复用 [`nonprod-call-evidence.sh`](../../deploys/test/nonprod-call-evidence.sh)：脚本提前开启抓包与 PJSIP logger，Agent 核对 `AGENT_EVIDENCE_ROOT` 下绑定本通 event/线路/原始号码的活跃凭证。调试条件缺失拒绝拨号；脚本不会因为新增参数而自动启动。 -此参数仅属于 Agent，Dispatcher 不接受。业务授权、任务/线路时段、额度和 HEP 原生 SIP 响应采集规则未取消；没有 tcpdump 不等于绕过业务门禁。 +`--capture` 仅属于 Agent，Dispatcher 不接受。Dispatcher、Agent 的 `--debug/-D` 均只开启详细日志，不与抓包联动；独立 `sip-call` 的 `--debug/-D` 仍保留本文原有抓包含义。业务授权、任务/线路时段、额度和 HEP 原生 SIP 响应采集规则未取消；没有 tcpdump 不等于绕过业务门禁。 ## 本地验证范围 diff --git a/cmd/sip-go-agent/agent_command.go b/cmd/sip-go-agent/agent_command.go index d262e05..39f3de6 100644 --- a/cmd/sip-go-agent/agent_command.go +++ b/cmd/sip-go-agent/agent_command.go @@ -15,25 +15,30 @@ import ( agentpb "git.ipao.vip/rogee/go-sip/gen/agent" "git.ipao.vip/rogee/go-sip/internal/asterisk" "git.ipao.vip/rogee/go-sip/internal/config" + "git.ipao.vip/rogee/go-sip/internal/logging" "git.ipao.vip/rogee/go-sip/internal/rpc" "github.com/spf13/cobra" "google.golang.org/grpc" "google.golang.org/grpc/credentials" + "log/slog" ) func newAgentCommand() *cobra.Command { var mode string - var debug bool + var debug, capture bool command := &cobra.Command{ Use: "agent", Short: "Run the bound Agent", Args: cobra.NoArgs, SilenceUsage: true, RunE: func(cmd *cobra.Command, _ []string) error { + logging.Configure("agent", debug, os.Stderr) + logging.ObserveDefaultHTTP("agent") + slog.Info("Agent startup", "mode", mode, "debug", debug, "capture", capture) settings, err := config.LoadAgentEnvironment(mode) if err != nil { return err } - settings.Debug = debug - if err := settings.ValidateDebug(); err != nil { + settings.Debug, settings.Capture = debug, capture + if err := settings.ValidateCapture(); err != nil { return err } ca, err := readAgentPEM("MTLS_CA_FILE", settings.CAFile) @@ -85,7 +90,7 @@ func newAgentCommand() *cobra.Command { if err != nil { return errors.New("Agent mTLS Dispatcher certificate configuration is invalid") } - connection, connectErr := grpc.NewClient(settings.DispatcherEndpoint, grpc.WithTransportCredentials(credentials.NewTLS(clientTLS))) + connection, connectErr := grpc.NewClient(settings.DispatcherEndpoint, grpc.WithTransportCredentials(credentials.NewTLS(clientTLS)), grpc.WithChainUnaryInterceptor(logging.UnaryClientInterceptor("agent", "dispatcher"))) if connectErr != nil { return errors.New("Agent cannot create pinned Dispatcher connection") } @@ -106,7 +111,7 @@ func newAgentCommand() *cobra.Command { return fmt.Errorf("Agent cannot listen on configured local address: %w", err) } defer listener.Close() - server := grpc.NewServer(grpc.Creds(credentials.NewTLS(serverTLS))) + server := grpc.NewServer(grpc.Creds(credentials.NewTLS(serverTLS)), grpc.ChainUnaryInterceptor(logging.UnaryServerInterceptor("agent", "dispatcher"))) agentpb.RegisterAgentControlServiceServer(server, handler) if mode == "nonprod-real" { startRealAgentRecovery(cmd.Context(), settings, recoveryClient, handler) @@ -151,7 +156,8 @@ func newAgentCommand() *cobra.Command { return nil }, } - command.Flags().BoolVarP(&debug, "debug", "D", false, "Require per-call evidence from the nonproduction capture script") + command.Flags().BoolVarP(&debug, "debug", "D", false, "Log detailed redacted interface and MQ data") + command.Flags().BoolVar(&capture, "capture", false, "Require per-call evidence from the nonproduction capture script") command.Flags().StringVar(&mode, "mode", "mock", "isolated Mock, SIP-only or explicit nonprod-real mode") return command } diff --git a/cmd/sip-go-agent/agent_real.go b/cmd/sip-go-agent/agent_real.go index 82a640a..1e16c47 100644 --- a/cmd/sip-go-agent/agent_real.go +++ b/cmd/sip-go-agent/agent_real.go @@ -40,7 +40,7 @@ func newRealAgentServer(ctx context.Context, settings config.AgentEnvironment, d if value.Kind() == reflect.Pointer && value.IsNil() { return nil, errors.New("real Agent requires a live pinned Dispatcher transport") } - if err := settings.ValidateDebug(); err != nil { + if err := settings.ValidateCapture(); err != nil { return nil, err } for _, path := range []string{settings.AsteriskConfigDir, settings.AsteriskBin, settings.AsteriskLibraryDir, settings.RecoveryRoot} { @@ -88,7 +88,7 @@ func newRealAgentServer(ctx context.Context, settings config.AgentEnvironment, d Journal: &agent.ResultJournal{Root: settings.RecoveryRoot}, } return (&rpc.ApprovedRecordedRealCall{ - Loader: loader, MediaPayloadType: 118, EvidenceRoot: settings.EvidenceRoot, Debug: settings.Debug, + Loader: loader, MediaPayloadType: 118, EvidenceRoot: settings.EvidenceRoot, Capture: settings.Capture, MaxWAVBytes: 64 << 20, ReportTimeout: 15 * time.Minute, Delivery: delivery, Mirror: mirror, }).Prepare(execution) }, diff --git a/cmd/sip-go-agent/agent_setup.go b/cmd/sip-go-agent/agent_setup.go index 3aa3137..f0b479f 100644 --- a/cmd/sip-go-agent/agent_setup.go +++ b/cmd/sip-go-agent/agent_setup.go @@ -21,6 +21,7 @@ import ( agentpb "git.ipao.vip/rogee/go-sip/gen/agent" "git.ipao.vip/rogee/go-sip/internal/agent" "git.ipao.vip/rogee/go-sip/internal/config" + "git.ipao.vip/rogee/go-sip/internal/logging" "git.ipao.vip/rogee/go-sip/internal/rpc" "git.ipao.vip/rogee/go-sip/internal/tenant" "github.com/google/uuid" @@ -154,7 +155,10 @@ func localMockHTTPClient(trustPEM []byte) (*http.Client, error) { if !roots.AppendCertsFromPEM(trustPEM) { return nil, errors.New("Mock HTTPS requires an approved trust bundle") } - transport := http.DefaultTransport.(*http.Transport).Clone() + transport, err := logging.CloneDefaultTransport() + if err != nil { + return nil, err + } transport.Proxy = nil transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12, RootCAs: roots} transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { diff --git a/cmd/sip-go-agent/debug_test.go b/cmd/sip-go-agent/debug_test.go index 817f1e1..e4067c0 100644 --- a/cmd/sip-go-agent/debug_test.go +++ b/cmd/sip-go-agent/debug_test.go @@ -2,23 +2,55 @@ package main import "testing" -func TestDebugFlagIsExplicitAndAgentOnly(t *testing.T) { +func TestDebugFlagOnlyEnablesDetailedLogsForBothRoles(t *testing.T) { for _, flag := range []string{"--debug", "-D"} { - command := newAgentCommand() - if err := command.ParseFlags([]string{flag}); err != nil { - t.Fatal(err) - } - enabled, err := command.Flags().GetBool("debug") - if err != nil || !enabled { - t.Fatalf("%s: %v %v", flag, enabled, err) - } - if err := newDispatcherCommand().ParseFlags([]string{flag}); err == nil { - t.Fatal("Dispatcher must not silently accept an Agent capture flag") + for _, role := range []string{"agent", "dispatcher"} { + command := newAgentCommand() + if role == "dispatcher" { + command = newDispatcherCommand() + } + if err := command.ParseFlags([]string{flag}); err != nil { + t.Fatalf("%s %s: %v", role, flag, err) + } + enabled, err := command.Flags().GetBool("debug") + if err != nil || !enabled { + t.Fatalf("%s %s: %v %v", role, flag, enabled, err) + } + if role == "agent" { + capture, err := command.Flags().GetBool("capture") + if err != nil || capture { + t.Fatalf("debug enabled capture: %v %v", capture, err) + } + } } } + for _, command := range []*struct { + role string + enabled bool + }{{role: "agent"}, {role: "dispatcher"}} { + c := newAgentCommand() + if command.role == "dispatcher" { + c = newDispatcherCommand() + } + if enabled, _ := c.Flags().GetBool("debug"); enabled { + t.Fatal("debug must be off by default") + } + } +} + +func TestCaptureFlagIsSeparateAndAgentOnly(t *testing.T) { command := newAgentCommand() - enabled, err := command.Flags().GetBool("debug") - if err != nil || enabled { - t.Fatal("capture must be disabled by default") + if err := command.ParseFlags([]string{"--capture"}); err != nil { + t.Fatal(err) + } + capture, err := command.Flags().GetBool("capture") + if err != nil || !capture { + t.Fatal(capture, err) + } + if debug, _ := command.Flags().GetBool("debug"); debug { + t.Fatal("capture must not enable raw logging") + } + if err := newDispatcherCommand().ParseFlags([]string{"--capture"}); err == nil { + t.Fatal("Dispatcher must reject capture") } } diff --git a/cmd/sip-go-agent/dispatcher_command.go b/cmd/sip-go-agent/dispatcher_command.go index e777ecc..3876dd0 100644 --- a/cmd/sip-go-agent/dispatcher_command.go +++ b/cmd/sip-go-agent/dispatcher_command.go @@ -4,8 +4,10 @@ import ( "context" "errors" "fmt" + "git.ipao.vip/rogee/go-sip/internal/logging" "log/slog" "net" + "os" "time" agentpb "git.ipao.vip/rogee/go-sip/gen/agent" @@ -23,13 +25,18 @@ import ( func newDispatcherCommand() *cobra.Command { var mode string + var debug bool command := &cobra.Command{ Use: "dispatcher", Short: "Run the bound Dispatcher", Args: cobra.NoArgs, SilenceUsage: true, RunE: func(cmd *cobra.Command, _ []string) error { + logging.Configure("dispatcher", debug, os.Stderr) + logging.ObserveDefaultHTTP("dispatcher") + slog.Info("Dispatcher startup", "mode", mode, "debug", debug) return runDispatcher(cmd.Context(), mode) }, } + command.Flags().BoolVarP(&debug, "debug", "D", false, "Log detailed redacted interface and MQ data") command.Flags().StringVar(&mode, "mode", "mock", "isolated Mock, SIP-only or explicit nonprod-real mode") return command } @@ -101,7 +108,7 @@ func runDispatcher(ctx context.Context, mode string) (result error) { return err } defer func() { result = errors.Join(result, broker.Close()) }() - connection, err := grpc.NewClient(agentEndpoint.Address, grpc.WithTransportCredentials(credentials.NewTLS(clientTLS))) + connection, err := grpc.NewClient(agentEndpoint.Address, grpc.WithTransportCredentials(credentials.NewTLS(clientTLS)), grpc.WithChainUnaryInterceptor(logging.UnaryClientInterceptor("dispatcher", "agent"))) if err != nil { return errors.New("Dispatcher cannot create verified Agent connection") } @@ -156,7 +163,7 @@ func runDispatcher(ctx context.Context, mode string) (result error) { if err != nil { return fmt.Errorf("Dispatcher cannot listen on configured local address: %w", err) } - server := grpc.NewServer(grpc.Creds(credentials.NewTLS(serverTLS))) + server := grpc.NewServer(grpc.Creds(credentials.NewTLS(serverTLS)), grpc.ChainUnaryInterceptor(logging.UnaryServerInterceptor("dispatcher", "agent"))) agentpb.RegisterAgentControlServiceServer(server, &rpc.RecordingServer{ DispatcherID: settings.DispatcherID, Store: database, OSS: ossClient, TrustedFingerprints: settings.PeerFingerprints, AuthorizeSession: coordinator.AuthorizeInboundMeta, diff --git a/cmd/sip-go-agent/dispatcher_http_test.go b/cmd/sip-go-agent/dispatcher_http_test.go index 4d171c7..1c24c33 100644 --- a/cmd/sip-go-agent/dispatcher_http_test.go +++ b/cmd/sip-go-agent/dispatcher_http_test.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/pem" + "git.ipao.vip/rogee/go-sip/internal/logging" "io" "net" "net/http" @@ -24,7 +25,10 @@ func TestDispatcherSaaSHTTPClientAllowsRealTargetsAndKeepsMockLocal(t *testing.T // Resolve the nonlocal SaaS hostname to the isolated test server. No request // leaves this machine; Mock must still reject the original target hostname. original := http.DefaultTransport - transport := original.(*http.Transport).Clone() + transport, err := logging.CloneDefaultTransport() + if err != nil { + t.Fatal(err) + } transport.Proxy = nil transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { if address == "saas.example.invalid:80" { diff --git a/cmd/sip-go-agent/dispatcher_sip.go b/cmd/sip-go-agent/dispatcher_sip.go index 596ea26..af76292 100644 --- a/cmd/sip-go-agent/dispatcher_sip.go +++ b/cmd/sip-go-agent/dispatcher_sip.go @@ -10,6 +10,7 @@ import ( agentpb "git.ipao.vip/rogee/go-sip/gen/agent" "git.ipao.vip/rogee/go-sip/internal/config" "git.ipao.vip/rogee/go-sip/internal/dispatcher" + "git.ipao.vip/rogee/go-sip/internal/logging" "git.ipao.vip/rogee/go-sip/internal/mq" "git.ipao.vip/rogee/go-sip/internal/rpc" "git.ipao.vip/rogee/go-sip/internal/store" @@ -67,7 +68,7 @@ func runSIPOnlyDispatcher(ctx context.Context, settings config.DispatcherRuntime return err } defer func() { result = errors.Join(result, broker.Close()) }() - connection, err := grpc.NewClient(endpoint.Address, grpc.WithTransportCredentials(credentials.NewTLS(clientTLS))) + connection, err := grpc.NewClient(endpoint.Address, grpc.WithTransportCredentials(credentials.NewTLS(clientTLS)), grpc.WithChainUnaryInterceptor(logging.UnaryClientInterceptor("dispatcher", "agent"))) if err != nil { return errors.New("SIP-only Dispatcher cannot create pinned Agent connection") } diff --git a/deploys/production/README.md b/deploys/production/README.md index ea19a29..205b00b 100644 --- a/deploys/production/README.md +++ b/deploys/production/README.md @@ -48,10 +48,14 @@ 验收包括三项服务安装、启用并启动，Asterisk/Agent `enabled+active`，外部 RabbitMQ 登录/通道、Agent监听、Asterisk基础运行及版本/hash。SaaS 未启动不阻断部署：Dispatcher可明确等待或自动重试，新执行准入保持关闭；SaaS读取、其业务队列供给、业务会话及新SIP加载延期，不把此状态写成业务已就绪。证书、配置、MQ登录及Asterisk故障不得归为SaaS不可用。没有HTTP响应须按真实错误类型判定，证书或未知传输错误阻断部署；等待状态必须是有最新对应错误的 `activating/auto-restart`，`failed` 不算成功。无获批任务、未接通或AI未调用时如实记录未验证。 +## 日志模式 + +部署入口的 `--debug` 同时开启 Dispatcher、Agent 的详细收发日志，不启用抓包。省略时使用工作流程日志；Agent 抓包独立为 `--capture`，本入口不自动设置。敏感内容排除及查看命令见 [日志说明](../../docs/logging.md)。 + ## 一键入口 ```bash -python3 deploys/production/production.py deploy +python3 deploys/production/production.py deploy --debug python3 deploys/production/production.py status python3 deploys/production/production.py stop python3 deploys/production/production.py start diff --git a/deploys/production/production.py b/deploys/production/production.py index 13b36ba..c868cb9 100644 --- a/deploys/production/production.py +++ b/deploys/production/production.py @@ -117,11 +117,14 @@ def main(): parser.add_argument('--host', default='server.sip') parser.add_argument('--known-hosts', type=Path, default=REPO / '.local/agent-call-known_hosts') parser.add_argument('--source-dir', type=Path, default=REPO) + parser.add_argument('--debug', action='store_true', help='Enable detailed redacted logs for both roles without packet capture') parser.add_argument('--reset-state', action='store_true') parser.add_argument('--confirm-reset', action='store_true', help='explicit current operator confirmation; never automatic') args = parser.parse_args() if args.host != 'server.sip': parser.error('this deployment is authorized only for server.sip') + if args.debug and args.action != 'deploy': + parser.error('--debug is only valid with deploy') if args.reset_state and (args.action != 'deploy' or not args.confirm_reset): parser.error('--reset-state requires deploy and --confirm-reset') if not args.known_hosts.is_file(): @@ -131,7 +134,7 @@ def main(): tool_dir = '/home/rogee/.local/share/go-sip-tools' ssh("python3 -c \"from pathlib import Path;import os;p=Path('" + tool_dir + "');p.mkdir(mode=0o700,exist_ok=True);os.chmod(p,0o700)\"") checked(['scp', '-q', *opts, str(HERE / 'remote.py'), args.host + ':' + tool_dir + '/production-remote.py'], timeout=30) - payload = {'action': args.action, 'reset_state': args.reset_state, 'confirm_reset': args.confirm_reset} + payload = {'action': args.action, 'reset_state': args.reset_state, 'confirm_reset': args.confirm_reset, 'debug': args.debug} if args.action in ('deploy', 'bootstrap', 'retire-mocks'): payload['source'] = sources(args.source_dir) if args.action == 'deploy': diff --git a/deploys/production/remote.py b/deploys/production/remote.py index 7f8ab0d..dbacd94 100644 --- a/deploys/production/remote.py +++ b/deploys/production/remote.py @@ -364,6 +364,9 @@ def validate_environments(probe, roles): def install(payload, p): + debug = payload.get('debug', False) + if not isinstance(debug, bool): + raise ValueError('debug must be a boolean') facts = host_check(p) source = payload['source'] # Changing ownership of existing durable data is never a normal deployment. @@ -394,7 +397,7 @@ def install(payload, p): shutil.copy2(candidate, release / name) (release / name).chmod(0o700) validate_environments(release / 'production-probe', roles) - write_private(release / 'manifest.json', json.dumps({'commit': payload['commit'], 'hashes': payload['hashes'], 'production_approval': False}, indent=2)) + write_private(release / 'manifest.json', json.dumps({'commit': payload['commit'], 'hashes': payload['hashes'], 'production_approval': False, 'debug': debug}, indent=2)) current = p['current'] if current.exists() and not current.is_symlink(): raise ValueError('current artifact pointer is not a symlink') @@ -404,7 +407,7 @@ def install(payload, p): tmp.symlink_to(release) os.replace(tmp, current) for role in ('agent', 'dispatcher'): - unit = '[Unit]\nDescription=go-sip production deployment ' + role + '\nAfter=go-sip-asterisk.service network-online.target\n\n[Service]\nType=simple\nEnvironmentFile=' + str(p['config'] / (role + '.env')) + '\nExecStart=' + str(current / 'sip-go-agent') + ' ' + role + ' --mode nonprod-real\nRestart=on-failure\nRestartSec=10\nUMask=0077\n\n[Install]\nWantedBy=default.target\n' + unit = '[Unit]\nDescription=go-sip production deployment ' + role + '\nAfter=go-sip-asterisk.service network-online.target\n\n[Service]\nType=simple\nEnvironmentFile=' + str(p['config'] / (role + '.env')) + '\nExecStart=' + str(current / 'sip-go-agent') + ' ' + role + ' --mode nonprod-real' + (' --debug' if debug else '') + '\nRestart=on-failure\nRestartSec=10\nUMask=0077\n\n[Install]\nWantedBy=default.target\n' write_private(p['units'] / ('go-sip-' + role + '.service'), unit) run(['systemctl', '--user', 'daemon-reload']) run(['sudo', '-n', 'loginctl', 'enable-linger', p['home'].name]) diff --git a/deploys/production/test_lifecycle.py b/deploys/production/test_lifecycle.py index c94640f..048fdab 100644 --- a/deploys/production/test_lifecycle.py +++ b/deploys/production/test_lifecycle.py @@ -69,6 +69,20 @@ class LifecycleTest(unittest.TestCase): self.assertFalse(manifest['production_approval']) self.assertEqual(set(manifest['hashes']), {'sip-go-agent', 'production-probe'}) + def test_debug_deployment_enables_both_roles_without_capture(self): + remote.install(dict(self.payload, debug=True), self.p) + for role in ('agent', 'dispatcher'): + unit = (self.p['units'] / ('go-sip-' + role + '.service')).read_text() + self.assertIn('--mode nonprod-real --debug', unit) + self.assertNotIn('--capture', unit) + manifest = json.loads((self.p['current'] / 'manifest.json').read_text()) + self.assertTrue(manifest['debug']) + + def test_debug_deployment_rejects_non_boolean(self): + with self.assertRaisesRegex(ValueError, 'debug'): + remote.install(dict(self.payload, debug='false'), self.p) + self.assertNotIn('stop', self.calls) + def test_explicit_reset_stops_and_archives_before_deleting(self): original = self.state.read_bytes() def backup(*args): diff --git a/deploys/test/README.md b/deploys/test/README.md index 3505dbf..d204029 100644 --- a/deploys/test/README.md +++ b/deploys/test/README.md @@ -17,7 +17,7 @@ systemd service or add it to a production host. ## Native Asterisk validation `nonprod-call-evidence.sh` is the capture-first wrapper used with -`sip-go-agent agent --debug/-D` for +`sip-go-agent agent --capture` for non-production `mock`, `mixed` and explicit `nonprod-real` call checks. It runs on a validation host with native Asterisk and required diagnostics; it is not an Asterisk or Agent replacement and is not containerized. Run it explicitly with the current @@ -33,7 +33,7 @@ restricted evidence ownership is recorded in `diagnostic-errors.txt` and fails the check; an already failed call keeps its nonzero result. Once live tcpdump and the PJSIP logger are running, the script creates a root-owned, group-readable `.active` capture arm under `--proof-root` (default -`/run/sip-go-agent/nonprod-armed`). A debug-enabled real Agent must set `AGENT_EVIDENCE_ROOT` +`/run/sip-go-agent/nonprod-armed`). A capture-enabled real Agent must set `AGENT_EVIDENCE_ROOT` to this directory; it checks the exact approved event ID, trunk, raw callee, recent arm and live capture PID before origination. The script removes the arm before stopping capture. Run one approved call per invocation, with @@ -56,7 +56,7 @@ mirror as described in [`cell/README.md`](../cell/README.md), with `AGENT_HEP_LISTEN_ADDR=127.0.0.1:` matching `capture_address` in the private `hep.conf`. Confirm `res_hep` and `res_hep_pjsip` are running and the Agent's UDP listener is bound before any explicitly authorized trial. The -capture-first wrapper is required only when Agent `--debug/-D` is enabled; +capture-first wrapper is required only when Agent `--capture` is enabled; normal Agent operation does not require tcpdump or external capture proofs. HEP's native SIP-response mirror, call-window checks and caller approval remain independent requirements. The standalone no-AI single-call tool and its optional diff --git a/docs/logging.md b/docs/logging.md new file mode 100644 index 0000000..a6957ca --- /dev/null +++ b/docs/logging.md @@ -0,0 +1,39 @@ +# Dispatcher / Agent 日志 + +两个角色统一由 Logrus 输出一行一条的 JSON 日志。已有 slog 和标准 log 调用交给同一输出，不另设日志系统。 + +## 启动 + +```sh +sip-go-agent dispatcher --mode nonprod-real --debug +sip-go-agent agent --mode nonprod-real -D +``` + +不传 `--debug/-D`：记录工作流程、交互方向、阶段、关联身份、结果、耗时和错误，不记录收发正文。传入后：保留这些记录，并增加实际 HTTP、MQ、Unary RPC 收发正文的安全副本；不截断非敏感 JSON 业务字段。 + +Agent 的抓包开关独立为 `--capture`：只在显式开启时要求 `AGENT_EVIDENCE_ROOT` 及逐通有效的外部抓包凭证。`--debug/-D` 不启动抓包、不要求 tcpdump 或证据凭证；`--capture` 也不自动开启详细日志。独立 `sip-call` 的 `--debug/-D` 仍保留原有抓包含义，本次不改。 + +## 记录内容 + +- HTTP：请求开始、实际被读取的请求/响应 JSON、响应状态、正文读取结果。认证头不记录；URL 的认证、查询和 fragment 部分删除。SDK/native ARI 的 HTTP 客户端使用同一观察输出，保留 TLS、超时、重定向和 WebSocket 升级行为。 +- MQ：实际入站/出站字节；接收、校验、处理、ACK/NACK/Reject、发布及 broker confirm。`publisher_confirmed` 只证明 RabbitMQ 确认，记录 `saas_processed=false`，不能据此声称 SaaS 已处理。 +- RPC：实际边界的 protobuf 请求/响应投影，明确 `wire_encoding=protobuf`。冻结配置中的 `*_json` 字节展开为 JSON，其余字节仅记录长度；不是业务内部 CallSpec，也不声称 JSON 是 protobuf 原始线格式。客户端响应收到和服务端响应准备完成分别记录，服务端返回不能冒充对端收讫。 +- 业务流程：持久接纳/重复忽略、等待准入、选线、预占、请求执行、Agent 排程/执行/结束、结果回报及恢复。 + +`event_id/task_id/call_id/request_id` 等已有标识用于关联记录；只有组件实际掌握的标识才输出。 + +## 正文边界 + +密钥、Token、认证信息、完整 prompt/messages/transcript/text/content/结束语、原始 SIP 内容及音频不输出。敏感字段按键名隐藏，重复键逐个处理；其他 JSON 的空白、重复键和数字写法保留，原业务字节与对象不修改。HTTP 二进制和升级后的 WebSocket 流不进入正文日志。 + +非法 JSON、无法安全展开的嵌套数据或部分读取不能退回打印未经处理的正文：输出 `redaction_error`、字节数及 SHA-256。HTTP `body_complete=false` 只表示已观测部分，不能把它当作完整网络正文；原读取/关闭错误继续返回，不因日志改变重试、确认或业务结果。 + +## 固定环境部署及查看 + +```sh +python3 deploys/production/production.py deploy --host server.sip --debug +journalctl --user -u go-sip-dispatcher.service -f -o cat +journalctl --user -u go-sip-agent.service -f -o cat +``` + +部署时 `--debug` 同时写入两个固定服务，不增加 `--capture`，普通重部署不清数据。实际收发详情仅出现于该交互确实发生时；不得为了验证日志生成外呼消息或独立调用真实 AI/OSS。 diff --git a/go.mod b/go.mod index 71b3def..55eb2b1 100644 --- a/go.mod +++ b/go.mod @@ -46,6 +46,7 @@ require ( github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/rotisserie/eris v0.4.1 // indirect + github.com/sirupsen/logrus v1.9.4 // indirect github.com/spf13/pflag v1.0.9 // indirect github.com/tidwall/gjson v1.19.0 // indirect github.com/tidwall/match v1.1.1 // indirect diff --git a/go.sum b/go.sum index 0476f84..cf51258 100644 --- a/go.sum +++ b/go.sum @@ -96,6 +96,8 @@ github.com/santhosh-tekuri/jsonschema/v6 v6.0.3 h1:1EYB5IzjZawrrnELUi78f9fPu57Hu github.com/santhosh-tekuri/jsonschema/v6 v6.0.3/go.mod h1:JXeL+ps8p7/KNMjDQk3TCwPpBy0wYklyWTfbkIzdIFU= github.com/shirou/gopsutil/v4 v4.26.8 h1:YQMTF/1J50B5+Y0vlo1eDRf5DoR7Gk69hY+8wjYkQeo= github.com/shirou/gopsutil/v4 v4.26.8/go.mod h1:5O9FjBiXoTDFatIWjZZosqj4pV0DRtLx598xGbBehzM= +github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= +github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= github.com/spf13/cobra v1.10.1 h1:lJeBwCfmrnXthfAupyUTzJ/J4Nc1RsHC/mSRU2dll/s= github.com/spf13/cobra v1.10.1/go.mod h1:7SmJGaTHFVBY0jW4NXGluQoLvhqFQM+6XSKD+P4XaB0= github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY= diff --git a/internal/agent/recording_scan.go b/internal/agent/recording_scan.go index 78dc12b..bd5af23 100644 --- a/internal/agent/recording_scan.go +++ b/internal/agent/recording_scan.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io/fs" + "log/slog" "os" "path/filepath" "strings" @@ -16,6 +17,7 @@ import ( // Blocked/corrupt entries remain untouched and are returned as errors while // other independent entries continue; none can produce a fabricated result. func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant, report RecordingResultReporter) (int, error) { + slog.InfoContext(ctx, "Recording recovery scan", "stage", "recovery_started") if strings.TrimSpace(r.Root) == "" { return 0, errors.New("recording recovery root is required") } @@ -96,6 +98,11 @@ func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant, } return nil }) + if len(blocked) > 0 || walkErr != nil { + slog.ErrorContext(ctx, "Recording recovery blocked", "stage", "recovery_blocked", "processed", processed, "blocked", len(blocked), "walk_error", walkErr) + } else { + slog.InfoContext(ctx, "Recording recovery scan complete", "stage", "recovery_completed", "processed", processed) + } return processed, errors.Join(append(blocked, walkErr)...) } diff --git a/internal/agent/result_journal.go b/internal/agent/result_journal.go index 1cc8a7a..dd7cc0f 100644 --- a/internal/agent/result_journal.go +++ b/internal/agent/result_journal.go @@ -9,7 +9,7 @@ import ( "errors" "fmt" "io" - "log" + "log/slog" "os" "path/filepath" "strings" @@ -249,7 +249,7 @@ func (j ResultJournal) RecoverWithEnd(ctx context.Context, confirmEnd, report fu entry.State = "ready" } if entry.State != "ready" { - log.Printf("Agent recovery blocked stage=original_upload_outcome event_digest=%s", item.Name()[:12]) + slog.WarnContext(ctx, "Agent result recovery blocked", "stage", "original_upload_outcome", "event_digest", item.Name()[:12]) blocked = append(blocked, errors.New("original upload outcome requires inspection")) continue } @@ -265,5 +265,10 @@ func (j ResultJournal) RecoverWithEnd(ctx context.Context, confirmEnd, report fu blocked = append(blocked, fmt.Errorf("confirm original result: %w", err)) } } + if len(blocked) > 0 { + slog.ErrorContext(ctx, "Agent result recovery incomplete", "stage", "result_recovery_blocked", "blocked", len(blocked)) + } else { + slog.InfoContext(ctx, "Agent result recovery complete", "stage", "result_recovery_completed") + } return errors.Join(blocked...) } diff --git a/internal/agent/upload.go b/internal/agent/upload.go index 80bc67a..ee5a008 100644 --- a/internal/agent/upload.go +++ b/internal/agent/upload.go @@ -7,6 +7,7 @@ import ( "encoding/hex" "errors" "fmt" + "git.ipao.vip/rogee/go-sip/internal/logging" "hash" "io" "net/http" @@ -162,7 +163,7 @@ func (c UploadClient) putValidated(ctx context.Context, grant *agentpb.UploadGra if client == nil { client = &http.Client{} } - copyClient := *client + copyClient := *logging.HTTPClient(client, "agent", "oss") copyClient.CheckRedirect = func(_ *http.Request, _ []*http.Request) error { return http.ErrUseLastResponse } resp, err := copyClient.Do(req) if err != nil { diff --git a/internal/ai/pipeline.go b/internal/ai/pipeline.go index 8baec1a..357d555 100644 --- a/internal/ai/pipeline.go +++ b/internal/ai/pipeline.go @@ -11,6 +11,7 @@ import ( "sync" "git.ipao.vip/rogee/doubao-speech-go" + "git.ipao.vip/rogee/go-sip/internal/logging" "github.com/openai/openai-go/v3" "github.com/openai/openai-go/v3/option" ) @@ -195,6 +196,7 @@ func (b Binding) Complete(ctx context.Context, finalUserText string) (string, er option.WithAPIKey(b.LLM.Provider.Credential), option.WithBaseURL(strings.TrimRight(b.LLM.Provider.Endpoint, "/")), option.WithMaxRetries(0), + option.WithHTTPClient(logging.HTTPClient(nil, "agent", "llm")), ) body := cloneParams(b.LLM.Params) body["model"], _ = json.Marshal(b.LLM.Model) diff --git a/internal/config/agent_debug_test.go b/internal/config/agent_debug_test.go index ec76d51..91cf32c 100644 --- a/internal/config/agent_debug_test.go +++ b/internal/config/agent_debug_test.go @@ -5,7 +5,7 @@ import ( "testing" ) -func TestAgentDebugEvidenceIsOptIn(t *testing.T) { +func TestAgentCaptureEvidenceIsIndependentOfDebug(t *testing.T) { setAgentEnvironment(t) t.Setenv("AGENT_MOCK_SCENARIO_FILE", "") t.Setenv("AGENT_MOCK_APPLIED_SIP_FILE", "") @@ -14,17 +14,18 @@ func TestAgentDebugEvidenceIsOptIn(t *testing.T) { } s, err := LoadAgentEnvironment("nonprod-real") if err != nil { - t.Fatalf("normal operation must not require external capture: %v", err) - } - if err = s.ValidateDebug(); err != nil { - t.Fatal(err) + t.Fatalf("normal operation requires no external capture: %v", err) } s.Debug = true - if err = s.ValidateDebug(); err == nil || !strings.Contains(err.Error(), "AGENT_EVIDENCE_ROOT") { - t.Fatalf("debug requires an absolute evidence root: %v", err) + if err = s.ValidateCapture(); err != nil { + t.Fatalf("Debug logging must not require capture: %v", err) + } + s.Capture = true + if err = s.ValidateCapture(); err == nil || !strings.Contains(err.Error(), "AGENT_EVIDENCE_ROOT") { + t.Fatalf("capture requires absolute evidence root: %v", err) } s.EvidenceRoot = "/tmp/evidence" - if err = s.ValidateDebug(); err != nil { + if err = s.ValidateCapture(); err != nil { t.Fatal(err) } } diff --git a/internal/config/agent_runtime.go b/internal/config/agent_runtime.go index 6e5e15b..4a62b39 100644 --- a/internal/config/agent_runtime.go +++ b/internal/config/agent_runtime.go @@ -16,6 +16,7 @@ import ( // durable file, media adapter, credential or network connection. type AgentEnvironment struct { Debug bool + Capture bool Mode string AgentID string CellID string @@ -136,10 +137,10 @@ func LoadAgentEnvironment(mode string) (AgentEnvironment, error) { return settings, nil } -// ValidateDebug only requires external capture configuration when explicitly enabled. -func (s AgentEnvironment) ValidateDebug() error { - if s.Debug && s.Mode == "nonprod-real" && !filepath.IsAbs(s.EvidenceRoot) { - return errors.New("AGENT_EVIDENCE_ROOT must be absolute when --debug/-D is enabled") +// ValidateCapture requires external evidence only for explicit packet capture. +func (s AgentEnvironment) ValidateCapture() error { + if s.Capture && s.Mode == "nonprod-real" && !filepath.IsAbs(s.EvidenceRoot) { + return errors.New("AGENT_EVIDENCE_ROOT must be absolute when --capture is enabled") } return nil } diff --git a/internal/configread/client.go b/internal/configread/client.go index a484f7e..a38992f 100644 --- a/internal/configread/client.go +++ b/internal/configread/client.go @@ -10,6 +10,8 @@ import ( "net/http" "net/url" "strings" + + "git.ipao.vip/rogee/go-sip/internal/logging" ) const dispatcherIDHeader = "X-DISPATCHER-id" @@ -49,7 +51,7 @@ func NewClient(baseURL, dispatcherID, secret string, httpClient *http.Client) (* if httpClient == nil { httpClient = http.DefaultClient } - return &Client{baseURL: strings.TrimRight(baseURL, "/"), dispatcherID: dispatcherID, secret: secret, httpClient: httpClient}, nil + return &Client{baseURL: strings.TrimRight(baseURL, "/"), dispatcherID: dispatcherID, secret: secret, httpClient: logging.HTTPClient(httpClient, "dispatcher", "saas")}, nil } func (c *Client) DispatcherID() string { return c.dispatcherID } diff --git a/internal/dispatcher/execute.go b/internal/dispatcher/execute.go index b3eda84..187c9f6 100644 --- a/internal/dispatcher/execute.go +++ b/internal/dispatcher/execute.go @@ -5,7 +5,7 @@ import ( "encoding/json" "errors" "fmt" - "log" + "log/slog" "time" "git.ipao.vip/rogee/go-sip/internal/configread" @@ -90,11 +90,14 @@ func (c *ExecuteController) ProcessExecute(ctx context.Context, body []byte) err TenantID: event.TenantID, TaskID: event.Payload.TaskID, Callee: event.Payload.Callee, IssuedAt: event.IssuedAt, } - stored, _, err := c.Store.RecordExecute(cmd) + logger := slog.With("event_id", cmd.EventID, "task_id", cmd.TaskID, "tenant_id", cmd.TenantID) + stored, inserted, err := c.Store.RecordExecute(cmd) if errors.Is(err, store.ErrEventRejected) { + logger.Info("Rejected call identity replayed without execution", "stage", "idempotent_rejection") return nil } if errors.Is(err, store.ErrStopped) { + logger.Info("Call instruction ignored", "stage", "admission_closed", "reason", "task stopped") // The stopped task never accepted this old queued command. RabbitMQ // may ACK it without creating an execution or a per-call result. return nil @@ -102,7 +105,9 @@ func (c *ExecuteController) ProcessExecute(ctx context.Context, body []byte) err if err != nil { return fmt.Errorf("durable inbound call %q: %w", cmd.EventID, err) } + logger.Info("Call instruction saved", "stage", "inbox_saved", "inserted", inserted, "status", stored.Status) if stored.Status != "pending" { + logger.Info("Call instruction already handled", "stage", "idempotent_skip", "status", stored.Status) return nil } if !validRawCallee(cmd.Callee) { @@ -147,6 +152,8 @@ func agentProvedCallNotIssued(err error) bool { } func (c *ExecuteController) dispatchPending(ctx context.Context, cmd store.ExecuteCommand) error { + logger := slog.With("event_id", cmd.EventID, "task_id", cmd.TaskID, "tenant_id", cmd.TenantID) + logger.Info("Call admission check", "stage", "select_line") at := c.Now() issued, err := time.Parse(time.RFC3339Nano, cmd.IssuedAt) if err != nil { @@ -176,6 +183,7 @@ func (c *ExecuteController) dispatchPending(ctx context.Context, cmd store.Execu } choice, err := SelectTrunk(snapshot, cmd.Callee, at, occupied, taskOccupied, loaded) if errors.Is(err, ErrRuleWait) { + logger.Info("Call instruction waiting", "stage", "admission_wait", "reason", err) return nil } if err != nil { @@ -235,18 +243,20 @@ func (c *ExecuteController) dispatchPending(ctx context.Context, cmd store.Execu if err != nil { return errors.Join(fmt.Errorf("read reserved task trunk limit: %w", err), c.Store.CancelReservationBeforeOrigin(cmd.DispatcherID, cmd.EventID)) } - log.Printf("Dispatcher call reserved before originate: event_id=%q task_revision=%d trunk_id=%q task_trunk_limit=%d task_trunk_used_before=%d", cmd.EventID, snapshot.Task.TaskRevision, choice.TrunkID, limits[choice.TrunkID], taskOccupied[choice.TrunkID]) + logger.Info("Call reservation saved", "stage", "reserved", "task_revision", snapshot.Task.TaskRevision, "trunk_id", choice.TrunkID, "task_trunk_limit", limits[choice.TrunkID], "task_trunk_used_before", taskOccupied[choice.TrunkID]) + logger.Info("Call execution request", "direction", "dispatcher->agent", "stage", "originate_requested", "trunk_id", choice.TrunkID) if err := c.Originator.Originate(ctx, spec); err != nil { if agentProvedCallNotIssued(err) { rejectErr := c.Store.RejectUnissuedExecute(cmd.DispatcherID, cmd.EventID, "Agent refused before outbound call") if rejectErr == nil { - log.Printf("Dispatcher call refused before dial: event_id=%q agent_code=%s", cmd.EventID, status.Code(err)) + logger.Warn("Agent refused before dial", "direction", "agent->dispatcher", "stage", "rejected_before_dial", "agent_code", status.Code(err)) } return errors.Join(fmt.Errorf("Agent rejected call %q before dialing: %w", cmd.EventID, err), rejectErr) } unknownErr := c.Store.MarkExecuteUnknown(cmd.DispatcherID, cmd.EventID) return errors.Join(fmt.Errorf("originator outcome unknown for call %q: %w", cmd.EventID, err), unknownErr) } + logger.Info("Agent scheduled call", "direction", "agent->dispatcher", "stage", "originate_scheduled", "trunk_id", choice.TrunkID) if err := c.Store.MarkExecuteDispatched(cmd.DispatcherID, cmd.EventID); err != nil { return fmt.Errorf("originated call %q has no durable acknowledgment: %w", cmd.EventID, err) } diff --git a/internal/logging/http.go b/internal/logging/http.go new file mode 100644 index 0000000..4d5ffeb --- /dev/null +++ b/internal/logging/http.go @@ -0,0 +1,166 @@ +package logging + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "log/slog" + "mime" + "net/http" + "strings" + "sync" + "time" +) + +// HTTPClient clones the client, retaining its TLS, redirect, timeout and custom +// transport settings. Only bytes actually read at the boundary are observed. +func HTTPClient(client *http.Client, local, remote string) *http.Client { + if client == nil { + client = http.DefaultClient + } + clone := *client + transport := client.Transport + if transport == nil { + transport = http.DefaultTransport + } + if observed, ok := transport.(*httpObserver); ok { + transport = observed.next + } + clone.Transport = &httpObserver{next: transport, local: local, remote: remote} + return &clone +} + +// ObserveDefaultHTTP covers library-owned HTTP clients (including native ARI) +// whose supported API does not expose a transport. Call only at service startup. +func ObserveDefaultHTTP(local string) { + next := http.DefaultTransport + if observed, ok := next.(*httpObserver); ok { + next = observed.next + } + http.DefaultTransport = &httpObserver{next: next, local: local} +} + +// CloneDefaultTransport retains explicit transport settings used by isolated +// TLS tests. A custom non-HTTP transport is an error, not a silent fallback. +func CloneDefaultTransport() (*http.Transport, error) { + base := http.DefaultTransport + if observed, ok := base.(*httpObserver); ok { + base = observed.next + } + transport, ok := base.(*http.Transport) + if !ok { + return nil, fmt.Errorf("default HTTP transport cannot be cloned: %T", base) + } + return transport.Clone(), nil +} + +type httpObserver struct { + next http.RoundTripper + local, remote string +} + +func (t *httpObserver) RoundTrip(req *http.Request) (*http.Response, error) { + started := time.Now() + remote := t.remote + if remote == "" { + remote = req.URL.Host + if remote == "127.0.0.1:8088" { + remote = "asterisk" + } + } + attrs := []any{"transport", "http", "method", req.Method, "url", req.URL.String()} + ctx := req.Context() + slog.InfoContext(ctx, "HTTP request", append(attrs, "direction", t.local+"->"+remote, "stage", "request_start")...) + clone := req.Clone(ctx) + if req.Body != nil { + clone.Body = newObservedBody(ctx, req.Body, req.Header.Get("Content-Type"), "HTTP request data", append(attrs, "direction", t.local+"->"+remote)...) + } + res, err := t.next.RoundTrip(clone) + result := append(attrs, "direction", remote+"->"+t.local, "duration_ms", time.Since(started).Milliseconds()) + if err != nil { + slog.ErrorContext(ctx, "HTTP failed", append(result, "stage", "transport_failed", "error", err)...) + return res, err + } + slog.InfoContext(ctx, "HTTP response", append(result, "stage", "response_received", "http_status", res.StatusCode, "content_type", res.Header.Get("Content-Type"))...) + // HTTP upgrades require the original read/write stream (ASR/TTS WebSocket). + if res.StatusCode == http.StatusSwitchingProtocols { + return res, nil + } + if res.Body != nil { + res.Body = newObservedBody(ctx, res.Body, res.Header.Get("Content-Type"), "HTTP response data", append(result, "http_status", res.StatusCode)...) + } + return res, nil +} + +type observedBody struct { + source io.ReadCloser + ctx context.Context + message string + attrs []any + collect bool + mu sync.Mutex + data bytes.Buffer + count int64 + done bool + started time.Time +} + +func newObservedBody(ctx context.Context, source io.ReadCloser, contentType, message string, attrs ...any) *observedBody { + media, _, err := mime.ParseMediaType(contentType) + isJSON := err == nil && (media == "application/json" || strings.HasSuffix(media, "+json")) + mode := "binary_omitted" + if isJSON { + mode = "json" + } + return &observedBody{source: source, ctx: ctx, message: message, attrs: append(append([]any{}, attrs...), "content_type", contentType, "body_mode", mode), collect: isJSON && slog.Default().Enabled(ctx, slog.LevelDebug), started: time.Now()} +} +func (r *observedBody) Read(p []byte) (int, error) { + n, err := r.source.Read(p) + r.mu.Lock() + defer r.mu.Unlock() + r.count += int64(n) + if r.collect && !r.done { + r.data.Write(p[:n]) + } + if errors.Is(err, io.EOF) { + r.finishLocked(true, nil) + } else if err != nil { + r.finishLocked(false, err) + } + return n, err +} +func (r *observedBody) Close() error { + err := r.source.Close() + r.mu.Lock() + defer r.mu.Unlock() + r.finishLocked(false, err) + if err != nil { + slog.ErrorContext(r.ctx, "HTTP body close failed", append(r.attrs, "stage", "body_close_failed", "error", err)...) + } + return err +} +func (r *observedBody) finishLocked(complete bool, err error) { + if r.done { + return + } + r.done = true + stage := "body_read" + if !complete { + stage = "body_read_incomplete" + } + attrs := append(append([]any{}, r.attrs...), "stage", stage, "bytes_read", r.count, "body_duration_ms", time.Since(r.started).Milliseconds()) + if err != nil { + slog.ErrorContext(r.ctx, "HTTP body read failed", append(attrs, "error", err)...) + } else { + slog.InfoContext(r.ctx, "HTTP body observation", attrs...) + } + if r.collect { + // A decoder may close after consuming a valid JSON value without reading + // EOF. Log observed bytes without claiming a complete wire body; malformed + // prefixes still produce an explicit safe redaction diagnostic. + Body(r.ctx, r.message, r.data.Bytes(), append(attrs, "body_complete", complete)...) + } + r.data.Reset() +} diff --git a/internal/logging/http_regression_test.go b/internal/logging/http_regression_test.go new file mode 100644 index 0000000..a681d16 --- /dev/null +++ b/internal/logging/http_regression_test.go @@ -0,0 +1,94 @@ +package logging + +import ( + "bytes" + "context" + "errors" + "io" + "net/http" + "strings" + "testing" +) + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +type upgradeStream struct{ bytes.Buffer } + +func (*upgradeStream) Close() error { return nil } + +func TestHTTPUpgradeRetainsOriginalReadWriteStream(t *testing.T) { + installTestLogger(t, true) + stream := &upgradeStream{} + client := HTTPClient(&http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: 101, Header: make(http.Header), Body: stream}, nil + })}, "agent", "asr") + req, _ := http.NewRequest(http.MethodGet, "http://example.test/ws", nil) + res, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + if res.Body != stream { + t.Fatal("WebSocket upgraded stream was decorated") + } + if _, ok := res.Body.(io.ReadWriteCloser); !ok { + t.Fatal("WebSocket writer lost") + } +} + +func TestDefaultTransportObservationClonesBaseAndDoesNotDoubleLog(t *testing.T) { + previous := http.DefaultTransport + t.Cleanup(func() { http.DefaultTransport = previous }) + http.DefaultTransport = &http.Transport{} + ObserveDefaultHTTP("agent") + clone, err := CloneDefaultTransport() + if err != nil || clone == nil { + t.Fatal(err) + } + out := installTestLogger(t, false) + base := roundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: 200, Header: make(http.Header), Body: http.NoBody}, nil + }) + http.DefaultTransport = &httpObserver{next: base, local: "agent"} + client := HTTPClient(nil, "agent", "llm") + req, _ := http.NewRequest(http.MethodGet, "http://example.test/", nil) + res, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + res.Body.Close() + if strings.Count(out.String(), `"stage":"request_start"`) != 1 { + t.Fatal(out.String()) + } + if _, err = CloneDefaultTransport(); err == nil { + t.Fatal("custom unsupported transport must be explicit") + } +} + +func TestHTTPReadFailureIsReportedWithoutLeakingIncompleteBody(t *testing.T) { + out := installTestLogger(t, true) + expected := errors.New("read failed") + body := newObservedBody(context.Background(), &brokenBody{err: expected}, "application/json", "HTTP response data") + if _, err := io.ReadAll(body); !errors.Is(err, expected) { + t.Fatal(err) + } + body.Close() + if strings.Contains(out.String(), "private") || !strings.Contains(out.String(), "body_read_incomplete") || !strings.Contains(out.String(), "read failed") { + t.Fatal(out.String()) + } +} + +type brokenBody struct { + err error + read bool +} + +func (b *brokenBody) Read(p []byte) (int, error) { + if b.read { + return 0, b.err + } + b.read = true + return copy(p, `{"api_key":"private`), b.err +} +func (*brokenBody) Close() error { return nil } diff --git a/internal/logging/http_test.go b/internal/logging/http_test.go new file mode 100644 index 0000000..1979077 --- /dev/null +++ b/internal/logging/http_test.go @@ -0,0 +1,88 @@ +package logging + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestHTTPLogsRawBoundaryBytesAndPreservesBodies(t *testing.T) { + for _, debug := range []bool{false, true} { + t.Run(map[bool]string{false: "normal", true: "debug"}[debug], func(t *testing.T) { + out := installTestLogger(t, debug) + requestBody := `{ "task_id":"4", "api_key":"private-key", "params":{"speed":1.0}}` + responseBody := `{ "status":"ok", "content":"private-dialogue", "n":1e+02 }` + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, e := io.ReadAll(r.Body) + if e != nil || string(b) != requestBody { + t.Errorf("business body changed: %s %v", b, e) + } + if r.Header.Get("Authorization") != "Bearer private-header" { + t.Error("header changed") + } + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Set-Cookie", "private-cookie") + w.Write([]byte(responseBody)) + })) + defer server.Close() + req, e := http.NewRequestWithContext(context.Background(), http.MethodPost, server.URL+"/task/4", strings.NewReader(requestBody)) + if e != nil { + t.Fatal(e) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer private-header") + res, e := HTTPClient(server.Client(), "dispatcher", "saas").Do(req) + if e != nil { + t.Fatal(e) + } + b, e := io.ReadAll(res.Body) + res.Body.Close() + if e != nil || string(b) != responseBody { + t.Fatal("response changed", e) + } + text := out.String() + for _, secret := range []string{"private-key", "private-dialogue", "private-header", "private-cookie"} { + if strings.Contains(text, secret) { + t.Fatalf("leak: %s", text) + } + } + for _, expected := range []string{"dispatcher->saas", "saas->dispatcher", "request_start", "response_received", "body_read", "duration_ms"} { + if !strings.Contains(text, expected) { + t.Fatalf("missing %s: %s", expected, text) + } + } + if strings.Contains(text, `\"n\":1e+02`) != debug { + t.Fatal("debug raw bytes or normal omission failed", text) + } + }) + } +} + +func TestHTTPBinaryIsNotLoggedAndSignedURLIsMasked(t *testing.T) { + out := installTestLogger(t, true) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + w.Header().Set("Content-Type", "audio/wav") + w.Write([]byte("private-audio")) + })) + defer server.Close() + req, _ := http.NewRequest(http.MethodPut, server.URL+"/recording?signature=private-url", strings.NewReader("private-recording")) + req.Header.Set("Content-Type", "audio/wav") + res, e := HTTPClient(server.Client(), "agent", "oss").Do(req) + if e != nil { + t.Fatal(e) + } + io.Copy(io.Discard, res.Body) + res.Body.Close() + for _, secret := range []string{"private-url", "private-recording", "private-audio"} { + if strings.Contains(out.String(), secret) { + t.Fatal(out.String()) + } + } + if !strings.Contains(out.String(), "binary_omitted") { + t.Fatal(out.String()) + } +} diff --git a/internal/logging/logging.go b/internal/logging/logging.go new file mode 100644 index 0000000..6c3f4f8 --- /dev/null +++ b/internal/logging/logging.go @@ -0,0 +1,114 @@ +// Package logging provides the single Logrus output for both service roles. +// The slog handler keeps existing structured call sites and injected loggers, +// without maintaining a second output or a second level configuration. +package logging + +import ( + "context" + "crypto/sha256" + "fmt" + "io" + "log/slog" + "time" + + "github.com/sirupsen/logrus" +) + +func Configure(role string, debug bool, output io.Writer) { + backend := logrus.New() + backend.SetOutput(output) + backend.SetFormatter(&logrus.JSONFormatter{TimestampFormat: time.RFC3339Nano, DisableHTMLEscape: true}) + if debug { + backend.SetLevel(logrus.DebugLevel) + } else { + backend.SetLevel(logrus.InfoLevel) + } + slog.SetDefault(slog.New(&handler{backend: backend, fields: logrus.Fields{"role": role}})) +} + +type handler struct { + backend *logrus.Logger + fields logrus.Fields + group string +} + +func level(l slog.Level) logrus.Level { + switch { + case l >= slog.LevelError: + return logrus.ErrorLevel + case l >= slog.LevelWarn: + return logrus.WarnLevel + case l >= slog.LevelInfo: + return logrus.InfoLevel + default: + return logrus.DebugLevel + } +} +func (h *handler) Enabled(_ context.Context, l slog.Level) bool { + return h.backend.IsLevelEnabled(level(l)) +} +func (h *handler) Handle(_ context.Context, r slog.Record) error { + fields := h.copyFields() + r.Attrs(func(a slog.Attr) bool { h.add(fields, a); return true }) + h.backend.WithFields(fields).WithTime(r.Time).Log(level(r.Level), cleanString(r.Message)) + return nil +} +func (h *handler) copyFields() logrus.Fields { + fields := make(logrus.Fields, len(h.fields)+8) + for k, v := range h.fields { + fields[k] = v + } + return fields +} +func (h *handler) add(fields logrus.Fields, a slog.Attr) { + a.Value = a.Value.Resolve() + if a.Equal(slog.Attr{}) { + return + } + if a.Value.Kind() == slog.KindGroup { + group := h.group + if a.Key != "" { + group += a.Key + "." + } + nested := *h + nested.group = group + for _, child := range a.Value.Group() { + nested.add(fields, child) + } + return + } + fields[h.group+a.Key] = safeValue(a.Key, a.Value.Any()) +} +func (h *handler) WithAttrs(attrs []slog.Attr) slog.Handler { + clone := *h + clone.fields = h.copyFields() + for _, a := range attrs { + clone.add(clone.fields, a) + } + return &clone +} +func (h *handler) WithGroup(name string) slog.Handler { + if name == "" { + return h + } + clone := *h + clone.group += name + "." + return &clone +} + +// Body observes actual boundary bytes. Redaction happens only when Debug is on +// and never mutates the original data. Unsafe/malformed bodies produce an +// explicit diagnostic, not an unsanitized fallback or a successful parse claim. +func Body(ctx context.Context, message string, body []byte, attrs ...any) { + if !slog.Default().Enabled(ctx, slog.LevelDebug) { + return + } + attrs = append(append([]any{}, attrs...), "body_bytes", len(body), "body_sha256", fmt.Sprintf("%x", sha256.Sum256(body))) + safe, err := RedactJSON(body) + if err != nil { + attrs = append(attrs, "redaction_error", err.Error(), "body", "[OMITTED: unsafe or invalid JSON]") + } else { + attrs = append(attrs, "body", safe, "body_encoding", "json") + } + slog.DebugContext(ctx, message, attrs...) +} diff --git a/internal/logging/logging_test.go b/internal/logging/logging_test.go new file mode 100644 index 0000000..37a5d35 --- /dev/null +++ b/internal/logging/logging_test.go @@ -0,0 +1,101 @@ +package logging + +import ( + "bytes" + "context" + "encoding/json" + "log" + "log/slog" + "strings" + "testing" +) + +func installTestLogger(t *testing.T, debug bool) *bytes.Buffer { + t.Helper() + previous := slog.Default() + out := new(bytes.Buffer) + Configure("dispatcher", debug, out) + t.Cleanup(func() { slog.SetDefault(previous) }) + return out +} + +func TestLogrusModesIncludeWorkflowButOnlyDebugIncludesBody(t *testing.T) { + for _, debug := range []bool{false, true} { + t.Run(map[bool]string{false: "normal", true: "debug"}[debug], func(t *testing.T) { + out := installTestLogger(t, debug) + slog.Info("HTTP request", "direction", "dispatcher->saas", "stage", "send", "task_id", "4") + Body(context.Background(), "HTTP request data", []byte(`{ "task_id" : "4", "concurrency":2 }`), "direction", "dispatcher->saas") + log.Print("existing workflow") + text := out.String() + for _, expected := range []string{`"role":"dispatcher"`, `"level":"info"`, `"stage":"send"`, `existing workflow`} { + if !strings.Contains(text, expected) { + t.Fatalf("missing %s: %s", expected, text) + } + } + if strings.Contains(text, `\"concurrency\":2`) != debug { + t.Fatalf("debug body=%v: %s", debug, text) + } + for _, line := range strings.Split(strings.TrimSpace(text), "\n") { + if !json.Valid([]byte(line)) { + t.Fatalf("not Logrus JSON: %s", line) + } + } + }) + } +} + +func TestRawJSONRedactionPreservesNonSensitiveBytesAndDuplicates(t *testing.T) { + body := []byte("{ \"task_id\":\"4\", \"n\":1e+02, \"n\":2, \"api_key\":\"key-secret\", \"api_key\":\"second-secret\", \"nested\":{\"accessKeySecret\":\"third-secret\",\"params\":{\"speed\":1.0}},\"prompt\":\"private prompt\",\"transcript\":[{\"text\":\"private talk\"}],\"session_credential\":\"session-secret\",\"target_url\":\"https://bucket.test/key?signature=private\",\"raw\":\"private SIP\" }") + original := bytes.Clone(body) + safe, err := RedactJSON(body) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(body, original) { + t.Fatal("redaction modified business bytes") + } + for _, secret := range []string{"key-secret", "second-secret", "third-secret", "private prompt", "private talk", "session-secret", "signature=private", "private SIP"} { + if strings.Contains(safe, secret) { + t.Fatalf("leaked %s", secret) + } + } + for _, unchanged := range []string{`"task_id":"4"`, `"n":1e+02, "n":2`, `"params":{"speed":1.0}`} { + if !strings.Contains(safe, unchanged) { + t.Fatalf("raw representation changed: %s", safe) + } + } + if !json.Valid([]byte(safe)) { + t.Fatalf("invalid sanitized JSON: %s", safe) + } +} + +func TestInvalidBodyProducesExplicitDiagnosticWithoutRawLeak(t *testing.T) { + out := installTestLogger(t, true) + Body(context.Background(), "MQ receive data", []byte(`{"api_key":"do-not-print`), "direction", "mq->dispatcher") + text := out.String() + if strings.Contains(text, "do-not-print") || !strings.Contains(text, "redaction_error") || !strings.Contains(text, "body_sha256") { + t.Fatalf("unsafe/hidden invalid body: %s", text) + } +} + +func TestDebugBodyIsNotTruncated(t *testing.T) { + out := installTestLogger(t, true) + value := strings.Repeat("x", 150000) + "end-of-business-data" + body := []byte(`{"description":"` + value + `"}`) + Body(context.Background(), "MQ data", body) + var record map[string]any + if err := json.Unmarshal(bytes.TrimSpace(out.Bytes()), &record); err != nil { + t.Fatal(err) + } + if record["body"] != string(body) { + t.Fatal("raw business body was truncated or rewritten") + } +} + +func TestSensitiveWorkflowFieldsAndURLsNeverLeak(t *testing.T) { + out := installTestLogger(t, true) + slog.Info("workflow", "Authorization", "Bearer do-not-print", "apiKey", "do-not-print", "prompt", "do-not-print", "url", "https://user:pass@host/path?signature=do-not-print") + if strings.Contains(out.String(), "do-not-print") || strings.Contains(out.String(), "user:pass") { + t.Fatal(out.String()) + } +} diff --git a/internal/logging/redact.go b/internal/logging/redact.go new file mode 100644 index 0000000..df15742 --- /dev/null +++ b/internal/logging/redact.go @@ -0,0 +1,184 @@ +package logging + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/url" + "regexp" + "sort" + "strings" + "unicode" +) + +const redacted = "[REDACTED]" + +func sensitive(key string) bool { + normalized := strings.Map(func(r rune) rune { + if unicode.IsLetter(r) || unicode.IsDigit(r) { + return unicode.ToLower(r) + } + return -1 + }, key) + switch normalized { + case "content", "headers", "history", "sentence", "sentences", "words", "openingremark", "openingremarks", "greeting", "sessionnonce", "credential", "credentials", "key", "authorization", "proxyauthorization", "apikey", "accesskeyid", "accesskeysecret", "appkey", "appsecret", "secret", "secretkey", "password", "passwd", "token", "accesstoken", "refreshtoken", "uploadtoken", "providertoken", "sessioncredential", "clientnonce", "servernonce", "cookie", "setcookie", "prompt", "systemprompt", "instructions", "messages", "messagecontent", "transcript", "transcription", "finaltranscript", "text", "utterance", "audio", "audiodata", "recordingdata", "raw", "closingremark", "signature", "targeturl": + return true + } + return strings.HasSuffix(normalized, "password") || strings.HasSuffix(normalized, "secret") || strings.HasSuffix(normalized, "apikey") || strings.HasSuffix(normalized, "token") || strings.HasSuffix(normalized, "text") +} + +type replacement struct { + start, end int + value string +} + +// RedactJSON preserves the original lexical representation, including duplicate +// keys, numeric spelling and whitespace, except explicitly sensitive values. +// A JSON re-marshal would hide precisely the malformed/duplicate input under investigation. +func RedactJSON(body []byte) (string, error) { + if len(body) == 0 { + return "", nil + } + if !json.Valid(body) { + return "", errors.New("body is not valid JSON; original omitted") + } + dec := json.NewDecoder(strings.NewReader(string(body))) + dec.UseNumber() + var replacements []replacement + var value func(bool) error + startOffset := func() int { + n := int(dec.InputOffset()) + for n < len(body) && (body[n] == ' ' || body[n] == '\n' || body[n] == '\r' || body[n] == '\t' || body[n] == ':' || body[n] == ',') { + n++ + } + return n + } + value = func(hide bool) error { + start := startOffset() + token, err := dec.Token() + if err != nil { + return err + } + switch token { + case json.Delim('{'): + for dec.More() { + key, err := dec.Token() + if err != nil { + return err + } + if err := value(!hide && sensitive(key.(string))); err != nil { + return err + } + } + if _, err := dec.Token(); err != nil { + return err + } + case json.Delim('['): + for dec.More() { + if err := value(false); err != nil { + return err + } + } + if _, err := dec.Token(); err != nil { + return err + } + default: + if s, ok := token.(string); ok && !hide { + clean := cleanString(s) + if clean != s { + encoded, _ := json.Marshal(clean) + replacements = append(replacements, replacement{start, int(dec.InputOffset()), string(encoded)}) + } + } + } + if hide { + end := int(dec.InputOffset()) + // Hiding an entire object/array also hides every replacement inside it. + kept := replacements[:0] + for _, r := range replacements { + if r.start < start || r.end > end { + kept = append(kept, r) + } + } + replacements = append(kept, replacement{start, end, `"[REDACTED]"`}) + } + return nil + } + if err := value(false); err != nil { + return "", fmt.Errorf("JSON redaction failed: %w", err) + } + if _, err := dec.Token(); err != io.EOF { + return "", errors.New("unexpected trailing JSON data") + } + sort.Slice(replacements, func(i, j int) bool { return replacements[i].start < replacements[j].start }) + var out strings.Builder + cursor := 0 + for _, r := range replacements { + out.Write(body[cursor:r.start]) + out.WriteString(r.value) + cursor = r.end + } + out.Write(body[cursor:]) + return out.String(), nil +} + +var ( + urlInText = regexp.MustCompile(`https?://[^\s<>"']+`) + bearerInText = regexp.MustCompile(`(?i)\bBearer\s+[^\s,;"']+`) + secretInText = regexp.MustCompile(`(?i)\b(api[_-]?key|access[_-]?key[_-]?(secret|id)|password|token|secret|signature)\s*[=:]\s*[^\s,;"']+`) +) + +func cleanString(s string) string { + s = urlInText.ReplaceAllStringFunc(s, func(raw string) string { + u, err := url.Parse(raw) + if err != nil { + return "[REDACTED_URL]" + } + if u.RawQuery != "" || u.User != nil || u.Fragment != "" { + u.User = nil + u.RawQuery = "" + u.Fragment = "" + return u.String() + "[REDACTED_URL_PARTS]" + } + return raw + }) + s = bearerInText.ReplaceAllString(s, "Bearer "+redacted) + return secretInText.ReplaceAllString(s, "$1="+redacted) +} + +func safeValue(key string, v any) any { + if sensitive(key) { + return redacted + } + switch x := v.(type) { + case error: + return cleanString(x.Error()) + case string: + if json.Valid([]byte(x)) && (strings.HasPrefix(strings.TrimSpace(x), "{") || strings.HasPrefix(strings.TrimSpace(x), "[")) { + safe, err := RedactJSON([]byte(x)) + if err != nil { + return "[OMITTED: JSON redaction failed]" + } + return safe + } + return cleanString(x) + case []byte: + return fmt.Sprintf("[BINARY: %d bytes]", len(x)) + default: + // Existing structured logs only need scalar identity/count/status fields. + // Sanitize containers instead of allowing Logrus to serialize unchecked structs. + body, err := json.Marshal(v) + if err != nil { + return "[OMITTED: unsupported log value]" + } + if len(body) > 0 && (body[0] == '{' || body[0] == '[') { + safe, err := RedactJSON(body) + if err != nil { + return "[OMITTED: unsafe log value]" + } + return json.RawMessage(safe) + } + return v + } +} diff --git a/internal/logging/rpc.go b/internal/logging/rpc.go new file mode 100644 index 0000000..352d308 --- /dev/null +++ b/internal/logging/rpc.go @@ -0,0 +1,141 @@ +package logging + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "strings" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoreflect" +) + +func UnaryClientInterceptor(local, remote string) grpc.UnaryClientInterceptor { + return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoke grpc.UnaryInvoker, opts ...grpc.CallOption) error { + attrs := rpcIdentity(method, req) + started := time.Now() + slog.InfoContext(ctx, "RPC request", append(attrs, "direction", local+"->"+remote, "stage", "invoke")...) + rpcBody(ctx, "RPC request data", req, append(attrs, "direction", local+"->"+remote, "stage", "invoke")...) + err := invoke(ctx, method, req, reply, cc, opts...) + result := append(attrs, "direction", remote+"->"+local, "duration_ms", time.Since(started).Milliseconds(), "rpc_status", status.Code(err).String()) + if err != nil { + slog.ErrorContext(ctx, "RPC failed", append(result, "stage", "response_failed", "error", err)...) + return err + } + slog.InfoContext(ctx, "RPC response", append(result, "stage", "response_received")...) + rpcBody(ctx, "RPC response data", reply, append(result, "stage", "response_received")...) + return nil + } +} + +func UnaryServerInterceptor(local, remote string) grpc.UnaryServerInterceptor { + return func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handle grpc.UnaryHandler) (any, error) { + attrs := rpcIdentity(info.FullMethod, req) + started := time.Now() + slog.InfoContext(ctx, "RPC request", append(attrs, "direction", remote+"->"+local, "stage", "request_received")...) + rpcBody(ctx, "RPC request data", req, append(attrs, "direction", remote+"->"+local, "stage", "request_received")...) + reply, err := handle(ctx, req) + result := append(attrs, "direction", local+"->"+remote, "duration_ms", time.Since(started).Milliseconds(), "rpc_status", status.Code(err).String()) + if err != nil { + slog.ErrorContext(ctx, "RPC failed", append(result, "stage", "response_failed", "error", err)...) + return reply, err + } + // Returning from a handler is not evidence of delivery to the caller. + slog.InfoContext(ctx, "RPC response", append(result, "stage", "response_ready")...) + rpcBody(ctx, "RPC response data", reply, append(result, "stage", "response_ready")...) + return reply, nil + } +} + +func rpcIdentity(method string, v any) []any { + attrs := []any{"transport", "grpc", "method", method} + p, ok := v.(proto.Message) + if !ok || p == nil { + return attrs + } + var collect func(protoreflect.Message) + collect = func(m protoreflect.Message) { + m.Range(func(f protoreflect.FieldDescriptor, v protoreflect.Value) bool { + switch string(f.Name()) { + case "request_id", "trace_id", "operation_id", "dispatcher_id", "tenant_id", "agent_id", "cell_id", "boot_id", "source_event_id", "task_id", "call_id", "execution_id": + if !f.IsList() && !f.IsMap() { + attrs = append(attrs, string(f.Name()), v.Interface()) + } + case "meta": + if f.Kind() == protoreflect.MessageKind { + collect(v.Message()) + } + } + return true + }) + } + collect(p.ProtoReflect()) + return attrs +} + +func rpcBody(ctx context.Context, message string, v any, attrs ...any) { + if !slog.Default().Enabled(ctx, slog.LevelDebug) { + return + } + p, ok := v.(proto.Message) + if !ok || p == nil { + return + } + body, err := rpcJSON(p) + if err != nil { + slog.DebugContext(ctx, message, append(attrs, "redaction_error", err.Error(), "body", "[OMITTED: RPC projection failed]")...) + return + } + Body(ctx, message, body, append(attrs, "wire_encoding", "protobuf", "log_encoding", "protobuf_json_with_embedded_json")...) +} + +// JSON payload bytes are observed as JSON instead of opaque base64. All other +// byte fields are represented only by length. No request object is modified. +func rpcJSON(p proto.Message) ([]byte, error) { + encoded, err := (protojson.MarshalOptions{UseProtoNames: true}).Marshal(p) + if err != nil { + return nil, err + } + var fields map[string]json.RawMessage + if err = json.Unmarshal(encoded, &fields); err != nil { + return nil, err + } + var projectionErr error + p.ProtoReflect().Range(func(f protoreflect.FieldDescriptor, v protoreflect.Value) bool { + if f.IsList() || f.IsMap() { + return true + } + name := string(f.Name()) + if f.Kind() == protoreflect.BytesKind { + raw := v.Bytes() + if strings.HasSuffix(name, "_json") { + safe, e := RedactJSON(raw) + if e != nil { + fields[name], _ = json.Marshal("[OMITTED: invalid embedded JSON]") + fields[name+"_redaction_error"], _ = json.Marshal(e.Error()) + } else if len(raw) > 0 { + fields[name] = json.RawMessage(safe) + } + } else { + fields[name], _ = json.Marshal(fmt.Sprintf("[BINARY: %d bytes]", len(raw))) + } + } else if f.Kind() == protoreflect.MessageKind { + nested, e := rpcJSON(v.Message().Interface()) + if e != nil { + projectionErr = e + return false + } + fields[name] = nested + } + return true + }) + if projectionErr != nil { + return nil, projectionErr + } + return json.Marshal(fields) +} diff --git a/internal/logging/rpc_test.go b/internal/logging/rpc_test.go new file mode 100644 index 0000000..376626f --- /dev/null +++ b/internal/logging/rpc_test.go @@ -0,0 +1,65 @@ +package logging + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "strings" + "testing" + + agentpb "git.ipao.vip/rogee/go-sip/gen/agent" + "google.golang.org/grpc" +) + +func TestRPCLogsObserveActualMessageWithoutMutatingIt(t *testing.T) { + out := installTestLogger(t, true) + raw := []byte(`{"task_id":"4","params":{"speed":1.0},"api_key":"private-key","prompt":"private-prompt"}`) + req := &agentpb.ExecuteApprovedRequest{Meta: &agentpb.RequestMeta{RequestId: "request-1"}, CallId: "execution-1", TaskConfigJson: raw, ProvidersJson: []byte(`[{"api_key":"private-session"}]`)} + res := &agentpb.ExecuteApprovedResponse{} + interceptor := UnaryClientInterceptor("dispatcher", "agent") + err := interceptor(context.Background(), "/sip.agent.v1.AgentControlService/ExecuteApproved", req, res, nil, func(_ context.Context, _ string, sent, received any, _ *grpc.ClientConn, _ ...grpc.CallOption) error { + if sent != req || received != res || !bytes.Equal(req.TaskConfigJson, raw) { + t.Fatal("logger mutated RPC") + } + return nil + }) + if err != nil { + t.Fatal(err) + } + text := out.String() + for _, secret := range []string{"private-key", "private-prompt", "private-session"} { + if strings.Contains(text, secret) { + t.Fatalf("leaked %s", secret) + } + } + for _, expected := range []string{"request-1", "execution-1", "dispatcher->agent", "agent->dispatcher", "response_received", `\"speed\":1.0`} { + if !strings.Contains(text, expected) { + t.Fatalf("missing %s: %s", expected, text) + } + } +} + +func TestRPCNormalModeAndErrorsNeverIncludeBody(t *testing.T) { + out := installTestLogger(t, false) + req := &agentpb.ReportCallEndedRequest{SourceEventId: "execution-2"} + expected := errors.New("connection failed") + _, err := UnaryServerInterceptor("agent", "dispatcher")(context.Background(), req, &grpc.UnaryServerInfo{FullMethod: "/ReportCallEnded"}, func(context.Context, any) (any, error) { return nil, expected }) + if !errors.Is(err, expected) { + t.Fatalf("error changed: %v", err) + } + text := out.String() + if strings.Contains(text, "private-token") || strings.Contains(text, `"body"`) { + t.Fatal(text) + } + for _, part := range []string{"execution-2", "response_failed", "duration_ms", "connection failed"} { + if !strings.Contains(text, part) { + t.Fatalf("missing %s", part) + } + } + for _, line := range strings.Split(strings.TrimSpace(text), "\n") { + if !json.Valid([]byte(line)) { + t.Fatal(line) + } + } +} diff --git a/internal/mq/broker.go b/internal/mq/broker.go index cf58c82..53bf7c3 100644 --- a/internal/mq/broker.go +++ b/internal/mq/broker.go @@ -5,11 +5,13 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "strings" "sync" "time" "git.ipao.vip/rogee/go-sip/internal/contract" + "git.ipao.vip/rogee/go-sip/internal/logging" "git.ipao.vip/rogee/go-sip/internal/tenant" amqp "github.com/rabbitmq/amqp091-go" ) @@ -96,7 +98,19 @@ func (b *Broker) Close() error { func (b *Broker) Done() <-chan *amqp.Error { return b.closed } -func (b *Broker) Publish(ctx context.Context, exchange, routingKey string, body []byte) error { +func (b *Broker) Publish(ctx context.Context, exchange, routingKey string, body []byte) (publishErr error) { + started := time.Now() + attrs := append(messageIdentity(body), "transport", "mq", "exchange", exchange, "routing_key", routingKey) + slog.InfoContext(ctx, "MQ publish requested", append(attrs, "direction", "dispatcher->mq", "stage", "publish_requested", "body_bytes", len(body))...) + logging.Body(ctx, "MQ publish data", body, append(attrs, "direction", "dispatcher->mq", "stage", "publish_requested")...) + defer func() { + result := append(append([]any{}, attrs...), "duration_ms", time.Since(started).Milliseconds()) + if publishErr != nil { + slog.ErrorContext(ctx, "MQ publish failed", append(result, "direction", "dispatcher->mq", "stage", "publish_failed", "error", publishErr)...) + } else { + slog.InfoContext(ctx, "MQ publish confirmed", append(result, "direction", "mq->dispatcher", "stage", "publisher_confirmed", "saas_processed", false)...) + } + }() if exchange != ResultsExchange || routingKey != b.resultRoute || len(body) == 0 || len(body) > MaxMessageBytes { return fmt.Errorf("publish requires result exchange, Dispatcher route and 1..%d body bytes", MaxMessageBytes) } @@ -125,6 +139,7 @@ func (b *Broker) Publish(ctx context.Context, exchange, routingKey string, body if len(identity.EventID) > 255 { return errors.New("outbound event identity exceeds AMQP limit") } + slog.InfoContext(ctx, "MQ outbound validated", append(attrs, "direction", "dispatcher->mq", "stage", "validated")...) messageID := identity.EventID ctx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() @@ -145,6 +160,7 @@ func (b *Broker) Publish(ctx context.Context, exchange, routingKey string, body return fmt.Errorf("enable current publisher confirms: %w", err) } returned := channel.NotifyReturn(make(chan amqp.Return, 1)) + slog.InfoContext(ctx, "MQ publish", append(attrs, "direction", "dispatcher->mq", "stage", "publish_start")...) confirmation, err := channel.PublishWithDeferredConfirmWithContext(ctx, exchange, routingKey, true, false, amqp.Publishing{ ContentType: "application/json", DeliveryMode: amqp.Persistent, Body: body, MessageId: messageID, }) @@ -348,6 +364,7 @@ func (b *Broker) DrainControlPredeclared(ctx context.Context, queue string, hand if !ok { return processed, nil } + observeDelivery(ctx, queue, &delivery) if err := b.validateInbound(queue, delivery.RoutingKey, delivery.Body); err != nil { if err := b.rejectInbound(ctx, queue, delivery, err); err != nil { return processed, err @@ -355,7 +372,9 @@ func (b *Broker) DrainControlPredeclared(ctx context.Context, queue string, hand processed++ continue } + slog.InfoContext(ctx, "MQ inbound validated", append(messageIdentity(delivery.Body), "direction", "mq->dispatcher", "queue", queue, "stage", "validated")...) if err := handler(ctx, delivery.RoutingKey, delivery.Body); err != nil { + slog.ErrorContext(ctx, "MQ handler failed", append(messageIdentity(delivery.Body), "stage", "handler_failed", "error", err, "permanent", IsPermanent(err))...) if IsPermanent(err) { if rejectErr := delivery.Reject(false); rejectErr != nil { return processed, fmt.Errorf("control handler error %v; reject: %w", err, rejectErr) @@ -449,13 +468,16 @@ func (b *Broker) consumeDeliveries(ctx context.Context, dispatcherID, queue stri if !ok { return errors.New("rabbitmq current delivery channel closed") } + observeDelivery(ctx, queue, &delivery) if err := b.validateInbound(queue, delivery.RoutingKey, delivery.Body); err != nil { if err := b.rejectInbound(ctx, queue, delivery, err); err != nil { return err } continue } + slog.InfoContext(ctx, "MQ inbound validated", append(messageIdentity(delivery.Body), "direction", "mq->dispatcher", "queue", queue, "stage", "validated")...) if err := handler(ctx, delivery.RoutingKey, delivery.Body); err != nil { + slog.ErrorContext(ctx, "MQ handler failed", append(messageIdentity(delivery.Body), "stage", "handler_failed", "error", err, "permanent", IsPermanent(err))...) if IsPermanent(err) { if rejectErr := delivery.Reject(false); rejectErr != nil { return fmt.Errorf("permanent current handler error %v; reject: %w", err, rejectErr) diff --git a/internal/mq/logging.go b/internal/mq/logging.go new file mode 100644 index 0000000..e035f8d --- /dev/null +++ b/internal/mq/logging.go @@ -0,0 +1,74 @@ +package mq + +import ( + "context" + "encoding/json" + "log/slog" + "time" + + "git.ipao.vip/rogee/go-sip/internal/logging" + amqp "github.com/rabbitmq/amqp091-go" +) + +func messageIdentity(body []byte) []any { + var m struct { + EventID string `json:"event_id"` + EventType string `json:"event_type"` + DispatcherID string `json:"dispatcher_id"` + Payload struct { + TaskID string `json:"task_id"` + } `json:"payload"` + } + if err := json.Unmarshal(body, &m); err != nil { + return nil + } + return []any{"event_id", m.EventID, "event_type", m.EventType, "dispatcher_id", m.DispatcherID, "task_id", m.Payload.TaskID} +} + +func observeDelivery(ctx context.Context, queue string, d *amqp.Delivery) { + attrs := append(messageIdentity(d.Body), "transport", "mq", "direction", "mq->dispatcher", "queue", queue, "routing_key", d.RoutingKey, "message_id", d.MessageId, "delivery_tag", d.DeliveryTag, "redelivered", d.Redelivered) + slog.InfoContext(ctx, "MQ delivery", append(attrs, "stage", "received", "body_bytes", len(d.Body))...) + logging.Body(ctx, "MQ receive data", d.Body, append(attrs, "stage", "received")...) + if d.Acknowledger != nil { + d.Acknowledger = &observedAcknowledger{next: d.Acknowledger, ctx: ctx, attrs: attrs, started: time.Now()} + } +} + +type observedAcknowledger struct { + next amqp.Acknowledger + ctx context.Context + attrs []any + started time.Time +} + +func (a *observedAcknowledger) report(stage string, err error) { + attrs := append(append([]any{}, a.attrs...), "direction", "dispatcher->mq", "stage", stage, "duration_ms", time.Since(a.started).Milliseconds()) + if err != nil { + slog.ErrorContext(a.ctx, "MQ disposition failed", append(attrs, "error", err)...) + } else { + slog.InfoContext(a.ctx, "MQ disposition", attrs...) + } +} +func (a *observedAcknowledger) Ack(tag uint64, multiple bool) error { + err := a.next.Ack(tag, multiple) + a.report("ack_sent", err) + return err +} +func (a *observedAcknowledger) Nack(tag uint64, multiple, requeue bool) error { + err := a.next.Nack(tag, multiple, requeue) + stage := "nack_sent" + if requeue { + stage = "requeue_requested" + } + a.report(stage, err) + return err +} +func (a *observedAcknowledger) Reject(tag uint64, requeue bool) error { + err := a.next.Reject(tag, requeue) + stage := "reject_sent" + if requeue { + stage = "requeue_requested" + } + a.report(stage, err) + return err +} diff --git a/internal/mq/logging_test.go b/internal/mq/logging_test.go new file mode 100644 index 0000000..3012abd --- /dev/null +++ b/internal/mq/logging_test.go @@ -0,0 +1,76 @@ +package mq + +import ( + "bytes" + "context" + "errors" + "log/slog" + "strings" + "testing" + + "git.ipao.vip/rogee/go-sip/internal/logging" + amqp "github.com/rabbitmq/amqp091-go" +) + +type logAck struct { + acks, rejects, nacks int + err error +} + +func (a *logAck) Ack(uint64, bool) error { a.acks++; return a.err } +func (a *logAck) Nack(uint64, bool, bool) error { a.nacks++; return a.err } +func (a *logAck) Reject(uint64, bool) error { a.rejects++; return a.err } + +func TestMQReceiveAndDispositionLogsPreserveMessageAndErrors(t *testing.T) { + for _, debug := range []bool{false, true} { + t.Run(map[bool]string{false: "normal", true: "debug"}[debug], func(t *testing.T) { + prior := slog.Default() + defer slog.SetDefault(prior) + var out bytes.Buffer + logging.Configure("dispatcher", debug, &out) + body := []byte(`{ "event_id":"event-1", "event_type":"call.execute", "payload":{"task_id":"4","callee":"12345"},"api_key":"private"}`) + original := bytes.Clone(body) + ack := &logAck{} + delivery := amqp.Delivery{Body: body, MessageId: "event-1", DeliveryTag: 1, Acknowledger: ack} + observeDelivery(context.Background(), "task-4", &delivery) + if !bytes.Equal(delivery.Body, original) { + t.Fatal("message changed") + } + if err := delivery.Ack(false); err != nil || ack.acks != 1 { + t.Fatal(err, ack) + } + expected := errors.New("ack transport failed") + ack.err = expected + if err := delivery.Reject(false); !errors.Is(err, expected) || ack.rejects != 1 { + t.Fatal("disposition error lost", err) + } + text := out.String() + for _, part := range []string{"received", "ack_sent", "reject_sent", "ack transport failed", "event-1", "task-4"} { + if !strings.Contains(text, part) { + t.Fatalf("missing %s: %s", part, text) + } + } + if strings.Contains(text, "private") { + t.Fatal(text) + } + if strings.Contains(text, `\"callee\":\"12345\"`) != debug { + t.Fatal("mode failed", text) + } + }) + } +} + +func TestMQRejectedPublishDoesNotClaimConfirmation(t *testing.T) { + previous := slog.Default() + defer slog.SetDefault(previous) + var out bytes.Buffer + logging.Configure("dispatcher", true, &out) + b := &Broker{} + if err := b.Publish(context.Background(), "invalid", "invalid", []byte(`{"api_key":"private"}`)); err == nil { + t.Fatal("invalid publish accepted") + } + text := out.String() + if strings.Contains(text, "private") || strings.Contains(text, "publisher_confirmed") || !strings.Contains(text, "publish_failed") { + t.Fatal(text) + } +} diff --git a/internal/rpc/approved_debug_test.go b/internal/rpc/approved_debug_test.go index 9847284..6e13be4 100644 --- a/internal/rpc/approved_debug_test.go +++ b/internal/rpc/approved_debug_test.go @@ -7,12 +7,12 @@ import ( "time" ) -func TestRealCallDebugControlsExternalEvidenceOnly(t *testing.T) { +func TestRealCallCaptureControlsExternalEvidenceOnly(t *testing.T) { for _, debug := range []bool{false, true} { t.Run(map[bool]string{false: "normal", true: "debug"}[debug], func(t *testing.T) { fixture, _, _, _, approved := recordedMockFixture(t, nil, time.Second) approved.CallerID, approved.DialedCallee, approved.RingTimeout = "BD93205882", "7089"+approved.Callee, time.Second - r := &ApprovedRecordedRealCall{Debug: debug, Mirror: asterisk.NewHEPMirror(), MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, Delivery: fixture.Delivery, originator: func(ctx context.Context, _ asterisk.NativeDial) (realMedia, error) { + r := &ApprovedRecordedRealCall{Capture: debug, Mirror: asterisk.NewHEPMirror(), MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, Delivery: fixture.Delivery, originator: func(ctx context.Context, _ asterisk.NativeDial) (realMedia, error) { t.Fatal("prepare must not dial") return realMedia{}, nil }} diff --git a/internal/rpc/approved_recorded_real.go b/internal/rpc/approved_recorded_real.go index 8b12869..8754d0f 100644 --- a/internal/rpc/approved_recorded_real.go +++ b/internal/rpc/approved_recorded_real.go @@ -24,7 +24,7 @@ type ApprovedRecordedRealCall struct { Loader asterisk.Loader MediaPayloadType uint8 EvidenceRoot string - Debug bool + Capture bool MaxWAVBytes int64 ReportTimeout time.Duration Delivery *agent.RecordingDelivery @@ -63,7 +63,7 @@ func (r *ApprovedRecordedRealCall) Prepare(approved ApprovedExecution) (func(con return nil, errors.New("native Asterisk configuration required for real call") } } - if r.Debug { + if r.Capture { if err := verifyCallEvidence(r.EvidenceRoot, approved.SourceEventID, approved.SelectedTrunkID, approved.Callee); err != nil { return nil, err } @@ -106,7 +106,7 @@ func (r *ApprovedRecordedRealCall) Prepare(approved ApprovedExecution) (func(con if ctx == nil || ctx.Err() != nil || !time.Now().Before(approved.DialBefore) { return errors.New("real call instruction expired or cancelled before origination") } - if r.Debug { + if r.Capture { if err := verifyCallEvidence(r.EvidenceRoot, approved.SourceEventID, approved.SelectedTrunkID, approved.Callee); err != nil { return err } diff --git a/internal/rpc/approved_worker.go b/internal/rpc/approved_worker.go index 765eae9..b5c8ba7 100644 --- a/internal/rpc/approved_worker.go +++ b/internal/rpc/approved_worker.go @@ -3,7 +3,7 @@ package rpc import ( "context" "errors" - "log" + "log/slog" "time" "git.ipao.vip/rogee/go-sip/internal/agent" @@ -82,17 +82,21 @@ func (w *ApprovedCallWorker) Originate(request context.Context, approved Approve issued <- ErrApprovedDialExpired return } + logger := slog.With("event_id", approved.SourceEventID, "task_id", approved.TaskID, "call_id", approved.CallID) + logger.Info("Agent call scheduled", "stage", "scheduled") issued <- nil // the Unary response may now acknowledge scheduled work + logger.Info("Agent call execution", "stage", "execution_started") runErr := prepared(callContext) // Physical execution resources are released independently of any later // failure reporting, OSS work or MQ publisher confirmation. release() if runErr == nil { + logger.Info("Agent call execution complete", "stage", "execution_completed") return } - log.Printf("Agent approved call ended with error: event_id=%q task_id=%q cause_type=%T", approved.SourceEventID, approved.TaskID, runErr) + logger.Error("Agent call execution failed", "stage", "execution_failed", "error", runErr) if err := w.OnFailure(approved, runErr); err != nil { - log.Printf("Agent approved call failure reporting failed: event_id=%q task_id=%q cause_type=%T", approved.SourceEventID, approved.TaskID, err) + logger.Error("Agent call failure reporting failed", "stage", "failure_report_failed", "error", err) } }() return <-issued