coulomb/qonto-assistant
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add SecurityPractice.md, Security Genome record, and deny-escalation lockout

Browse source 
Design doc for hardening qonto-assistant before deployment to
railiance01: this is the first fleet service that must be
internet-reachable (external harness clients, not just in-cluster
jobs) while holding a real bank credential. Covers identity (key-cape
in place of the interim bearer token), authorization (finance.qonto.read
in flex-auth + tenant-engine capability roles instead of the
hardcoded default_tenant_id), network exposure (facade-only internet
address), isolation profile, and a Kings Guard mapping (the existing
audit stream is already Immune-Observation-shaped; nothing to rebuild
later).

Ships one concrete, dependency-free piece of that design now:
DenyEscalationTracker locks out an actor who repeatedly triggers
arg_constraint/credential_exfil denies within a short window, closing
the gap where a probing client could retry indefinitely at whatever
rate the existing rate limiter otherwise allows. Wired through
CapabilityService, on by default, configurable via
QONTO_DENY_ESCALATION_* env vars. Ordinary denies (authz_denied,
tenant_scope) never count toward it.

Also adds specs/security-genome.yaml (kings-guard's genome-record
shape, populated now so no rework is needed once a consumer exists).

Verified: pytest -> 39 passed (8 new); REST and MCP smoke scripts both
pass against fixtures; compileall clean.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-07-23 22:59:06 +02:00
parent 78eb2a819c
commit 3faf1fed71
9 changed files with 655 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

246
docs/SecurityPractice.md Normal file
View file

