feat(pebble): phase-0 — search, audit hardening, L1 chat orchestrator

financial_forecasting/auth.py

@@ -142,6 +142,50 @@ async def get_current_user_dep(request: Request) -> Optional[Dict]:
 
 _BEDROCK_INTERNAL_API_KEY = os.getenv("BEDROCK_INTERNAL_API_KEY", "")
 
+# HTTP methods considered "writes" for the kill switch below. Idempotent reads
+# (GET, HEAD, OPTIONS) remain available even when the switch is on so that
+# Pebble's research/lookup paths keep working during write-side incidents.
+_WRITE_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"})
+
+# Default scope grant for the internal-key principal. v1.0 grants the
+# superset; later phases tighten via per-call X-Pebble-Scopes header.
+# `*` is interpreted as "all scopes" by check_permission_or_internal.
+_DEFAULT_INTERNAL_SCOPES = ("*",)
+
+
+def _pebble_writes_disabled() -> bool:
+    """Read the kill switch fresh on every call so operators can flip it
+    without a redeploy. Truthy values: "true" / "1" / "yes" (case-insensitive).
+    """
+    return os.getenv("PEBBLE_WRITES_DISABLED", "").strip().lower() in {"true", "1", "yes"}
+
+
+def _parse_scopes(header_value: str) -> tuple[str, ...]:
+    """Parse the optional X-Pebble-Scopes header. Comma-separated scope
+    strings; empty / unset header = full grant via the wildcard ``*``.
+    """
+    raw = (header_value or "").strip()
+    if not raw:
+        return _DEFAULT_INTERNAL_SCOPES
+    scopes = tuple(s.strip() for s in raw.split(",") if s.strip())
+    return scopes or _DEFAULT_INTERNAL_SCOPES
+
+
+def _is_valid_originating_user_email(email: str) -> bool:
+    """Lightweight shape check on X-Originating-User. We don't verify
+    membership in org_users here (would require a DB hit on every
+    request); the route-level audit insert (FK to org_users via
+    originating_user_email lookup) is the durable check. Spoofs against
+    the API key still produce auditable rows.
+    """
+    if not email or len(email) > 254:
+        return False
+    if "@" not in email or email.startswith("@") or email.endswith("@"):
+        return False
+    if any(c.isspace() for c in email):
+        return False
+    return True
+
 
 async def require_auth_or_internal(request: Request) -> Dict:
     """Authorize via internal API key (service-to-service) or user JWT.
@@ -150,13 +194,101 @@ async def require_auth_or_internal(request: Request) -> Dict:
     synthetic service user dict.  Otherwise falls back to require_auth.
     Dev mode: if BEDROCK_INTERNAL_API_KEY is empty, internal key check
     is skipped and only JWT auth is tried.
+
+    **Mandatory headers when using internal-key auth (Phase 0.2):**
+
+    - ``X-Originating-User``: the human whose session triggered this
+      service-to-service call. Required on EVERY internal-key request,
+      reads and writes alike. The email is propagated to
+      ``bedrock.pebble_write_audit`` and to permission resolution so
+      that Pebble acts as a delegated principal, not a god principal.
+      Three independent adversarial reviews flagged the un-attributed
+      "service:pebble" pattern as a 1.0 blocker; this enforcement
+      closes that gap.
+    - ``X-Request-Id``: a UUIDv7 (or any UUID) for replay defense via
+      ``UNIQUE(request_id)`` on ``bedrock.pebble_write_audit``. Format
+      checked here; uniqueness checked at the audit-row INSERT in the
+      route handler.
+
+    Optional header:
+
+    - ``X-Pebble-Scopes``: comma-separated scopes the caller is
+      requesting for THIS request. Default = ``("*",)`` (full grant).
+      ``check_permission_or_internal`` verifies the matching scope is
+      present. Future tightening: Pebble will request narrow scopes
+      per call instead of carrying the full grant.
+
+    Kill switch: if PEBBLE_WRITES_DISABLED env is truthy AND the caller
+    is using a valid internal key AND the request method is a write
+    (POST/PUT/PATCH/DELETE), return 503. Reads via internal key and all
+    JWT-authenticated requests are unaffected.
     """
     internal_key = request.headers.get("X-Internal-Key", "")
     if _BEDROCK_INTERNAL_API_KEY and internal_key:
         if hmac.compare_digest(internal_key, _BEDROCK_INTERNAL_API_KEY):
