diff --git a/README.md b/README.md index 4677017..4c7cb37 100644 --- a/README.md +++ b/README.md @@ -280,6 +280,7 @@ raglogs ingest --adapter k8s ./node-logs.tar.gz | `--since` | Window for pull adapters, e.g. `30m`, `1h`, `24h` (default last 1h) | | `--from` / `--to` | Explicit ISO 8601 window bounds (pull adapters) | | `--resume-job` | Prior ingestion job UUID to resume pagination cursors from | +| `--scope` | Isolation scope (CLI default `default`). Service requests require a resolvable scope. | **Loki** @@ -636,6 +637,7 @@ Mint, list, and revoke HTTP API keys. The API bearer token is printed **once** a ```bash raglogs keys create --role query --scope default --name "ci" +raglogs keys create --role query --scope incident:INC-9 --allow-scope-override --name "ci-override" raglogs keys list raglogs keys revoke ``` @@ -643,7 +645,8 @@ raglogs keys revoke | Flag | Description | |---|---| | `--role` | `ingest`, `query`, or `admin` (default `query`) | -| `--scope` | Stored on the key for later tenant isolation (G8). Default `default`. Not used to filter log queries yet. | +| `--scope` | Pin the key to this isolation scope (enforced on every service read/write). Default `default`. Convention: `incident:`, `service:`, `env:`. | +| `--allow-scope-override` | Allow the caller to pass a request `scope` other than the key's pin. Pinned by default. | | `--name` | Optional label | Requires a migrated database (`raglogs init`). See [HTTP API authentication](#http-api-authentication). @@ -945,7 +948,13 @@ A valid key with the wrong role returns **403**: {"error_code": "AUTH_FORBIDDEN", "message": "…"} ``` -Keys are stored argon2-hashed with a short indexed prefix. `scope` defaults to `default` and is stored for later isolation work — this release does **not** filter clusters or logs by scope. Each new key also gets a `whsec_…` webhook signing secret (shown once; `keys list` shows `whsec_****` only). +Keys are stored argon2-hashed with a short indexed prefix. Each key is **pinned** to a `scope` (default `default`) and every service ingest/query is filtered by that scope — including baseline comparison — so one incident's logs cannot contaminate another. Mint with `--allow-scope-override` to let the caller pass a request `scope`. A service request with no resolvable scope returns **400**: + +```json +{"error_code": "SCOPE_REQUIRED", "message": "…"} +``` + +A pinned key that sends a different non-empty scope returns **403** `SCOPE_MISMATCH`. The CLI is scope-optional and defaults to `default` (`raglogs ingest --scope incident:INC-9`, `raglogs explain --scope …`). Each new key also gets a `whsec_…` webhook signing secret (shown once; `keys list` shows `whsec_****` only). Optional OIDC: set `AUTH_MODE=oidc` or `both` and `OIDC_ISSUER`. A JWT (three dotted segments) is validated via JWKS (`iss`, `exp`, and `aud` when `OIDC_AUDIENCE` is set). Role comes from claim `raglogs_role` or `roles`, defaulting to `query`. When `AUTH_MODE=api_key`, JWTs are rejected. @@ -994,7 +1003,7 @@ curl -X POST http://localhost:8000/v1/ingestions/$ID:stop **Backpressure.** When pending worker jobs ≥ `INGEST_QUEUE_MAX` (default 100), `POST /v1/ingestions` and `POST /v1/ingestions/lines` return **429** with `Retry-After` (`INGEST_RETRY_AFTER_SECONDS`, default 5) and body `{"error_code":"INGEST_QUEUE_FULL","message":"..."}`. This is a queue-depth stand-in, not full API/LLM rate limiting. -**Idempotency-Key.** `POST /v1/ingestions` (batch enqueue and tail create; also the deprecated `/ingestions` alias) honors an `Idempotency-Key` header (max 256 characters). A repeat within `INGEST_IDEMPOTENCY_TTL_SECONDS` (default 86400) returns the original **202** job — the same `worker_job_id` for batch, the same `ingestion_job_id` for tail — instead of starting a new one. Empty keys return **400**. GET routes ignore the header. `POST /v1/ingestions/lines` does not use the header; duplicate push/tail lines are handled by content dedup instead. +**Idempotency-Key.** `POST /v1/ingestions` (batch enqueue and tail create; also the deprecated `/ingestions` alias) honors an `Idempotency-Key` header (max 256 characters). A repeat **in the same isolation scope** within `INGEST_IDEMPOTENCY_TTL_SECONDS` (default 86400) returns the original **202** job — the same `worker_job_id` for batch, the same `ingestion_job_id` for tail — instead of starting a new one. Reusing another scope's key returns **409** `IDEMPOTENCY_SCOPE_CONFLICT`. Empty keys return **400**. GET routes ignore the header. `POST /v1/ingestions/lines` does not use the header; duplicate push/tail lines are handled by content dedup instead. ```bash curl -X POST http://localhost:8000/v1/ingestions \ @@ -1003,11 +1012,11 @@ curl -X POST http://localhost:8000/v1/ingestions \ -d '{"paths":["/var/log/app"]}' ``` -**Content dedup.** Every persist path (`ingest_files`, `ingest_from_source` including tail ticks, and push `/lines`) stores `original_line_hash` (SHA-256 of the **raw** line, distinct from the normalized fingerprint) and upserts on `(scope, source_ref, original_line_hash, timestamp)`. Re-reading the same physical lines is a no-op, so cluster counts stay stable across overlapping windows and tail/push retries. Missing `source_ref` is stored as `""` so uniqueness works (Postgres NULLs are distinct). `scope` defaults to `"default"`, or the API key's scope when a principal is present; queries are **not** filtered by scope yet. Duplicate lines are skipped, not errors. +**Content dedup.** Every persist path (`ingest_files`, `ingest_from_source` including tail ticks, and push `/lines`) stores `original_line_hash` (SHA-256 of the **raw** line, distinct from the normalized fingerprint) and upserts on `(scope, source_ref, original_line_hash, timestamp)`. Re-reading the same physical lines is a no-op, so cluster counts stay stable across overlapping windows and tail/push retries. Missing `source_ref` is stored as `""` so uniqueness works (Postgres NULLs are distinct). `scope` defaults to `"default"` (CLI) or is resolved from the API key / request (service). Queries, ingest lists, and baseline comparison are filtered by the same scope. Duplicate lines are skipped, not errors. **Completion callbacks.** Optional `callback_url` on `POST /v1/ingestions` (http or https only; `file:` and empty hosts are rejected). When a **batch** worker job reaches a terminal state (`done` / `failed`), raglogs POSTs an HMAC-SHA256-signed JSON body to that URL. Delivery is fail-open: retries with jittered exponential backoff (`WEBHOOK_MAX_RETRIES`, default 5 extra attempts) on 5xx, 429, and connect errors; 4xx other than 429 are not retried. Failures are logged and **do not** change ingest status — poll `GET /v1/ingestions/jobs/{worker_job_id}` still works. -`job_id` in the payload is the **ingestion_job_id** when ingest created a row; if the worker failed before that, it is the `worker_job_id`. `scope` is the authenticated key's scope, or `"default"` when auth is off. `counts.clusters` is `0` (clustering is not part of ingest). Worker `done` maps to `"succeeded"` (or `"partial"` when `error_count > 0`); `failed` maps to `"failed"`. +`job_id` in the payload is the **ingestion_job_id** when ingest created a row; if the worker failed before that, it is the `worker_job_id`. `scope` is the resolved isolation scope (API key pin, request override when allowed, or `"default"` when auth is off). `counts.clusters` is `0` (clustering is not part of ingest). Worker `done` maps to `"succeeded"` (or `"partial"` when `error_count > 0`); `failed` maps to `"failed"`. ```bash curl -X POST http://localhost:8000/v1/ingestions \ diff --git a/migrations/versions/0008_scope_isolation.py b/migrations/versions/0008_scope_isolation.py new file mode 100644 index 0000000..bea57ae --- /dev/null +++ b/migrations/versions/0008_scope_isolation.py @@ -0,0 +1,53 @@ +"""scope isolation: ingestion_jobs.scope, log_entries indexes, key override flag + +Revision ID: 0008_scope_isolation +Revises: 0007_ingest_idempotency +Create Date: 2026-08-17 00:00:00.000000 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa + +from alembic import op + +revision: str = "0008_scope_isolation" +down_revision: Union[str, None] = "0007_ingest_idempotency" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "ingestion_jobs", + sa.Column("scope", sa.String(255), nullable=False, server_default="default"), + ) + op.create_index( + "ix_log_entries_scope_timestamp", + "log_entries", + ["scope", "timestamp"], + ) + op.create_index( + "ix_log_entries_scope_service_environment_fingerprint", + "log_entries", + ["scope", "service", "environment", "fingerprint"], + ) + op.add_column( + "api_keys", + sa.Column( + "allow_scope_override", + sa.Boolean(), + nullable=False, + server_default=sa.false(), + ), + ) + + +def downgrade() -> None: + op.drop_column("api_keys", "allow_scope_override") + op.drop_index( + "ix_log_entries_scope_service_environment_fingerprint", + table_name="log_entries", + ) + op.drop_index("ix_log_entries_scope_timestamp", table_name="log_entries") + op.drop_column("ingestion_jobs", "scope") diff --git a/src/api/app.py b/src/api/app.py index fc14cb7..3723c7e 100644 --- a/src/api/app.py +++ b/src/api/app.py @@ -3,12 +3,13 @@ from pathlib import Path import re -from fastapi import APIRouter, FastAPI -from fastapi.responses import ORJSONResponse +from fastapi import APIRouter, FastAPI, Request +from fastapi.responses import JSONResponse, ORJSONResponse from fastapi.routing import APIRoute from fastapi.staticfiles import StaticFiles from src.api.auth.middleware import AuthMiddleware +from src.api.auth.scope import ScopeResolutionError, scope_error_response from src.api.deprecation import DeprecationHeaderMiddleware from src.api.routes import ask, clusters, compare_windows, config, explain, health, ingestions, timeline, ui @@ -63,6 +64,13 @@ def generate(route: APIRoute) -> str: app.add_middleware(AuthMiddleware) app.add_middleware(DeprecationHeaderMiddleware) + +@app.exception_handler(ScopeResolutionError) +async def handle_scope_resolution_error( + request: Request, exc: ScopeResolutionError +) -> JSONResponse: + return scope_error_response(exc) + app.include_router(health.router, tags=["health"]) app.include_router(ui.router, tags=["ui"]) diff --git a/src/api/auth/keys.py b/src/api/auth/keys.py index 2146dfe..9599e31 100644 --- a/src/api/auth/keys.py +++ b/src/api/auth/keys.py @@ -32,6 +32,7 @@ class ApiKeyInfo: revoked_at: datetime | None created_at: datetime | None webhook_secret_preview: str | None = None + allow_scope_override: bool = False @dataclass(frozen=True) @@ -46,6 +47,7 @@ class ApiKeyRecord: scope: str revoked_at: datetime | None created_at: datetime | None = None + allow_scope_override: bool = False def generate_api_key() -> str: @@ -127,6 +129,7 @@ def lookup_api_key(token: str) -> ApiKeyRecord | None: scope=row.scope, revoked_at=row.revoked_at, created_at=row.created_at, + allow_scope_override=bool(getattr(row, "allow_scope_override", False)), ) for row in rows ] @@ -138,6 +141,7 @@ def create_api_key( role: str, scope: str = DEFAULT_SCOPE, name: str | None = None, + allow_scope_override: bool = False, ) -> tuple[str, str, ApiKeyInfo]: """Persist a hashed key and return (api_key, webhook_secret, metadata). @@ -158,6 +162,7 @@ def create_api_key( role=role, scope=scope, name=name, + allow_scope_override=allow_scope_override, ) return plaintext, webhook_secret, record @@ -174,6 +179,7 @@ def _key_info(row: Any) -> ApiKeyInfo: webhook_secret_preview=mask_webhook_secret( getattr(row, "webhook_secret", None) ), + allow_scope_override=bool(getattr(row, "allow_scope_override", False)), ) @@ -184,6 +190,7 @@ def _persist_key( role: str, scope: str, name: str | None, + allow_scope_override: bool = False, ) -> ApiKeyInfo: from src.db.models import ApiKey from src.db.session import get_db @@ -195,6 +202,7 @@ def _persist_key( role=role, scope=scope, webhook_secret=webhook_secret, + allow_scope_override=allow_scope_override, ) with get_db() as db: db.add(row) diff --git a/src/api/auth/middleware.py b/src/api/auth/middleware.py index c812d67..7a1d1c3 100644 --- a/src/api/auth/middleware.py +++ b/src/api/auth/middleware.py @@ -24,6 +24,7 @@ class AuthPrincipal: auth_method: str key_id: str | None = None subject: str | None = None + allow_scope_override: bool = False def _auth_error(status_code: int, error_code: str, message: str) -> JSONResponse: @@ -62,6 +63,7 @@ def _authenticate_token(token: str, settings: Any) -> AuthPrincipal | None: scope=principal.scope, auth_method="oidc", subject=principal.subject, + allow_scope_override=False, ) if mode not in ("api_key", "both"): @@ -76,6 +78,7 @@ def _authenticate_token(token: str, settings: Any) -> AuthPrincipal | None: scope=record.scope, auth_method="api_key", key_id=str(record.id), + allow_scope_override=getattr(record, "allow_scope_override", False) is True, ) diff --git a/src/api/auth/roles.py b/src/api/auth/roles.py index bdd8d9b..d0e52b4 100644 --- a/src/api/auth/roles.py +++ b/src/api/auth/roles.py @@ -1,7 +1,7 @@ """Route → allowed role mapping. -`admin` is included in every non-exempt set. Scope is stored on keys for later -G8 isolation; this module does not filter log queries by scope. +`admin` is included in every non-exempt set. Scope isolation (G8) is enforced +in ``src.api.auth.scope`` and query filters, not in this role map. A leading `/v1` or `/v2` (any `/v`) is stripped before matching, so `POST /v1/ingestions` uses the same roles as `POST /ingestions`. diff --git a/src/api/auth/scope.py b/src/api/auth/scope.py new file mode 100644 index 0000000..eca56e9 --- /dev/null +++ b/src/api/auth/scope.py @@ -0,0 +1,137 @@ +"""Resolve the isolation scope for an HTTP request (G8). + +Service requests always end with a non-empty scope, or ``400 SCOPE_REQUIRED``. +Pinned API keys (and OIDC principals) cannot switch to another scope. +""" + +from __future__ import annotations + +from typing import Optional + +from fastapi import Request +from starlette.responses import JSONResponse + +from src.api.auth.middleware import AuthPrincipal +from src.db.models import DEFAULT_LOG_SCOPE + +ERROR_SCOPE_REQUIRED = "SCOPE_REQUIRED" +ERROR_SCOPE_MISMATCH = "SCOPE_MISMATCH" + + +class ScopeResolutionError(Exception): + """Raised when a service request has no resolvable scope, or a pinned mismatch.""" + + def __init__(self, status_code: int, error_code: str, message: str) -> None: + self.status_code = status_code + self.error_code = error_code + self.message = message + super().__init__(message) + + +def scope_error_response(exc: ScopeResolutionError) -> JSONResponse: + return JSONResponse( + status_code=exc.status_code, + content={"error_code": exc.error_code, "message": exc.message}, + ) + + +def normalize_requested_scope(value: Optional[str]) -> Optional[str]: + """Strip whitespace. Empty / None means "not provided".""" + if value is None: + return None + stripped = value.strip() + return stripped or None + + +def resolve_scope( + *, + requested_scope: Optional[str], + auth_enabled: bool, + principal: Optional[AuthPrincipal], +) -> str: + """Return the scope to stamp and filter with. + + Auth off: explicit request scope, else ``default`` (unauthenticated tests). + Pinned key / OIDC: always the principal scope; a different request scope + is ``403 SCOPE_MISMATCH``. Override-allowed keys use the request scope + when present, else the key scope. Empty after resolution → ``400``. + """ + requested = normalize_requested_scope(requested_scope) + + if not auth_enabled: + resolved = requested or DEFAULT_LOG_SCOPE + if not resolved: + raise ScopeResolutionError( + 400, + ERROR_SCOPE_REQUIRED, + "A non-empty scope is required", + ) + return resolved + + if principal is None: + raise ScopeResolutionError( + 400, + ERROR_SCOPE_REQUIRED, + "A resolvable scope is required for this request", + ) + + pinned = bool( + principal.auth_method == "oidc" or not principal.allow_scope_override + ) + principal_scope = normalize_requested_scope(principal.scope) + + if pinned: + if not principal_scope: + raise ScopeResolutionError( + 400, + ERROR_SCOPE_REQUIRED, + "API key has no scope; mint a key with --scope", + ) + if requested is not None and requested != principal_scope: + raise ScopeResolutionError( + 403, + ERROR_SCOPE_MISMATCH, + "Request scope does not match the API key's pinned scope", + ) + return principal_scope + + resolved = requested or principal_scope + if not resolved: + raise ScopeResolutionError( + 400, + ERROR_SCOPE_REQUIRED, + "Pass scope on the request or mint a key with a default scope", + ) + return resolved + + +def requested_scope_from_http( + request: Request, + body_scope: Optional[str] = None, +) -> Optional[str]: + """Prefer a JSON-body ``scope``; otherwise the ``scope`` query parameter.""" + from_body = normalize_requested_scope(body_scope) + if from_body is not None: + return from_body + return normalize_requested_scope(request.query_params.get("scope")) + + +def bind_request_scope( + request: Request, + body_scope: Optional[str] = None, +) -> str: + """Resolve scope, store it on ``request.state``, and return it.""" + from src.config import get_settings + + settings = get_settings() + principal = getattr(request.state, "auth_principal", None) + if principal is not None and not isinstance(principal, AuthPrincipal): + principal = None + + resolved = resolve_scope( + requested_scope=requested_scope_from_http(request, body_scope), + auth_enabled=bool(settings.auth_enabled), + principal=principal, + ) + request.state.resolved_scope = resolved + return resolved diff --git a/src/api/routes/ask.py b/src/api/routes/ask.py index 6aac79e..23726d5 100644 --- a/src/api/routes/ask.py +++ b/src/api/routes/ask.py @@ -2,7 +2,7 @@ from datetime import datetime from typing import Optional -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel from src.api.schemas.v1 import ( @@ -22,6 +22,7 @@ class AskRequest(BaseModel): to_time: Optional[datetime] = None service: Optional[str] = None ingestion_job_id: Optional[str] = None + scope: Optional[str] = None @router.post( @@ -29,11 +30,14 @@ class AskRequest(BaseModel): response_model=AskResponse, response_model_exclude_unset=True, ) -def ask_endpoint(request: AskRequest) -> AskResponse: +def ask_endpoint(request: AskRequest, http_request: Request) -> AskResponse: + from src.api.auth.scope import bind_request_scope from src.core.retrieval.question_router import answer_question from src.db.session import get_db from src.utils.time import resolve_window + scope = bind_request_scope(http_request, request.scope) + window_start, window_end = None, None if request.since or request.from_time: try: @@ -61,6 +65,7 @@ def ask_endpoint(request: AskRequest) -> AskResponse: window_end=window_end, service=request.service, ingestion_job_id=ingestion_job_id, + scope=scope, ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) diff --git a/src/api/routes/clusters.py b/src/api/routes/clusters.py index 09ed55b..eedc75d 100644 --- a/src/api/routes/clusters.py +++ b/src/api/routes/clusters.py @@ -10,7 +10,6 @@ SCHEMA_VERSION, ClustersResponse, llm_rules_only, - scope_from_request, window_from_bounds, ) @@ -25,6 +24,7 @@ class ClustersRequest(BaseModel): env: Optional[str] = None top: int = 15 ingestion_job_id: Optional[str] = None + scope: Optional[str] = None @router.post( @@ -34,10 +34,13 @@ class ClustersRequest(BaseModel): response_model_by_alias=True, ) def clusters_endpoint(request: ClustersRequest, http_request: Request) -> ClustersResponse: + from src.api.auth.scope import bind_request_scope from src.core.clustering.clusterer import run_clustering from src.db.session import get_db from src.utils.time import resolve_window + scope = bind_request_scope(http_request, request.scope) + try: window_start, window_end = resolve_window( since=request.since, @@ -65,13 +68,14 @@ def clusters_endpoint(request: ClustersRequest, http_request: Request) -> Cluste max_clusters=request.top, save_to_db=False, ingestion_job_id=ingestion_job_id, + scope=scope, ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) return ClustersResponse( schema_version=SCHEMA_VERSION, - scope=scope_from_request(http_request), + scope=scope, window=window_from_bounds(window_start, window_end), ingestion_job_id=request.ingestion_job_id, llm=llm_rules_only(), diff --git a/src/api/routes/compare_windows.py b/src/api/routes/compare_windows.py index 70a4959..326c72d 100644 --- a/src/api/routes/compare_windows.py +++ b/src/api/routes/compare_windows.py @@ -26,6 +26,7 @@ class CompareRequest(BaseModel): all_ingestions: bool = False ingestion_job_id: Optional[str] = None format: Literal["json", "text"] = "json" + scope: Optional[str] = None def _ensure_utc(dt: datetime) -> datetime: @@ -178,13 +179,15 @@ def row(sigil: str, msg: str, count_a: Optional[int], count_b: Optional[int]) -> response_model_by_alias=True, ) def compare_endpoint(request: CompareRequest, http_request: Request) -> CompareResponse: - from src.api.schemas.v1 import scope_from_request + from src.api.auth.scope import bind_request_scope from src.core.clustering.clusterer import run_clustering from src.core.compare.differ import compare_windows as run_compare_windows from src.core.explain.evidence import assemble_evidence from src.core.explain.summarizer import get_latest_ingestion_job_id from src.db.session import get_db + scope = bind_request_scope(http_request, request.scope) + try: a_start, a_end, b_start, b_end = _resolve_compare_windows(request) except ValueError as e: @@ -203,7 +206,7 @@ def compare_endpoint(request: CompareRequest, http_request: Request) -> CompareR if ingestion_job_id is not None: job_id = ingestion_job_id elif not request.all_ingestions: - job_id = get_latest_ingestion_job_id(db) + job_id = get_latest_ingestion_job_id(db, scope=scope) _, clusters_a = run_clustering( db=db, @@ -214,6 +217,7 @@ def compare_endpoint(request: CompareRequest, http_request: Request) -> CompareR save_to_db=False, ingestion_job_id=job_id, max_clusters=50, + scope=scope, ) _, clusters_b = run_clustering( db=db, @@ -224,6 +228,7 @@ def compare_endpoint(request: CompareRequest, http_request: Request) -> CompareR save_to_db=False, ingestion_job_id=job_id, max_clusters=50, + scope=scope, ) packet_a = assemble_evidence( @@ -234,6 +239,7 @@ def compare_endpoint(request: CompareRequest, http_request: Request) -> CompareR service_filter=request.service, environment_filter=request.env, ingestion_job_id=job_id, + scope=scope, ) packet_b = assemble_evidence( db=db, @@ -243,6 +249,7 @@ def compare_endpoint(request: CompareRequest, http_request: Request) -> CompareR service_filter=request.service, environment_filter=request.env, ingestion_job_id=job_id, + scope=scope, ) result = run_compare_windows( @@ -259,7 +266,7 @@ def compare_endpoint(request: CompareRequest, http_request: Request) -> CompareR except Exception as e: raise HTTPException(status_code=500, detail=str(e)) - body = _compare_result_to_response(result, scope=scope_from_request(http_request)) + body = _compare_result_to_response(result, scope=scope) if request.format == "text": rendered = format_compare_plain(result) body = body.model_copy(update={"rendered_text": rendered, "text": rendered}) diff --git a/src/api/routes/explain.py b/src/api/routes/explain.py index 51f66d3..094dd05 100644 --- a/src/api/routes/explain.py +++ b/src/api/routes/explain.py @@ -3,7 +3,7 @@ POST /query/explain - optional ingestion_job_id scopes analysis to a specific ingest - - caches results in the explanations table keyed on (window, service, env, ingestion_job_id) + - caches results in the explanations table keyed on (window, service, env, ingestion_job_id, scope) - returns cached result on repeated calls for the same window+filters """ from __future__ import annotations @@ -21,7 +21,6 @@ ExplainResponse, explain_from_cached, explain_from_result, - scope_from_request, ) if TYPE_CHECKING: @@ -42,11 +41,13 @@ class ExplainRequest(BaseModel): ingestion_job_id: Optional[str] = None # scope to a specific ingest force_refresh: bool = False # bypass cache format: Literal["json", "markdown"] = "json" + scope: Optional[str] = None def _cache_key(window_start: datetime, window_end: datetime, service: Optional[str], env: Optional[str], - ingestion_job_id: Optional[str]) -> str: + ingestion_job_id: Optional[str], + scope: str = "default") -> str: """SHA-256 of the canonical filter string — used to detect cache hits.""" canonical = json.dumps({ "ws": window_start.isoformat(), @@ -54,6 +55,7 @@ def _cache_key(window_start: datetime, window_end: datetime, "svc": service, "env": env, "ijid": ingestion_job_id, + "scope": scope, }, sort_keys=True) return hashlib.sha256(canonical.encode()).hexdigest() @@ -183,10 +185,13 @@ def _save_to_cache(db, cache_hash: str, window_start: datetime, window_end: date response_model_by_alias=True, ) def explain_endpoint(request: ExplainRequest, http_request: Request) -> ExplainResponse: + from src.api.auth.scope import bind_request_scope from src.core.explain.summarizer import explain_window from src.db.session import get_db from src.utils.time import resolve_window + scope = bind_request_scope(http_request, request.scope) + try: window_start, window_end = resolve_window( since=request.since, @@ -203,8 +208,9 @@ def explain_endpoint(request: ExplainRequest, http_request: Request) -> ExplainR except ValueError: raise HTTPException(status_code=400, detail="Invalid ingestion_job_id") - cache_hash = _cache_key(window_start, window_end, request.service, request.env, request.ingestion_job_id) - scope = scope_from_request(http_request) + cache_hash = _cache_key( + window_start, window_end, request.service, request.env, request.ingestion_job_id, scope + ) try: with get_db() as db: @@ -234,6 +240,7 @@ def explain_endpoint(request: ExplainRequest, http_request: Request) -> ExplainR max_clusters=request.max_clusters, baseline_window_str=request.baseline_window, ingestion_job_id=ingestion_job_id, + scope=scope, ) body = explain_from_result( diff --git a/src/api/routes/ingestions.py b/src/api/routes/ingestions.py index 849e5d6..78954e2 100644 --- a/src/api/routes/ingestions.py +++ b/src/api/routes/ingestions.py @@ -46,6 +46,7 @@ class IngestRequest(BaseModel): with_embeddings: bool = False mode: Literal["batch", "tail"] = "batch" callback_url: Optional[str] = None + scope: Optional[str] = None @field_validator("callback_url") @classmethod @@ -249,12 +250,16 @@ def _ingest_payload( def _callback_meta_from_request(http_request: Request) -> dict[str, Any]: """Stash scope + api_key_id for HMAC at delivery time (not the bearer token).""" meta: dict[str, Any] = {"scope": "default"} + resolved = getattr(http_request.state, "resolved_scope", None) + if isinstance(resolved, str) and resolved.strip(): + meta["scope"] = resolved.strip() principal = getattr(http_request.state, "auth_principal", None) if principal is None: return meta - scope = getattr(principal, "scope", None) - if scope: - meta["scope"] = scope + if not isinstance(resolved, str) or not resolved.strip(): + scope = getattr(principal, "scope", None) + if scope: + meta["scope"] = scope key_id = getattr(principal, "key_id", None) if key_id: meta["api_key_id"] = key_id @@ -262,9 +267,66 @@ def _callback_meta_from_request(http_request: Request) -> dict[str, Any]: def _scope_from_request(http_request: Request) -> str: + resolved = getattr(http_request.state, "resolved_scope", None) + if isinstance(resolved, str) and resolved.strip(): + return resolved.strip() return str(_callback_meta_from_request(http_request).get("scope") or "default") +def _payload_scope(payload: Any, fallback: str = "default") -> str: + if not isinstance(payload, dict): + return fallback + value = payload.get("scope") + if isinstance(value, str) and value.strip(): + return value.strip() + return fallback + + +def _column_scope(job: Any, fallback: str = "default") -> str: + value = getattr(job, "scope", None) + if isinstance(value, str) and value.strip(): + return value.strip() + return fallback + + +def _scope_of_idempotency_row(db: Any, row: Any) -> str: + """Scope of the job this Idempotency-Key is bound to (payload or column).""" + worker_job_id = getattr(row, "worker_job_id", None) + if worker_job_id is not None: + from src.db.models import WorkerJob + + job = db.query(WorkerJob).filter(WorkerJob.id == worker_job_id).first() + if job is not None: + return _payload_scope(getattr(job, "payload_json", None)) + + ingestion_job_id = getattr(row, "ingestion_job_id", None) + if ingestion_job_id is not None: + from src.db.models import IngestionJob + + job = db.query(IngestionJob).filter(IngestionJob.id == ingestion_job_id).first() + if job is not None: + column = _column_scope(job, fallback="") + if column: + return column + meta = getattr(job, "metadata_json", None) + if isinstance(meta, dict): + return _payload_scope(meta) + return "default" + + +def _idempotency_scope_conflict() -> JSONResponse: + return JSONResponse( + status_code=409, + content={ + "error_code": "IDEMPOTENCY_SCOPE_CONFLICT", + "message": ( + "Idempotency-Key is already bound to a different scope; " + "use a distinct key or wait for the original mapping to expire." + ), + }, + ) + + def _enqueued_from_idempotency(row: Any) -> EnqueuedResponse: mode = getattr(row, "mode", None) or "batch" ingestion_job_id = getattr(row, "ingestion_job_id", None) @@ -284,7 +346,9 @@ def _enqueued_from_idempotency(row: Any) -> EnqueuedResponse: ) -def _replay_if_active(db: Any, key: Optional[str]) -> Optional[EnqueuedResponse]: +def _replay_if_active( + db: Any, key: Optional[str], scope: str +) -> Optional[EnqueuedResponse | JSONResponse]: if not key: return None from datetime import datetime, timezone @@ -294,6 +358,8 @@ def _replay_if_active(db: Any, key: Optional[str]) -> Optional[EnqueuedResponse] row = lookup_active_key(db, key, datetime.now(tz=timezone.utc)) if row is None: return None + if _scope_of_idempotency_row(db, row) != scope: + return _idempotency_scope_conflict() import structlog structlog.get_logger().info( @@ -311,7 +377,8 @@ def _bind_idempotency_key( worker_job_id: Optional[uuid.UUID], ingestion_job_id: Optional[uuid.UUID], mode: str, -) -> Optional[EnqueuedResponse]: + scope: str, +) -> Optional[EnqueuedResponse | JSONResponse]: """Persist the key mapping. Returns a replay response on unique-key race.""" if not key: return None @@ -331,6 +398,8 @@ def _bind_idempotency_key( ttl_seconds=settings.ingest_idempotency_ttl_seconds, ) if is_replay: + if _scope_of_idempotency_row(db, _row) != scope: + return _idempotency_scope_conflict() return _enqueued_from_idempotency(_row) return None @@ -348,6 +417,11 @@ def _create_tail_job( source_name = request.source_name or f"tail:{request.adapter}" source = _get_or_create_source(db, name=source_name, type_=request.adapter) now = datetime.now(tz=timezone.utc) + scope = "default" + if callback_meta: + raw = callback_meta.get("scope") + if isinstance(raw, str) and raw.strip(): + scope = raw.strip() job = IngestionJob( id=uuid.uuid4(), source_id=source.id, @@ -358,6 +432,7 @@ def _create_tail_job( source_ref=source_name, mode="tail", consecutive_errors=0, + scope=scope, ) db.add(job) db.flush() @@ -377,7 +452,10 @@ def _parse_ingestion_job_id(ingestion_job_id: str) -> uuid.UUID: def _apply_tail_action( - ingestion_job_id: str, action: Literal["pause", "resume", "stop"] + ingestion_job_id: str, + action: Literal["pause", "resume", "stop"], + *, + scope: str, ) -> TailLifecycleResponse: from src.core.ingestion.tail import TailLifecycleError, apply_tail_lifecycle from src.db.models import IngestionJob @@ -385,7 +463,11 @@ def _apply_tail_action( ij_uuid = _parse_ingestion_job_id(ingestion_job_id) with get_db() as db: - job = db.query(IngestionJob).filter(IngestionJob.id == ij_uuid).first() + job = ( + db.query(IngestionJob) + .filter(IngestionJob.id == ij_uuid, IngestionJob.scope == scope) + .first() + ) if not job: raise HTTPException(status_code=404, detail="Ingestion job not found") try: @@ -434,6 +516,7 @@ def create_ingestion( Optional ``Idempotency-Key`` (also accepted as ``idempotency-key``): a repeat within the TTL returns the original 202 job rather than a new one. """ + from src.api.auth.scope import bind_request_scope from src.core.ingestion.idempotency import ( InvalidIdempotencyKey, parse_idempotency_key, @@ -441,6 +524,8 @@ def create_ingestion( from src.db.models import WorkerJob from src.db.session import get_db + resolved = bind_request_scope(http_request, request.scope) + try: parsed_key = parse_idempotency_key(idempotency_key) except InvalidIdempotencyKey as e: @@ -461,7 +546,7 @@ def create_ingestion( callback_meta = _callback_meta_from_request(http_request) with get_db() as db: - replayed = _replay_if_active(db, parsed_key) + replayed = _replay_if_active(db, parsed_key, resolved) if replayed is not None: return replayed @@ -481,6 +566,7 @@ def create_ingestion( if response.ingestion_job_id else None, mode="tail", + scope=resolved, ) if race is not None: from src.db.models import IngestionJob @@ -513,6 +599,7 @@ def create_ingestion( worker_job_id=job.id, ingestion_job_id=None, mode="batch", + scope=resolved, ) if race is not None: db.delete(job) @@ -523,21 +610,27 @@ def create_ingestion( @router.get("", response_model=IngestionListResponse) -def list_ingestions(): +def list_ingestions(http_request: Request, scope: Optional[str] = None): """List recent completed ingestion jobs, newest first. Used by the web UI's ingestion picker.""" from sqlalchemy import desc, select + from src.api.auth.scope import bind_request_scope from src.db.models import IngestionJob, Source + from src.db.scope_filter import filter_ingestion_jobs_by_scope from src.db.session import get_db + resolved = bind_request_scope(http_request, scope) + with get_db() as db: - rows = db.execute( + stmt = ( select(IngestionJob, Source.name) .join(Source, Source.id == IngestionJob.source_id) .where(IngestionJob.status == "completed") .order_by(desc(IngestionJob.finished_at)) .limit(25) - ).all() + ) + stmt = filter_ingestion_jobs_by_scope(stmt, resolved) + rows = db.execute(stmt).all() return IngestionListResponse( ingestions=[ @@ -555,14 +648,17 @@ def list_ingestions(): @router.get("/jobs/{worker_job_id}", response_model=WorkerJobStatus) -def get_worker_job_status(worker_job_id: str): +def get_worker_job_status(worker_job_id: str, http_request: Request): """ Poll the status of an enqueued ingest job. When status == 'done', result.ingestion_job_id is ready for /query/explain. """ + from src.api.auth.scope import bind_request_scope from src.db.models import WorkerJob from src.db.session import get_db + resolved = bind_request_scope(http_request, None) + try: wj_uuid = uuid.UUID(worker_job_id) except ValueError: @@ -572,6 +668,8 @@ def get_worker_job_status(worker_job_id: str): job = db.query(WorkerJob).filter(WorkerJob.id == wj_uuid).first() if not job: raise HTTPException(status_code=404, detail="Worker job not found") + if _payload_scope(getattr(job, "payload_json", None)) != resolved: + raise HTTPException(status_code=404, detail="Worker job not found") return WorkerJobStatus( worker_job_id=str(job.id), @@ -588,7 +686,7 @@ def get_worker_job_status(worker_job_id: str): @router.get("/latest", response_model=LatestIngestionResponse) -def get_latest_ingestion(): +def get_latest_ingestion(http_request: Request, scope: Optional[str] = None): """ Return the ID of the most recently completed ingestion job, or null if none exists yet — the same default the CLI and GET /ingestions apply @@ -599,11 +697,14 @@ def get_latest_ingestion(): Registered before /{ingestion_job_id} so "latest" isn't swallowed by that path param. """ + from src.api.auth.scope import bind_request_scope from src.core.explain.summarizer import get_latest_ingestion_job_id from src.db.session import get_db + resolved = bind_request_scope(http_request, scope) + with get_db() as db: - job_id = get_latest_ingestion_job_id(db) + job_id = get_latest_ingestion_job_id(db, scope=resolved) return LatestIngestionResponse(ingestion_job_id=str(job_id) if job_id else None) @@ -625,11 +726,14 @@ def get_latest_ingestion(): ) async def push_ingestion_lines(request: Request) -> PushLinesResponse: """Push NDJSON of raw or pre-parsed log lines. Persisted synchronously.""" + from src.api.auth.scope import bind_request_scope from src.config import get_settings from src.core.ingestion.push import NdjsonParseError, parse_ndjson_payload from src.core.ingestion.service import ingest_push_lines from src.db.session import get_db + bind_request_scope(request, None) + settings = get_settings() body = (await request.body()).decode("utf-8", errors="replace") try: @@ -662,30 +766,45 @@ def push_lines_get_not_allowed() -> None: @router.post("/{ingestion_job_id}:pause", response_model=TailLifecycleResponse) -def pause_tail_job(ingestion_job_id: str) -> TailLifecycleResponse: - return _apply_tail_action(ingestion_job_id, "pause") +def pause_tail_job(ingestion_job_id: str, http_request: Request) -> TailLifecycleResponse: + from src.api.auth.scope import bind_request_scope + + scope = bind_request_scope(http_request, None) + return _apply_tail_action(ingestion_job_id, "pause", scope=scope) @router.post("/{ingestion_job_id}:resume", response_model=TailLifecycleResponse) -def resume_tail_job(ingestion_job_id: str) -> TailLifecycleResponse: - return _apply_tail_action(ingestion_job_id, "resume") +def resume_tail_job(ingestion_job_id: str, http_request: Request) -> TailLifecycleResponse: + from src.api.auth.scope import bind_request_scope + + scope = bind_request_scope(http_request, None) + return _apply_tail_action(ingestion_job_id, "resume", scope=scope) @router.post("/{ingestion_job_id}:stop", response_model=TailLifecycleResponse) -def stop_tail_job(ingestion_job_id: str) -> TailLifecycleResponse: - return _apply_tail_action(ingestion_job_id, "stop") +def stop_tail_job(ingestion_job_id: str, http_request: Request) -> TailLifecycleResponse: + from src.api.auth.scope import bind_request_scope + + scope = bind_request_scope(http_request, None) + return _apply_tail_action(ingestion_job_id, "stop", scope=scope) @router.get("/{ingestion_job_id}", response_model=IngestionJobDetail) -def get_ingestion_detail(ingestion_job_id: str): +def get_ingestion_detail(ingestion_job_id: str, http_request: Request): """Fetch an IngestionJob record by its ID.""" + from src.api.auth.scope import bind_request_scope from src.db.models import IngestionJob from src.db.session import get_db + scope = bind_request_scope(http_request, None) ij_uuid = _parse_ingestion_job_id(ingestion_job_id) with get_db() as db: - job = db.query(IngestionJob).filter(IngestionJob.id == ij_uuid).first() + job = ( + db.query(IngestionJob) + .filter(IngestionJob.id == ij_uuid, IngestionJob.scope == scope) + .first() + ) if not job: raise HTTPException(status_code=404, detail="Ingestion job not found") diff --git a/src/api/routes/timeline.py b/src/api/routes/timeline.py index db2c0b8..910fa5e 100644 --- a/src/api/routes/timeline.py +++ b/src/api/routes/timeline.py @@ -13,7 +13,6 @@ TimelineEventModel, TimelineResponse, llm_rules_only, - scope_from_request, window_from_bounds, ) from src.core.timeline.plain_text import format_timeline_plain @@ -30,6 +29,7 @@ class TimelineRequest(BaseModel): all_ingestions: bool = False ingestion_job_id: Optional[str] = None format: Literal["json", "text"] = "json" + scope: Optional[str] = None def _events_to_models(events) -> list[TimelineEventModel]: @@ -54,6 +54,7 @@ def _events_to_models(events) -> list[TimelineEventModel]: response_model_by_alias=True, ) def timeline_endpoint(request: TimelineRequest, http_request: Request) -> TimelineResponse: + from src.api.auth.scope import bind_request_scope from src.core.clustering.clusterer import run_clustering from src.core.explain.evidence import assemble_evidence from src.core.explain.summarizer import get_latest_ingestion_job_id @@ -61,6 +62,8 @@ def timeline_endpoint(request: TimelineRequest, http_request: Request) -> Timeli from src.db.session import get_db from src.utils.time import resolve_window + scope = bind_request_scope(http_request, request.scope) + try: window_start, window_end = resolve_window( since=request.since, @@ -83,7 +86,7 @@ def timeline_endpoint(request: TimelineRequest, http_request: Request) -> Timeli if ingestion_job_id is not None: job_id = ingestion_job_id elif not request.all_ingestions: - job_id = get_latest_ingestion_job_id(db) + job_id = get_latest_ingestion_job_id(db, scope=scope) _, clusters = run_clustering( db=db, @@ -93,6 +96,7 @@ def timeline_endpoint(request: TimelineRequest, http_request: Request) -> Timeli environment=request.env, save_to_db=False, ingestion_job_id=job_id, + scope=scope, ) packet = assemble_evidence( @@ -103,6 +107,7 @@ def timeline_endpoint(request: TimelineRequest, http_request: Request) -> Timeli service_filter=request.service, environment_filter=request.env, ingestion_job_id=job_id, + scope=scope, ) events = build_timeline(packet) @@ -112,7 +117,7 @@ def timeline_endpoint(request: TimelineRequest, http_request: Request) -> Timeli body = TimelineResponse( schema_version=SCHEMA_VERSION, - scope=scope_from_request(http_request), + scope=scope, window=window_from_bounds(window_start, window_end), events=_events_to_models(events), llm=llm_rules_only(), diff --git a/src/api/schemas/v1.py b/src/api/schemas/v1.py index 647672c..b0f6646 100644 --- a/src/api/schemas/v1.py +++ b/src/api/schemas/v1.py @@ -196,12 +196,12 @@ class ClustersResponse(BaseModel): def scope_from_request(http_request: Optional[Request] = None) -> str: - """Principal scope when auth attached a principal; otherwise ``default``. - - Does not filter logs (G8). Exposed so consumers can pin the response. - """ + """Resolved G8 scope when ``bind_request_scope`` ran; else principal or ``default``.""" if http_request is None: return "default" + resolved = getattr(http_request.state, "resolved_scope", None) + if isinstance(resolved, str) and resolved.strip(): + return resolved.strip() principal = getattr(http_request.state, "auth_principal", None) if principal is None: return "default" diff --git a/src/cli/commands/ask.py b/src/cli/commands/ask.py index b4307a2..3941496 100644 --- a/src/cli/commands/ask.py +++ b/src/cli/commands/ask.py @@ -15,6 +15,11 @@ def ask_cmd( since: Optional[str] = typer.Option(None, "--since", help="Time window e.g. 2h"), service: Optional[str] = typer.Option(None, "--service", help="Filter by service"), fmt: str = typer.Option("text", "--format", help="Output format: text|json"), + scope: str = typer.Option( + "default", + "--scope", + help="Isolation scope (CLI default: default)", + ), ): """Ask a natural language question about your logs.""" if ctx.invoked_subcommand is not None: @@ -46,6 +51,7 @@ def ask_cmd( window_start=window_start, window_end=window_end, service=service, + scope=scope, ) except Exception as e: console.print(f"[red]Error:[/red] {e}") diff --git a/src/cli/commands/clusters.py b/src/cli/commands/clusters.py index b8e8299..8257e4a 100644 --- a/src/cli/commands/clusters.py +++ b/src/cli/commands/clusters.py @@ -17,6 +17,11 @@ def clusters_cmd( fmt: str = typer.Option("text", "--format", help="Output format: text|json"), all_ingestions: bool = typer.Option(False, "--all-ingestions", help="Include all historical ingestion data (not just latest)"), ingestion_job: Optional[str] = typer.Option(None, "--ingestion-job", help="Scope to a specific ingestion job UUID"), + scope: str = typer.Option( + "default", + "--scope", + help="Isolation scope (CLI default: default)", + ), ): """List top log clusters in a time window.""" import uuid @@ -42,7 +47,7 @@ def clusters_cmd( if ingestion_job: job_id = uuid.UUID(ingestion_job) elif not all_ingestions: - job_id = get_latest_ingestion_job_id(db) + job_id = get_latest_ingestion_job_id(db, scope=scope) _, clusters = run_clustering( db=db, window_start=window_start, @@ -52,6 +57,7 @@ def clusters_cmd( max_clusters=top, save_to_db=False, ingestion_job_id=job_id, + scope=scope, ) except Exception as e: console.print(f"[red]Error:[/red] {e}") diff --git a/src/cli/commands/compare.py b/src/cli/commands/compare.py index 8f46f7f..4f78cbf 100644 --- a/src/cli/commands/compare.py +++ b/src/cli/commands/compare.py @@ -24,7 +24,6 @@ import typer from rich.console import Console -from rich.text import Text console = Console() @@ -59,16 +58,21 @@ def compare_cmd( help="Include all historical ingestion data"), fmt: str = typer.Option("text", "--format", help="Output format: text|json"), + scope: str = typer.Option( + "default", + "--scope", + help="Isolation scope (CLI default: default)", + ), ): """Diff two time windows — see exactly what changed.""" - from datetime import datetime, timedelta, timezone + from datetime import datetime, timezone from src.core.clustering.clusterer import run_clustering from src.core.compare.differ import compare_windows from src.core.explain.evidence import assemble_evidence from src.core.explain.summarizer import get_latest_ingestion_job_id from src.db.session import get_db - from src.utils.time import format_window, parse_duration, resolve_window + from src.utils.time import parse_duration now = datetime.now(tz=timezone.utc) @@ -104,20 +108,20 @@ def compare_cmd( with console.status("[cyan]Comparing windows...[/cyan]"): try: with get_db() as db: - job_id = None if all_ingestions else get_latest_ingestion_job_id(db) + job_id = None if all_ingestions else get_latest_ingestion_job_id(db, scope=scope) # Cluster both windows independently (no baseline comparison needed) _, clusters_a = run_clustering( db=db, window_start=a_start, window_end=a_end, service=service, environment=env, save_to_db=False, ingestion_job_id=job_id, - max_clusters=50, + max_clusters=50, scope=scope, ) _, clusters_b = run_clustering( db=db, window_start=b_start, window_end=b_end, service=service, environment=env, save_to_db=False, ingestion_job_id=job_id, - max_clusters=50, + max_clusters=50, scope=scope, ) # Get trigger candidates from evidence assembly for both windows @@ -125,11 +129,13 @@ def compare_cmd( db=db, window_start=a_start, window_end=a_end, clusters=clusters_a, service_filter=service, environment_filter=env, ingestion_job_id=job_id, + scope=scope, ) packet_b = assemble_evidence( db=db, window_start=b_start, window_end=b_end, clusters=clusters_b, service_filter=service, environment_filter=env, ingestion_job_id=job_id, + scope=scope, ) result = compare_windows( diff --git a/src/cli/commands/explain.py b/src/cli/commands/explain.py index 02f158b..8e419db 100644 --- a/src/cli/commands/explain.py +++ b/src/cli/commands/explain.py @@ -19,6 +19,11 @@ def explain_cmd( no_llm: bool = typer.Option(False, "--no-llm", help="Skip LLM, use deterministic templates"), all_ingestions: bool = typer.Option(False, "--all-ingestions", help="Include all historical ingestion data (not just latest)"), ingestion_job: Optional[str] = typer.Option(None, "--ingestion-job", help="Scope analysis to a specific ingestion job UUID"), + scope: str = typer.Option( + "default", + "--scope", + help="Isolation scope (CLI default: default)", + ), ): """Explain what happened in a time window. This is the core command.""" import uuid @@ -49,7 +54,7 @@ def explain_cmd( job_id = uuid.UUID(ingestion_job) elif not all_ingestions: # Default: scope to latest ingestion job to avoid count inflation - job_id = get_latest_ingestion_job_id(db) + job_id = get_latest_ingestion_job_id(db, scope=scope) if job_id: ui.print(f"[dim]Scoped to latest ingestion: {job_id}[/dim]") result = explain_window( @@ -62,6 +67,7 @@ def explain_cmd( max_clusters=max_clusters, baseline_window_str=baseline_window, ingestion_job_id=job_id, + scope=scope, ) except Exception as e: ui.print(f"[red]Error:[/red] {e}") diff --git a/src/cli/commands/ingest.py b/src/cli/commands/ingest.py index 9797941..b76970b 100644 --- a/src/cli/commands/ingest.py +++ b/src/cli/commands/ingest.py @@ -36,6 +36,11 @@ def ingest_cmd( from_time: Optional[str] = typer.Option(None, "--from", help="Window start, ISO 8601 (non-file adapters)"), to_time: Optional[str] = typer.Option(None, "--to", help="Window end, ISO 8601 (non-file adapters)"), resume_job: Optional[str] = typer.Option(None, "--resume-job", help="Prior ingestion job UUID to resume cursors from (non-file adapters)"), + scope: str = typer.Option( + "default", + "--scope", + help="Isolation scope (CLI default: default). Convention: incident:, service:, env:", + ), ): """Ingest logs into the database.""" from src.db.session import get_db @@ -84,6 +89,7 @@ def on_progress(lines_read: int, parsed: int): fmt=fmt, progress_callback=on_progress, with_embeddings=with_embeddings, + scope=scope, ) else: import uuid @@ -107,6 +113,15 @@ def on_progress(lines_read: int, parsed: int): prior = db.query(IngestionJob).filter(IngestionJob.id == uuid.UUID(resume_job)).first() if prior is None: raise ValueError(f"No ingestion job found with id {resume_job}") + prior_scope = getattr(prior, "scope", None) + if ( + isinstance(prior_scope, str) + and prior_scope.strip() + and prior_scope.strip() != scope + ): + raise ValueError( + f"Ingestion job {resume_job} is in a different scope" + ) resume_cursors = (prior.metadata_json or {}).get("cursors") resume_completed_streams = (prior.metadata_json or {}).get("completed_streams") @@ -132,6 +147,7 @@ def on_progress(lines_read: int, parsed: int): resume_completed_streams=resume_completed_streams, progress_callback=on_progress, with_embeddings=with_embeddings, + scope=scope, ) job_id = str(job.id) # read inside session before it closes except ValueError as e: diff --git a/src/cli/commands/keys.py b/src/cli/commands/keys.py index 4a049f3..053afc1 100644 --- a/src/cli/commands/keys.py +++ b/src/cli/commands/keys.py @@ -29,7 +29,12 @@ def create_cmd( scope: str = typer.Option( "default", "--scope", - help="Key scope (stored for later isolation; unused today)", + help="Pin this key to a scope (enforced on every service read/write). Convention: incident:, service:, env:", + ), + allow_scope_override: bool = typer.Option( + False, + "--allow-scope-override", + help="Allow the caller to pass a request scope other than the key's pinned scope", ), name: Optional[str] = typer.Option( None, "--name", help="Optional label for this key" @@ -45,7 +50,10 @@ def create_cmd( try: plaintext, webhook_secret, info = create_api_key( - role=role, scope=scope, name=name + role=role, + scope=scope, + name=name, + allow_scope_override=allow_scope_override, ) except Exception as exc: console.print(f"[red]Error:[/red] {exc}") @@ -66,7 +74,8 @@ def create_cmd( ) ) console.print( - f"[dim]id={info.id} prefix={info.key_prefix} role={info.role} scope={info.scope}[/dim]" + f"[dim]id={info.id} prefix={info.key_prefix} role={info.role} " + f"scope={info.scope} allow_scope_override={info.allow_scope_override}[/dim]" ) console.print( "[dim]HMAC ingest callbacks with the webhook secret (whsec_…), not the API key.[/dim]" @@ -89,6 +98,7 @@ def list_cmd() -> None: table.add_column("prefix") table.add_column("role") table.add_column("scope") + table.add_column("override") table.add_column("webhook") table.add_column("name") table.add_column("created") @@ -102,6 +112,7 @@ def list_cmd() -> None: key.key_prefix, key.role, key.scope, + "yes" if key.allow_scope_override else "no", key.webhook_secret_preview or "", key.name or "", created, diff --git a/src/cli/commands/timeline.py b/src/cli/commands/timeline.py index 5322563..7b9aa0f 100644 --- a/src/cli/commands/timeline.py +++ b/src/cli/commands/timeline.py @@ -24,9 +24,6 @@ import typer from rich.console import Console -from rich.text import Text -from rich import box -from rich.table import Table console = Console() @@ -50,6 +47,11 @@ def timeline_cmd( all_ingestions: bool = typer.Option(False, "--all-ingestions", help="Include all ingestion data"), ingestion_job: Optional[str] = typer.Option(None, "--ingestion-job", help="Scope to specific ingestion job UUID"), fmt: str = typer.Option("text", "--format", help="Output format: text|json"), + scope: str = typer.Option( + "default", + "--scope", + help="Isolation scope (CLI default: default)", + ), ): """Reconstruct the sequence of events in an incident window.""" import uuid @@ -60,7 +62,7 @@ def timeline_cmd( from src.core.explain.summarizer import get_latest_ingestion_job_id from src.core.timeline.builder import build_timeline from src.db.session import get_db - from src.utils.time import format_window, resolve_window + from src.utils.time import resolve_window try: from_dt = datetime.fromisoformat(from_time) if from_time else None @@ -77,7 +79,7 @@ def timeline_cmd( if ingestion_job: job_id = uuid.UUID(ingestion_job) elif not all_ingestions: - job_id = get_latest_ingestion_job_id(db) + job_id = get_latest_ingestion_job_id(db, scope=scope) _, clusters = run_clustering( db=db, @@ -87,6 +89,7 @@ def timeline_cmd( environment=env, save_to_db=False, ingestion_job_id=job_id, + scope=scope, ) packet = assemble_evidence( @@ -97,6 +100,7 @@ def timeline_cmd( service_filter=service, environment_filter=env, ingestion_job_id=job_id, + scope=scope, ) events = build_timeline(packet) diff --git a/src/config/settings.py b/src/config/settings.py index ad365a0..edb86bb 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -37,8 +37,8 @@ class Settings(BaseSettings): # HTTP API authentication (G2). Default off so local demo and existing # TestClient tests stay unauthenticated. Production/Docker should set - # AUTH_ENABLED=true. Scope on keys is stored for later G8 isolation; - # queries are not filtered by scope yet. + # AUTH_ENABLED=true. Keys are pinned to a scope (G8) unless minted with + # allow_scope_override; every service read/write is filtered by scope. auth_enabled: bool = False auth_mode: Literal["api_key", "oidc", "both"] = "api_key" oidc_issuer: str = "" diff --git a/src/core/clustering/baseline.py b/src/core/clustering/baseline.py index 751cedd..4b3d7b5 100644 --- a/src/core/clustering/baseline.py +++ b/src/core/clustering/baseline.py @@ -5,7 +5,11 @@ from sqlalchemy import func, select from sqlalchemy.orm import Session -from src.db.models import LogEntry, IngestionJob +from src.db.models import DEFAULT_LOG_SCOPE, IngestionJob, LogEntry +from src.db.scope_filter import ( + filter_ingestion_jobs_by_scope, + filter_log_entries_by_scope, +) def get_baseline_counts( @@ -15,6 +19,7 @@ def get_baseline_counts( service: Optional[str] = None, environment: Optional[str] = None, exclude_ingestion_job_id: Optional[uuid.UUID] = None, + scope: str = DEFAULT_LOG_SCOPE, ) -> dict[str, int]: """ Return a dict of {fingerprint: count} for the baseline window. @@ -31,6 +36,7 @@ def get_baseline_counts( LogEntry.fingerprint.isnot(None), ) ) + q = filter_log_entries_by_scope(q, scope) if service: q = q.where(LogEntry.service == service) @@ -45,11 +51,11 @@ def get_baseline_counts( if current_job and current_job.started_at: # Exclude all jobs that started at or after this job (including itself) - later_jobs = db.execute( - select(IngestionJob.id).where( - IngestionJob.started_at >= current_job.started_at - ) - ).scalars().all() + later_stmt = select(IngestionJob.id).where( + IngestionJob.started_at >= current_job.started_at + ) + later_stmt = filter_ingestion_jobs_by_scope(later_stmt, scope) + later_jobs = db.execute(later_stmt).scalars().all() if later_jobs: q = q.where(LogEntry.ingestion_job_id.notin_(later_jobs)) else: diff --git a/src/core/clustering/clusterer.py b/src/core/clustering/clusterer.py index 4125523..efae84e 100644 --- a/src/core/clustering/clusterer.py +++ b/src/core/clustering/clusterer.py @@ -11,7 +11,8 @@ from src.core.clustering.scoring import compute_importance_score from src.core.embeddings.provider import EmbeddingsProvider from src.core.normalization.patterns import is_trigger_message -from src.db.models import Cluster, ClusterMember, ClusterRun, LogEntry +from src.db.models import DEFAULT_LOG_SCOPE, Cluster, ClusterMember, ClusterRun, LogEntry +from src.db.scope_filter import filter_log_entries_by_scope from src.utils.time import resolve_baseline_window @@ -42,6 +43,7 @@ def run_clustering( max_clusters: int = 50, save_to_db: bool = True, ingestion_job_id: Optional[uuid.UUID] = None, + scope: str = DEFAULT_LOG_SCOPE, ) -> tuple["ClusterRun", list[ClusterData]]: """ Main clustering pipeline for a time window. @@ -60,6 +62,7 @@ def run_clustering( LogEntry.timestamp <= window_end, LogEntry.fingerprint.isnot(None), ) + q = filter_log_entries_by_scope(q, scope) if service: q = q.where(LogEntry.service == service) @@ -104,7 +107,14 @@ def run_clustering( baseline_counts: dict[str, int] = {} else: baseline_start, baseline_end = resolve_baseline_window(window_start, window_end, baseline_window_str) - baseline_counts = get_baseline_counts(db, baseline_start, baseline_end, service=service, environment=environment) + baseline_counts = get_baseline_counts( + db, + baseline_start, + baseline_end, + service=service, + environment=environment, + scope=scope, + ) # 4. Build cluster data clusters: list[ClusterData] = [] diff --git a/src/core/explain/evidence.py b/src/core/explain/evidence.py index 5f1d112..cc23a3c 100644 --- a/src/core/explain/evidence.py +++ b/src/core/explain/evidence.py @@ -1,15 +1,15 @@ import uuid -from dataclasses import dataclass, field +from dataclasses import dataclass from datetime import datetime, timezone -from typing import Optional, Union +from typing import Optional from sqlalchemy import select from sqlalchemy.orm import Session from src.core.clustering.clusterer import ClusterData from src.core.normalization.patterns import is_trigger_message -from src.db.models import LogEntry -from src.utils.time import format_window +from src.db.models import DEFAULT_LOG_SCOPE, LogEntry +from src.db.scope_filter import filter_log_entries_by_scope def _trunc(text: str, max_len: int) -> str: @@ -51,6 +51,7 @@ def find_trigger_candidates( window_end: datetime, lookback_minutes: int = 10, ingestion_job_id: Optional[uuid.UUID] = None, + scope: str = DEFAULT_LOG_SCOPE, ) -> list[TriggerCandidate]: """ Find likely trigger events in a window slightly before the main window. @@ -63,6 +64,7 @@ def find_trigger_candidates( LogEntry.timestamp >= search_start, LogEntry.timestamp <= window_end, ) + q = filter_log_entries_by_scope(q, scope) if ingestion_job_id: q = q.where(LogEntry.ingestion_job_id == ingestion_job_id) q = q.limit(5000) @@ -99,12 +101,20 @@ def find_trigger_candidates( return deduped[:3] -def count_logs_in_window(db: Session, window_start: datetime, window_end: datetime, service: Optional[str] = None, ingestion_job_id: Optional[uuid.UUID] = None) -> int: +def count_logs_in_window( + db: Session, + window_start: datetime, + window_end: datetime, + service: Optional[str] = None, + ingestion_job_id: Optional[uuid.UUID] = None, + scope: str = DEFAULT_LOG_SCOPE, +) -> int: from sqlalchemy import func q = select(func.count(LogEntry.id)).where( LogEntry.timestamp >= window_start, LogEntry.timestamp <= window_end, ) + q = filter_log_entries_by_scope(q, scope) if service: q = q.where(LogEntry.service == service) if ingestion_job_id: @@ -122,16 +132,24 @@ def assemble_evidence( environment_filter: Optional[str] = None, max_evidence_items: int = 8, ingestion_job_id: Optional[uuid.UUID] = None, + scope: str = DEFAULT_LOG_SCOPE, ) -> EvidencePacket: """ Assemble an evidence packet from clusters and window data. """ - total_logs = count_logs_in_window(db, window_start, window_end, service=service_filter, ingestion_job_id=ingestion_job_id) + total_logs = count_logs_in_window( + db, + window_start, + window_end, + service=service_filter, + ingestion_job_id=ingestion_job_id, + scope=scope, + ) # Error/warn clusters only for primary analysis significant_clusters = [ c for c in clusters - if any(l in ("error", "fatal", "warn", "critical") for l in c.levels) + if any(lvl in ("error", "fatal", "warn", "critical") for lvl in c.levels) ] if not significant_clusters: @@ -143,7 +161,14 @@ def assemble_evidence( secondary = sorted(significant_clusters[1:], key=lambda c: c.count, reverse=True)[:4] if len(significant_clusters) > 1 else [] # Trigger candidates (look back up to 10 min before window start) - triggers = find_trigger_candidates(db, window_start, window_end, lookback_minutes=10, ingestion_job_id=ingestion_job_id) + triggers = find_trigger_candidates( + db, + window_start, + window_end, + lookback_minutes=10, + ingestion_job_id=ingestion_job_id, + scope=scope, + ) # Collect affected services services_set: set[str] = set() diff --git a/src/core/explain/summarizer.py b/src/core/explain/summarizer.py index 1686049..07680fd 100644 --- a/src/core/explain/summarizer.py +++ b/src/core/explain/summarizer.py @@ -7,12 +7,12 @@ from src.config import get_settings from src.core.clustering.clusterer import run_clustering -from src.db.models import IngestionJob from src.core.explain.confidence import compute_confidence from src.core.explain.evidence import EvidencePacket, assemble_evidence from src.core.explain.templates import render_insufficient_evidence, render_text_summary from src.core.llm.provider import NoopLLMProvider, build_llm_provider -from src.utils.time import resolve_baseline_window +from src.db.models import DEFAULT_LOG_SCOPE, IngestionJob +from src.db.scope_filter import filter_ingestion_jobs_by_scope @dataclass @@ -30,12 +30,21 @@ class ExplainResult: mode: str = "rules" -def get_latest_ingestion_job_id(db: Session) -> Optional[uuid.UUID]: - """Return the ID of the most recently completed ingestion job.""" - from sqlalchemy import select, desc - job = db.execute( - select(IngestionJob).where(IngestionJob.status == "completed").order_by(desc(IngestionJob.finished_at)).limit(1) - ).scalar_one_or_none() +def get_latest_ingestion_job_id( + db: Session, + scope: str = DEFAULT_LOG_SCOPE, +) -> Optional[uuid.UUID]: + """Return the ID of the most recently completed ingestion job in ``scope``.""" + from sqlalchemy import desc, select + + stmt = ( + select(IngestionJob) + .where(IngestionJob.status == "completed") + .order_by(desc(IngestionJob.finished_at)) + .limit(1) + ) + stmt = filter_ingestion_jobs_by_scope(stmt, scope) + job = db.execute(stmt).scalar_one_or_none() return job.id if job else None @@ -49,6 +58,7 @@ def explain_window( max_clusters: int = 10, baseline_window_str: Optional[str] = None, ingestion_job_id: Optional[uuid.UUID] = None, + scope: str = DEFAULT_LOG_SCOPE, ) -> ExplainResult: """ Full explain pipeline for a time window. @@ -67,6 +77,7 @@ def explain_window( max_clusters=max_clusters, save_to_db=True, ingestion_job_id=ingestion_job_id, + scope=scope, ) # 2. Assemble evidence @@ -79,6 +90,7 @@ def explain_window( environment_filter=environment, max_evidence_items=settings.max_evidence_items, ingestion_job_id=ingestion_job_id, + scope=scope, ) # 3. Confidence @@ -110,7 +122,7 @@ def explain_window( if llm_text: summary_text = llm_text mode = "llm" - except Exception as e: + except Exception: # Degrade gracefully pass diff --git a/src/core/ingestion/service.py b/src/core/ingestion/service.py index a1d1fb2..e684bc7 100644 --- a/src/core/ingestion/service.py +++ b/src/core/ingestion/service.py @@ -277,6 +277,7 @@ def ingest_files( source_adapter="file", source_ref=", ".join(str(p) for p in files[:5]), mode="batch", + scope=scope or DEFAULT_LOG_SCOPE, ) db.add(job) db.flush() @@ -449,6 +450,7 @@ def ingest_from_source( source_adapter=spec.adapter, source_ref=", ".join(r.stream_id for r in refs[:5]), mode="batch", + scope=scope or DEFAULT_LOG_SCOPE, ) db.add(job) db.flush() @@ -569,6 +571,7 @@ def ingest_push_lines( source_adapter="push", source_ref="push", mode="push", + scope=scope or DEFAULT_LOG_SCOPE, ) db.add(job) db.flush() diff --git a/src/core/ingestion/tail.py b/src/core/ingestion/tail.py index 3a7ece4..91341c5 100644 --- a/src/core/ingestion/tail.py +++ b/src/core/ingestion/tail.py @@ -226,6 +226,12 @@ def tick_one_tail_job( source_name = meta.get("source_name") with_embeddings = bool(meta.get("with_embeddings") or False) + column_scope = getattr(job, "scope", None) + if isinstance(column_scope, str) and column_scope.strip(): + tick_scope = column_scope.strip() + else: + tick_scope = str(meta.get("scope") or "default") + try: _, _stats = ingest_from_source( db=db, @@ -238,7 +244,7 @@ def tick_one_tail_job( with_embeddings=with_embeddings, existing_job=job, finalize=False, - scope=str(meta.get("scope") or "default"), + scope=tick_scope, ) except Exception as exc: new_count, should_pause = consecutive_errors_after_failure( diff --git a/src/core/retrieval/question_router.py b/src/core/retrieval/question_router.py index b7097d8..f05e793 100644 --- a/src/core/retrieval/question_router.py +++ b/src/core/retrieval/question_router.py @@ -28,7 +28,8 @@ from src.config.settings import Settings from src.core.embeddings.provider import get_embeddings_provider from src.core.embeddings.store import STORED_EMBEDDING_DIMS -from src.db.models import LogEmbedding, LogEntry +from src.db.models import DEFAULT_LOG_SCOPE, LogEmbedding, LogEntry +from src.db.scope_filter import filter_log_entries_by_scope log = structlog.get_logger() @@ -89,8 +90,10 @@ def _filter_log_entries( service: Optional[str], level_bias: Optional[str], ingestion_job_id: Optional[uuid.UUID], + scope: str = DEFAULT_LOG_SCOPE, ) -> Select: - """Apply the shared window / service / level / job filters used by ask search.""" + """Apply the shared window / service / level / job / scope filters used by ask search.""" + q = filter_log_entries_by_scope(q, scope) if window_start: q = q.where(LogEntry.timestamp >= window_start) if window_end: @@ -113,6 +116,7 @@ def search_logs( level_bias: Optional[str], limit: int = 500, ingestion_job_id: Optional[uuid.UUID] = None, + scope: str = DEFAULT_LOG_SCOPE, ) -> list[LogEntry]: q = select(LogEntry) @@ -120,7 +124,9 @@ def search_logs( conditions = [LogEntry.normalized_message.ilike(f"%{kw}%") for kw in keywords[:6]] q = q.where(or_(*conditions)) - q = _filter_log_entries(q, window_start, window_end, service, level_bias, ingestion_job_id) + q = _filter_log_entries( + q, window_start, window_end, service, level_bias, ingestion_job_id, scope=scope + ) q = q.order_by(LogEntry.timestamp.desc()).limit(limit) return list(db.execute(q).scalars().all()) @@ -135,6 +141,7 @@ def search_logs_semantic( limit: int = 100, ingestion_job_id: Optional[uuid.UUID] = None, min_similarity: float = 0.75, + scope: str = DEFAULT_LOG_SCOPE, ) -> list[LogEntry]: """Nearest-neighbor log lines via pgvector cosine similarity. @@ -151,7 +158,9 @@ def search_logs_semantic( .join(LogEmbedding, LogEmbedding.log_entry_id == LogEntry.id) .where(similarity >= min_similarity) ) - q = _filter_log_entries(q, window_start, window_end, service, level_bias, ingestion_job_id) + q = _filter_log_entries( + q, window_start, window_end, service, level_bias, ingestion_job_id, scope=scope + ) q = q.order_by(distance.asc()).limit(limit) return list(db.execute(q).scalars().all()) @@ -186,6 +195,7 @@ def fetch_fallback_clusters( service: Optional[str], level_bias: Optional[str], ingestion_job_id: Optional[uuid.UUID] = None, + scope: str = DEFAULT_LOG_SCOPE, ) -> list[LogEntry]: """ Fallback: fetch the most significant error/warn logs from the window @@ -197,6 +207,7 @@ def fetch_fallback_clusters( .where(LogEntry.timestamp >= window_start, LogEntry.timestamp <= window_end) .where(LogEntry.level.in_(["error", "fatal", "warn", "critical"])) ) + q = filter_log_entries_by_scope(q, scope) if service: q = q.where(LogEntry.service == service) if ingestion_job_id: @@ -235,6 +246,7 @@ def answer_question( window_end: Optional[datetime] = None, service: Optional[str] = None, ingestion_job_id: Optional[uuid.UUID] = None, + scope: str = DEFAULT_LOG_SCOPE, ) -> AskResult: from src.config import get_settings from src.core.llm.provider import NoopLLMProvider, build_llm_provider @@ -258,6 +270,7 @@ def answer_question( level_bias=level_bias, ingestion_job_id=ingestion_job_id, settings=settings, + scope=scope, ) if not matching: @@ -333,6 +346,7 @@ def _retrieve_matching_logs( level_bias: Optional[str], ingestion_job_id: Optional[uuid.UUID], settings: Settings, + scope: str = DEFAULT_LOG_SCOPE, ) -> tuple[list[LogEntry], str]: """Semantic-first retrieval, then keyword, then error/warn fallback. @@ -356,6 +370,7 @@ def _retrieve_matching_logs( limit=settings.ask_semantic_top_k, ingestion_job_id=ingestion_job_id, min_similarity=settings.ask_semantic_min_similarity, + scope=scope, ) if semantic_hits: return semantic_hits, "semantic" @@ -365,6 +380,7 @@ def _retrieve_matching_logs( matching = search_logs( db, keywords, window_start, window_end, service, level_bias, ingestion_job_id=ingestion_job_id, + scope=scope, ) if matching: return matching, "keyword" @@ -372,6 +388,7 @@ def _retrieve_matching_logs( matching = fetch_fallback_clusters( db, window_start, window_end, service, level_bias, ingestion_job_id=ingestion_job_id, + scope=scope, ) return matching, "fallback" diff --git a/src/db/models.py b/src/db/models.py index 2a2e1d1..daa1c14 100644 --- a/src/db/models.py +++ b/src/db/models.py @@ -4,6 +4,7 @@ from pgvector.sqlalchemy import Vector from sqlalchemy import ( + Boolean, DateTime, Float, ForeignKey, @@ -61,6 +62,9 @@ class IngestionJob(Base): cursor: Mapped[str | None] = mapped_column(Text, nullable=True) last_polled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) consecutive_errors: Mapped[int] = mapped_column(Integer, default=0) + scope: Mapped[str] = mapped_column( + String(255), nullable=False, default=DEFAULT_LOG_SCOPE, server_default="default" + ) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) source: Mapped["Source"] = relationship("Source", back_populates="jobs") @@ -101,6 +105,14 @@ class LogEntry(Base): __table_args__ = ( Index("ix_log_entries_timestamp_service", "timestamp", "service"), Index("ix_log_entries_source_adapter", "source_adapter"), + Index("ix_log_entries_scope_timestamp", "scope", "timestamp"), + Index( + "ix_log_entries_scope_service_environment_fingerprint", + "scope", + "service", + "environment", + "fingerprint", + ), Index( LOG_ENTRY_DEDUP_INDEX, "scope", @@ -208,6 +220,9 @@ class ApiKey(Base): key_hash: Mapped[str] = mapped_column(Text, nullable=False) role: Mapped[str] = mapped_column(String(20), nullable=False) scope: Mapped[str] = mapped_column(String(255), nullable=False, default="default") + allow_scope_override: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False, server_default=text("false") + ) webhook_secret: Mapped[str | None] = mapped_column(Text, nullable=True) revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) diff --git a/src/db/scope_filter.py b/src/db/scope_filter.py new file mode 100644 index 0000000..b97b462 --- /dev/null +++ b/src/db/scope_filter.py @@ -0,0 +1,32 @@ +"""Helpers that pin LogEntry / IngestionJob queries to a single scope. + +Every analysis and ingest-list path must go through these so one scope's +rows cannot appear in another's SQL (including baseline comparison). +""" + +from __future__ import annotations + +from sqlalchemy import Select +from sqlalchemy.sql.elements import ColumnElement + +from src.db.models import IngestionJob, LogEntry + + +def log_entry_scope_clause(scope: str) -> ColumnElement[bool]: + """``LogEntry.scope == scope`` — use in additional ``where()`` calls.""" + return LogEntry.scope == scope + + +def ingestion_job_scope_clause(scope: str) -> ColumnElement[bool]: + """``IngestionJob.scope == scope`` — use in additional ``where()`` calls.""" + return IngestionJob.scope == scope + + +def filter_log_entries_by_scope(stmt: Select, scope: str) -> Select: + """Restrict a ``LogEntry`` select to ``scope``.""" + return stmt.where(log_entry_scope_clause(scope)) + + +def filter_ingestion_jobs_by_scope(stmt: Select, scope: str) -> Select: + """Restrict an ``IngestionJob`` select to ``scope``.""" + return stmt.where(ingestion_job_scope_clause(scope)) diff --git a/src/worker/runner.py b/src/worker/runner.py index 1c3fe97..46992bc 100644 --- a/src/worker/runner.py +++ b/src/worker/runner.py @@ -102,6 +102,16 @@ def run_ingest_job(db, worker_job) -> dict: raise ValueError( f"No ingestion job found with id {payload['resume_ingestion_job_id']}" ) + payload_scope = str(payload.get("scope") or "default") + prior_scope = getattr(prior, "scope", None) + if ( + isinstance(prior_scope, str) + and prior_scope.strip() + and prior_scope.strip() != payload_scope + ): + raise ValueError( + "resume_ingestion_job_id is in a different scope than this ingest" + ) resume_cursors = (prior.metadata_json or {}).get("cursors") resume_completed_streams = (prior.metadata_json or {}).get( "completed_streams" diff --git a/tests/unit/test_ingest_idempotency.py b/tests/unit/test_ingest_idempotency.py index 496eb51..12c7074 100644 --- a/tests/unit/test_ingest_idempotency.py +++ b/tests/unit/test_ingest_idempotency.py @@ -296,3 +296,106 @@ def capture_add(obj: object) -> None: if isinstance(obj, IngestIdempotencyKey) ] assert added_keys == [] + + def test_cross_scope_key_reuse_does_not_return_other_scopes_job(self) -> None: + stored: dict[str, IngestIdempotencyKey | None] = {"row": None} + jobs: dict[str, WorkerJob | None] = {"job": None} + mock_db = _ctx_db() + + def query_side_effect(model: object) -> MagicMock: + q = MagicMock() + name = getattr(model, "__name__", "") + if name == "IngestIdempotencyKey": + q.filter.return_value.first.return_value = stored["row"] + elif name == "WorkerJob": + q.filter.return_value.first.return_value = jobs["job"] + else: + q.filter.return_value.first.return_value = None + return q + + def capture_add(obj: object) -> None: + if isinstance(obj, WorkerJob): + obj.id = uuid.uuid4() + jobs["job"] = obj + if isinstance(obj, IngestIdempotencyKey): + stored["row"] = obj + if getattr(obj, "expires_at", None) is None: + obj.expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=1) + + mock_db.query.side_effect = query_side_effect + mock_db.add.side_effect = capture_add + + with ( + patch("src.adapters.file.adapter.discover_files", return_value=["f.log"]), + patch("src.db.session.get_db", side_effect=lambda: mock_db), + ): + first = client.post( + "/v1/ingestions", + json={"paths": ["/logs"], "scope": "incident:A"}, + headers={"Idempotency-Key": "shared-across-tenants"}, + ) + second = client.post( + "/v1/ingestions", + json={"paths": ["/logs"], "scope": "incident:B"}, + headers={"Idempotency-Key": "shared-across-tenants"}, + ) + + assert first.status_code == 202 + first_id = first.json()["worker_job_id"] + assert first_id + assert jobs["job"] is not None + assert jobs["job"].payload_json["scope"] == "incident:A" + + assert second.status_code == 409 + body = second.json() + assert body["error_code"] == "IDEMPOTENCY_SCOPE_CONFLICT" + assert "worker_job_id" not in body + assert "ingestion_job_id" not in body + assert first_id not in str(body) + + def test_same_explicit_scope_still_replays(self) -> None: + stored: dict[str, IngestIdempotencyKey | None] = {"row": None} + jobs: dict[str, WorkerJob | None] = {"job": None} + mock_db = _ctx_db() + + def query_side_effect(model: object) -> MagicMock: + q = MagicMock() + name = getattr(model, "__name__", "") + if name == "IngestIdempotencyKey": + q.filter.return_value.first.return_value = stored["row"] + elif name == "WorkerJob": + q.filter.return_value.first.return_value = jobs["job"] + else: + q.filter.return_value.first.return_value = None + return q + + def capture_add(obj: object) -> None: + if isinstance(obj, WorkerJob): + obj.id = uuid.uuid4() + jobs["job"] = obj + if isinstance(obj, IngestIdempotencyKey): + stored["row"] = obj + if getattr(obj, "expires_at", None) is None: + obj.expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=1) + + mock_db.query.side_effect = query_side_effect + mock_db.add.side_effect = capture_add + + with ( + patch("src.adapters.file.adapter.discover_files", return_value=["f.log"]), + patch("src.db.session.get_db", side_effect=lambda: mock_db), + ): + first = client.post( + "/v1/ingestions", + json={"paths": ["/logs"], "scope": "incident:A"}, + headers={"Idempotency-Key": "same-scope"}, + ) + second = client.post( + "/v1/ingestions", + json={"paths": ["/logs"], "scope": "incident:A"}, + headers={"Idempotency-Key": "same-scope"}, + ) + + assert first.status_code == 202 + assert second.status_code == 202 + assert first.json()["worker_job_id"] == second.json()["worker_job_id"] diff --git a/tests/unit/test_scope.py b/tests/unit/test_scope.py new file mode 100644 index 0000000..1c84e10 --- /dev/null +++ b/tests/unit/test_scope.py @@ -0,0 +1,459 @@ +"""Unit tests for G8 scope isolation (no database).""" + +from __future__ import annotations + +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch +from uuid import uuid4 + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import select + +from src.api.app import app +from src.api.auth.middleware import AuthPrincipal +from src.api.auth.scope import ( + ERROR_SCOPE_MISMATCH, + ERROR_SCOPE_REQUIRED, + ScopeResolutionError, + resolve_scope, +) +from src.config.settings import Settings +from src.core.clustering.baseline import get_baseline_counts +from src.core.retrieval.question_router import search_logs, search_logs_semantic +from src.db.models import IngestionJob, LogEntry +from src.db.scope_filter import ( + filter_ingestion_jobs_by_scope, + filter_log_entries_by_scope, +) + +client = TestClient(app, raise_server_exceptions=False) + +WINDOW_START = datetime(2026, 8, 16, 5, 0, tzinfo=timezone.utc) +WINDOW_END = datetime(2026, 8, 16, 7, 0, tzinfo=timezone.utc) + + +def _principal( + *, + scope: str = "incident:A", + allow_scope_override: bool = False, + auth_method: str = "api_key", + role: str = "query", +) -> AuthPrincipal: + return AuthPrincipal( + role=role, + scope=scope, + auth_method=auth_method, + key_id=str(uuid4()), + allow_scope_override=allow_scope_override, + ) + + +def _compiled(stmt) -> str: + return str(stmt.compile(compile_kwargs={"literal_binds": False})).lower() + + +def _params(stmt) -> list[object]: + return list(stmt.compile(compile_kwargs={"literal_binds": False}).params.values()) + + +def _auth_settings(**kwargs: object) -> Settings: + values: dict[str, object] = {"auth_enabled": True, "auth_mode": "api_key"} + values.update(kwargs) + return Settings(_env_file=None, **values) + + +def _key( + role: str, + *, + scope: str = "default", + allow_scope_override: bool = False, +) -> MagicMock: + record = MagicMock() + record.id = uuid4() + record.name = "test" + record.key_prefix = "rlk_testhash" + record.role = role + record.scope = scope + record.allow_scope_override = allow_scope_override + record.revoked_at = None + return record + + +def _ctx_db() -> MagicMock: + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + mock_db.execute.return_value.scalar_one.return_value = 0 + mock_db.execute.return_value.all.return_value = [] + mock_db.execute.return_value.scalars.return_value.all.return_value = [] + mock_db.execute.return_value.scalar_one_or_none.return_value = None + return mock_db + + +# ── Resolution ──────────────────────────────────────────────────────────────── + + +class TestResolveScope: + def test_auth_off_defaults_to_default(self) -> None: + assert ( + resolve_scope(requested_scope=None, auth_enabled=False, principal=None) + == "default" + ) + + def test_auth_off_uses_explicit_request_scope(self) -> None: + assert ( + resolve_scope( + requested_scope="incident:INC-9", + auth_enabled=False, + principal=None, + ) + == "incident:INC-9" + ) + + def test_auth_off_strips_whitespace(self) -> None: + assert ( + resolve_scope( + requested_scope=" env:prod ", + auth_enabled=False, + principal=None, + ) + == "env:prod" + ) + + def test_auth_off_blank_request_falls_back_to_default(self) -> None: + assert ( + resolve_scope(requested_scope=" ", auth_enabled=False, principal=None) + == "default" + ) + + def test_pinned_uses_key_scope(self) -> None: + assert ( + resolve_scope( + requested_scope=None, + auth_enabled=True, + principal=_principal(scope="incident:A"), + ) + == "incident:A" + ) + + def test_pinned_same_request_scope_ok(self) -> None: + assert ( + resolve_scope( + requested_scope="incident:A", + auth_enabled=True, + principal=_principal(scope="incident:A"), + ) + == "incident:A" + ) + + def test_pinned_mismatch_raises(self) -> None: + with pytest.raises(ScopeResolutionError) as exc: + resolve_scope( + requested_scope="incident:B", + auth_enabled=True, + principal=_principal(scope="incident:A"), + ) + assert exc.value.status_code == 403 + assert exc.value.error_code == ERROR_SCOPE_MISMATCH + + def test_oidc_is_pinned_even_if_override_flag_were_set(self) -> None: + principal = AuthPrincipal( + role="query", + scope="default", + auth_method="oidc", + allow_scope_override=True, + ) + with pytest.raises(ScopeResolutionError) as exc: + resolve_scope( + requested_scope="incident:X", + auth_enabled=True, + principal=principal, + ) + assert exc.value.error_code == ERROR_SCOPE_MISMATCH + + def test_override_uses_request_scope(self) -> None: + assert ( + resolve_scope( + requested_scope="incident:B", + auth_enabled=True, + principal=_principal( + scope="incident:A", allow_scope_override=True + ), + ) + == "incident:B" + ) + + def test_override_falls_back_to_key_scope(self) -> None: + assert ( + resolve_scope( + requested_scope=None, + auth_enabled=True, + principal=_principal( + scope="incident:A", allow_scope_override=True + ), + ) + == "incident:A" + ) + + def test_override_both_empty_is_scope_required(self) -> None: + with pytest.raises(ScopeResolutionError) as exc: + resolve_scope( + requested_scope=" ", + auth_enabled=True, + principal=_principal(scope="", allow_scope_override=True), + ) + assert exc.value.status_code == 400 + assert exc.value.error_code == ERROR_SCOPE_REQUIRED + + def test_auth_on_without_principal_is_scope_required(self) -> None: + with pytest.raises(ScopeResolutionError) as exc: + resolve_scope( + requested_scope=None, auth_enabled=True, principal=None + ) + assert exc.value.error_code == ERROR_SCOPE_REQUIRED + + def test_pinned_empty_key_scope_is_scope_required(self) -> None: + with pytest.raises(ScopeResolutionError) as exc: + resolve_scope( + requested_scope=None, + auth_enabled=True, + principal=_principal(scope=" "), + ) + assert exc.value.error_code == ERROR_SCOPE_REQUIRED + + +# ── SQL isolation ───────────────────────────────────────────────────────────── + + +class TestScopeFilterSQL: + def test_log_entry_helper_binds_scope(self) -> None: + stmt = filter_log_entries_by_scope(select(LogEntry), "incident:A") + sql = _compiled(stmt) + assert "log_entries.scope" in sql + assert "incident:A" in _params(stmt) + + def test_ingestion_job_helper_binds_scope(self) -> None: + stmt = filter_ingestion_jobs_by_scope(select(IngestionJob), "incident:B") + sql = _compiled(stmt) + assert "ingestion_jobs.scope" in sql + assert "incident:B" in _params(stmt) + + def test_two_scopes_compile_to_different_binds(self) -> None: + a = filter_log_entries_by_scope(select(LogEntry), "incident:A") + b = filter_log_entries_by_scope(select(LogEntry), "incident:B") + assert "incident:A" in _params(a) + assert "incident:B" in _params(b) + assert "incident:B" not in _params(a) + assert "incident:A" not in _params(b) + + def test_baseline_sql_includes_scope(self) -> None: + db = MagicMock() + db.execute.return_value.all.return_value = [] + get_baseline_counts( + db, WINDOW_START, WINDOW_END, scope="incident:A" + ) + stmt = db.execute.call_args[0][0] + sql = _compiled(stmt) + assert "log_entries.scope" in sql + assert "incident:A" in _params(stmt) + + def test_search_logs_sql_includes_scope(self) -> None: + db = MagicMock() + db.execute.return_value.scalars.return_value.all.return_value = [] + search_logs( + db, ["error"], WINDOW_START, WINDOW_END, None, None, scope="incident:A" + ) + stmt = db.execute.call_args[0][0] + sql = _compiled(stmt) + assert "log_entries.scope" in sql + assert "incident:A" in _params(stmt) + + def test_semantic_search_sql_includes_scope(self) -> None: + from src.core.embeddings.store import STORED_EMBEDDING_DIMS + + db = MagicMock() + db.execute.return_value.scalars.return_value.all.return_value = [] + search_logs_semantic( + db, + [0.01] * STORED_EMBEDDING_DIMS, + WINDOW_START, + WINDOW_END, + None, + None, + scope="incident:A", + ) + stmt = db.execute.call_args[0][0] + sql = _compiled(stmt) + assert "log_entries.scope" in sql + assert "incident:A" in _params(stmt) + + +# ── API TestClient (mocked DB) ──────────────────────────────────────────────── + + +class TestScopeApi: + def test_auth_off_explain_passes_default_scope(self) -> None: + mock_result = MagicMock() + mock_result.window_start = WINDOW_START + mock_result.window_end = WINDOW_END + mock_result.summary_text = "ok" + mock_result.confidence = "low" + mock_result.mode = "rules" + mock_result.total_logs = 0 + mock_result.services_affected = [] + mock_result.primary_cluster = None + mock_result.secondary_clusters = [] + mock_result.trigger_candidates = [] + mock_result.evidence_items = [] + mock_db = _ctx_db() + + with patch("src.db.session.get_db", side_effect=lambda: mock_db), \ + patch( + "src.core.explain.summarizer.explain_window", + return_value=mock_result, + ) as mock_explain, \ + patch("src.api.routes.explain._load_from_cache", return_value=None), \ + patch("src.api.routes.explain._save_to_cache"): + resp = client.post("/v1/query/explain", json={"since": "1h", "no_llm": True}) + + assert resp.status_code == 200 + assert resp.json()["scope"] == "default" + assert mock_explain.call_args.kwargs["scope"] == "default" + + def test_auth_off_explicit_request_scope(self) -> None: + mock_result = MagicMock() + mock_result.window_start = WINDOW_START + mock_result.window_end = WINDOW_END + mock_result.summary_text = "ok" + mock_result.confidence = "low" + mock_result.mode = "rules" + mock_result.total_logs = 0 + mock_result.services_affected = [] + mock_result.primary_cluster = None + mock_result.secondary_clusters = [] + mock_result.trigger_candidates = [] + mock_result.evidence_items = [] + mock_db = _ctx_db() + + with patch("src.db.session.get_db", side_effect=lambda: mock_db), \ + patch( + "src.core.explain.summarizer.explain_window", + return_value=mock_result, + ) as mock_explain, \ + patch("src.api.routes.explain._load_from_cache", return_value=None), \ + patch("src.api.routes.explain._save_to_cache"): + resp = client.post( + "/v1/query/explain", + json={"since": "1h", "no_llm": True, "scope": "incident:CLI"}, + ) + + assert resp.status_code == 200 + assert resp.json()["scope"] == "incident:CLI" + assert mock_explain.call_args.kwargs["scope"] == "incident:CLI" + + def test_pinned_key_mismatch_is_403(self) -> None: + settings = _auth_settings() + with patch("src.config.get_settings", return_value=settings), \ + patch( + "src.api.auth.keys.lookup_api_key", + return_value=_key("query", scope="incident:A"), + ): + resp = client.post( + "/v1/query/explain", + json={"since": "1h", "scope": "incident:B"}, + headers={"Authorization": "Bearer rlk_queryrolexx"}, + ) + assert resp.status_code == 403 + body = resp.json() + assert body["error_code"] == ERROR_SCOPE_MISMATCH + assert "detail" not in body + + def test_override_key_uses_request_scope(self) -> None: + mock_result = MagicMock() + mock_result.window_start = WINDOW_START + mock_result.window_end = WINDOW_END + mock_result.summary_text = "ok" + mock_result.confidence = "low" + mock_result.mode = "rules" + mock_result.total_logs = 0 + mock_result.services_affected = [] + mock_result.primary_cluster = None + mock_result.secondary_clusters = [] + mock_result.trigger_candidates = [] + mock_result.evidence_items = [] + mock_db = _ctx_db() + settings = _auth_settings() + key = _key("query", scope="incident:A", allow_scope_override=True) + + with patch("src.config.get_settings", return_value=settings), \ + patch("src.api.auth.keys.lookup_api_key", return_value=key), \ + patch("src.db.session.get_db", side_effect=lambda: mock_db), \ + patch( + "src.core.explain.summarizer.explain_window", + return_value=mock_result, + ) as mock_explain, \ + patch("src.api.routes.explain._load_from_cache", return_value=None), \ + patch("src.api.routes.explain._save_to_cache"): + resp = client.post( + "/v1/query/explain", + json={"since": "1h", "no_llm": True, "scope": "incident:B"}, + headers={"Authorization": "Bearer rlk_queryrolexx"}, + ) + + assert resp.status_code == 200 + assert resp.json()["scope"] == "incident:B" + assert mock_explain.call_args.kwargs["scope"] == "incident:B" + + def test_two_keys_cannot_list_each_others_ingestions(self) -> None: + settings = _auth_settings() + mock_db = _ctx_db() + + def _list(scope: str) -> None: + with patch("src.config.get_settings", return_value=settings), \ + patch( + "src.api.auth.keys.lookup_api_key", + return_value=_key("query", scope=scope), + ), \ + patch("src.db.session.get_db", side_effect=lambda: mock_db): + resp = client.get( + "/v1/ingestions", + headers={"Authorization": "Bearer rlk_queryrolexx"}, + ) + assert resp.status_code == 200 + stmt = mock_db.execute.call_args[0][0] + sql = _compiled(stmt) + assert "ingestion_jobs.scope" in sql + assert scope in _params(stmt) + + _list("incident:A") + _list("incident:B") + + def test_ingest_stamps_resolved_scope_on_worker_payload(self) -> None: + settings = _auth_settings() + key = _key("ingest", scope="incident:INC-9") + mock_db = _ctx_db() + added: list[object] = [] + + def capture_add(obj: object) -> None: + added.append(obj) + obj.id = uuid4() # type: ignore[attr-defined] + + mock_db.add.side_effect = capture_add + + with patch("src.config.get_settings", return_value=settings), \ + patch("src.api.auth.keys.lookup_api_key", return_value=key), \ + patch( + "src.adapters.file.adapter.discover_files", + return_value=["f.log"], + ), \ + patch("src.db.session.get_db", side_effect=lambda: mock_db): + resp = client.post( + "/v1/ingestions", + json={"paths": ["/logs"]}, + headers={"Authorization": "Bearer rlk_ingestrole1"}, + ) + + assert resp.status_code == 202 + jobs = [obj for obj in added if hasattr(obj, "payload_json")] + assert jobs[0].payload_json["scope"] == "incident:INC-9" diff --git a/tests/unit/test_worker.py b/tests/unit/test_worker.py index 87c6b40..37ad1b8 100644 --- a/tests/unit/test_worker.py +++ b/tests/unit/test_worker.py @@ -5,8 +5,7 @@ claim/dispatch/failure state machine without I/O. """ import pytest -from datetime import datetime, timezone -from unittest.mock import MagicMock, patch, call +from unittest.mock import MagicMock, patch from src.worker.runner import claim_next_job, process_one @@ -112,7 +111,7 @@ def test_returns_true_when_job_processed(self): "error_count": 2, "services_detected": ["billing-worker"], "duration_seconds": 0.5, - }) as mock_run: + }): result = process_one(db) assert result is True @@ -299,6 +298,32 @@ def test_unknown_resume_job_fails_the_worker_job(self): assert "00000000-0000-0000-0000-000000000000" in job.error mock_ingest.assert_not_called() + def test_resume_job_in_other_scope_fails_the_worker_job(self): + """resume_ingestion_job_id must belong to the same isolation scope.""" + db = _mock_db() + job = _mock_worker_job(payload={ + "adapter": "cloudwatch", + "params": {"log_group": "/aws/lambda/x"}, + "service": None, + "env": None, + "source_name": None, + "format": "auto", + "scope": "incident:B", + "resume_ingestion_job_id": "11111111-1111-1111-1111-111111111111", + }) + db.execute.return_value.scalar_one_or_none.return_value = job + prior = MagicMock() + prior.scope = "incident:A" + db.query.return_value.filter.return_value.first.return_value = prior + + with patch("src.core.ingestion.service.ingest_from_source") as mock_ingest: + result = process_one(db) + + assert result is True + assert job.status == "failed" + assert "different scope" in job.error + mock_ingest.assert_not_called() + def test_unknown_job_type_fails_gracefully(self): db = _mock_db() job = _mock_worker_job(job_type="unknown_type")