diff --git a/api/app.py b/api/app.py index e39448d..942a6f3 100644 --- a/api/app.py +++ b/api/app.py @@ -10,17 +10,17 @@ from flask_cors import CORS from api.models.finding import DatabaseManager +from api.observability import configure_logging, get_request_id, init_app, init_sentry load_dotenv() -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s %(levelname)s %(name)s: %(message)s", -) +configure_logging() +init_sentry() logger = logging.getLogger(__name__) -# Paths that are always public regardless of environment or demo mode -_ALWAYS_PUBLIC = {"/", "/health"} +# Paths that are always public regardless of environment or demo mode. +# /ready and /metrics are probe/scrape endpoints and must never require auth. +_ALWAYS_PUBLIC = {"/", "/health", "/ready", "/metrics"} _INSECURE_JWT_DEFAULT = "change-me-in-production" _MIN_JWT_SECRET_LENGTH = 32 @@ -92,6 +92,14 @@ def create_app() -> Flask: """ app = Flask(__name__) + # ------------------------------------------------------------------ # + # Observability # + # ------------------------------------------------------------------ # + # Registered first so request IDs and request timing are available to + # every later before_request handler (including JWT auth) and to the + # error handlers. Also mounts the public /metrics endpoint. + init_app(app) + # ------------------------------------------------------------------ # # Configuration & Security # # ------------------------------------------------------------------ # @@ -164,7 +172,12 @@ def verify_jwt() -> None: auth = request.headers.get("Authorization", "") if not auth.startswith("Bearer "): - return jsonify({"error": "Missing or malformed Authorization header"}), 401 + return jsonify( + { + "error": "Missing or malformed Authorization header", + "request_id": get_request_id(), + } + ), 401 token = auth.split(" ", 1)[1] try: @@ -175,9 +188,10 @@ def verify_jwt() -> None: ) g.user = payload except jwt.ExpiredSignatureError: - return jsonify({"error": "Token has expired"}), 401 + return jsonify({"error": "Token has expired", "request_id": get_request_id()}), 401 except jwt.InvalidTokenError as exc: - return jsonify({"error": f"Invalid token: {exc}"}), 401 + logger.warning("Invalid JWT token: %s", exc) + return jsonify({"error": "Invalid token", "request_id": get_request_id()}), 401 return None @@ -214,32 +228,44 @@ def index(): @app.get("/health") def health(): + """Liveness probe: process is up. Deliberately does not touch the DB.""" return jsonify({"status": "ok"}) + @app.get("/ready") + def ready(): + """Readiness probe: 200 when the database is reachable, else 503.""" + try: + db = DatabaseManager() + db.ping() + return jsonify({"status": "ready"}), 200 + except Exception as exc: + logger.warning("Readiness check failed: %s", exc) + return jsonify({"status": "not_ready", "error": "database_unreachable"}), 503 + # ------------------------------------------------------------------ # # Error handlers # # ------------------------------------------------------------------ # @app.errorhandler(400) def bad_request(exc): - return jsonify({"error": "Bad request", "detail": str(exc)}), 400 + return jsonify({"error": "Bad request", "detail": str(exc), "request_id": get_request_id()}), 400 @app.errorhandler(401) def unauthorized(exc): - return jsonify({"error": "Unauthorized"}), 401 + return jsonify({"error": "Unauthorized", "request_id": get_request_id()}), 401 @app.errorhandler(403) def forbidden(exc): - return jsonify({"error": "Forbidden"}), 403 + return jsonify({"error": "Forbidden", "request_id": get_request_id()}), 403 @app.errorhandler(404) def not_found(exc): - return jsonify({"error": "Not found"}), 404 + return jsonify({"error": "Not found", "request_id": get_request_id()}), 404 @app.errorhandler(500) def internal_error(exc): logger.error("Unhandled exception: %s", exc) - return jsonify({"error": "Internal server error"}), 500 + return jsonify({"error": "Internal server error", "request_id": get_request_id()}), 500 logger.info("OpenShield API created - %d blueprints registered", len(app.blueprints)) return app diff --git a/api/models/finding.py b/api/models/finding.py index e40ea86..2792362 100644 --- a/api/models/finding.py +++ b/api/models/finding.py @@ -103,6 +103,18 @@ def close(self) -> None: self.conn = None logger.debug("Database connection closed") + def ping(self) -> bool: + """Execute a trivial query to confirm database connectivity. + + Used by the /ready probe. Raises on failure so the caller can map it + to a 503; returns True when the database answers. + """ + conn = self._get_conn() + with conn.cursor() as cur: + cur.execute("SELECT 1") + cur.fetchone() + return True + # ------------------------------------------------------------------ # # Schema # # ------------------------------------------------------------------ # diff --git a/api/observability.py b/api/observability.py new file mode 100644 index 0000000..8f48ecd --- /dev/null +++ b/api/observability.py @@ -0,0 +1,180 @@ +"""Shared observability layer: structured logging, request IDs, Prometheus +metrics and optional Sentry error tracking. + +This module is intentionally free of any project imports so that it can be +used from both the API (``api.app``) and the background worker +(``scanner.worker``) without creating an import cycle. + +Metric cardinality note: no metric is ever labelled with a request ID, +scan ID, subscription ID, resource ID, error message or free-form user +input. Only low-cardinality, bounded values (HTTP method, view name, +status code, scan status, rule ID and provider name) are used as labels. +""" + +import logging +import os +import time +import uuid + +from flask import Flask, Response, g, request +from prometheus_client import ( + CONTENT_TYPE_LATEST, + Counter, + Gauge, + Histogram, + generate_latest, +) + +# python-json-logger moved ``JsonFormatter`` from ``pythonjsonlogger.jsonlogger`` +# to ``pythonjsonlogger.json`` in 3.1. Prefer the new path, fall back for <3.1. +try: # pragma: no cover - trivial import shim + from pythonjsonlogger.json import JsonFormatter +except ImportError: # pragma: no cover + from pythonjsonlogger.jsonlogger import JsonFormatter + +logger = logging.getLogger(__name__) + +REQUEST_ID_HEADER = "X-Request-ID" + +# --------------------------------------------------------------------------- # +# Prometheus metrics # +# --------------------------------------------------------------------------- # +# Defined at module import so they are process-wide singletons. Re-importing +# this module (e.g. across repeated create_app() calls in tests) reuses the +# same collectors and never double-registers. + +HTTP_REQUESTS_TOTAL = Counter( + "openshield_http_requests_total", + "Total HTTP requests processed by the API.", + ["method", "endpoint", "status"], +) +HTTP_REQUEST_DURATION_SECONDS = Histogram( + "openshield_http_request_duration_seconds", + "HTTP request latency in seconds.", + ["method", "endpoint"], +) +SCANS_TOTAL = Counter( + "openshield_scans_total", + "Total scans processed by the worker, by terminal status.", + ["status"], +) +SCAN_DURATION_SECONDS = Histogram( + "openshield_scan_duration_seconds", + "Wall-clock duration of a single scan execution in seconds.", +) +PENDING_SCANS = Gauge( + "openshield_pending_scans", + "Number of scans currently waiting in the pending queue.", +) +RULE_ERRORS_TOTAL = Counter( + "openshield_rule_errors_total", + "Total number of times a scanner rule raised an exception.", + ["rule_id"], +) +NVD_REQUEST_LATENCY_SECONDS = Histogram( + "openshield_nvd_request_latency_seconds", + "Latency of outbound NVD HTTP requests in seconds.", +) +LLM_PROVIDER_LATENCY_SECONDS = Histogram( + "openshield_llm_provider_latency_seconds", + "Latency of outbound LLM provider requests in seconds.", + ["provider"], +) + + +# --------------------------------------------------------------------------- # +# Structured logging # +# --------------------------------------------------------------------------- # +def configure_logging(level: int = logging.INFO) -> None: + """Configure the root logger to emit structured JSON to stderr. + + Safe to call multiple times — it replaces the root handlers each call so + repeated invocations (API import, worker start, tests) do not stack + duplicate handlers. Any ``extra={...}`` fields passed to log calls are + included as top-level JSON keys. + """ + handler = logging.StreamHandler() + handler.setFormatter(JsonFormatter("%(asctime)s %(levelname)s %(name)s %(message)s")) + root = logging.getLogger() + root.handlers = [handler] + root.setLevel(level) + + +# --------------------------------------------------------------------------- # +# Sentry (optional) # +# --------------------------------------------------------------------------- # +def init_sentry() -> bool: + """Initialise Sentry error tracking only when ``SENTRY_DSN`` is set. + + Returns ``True`` when Sentry was initialised, ``False`` otherwise (DSN + unset, or ``sentry-sdk`` not installed). Never raises. + """ + dsn = os.environ.get("SENTRY_DSN") + if not dsn: + return False + try: + import sentry_sdk + except ImportError: + logger.warning("SENTRY_DSN is set but sentry-sdk is not installed; skipping.") + return False + + try: + sentry_sdk.init( + dsn=dsn, + environment=os.environ.get("OPENSHIELD_ENV", "development"), + traces_sample_rate=float(os.environ.get("SENTRY_TRACES_SAMPLE_RATE", "0.0")), + ) + except Exception as exc: + # A malformed DSN must never take down the API or the worker. + logger.warning("Failed to initialise Sentry: %s", exc) + return False + logger.info("Sentry error tracking initialised.") + return True + + +# --------------------------------------------------------------------------- # +# Request IDs # +# --------------------------------------------------------------------------- # +def get_request_id() -> str: + """Return the request ID bound to the current request context. + + Falls back to a freshly generated UUID if the middleware has not run + (e.g. outside a request), so callers always receive a usable value. + """ + rid = getattr(g, "request_id", None) + if not rid: + rid = str(uuid.uuid4()) + g.request_id = rid + return rid + + +# --------------------------------------------------------------------------- # +# Flask wiring # +# --------------------------------------------------------------------------- # +def init_app(app: Flask) -> None: + """Register request-ID and Prometheus middleware and the /metrics route. + + Must be called before any other ``before_request`` handlers (such as JWT + auth) so that ``g.request_id`` is available to them and to error handlers. + """ + + @app.before_request + def _start_observability() -> None: + g.request_id = request.headers.get(REQUEST_ID_HEADER) or str(uuid.uuid4()) + g.request_start_time = time.perf_counter() + + @app.after_request + def _record_observability(response: Response) -> Response: + response.headers[REQUEST_ID_HEADER] = get_request_id() + + endpoint = request.endpoint or "unknown" + method = request.method + HTTP_REQUESTS_TOTAL.labels(method=method, endpoint=endpoint, status=response.status_code).inc() + start = getattr(g, "request_start_time", None) + if start is not None: + HTTP_REQUEST_DURATION_SECONDS.labels(method=method, endpoint=endpoint).observe(time.perf_counter() - start) + return response + + @app.get("/metrics") + def metrics() -> Response: + return Response(generate_latest(), content_type=CONTENT_TYPE_LATEST) diff --git a/api/services/ai_provider.py b/api/services/ai_provider.py index 9134632..9af4c16 100644 --- a/api/services/ai_provider.py +++ b/api/services/ai_provider.py @@ -3,6 +3,8 @@ import logging import requests +from api.observability import LLM_PROVIDER_LATENCY_SECONDS + logger = logging.getLogger(__name__) PROVIDERS = ("anthropic", "groq", "gemini") @@ -23,11 +25,12 @@ def get_completion(provider: str, api_key: str, prompt: str, model: str = None) resolved_model = model or DEFAULT_MODELS[provider] - if provider == "anthropic": - return _anthropic(api_key, prompt, resolved_model) - if provider == "groq": - return _groq(api_key, prompt, resolved_model) - return _gemini(api_key, prompt, resolved_model) + with LLM_PROVIDER_LATENCY_SECONDS.labels(provider=provider).time(): + if provider == "anthropic": + return _anthropic(api_key, prompt, resolved_model) + if provider == "groq": + return _groq(api_key, prompt, resolved_model) + return _gemini(api_key, prompt, resolved_model) def _anthropic(api_key: str, prompt: str, model: str) -> str: diff --git a/requirements.txt b/requirements.txt index 5dee5a1..12a58f3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -27,5 +27,8 @@ azure-keyvault-keys==4.9.0 chromadb==0.4.24 sentence-transformers==2.7.0 numpy<2.0 +prometheus-client>=0.19.0 +python-json-logger>=2.0.7 +sentry-sdk>=1.40.0 pytest>=7.4.0 pytest-cov>=4.1.0 diff --git a/scanner/engine.py b/scanner/engine.py index 7335186..661664f 100644 --- a/scanner/engine.py +++ b/scanner/engine.py @@ -7,6 +7,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional +from api.observability import RULE_ERRORS_TOTAL from scanner.azure_client import AzureClient logger = logging.getLogger(__name__) @@ -121,6 +122,7 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]: findings.extend(rule_findings) logger.info("Rule %s produced %d finding(s)", rule_id, len(rule_findings)) except Exception as exc: + RULE_ERRORS_TOTAL.labels(rule_id=rule_id).inc() logger.error("Rule %s raised an exception: %s", rule_id, exc, exc_info=True) completed_at = datetime.now(timezone.utc).isoformat() diff --git a/scanner/nvd_client.py b/scanner/nvd_client.py index aa02b50..a922f97 100644 --- a/scanner/nvd_client.py +++ b/scanner/nvd_client.py @@ -24,6 +24,8 @@ import json from typing import Optional +from api.observability import NVD_REQUEST_LATENCY_SECONDS + logger = logging.getLogger(__name__) _NVD_BASE_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0" @@ -143,8 +145,9 @@ def query_nvd(keyword: str, results_per_page: int = _RESULTS_PER_PAGE) -> list[d headers={"User-Agent": "OpenShield/0.1 (github.com/openshield-org/openshield)"}, ) # URL host is the hardcoded NVD API base, not user-controlled - with urllib.request.urlopen(req, timeout=10) as resp: # nosec B310 - data = json.loads(resp.read()) + with NVD_REQUEST_LATENCY_SECONDS.time(): + with urllib.request.urlopen(req, timeout=10) as resp: # nosec B310 + data = json.loads(resp.read()) vulnerabilities = data.get("vulnerabilities", []) results = [parsed for item in vulnerabilities if (parsed := _parse_cve_item(item)) is not None] diff --git a/scanner/worker.py b/scanner/worker.py index ffdd27a..21a94c6 100644 --- a/scanner/worker.py +++ b/scanner/worker.py @@ -12,12 +12,16 @@ from datetime import datetime, timezone from api.models.finding import DatabaseManager +from api.observability import ( + PENDING_SCANS, + SCAN_DURATION_SECONDS, + SCANS_TOTAL, + configure_logging, + init_sentry, +) from scanner.engine import ScanEngine -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s %(levelname)s %(name)s: %(message)s", -) +configure_logging() logger = logging.getLogger("scanner.worker") POLL_INTERVAL_SECONDS = 5 @@ -30,6 +34,9 @@ def run_worker(): logger.error("DATABASE_URL environment variable is not set") return + # Initialise Sentry only when SENTRY_DSN is configured (no-op otherwise). + init_sentry() + db = DatabaseManager(db_url) logger.info("OpenShield Background Worker started. Polling every %ds", POLL_INTERVAL_SECONDS) @@ -38,7 +45,10 @@ def run_worker(): # 1. Cleanup stale scans from previous crashes db.recover_stale_scans(timeout_minutes=60) - # 2. Atomic claim + # 2. Publish current queue depth + PENDING_SCANS.set(len(db.get_pending_scans())) + + # 3. Atomic claim scan = db.claim_next_pending_scan() if not scan: time.sleep(POLL_INTERVAL_SECONDS) @@ -47,8 +57,14 @@ def run_worker(): scan_id = str(scan["scan_id"]) subscription_id = scan["subscription_id"] - logger.info("Starting scan %s for %s", scan_id, subscription_id) + logger.info( + "Starting scan %s for %s", + scan_id, + subscription_id, + extra={"scan_id": scan_id}, + ) + scan_start = time.perf_counter() try: engine = ScanEngine(subscription_id) result = engine.run_scan(scan_id) @@ -58,14 +74,27 @@ def run_worker(): result["status"] = "completed" db.save_scan(result) - logger.info("Successfully completed scan %s", scan_id) + SCANS_TOTAL.labels(status="completed").inc() + logger.info( + "Successfully completed scan %s", + scan_id, + extra={"scan_id": scan_id}, + ) except Exception as exc: error_msg = f"{str(exc)}\n{traceback.format_exc()}" - logger.error("Scan %s failed: %s", scan_id, error_msg) + SCANS_TOTAL.labels(status="failed").inc() + logger.error( + "Scan %s failed: %s", + scan_id, + error_msg, + extra={"scan_id": scan_id}, + ) # Sanitize public error message public_error = "An internal error occurred during the scan. Please check the logs." db.update_scan_status(scan_id, "failed", error_message=public_error) + finally: + SCAN_DURATION_SECONDS.observe(time.perf_counter() - scan_start) except Exception as exc: logger.error("Worker loop encountered an error: %s", exc) diff --git a/tests/test_observability.py b/tests/test_observability.py new file mode 100644 index 0000000..d0f425e --- /dev/null +++ b/tests/test_observability.py @@ -0,0 +1,168 @@ +"""Tests for the observability layer (issue #159). + +Covers the liveness/readiness probes, the Prometheus /metrics endpoint, +request-ID propagation, request_id in auth-failure responses, and the +conditional Sentry initialisation. +""" + +import uuid +from unittest.mock import MagicMock + +import pytest + +from api.app import create_app +from api.observability import SCANS_TOTAL, init_sentry + + +def _build_app(db_mock, monkeypatch): + """Build a fresh app whose DatabaseManager is replaced by ``db_mock``.""" + monkeypatch.setattr("api.app.DatabaseManager", MagicMock(return_value=db_mock)) + app = create_app() + app.config["TESTING"] = True + return app + + +# --------------------------------------------------------------------------- # +# Liveness / readiness # +# --------------------------------------------------------------------------- # +def test_health_is_public_and_independent_of_db(monkeypatch): + """/health must return 200 without ever touching the database.""" + failing_db = MagicMock() + failing_db.ping.side_effect = Exception("db is down") + app = _build_app(failing_db, monkeypatch) + + resp = app.test_client().get("/health") + + assert resp.status_code == 200 + assert resp.get_json() == {"status": "ok"} + failing_db.ping.assert_not_called() + + +def test_ready_returns_200_when_db_reachable(monkeypatch): + healthy_db = MagicMock() + healthy_db.ping.return_value = True + app = _build_app(healthy_db, monkeypatch) + + resp = app.test_client().get("/ready") + + assert resp.status_code == 200 + assert resp.get_json()["status"] == "ready" + healthy_db.ping.assert_called_once() + + +def test_ready_returns_503_when_db_unavailable(monkeypatch): + failing_db = MagicMock() + failing_db.ping.side_effect = Exception("connection refused") + app = _build_app(failing_db, monkeypatch) + + resp = app.test_client().get("/ready") + + assert resp.status_code == 503 + assert resp.get_json() == {"status": "not_ready", "error": "database_unreachable"} + + +# --------------------------------------------------------------------------- # +# Prometheus metrics # +# --------------------------------------------------------------------------- # +def test_metrics_endpoint_returns_prometheus_text(client): + resp = client.get("/metrics") + + assert resp.status_code == 200 + assert "text/plain" in resp.content_type + body = resp.get_data(as_text=True) + # A registered, unlabelled histogram is always emitted even at zero. + assert "openshield_scan_duration_seconds" in body + + +# --------------------------------------------------------------------------- # +# Request IDs # +# --------------------------------------------------------------------------- # +def test_client_supplied_request_id_is_echoed(client): + supplied = "client-supplied-123" + resp = client.get("/health", headers={"X-Request-ID": supplied}) + + assert resp.headers.get("X-Request-ID") == supplied + + +def test_request_id_is_generated_when_missing(client): + resp = client.get("/health") + + returned = resp.headers.get("X-Request-ID") + assert returned + # A generated ID is a valid UUID. + uuid.UUID(returned) + + +def test_auth_failure_json_includes_request_id(client): + # POST to a protected route without a Bearer token → 401 from JWT middleware. + resp = client.post("/api/scans") + + assert resp.status_code == 401 + payload = resp.get_json() + assert "request_id" in payload + # The body request_id matches the echoed response header. + assert payload["request_id"] == resp.headers.get("X-Request-ID") + + +# --------------------------------------------------------------------------- # +# Sentry # +# --------------------------------------------------------------------------- # +def test_sentry_not_initialized_when_dsn_unset(monkeypatch): + monkeypatch.delenv("SENTRY_DSN", raising=False) + fake_init = MagicMock() + monkeypatch.setattr("sentry_sdk.init", fake_init) + + assert init_sentry() is False + fake_init.assert_not_called() + + +def test_sentry_initialized_when_dsn_set(monkeypatch): + monkeypatch.setenv("SENTRY_DSN", "https://public@o0.ingest.sentry.io/1") + fake_init = MagicMock() + monkeypatch.setattr("sentry_sdk.init", fake_init) + + assert init_sentry() is True + fake_init.assert_called_once() + assert fake_init.call_args.kwargs["dsn"] == "https://public@o0.ingest.sentry.io/1" + + +# --------------------------------------------------------------------------- # +# Worker metrics do not break the worker # +# --------------------------------------------------------------------------- # +def test_worker_records_scan_metrics_on_success(monkeypatch): + """The worker's success path increments the completed scan counter.""" + from scanner import worker + + class _Stop(BaseException): + pass + + scan_id = str(uuid.uuid4()) + sub_id = "00000000-0000-0000-0000-000000000000" + + mock_db = MagicMock() + mock_db.recover_stale_scans.side_effect = [None, _Stop()] + mock_db.claim_next_pending_scan.side_effect = [ + {"scan_id": scan_id, "subscription_id": sub_id}, + None, + ] + mock_engine = MagicMock() + mock_engine.run_scan.return_value = { + "scan_id": scan_id, + "subscription_id": sub_id, + "findings": [], + "total_findings": 0, + "started_at": "2026-01-01T00:00:00Z", + } + + monkeypatch.setattr(worker, "DatabaseManager", MagicMock(return_value=mock_db)) + monkeypatch.setattr(worker, "ScanEngine", MagicMock(return_value=mock_engine)) + monkeypatch.setattr(worker.os.environ, "get", lambda *a, **k: "postgresql://x") + monkeypatch.setattr(worker.time, "sleep", lambda *a, **k: None) + + before = SCANS_TOTAL.labels(status="completed")._value.get() + with pytest.raises(_Stop): + worker.run_worker() + after = SCANS_TOTAL.labels(status="completed")._value.get() + + assert after == before + 1 + mock_db.save_scan.assert_called_once()