@ -0,0 +1,246 @@
# Security Practice — Internet Exposure, Governance, and Scale-to-Zero
> Status: draft, 2026-07-23. Written before deployment to `railiance01`.
> Context: this is the first fleet service that (a) must be reachable by
> clients outside the cluster (laptop-based Claude Code/Cursor/Codex
> sessions), and (b) is the sole holder of a real company bank credential.
> That combination raises the risk class above every other agent-facing
> service shipped so far (`llm-connect`, mail-triage), which are
> cluster-internal only.
Related: `specs/ArchitectureBlueprint.md` (service architecture and v1
policy), `docs/mcp-integration.md` (current auth model and its called-out
gaps), `docs/operator-runbook.md` (day-2 operation).
This document assumes the [`kings-guard`](../../kings-guard) NetKingdom
Immune Architecture exists as designed in
`kings-guard/specs/NetKingdomImmuneArchitecture.md` (currently draft, no
implementation) and describes how `qonto-assistant` would sit inside it,
alongside the real, already-running NetKingdom components (`key-cape`,
`flex-auth`, `tenant-engine`, OpenBao/`ops-warden`). Concrete requirements
handed to `kings-guard` to prioritize its own build are tracked as an
intake — see `KG-WP-0002` and the corresponding intake record.
---
## 1. Why this service is a different risk class
| Property | Prior services (`llm-connect`, mail-triage) | `qonto-assistant` |
| --- | --- | --- |
| Client population | in-cluster jobs/timers only | in-cluster **and** external harness sessions (laptops) |
| Network exposure | cluster-internal DNS only | needs a real internet-reachable address |
| Credential held | none / low-value | live company bank API key (`tenants/binky/qonto-api`) |
| Data classification | operational metadata | tenant-confidential financial data |
| Blast radius if compromised | mail queue drift | real financial data disclosure; reputational and possibly regulatory impact |
Every control below exists to answer one question: **what stops an
internet-reachable process, holding a real bank credential, from becoming
the fleet's first genuinely damaging incident?**
---
## 2. Layered design
```text
Internet
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Facade / Activator (always-on, holds NO bank credential) │
│ - key-cape token verification │
│ - flex-auth pre-check (finance.qonto.read) │
│ - wakes the real service only after both pass │
│ - idle-timeout scale-back-to-zero │
└───────────────────────────┬───────────────────────────────────┘
│ (only on authenticated+authorized wake)
▼
┌─────────────────────────────────────────────────────────────┐
│ qonto-assistant (scaled 0↔1, isolation profile I1/I2) │
│ - policy kernel (default-deny; existing, unchanged) │
│ - REST + MCP surfaces (existing, unchanged) │
│ - OpenBao fetch on cold start only, short TTL │
│ - audit emission shaped as Kings Guard Immune Observations │
└───────────────────────────┬───────────────────────────────────┘
│
▼
OpenBao (tenants/binky/qonto-api)
│
▼
Qonto thirdparty API
```
The facade and the scale-to-zero controller are **the same component**.
Building it once satisfies both the cost concern (service is idle most of
the time and should not burn compute) and the security concern (nothing
reachable from the internet is allowed to trigger a cold start of the
credential-holding process without first clearing identity and
authorization checks). This is deliberate: a "wake on any request" facade
without the auth gate would just be a new, free DoS/cost-exhaustion lever.
---
## 3. Identity — replace the interim bearer token with `key-cape`
`docs/mcp-integration.md` already documents today's interim state: a
shared-secret bearer token (`QONTO_ASSISTANT_MCP_TOKEN`) stands in for real
workload/client identity, explicitly because "no OIDC issuer exists in this
fleet yet."
That premise no longer holds for this deployment. `key-cape`
(Authelia + LLDAP + privacyIDEA) is the fleet's real, already-running IAM
Profile issuer — it is SSO today for ops/Temporal UI (`NK-WP-0021`). Before
`qonto-assistant` is internet-facing:
- The facade verifies a `key-cape`-issued IAM Profile token (human operator
session or service/workload identity) on every inbound request, before
any wake decision.
- The shared-secret `QONTO_ASSISTANT_MCP_TOKEN` becomes fixture/local-dev
only (as already documented) and must never be the auth boundary for a
real deployment.
- Actor claims (`X-Actor-ID`, `X-Tenant-ID`, `X-Actor-Lane`, `X-Actor-Scopes`)
stop being self-asserted headers and are instead derived from the
verified token — closing the gap `docs/mcp-integration.md` §Auth model
calls out explicitly ("self-asserted today, not cryptographically bound").
## 4. Authorization — register `finance.qonto.read` in `flex-auth` now
`config.py` already reserves `QONTO_ASSISTANT_REQUIRED_SCOPE` /
`QONTO_ASSISTANT_ENFORCE_SCOPE` for exactly this and ships with enforcement
off. Given internet exposure, this should not wait for "Phase 3" — it is
the actual authorization control point standing between an authenticated
caller and a live bank-data read.
- Register `finance.qonto.read` (and, if useful later, a narrower
`finance.qonto.read.transactions` / `.export`) as a `flex-auth` resource.
- The facade (or the assistant itself, on cold-start request) calls
`flex-auth` for a live decision rather than trusting a cached claim for
this resource class — mirroring the same reasoning `tenant-engine`
applies to itself: *"a stale `VEN` grant surviving a plan cancellation is
not an acceptable risk for money-movement or credential-vending
actions"* — read-only finance visibility deserves the same live-check
discipline, not just write paths.
- Gate on **live tenant capability role** via `tenant-engine`
(`VEN`/`CUS`, non-exclusive) and plan, replacing the current hardcoded
`default_tenant_id="binky"` assumption in `config.py`. A tenant whose
plan lapses should lose read access the same request cycle, not whenever
a cache expires.
## 5. Network exposure — the facade is the only internet-facing thing
`qonto-assistant`'s raw REST/MCP port must never be bound to a
publicly-reachable address, in any deployment. Concretely:
- The Kubernetes `Service`/ingress in front of `qonto-assistant` itself
should remain cluster-internal (`ClusterIP`), matching the pattern
`railiance-rhythm`'s host timers already use for `llm-connect`.
- The facade is the sole component with an external address. It is
deliberately dumb: no policy kernel, no bank credential, no Qonto client
— its only job is auth-gate, wake, proxy, and idle-timeout.
- Failure mode: if `key-cape` or `flex-auth` is unreachable, the facade
**fails closed** (refuses to wake the backend), per Kings Guard principle
6.11 ("fail securely, not blindly") — an outage in the auth path must not
silently become an open-access mode.
## 6. Scale-to-zero — cost control and security control together
- Facade holds the public address; backend `Deployment` runs at `replicas:
0` when idle.
- On an authenticated+authorized request, facade scales the backend to 1,
waits on `GET /v1/health`, then proxies through.
- An idle-timeout watcher (same component or a sidecar) scales back to 0
after a configurable window (proposed default: 10–15 minutes of no
traffic) — tunable per deployment, not hardcoded.
- If the target cluster already runs Knative Serving, this is close to the
built-in Activator + scale-to-zero autoscaler pattern and should reuse it
rather than reinventing one. Otherwise a minimal custom
controller is small and narrow in scope (one Deployment, one Service,
one wake/idle state machine) and should be scoped as its own follow-on
task rather than folded into `qonto-assistant`'s own codebase — see
`QONTO-WP-0004`.
- Security side effect, not just a cost one: because OpenBao secret fetch
already uses a short TTL (`qonto_secret_ttl_seconds`), scale-to-zero
shrinks the bank credential's residency window in process memory to
"only while an authenticated request is actually being served" — smaller
than today's always-on posture, for free.
## 7. Isolation profile
Per `kings-guard`'s isolation-profile classification
(`NetKingdomImmuneArchitecture.md` §11), given:
- tenant-confidential financial data classification,
- internet reachability (via the facade),
- sole custody of a real bank credential,
`qonto-assistant` should run at **I1 Reinforced** at minimum (dedicated
node pool or sandboxed runtime) rather than the shared/logical-isolation
default (I0) other low-risk internal services use. **I2 Dedicated**
(tenant-specific control/compute plane) is worth considering given it is
the single component in the fleet holding this particular credential — the
decision belongs to whoever owns Railiance placement for this workload, not
to this repo alone.
## 8. Recovery
`qonto-assistant` already has the properties Kings Guard's "recovery is a
first-class capability" principle (AD-006) asks for, largely as a side
effect of how it was built, not extra work:
- **Stateless** — no persisted secrets or session state; redeployable at
any time.
- **Short-TTL credential** — OpenBao lease expires quickly; a fresh pod
fetches its own, it never inherits a stale one.
- **Cheap reconstitution** — "kill the pod, let the facade cold-start a
fresh one on the next legitimate request" is a complete, low-cost
recovery procedure with no data-loss risk (`maximum_data_loss: PT0S`
applies trivially since there is no persisted data).
## 9. Kings Guard mapping (prep now, cheap; enforcement later)
Nothing below requires `kings-guard` to exist yet. It is preparation so
`qonto-assistant` needs zero rework once a sentinel-mesh or decision plane
does exist.
### 9.1 Security Genome record
See `specs/security-genome.yaml` for the concrete record following
`NetKingdomImmuneArchitecture.md` §9.1's schema — declared purpose,
capabilities provided/consumed, expected egress (Qonto API + OpenBao only,
nothing else), data classification, and recovery expectations.
### 9.2 Audit stream is already observation-shaped
`AuditLogger`'s existing event shape (actor, capability, decision,
deny_reason, latency, upstream HTTP status, policy version — see
`tests/test_audit.py`, `tests/test_audit_parity.py`) already matches Kings
Guard's Immune Observation contract closely enough that no schema rework
should be needed later — just a new consumer pointed at the same stream.
### 9.3 A concrete, actionable-today signal
This does not require any Kings Guard component: repeated
`arg_constraint` or `credential_exfil` deny reasons from the same actor
within a short window is a real, current signal. It should trip a
tightened rate limit or a temporary lockout for that actor now, using the
audit stream and rate limiter that already exist — this is a Fast Local
Loop (`NetKingdomImmuneArchitecture.md` §14.1) response that does not need
to wait for any future component.
---
## 10. What this repo can do unilaterally vs. what it depends on
| Item | Owner | Status |
| --- | --- | --- |
| Security Genome record | `qonto-assistant` | can ship now |
| Audit-stream shape review against Immune Observation contract | `qonto-assistant` | can ship now |
| Actor lockout on repeated deny signals | `qonto-assistant` | can ship now |
| `key-cape` token verification in place of bearer token | `qonto-assistant` + `key-cape` | needs `key-cape` client integration support |
| `finance.qonto.read` resource + live decision call | `qonto-assistant` + `flex-auth` | needs the resource registered in `flex-auth` |
| Live tenant-role gate | `qonto-assistant` + `tenant-engine` | needs a `tenant-engine` lookup/cache API call wired in |
| Facade / scale-to-zero activator | new component (home TBD — Railiance or a dedicated repo) | design only so far |
| I1/I2 isolation placement on `railiance01` | Railiance | needs a placement decision |
| Sentinel-mesh / decision-plane consumption of the audit stream | `kings-guard` | does not exist yet — see intake |
Tracked as `QONTO-WP-0004` in this repo, with the `kings-guard`-owned
portion tracked as an intake against `KG-WP-0002` (pilot-lane selection).