+            if _pebble_writes_disabled() and request.method in _WRITE_METHODS:
+                logger.warning(
+                    "pebble_writes_disabled: blocking %s %s for service caller",
+                    request.method,
+                    request.url.path,
+                )
+                raise HTTPException(
+                    status_code=503,
+                    detail={
+                        "error": "pebble_writes_disabled",
+                        "message": (
+                            "Pebble service-account writes are temporarily "
+                            "disabled. Reads remain available."
+                        ),
+                    },
+                )
+
+            originating_user = request.headers.get("X-Originating-User", "").strip()
+            if not _is_valid_originating_user_email(originating_user):
+                logger.warning(
+                    "internal_key_missing_originating_user: %s %s",
+                    request.method,
+                    request.url.path,
+                )
+                raise HTTPException(
+                    status_code=401,
+                    detail={
+                        "error": "originating_user_required",
+                        "message": (
+                            "X-Originating-User header is required on every "
+                            "internal-key request. Pebble acts on behalf of a "
+                            "specific user, never as itself."
+                        ),
+                    },
+                )
+
+            request_id = request.headers.get("X-Request-Id", "").strip()
+            if not request_id:
+                logger.warning(
+                    "internal_key_missing_request_id: %s %s",
+                    request.method,
+                    request.url.path,
+                )
+                raise HTTPException(
+                    status_code=401,
+                    detail={
+                        "error": "request_id_required",
+                        "message": (
+                            "X-Request-Id header is required on every "
+                            "internal-key request for replay defense. "
+                            "Pass a UUID."
+                        ),
+                    },
+                )
+
+            scopes = _parse_scopes(request.headers.get("X-Pebble-Scopes", ""))
+
             return {
                 "user_id": "service:pebble",
                 "email": "pebble@internal",
                 "is_service": True,
+                "originating_user_email": originating_user,
+                "request_id": request_id,
+                "scopes": scopes,
             }
     return await require_auth(request)

financial_forecasting/db/migrations/2026-05-06-pebble-write-audit.sql

@@ -0,0 +1,150 @@
+-- 2026-05-06: bedrock.pebble_write_audit + bedrock.search_audit_access_log
+--
+-- Phase 0.1 + 0.7 of the Pebble 1.0 plan (tasks/pebble-search-spec.md).
+--
+-- Why:
+--     When Pebble (or any service-to-service caller using
+--     X-Internal-Key) writes through Bedrock's API, today's audit
+--     trail collapses to a single synthetic user "service:pebble"
+--     across logger.info lines and SF "LastModifiedById" — the
+--     originating end-user is unrecoverable. Three independent
+--     reviews (security/UX/architecture adversaries) flagged this
+--     as a 1.0 blocker.
+--
+--     bedrock.pebble_write_audit captures every internal-key write
+--     with the originating user, request id, route, payload hash,
+--     and response status. UNIQUE(request_id) doubles as the
+--     idempotency / replay-defense store for X-Request-Id (Phase
+--     0.7) — a duplicate request_id within 24h yields a 409 from
+--     the API rather than a duplicate write.
+--
+--     bedrock.search_audit_access_log captures admins reading
+--     other users' search history when query_text is revealed.
+--     Lives separately from the search_audit table itself so that
+--     RTBF / DSAR purges of search_audit cannot also erase the
+--     accountability trail of who looked at what.
+--
+-- Related:
+--     * tasks/pebble-search-spec.md (decisions §3, §4)
+--     * tasks/pebble-search-spec-security.md §1.5, §2, §10.4
+--     * tasks/pebble-overhaul-plan.md §0.1, §0.7
+--     * financial_forecasting/auth.py:require_auth_or_internal
+--       (enforces X-Originating-User; this table receives the
+--       audit row from each write route)
+--
+-- Idempotent — safe to re-run.
+--
+-- Apply as bedrock owner:
+--     psql "$DATABASE_URL" -f 2026-05-06-pebble-write-audit.sql
+
+CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
+CREATE EXTENSION IF NOT EXISTS citext;
+
+CREATE TABLE IF NOT EXISTS bedrock.pebble_write_audit (
+    id                          UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    occurred_at                 TIMESTAMPTZ NOT NULL DEFAULT now(),
+
+    -- Idempotency / replay defense. Required on every internal-key
+    -- write; UNIQUE means a 24h replay returns 409 from INSERT.
+    request_id                  UUID NOT NULL,
+
+    -- Routing / classification.
+    route                       TEXT NOT NULL,                   -- e.g. "/api/opportunities/update-stage"
+    http_method                 TEXT NOT NULL CHECK (http_method IN
+                                  ('POST', 'PUT', 'PATCH', 'DELETE')),
+
+    -- Subject. nullable so an early-failure audit row (e.g. invalid
+    -- payload) still gets written for forensics.
+    sf_object_type              TEXT,                            -- 'Opportunity' | 'Account' | 'Contact' | 'Task' | 'Payment' | 'Award' | 'Project'
+    sf_object_id                TEXT,
+
+    -- Attribution. originating_user_email is the human whose
+    -- session triggered the workflow; service_user is "service:pebble"
+    -- (or future siblings). Both required.
+    originating_user_email      CITEXT NOT NULL,
+    service_user                TEXT NOT NULL DEFAULT 'service:pebble',
+
+    -- What was sent. payload_hash = sha256 of canonical JSON; full
+    -- payload kept for 90 days where the route opts in (sensitive
+    -- routes pass NULL to suppress).
+    payload_hash                TEXT,
+    payload                     JSONB,
+
+    -- What came back.
+    response_status             SMALLINT NOT NULL,
+    response_summary            JSONB,                           -- e.g. {"award_created": true, "stage": "..."}
+    error_class                 TEXT,                            -- ExceptionClass name when status >= 400
+
+    -- Side-effect attestation. ensure_for_opp + future event-bus
+    -- subscribers write back here so a forensic reader can see the
+    -- full chain from one INSERT (vs. correlating across 3 tables).
+    side_effects                JSONB,                           -- e.g. {"award_created": true, "activity_logged": true}
+
+    -- Latency for ops dashboards.
+    latency_ms                  INT,
+
+    -- Multi-tenant outer guard. Single value today, present so
+    -- search_audit / pebble_audit share an enforcement pattern.
+    org_id                      TEXT NOT NULL DEFAULT 'pursuit',
+
+    -- Replay defense: the duplicate-request_id detector relies on this.
+    UNIQUE (request_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_pebble_write_audit_user_time
+    ON bedrock.pebble_write_audit(originating_user_email, occurred_at DESC);
+CREATE INDEX IF NOT EXISTS idx_pebble_write_audit_object
+    ON bedrock.pebble_write_audit(sf_object_type, sf_object_id, occurred_at DESC)
+    WHERE sf_object_id IS NOT NULL;
+CREATE INDEX IF NOT EXISTS idx_pebble_write_audit_errors
+    ON bedrock.pebble_write_audit(occurred_at DESC)
+    WHERE response_status >= 400;
+CREATE INDEX IF NOT EXISTS idx_pebble_write_audit_route_time
+    ON bedrock.pebble_write_audit(route, occurred_at DESC);
+
+COMMENT ON TABLE bedrock.pebble_write_audit IS
+    'Audit row for every internal-key write through Bedrock (Phase 0.1). UNIQUE(request_id) doubles as 24h replay defense (Phase 0.7).';
+COMMENT ON COLUMN bedrock.pebble_write_audit.originating_user_email IS
+    'Mandatory. The human whose session triggered the workflow. Sourced from X-Originating-User header. require_auth_or_internal rejects calls without it.';
+COMMENT ON COLUMN bedrock.pebble_write_audit.request_id IS
+    'Mandatory. UUIDv7 from X-Request-Id header. Replay window = 24h via UNIQUE constraint; deduplicate at the route via INSERT...ON CONFLICT (request_id) DO NOTHING RETURNING id.';
+
+-- ---------------------------------------------------------------------------
+-- Admin access log for revealing search_audit.query_text
+-- ---------------------------------------------------------------------------
+-- search_audit itself comes in a later migration (Layer 1). This table is
+-- separate so RTBF/DSAR purges of search_audit cannot also erase
+-- accountability.
+CREATE TABLE IF NOT EXISTS bedrock.search_audit_access_log (
+    id                  UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    accessed_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
+    admin_email         CITEXT NOT NULL,
+    target_user_email   CITEXT NOT NULL,
+    revealed_query_text BOOLEAN NOT NULL,
+    reason              TEXT NOT NULL,
+    request_id          UUID NOT NULL,
+    org_id              TEXT NOT NULL DEFAULT 'pursuit'
+);
+
+CREATE INDEX IF NOT EXISTS idx_search_audit_access_admin
+    ON bedrock.search_audit_access_log(admin_email, accessed_at DESC);
+CREATE INDEX IF NOT EXISTS idx_search_audit_access_target
+    ON bedrock.search_audit_access_log(target_user_email, accessed_at DESC);
+
+COMMENT ON TABLE bedrock.search_audit_access_log IS
+    'Append-only log of admins viewing other users search history with query_text revealed. Survives RTBF purges.';
+
+-- ---------------------------------------------------------------------------
+-- Grants — bedrock_user owns these tables and writes through the API.
+-- ---------------------------------------------------------------------------
+DO $$
+BEGIN
+    IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'bedrock_user') THEN
+        GRANT SELECT, INSERT ON bedrock.pebble_write_audit TO bedrock_user;
+        GRANT SELECT, INSERT ON bedrock.search_audit_access_log TO bedrock_user;
+        -- UPDATE intentionally NOT granted on pebble_write_audit:
+        -- audit rows are append-only (immutability matters for forensics).
+        -- DELETE limited to a separate retention job role, added when
+        -- the 90d cleanup script lands.
+    END IF;
+END $$;