diff --git a/services/ingestion/app/integration_router.py b/services/ingestion/app/integration_router.py index 4eba6eff8..a64e16804 100644 --- a/services/ingestion/app/integration_router.py +++ b/services/ingestion/app/integration_router.py @@ -7,23 +7,42 @@ - Triggering manual syncs - CSV upload for the generic CSV/SFTP connector - Receiving inbound webhooks from external systems + +Hardening (findings 2-6; finding 1 credential store deferred): + +- Rate limits on configure / sync / csv-upload / disconnect / profile + mutations and the inbound webhook (shared slowapi limiter). +- Webhook path fails closed with a uniform 401 body (no secret-state + oracle) and rejects replayed signature digests inside a short TTL + nonce window. +- CSV ingest runs every cell through ``sanitize_cell`` before events + leave the upload handler (formula-injection neutralization). +- Request models are bounded (``extra="forbid"`` + max_length / ge/le). +- Tenant-bound audit events on configure / disconnect / sync / + csv-upload / profile mutations attribute the signed-in principal + (``ctx.principal_id`` / ``principal_kind`` / ``actor_email``) and + never carry secret material. """ from __future__ import annotations +import hashlib import json import logging +import threading +import time import uuid from datetime import datetime, timezone from typing import Any, Dict, List, Literal, Optional from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field from shared.database import get_db_safe from shared.rate_limit import limiter from shared.tenant_context import TenantContext from sqlalchemy import text +from .shared.csv_safety import sanitize_cell from .shared.upload_limits import ( MAX_CSV_FILE_SIZE_BYTES, read_upload_with_limit, @@ -35,29 +54,61 @@ router = APIRouter(prefix="/api/v1/integrations", tags=["Integrations"]) +# Abuse budgets (finding 2). Credential configure and sync are the highest +# risk; webhook already had a limit and keeps 60/min. Profile edits are +# low-frequency human actions. +_CONFIGURE_RATE_LIMIT = "10/minute" +_SYNC_RATE_LIMIT = "10/minute" +_CSV_UPLOAD_RATE_LIMIT = "20/minute" +_DISCONNECT_RATE_LIMIT = "20/minute" +_PROFILE_MUTATION_RATE_LIMIT = "30/minute" +_WEBHOOK_RATE_LIMIT = "60/minute" + +# Replay window for webhook signature digests (finding 3). Same digest +# inside this TTL is treated as a replay and rejected with the same +# uniform 401 body as any other verification failure (no oracle). +_WEBHOOK_NONCE_TTL_SECONDS = 300 +_webhook_nonce_lock = threading.Lock() +_webhook_nonce_seen: dict[str, float] = {} + +# Uniform webhook rejection body — never distinguish missing secret, +# missing signature, bad digest, stale timestamp, or replay. +_WEBHOOK_REJECTED_DETAIL = {"error": "webhook_rejected", "message": "Webhook verification failed"} + + # ── Models ──────────────────────────────────────────────────── class ConnectorConfigRequest(BaseModel): - """Request to configure a connector.""" + """Request to configure a connector. - connector_id: str - api_key: Optional[str] = None - api_secret: Optional[str] = None - oauth_client_id: Optional[str] = None - oauth_client_secret: Optional[str] = None - webhook_secret: Optional[str] = None - base_url: Optional[str] = None - extra: Dict[str, Any] = Field(default_factory=dict) + Finding 5: bounded + ``extra="forbid"``. Secret fields are accepted + on write but never echoed by the configure response. Finding 1 + (credential store) is deferred — secrets still live in the + process-local registry until that lands. + """ + + model_config = ConfigDict(extra="forbid") + + connector_id: str = Field(..., min_length=1, max_length=120) + api_key: Optional[str] = Field(default=None, max_length=4096) + api_secret: Optional[str] = Field(default=None, max_length=4096) + oauth_client_id: Optional[str] = Field(default=None, max_length=512) + oauth_client_secret: Optional[str] = Field(default=None, max_length=4096) + webhook_secret: Optional[str] = Field(default=None, max_length=4096) + base_url: Optional[str] = Field(default=None, max_length=2048) + extra: Dict[str, Any] = Field(default_factory=dict, max_length=50) class SyncRequest(BaseModel): """Request to trigger a manual sync.""" - connector_id: str - since: Optional[str] = None # ISO 8601 - limit: int = 100 + model_config = ConfigDict(extra="forbid") + + connector_id: str = Field(..., min_length=1, max_length=120) + since: Optional[str] = Field(default=None, max_length=40) # ISO 8601 + limit: int = Field(default=100, ge=1, le=1000) ProfileSourceType = Literal["csv", "edi", "epcis", "api", "webhook", "spreadsheet", "supplier_portal"] @@ -100,10 +151,12 @@ class IntegrationProfile(BaseModel): class CreateIntegrationProfileRequest(BaseModel): """Create a saved mapping profile for a supplier or source system.""" + model_config = ConfigDict(extra="forbid") + display_name: str = Field(..., min_length=2, max_length=120) source_type: ProfileSourceType = "csv" - field_mapping: Dict[str, str] = Field(default_factory=lambda: dict(DEFAULT_PROFILE_MAPPING)) - default_cte_type: str = "shipping" + field_mapping: Dict[str, str] = Field(default_factory=lambda: dict(DEFAULT_PROFILE_MAPPING), max_length=64) + default_cte_type: str = Field(default="shipping", max_length=64) status: ProfileStatus = "active" confidence: float = Field(default=0.75, ge=0, le=1) supplier_id: Optional[str] = Field(default=None, max_length=120) @@ -114,10 +167,12 @@ class CreateIntegrationProfileRequest(BaseModel): class UpdateIntegrationProfileRequest(BaseModel): """Patch mutable fields on a saved integration profile.""" + model_config = ConfigDict(extra="forbid") + display_name: Optional[str] = Field(default=None, min_length=2, max_length=120) source_type: Optional[ProfileSourceType] = None - field_mapping: Optional[Dict[str, str]] = None - default_cte_type: Optional[str] = None + field_mapping: Optional[Dict[str, str]] = Field(default=None, max_length=64) + default_cte_type: Optional[str] = Field(default=None, max_length=64) status: Optional[ProfileStatus] = None confidence: Optional[float] = Field(default=None, ge=0, le=1) supplier_id: Optional[str] = Field(default=None, max_length=120) @@ -126,6 +181,8 @@ class UpdateIntegrationProfileRequest(BaseModel): class MappingPreviewRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + events: List[Dict[str, Any]] = Field(default_factory=list, max_length=50) @@ -352,6 +409,146 @@ def _apply_profile_mapping(event: Dict[str, Any], profile: IntegrationProfile) - return mapped, missing +# ── Hardening helpers (findings 3, 4, 6) ─────────────────────── + + +def _sanitize_value(value: Any) -> Any: + """Recursively neutralize CSV formula prefixes in string leaves.""" + if isinstance(value, str): + return sanitize_cell(value) + if isinstance(value, dict): + return {str(k): _sanitize_value(v) for k, v in value.items()} + if isinstance(value, list): + return [_sanitize_value(v) for v in value] + return value + + +def _sanitize_ingest_events(events: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Apply ``sanitize_cell`` to every string in parse_csv ingest dicts.""" + return [_sanitize_value(event) for event in events] + + +def _webhook_nonce_key(tenant_id: str, connector_id: str, payload: bytes, signature_header: str) -> str: + """Stable digest for replay detection (tenant + connector + body + sig).""" + material = b"|".join( + [ + tenant_id.encode("utf-8"), + connector_id.encode("utf-8"), + payload, + signature_header.encode("utf-8"), + ] + ) + return hashlib.sha256(material).hexdigest() + + +def _webhook_nonce_seen_before(nonce_key: str) -> bool: + """Return True if this nonce was already accepted inside the TTL window. + + Does not record the key — call ``_webhook_nonce_remember`` only after + successful verification so a failed attempt does not burn the nonce. + """ + now = time.monotonic() + with _webhook_nonce_lock: + stale = [k for k, ts in _webhook_nonce_seen.items() if now - ts > _WEBHOOK_NONCE_TTL_SECONDS] + for k in stale: + _webhook_nonce_seen.pop(k, None) + return nonce_key in _webhook_nonce_seen + + +def _webhook_nonce_remember(nonce_key: str) -> None: + """Record a successfully verified webhook digest for the replay TTL.""" + with _webhook_nonce_lock: + _webhook_nonce_seen[nonce_key] = time.monotonic() + + +def _reset_webhook_nonces_for_tests() -> None: + """Test-only: clear the in-process webhook nonce cache.""" + with _webhook_nonce_lock: + _webhook_nonce_seen.clear() + + +def _audit_task_done(task) -> None: + """Fail loud, not silent: a detached audit write that raised must log.""" + import asyncio + + try: + task.result() + except asyncio.CancelledError: + pass + except Exception as exc: # pragma: no cover - defensive + logger.error("integration_audit_write_failed error=%s", exc) + + +def _audit_integration_event( + *, + ctx: TenantContext, + event_type_name: str, + action: str, + tenant_id: str, + connector_or_profile_id: str, + resource_type: str, + changed_fields: list[str], +) -> None: + """Fire-and-forget tenant-bound audit for an integration mutation. + + Best-effort: a successful mutation never fails because the audit + write could not be scheduled. Details carry field NAMES and ids + only — never api_key / api_secret / oauth / webhook_secret / extra + secret material (finding 1 / finding 6). + """ + try: + import asyncio + + from shared.audit_logging import ( + AuditActor, + AuditEventCategory, + AuditEventType, + AuditLogger, + AuditResource, + AuditSeverity, + ) + + event_type = getattr(AuditEventType, event_type_name) + audit = AuditLogger.get_instance() + actor = AuditActor( + actor_id=ctx.principal_id, + actor_type=ctx.principal_kind, + email=ctx.actor_email, + tenant_id=tenant_id, + ) + resource = AuditResource( + resource_type=resource_type, + resource_id=connector_or_profile_id, + tenant_id=tenant_id, + ) + coro = audit.log( + event_type=event_type, + category=AuditEventCategory.CONFIGURATION, + severity=AuditSeverity.INFO, + actor=actor, + action=action, + outcome="success", + resource=resource, + message=f"integration {action}", + details={ + "tenant_id": tenant_id, + "changed_fields": changed_fields, + "resource_id": connector_or_profile_id, + }, + tags=["integration", "connector"], + ) + try: + loop = asyncio.get_running_loop() + loop.create_task(coro).add_done_callback(_audit_task_done) + except RuntimeError: + try: + coro.close() + except Exception: + pass + except Exception: # pragma: no cover - defensive + logger.debug("audit_integration_event: could not emit audit event", exc_info=True) + + # ── Endpoints ───────────────────────────────────────────────── @@ -376,9 +573,11 @@ async def list_integration_profiles( response_model=IntegrationProfile, summary="Create a saved integration profile", ) +@limiter.limit(_PROFILE_MUTATION_RATE_LIMIT) async def create_integration_profile( + request: Request, tenant_id: str, - request: CreateIntegrationProfileRequest, + body: CreateIntegrationProfileRequest, ctx: TenantContext = Depends(require_tenant_context), ) -> IntegrationProfile: """Create a reusable field mapping profile for supplier onboarding.""" @@ -389,9 +588,19 @@ async def create_integration_profile( tenant_id=tenant_id, created_at=now, updated_at=now, - **request.model_dump(), + **body.model_dump(), ) - return _save_profile(profile) + saved = _save_profile(profile) + _audit_integration_event( + ctx=ctx, + event_type_name="DATA_CREATE", + action="profile_create", + tenant_id=tenant_id, + connector_or_profile_id=saved.profile_id, + resource_type="integration_profile", + changed_fields=sorted(body.model_dump(exclude_unset=True).keys()) or ["display_name"], + ) + return saved @router.get( @@ -414,18 +623,30 @@ async def get_integration_profile( response_model=IntegrationProfile, summary="Update a saved integration profile", ) +@limiter.limit(_PROFILE_MUTATION_RATE_LIMIT) async def update_integration_profile( + request: Request, tenant_id: str, profile_id: str, - request: UpdateIntegrationProfileRequest, + body: UpdateIntegrationProfileRequest, ctx: TenantContext = Depends(require_tenant_context), ) -> IntegrationProfile: """Patch a saved supplier/source mapping profile.""" tenant_id = ctx.tenant_id profile = _get_profile_or_404(tenant_id, profile_id) - updates = request.model_dump(exclude_unset=True) + updates = body.model_dump(exclude_unset=True) profile = profile.model_copy(update={**updates, "updated_at": _now_iso()}) - return _save_profile(profile) + saved = _save_profile(profile) + _audit_integration_event( + ctx=ctx, + event_type_name="DATA_UPDATE", + action="profile_update", + tenant_id=tenant_id, + connector_or_profile_id=profile_id, + resource_type="integration_profile", + changed_fields=sorted(updates.keys()) or ["profile_id"], + ) + return saved @router.post( @@ -433,10 +654,12 @@ async def update_integration_profile( response_model=MappingPreviewResponse, summary="Preview profile mapping against sample events", ) +@limiter.limit(_PROFILE_MUTATION_RATE_LIMIT) async def preview_integration_profile_mapping( + request: Request, tenant_id: str, profile_id: str, - request: MappingPreviewRequest, + body: MappingPreviewRequest, ctx: TenantContext = Depends(require_tenant_context), ) -> MappingPreviewResponse: """Apply a saved profile to sample events without committing records.""" @@ -444,7 +667,7 @@ async def preview_integration_profile_mapping( profile = _get_profile_or_404(tenant_id, profile_id) mapped_events: list[Dict[str, Any]] = [] missing_fields: Dict[str, List[str]] = {} - for index, event in enumerate(request.events): + for index, event in enumerate(body.events): mapped, missing = _apply_profile_mapping(event, profile) mapped_events.append(mapped) if missing: @@ -491,7 +714,9 @@ async def get_integration_statuses( "/configure/{tenant_id}", summary="Configure an integration connector", ) +@limiter.limit(_CONFIGURE_RATE_LIMIT) async def configure_connector( + request: Request, tenant_id: str, config_req: ConnectorConfigRequest, ctx: TenantContext = Depends(require_tenant_context), @@ -584,6 +809,29 @@ async def configure_connector( # takes effect immediately for this tenant. remove_connector(tenant_id, connector_id) connector = get_or_create_connector(tenant_id, connector_id, config) + # Finding 6: audit configure with field NAMES only — never secret values. + changed = [ + name + for name, present in ( + ("api_key", bool(config_req.api_key)), + ("api_secret", bool(config_req.api_secret)), + ("oauth_client_id", bool(config_req.oauth_client_id)), + ("oauth_client_secret", bool(config_req.oauth_client_secret)), + ("webhook_secret", bool(config_req.webhook_secret)), + ("base_url", bool(config_req.base_url)), + ("extra", bool(config_req.extra)), + ) + if present + ] + _audit_integration_event( + ctx=ctx, + event_type_name="CONFIG_CHANGE", + action="configure", + tenant_id=tenant_id, + connector_or_profile_id=connector_id, + resource_type="integration_connector", + changed_fields=changed or ["connector_id"], + ) return { "configured": True, "connector_id": connector_id, @@ -628,7 +876,9 @@ async def test_connection( "/sync/{tenant_id}", summary="Trigger a manual sync", ) +@limiter.limit(_SYNC_RATE_LIMIT) async def trigger_sync( + request: Request, tenant_id: str, sync_req: SyncRequest, ctx: TenantContext = Depends(require_tenant_context), @@ -659,6 +909,15 @@ async def trigger_sync( result = await connector.sync(since=since, limit=sync_req.limit) + _audit_integration_event( + ctx=ctx, + event_type_name="DATA_BULK_OPERATION", + action="sync", + tenant_id=tenant_id, + connector_or_profile_id=connector_id, + resource_type="integration_connector", + changed_fields=["sync"], + ) return { "connector_id": result.connector_id, "events_fetched": result.events_fetched, @@ -674,11 +933,13 @@ async def trigger_sync( "/csv-upload/{tenant_id}", summary="Upload CSV for CTE import", ) +@limiter.limit(_CSV_UPLOAD_RATE_LIMIT) async def upload_csv( + request: Request, tenant_id: str, file: UploadFile = File(...), - source_system: str = Query(default="csv_upload", description="Source ERP system name"), - default_cte_type: str = Query(default="receiving", description="Default CTE type for rows"), + source_system: str = Query(default="csv_upload", description="Source ERP system name", max_length=120), + default_cte_type: str = Query(default="receiving", description="Default CTE type for rows", max_length=64), ctx: TenantContext = Depends(require_tenant_context), ): """Upload a CSV file and import rows as CTE events. @@ -717,9 +978,20 @@ async def upload_csv( connector = CSVSFTPConnector(config) events = connector.parse_csv(csv_text, source_file=safe_filename) - # Convert to ingest format - ingest_events = [e.to_ingest_dict() for e in events] + # Convert to ingest format, then neutralize CSV formula prefixes + # (finding 4). Applied at the HTTP ingest surface so every string + # that leaves parse_csv is sanitized before it reaches callers. + ingest_events = _sanitize_ingest_events([e.to_ingest_dict() for e in events]) + _audit_integration_event( + ctx=ctx, + event_type_name="DATA_CREATE", + action="csv_upload", + tenant_id=tenant_id, + connector_or_profile_id="csv_sftp", + resource_type="integration_csv_upload", + changed_fields=["csv_upload"], + ) return { "filename": safe_filename, "rows_parsed": len(events), @@ -733,7 +1005,7 @@ async def upload_csv( "/webhook/{tenant_id}/{connector_id}", summary="Receive inbound webhook from external system", ) -@limiter.limit("60/minute") +@limiter.limit(_WEBHOOK_RATE_LIMIT) async def receive_webhook( tenant_id: str, connector_id: str, @@ -746,10 +1018,14 @@ async def receive_webhook( Note: This endpoint does NOT require API key auth — it uses the connector's webhook secret for verification instead. - Connectors must fail closed (raise ValueError, mapped to 401) - when their webhook secret is not configured or the signature - is missing or invalid. The response carries only an accepted - count, never the normalized inspection data. + Connectors must fail closed (raise ValueError, mapped to a + uniform 401) when their webhook secret is not configured or + the signature is missing or invalid. Finding 3: the 401 body + never distinguishes missing-secret / bad-digest / replay + (no secret-state oracle), and a repeated signature digest + inside the nonce TTL is rejected as a replay. The response + carries only an accepted count, never the normalized + inspection data. """ from shared.external_connectors.registry import ( get_tenant_connectors, @@ -767,9 +1043,23 @@ async def receive_webhook( payload = await request.body() headers = dict(request.headers) + # Prefer the common HMAC header names; empty string if absent so + # the nonce key is still stable for a given body. + signature_header = ( + headers.get("x-signature") + or headers.get("x-hub-signature-256") + or headers.get("x-safetyculture-signature") + or "" + ) + nonce_key = _webhook_nonce_key(tenant_id, connector_id, payload, signature_header) + if _webhook_nonce_seen_before(nonce_key): + # Replay inside the TTL — same uniform body as any other + # verification failure (finding 3). + raise HTTPException(status_code=401, detail=_WEBHOOK_REJECTED_DETAIL) try: events = await connector.handle_webhook(payload, headers) + _webhook_nonce_remember(nonce_key) logger.info( "webhook_received connector=%s tenant=%s events=%d", connector_id, @@ -785,19 +1075,24 @@ async def receive_webhook( "connector_id": connector_id, } except NotImplementedError: + # NotImplemented is not a secret-state signal; keep 501 so + # clients can disable an unsupported webhook. raise HTTPException( status_code=501, detail=f"Connector '{connector_id}' does not support webhooks", ) - except ValueError as exc: - raise HTTPException(status_code=401, detail=str(exc)) + except ValueError: + # Finding 3: uniform body — never echo str(exc) (oracle). + raise HTTPException(status_code=401, detail=_WEBHOOK_REJECTED_DETAIL) @router.delete( "/disconnect/{tenant_id}/{connector_id}", summary="Disconnect an integration", ) +@limiter.limit(_DISCONNECT_RATE_LIMIT) async def disconnect_connector( + request: Request, tenant_id: str, connector_id: str, ctx: TenantContext = Depends(require_tenant_context), @@ -811,4 +1106,13 @@ async def disconnect_connector( tenant_id = ctx.tenant_id connector_id = resolve_connector_id(connector_id) remove_connector(tenant_id, connector_id) + _audit_integration_event( + ctx=ctx, + event_type_name="CONFIG_CHANGE", + action="disconnect", + tenant_id=tenant_id, + connector_or_profile_id=connector_id, + resource_type="integration_connector", + changed_fields=["disconnect"], + ) return {"disconnected": True, "connector_id": connector_id} diff --git a/services/ingestion/tests/test_integration_hardening.py b/services/ingestion/tests/test_integration_hardening.py new file mode 100644 index 000000000..0dd904fc5 --- /dev/null +++ b/services/ingestion/tests/test_integration_hardening.py @@ -0,0 +1,669 @@ +"""Hardening tests for the integrations API router (findings 2-6). + +Finding 1 (credential store) is deferred — see +``/workspace/devops/router-reviews/integ-cred-options.md``. + +Covers (data-free, no live DB/network): + +- Rate limits on configure / sync / csv-upload / disconnect / profile + mutations and the inbound webhook (429 past the budget). +- Webhook verification failures return a uniform 401 body (no + secret-state oracle); a repeated signature digest inside the nonce + TTL is rejected as a replay. +- CSV upload runs every string leaf through ``sanitize_cell`` so + formula-injection prefixes never leave the ingest surface. +- Request models are bounded (``extra="forbid"`` + max_length / ge/le). +- Successful mutations emit tenant-bound audit events attributed to + ``ctx.principal_id`` / ``principal_kind`` / ``actor_email``, carrying + field NAMES only (never secret values). +- Negative auth: unauthenticated mutations are rejected; a context + pinned to one tenant cannot configure / sync / disconnect another + tenant's connector, and writes land under ``ctx.tenant_id``. + +Synthetic fixtures only; pragma allowlist used solely for fake secrets. +""" + +from __future__ import annotations + +import io +import sys +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi import FastAPI, Request +from fastapi.testclient import TestClient +from pydantic import ValidationError +from shared.rate_limit import limiter +from shared.tenant_context import TenantContext +from slowapi import _rate_limit_exceeded_handler +from slowapi.errors import RateLimitExceeded + +service_dir = Path(__file__).parent.parent +sys.path.insert(0, str(service_dir)) + +from app import integration_router as ir # noqa: E402 +from app.integration_router import ( # noqa: E402 + ConnectorConfigRequest, + CreateIntegrationProfileRequest, + SyncRequest, + UpdateIntegrationProfileRequest, + router, +) +from shared.external_connectors import registry as registry_mod # noqa: E402 +from shared.external_connectors.base import ConnectionStatus, ConnectorConfig # noqa: E402 + +TENANT_A = "tenant-a" +TENANT_B = "tenant-b" +CONNECTOR_ID = "safetyculture" + +# Synthetic only — never a real credential. +FAKE_API_KEY = ( + "sk_test_FAKE_INTEGRATION_DO_NOT_USE" # pragma: allowlist secret +) +FAKE_WEBHOOK_SECRET = ( + "whsec_test_FAKE_DO_NOT_USE_0001" # pragma: allowlist secret +) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +def _build_app(*, pinned_tenant: str | None = TENANT_A, authed: bool = True) -> FastAPI: + app = FastAPI() + app.state.limiter = limiter + app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) + app.include_router(router) + + if authed: + + async def _ctx(request: Request) -> TenantContext: + return TenantContext( + tenant_id=pinned_tenant or TENANT_A, + principal_kind="test", + principal_id="router-test", + actor_email="[email protected]", + ) + + app.dependency_overrides[ir.require_tenant_context] = _ctx + # Also bypass the legacy API-key gate when present. + try: + from app.webhook_compat import _verify_api_key + + app.dependency_overrides[_verify_api_key] = lambda: None + except Exception: # pragma: no cover + pass + return app + + +def _reset_limiter() -> None: + try: + limiter.reset() + except Exception: # pragma: no cover + storage = getattr(limiter, "_storage", None) or getattr( + getattr(limiter, "limiter", None), "storage", None + ) + if storage is not None and hasattr(storage, "reset"): + storage.reset() + + +@pytest.fixture(autouse=True) +def _isolate(monkeypatch: pytest.MonkeyPatch): + _reset_limiter() + ir._reset_webhook_nonces_for_tests() + ir._profile_store.clear() + monkeypatch.setattr(registry_mod, "_CONNECTOR_CLASSES", {}) + monkeypatch.setattr(registry_mod, "_CONNECTOR_ALIASES", {}) + monkeypatch.setattr(registry_mod, "_ACTIVE_CONNECTORS", {}) + monkeypatch.setattr(ir, "get_db_safe", lambda: None) + yield + _reset_limiter() + ir._reset_webhook_nonces_for_tests() + ir._profile_store.clear() + + +class _FakeConnector: + def __init__(self, status: ConnectionStatus = ConnectionStatus.CONNECTED): + self.status = status + self.handle_calls = 0 + + async def test_connection(self): + return True + + async def sync(self, *, since=None, limit=100): + class _SyncResult: + connector_id = CONNECTOR_ID + events_fetched = 1 + events_accepted = 1 + events_rejected = 0 + errors: list = [] + duration_ms = 1 + success = True + + return _SyncResult() + + async def handle_webhook(self, payload: bytes, headers: dict[str, str]): + self.handle_calls += 1 + return [] + + def get_connector_info(self): + return {"category": "food_safety"} + + +def _register_fake(monkeypatch: pytest.MonkeyPatch, connector: _FakeConnector | None = None): + connector = connector or _FakeConnector() + + class _Cls: + def __init__(self, config): + self.config = config + self.status = ConnectionStatus.CONNECTED + + def get_connector_info(self): + return {"category": "food_safety"} + + monkeypatch.setattr( + registry_mod, + "get_connector_class", + lambda cid: _Cls if cid == CONNECTOR_ID else None, + ) + monkeypatch.setattr( + registry_mod, + "resolve_connector_id", + lambda cid: cid, + ) + monkeypatch.setattr( + registry_mod, + "remove_connector", + lambda tid, cid: None, + ) + monkeypatch.setattr( + registry_mod, + "get_or_create_connector", + lambda tid, cid, config: connector + if isinstance(connector, _FakeConnector) + else _FakeConnector(), + ) + monkeypatch.setattr( + registry_mod, + "get_tenant_connectors", + lambda tid: {CONNECTOR_ID: connector} if tid == TENANT_A else {}, + ) + return connector + + +# --------------------------------------------------------------------------- +# Finding 5 — bounded request models +# --------------------------------------------------------------------------- + + +def test_configure_model_forbids_extra_fields() -> None: + with pytest.raises(ValidationError): + ConnectorConfigRequest(connector_id=CONNECTOR_ID, unexpected="x") + + +def test_configure_model_bounds_connector_id() -> None: + with pytest.raises(ValidationError): + ConnectorConfigRequest(connector_id="x" * 121) + + +def test_configure_model_bounds_secret_length() -> None: + with pytest.raises(ValidationError): + ConnectorConfigRequest(connector_id=CONNECTOR_ID, api_key="k" * 4097) + + +def test_sync_model_bounds_limit() -> None: + with pytest.raises(ValidationError): + SyncRequest(connector_id=CONNECTOR_ID, limit=0) + with pytest.raises(ValidationError): + SyncRequest(connector_id=CONNECTOR_ID, limit=1001) + ok = SyncRequest(connector_id=CONNECTOR_ID, limit=1000) + assert ok.limit == 1000 + + +def test_profile_create_forbids_extra() -> None: + with pytest.raises(ValidationError): + CreateIntegrationProfileRequest(display_name="Acme", rogue=True) + + +def test_profile_update_forbids_extra() -> None: + with pytest.raises(ValidationError): + UpdateIntegrationProfileRequest(notes="ok", not_a_field=1) + + +def test_configure_rejects_extra_with_422() -> None: + # Validation is pre-handler; no registry stub required. + client = TestClient(_build_app()) + resp = client.post( + f"/api/v1/integrations/configure/{TENANT_A}", + json={"connector_id": CONNECTOR_ID, "extra_field": "nope"}, + ) + assert resp.status_code == 422 + + +# --------------------------------------------------------------------------- +# Finding 2 — rate limits +# --------------------------------------------------------------------------- + + +def test_configure_rate_limited(monkeypatch: pytest.MonkeyPatch) -> None: + _register_fake(monkeypatch) + client = TestClient(_build_app()) + saw_429 = False + for _ in range(20): + resp = client.post( + f"/api/v1/integrations/configure/{TENANT_A}", + json={"connector_id": CONNECTOR_ID, "api_key": FAKE_API_KEY}, + ) + if resp.status_code == 429: + saw_429 = True + break + assert saw_429 + + +def test_sync_rate_limited(monkeypatch: pytest.MonkeyPatch) -> None: + _register_fake(monkeypatch) + client = TestClient(_build_app()) + saw_429 = False + for _ in range(20): + resp = client.post( + f"/api/v1/integrations/sync/{TENANT_A}", + json={"connector_id": CONNECTOR_ID, "limit": 10}, + ) + if resp.status_code == 429: + saw_429 = True + break + assert saw_429 + + +def test_disconnect_rate_limited(monkeypatch: pytest.MonkeyPatch) -> None: + _register_fake(monkeypatch) + client = TestClient(_build_app()) + saw_429 = False + for _ in range(30): + resp = client.delete( + f"/api/v1/integrations/disconnect/{TENANT_A}/{CONNECTOR_ID}", + ) + if resp.status_code == 429: + saw_429 = True + break + assert saw_429 + + +def test_profile_create_rate_limited(monkeypatch: pytest.MonkeyPatch) -> None: + client = TestClient(_build_app()) + saw_429 = False + for i in range(40): + resp = client.post( + f"/api/v1/integrations/profiles/{TENANT_A}", + json={"display_name": f"Profile {i}"}, + ) + if resp.status_code == 429: + saw_429 = True + break + assert saw_429 + + +def test_webhook_rate_limited(monkeypatch: pytest.MonkeyPatch) -> None: + connector = _register_fake(monkeypatch) + + # Vary the signature each call so the nonce cache does not 401 us first. + class _Varying(_FakeConnector): + def __init__(self): + super().__init__() + self.n = 0 + + async def handle_webhook(self, payload, headers): + self.n += 1 + return [] + + varying = _Varying() + monkeypatch.setattr( + registry_mod, + "get_tenant_connectors", + lambda tid: {CONNECTOR_ID: varying}, + ) + client = TestClient(_build_app()) + saw_429 = False + for i in range(70): + resp = client.post( + f"/api/v1/integrations/webhook/{TENANT_A}/{CONNECTOR_ID}", + content=b'{"n":%d}' % i, + headers={"x-signature": f"sha256=sig{i}"}, + ) + if resp.status_code == 429: + saw_429 = True + break + assert saw_429 + + +# --------------------------------------------------------------------------- +# Finding 3 — webhook oracle + nonce replay +# --------------------------------------------------------------------------- + + +def test_webhook_value_error_uniform_401(monkeypatch: pytest.MonkeyPatch) -> None: + class _Bad(_FakeConnector): + async def handle_webhook(self, payload, headers): + raise ValueError("webhook secret not configured for tenant") + + monkeypatch.setattr( + registry_mod, + "resolve_connector_id", + lambda cid: cid, + ) + monkeypatch.setattr( + registry_mod, + "get_tenant_connectors", + lambda tid: {CONNECTOR_ID: _Bad()}, + ) + client = TestClient(_build_app()) + resp = client.post( + f"/api/v1/integrations/webhook/{TENANT_A}/{CONNECTOR_ID}", + content=b'{"ping":true}', + headers={"x-signature": "sha256=abc"}, + ) + assert resp.status_code == 401 + detail = resp.json()["detail"] + assert detail == { + "error": "webhook_rejected", + "message": "Webhook verification failed", + } + # Oracle check: connector-specific wording must not appear. + dumped = str(detail).lower() + assert "secret" not in dumped + assert "not configured" not in dumped + + +def test_webhook_missing_secret_same_body_as_bad_sig( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class _Missing(_FakeConnector): + async def handle_webhook(self, payload, headers): + raise ValueError("missing webhook secret") + + class _BadSig(_FakeConnector): + async def handle_webhook(self, payload, headers): + raise ValueError("invalid HMAC digest") + + monkeypatch.setattr(registry_mod, "resolve_connector_id", lambda cid: cid) + client = TestClient(_build_app()) + + monkeypatch.setattr( + registry_mod, + "get_tenant_connectors", + lambda tid: {CONNECTOR_ID: _Missing()}, + ) + r1 = client.post( + f"/api/v1/integrations/webhook/{TENANT_A}/{CONNECTOR_ID}", + content=b'{"a":1}', + headers={"x-signature": "sha256=one"}, + ) + monkeypatch.setattr( + registry_mod, + "get_tenant_connectors", + lambda tid: {CONNECTOR_ID: _BadSig()}, + ) + r2 = client.post( + f"/api/v1/integrations/webhook/{TENANT_A}/{CONNECTOR_ID}", + content=b'{"a":2}', + headers={"x-signature": "sha256=two"}, + ) + assert r1.status_code == 401 and r2.status_code == 401 + assert r1.json() == r2.json() + + +def test_webhook_replay_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + connector = _FakeConnector() + monkeypatch.setattr(registry_mod, "resolve_connector_id", lambda cid: cid) + monkeypatch.setattr( + registry_mod, + "get_tenant_connectors", + lambda tid: {CONNECTOR_ID: connector}, + ) + client = TestClient(_build_app()) + headers = {"x-signature": "sha256=replay-me"} + body = b'{"event":"inspection.completed"}' + first = client.post( + f"/api/v1/integrations/webhook/{TENANT_A}/{CONNECTOR_ID}", + content=body, + headers=headers, + ) + assert first.status_code == 200 + assert first.json()["accepted"] == 0 + second = client.post( + f"/api/v1/integrations/webhook/{TENANT_A}/{CONNECTOR_ID}", + content=body, + headers=headers, + ) + assert second.status_code == 401 + assert second.json()["detail"]["error"] == "webhook_rejected" + # Second call must not reach the connector again. + assert connector.handle_calls == 1 + + +# --------------------------------------------------------------------------- +# Finding 4 — sanitize_cell on CSV upload path +# --------------------------------------------------------------------------- + + +def test_csv_upload_sanitizes_formula_cells(monkeypatch: pytest.MonkeyPatch) -> None: + class _CsvConn: + def __init__(self, config): + self.config = config + + def parse_csv(self, csv_text, source_file=None): + class _Evt: + def to_ingest_dict(self): + return { + "product_description": "=CMD|' /C calc'!A0", + "notes": "+1234+0", + "nested": {"lot": "@SUM(A1:A2)"}, + } + + return [_Evt()] + + monkeypatch.setattr( + "shared.external_connectors.csv_sftp.CSVSFTPConnector", + _CsvConn, + ) + client = TestClient(_build_app()) + csv_bytes = b"product_description,notes\nplain,plain\n" + resp = client.post( + f"/api/v1/integrations/csv-upload/{TENANT_A}", + files={"file": ("upload.csv", io.BytesIO(csv_bytes), "text/csv")}, + ) + assert resp.status_code == 200, resp.text + events = resp.json()["events"] + assert len(events) == 1 + # safe_cell prefixes a single quote in front of dangerous prefixes. + assert events[0]["product_description"].startswith("'") + assert events[0]["notes"].startswith("'") + assert events[0]["nested"]["lot"].startswith("'") + + +# --------------------------------------------------------------------------- +# Finding 6 — audit events with signed-in principal +# --------------------------------------------------------------------------- + + +class _FakeAuditLogger: + def __init__(self, sink: list) -> None: + self._sink = sink + + def log(self, **kwargs): + self._sink.append(kwargs) + + async def _noop() -> None: + return None + + return _noop() + + +def _install_fake_audit(monkeypatch: pytest.MonkeyPatch) -> list[dict]: + import shared.audit_logging as audit_mod + + sink: list[dict] = [] + fake = _FakeAuditLogger(sink) + monkeypatch.setattr( + audit_mod.AuditLogger, "get_instance", classmethod(lambda cls: fake) + ) + return sink + + +def test_configure_emits_audit_with_principal( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _register_fake(monkeypatch) + sink = _install_fake_audit(monkeypatch) + client = TestClient(_build_app()) + resp = client.post( + f"/api/v1/integrations/configure/{TENANT_A}", + json={ + "connector_id": CONNECTOR_ID, + "api_key": FAKE_API_KEY, + "webhook_secret": FAKE_WEBHOOK_SECRET, + }, + ) + assert resp.status_code == 200, resp.text + assert len(sink) == 1 + call = sink[0] + actor = call["actor"] + assert actor.actor_id == "router-test" + assert actor.actor_type == "test" + assert actor.email == "[email protected]" + assert actor.tenant_id == TENANT_A + # Field NAMES only — never the synthetic secret values. + details = call["details"] + assert "api_key" in details["changed_fields"] + assert "webhook_secret" in details["changed_fields"] + blob = str(details) + str(call.get("message", "")) + assert FAKE_API_KEY not in blob + assert FAKE_WEBHOOK_SECRET not in blob + + +def test_disconnect_emits_audit(monkeypatch: pytest.MonkeyPatch) -> None: + _register_fake(monkeypatch) + sink = _install_fake_audit(monkeypatch) + client = TestClient(_build_app()) + resp = client.delete( + f"/api/v1/integrations/disconnect/{TENANT_A}/{CONNECTOR_ID}", + ) + assert resp.status_code == 200 + assert len(sink) == 1 + assert sink[0]["action"] == "disconnect" + assert sink[0]["actor"].actor_id == "router-test" + + +def test_profile_create_emits_audit(monkeypatch: pytest.MonkeyPatch) -> None: + sink = _install_fake_audit(monkeypatch) + client = TestClient(_build_app()) + resp = client.post( + f"/api/v1/integrations/profiles/{TENANT_A}", + json={"display_name": "Acme CSV"}, + ) + assert resp.status_code == 200, resp.text + assert len(sink) == 1 + assert sink[0]["action"] == "profile_create" + assert sink[0]["actor"].actor_id == "router-test" + assert "display_name" in sink[0]["details"]["changed_fields"] + + +# --------------------------------------------------------------------------- +# Negative auth + wrong-tenant +# --------------------------------------------------------------------------- + + +def test_unauthenticated_configure_rejected() -> None: + client = TestClient(_build_app(authed=False)) + resp = client.post( + f"/api/v1/integrations/configure/{TENANT_A}", + json={"connector_id": CONNECTOR_ID, "api_key": FAKE_API_KEY}, + ) + assert resp.status_code in (401, 403, 422) + + +def test_unauthenticated_sync_rejected() -> None: + client = TestClient(_build_app(authed=False)) + resp = client.post( + f"/api/v1/integrations/sync/{TENANT_A}", + json={"connector_id": CONNECTOR_ID}, + ) + assert resp.status_code in (401, 403, 422) + + +def test_unauthenticated_disconnect_rejected() -> None: + client = TestClient(_build_app(authed=False)) + resp = client.delete( + f"/api/v1/integrations/disconnect/{TENANT_A}/{CONNECTOR_ID}", + ) + assert resp.status_code in (401, 403, 422) + + +def test_wrong_tenant_cannot_sync_other_tenant_connector( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Pinned ctx.tenant_id=TENANT_B must not see TENANT_A's connectors, + even when the path says TENANT_A.""" + connector = _FakeConnector() + monkeypatch.setattr(registry_mod, "resolve_connector_id", lambda cid: cid) + # Registry only has TENANT_A's connector. + monkeypatch.setattr( + registry_mod, + "get_tenant_connectors", + lambda tid: {CONNECTOR_ID: connector} if tid == TENANT_A else {}, + ) + # Pin context to TENANT_B — path still says TENANT_A. + client = TestClient(_build_app(pinned_tenant=TENANT_B)) + resp = client.post( + f"/api/v1/integrations/sync/{TENANT_A}", + json={"connector_id": CONNECTOR_ID}, + ) + # Handler uses ctx.tenant_id, so lookup is under B → 404. + assert resp.status_code == 404 + + +def test_configure_writes_under_ctx_tenant( + monkeypatch: pytest.MonkeyPatch, +) -> None: + seen: dict[str, Any] = {} + + class _Cls: + def __init__(self, config): + self.config = config + self.status = ConnectionStatus.CONNECTED + + def get_connector_info(self): + return {"category": "food_safety"} + + def _create(tid, cid, config): + seen["tenant_id"] = tid + seen["config_tenant"] = config.tenant_id + c = _FakeConnector() + c.status = ConnectionStatus.CONNECTED + return c + + monkeypatch.setattr(registry_mod, "get_connector_class", lambda cid: _Cls) + monkeypatch.setattr(registry_mod, "resolve_connector_id", lambda cid: cid) + monkeypatch.setattr(registry_mod, "remove_connector", lambda tid, cid: None) + monkeypatch.setattr(registry_mod, "get_or_create_connector", _create) + + # Path claims TENANT_B, but ctx is pinned to TENANT_A. + client = TestClient(_build_app(pinned_tenant=TENANT_A)) + resp = client.post( + f"/api/v1/integrations/configure/{TENANT_B}", + json={"connector_id": CONNECTOR_ID, "api_key": FAKE_API_KEY}, + ) + assert resp.status_code == 200, resp.text + assert seen["tenant_id"] == TENANT_A + assert seen["config_tenant"] == TENANT_A + + +def test_integration_stays_out_of_production_enabled() -> None: + """Regression guard: do not silently enable the held integration router.""" + from server.router_policy import PRODUCTION_ENABLED_ROUTERS, VALID_ROUTER_NAMES + + assert "integration" in VALID_ROUTER_NAMES + assert "integration" not in PRODUCTION_ENABLED_ROUTERS diff --git a/services/ingestion/tests/test_integration_router_1342.py b/services/ingestion/tests/test_integration_router_1342.py index dc94bc5f7..4fdb59225 100644 --- a/services/ingestion/tests/test_integration_router_1342.py +++ b/services/ingestion/tests/test_integration_router_1342.py @@ -35,6 +35,9 @@ import pytest from fastapi import FastAPI, Request from fastapi.testclient import TestClient +from shared.rate_limit import limiter +from slowapi import _rate_limit_exceeded_handler +from slowapi.errors import RateLimitExceeded # Make services/ingestion importable. service_dir = Path(__file__).parent.parent @@ -99,6 +102,10 @@ def to_ingest_dict(self): def app_with_auth_bypass(): """Build a FastAPI app with auth context stubbed for router contract tests.""" app = FastAPI() + # slowapi needs the limiter on app.state + a 429 handler once + # mutation routes carry @limiter.limit (findings 2 hardening). + app.state.limiter = limiter + app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.include_router(integration_router.router) # Dependency-override bypasses the auth gate so other contracts @@ -132,6 +139,19 @@ def _clean_registry(monkeypatch): monkeypatch.setattr(registry_mod, "_CONNECTOR_CLASSES", {}) monkeypatch.setattr(registry_mod, "_CONNECTOR_ALIASES", {}) monkeypatch.setattr(registry_mod, "_ACTIVE_CONNECTORS", {}) + try: + limiter.reset() + except Exception: # pragma: no cover - defensive + pass + integration_router._reset_webhook_nonces_for_tests() + integration_router._profile_store.clear() + yield + try: + limiter.reset() + except Exception: # pragma: no cover - defensive + pass + integration_router._reset_webhook_nonces_for_tests() + integration_router._profile_store.clear() # --------------------------------------------------------------------------- @@ -697,7 +717,14 @@ async def handle_webhook(self, payload, headers): # type: ignore[override] json={"event": "ping"}, ) assert resp.status_code == 401 - assert "invalid signature" in resp.json()["detail"] + # Finding 3: uniform rejection body — never echo the connector's + # ValueError text (secret-state oracle). + detail = resp.json()["detail"] + assert detail == { + "error": "webhook_rejected", + "message": "Webhook verification failed", + } + assert "invalid signature" not in str(detail) def test_webhook_payload_passed_as_raw_bytes(self, client, monkeypatch): """The connector receives the raw request body (not JSON-decoded) @@ -826,6 +853,8 @@ class TestAuthGate: @pytest.fixture def app_with_real_auth(self): app = FastAPI() + app.state.limiter = limiter + app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.include_router(integration_router.router) return app diff --git a/services/ingestion/tests/test_safetyculture_webhook_auth.py b/services/ingestion/tests/test_safetyculture_webhook_auth.py index e13e6821c..8b1ef00cc 100644 --- a/services/ingestion/tests/test_safetyculture_webhook_auth.py +++ b/services/ingestion/tests/test_safetyculture_webhook_auth.py @@ -240,6 +240,11 @@ async def _tenant_ctx(request: Request) -> TenantContext: return TestClient(app) +# The router answers every verification failure with one uniform 401 body, +# so a caller cannot tell a missing secret from a missing signature. +_UNIFORM_REJECTION = {"error": "webhook_rejected", "message": "Webhook verification failed"} + + def _post(client: TestClient, body: bytes, headers: Dict[str, str] | None = None): return client.post( f"/api/v1/integrations/webhook/{TENANT}/{CONNECTOR_ID}", @@ -252,7 +257,7 @@ def test_router_missing_signature_returns_401(client, monkeypatch): monkeypatch.setattr(registry_mod, "get_tenant_connectors", lambda _tid: {CONNECTOR_ID: _connector()}) resp = _post(client, json.dumps(PAYLOAD).encode()) assert resp.status_code == 401 - assert "signature" in resp.json()["detail"].lower() + assert resp.json()["detail"] == _UNIFORM_REJECTION def test_router_bad_signature_returns_401(client, monkeypatch): @@ -268,7 +273,7 @@ def test_router_missing_secret_returns_401(client, monkeypatch): body = json.dumps(PAYLOAD).encode() resp = _post(client, body, {"X-Sc-Signature": _sign(body)}) assert resp.status_code == 401 - assert "not configured" in resp.json()["detail"].lower() + assert resp.json()["detail"] == _UNIFORM_REJECTION def test_router_valid_signature_returns_count_only(client, monkeypatch): @@ -295,7 +300,7 @@ def test_router_nested_shape_missing_signature_returns_401(client, monkeypatch): monkeypatch.setattr(registry_mod, "get_tenant_connectors", lambda _tid: {CONNECTOR_ID: _connector()}) resp = _post(client, json.dumps(NESTED_PAYLOAD).encode()) assert resp.status_code == 401 - assert "signature" in resp.json()["detail"].lower() + assert resp.json()["detail"] == _UNIFORM_REJECTION def test_reconfigure_rotates_webhook_secret(client, monkeypatch):