fix(api): rate-limit /ready and /metrics, expose db pool telemetry

CHANGELOG.md

@@ -25,12 +25,14 @@ OpenShield uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 - Harvest Now Decrypt Later exposure window calculation per cryptographic asset
 - Deterministic Render deployment workflow for separate API and worker services
 - Terraform configuration for Render, Vercel, and GitHub OIDC
+- Database connection pool utilization and exhaustion telemetry (`openshield_db_pool_connections_*`) on `/metrics`
 
 ### Fixed
 
 - High-severity CodeQL findings in Python and JavaScript code
 - Security findings identified during Semgrep analysis
 - Sensitive identity metadata removed from scanner debug logging
+- `/ready` and `/metrics` rate-limited per source IP so the unauthenticated probe/scrape surface can no longer be used to exhaust the database connection pool
 
 ### Security
 

api/app.py

@@ -10,8 +10,15 @@
 from flask_cors import CORS
 from werkzeug.middleware.proxy_fix import ProxyFix
 
-from api.models.finding import DatabaseManager
-from api.observability import configure_logging, get_request_id, init_app, init_sentry
+from api.models.finding import DatabaseManager, get_pool_stats
+from api.observability import (
+    configure_logging,
+    get_request_id,
+    init_app,
+    init_sentry,
+    probe_rate_limit,
+    set_pool_stats_provider,
+)
 
 load_dotenv()
 
@@ -42,6 +49,14 @@
 _KNOWN_ROLES = {"viewer", "operator", "admin"}
 _WRITE_ROLES = {"operator", "admin"}
 