13
docs/operator-runbook.md
View file

@ -206,3 +206,16 @@ That consumer-side write remains outside this repo.
- The service supports a `bearer` auth mode for future upstream evolution, but
the current dogfood path remains `legacy_api_key` because that is the proven
BINKY-WP-0005 header mode.
## Deny-escalation lockout
On by default (`QONTO_DENY_ESCALATION_ENABLED=true`). An actor who triggers
`arg_constraint` or `credential_exfil` policy denials `QONTO_DENY_ESCALATION_THRESHOLD`
times (default 3) within `QONTO_DENY_ESCALATION_WINDOW_SECONDS` (default 60s)
is locked out for `QONTO_DENY_ESCALATION_LOCKOUT_SECONDS` (default 300s) —
every request from that actor is rejected with `actor_locked_out` before the
policy kernel is even consulted, regardless of which capability they call
next. Ordinary denies (`authz_denied`, `tenant_scope`, `unknown_capability`)
never count toward this — only the two reason classes that indicate
probing/exfiltration rather than a client mistake. See
`docs/SecurityPractice.md` §9.3 and `src/qonto_assistant/security_watch.py`.

86
specs/security-genome.yaml Normal file
View file

@ -0,0 +1,86 @@
# Security Genome record for qonto-assistant.
#
# Schema per kings-guard/specs/NetKingdomImmuneArchitecture.md §9.1
# ("Minimum Genome Record"). Kings Guard does not exist as a running system
# yet — this record is written now so no rework is needed once an admission
# or posture-assessment consumer does exist. See docs/SecurityPractice.md.
security_genome_record:
id: kg:genome:qonto-assistant
version: 0.1.0
ownership:
tenant_id: binky
accountable_owner: bernd.worsch
operational_owner: qonto-assistant
purpose:
intent: >
Sole holder of the company Qonto bank API credential; exposes
read-only finance capabilities (org summary, transactions,
CostRunRate hints) to authorized agent harnesses and operators via a
governed REST + MCP surface, so no client ever holds the bank key
directly.
criticality: high
identities:
workload_identity: "TBD — pending key-cape/workload-identity integration (QONTO-WP-0004)"
deployment_identity: "TBD — pending Railiance placement decision"
capabilities:
provides:
- finance.qonto.read
consumes:
- openbao-secret.read # tenants/binky/qonto-api
- qonto-thirdparty-api.read
communication:
ingress:
- caller: facade-activator # not yet built — see docs/SecurityPractice.md §6
protocol: https
egress:
- destination: openbao
protocol: https
- destination: qonto-thirdparty-api
protocol: https
# No other egress is expected or permitted. Any additional outbound
# destination observed at runtime is, by definition, a deviation from
# this genome.
data:
classifications:
- tenant-confidential
- financial
persistence: ephemeral
export_allowed: false
provenance:
signed_artifact_required: false # not yet implemented fleet-wide
approved_registry: "TBD"
sbom_required: false # not yet implemented; tracked as a gap
reproducible_build_preferred: true
expected_behavior:
max_request_rate: "per QONTO_RATE_LIMIT_REQUESTS / QONTO_RATE_LIMIT_WINDOW_SECONDS (config.py)"
external_network_access: false # beyond the two declared egress destinations
interactive_shell: prohibited
privilege_escalation: prohibited
recovery:
redeployable: true
maximum_recovery_time: PT5M # stateless; cold-start dominated by scale-to-zero wake latency
maximum_data_loss: PT0S # no persisted state
tolerances:
- id: local-dev-shared-secret-auth
description: >
QONTO_ASSISTANT_MCP_TOKEN shared-secret bearer auth is tolerated
only for fixture-backed local development, never for a deployment
holding real Qonto credentials.
expires_at: null # revisit once key-cape integration (QONTO-WP-0004) lands
- id: self-asserted-actor-claims
description: >
X-Actor-* headers are self-asserted, not yet cryptographically
bound to verified identity. Tolerated until key-cape/flex-auth
integration (QONTO-WP-0004) closes this.
expires_at: null

10
src/qonto_assistant/app.py
View file

@ -21,6 +21,7 @@ from qonto_assistant.mcp_server import create_mcp_server
from qonto_assistant.policy import PolicyEngine
from qonto_assistant.qonto_client import FixtureQontoClient, QontoClient
from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter
from qonto_assistant.security_watch import DenyEscalationTracker
from qonto_assistant.service import CapabilityService
@ -171,6 +172,15 @@ def _build_service(
),
concurrency_limiter=concurrency_limiter
or ConcurrencyLimiter(limit=settings.max_concurrency),
deny_escalation_tracker=(
DenyEscalationTracker(
threshold=settings.deny_escalation_threshold,
window_seconds=settings.deny_escalation_window_seconds,
lockout_seconds=settings.deny_escalation_lockout_seconds,
)
if settings.deny_escalation_enabled
else None
),
)

8
src/qonto_assistant/config.py
View file

@ -28,6 +28,10 @@ class Settings:
rate_limit_requests: int
rate_limit_window_seconds: int
max_concurrency: int
deny_escalation_enabled: bool
deny_escalation_threshold: int
deny_escalation_window_seconds: int
deny_escalation_lockout_seconds: int
credential_source: str
openbao_path: str
openbao_command: str
@ -66,6 +70,10 @@ class Settings:
rate_limit_requests=int(os.getenv("QONTO_RATE_LIMIT_REQUESTS", "20")),
rate_limit_window_seconds=int(os.getenv("QONTO_RATE_LIMIT_WINDOW_SECONDS", "60")),
max_concurrency=int(os.getenv("QONTO_MAX_CONCURRENCY", "4")),
deny_escalation_enabled=os.getenv("QONTO_DENY_ESCALATION_ENABLED", "true").lower() == "true",
deny_escalation_threshold=int(os.getenv("QONTO_DENY_ESCALATION_THRESHOLD", "3")),
deny_escalation_window_seconds=int(os.getenv("QONTO_DENY_ESCALATION_WINDOW_SECONDS", "60")),
deny_escalation_lockout_seconds=int(os.getenv("QONTO_DENY_ESCALATION_LOCKOUT_SECONDS", "300")),
credential_source=os.getenv("QONTO_CREDENTIAL_SOURCE", "env"),
openbao_path=os.getenv("QONTO_OPENBAO_PATH", "tenants/binky/qonto-api"),
openbao_command=os.getenv("QONTO_OPENBAO_COMMAND", "bao"),

85
src/qonto_assistant/security_watch.py Normal file
View file

@ -0,0 +1,85 @@
from __future__ import annotations
import time
from collections import defaultdict, deque
from collections.abc import Callable
from threading import Lock
from qonto_assistant.errors import QontoAssistantError
# Deny reasons that indicate probing or exfiltration attempts rather than
# ordinary client mistakes (e.g. a too-large page_size is `arg_constraint`
# but so is a scripted attempt to walk every constraint boundary; treating
# it as escalation-worthy is deliberately conservative). See
# docs/SecurityPractice.md §9.3 — this is a Fast Local Loop response
# (NetKingdomImmuneArchitecture.md §14.1) that needs no external component.
ESCALATING_DENY_REASONS = frozenset({"arg_constraint", "credential_exfil"})
class ActorLockedOutError(QontoAssistantError):
"""Raised when an actor is temporarily locked out after repeated
escalation-worthy policy denials."""
error_code = "actor_locked_out"
status_code = 429
def __init__(self, actor_key: str, *, retry_after_seconds: float) -> None:
super().__init__(f"actor {actor_key} locked out for {retry_after_seconds:.0f}s")
self.actor_key = actor_key
self.retry_after_seconds = retry_after_seconds
class DenyEscalationTracker:
"""Tracks escalation-worthy policy denials per actor and imposes a
temporary lockout once a threshold is crossed within a window.
This is intentionally simple in-process state, not a replacement for a
real decision/response plane (kings-guard, once it exists). It closes
the gap today: without it, an actor can retry a credential-exfil or
arg-constraint probe indefinitely at whatever rate the rate limiter
otherwise allows.
"""
def __init__(
self,
*,
threshold: int,
window_seconds: int,
lockout_seconds: int,
clock: Callable[[], float] = time.monotonic,
) -> None:
self.threshold = threshold
self.window_seconds = window_seconds
self.lockout_seconds = lockout_seconds
self.clock = clock
self._deny_events: dict[str, deque[float]] = defaultdict(deque)
self._locked_until: dict[str, float] = {}
self._lock = Lock()
def check(self, actor_key: str) -> None:
"""Raise ActorLockedOutError if the actor is currently locked out."""
now = self.clock()
with self._lock:
locked_until = self._locked_until.get(actor_key)
if locked_until is not None:
if now < locked_until:
raise ActorLockedOutError(actor_key, retry_after_seconds=locked_until - now)
del self._locked_until[actor_key]
self._deny_events.pop(actor_key, None)
def record_deny(self, actor_key: str, reason: str) -> None:
"""Record a policy deny outcome; escalate to a lockout if the actor
has crossed the threshold of escalation-worthy denials within the
window."""
if reason not in ESCALATING_DENY_REASONS:
return
now = self.clock()
with self._lock:
window_start = now - self.window_seconds
bucket = self._deny_events[actor_key]
bucket.append(now)
while bucket and bucket[0] < window_start:
bucket.popleft()
if len(bucket) >= self.threshold:
self._locked_until[actor_key] = now + self.lockout_seconds
bucket.clear()

25
src/qonto_assistant/service.py
View file

@ -12,6 +12,7 @@ from qonto_assistant.errors import InvalidRequestError, PolicyDeniedError, Upstr
from qonto_assistant.policy import PolicyEngine
from qonto_assistant.qonto_client import QontoClientProtocol
from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter
from qonto_assistant.security_watch import ActorLockedOutError, DenyEscalationTracker
class CapabilityService:
@ -23,12 +24,14 @@ class CapabilityService:
audit_logger: AuditLogger,
rate_limiter: RateLimiter,
concurrency_limiter: ConcurrencyLimiter,
deny_escalation_tracker: DenyEscalationTracker | None = None,
) -> None:
self.client = client
self.policy = policy
self.audit_logger = audit_logger
self.rate_limiter = rate_limiter
self.concurrency_limiter = concurrency_limiter
self.deny_escalation_tracker = deny_escalation_tracker
def get_accounts(
self, *, claims: ActorClaims, request_id: str, protocol: ProtocolName = "rest"
@ -140,8 +143,29 @@ class CapabilityService:
protocol=protocol,
)
started = time.perf_counter()
actor_key = f"{claims.tenant_id}:{claims.actor_id}"
if self.deny_escalation_tracker is not None:
try:
self.deny_escalation_tracker.check(actor_key)
except ActorLockedOutError:
self._emit_audit(
request_id=request_id,
claims=claims,
capability_id=capability_id,
decision="deny",
deny_reason="actor_locked_out",
latency_ms=_latency_ms(started),
result_count=None,
qonto_http_status=None,
protocol=protocol,
)
raise
decision = self.policy.decide(request)
if not decision.allowed:
if self.deny_escalation_tracker is not None:
self.deny_escalation_tracker.record_deny(actor_key, decision.reason)
self._emit_audit(
request_id=request_id,
claims=claims,
@ -155,7 +179,6 @@ class CapabilityService:
)
raise PolicyDeniedError(decision)
actor_key = f"{claims.tenant_id}:{claims.actor_id}"
self.rate_limiter.check(actor_key)
with self.concurrency_limiter.slot(actor_key):

