# `core/hardening` — API hardening kit

A self-contained package that adds production-grade API hardening to this Flask/uWSGI service:
rate limiting, response compression, security headers, a request body-size limit, a readiness
probe, a thread-safe **circuit breaker** with **per-resource outbound guards**, and Datadog
observability. Everything merges dark (config-gated, safe defaults) and is designed to be lifted
into a shared library once a second service adopts it.

---

## Wiring it into an app — `init_hardening`

```python
from core.hardening import init_hardening

def create_app(config):
    app = Flask(...)
    ...  # DB, marshmallow, blueprints, etc.
    init_hardening(app, config)  # LAST statement -- see ordering below
    return app
```

`init_hardening(app, config)` is the single composition root. It:

1. Calls `validate_config(config)` first, so a bad value fails the boot before any middleware
   installs (see the misconfig table below).
2. Installs **ProxyFix**, so `request.remote_addr` is the real client IP before anything
   downstream (the rate limiter's `principal_key()`) keys on it.
3. Registers the **leak-free error handlers** (`errors.py`) before anything can raise the
   exceptions they catch.
4. Installs the **body limit**, **security headers**, and **compression** (order-independent
   among themselves).
5. Installs **rate limiting last** — `setup_rate_limiting` walks `app.view_functions` to resolve
   `@rate_category` markers, which is only fully populated once every blueprint has registered.
   This is why `init_hardening` itself must be called *after* blueprint registration, as the last
   statement in `create_app`.

### Misconfig table

`validate_config` turns each of these into a boot-time `ValueError` instead of a runtime surprise:

| Bad value | Symptom without validation | Caught by |
|---|---|---|
| `RATELIMIT_MODE` not `shadow`/`enforce` | `setup_rate_limiting` raises deep inside app boot with a less obvious traceback | `RateLimitMode(config.RATELIMIT_MODE)` |
| `RATELIMIT_DEFAULT` empty or malformed | Same — `parse_many` raises during `setup_rate_limiting` | `parse_many(config.RATELIMIT_DEFAULT)` |
| `RATELIMIT_STORAGE_URI` missing a scheme | May not fail until the first request hits the limiter's storage backend | `urlparse(...).scheme` check |
| `MAX_CONTENT_LENGTH_MB` out of `[1, 10]` | Caught separately in `core/config.py`'s `_max_content_length_bytes` at import time | N/A (config module, not this package) |

---

## Inbound HTTP hardening

| Module | What it does |
|---|---|
| `security_headers.py` | App-owned response headers (nosniff, frame `DENY`, referrer policy, COOP, a locked-down CSP) plus an authoritative `Cache-Control` (mutating methods and `@no_store` views are forced to `no-store`). HSTS and stripping the `Server` banner are the edge's job. |
| `compression.py` | gzip/brotli on JSON/text responses via `flask-compress`. `COMPRESS_STREAMS=False` leaves `stream_with_context` exports uncompressed so they aren't buffered into memory. |
| `body_limit.py` | Observes `Content-Length` (feeds the `hardening.request.body_bytes` distribution) and renders a generic 413 on `RequestEntityTooLarge`. The ceiling itself is `MAX_CONTENT_LENGTH`, enforced by Werkzeug. |
| `client_ip.py` | `install_proxyfix` trusts `PROXYFIX_X_FOR` forwarded hops; `principal_key()` resolves the rate-limit bucket (authenticated principal → requestor-service header → IP, in that trust order). |
| `rate_limit.py` | Self-driven, tiered (`@rate_category`) rate limiting with a `shadow`\|`enforce` mode — see **Staged enable order** below. Publishes `X-RateLimit-*` headers in both modes. |
| `errors.py` | The two leak-free handlers: `CircuitBreakerError` → generic 503, rate-limit breach → generic 429 (no resource name or limit-policy string ever reaches the client). |
| `readiness.py` | `is_ready()` — a cached, short-timeout DB probe on a dedicated `NullPool` engine, served at `/ready/` (`core/blueprints/base.py`). |

Both `/hello/` (liveness) and `/ready/` (readiness) are excluded from access/request logging via
`Config.HEALTH_CHECK_PATHS` — they're polled frequently by the load balancer and would otherwise
drown out real request logs.

### Staged enable order

Rate limiting is the one piece with a live rollout gate, driven by `RATELIMIT_MODE`:

1. **`shadow`** (default): every request is checked against its limits, breaches are logged and
   counted (`hardening.ratelimit.rejected`), and `X-RateLimit-*` headers are published — but
   nothing is ever blocked. This is safe to enable in prod immediately; it changes no client
   outcome and lets the caps be validated against real traffic before they can bite.
2. **`enforce`**: breaches now raise, and the client gets a 429. Do not flip this until:
   - the shadow data (Datadog, `hardening.ratelimit.rejected` by category/key_type) confirms the
     caps are sized correctly for real traffic, **and**
   - the storage backend is a shared store (`RATELIMIT_STORAGE_URI=redis://...`), not
     `memory://` — otherwise each uWSGI worker/Fargate task keeps its own counters and a
     per-key limit is effectively multiplied by the pod count, **and**
   - the edge header-trust precondition below is signed off.

### Edge precondition (blocking for `enforce`)

`principal_key()`'s `service` tier reads the client-supplied `Orchard-Requestor-Service` header,
and `install_proxyfix`'s `remote_addr` resolution trusts `PROXYFIX_X_FOR` hops of
`X-Forwarded-For`. Both are spoofable unless the edge (HAProxy at ows-grass) strips and
re-injects `X-Forwarded-For`, `Orchard-Identity-Id`, `Orchard-User-Id`, and
`Orchard-Requestor-Service` so a client cannot forge them. Without that guarantee, per-principal
enforcement is bypassable (mint unlimited keys), escalatable (claim the internal tier), or
weaponizable (set a victim's id to burn their bucket). `PROXYFIX_X_FOR` must also be verified
empirically per environment (curl through the real edge, inspect the raw header — it may differ
qa/uat/prod).

---

## Outbound resilience — guarding downstream calls

Every call to an external dependency (another service, Airflow/MWAA, Snowflake) should go through
`call_downstream`. It wraps the call in that resource's circuit breaker, applies the right timeout,
normalizes failures, and (for idempotent calls) retries with backoff.

```python
from core.hardening.downstream import call_downstream
from core.hardening.resources import Resource

# httpx OwsClient -> ows-abacus-account  (the adapter injects an httpx.Timeout via timeout=)
resp = call_downstream(Resource.OWS_ABACUS_ACCOUNT,
                       lambda timeout: ows_client.get(url, timeout=timeout))

# requests -> MWAA / Airflow  (the adapter injects a (connect, read) tuple)
resp = call_downstream(Resource.AIRFLOW_MWAA,
                       lambda timeout: requests.post(url, json=body, timeout=timeout))

# owsrequest -> ows-collaborator  (no per-call timeout; relies on uWSGI harakiri + the breaker)
resp = call_downstream(Resource.OWS_COLLABORATOR,
                       lambda: owsrequest.get(url))

# Snowflake  (the adapter owns the executor + with-block and passes the executor in)
rows = call_downstream(Resource.SNOWFLAKE, lambda ex: ex.run(sql))

# An idempotent read that may retry (only when that resource's policy sets retries > 0)
data = call_downstream(Resource.OWS_ABACUS_ACCOUNT,
                       lambda timeout: ows_client.get(url, timeout=timeout),
                       idempotent=True)
```

**The `call` lambda's signature must match the resource's adapter** (`timeout=`, no-arg, or `ex`) —
that's the cost of letting each adapter own its client's quirks instead of forcing a uniform
signature. The call returns the client's response on success, or raises:

- **`DownstreamServerError`** — the downstream returned a 5xx.
- **`DownstreamTransportError`** — a timeout or connection failure (normalized from `httpx`/`requests`).
- **`CircuitBreakerError`** — the breaker is open; the call was rejected without running.

Both `Downstream*Error`s count against the resource's breaker; after `fail_max` consecutive failures
the breaker opens and fast-fails for `reset_timeout`, then admits a single probe.

### What a "guard" is

For each resource, a `ResourceGuard` bundles four things, built once at import in a `GuardRegistry`:

| Part | Source | Role |
|---|---|---|
| **policy** | `resources.py` `DOWNSTREAMS` | connect/read timeouts, `fail_max`, `reset_timeout`, `retries` |
| **breaker** | built from the policy | trips on classified failures (lock guards state only; I/O runs concurrently) |
| **adapter** | `guards.py` (or service-specific) | injects the timeout shape; normalizes failures to `Downstream*Error` |
| **classifier** | `guards.py` / service | decides which exceptions count against the breaker |

`call_downstream(resource, call, *, idempotent=False)` = `breaker.call(retry(adapter(call)))`. The
retry is **inside** the breaker, so all `retries+1` physical attempts count as **one** logical
failure, and a half-open probe may itself fan out to several attempts.

### Adding a new downstream

1. Add the resource to `Resource` (`resources.py`) and a `DownstreamPolicy` to `DOWNSTREAMS`.
2. In `downstream.py`, add a `(adapter, classifier)` entry to `_catalog()`. Reuse a generic adapter
   (`httpx_timeout_adapter` for httpx, `requests_timeout_adapter` for requests, `no_timeout_adapter` for
   owsrequest) and `is_downstream_failure`, or write a service-specific one (see `make_snowflake_adapter`).

`build_registry()` raises if any resource is missing a policy or adapter.

---

## Circuit breaker

The breaker is usable directly when you need to guard something other than a registered downstream:

```python
from core.hardening.breaker import CircuitBreaker, CircuitBreakerError

breaker = CircuitBreaker(name='thing', fail_max=5, reset_timeout=30,
                         success_threshold=2, count_failure=lambda e: True,
                         on_state_change=lambda name, old, new: ...)

# explicit form
result = breaker.call(lambda: do_io())

# decorator form (functools.wraps-preserving)
@breaker
def fetch(x):
    return do_io(x)
```

Design notes: the lock guards **only** state transitions, so wrapped I/O and the `on_state_change`
callback run concurrently (pybreaker serializes under its lock; this does not). Half-open admits
exactly one probe via a non-blocking `BoundedSemaphore`. The callback fires **outside** the lock and
is logged (not swallowed) on failure. A throwing `count_failure` classifier can't mask the real
downstream error. See DECISION_LOG D2–D4 (in abacus-docs) for why this is custom, not pybreaker.

### Breaker runbook

- **Reading state:** `observability.on_breaker_state` emits a per-state gauge
  (`hardening.breaker.state`, tagged `resource` + `state`) on every transition — 1 for the state
  just entered, 0 for the one just left. Graph all three states per resource to see open/half-open
  windows over time.
- **A resource's breaker is OPEN:** the downstream is failing past `fail_max` consecutive times.
  Check the downstream's own health/dashboards first — the breaker is a symptom, not the cause.
  It self-recovers: after `reset_timeout` it admits exactly one half-open probe; `success_threshold`
  consecutive successes re-close it, any failure re-opens it for another `reset_timeout`.
  There is no manual reset — restarting the process is the only way to force-clear state, and
  should not be needed under normal recovery.
  Since a `HALF_OPEN` window admits exactly one probe, don't be alarmed by a stretch of continued
  503s while a downstream is down — that is the breaker doing its job (fast-failing instead of
  piling up timeouts on an already-struggling dependency).
- **Tuning a resource's policy:** `fail_max` / `reset_timeout` / `retries` live in `resources.py`
  `DOWNSTREAMS`. `DownstreamPolicy.__post_init__` rejects `reset_timeout < read_timeout` and
  `retries > 0` on a resource whose client already retries internally (owsrequest) — both would
  either never let the breaker actually rest, or silently multiply retries.

---

## Reusing this in another service

The package is split so the mechanism extracts cleanly:

- **`guards.py`** (+ `breaker.py`, `policy.py`) — generic, service-agnostic mechanism: `GuardRegistry`,
  `ResourceGuard`, the generic HTTP adapters, `Downstream*Error`, `is_downstream_failure`, retry, the
  `Policy` Protocol, and the `DownstreamPolicy` type. Imports nothing service-specific — the part that
  lifts into a shared library.
- **`downstream.py`** — this service's **catalog**: `build_registry()` constructs a `GuardRegistry` and
  registers each `Resource` with its policy/adapter/classifier; `call_downstream` is a thin wrapper.
  Service-specific pieces (the `Resource` enum, `DOWNSTREAMS` values, `make_snowflake_adapter`, the
  Snowflake classifier) live here.

To adopt in another service, depend on `guards.py` (eventually the shared lib) and write your own
small catalog factory:

```python
from hardening.guards import GuardRegistry, httpx_timeout_adapter, is_downstream_failure

def build_registry():
    r = GuardRegistry(on_state_change=emit_metric)
    r.register("payments", payments_policy, httpx_timeout_adapter, is_downstream_failure)
    r.register("warehouse", wh_policy, my_warehouse_adapter, my_warehouse_classifier)
    return r

REGISTRY = build_registry()  # or stash on app.extensions via init_hardening

def call_downstream(name, call, *, idempotent=False, registry=None):
    return (registry or REGISTRY).call(name, call, idempotent=idempotent)
```

The resource identity is any hashable you choose (an enum, a string) — the registry doesn't own the
catalog, you do.

`init_hardening(app, config)` (the inbound half) is Flask-specific and depends on this service's
own `rate_policy.py`/`config.py`, so it isn't split out the same way yet — lifting it would mean
extracting `rate_policy.RATE_CATEGORIES` and the config attribute names as adopter-supplied inputs.

---

## Status

Built and active (behind config, safe defaults): circuit breaker + outbound guards, security
headers, compression, body limit, rate limiting (shadow), readiness, observability, the
`init_hardening` composition root, and the uWSGI robustness flags (`harakiri`, `reload-mercy`,
`die-on-term`, etc. — `uwsgi-start.sh`).

Gated on follow-on work: rate-limit `enforce` mode (needs the shared Redis store and the edge
header-trust precondition, both described above).