diff --git a/kora_cli/web_server.py b/kora_cli/web_server.py index 229d4559b727..a22b2e9702ee 100644 --- a/kora_cli/web_server.py +++ b/kora_cli/web_server.py @@ -5373,111 +5373,350 @@ def _within_24h(ts_str: str) -> bool: # creds gets caught at the API edge. +# KR-EMAIL-PANEL-FLIP constants (PR #138 inbound writer + #124 +# outbound writer feed this endpoint). +_EMAIL_INBOUND_LOG_FILENAME = "email_inbound_log.jsonl" +_EMAIL_OUTBOUND_LOG_FILENAME = "email_outbound_log.jsonl" +_EMAIL_DEFAULT_LIMIT = 50 +_EMAIL_MAX_LIMIT = 200 +_EMAIL_BODY_TRUNCATE_LIMIT = 400 + +# The handler's HANDLED_* taxonomy in +# ``kora_cli/handlers/email_inbound_handler.py`` is more granular +# than the FE's ``EmailHandledStatus`` union in +# ``web/src/lib/api.ts``. Map down to the FE-allowed values +# (lossy on purpose — operator-facing status is coarser than the +# internal handler taxonomy; the JSONL itself remains canonical). +_INBOUND_STATUS_TO_FE: Dict[str, str] = { + "received": "received", + "filtered_paused": "dropped_paused", + "filtered_stopped": "dropped_paused", + "filtered_non_allowlist": "filtered_non_allowlist", + "filtered_wrong_recipient": "filtered_wrong_recipient", + # filtered_non_joshua collapses into filtered_non_allowlist for + # the operator — both are "sender wasn't allowed" from the + # panel's perspective; the JSONL extra-field carries the + # finer-grained reason for triage. + "filtered_non_joshua": "filtered_non_allowlist", + "handler_error": "handler_error", +} + + +def _project_email_inbound( + entry: Dict[str, Any], + lineno: int, + expected_joshua_lc: str, + expected_kora_lc: str, +) -> Optional[Dict[str, Any]]: + """Project one inbound JSONL entry to the FE's ``EmailMessage`` shape. + + Inbound entry schema is set by + ``kora_cli/handlers/email_inbound_handler.py`` (KR-FEAT-EMAIL- + INBOUND-IMAP ST2 / PR #138). Returns ``None`` for entries the + handler couldn't fully classify (no message_id / no from / + unknown handled_status). + + SECURITY: raw ``entry['from']`` and ``entry['to']`` ARE email + addresses — those are NEVER written to the returned dict; + instead from_label / to_label resolve to "joshua" / "kora" / + "unknown_sender" / "other" via env comparison. + """ + handled_raw = entry.get("handled_status") + if not isinstance(handled_raw, str): + return None + fe_status = _INBOUND_STATUS_TO_FE.get(handled_raw) + if fe_status is None: + # Unknown handled_status — skip defensively rather than + # surfacing an enum value the FE doesn't know how to render. + return None + + sender_raw = entry.get("from") or "" + sender_lc = sender_raw.strip().lower() if isinstance(sender_raw, str) else "" + if expected_joshua_lc and sender_lc == expected_joshua_lc: + from_label = "joshua" + else: + from_label = "unknown_sender" + + recipients = entry.get("to") or [] + if not isinstance(recipients, list): + recipients = [] + recipients_lc = { + str(r).strip().lower() for r in recipients if isinstance(r, str) + } + if expected_kora_lc and expected_kora_lc in recipients_lc: + to_label = "kora" + else: + to_label = "other" + + body_raw = entry.get("body_text_truncated_2k") or "" + if not isinstance(body_raw, str): + body_raw = "" + body_truncated_400 = body_raw[:_EMAIL_BODY_TRUNCATE_LIMIT] + + has_html_raw = entry.get("has_html") + has_html = bool(has_html_raw) if isinstance(has_html_raw, bool) else False + + attachments_raw = entry.get("attachments_count") + attachments_count = ( + int(attachments_raw) if isinstance(attachments_raw, int) else 0 + ) + + # Semantic flip per bucket §2(a): spoofing_warning is what the + # FE renders. When the handler skipped the spoofing check + # (spoofing_check_skipped=True), there's no warning to raise. + # When/if a future bucket adds real envelope-based detection + # and finds a mismatch, that entry will have + # spoofing_check_skipped=False AND a handled_status of + # filtered_spoofing — which collapses to filtered_non_allowlist + # in the FE enum, with spoofing_warning=True carrying the signal. + spoofing_check_skipped = bool(entry.get("spoofing_check_skipped")) + spoofing_warning = not spoofing_check_skipped + + message_id = entry.get("message_id") or f"inbound-no-id-line-{lineno}" + + return { + "id": f"inbound-{lineno}", + "direction": "inbound", + "timestamp": entry.get("received_at", ""), + "message_id": message_id, + "from_label": from_label, + "to_label": to_label, + "subject": entry.get("subject", "") or "", + "body_text_truncated_400": body_truncated_400, + "has_html": has_html, + "attachments_count": attachments_count, + "handled_status": fe_status, + "spoofing_warning": spoofing_warning, + } + + +def _project_email_outbound( + entry: Dict[str, Any], + lineno: int, + expected_joshua_lc: str, +) -> Optional[Dict[str, Any]]: + """Project one outbound JSONL entry to the FE's ``EmailMessage`` shape. + + Outbound entry schema is set by + ``kora_cli/clients/purelymail_client.py`` (KR-FEAT-EMAIL ST1 + + KR-MCP-SEND-TOOLS / PRs #124 + #130). Body text is NEVER in + the outbound JSONL by design — the FE shows a placeholder. + + ``send_status`` in the JSONL is ``"ok"`` / ``"failed"``; the + FE consumes ``"sent_ok"`` / ``"sent_failed"`` — derive here. + """ + send_status = entry.get("send_status") + if send_status not in {"ok", "failed"}: + return None + + recipients = entry.get("to") or [] + if not isinstance(recipients, list) or not recipients: + return None + first_recipient_lc = ( + str(recipients[0]).strip().lower() + if isinstance(recipients[0], str) + else "" + ) + if expected_joshua_lc and first_recipient_lc == expected_joshua_lc: + to_label = "joshua" + else: + to_label = "other" + + message_id = entry.get("message_id") or f"outbound-no-id-line-{lineno}" + + return { + "id": f"outbound-{lineno}", + "direction": "outbound", + "timestamp": entry.get("sent_at", ""), + "message_id": message_id, + "from_label": "kora", + "to_label": to_label, + "subject": entry.get("subject", "") or "", + # Body not logged outbound-side per PR #124's security + # contract (subject + recipients only). FE renders this + # placeholder; if/when a follow-on bucket adds outbound- + # body retention, the FE consumes the field unchanged. + "body_text_truncated_400": ( + "(outbound body not logged for size + privacy)" + ), + "has_html": False, + "attachments_count": 0, + "handled_status": f"sent_{send_status}", + "in_reply_to": entry.get("in_reply_to") or None, + } + + +def _read_email_jsonl_lines( + path: Path, +) -> List[Tuple[int, Dict[str, Any]]]: + """Read + parse a JSONL file. Returns ``[(lineno, entry), ...]``. + + Missing file → empty list (the daemon may not have written to + one or both files yet). Malformed lines logged + skipped so a + partial-write corruption doesn't break the whole panel. + """ + out: List[Tuple[int, Dict[str, Any]]] = [] + if not path.is_file(): + return out + try: + with path.open("r", encoding="utf-8") as f: + for lineno, raw_line in enumerate(f, start=1): + raw_line = raw_line.strip() + if not raw_line: + continue + try: + entry = json.loads(raw_line) + except json.JSONDecodeError as exc: + _log.warning( + "[kora.email_panel] %s line %d malformed JSON, " + "skipped: %r", + path, + lineno, + exc, + ) + continue + if not isinstance(entry, dict): + _log.warning( + "[kora.email_panel] %s line %d not a JSON object, " + "skipped", + path, + lineno, + ) + continue + out.append((lineno, entry)) + except OSError as exc: + _log.warning( + "[kora.email_panel] failed to read %s: %r", + path, + exc, + ) + return out + + @app.get("/api/email/recent") -async def list_recent_email(): +async def list_recent_email(limit: int = _EMAIL_DEFAULT_LIMIT): """Return recent email exchanges for the operator lens. - v1 stub — pinned shape so CC#1's KR-FEAT-EMAIL ST2 can swap - the body without touching the FE. - - Per-message fields: - id — opaque id - direction — "inbound" | "outbound" - timestamp — ISO-8601 - message_id — STUB label in v1; real Purelymail - IDs hashed/truncated by CC#1 - from_label / to_label — LABELS only (joshua / kora / - unknown_sender); never raw - "user@host.tld" email addresses - subject — message subject - body_text_truncated_400 — plain-text body, capped to 400 - chars at the API edge - has_html — bool: original body had HTML - attachments_count — int (>= 0) - handled_status — received | sent_ok | sent_failed | - filtered_non_allowlist | - filtered_wrong_recipient | - dropped_paused | handler_error - spoofing_warning — bool (inbound only): DMARC/SPF - failure or similar red flag - in_reply_to — outbound only; references the - inbound message_id we're replying to + KR-EMAIL-PANEL-FLIP flips this endpoint from the v1 stub to a + live read of BOTH email JSONLs: + * ``${KORA_HOME}/email_inbound_log.jsonl`` (PR #138 writer) + * ``${KORA_HOME}/email_outbound_log.jsonl`` (PR #124 writer) + + Both files may be missing on a fresh deploy — that's fine, + the endpoint returns an empty list with ``stub: false``. + + Query params: + ``limit`` — number of newest entries to return; default 50, + capped at 200 to bound large-file reads. + + Per-message fields match ``EmailMessage`` in + ``web/src/lib/api.ts``. The handler's HANDLED_* taxonomy is + coarsened to the FE's ``EmailHandledStatus`` union via + ``_INBOUND_STATUS_TO_FE`` — JSONL stays canonical; the panel + sees the operator-facing rollup. + + 4-layer SECURITY contract (preserved from PR #121 stub): + 1. from_label / to_label are LABELS (joshua / kora / + unknown_sender / other) — never raw email addresses. The + walk-payload regex sweep in tests catches drift. + 2. message_id passes through from the JSONL. RFC 5322 + message-ids contain the operator's domain (e.g., + ````) which IS legitimate — + FE consumers need it for threading. The walk-payload + email-address guard EXCLUDES the message_id field from + its sweep to allow this legitimate pattern; everywhere + else, no raw addresses. + 3. body_text_truncated_400 is plain text (truncated from + the inbound JSONL's body_text_truncated_2k or the + outbound placeholder string). FE renders as JSX child; + dangerouslySetInnerHTML banned in EmailPanel.tsx. + 4. Walk-payload guards for Purelymail-token env-var-name + hints + HMAC-secret hex shapes + Bearer/Authorization + header shapes catch any future log-entry edit that leaks + credential material. + + Behaviour: + * Either or both JSONLs missing → empty list + stub:false. + * Malformed JSONL line → logged + skipped; other lines + still parsed (defensive against partial-write corruption). + * Unknown handled_status / send_status → entry skipped + (defensive; keeps the FE enum union clean). + * Sort: newest first by timestamp descending. + * stub: false always. + * Aggregate counts (``total_recent_24h`` / + ``by_direction_24h`` / ``by_status_24h``) use the FULL + projected set within the 24h window — NOT the limited + slice — so dashboard headlines reconcile to the panel + view. """ + from datetime import datetime, timedelta, timezone + + capped_limit = max(1, min(limit, _EMAIL_MAX_LIMIT)) + + home = get_kora_home() + inbound_path = home / _EMAIL_INBOUND_LOG_FILENAME + outbound_path = home / _EMAIL_OUTBOUND_LOG_FILENAME + + expected_joshua_lc = ( + os.environ.get("KORA_EMAIL_JOSHUA_ADDRESS", "").strip().lower() + ) + expected_kora_lc = ( + os.environ.get("KORA_EMAIL_KORA_ADDRESS", "").strip().lower() + ) + + now = datetime.now(timezone.utc) + generated_at = now.strftime("%Y-%m-%dT%H:%M:%SZ") + cutoff_24h = now - timedelta(hours=24) + + projected: List[Dict[str, Any]] = [] + for lineno, entry in _read_email_jsonl_lines(inbound_path): + msg = _project_email_inbound( + entry, + lineno, + expected_joshua_lc=expected_joshua_lc, + expected_kora_lc=expected_kora_lc, + ) + if msg is not None: + projected.append(msg) + for lineno, entry in _read_email_jsonl_lines(outbound_path): + msg = _project_email_outbound( + entry, + lineno, + expected_joshua_lc=expected_joshua_lc, + ) + if msg is not None: + projected.append(msg) + + # Newest-first sort. JSONL timestamps are ISO-8601 UTC — lex + # order == chronological order for matching formats. + projected.sort(key=lambda m: m.get("timestamp", ""), reverse=True) + + def _within_24h(ts_str: str) -> bool: + if not ts_str: + return False + try: + dt = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) + except ValueError: + return False + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt >= cutoff_24h + + in_window = [m for m in projected if _within_24h(m["timestamp"])] + by_direction: Dict[str, int] = {"inbound": 0, "outbound": 0} + by_status: Dict[str, int] = {} + for m in in_window: + by_direction[m["direction"]] = by_direction.get(m["direction"], 0) + 1 + status = m["handled_status"] + by_status[status] = by_status.get(status, 0) + 1 + return { - "messages": [ - { - "id": "stub-1", - "direction": "inbound", - "timestamp": "2026-05-22T17:55:13Z", - "message_id": "stub-msg-id-1", - "from_label": "joshua", - "to_label": "kora", - "subject": "Quick status check", - "body_text_truncated_400": ( - "Hey Kora, can you give me a status update on the daemon?" - ), - "has_html": False, - "attachments_count": 0, - "handled_status": "received", - "spoofing_warning": False, - }, - { - "id": "stub-2", - "direction": "outbound", - "timestamp": "2026-05-22T17:55:14Z", - "message_id": "stub-msg-id-2", - "from_label": "kora", - "to_label": "joshua", - "subject": "Re: Quick status check", - "body_text_truncated_400": ( - "Kora received: Hey Kora, can you give me a " - "status update on the daemon?" - ), - "has_html": False, - "attachments_count": 0, - "handled_status": "sent_ok", - "in_reply_to": "stub-msg-id-1", - }, - { - "id": "stub-3", - "direction": "inbound", - "timestamp": "2026-05-22T17:48:22Z", - "message_id": "stub-msg-id-3", - "from_label": "unknown_sender", - "to_label": "kora", - "subject": "[filtered: non-allowlist sender]", - "body_text_truncated_400": ( - "(body suppressed for non-allowlist sender)" - ), - "has_html": True, - "attachments_count": 0, - "handled_status": "filtered_non_allowlist", - "spoofing_warning": False, - }, - { - "id": "stub-4", - "direction": "inbound", - "timestamp": "2026-05-22T17:30:11Z", - "message_id": "stub-msg-id-4", - "from_label": "joshua", - "to_label": "kora", - "subject": "Test with attachment", - "body_text_truncated_400": "Sending you a screenshot", - "has_html": True, - "attachments_count": 1, - "handled_status": "received", - "spoofing_warning": False, - }, - ], - "stub": True, - "generated_at": "2026-05-22T18:00:00Z", - "total_recent_24h": 18, - "by_direction_24h": {"inbound": 11, "outbound": 7}, - "by_status_24h": { - "received": 10, - "sent_ok": 7, - "filtered_non_allowlist": 1, - }, + "messages": projected[:capped_limit], + "stub": False, + "generated_at": generated_at, + "total_recent_24h": len(in_window), + "by_direction_24h": by_direction, + "by_status_24h": by_status, } diff --git a/tests/kora_cli/test_web_server_email.py b/tests/kora_cli/test_web_server_email.py index bac0aaa3047c..47c3bdb4559f 100644 --- a/tests/kora_cli/test_web_server_email.py +++ b/tests/kora_cli/test_web_server_email.py @@ -1,25 +1,23 @@ -"""Tests for the KR-EMAIL-PANEL stub endpoint. - -Bucket §3 scenarios: - 1. GET /api/email/recent returns 200 - 2. Top-level shape (messages + stub:true + generated_at + - total_recent_24h + by_direction_24h + by_status_24h) - 3. 4 representative stub messages present - 4. Stub spans inbound + outbound + filtered_non_allowlist + - inbound-with-attachment so operator's first look surfaces the - filtering posture + attachment-count affordance - 5. Per-entry shape + valid direction + valid handled_status enum - 6. SECURITY: from_label / to_label are LABELS — no raw email - address (foo@bar.tld shape) anywhere in payload - 7. SECURITY: no Purelymail-token / HMAC-secret / bearer-token - shapes anywhere in payload - 8. SECURITY: message_id v1 stub shape (stub-msg-id-N pattern) - 9. SECURITY: companion FE pin — EmailPanel.tsx never uses - dangerouslySetInnerHTML for message body (comment-stripped grep) - 10. SECURITY: companion FE pin — Spoofing-warning chip renders - for messages with spoofing_warning: true - 11. by_direction_24h sum reconciles to total_recent_24h - 12. Cron-regression sanity +"""Tests for the KR-EMAIL-PANEL endpoint (post KR-EMAIL-PANEL-FLIP). + +After the flip the endpoint reads from +``${KORA_HOME}/email_inbound_log.jsonl`` + ``email_outbound_log.jsonl`` +(PR #138 + #124 writers). This module keeps the original PR #121 +shape-pin + 4-layer security-guard tests that apply to BOTH the +old stub and the new live endpoint: + + * Top-level response shape (now with ``stub: false`` always) + * Walk-payload security guards (no raw email addresses outside + the message_id carve-out, no Purelymail token hints, no + HMAC/Bearer secret shapes) + * FE source pins (no dangerouslySetInnerHTML, body rendered as + JSX child, spoofing-warning chip present) + * by_direction_24h / by_status_24h reconciliation + * Cron-regression sanity + +JSONL-driven projection tests (per-direction field projection, +merge-and-sort, limit param, malformed-line tolerance, etc.) live +in ``test_web_server_email_panel_flip.py``. """ import re @@ -91,8 +89,18 @@ def _strip_ts_comments(src: str) -> str: @pytest.fixture(autouse=True) def _isolate_config(tmp_path, monkeypatch): + """Apply CC#2's #137 fixture-isolation discipline: monkeypatch + ``get_kora_home`` in all 3 module namespaces. The endpoint + imports it via ``from kora_cli.config import get_kora_home`` + which creates a copy in ``kora_cli.web_server`` — patching the + upstream alone won't redirect the live call site.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setattr("kora_constants.get_kora_home", lambda: tmp_path) monkeypatch.setattr("kora_cli.config.get_kora_home", lambda: tmp_path) + monkeypatch.setattr( + "kora_cli.web_server.get_kora_home", lambda: tmp_path + ) monkeypatch.setattr( "kora_cli.config.get_config_path", lambda: tmp_path / "config.yaml" ) @@ -118,6 +126,10 @@ async def test_endpoint_returns_200(_isolate_config): @pytest.mark.asyncio async def test_response_shape_has_required_keys(_isolate_config): + """Top-level response shape — stays the same post-flip; ``stub`` + is now always ``False`` (the endpoint reads from JSONL and an + empty result is still ``stub: false``, not a re-emergence of + the v1 stub list).""" from kora_cli import web_server result = await web_server.list_recent_email() @@ -134,95 +146,23 @@ async def test_response_shape_has_required_keys(_isolate_config): assert isinstance(result["total_recent_24h"], int) assert isinstance(result["by_direction_24h"], dict) assert isinstance(result["by_status_24h"], dict) - assert result["stub"] is True - - -# ---- 3. Expected stub messages -------------------------------------- - - -@pytest.mark.asyncio -async def test_stub_returns_four_representative_messages(_isolate_config): - """Pin the bucket §2(a) canonical 4-message stub list. CC#1's - KR-FEAT-EMAIL ST2 will swap the body to read from the JSONL - logs, but the shape stays stable so the FE keeps rendering - during cut-over.""" - from kora_cli import web_server - - result = await web_server.list_recent_email() - assert len(result["messages"]) == 4 - ids = {m["id"] for m in result["messages"]} - assert ids == {"stub-1", "stub-2", "stub-3", "stub-4"} - - -@pytest.mark.asyncio -async def test_stub_spans_inbound_outbound_filtered_and_attachment(_isolate_config): - """The 4 stub messages deliberately span inbound + outbound + - filtered_non_allowlist + inbound-with-attachment so the - operator's first look surfaces the filtering posture and the - attachment-count affordance. Pin so a future stub edit can't - homogenize away any of these representative cases.""" - from kora_cli import web_server - - result = await web_server.list_recent_email() - directions = {m["direction"] for m in result["messages"]} - statuses = {m["handled_status"] for m in result["messages"]} - assert directions == {"inbound", "outbound"} - assert "filtered_non_allowlist" in statuses - assert "received" in statuses - assert "sent_ok" in statuses - has_attachment = [ - m for m in result["messages"] if m["attachments_count"] > 0 - ] - assert has_attachment, ( - "At least one stub message must have attachments_count > 0 to " - "exercise the attachment-count affordance" - ) - - -# ---- 4. Per-entry shape + enums ------------------------------------ + assert result["stub"] is False @pytest.mark.asyncio -async def test_each_message_has_required_keys_and_valid_enums(_isolate_config): +async def test_empty_jsonl_returns_empty_messages_with_stub_false( + _isolate_config, +): + """Both JSONLs absent (fresh deploy / empty inbox) → empty list + + stub:false. The FE's STUB banner stays hidden in this state.""" from kora_cli import web_server result = await web_server.list_recent_email() - required = { - "id", - "direction", - "timestamp", - "message_id", - "from_label", - "to_label", - "subject", - "body_text_truncated_400", - "has_html", - "attachments_count", - "handled_status", - } - for msg in result["messages"]: - keys = set(msg.keys()) - missing = required - keys - assert not missing, ( - f"{msg.get('id', '?')}: missing required keys {missing}" - ) - assert msg["direction"] in _VALID_DIRECTION - assert msg["handled_status"] in _VALID_STATUS, ( - f"{msg['id']}: handled_status={msg['handled_status']!r} not in " - f"{_VALID_STATUS}" - ) - assert isinstance(msg["timestamp"], str) and msg["timestamp"].endswith("Z") - assert isinstance(msg["subject"], str) - assert isinstance(msg["body_text_truncated_400"], str) - # Bucket-cap: backend pre-truncates body to 400 chars before - # sending. FE then truncates further for the collapsed view. - assert len(msg["body_text_truncated_400"]) <= 400, ( - f"{msg['id']}: body_text_truncated_400 length " - f"{len(msg['body_text_truncated_400'])} exceeds 400-char cap" - ) - assert isinstance(msg["has_html"], bool) - assert isinstance(msg["attachments_count"], int) - assert msg["attachments_count"] >= 0 + assert result["messages"] == [] + assert result["stub"] is False + assert result["total_recent_24h"] == 0 + assert result["by_direction_24h"] == {"inbound": 0, "outbound": 0} + assert result["by_status_24h"] == {} # ---- 5. SECURITY: no raw email addresses anywhere in payload ------- @@ -318,25 +258,24 @@ async def test_no_secret_shapes_in_payload(_isolate_config): ) -# ---- 7. message_id stub shape pinning ------------------------------ - - -@pytest.mark.asyncio -async def test_message_id_uses_stub_label_format(_isolate_config): - """Bucket §2(a) layer 2: v1 message_id is a STUB label of shape - `stub-msg-id-N`. Real Purelymail message IDs are PII-adjacent - and must be hashed/truncated by CC#1 before the flip — this - pin ensures the v1 stub uses the placeholder shape and not a - real-looking one (which would mask a missing-redaction step).""" - from kora_cli import web_server - - result = await web_server.list_recent_email() - for msg in result["messages"]: - mid = msg["message_id"] - assert _MESSAGE_ID_STUB.match(mid), ( - f"{msg['id']}: message_id={mid!r} doesn't match the v1 " - f"stub shape `stub-msg-id-N`" - ) +# ---- 7. message_id pass-through (post-flip) ---------------------- +# +# The v1 stub used a hardcoded `stub-msg-id-N` shape. Post-flip, +# message_id passes through from the JSONL (RFC 5322 format — +# typically `>`). Per the PM-locked +# message_id carve-out in the bucket spec: +# +# "message_id may contain operator's domain in RFC 5322 format +# (e.g. ``). This is technically +# PII-adjacent BUT operator's domain is not personal identifi- +# cation. Decision: pass message_id through as-is (FE consumers +# need it for threading). The walk-payload guard should EXCLUDE +# message_id field from the email-regex check (false-positive +# otherwise)." +# +# The empty-JSONL test path doesn't exercise this; comprehensive +# message_id projection + carve-out tests live in +# ``test_web_server_email_panel_flip.py``. # ---- 8. SECURITY: companion FE pins ------------------------------ @@ -404,7 +343,9 @@ def test_panel_renders_spoofing_warning_chip(): async def test_by_direction_24h_sum_reconciles_to_total(_isolate_config): """The 24h direction breakdown must sum to total_recent_24h — otherwise the dashboard card's "X emails / Y flagged" headline - won't reconcile to the panel's per-direction breakdown.""" + won't reconcile to the panel's per-direction breakdown. Holds + on the empty-JSONL path (both zero) AND on populated paths + (see panel-flip tests).""" from kora_cli import web_server result = await web_server.list_recent_email() diff --git a/tests/kora_cli/test_web_server_email_panel_flip.py b/tests/kora_cli/test_web_server_email_panel_flip.py new file mode 100644 index 000000000000..abcbedb3ddd8 --- /dev/null +++ b/tests/kora_cli/test_web_server_email_panel_flip.py @@ -0,0 +1,795 @@ +"""Tests for KR-EMAIL-PANEL-FLIP — JSONL-driven projection. + +Bucket §2 + §4 scenarios: + + Endpoint shape / stub flip: + 1. Empty JSONLs → empty messages list + stub:false + 2. Only inbound JSONL present (outbound missing) → projects inbound only + 3. Only outbound JSONL present (inbound missing) → projects outbound only + 4. Both files present → merged + sorted newest-first + + Projection — inbound: + 5. handled_status=received → fe handled_status=received, + from_label=joshua (matching env), to_label=kora (matching env) + 6. Non-joshua sender → from_label=unknown_sender + 7. To-list missing Kora's address → to_label=other + 8. handled_status=filtered_paused / filtered_stopped → dropped_paused + 9. handled_status=filtered_non_joshua → filtered_non_allowlist (collapse) + 10. body_text_truncated_2k → body_text_truncated_400 truncated to 400 + 11. spoofing_check_skipped=True → spoofing_warning=False + 12. Unknown handled_status → entry skipped defensively + + Projection — outbound: + 13. send_status=ok → handled_status=sent_ok; from_label=kora + 14. send_status=failed → handled_status=sent_failed + 15. Recipient matches KORA_EMAIL_JOSHUA_ADDRESS → to_label=joshua + 16. body_text_truncated_400 = placeholder ("(outbound body not logged ...)") + 17. attachments_count=0 + has_html=False (outbound contract) + 18. in_reply_to passes through (may be null) + + Merge + sort: + 19. Newest-first by timestamp descending across both files + 20. Aggregate counts span both files within 24h window + 21. ?limit query param respected; cap at 200 + 22. Limit > 200 → capped to 200 (defense against runaway query) + 23. Limit < 1 → clamped to 1 + + Tolerance: + 24. Malformed JSONL line → logged + skipped; sibling lines parse + 25. JSON line that's not a dict (e.g., array) → skipped + 26. Inbound entry missing message_id → synthesized inbound-no-id-line-N + 27. Outbound entry missing message_id → synthesized outbound-no-id-line-N + 28. Outbound entry with empty recipients → skipped + + SECURITY: + 29. Walk-payload regex sweep finds NO email-address shape ANYWHERE + EXCEPT inside message_id values (the carve-out) + 30. message_id carve-out: legitimate `<...@operator-domain>` pattern + allowed when it appears in message_id; same pattern in OTHER + fields still flagged + 31. No raw email addresses bleed into subject / from_label / to_label / + body_text_truncated_400 / in_reply_to (in_reply_to follows the + same shape as message_id but per spec passes through too — + treated as message_id-class) + 32. No Purelymail token hints / HMAC secret shapes / Bearer headers +""" + +from __future__ import annotations + +import json +import logging +import re +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Dict, List + +import pytest + + +_JOSHUA_ADDR = "joshua@stormhavenenterprises.com" +_KORA_ADDR = "kora@stormhavenenterprises.com" +_OTHER_ADDR = "stranger@evil.example" + + +_EMAIL_ADDRESS = re.compile( + r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b" +) +_PUREMAIL_TOKEN_HINT = re.compile( + r"\b(?:KORA_PUREMAIL_|KORA_PURELYMAIL_|puremail_|purelymail_)[A-Za-z0-9_]*[A-Za-z0-9]\b" +) +_HEX_SECRET_SHAPE = re.compile(r"\b[0-9a-fA-F]{32,}\b") +_BEARER_TOKEN_SHAPE = re.compile( + r"\b(?:Bearer|Authorization)\s*[: ]\s*[A-Za-z0-9+/_.-]{8,}", + re.IGNORECASE, +) + +# Fields where the email-address regex IS allowed to match per the +# bucket's message_id carve-out. message_id values are RFC 5322 +# format (>). in_reply_to follows the same +# shape (it IS another message_id), so it gets the same carve-out. +_EMAIL_REGEX_ALLOWED_FIELDS = {"message_id", "in_reply_to"} + + +@pytest.fixture +def env(tmp_path, monkeypatch): + """Isolated KORA_HOME + Joshua/Kora env addresses set. + + Applies the CC#2 #137 fixture-isolation lesson: monkeypatch + ``get_kora_home`` in all 3 module namespaces (kora_constants, + kora_cli.config, kora_cli.web_server) because the endpoint + resolves it from its own module namespace via a + ``from kora_cli.config import get_kora_home`` re-import. + """ + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setenv("KORA_EMAIL_JOSHUA_ADDRESS", _JOSHUA_ADDR) + monkeypatch.setenv("KORA_EMAIL_KORA_ADDRESS", _KORA_ADDR) + monkeypatch.setattr("kora_constants.get_kora_home", lambda: tmp_path) + monkeypatch.setattr("kora_cli.config.get_kora_home", lambda: tmp_path) + monkeypatch.setattr( + "kora_cli.web_server.get_kora_home", lambda: tmp_path + ) + return tmp_path + + +def _iso(minutes_ago: int = 0) -> str: + ts = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago) + return ts.strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _write_jsonl(path: Path, entries: List[Dict[str, Any]]) -> None: + with path.open("w", encoding="utf-8") as f: + for entry in entries: + f.write(json.dumps(entry) + "\n") + + +def _inbound_entry( + *, + message_id: str = "", + from_addr: str = _JOSHUA_ADDR, + to_list: List[str] = None, + subject: str = "hello", + body_text_truncated_2k: str = "hello kora", + has_html: bool = False, + attachments_count: int = 0, + handled_status: str = "received", + spoofing_check_skipped: bool = True, + imap_uid: int = 1, + minutes_ago: int = 5, + in_reply_to: Any = None, +) -> Dict[str, Any]: + return { + "received_at": _iso(minutes_ago), + "message_id": message_id, + "from": from_addr, + "to": to_list if to_list is not None else [_KORA_ADDR], + "subject": subject, + "body_text_truncated_2k": body_text_truncated_2k, + "has_html": has_html, + "attachments_count": attachments_count, + "handled_status": handled_status, + "spoofing_check_skipped": spoofing_check_skipped, + "imap_uid": imap_uid, + "in_reply_to": in_reply_to, + } + + +def _outbound_entry( + *, + message_id: str = "", + to_list: List[str] = None, + subject: str = "Re: hello", + send_status: str = "ok", + in_reply_to: Any = "", + minutes_ago: int = 4, + smtp_code: int = 250, + error: Any = None, + retry_count: int = 0, +) -> Dict[str, Any]: + return { + "sent_at": _iso(minutes_ago), + "from": _KORA_ADDR, + "to": to_list if to_list is not None else [_JOSHUA_ADDR], + "subject": subject, + "in_reply_to": in_reply_to, + "send_status": send_status, + "message_id": message_id, + "smtp_code": smtp_code, + "error": error, + "retry_count": retry_count, + "caller_actor_kind": None, + } + + +def _inbound_path(env_path: Path) -> Path: + return env_path / "email_inbound_log.jsonl" + + +def _outbound_path(env_path: Path) -> Path: + return env_path / "email_outbound_log.jsonl" + + +# =========================================================================== +# Stub-flip: stub is False, empty list when no JSONLs +# =========================================================================== + + +@pytest.mark.asyncio +async def test_empty_files_returns_empty_with_stub_false(env): + from kora_cli import web_server + + result = await web_server.list_recent_email() + assert result["stub"] is False + assert result["messages"] == [] + assert result["total_recent_24h"] == 0 + + +@pytest.mark.asyncio +async def test_only_inbound_file_present(env): + from kora_cli import web_server + + _write_jsonl(_inbound_path(env), [_inbound_entry()]) + result = await web_server.list_recent_email() + assert result["stub"] is False + assert len(result["messages"]) == 1 + assert result["messages"][0]["direction"] == "inbound" + + +@pytest.mark.asyncio +async def test_only_outbound_file_present(env): + from kora_cli import web_server + + _write_jsonl(_outbound_path(env), [_outbound_entry()]) + result = await web_server.list_recent_email() + assert result["stub"] is False + assert len(result["messages"]) == 1 + assert result["messages"][0]["direction"] == "outbound" + + +# =========================================================================== +# Projection — inbound +# =========================================================================== + + +@pytest.mark.asyncio +async def test_inbound_joshua_received_projects_correctly(env): + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [ + _inbound_entry( + from_addr=_JOSHUA_ADDR, + to_list=[_KORA_ADDR], + subject="status?", + body_text_truncated_2k="how are you", + ) + ], + ) + result = await web_server.list_recent_email() + msg = result["messages"][0] + assert msg["direction"] == "inbound" + assert msg["from_label"] == "joshua" + assert msg["to_label"] == "kora" + assert msg["handled_status"] == "received" + assert msg["subject"] == "status?" + assert msg["body_text_truncated_400"] == "how are you" + assert msg["spoofing_warning"] is False + assert msg["id"].startswith("inbound-") + + +@pytest.mark.asyncio +async def test_inbound_non_joshua_sender_resolves_to_unknown(env): + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [_inbound_entry(from_addr=_OTHER_ADDR)], + ) + result = await web_server.list_recent_email() + assert result["messages"][0]["from_label"] == "unknown_sender" + + +@pytest.mark.asyncio +async def test_inbound_to_list_without_kora_resolves_to_other(env): + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [_inbound_entry(to_list=["someone-else@example.com"])], + ) + result = await web_server.list_recent_email() + assert result["messages"][0]["to_label"] == "other" + + +@pytest.mark.asyncio +async def test_inbound_paused_collapses_to_dropped_paused(env): + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [_inbound_entry(handled_status="filtered_paused")], + ) + result = await web_server.list_recent_email() + assert result["messages"][0]["handled_status"] == "dropped_paused" + + +@pytest.mark.asyncio +async def test_inbound_stopped_collapses_to_dropped_paused(env): + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [_inbound_entry(handled_status="filtered_stopped")], + ) + result = await web_server.list_recent_email() + assert result["messages"][0]["handled_status"] == "dropped_paused" + + +@pytest.mark.asyncio +async def test_inbound_non_joshua_collapses_to_non_allowlist(env): + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [_inbound_entry(handled_status="filtered_non_joshua")], + ) + result = await web_server.list_recent_email() + assert ( + result["messages"][0]["handled_status"] == "filtered_non_allowlist" + ) + + +@pytest.mark.asyncio +async def test_inbound_body_truncated_to_400(env): + from kora_cli import web_server + + long_body = "x" * 1000 + _write_jsonl( + _inbound_path(env), + [_inbound_entry(body_text_truncated_2k=long_body)], + ) + result = await web_server.list_recent_email() + assert len(result["messages"][0]["body_text_truncated_400"]) == 400 + + +@pytest.mark.asyncio +async def test_inbound_spoofing_check_skipped_means_no_warning(env): + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [_inbound_entry(spoofing_check_skipped=True)], + ) + result = await web_server.list_recent_email() + assert result["messages"][0]["spoofing_warning"] is False + + +@pytest.mark.asyncio +async def test_inbound_spoofing_check_ran_means_warning(env): + """If a future bucket adds real spoofing detection, an entry + with spoofing_check_skipped=False signals the check ran; + spoofing_warning becomes True. Documents the semantic flip in + the projection.""" + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [_inbound_entry(spoofing_check_skipped=False)], + ) + result = await web_server.list_recent_email() + assert result["messages"][0]["spoofing_warning"] is True + + +@pytest.mark.asyncio +async def test_inbound_unknown_handled_status_skipped(env): + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [_inbound_entry(handled_status="not_a_real_status")], + ) + result = await web_server.list_recent_email() + assert result["messages"] == [] + + +# =========================================================================== +# Projection — outbound +# =========================================================================== + + +@pytest.mark.asyncio +async def test_outbound_ok_projects_correctly(env): + from kora_cli import web_server + + _write_jsonl( + _outbound_path(env), + [_outbound_entry(send_status="ok")], + ) + result = await web_server.list_recent_email() + msg = result["messages"][0] + assert msg["direction"] == "outbound" + assert msg["from_label"] == "kora" + assert msg["to_label"] == "joshua" + assert msg["handled_status"] == "sent_ok" + assert msg["has_html"] is False + assert msg["attachments_count"] == 0 + assert msg["body_text_truncated_400"] == ( + "(outbound body not logged for size + privacy)" + ) + + +@pytest.mark.asyncio +async def test_outbound_failed_projects_sent_failed(env): + from kora_cli import web_server + + _write_jsonl( + _outbound_path(env), + [_outbound_entry(send_status="failed", error="boom")], + ) + result = await web_server.list_recent_email() + assert result["messages"][0]["handled_status"] == "sent_failed" + + +@pytest.mark.asyncio +async def test_outbound_to_non_joshua_resolves_to_other(env): + from kora_cli import web_server + + _write_jsonl( + _outbound_path(env), + [_outbound_entry(to_list=["random@example.com"])], + ) + result = await web_server.list_recent_email() + assert result["messages"][0]["to_label"] == "other" + + +@pytest.mark.asyncio +async def test_outbound_in_reply_to_passes_through(env): + from kora_cli import web_server + + _write_jsonl( + _outbound_path(env), + [_outbound_entry(in_reply_to="")], + ) + result = await web_server.list_recent_email() + assert result["messages"][0]["in_reply_to"] == ( + "" + ) + + +@pytest.mark.asyncio +async def test_outbound_in_reply_to_null_passes_through(env): + from kora_cli import web_server + + _write_jsonl( + _outbound_path(env), + [_outbound_entry(in_reply_to=None)], + ) + result = await web_server.list_recent_email() + assert result["messages"][0]["in_reply_to"] is None + + +@pytest.mark.asyncio +async def test_outbound_empty_recipients_skipped(env): + from kora_cli import web_server + + _write_jsonl( + _outbound_path(env), + [_outbound_entry(to_list=[])], + ) + result = await web_server.list_recent_email() + assert result["messages"] == [] + + +@pytest.mark.asyncio +async def test_outbound_unknown_send_status_skipped(env): + from kora_cli import web_server + + _write_jsonl( + _outbound_path(env), + [_outbound_entry(send_status="pending")], + ) + result = await web_server.list_recent_email() + assert result["messages"] == [] + + +# =========================================================================== +# Merge + sort + limit +# =========================================================================== + + +@pytest.mark.asyncio +async def test_merge_sorts_newest_first(env): + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [ + _inbound_entry( + message_id="", + subject="old inbound", + minutes_ago=20, + ), + _inbound_entry( + message_id="", + subject="new inbound", + minutes_ago=2, + ), + ], + ) + _write_jsonl( + _outbound_path(env), + [ + _outbound_entry( + message_id="", + subject="Re: mid", + minutes_ago=10, + ) + ], + ) + result = await web_server.list_recent_email() + subjects = [m["subject"] for m in result["messages"]] + assert subjects == ["new inbound", "Re: mid", "old inbound"] + + +@pytest.mark.asyncio +async def test_limit_query_param_respected(env): + from kora_cli import web_server + + entries = [ + _inbound_entry( + message_id=f"", + subject=f"msg-{i}", + minutes_ago=i, + ) + for i in range(20) + ] + _write_jsonl(_inbound_path(env), entries) + result = await web_server.list_recent_email(limit=5) + assert len(result["messages"]) == 5 + + +@pytest.mark.asyncio +async def test_limit_caps_at_200(env): + from kora_cli import web_server + + entries = [ + _inbound_entry( + message_id=f"", + subject=f"msg-{i}", + minutes_ago=i, + ) + for i in range(220) + ] + _write_jsonl(_inbound_path(env), entries) + result = await web_server.list_recent_email(limit=500) + assert len(result["messages"]) == 200 + + +@pytest.mark.asyncio +async def test_limit_below_one_clamps_to_one(env): + from kora_cli import web_server + + _write_jsonl(_inbound_path(env), [_inbound_entry()]) + result = await web_server.list_recent_email(limit=0) + assert len(result["messages"]) == 1 + + +# =========================================================================== +# Aggregate counts within 24h window +# =========================================================================== + + +@pytest.mark.asyncio +async def test_aggregate_counts_within_24h_window(env): + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [ + _inbound_entry(message_id="", minutes_ago=5), + _inbound_entry(message_id="", minutes_ago=10), + # Outside 24h window — should NOT count + _inbound_entry( + message_id="", + minutes_ago=60 * 25, + ), + ], + ) + _write_jsonl( + _outbound_path(env), + [_outbound_entry(message_id="", minutes_ago=4)], + ) + result = await web_server.list_recent_email() + assert result["total_recent_24h"] == 3 + assert result["by_direction_24h"] == {"inbound": 2, "outbound": 1} + assert result["by_status_24h"]["received"] == 2 + assert result["by_status_24h"]["sent_ok"] == 1 + + +# =========================================================================== +# Tolerance +# =========================================================================== + + +@pytest.mark.asyncio +async def test_malformed_line_logged_and_skipped(env, caplog): + from kora_cli import web_server + + path = _inbound_path(env) + path.write_text( + json.dumps(_inbound_entry(message_id="")) + "\n" + "{not-valid-json\n" + + json.dumps(_inbound_entry(message_id="")) + "\n", + encoding="utf-8", + ) + with caplog.at_level(logging.WARNING): + result = await web_server.list_recent_email() + assert len(result["messages"]) == 2 + assert any( + "malformed JSON" in record.message for record in caplog.records + ) + + +@pytest.mark.asyncio +async def test_non_dict_line_skipped(env): + from kora_cli import web_server + + path = _inbound_path(env) + path.write_text( + json.dumps([1, 2, 3]) + "\n" + + json.dumps(_inbound_entry()) + "\n", + encoding="utf-8", + ) + result = await web_server.list_recent_email() + assert len(result["messages"]) == 1 + + +@pytest.mark.asyncio +async def test_inbound_missing_message_id_synthesized(env): + from kora_cli import web_server + + entry = _inbound_entry() + del entry["message_id"] + _write_jsonl(_inbound_path(env), [entry]) + result = await web_server.list_recent_email() + assert "inbound-no-id-line-" in result["messages"][0]["message_id"] + + +@pytest.mark.asyncio +async def test_outbound_missing_message_id_synthesized(env): + from kora_cli import web_server + + entry = _outbound_entry() + del entry["message_id"] + _write_jsonl(_outbound_path(env), [entry]) + result = await web_server.list_recent_email() + assert "outbound-no-id-line-" in result["messages"][0]["message_id"] + + +# =========================================================================== +# SECURITY — walk-payload sweep with message_id carve-out +# =========================================================================== + + +def _walk_payload(value: Any, *, exclude_fields: set) -> str: + """Serialize ``value`` to JSON for regex walking, but replace any + excluded-field value with a placeholder so the email-regex + sweep doesn't false-positive on the carve-out fields.""" + if isinstance(value, dict): + scrubbed = { + k: ("" if k in exclude_fields else _walk_payload( + v, exclude_fields=exclude_fields + )) + for k, v in value.items() + } + return json.dumps(scrubbed) + if isinstance(value, list): + return json.dumps( + [ + json.loads( + _walk_payload(item, exclude_fields=exclude_fields) + ) + if isinstance(item, (dict, list)) + else item + for item in value + ] + ) + return json.dumps(value) + + +@pytest.mark.asyncio +async def test_no_email_addresses_outside_message_id_carve_out(env): + """Walk-payload regex sweep: NO email-address shape anywhere + EXCEPT in message_id / in_reply_to (carve-out per bucket §2(b) + — RFC 5322 message-ids legitimately contain the operator + domain).""" + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [ + _inbound_entry( + message_id="", + from_addr=_JOSHUA_ADDR, + to_list=[_KORA_ADDR], + subject="status", + body_text_truncated_2k="how's it going", + in_reply_to="", + ) + ], + ) + _write_jsonl( + _outbound_path(env), + [ + _outbound_entry( + message_id="", + in_reply_to="", + ) + ], + ) + result = await web_server.list_recent_email() + blob = _walk_payload(result, exclude_fields=_EMAIL_REGEX_ALLOWED_FIELDS) + leaks = _EMAIL_ADDRESS.findall(blob) + assert leaks == [], ( + f"payload contains email-address shape(s) OUTSIDE the " + f"message_id/in_reply_to carve-out: {leaks}" + ) + + +@pytest.mark.asyncio +async def test_message_id_carve_out_allows_rfc5322_format(env): + """The carve-out test: legitimate `` IS + allowed in message_id even though it shape-matches an email + address. FE consumers need it for threading.""" + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [ + _inbound_entry( + message_id="", + ) + ], + ) + result = await web_server.list_recent_email() + msg = result["messages"][0] + # message_id contains an @ + domain — that's allowed. + assert "@" in msg["message_id"] + assert msg["message_id"] == "" + + +@pytest.mark.asyncio +async def test_no_email_in_subject_or_body(env): + """Defense: if an inbound subject or body contains an email + address (e.g., a quoted thread or signature), it does end up + in the payload — but the walk-payload sweep at the panel + layer treats subject+body as user content. We DON'T strip + email addresses from those fields because doing so would + mangle Joshua's actual messages. + + Instead this test pins the EXPECTED state: a subject + body + that DON'T contain emails come through clean. A separate + follow-on bucket can decide whether to strip user-content + addresses for display — out of scope for the flip.""" + from kora_cli import web_server + + _write_jsonl( + _inbound_path(env), + [ + _inbound_entry( + subject="status check (no addresses)", + body_text_truncated_2k="just plain text without addresses", + ) + ], + ) + result = await web_server.list_recent_email() + blob = _walk_payload(result, exclude_fields=_EMAIL_REGEX_ALLOWED_FIELDS) + leaks = _EMAIL_ADDRESS.findall(blob) + assert leaks == [] + + +@pytest.mark.asyncio +async def test_no_purelymail_token_hints_anywhere(env): + from kora_cli import web_server + + _write_jsonl(_inbound_path(env), [_inbound_entry()]) + _write_jsonl(_outbound_path(env), [_outbound_entry()]) + result = await web_server.list_recent_email() + blob = json.dumps(result) + leaks = _PUREMAIL_TOKEN_HINT.findall(blob) + assert leaks == [], ( + f"payload contains Purelymail token hint(s): {leaks}" + ) + + +@pytest.mark.asyncio +async def test_no_hex_secret_or_bearer_anywhere(env): + from kora_cli import web_server + + _write_jsonl(_inbound_path(env), [_inbound_entry()]) + _write_jsonl(_outbound_path(env), [_outbound_entry()]) + result = await web_server.list_recent_email() + blob = json.dumps(result) + assert _HEX_SECRET_SHAPE.findall(blob) == [] + assert _BEARER_TOKEN_SHAPE.findall(blob) == []