4
tests/test_api.py
View file

@ -34,6 +34,10 @@ def _settings() -> Settings:
rate_limit_requests=20,
rate_limit_window_seconds=60,
max_concurrency=4,
deny_escalation_enabled=True,
deny_escalation_threshold=3,
deny_escalation_window_seconds=60,
deny_escalation_lockout_seconds=300,
credential_source="env",
openbao_path="tenants/binky/qonto-api",
openbao_command="bao",

179
tests/test_deny_escalation.py Normal file
View file

@ -0,0 +1,179 @@
from pathlib import Path
import pytest
from qonto_assistant.audit import AuditLogger
from qonto_assistant.contracts import ActorClaims
from qonto_assistant.errors import PolicyDeniedError
from qonto_assistant.policy import PolicyEngine
from qonto_assistant.qonto_client import FixtureQontoClient
from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter
from qonto_assistant.security_watch import (
ActorLockedOutError,
DenyEscalationTracker,
)
from qonto_assistant.service import CapabilityService
POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml"
FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "qonto"
class _FakeClock:
def __init__(self) -> None:
self.now = 0.0
def __call__(self) -> float:
return self.now
def advance(self, seconds: float) -> None:
self.now += seconds
def _claims(actor_id: str = "prober") -> ActorClaims:
return ActorClaims(actor_id=actor_id, tenant_id="binky", lane="green")
def _service(tracker: DenyEscalationTracker, events: list[dict[str, object]]) -> CapabilityService:
return CapabilityService(
client=FixtureQontoClient(fixture_dir=FIXTURE_DIR),
policy=PolicyEngine.from_file(
POLICY_FILE,
required_scope="finance.qonto.read",
enforce_scope=False,
),
audit_logger=AuditLogger(sink=events.append),
rate_limiter=RateLimiter(limit=100, window_seconds=60),
concurrency_limiter=ConcurrencyLimiter(limit=4),
deny_escalation_tracker=tracker,
)
# --- Unit tests for the tracker itself -------------------------------------------------
def test_tracker_allows_denies_below_threshold() -> None:
clock = _FakeClock()
tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock)
tracker.record_deny("binky:prober", "arg_constraint")
tracker.record_deny("binky:prober", "arg_constraint")
tracker.check("binky:prober") # should not raise
def test_tracker_locks_out_after_threshold_within_window() -> None:
clock = _FakeClock()
tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock)
for _ in range(3):
tracker.record_deny("binky:prober", "arg_constraint")
with pytest.raises(ActorLockedOutError):
tracker.check("binky:prober")
def test_tracker_ignores_non_escalating_deny_reasons() -> None:
clock = _FakeClock()
tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock)
for _ in range(10):
tracker.record_deny("binky:prober", "tenant_scope")
tracker.record_deny("binky:prober", "authz_denied")
tracker.check("binky:prober") # should not raise -- neither reason escalates
def test_tracker_denies_outside_window_do_not_accumulate() -> None:
clock = _FakeClock()
tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock)
tracker.record_deny("binky:prober", "arg_constraint")
clock.advance(61)
tracker.record_deny("binky:prober", "arg_constraint")
clock.advance(61)
tracker.record_deny("binky:prober", "arg_constraint")
tracker.check("binky:prober") # should not raise -- each deny fell outside the prior window
def test_tracker_lockout_expires_after_lockout_window() -> None:
clock = _FakeClock()
tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock)
for _ in range(3):
tracker.record_deny("binky:prober", "arg_constraint")
with pytest.raises(ActorLockedOutError):
tracker.check("binky:prober")
clock.advance(301)
tracker.check("binky:prober") # should not raise -- lockout window elapsed
# Escalation state was cleared, not just the lockout: it takes a full
# fresh threshold of denies to lock out again.
tracker.record_deny("binky:prober", "arg_constraint")
tracker.record_deny("binky:prober", "arg_constraint")
tracker.check("binky:prober") # should not raise -- only 2 denies since the reset
def test_tracker_is_scoped_per_actor() -> None:
clock = _FakeClock()
tracker = DenyEscalationTracker(threshold=3, window_seconds=60, lockout_seconds=300, clock=clock)
for _ in range(3):
tracker.record_deny("binky:prober", "arg_constraint")
tracker.check("binky:other-actor") # should not raise -- different actor, untouched
# --- Integration through CapabilityService ---------------------------------------------
def test_service_locks_out_actor_after_repeated_arg_constraint_denies() -> None:
clock = _FakeClock()
tracker = DenyEscalationTracker(threshold=2, window_seconds=60, lockout_seconds=300, clock=clock)
events: list[dict[str, object]] = []
service = _service(tracker, events)
def _oversized_page_call() -> None:
service.list_transactions(
claims=_claims(),
request_id="req-probe",
account_slug=None,
page=1,
page_size=10_000, # far beyond max_per_page -> arg_constraint
window_days=31,
status="completed",
side=None,
protocol="rest",
)
# First two denies cross the threshold=2 escalation bar.
for _ in range(2):
with pytest.raises(PolicyDeniedError):
_oversized_page_call()
# A third call -- even to an unrelated, otherwise-allowed capability --
# is now rejected before the policy kernel is consulted at all.
with pytest.raises(ActorLockedOutError):
service.get_accounts(claims=_claims(), request_id="req-locked", protocol="rest")
lockout_events = [event for event in events if event.get("deny_reason") == "actor_locked_out"]
assert len(lockout_events) == 1
def test_service_does_not_lock_out_for_ordinary_denies() -> None:
clock = _FakeClock()
tracker = DenyEscalationTracker(threshold=2, window_seconds=60, lockout_seconds=300, clock=clock)
events: list[dict[str, object]] = []
service = _service(tracker, events)
red_lane_claims = ActorClaims(actor_id="prober", tenant_id="binky", lane="red")
for _ in range(5):
with pytest.raises(PolicyDeniedError):
service.get_accounts(claims=red_lane_claims, request_id="req-lane", protocol="rest")
# authz_denied (wrong lane) never escalates -- the actor should still be
# able to make a legitimate, correctly-scoped call afterwards.
payload = service.get_accounts(claims=_claims(), request_id="req-ok", protocol="rest")
assert "organization" in payload