+# Generous enough for legitimate manual or automated readiness checks from
+# one source, but bounded well under the default pool size
+# (DB_POOL_MAX_CONN=10) so a single caller can never claim more than half
+# the pool's capacity by itself, even if every allowed request in the
+# window lands at once. See probe_rate_limit's docstring for why this is
+# in-memory rather than the shared Postgres-backed rate_limit().
+_READY_MAX_REQUESTS_PER_WINDOW = 5
+
 
 def _is_production() -> bool:
     return (
@@ -121,6 +136,7 @@ def create_app() -> Flask:
     # every later before_request handler (including JWT auth) and to the
     # error handlers. Also mounts the public /metrics endpoint.
     init_app(app)
+    set_pool_stats_provider(get_pool_stats)
 
     # ------------------------------------------------------------------ #
     # Configuration & Security                                             #
@@ -269,6 +285,7 @@ def health():
         return jsonify({"status": "ok"})
 
     @app.get("/ready")
+    @probe_rate_limit(_READY_MAX_REQUESTS_PER_WINDOW)
     def ready():
         """Readiness probe: 200 when the database is reachable, else 503."""
         try:

api/models/finding.py

@@ -46,6 +46,44 @@ def _get_pool(dsn: str) -> "psycopg2.pool.ThreadedConnectionPool":
         return pool
 
 
+def get_pool_stats(dsn: Optional[str] = None) -> Dict[str, Any]:
+    """Return a point-in-time snapshot of the shared connection pool's utilization.
+
+    Reports only counts (in-use / idle / configured maximum) - never the DSN,
+    host, credentials, or anything else that could describe the database
+    deployment. Safe to expose on a public surface (Prometheus /metrics, the
+    /ready probe) precisely because it carries no operational secrets, only
+    capacity numbers an operator needs to see the pool approaching exhaustion
+    before it happens.
+
+    Returns zeroed stats with no pool created yet (no request has connected
+    since process start) rather than raising, so callers on the request path
+    (like /ready) never fail because of a stats lookup.
+    """
+    dsn = dsn or os.environ.get("DATABASE_URL", "")
+    with _POOLS_LOCK:
+        pool = _POOLS.get(dsn)
+
+    if pool is None:
+        return {"max_connections": _POOL_MAX_CONN, "in_use": 0, "idle": 0, "utilization_percent": 0.0}
+
+    # psycopg2's pool has no public stats API; _used/_pool/maxconn are the
+    # same attributes getconn()/putconn() themselves mutate under this same
+    # lock, so a snapshot taken while holding it can't land mid-mutation.
+    with pool._lock:
+        in_use = len(pool._used)
+        idle = len(pool._pool)
+        max_connections = pool.maxconn
+
+    utilization_percent = round((in_use / max_connections) * 100, 1) if max_connections else 0.0
+    return {
+        "max_connections": max_connections,
+        "in_use": in_use,
+        "idle": idle,
+        "utilization_percent": utilization_percent,
+    }
+
+
 FRAMEWORK_FILE_MAP = {
     "cis": "cis_azure_benchmark.json",
     "nist": "nist_csf.json",

api/observability.py

@@ -14,10 +14,14 @@
 import logging
 import os
 import re
+import threading
 import time
 import uuid
+from collections import OrderedDict, deque
+from functools import wraps
+from typing import Callable, Deque, Dict, Optional, Tuple
 
-from flask import Flask, Response, g, request
+from flask import Flask, Response, current_app, g, jsonify, request
 from prometheus_client import (
     CONTENT_TYPE_LATEST,
     Counter,
@@ -82,6 +86,49 @@
     "Latency of outbound LLM provider requests in seconds.",
     ["provider"],
 )
+DB_POOL_CONNECTIONS_IN_USE = Gauge(
+    "openshield_db_pool_connections_in_use",
+    "Connections currently checked out from the shared database pool.",
+)
+DB_POOL_CONNECTIONS_IDLE = Gauge(
+    "openshield_db_pool_connections_idle",
+    "Connections currently idle in the shared database pool.",
+)
+DB_POOL_CONNECTIONS_MAX = Gauge(
+    "openshield_db_pool_connections_max",
+    "Configured maximum size of the shared database pool.",
+)
+
+# Set by api.app at startup (see set_pool_stats_provider). Kept as a plain
+# module-level callable rather than importing api.models.finding directly -
+# this module stays free of project imports (see module docstring) so it
+# can be reused from the worker without pulling in the API's DB layer.
+_pool_stats_provider: Optional[Callable[[], Dict[str, int]]] = None
+
+
+def set_pool_stats_provider(fn: Callable[[], Dict[str, int]]) -> None:
+    """Register the callable /metrics uses to refresh the DB pool gauges.
+
+    Called lazily on every scrape (see ``metrics()`` below) rather than on a
+    timer, so the numbers are exact at scrape time instead of aging between
+    scrapes.
+    """
+    global _pool_stats_provider
+    _pool_stats_provider = fn
+
+
+def _refresh_pool_metrics() -> None:
+    if _pool_stats_provider is None:
+        return
+    try:
+        stats = _pool_stats_provider()
+        DB_POOL_CONNECTIONS_IN_USE.set(stats["in_use"])
+        DB_POOL_CONNECTIONS_IDLE.set(stats["idle"])
+        DB_POOL_CONNECTIONS_MAX.set(stats["max_connections"])
+    except Exception:
+        # A stats lookup must never take /metrics down - the rest of the
+        # scrape (HTTP/scan counters etc.) is still valid without it.
+        logger.exception("Failed to refresh database pool metrics")
 
 
 # --------------------------------------------------------------------------- #
@@ -150,6 +197,133 @@ def get_request_id() -> str:
     return rid
 
 
+# --------------------------------------------------------------------------- #
+# Probe/scrape rate limiting                                                   #
+# --------------------------------------------------------------------------- #
+# /health, /ready, and /metrics are exempt from JWT auth by design (see
+# api.app._ALWAYS_PUBLIC) so uptime checkers and Prometheus scrapers can
+# reach them without a token. That also makes /ready and /metrics the one
+# place an unauthenticated caller can trigger repeated backend work - a
+# pooled database checkout for /ready - without ever presenting a token.
+#
+# api.rate_limit.rate_limit (used elsewhere in the API) is the wrong tool
+# here: it does its own Postgres round trip per check, which would add
+# database load to /ready - the exact endpoint whose job is to protect the
+# database pool from overload - and risks checking out a second, separate
+# pooled connection under its own `g.db` alongside the one /ready's handler
+# already manages, only one of which the teardown handler would return.
+#
+# This limiter is pure in-memory and per-process instead. The connection
+# pool it protects (api.models.finding._POOLS) is itself process-local under
+# Gunicorn's multi-worker model, so a per-process budget is the matching
+# granularity, not a weaker substitute for a shared one: it bounds exactly
+# the pool a single worker process can exhaust. It is a defense-in-depth
+# backstop, not a replacement for restricting these paths at whatever
+# reverse proxy/CDN/WAF fronts the deployment - see the "Restricting
+# probe/scrape endpoints at the edge" section of docs/deployment/render.md.
+_PROBE_WINDOW_SECONDS = 10.0
+# Prometheus scrapes on a fixed interval (typically 15-30s) from a small,
+# stable set of scraper IPs, so this stays generous; it exists to blunt a
+# single caller hammering the endpoint, not to constrain normal scraping.
+_METRICS_MAX_REQUESTS_PER_WINDOW = 20
+# Hard ceiling on distinct (address, path) keys tracked at once. Same-key
+# cleanup alone isn't enough: a caller that continuously rotates its source
+# address (or a spoofed forwarded address wherever the trusted-proxy
+# boundary is misconfigured) creates a new one-shot dict entry per address,
+# and a key that's never revisited is never pruned by the per-call cleanup
+# below - making the tracking dict itself an unbounded memory sink. This
+# caps it regardless of how many distinct addresses show up.
+_PROBE_MAX_TRACKED_KEYS = 10_000
+# A full scan over every tracked key on every single request would undercut
+# the point of a cheap in-memory limiter, so the sweep below runs every Nth
+# call instead of every call. This bounds how long a key whose owner never
+# returns can survive to at most this many requests' worth of accumulation
+# - the hard cap above is what actually bounds worst-case memory regardless
+# of traffic shape or how the sweep is scheduled.
+_PROBE_SWEEP_INTERVAL = 200
+_probe_lock = threading.Lock()
+# An OrderedDict, not a plain dict: every touch (new hit or a sweep pruning
+# it survives) moves a key to the end, so the front is always the least
+# recently touched key - the correct, deterministic thing to evict first
+# when the hard cap above is reached.
+_probe_hits: "OrderedDict[Tuple[str, str], Deque[float]]" = OrderedDict()
+_probe_call_count = 0
+
+
+def _sweep_expired_probe_hits(now: float, window_seconds: float) -> None:
+    """Remove every tracked key whose hits have all expired.
+
+    Must be called with _probe_lock already held. This is the global
+    cleanup pass: the per-call logic in probe_rate_limit only ever prunes
+    the one key the current request touched, which does nothing for a
+    key that's never hit again.
+    """
+    cutoff = now - window_seconds
+    for key in list(_probe_hits.keys()):
+        hits = _probe_hits[key]
+        while hits and hits[0] < cutoff:
+            hits.popleft()
+        if not hits:
+            del _probe_hits[key]
+
+
+def probe_rate_limit(max_requests: int, window_seconds: float = _PROBE_WINDOW_SECONDS):
+    """Limit a probe/scrape view to ``max_requests`` per ``window_seconds`` per client IP.
+
+    The check runs before the wrapped view body, so a request rejected here
+    never reaches the database work it would otherwise trigger. Disabled in
+    testing mode, matching the convention in api.rate_limit.rate_limit.
+    """
+
+    def decorator(fn):
+        @wraps(fn)
+        def wrapped(*args, **kwargs):
+            if current_app.testing:
+                return fn(*args, **kwargs)
+
+            global _probe_call_count
+            key = (request.remote_addr or "unknown", request.path)
+            now = time.monotonic()
+            with _probe_lock:
+                _probe_call_count += 1
+                if _probe_call_count % _PROBE_SWEEP_INTERVAL == 0:
+                    _sweep_expired_probe_hits(now, window_seconds)
+
+                hits = _probe_hits.get(key)
+                if hits is not None:
+                    cutoff = now - window_seconds
+                    while hits and hits[0] < cutoff:
+                        hits.popleft()
+                    if not hits:
+                        del _probe_hits[key]
+                        hits = None
+
+                if hits is None:
+                    if len(_probe_hits) >= _PROBE_MAX_TRACKED_KEYS:
+                        # Deterministic eviction: drop the least recently
+                        # touched key, not an arbitrary/insertion-order one.
+                        _probe_hits.popitem(last=False)
+                    hits = deque()
+
+                allowed = len(hits) < max_requests
+                if allowed:
+                    hits.append(now)
+                # Re-inserting (or just touching) moves this key to the end -
+                # it's the most recently active key, so it's the last thing
+                # eviction should reach for.
+                _probe_hits[key] = hits
+                _probe_hits.move_to_end(key)
+
+            if not allowed:
+                return jsonify({"status": "rate_limited"}), 429, {"Retry-After": str(int(window_seconds))}
+
+            return fn(*args, **kwargs)
+
+        return wrapped
+
+    return decorator
+
+
 # --------------------------------------------------------------------------- #
 # Flask wiring                                                                 #
 # --------------------------------------------------------------------------- #
@@ -179,5 +353,7 @@ def _record_observability(response: Response) -> Response:
         return response
 
     @app.get("/metrics")
+    @probe_rate_limit(_METRICS_MAX_REQUESTS_PER_WINDOW)
     def metrics() -> Response:
+        _refresh_pool_metrics()
         return Response(generate_latest(), content_type=CONTENT_TYPE_LATEST)

docs/deployment/render.md

@@ -69,6 +69,33 @@ the Flask application. Leaving this value unset enables wildcard CORS and is not
 acceptable for a real staging or production environment. Do not add this setting to
 worker services.
 
+## Restricting probe/scrape endpoints at the edge
+
+`/health`, `/ready`, and `/metrics` are intentionally exempt from JWT auth (see
+`api.app._ALWAYS_PUBLIC`) so uptime checkers and Prometheus scrapers can reach them
+without a token. `/health` is a pure liveness check with no backend dependency and is
+meant to stay reachable from anywhere - it is what `render.yaml`'s
+`healthCheckPath` uses. `/ready` and `/metrics` are different: `/ready` checks out a
+pooled database connection on every call, and `/metrics` returns operational counters
+(including database pool utilization from `openshield_db_pool_connections_*` -
+counts only, never the DSN, host, or credentials - see `get_pool_stats()` in
+`api/models/finding.py`). Neither should be reachable by arbitrary internet clients.
+
+Render's Blueprint format has no path-based access control, so this repository
+cannot restrict `/ready`/`/metrics` from `render.yaml` itself. The application
+carries its own in-process, in-memory rate limit on both paths
+(`api.observability.probe_rate_limit`, wired in `api/app.py`) as a defense-in-depth
+backstop - deliberately not the shared Postgres-backed `api.rate_limit.rate_limit`
+used elsewhere, since that would add database load to the exact endpoint meant to
+protect the database from overload. That backstop bounds a single source hammering
+one process; it is not a substitute for restricting network reachability.
+
+Whoever operates the reverse proxy, CDN, or WAF in front of a Render deployment
+(Cloudflare or similar, if one is configured) should restrict `/ready` and `/metrics`
+to known monitoring/scraping source IPs or an internal network path, the same way
+they would for any other backend-only operational endpoint. This is an operational
+configuration step outside this repository, not something `render.yaml` can express.
+
 ## Coordinated deterministic deployment
 
 For the selected environment, the workflow: