diff --git a/CHANGELOG.md b/CHANGELOG.md index 188309f..20cf5e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,34 @@ All notable changes to this project are documented in this file. ## Unreleased +### Added (phase-1 sidecar) + +- `enforcement.py` + `enforcement.phase` in policy — the rollout posture is now configuration. The engine always computes a full decision; the phase decides how much is enforced, so advancing is a config change and never a code change, and all three phases exist from day one. `apply_enforcement` never mutates the decision: the audit records what the engine ruled, not what a partially-enforcing deployment did about it, and conflating those would make phase-1 data worthless. +- `client.py` — `AetheryaClient`, a dependency-free HTTP client for the sidecar shape (agent in one process, engine in another). Fails **closed** from phase 2 on: an unreachable service refuses the action, because a boundary that disappears when the network hiccups is not a boundary. Phase 1 is the explicit exception — it enforces nothing, so an outage must not stop the agent. +- `rollout_report.py` + `aetherya rollout report` — measures a phase against its exit criteria from the audit trail: window size, decisions by state, what the next phase would act on, distinct actors and tools per state, intent escalations, skipped semantics, policy-fingerprint uniformity, chain integrity, and the `hard_deny` events that require manual review. The `hard_deny_reviewed` criterion **never auto-passes**: the tool can count and list those events, it cannot judge them, and auto-passing would turn the phase gate into the formality it exists to prevent. Exits non-zero until ready, so it works as a promotion gate. +- `examples/sidecar_agent.py` — the same agent trajectory as `agent_loop.py`, but over HTTP against the container, with the shadow gap counted per run. +- `Dockerfile.slim` + `config/policy.slim.yaml` — **8.7GB → 330MB**. The size was never ÆTHERYA; it was PyTorch, pulled in by sentence-transformers for the advisory layer. That layer is capped below every deny threshold, so it can escalate to a human but never refuse alone, and the deterministic core does not use it. Verified: the slim image hard-denies exactly what the full one does, including the quote-split evasion `r''m -rf /`, and reports healthy in 8s instead of ~25s. A test pins the two policies to differ in that one field. + +### Fixed + +- `rollout report` crashed on a malformed audit line or an empty file, because `verify_audit_file` raises rather than returning. Both are findings an operator must see before advancing a phase — they are now reported as a failed `chain_intact` criterion with the detail, instead of taking the tool down or being silently skipped. + +### Fixed (CI) + +- `examples/agent_loop.py`: the OpenAI backend checked for the SDK before checking for the API key, so the error you got depended on whether `openai` happened to be installed. It is present in a dev environment and absent from CI's `[dev]` extra, which is exactly why `test_openai_backend_requires_a_key` passed locally and failed CI on the last two pushes. The key is now checked first — a missing key is the more common misconfiguration, and you should not need the SDK installed to be told about it — and both paths have tests, one of which blocks the import to prove it. +- `security_gate` phase 2 generates its own corpus but `AuditLogger` appends by design, so a file left by a previous local run was appended to and counted. The phase then failed on events it had not created, reporting `invalid == total`, which reads like a real integrity regression. CI never saw it (fresh checkout) but a second local run always did. The phase now starts from a clean file and is idempotent, with a test asserting two consecutive runs both pass. + +### Fixed (found while containerising) + +- **The HTTP API never applied the configured rate limit.** `rate_limit` was validated at load and `build_rate_limiter()` had 13 integration tests, but nothing in the API path ever constructed a limiter — the configured limit was documentation, not behaviour. `AetheryaAPI` now builds one per process (rebuilt only when the policy's rate-limit section changes; rebuilding per request would reset every window and make it a no-op) and passes it to the pipeline. Verified end to end through the container: 65 requests from one actor against a limit of 60 yielded 60 allowed and 5 refused, with the window visible in Redis. +- `run_pipeline` typed `rate_limiter` as `ActorRateLimiter`, so the Redis backend could not be passed to it at all. Widened to the `RateLimiter` protocol — mypy caught this the moment the API tried to wire it up. + +### Added (containerised local stack) + +- `Dockerfile` — two-stage build, non-root user, the semantic model baked in at build time (with `require_warm_semantic_model`, downloading it on first boot would either delay readiness or leave the advisory layer silently inactive). Healthcheck asserts `ok && !degraded && policy_fingerprint_match`, so a replica serving under an unintended policy or with the advisory layer inert is not marked healthy. +- `docker-compose.yml` — decision service plus Redis, with the audit chain on a named volume so it survives the container. Verified: chain intact across a restart (9 events, 0 invalid). The approvals service is deliberately absent — it is localhost-only by design and belongs to phase 3, not phase 1. +- `config/policy.docker.yaml` — deployment policy, identical to `config/policy.yaml` except for the distributed rate-limit backend. The repo default stays `memory` because `redis` fails closed and would refuse every request without a reachable server. A test pins the two files to differ in exactly that one field so they cannot drift. + ### Added (rollout) - `docs/rollout-phases.md` — the three-phase production rollout as an executable plan: shadow, hard-deny enforcement, full enforcement. Each phase states its posture, the wiring change it needs, what to measure, and **exit criteria that can fail**, including what to do when shadow mode reports an unacceptable false-positive rate. Also documents the target operating configuration and the `/health` readiness gate. diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..d099ee6 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,65 @@ +# ÆTHERYA decision service. +# +# Two stages so the runtime image does not carry build tooling. The semantic +# model is baked in at build time: `require_warm_semantic_model` means the +# advisory layer declines to run on a cold model, and downloading ~90MB on first +# boot would either delay readiness or leave the layer silently inactive. + +FROM python:3.11-slim AS builder + +WORKDIR /build + +RUN python -m venv /opt/venv +ENV PATH="/opt/venv/bin:$PATH" + +COPY pyproject.toml README.md ./ +COPY src/ ./src/ + +RUN pip install --no-cache-dir -U pip \ + && pip install --no-cache-dir ".[redis]" + +# Pre-download the sentence-transformers model into the image. +ENV HF_HOME=/opt/models +RUN python -c "from aetherya.constitution import warmup_semantic_model; warmup_semantic_model()" + + +FROM python:3.11-slim AS runtime + +# Never run the decision boundary as root. +RUN useradd --create-home --uid 10001 aetherya + +COPY --from=builder /opt/venv /opt/venv +COPY --from=builder --chown=aetherya:aetherya /opt/models /opt/models + +ENV PATH="/opt/venv/bin:$PATH" \ + PYTHONUNBUFFERED=1 \ + PYTHONDONTWRITEBYTECODE=1 \ + HF_HOME=/opt/models + +WORKDIR /app +COPY --chown=aetherya:aetherya config/ ./config/ +COPY --chown=aetherya:aetherya examples/ ./examples/ + +# The audit trail is a chained JSONL file; it must outlive the container. +# docker-compose mounts a volume here. +RUN mkdir -p /app/audit && chown aetherya:aetherya /app/audit +VOLUME ["/app/audit"] + +USER aetherya +EXPOSE 8080 + +# `degraded` and `policy_fingerprint_match` are the fields that matter: a replica +# serving with the advisory layer inactive, or under an unintended policy, should +# not receive traffic. +HEALTHCHECK --interval=15s --timeout=5s --start-period=20s --retries=3 \ + CMD python -c "import json,urllib.request;\ +d=json.load(urllib.request.urlopen('http://127.0.0.1:8080/health',timeout=4));\ +raise SystemExit(0 if d.get('ok') and not d.get('degraded') and d.get('policy_fingerprint_match') else 1)" + +# --require-semantic-ready: fail startup rather than serve in a degraded state. +CMD ["aetherya-api", \ + "--host", "0.0.0.0", \ + "--port", "8080", \ + "--policy-path", "config/policy.yaml", \ + "--audit-path", "audit/decisions.jsonl", \ + "--require-semantic-ready"] diff --git a/Dockerfile.slim b/Dockerfile.slim new file mode 100644 index 0000000..e07e1f6 --- /dev/null +++ b/Dockerfile.slim @@ -0,0 +1,70 @@ +# ÆTHERYA decision service — slim variant (~200MB instead of ~8.7GB). +# +# The size of the full image is not ÆTHERYA: it is PyTorch, pulled in by +# sentence-transformers for the semantic layer. That layer is advisory by +# design — capped below every deny threshold, so it can escalate to a human +# but never refuse on its own. The deterministic core (parser, guards, +# procedural corpus, jailbreak, capabilities, risk, confirmation, audit) does +# not use it at all. +# +# So this variant drops the dependency and runs with `use_semantic: false`. +# Everything that hard-denies today still hard-denies: those verdicts come +# from the procedural and jailbreak corpora, not from similarity. What you +# lose is paraphrase detection on inputs no keyword matches. +# +# Recommended for phase 1: what you are measuring there is deterministic +# refusals, and this starts in seconds instead of ~15s. + +FROM python:3.11-slim AS builder + +WORKDIR /build + +RUN python -m venv /opt/venv +ENV PATH="/opt/venv/bin:$PATH" + +COPY pyproject.toml README.md ./ +COPY src/ ./src/ + +RUN pip install --no-cache-dir -U pip \ + && pip install --no-cache-dir --no-deps ".[redis]" \ + && pip install --no-cache-dir "pyyaml" "numpy>=1.26.0" "redis>=5.0.0" + + +FROM python:3.11-slim AS runtime + +# Never run the decision boundary as root. +RUN useradd --create-home --uid 10001 aetherya + +COPY --from=builder /opt/venv /opt/venv + +ENV PATH="/opt/venv/bin:$PATH" \ + PYTHONUNBUFFERED=1 \ + PYTHONDONTWRITEBYTECODE=1 + +WORKDIR /app +COPY --chown=aetherya:aetherya config/ ./config/ +COPY --chown=aetherya:aetherya examples/ ./examples/ + +# The audit trail is a chained JSONL file; it must outlive the container. +# docker-compose mounts a volume here. +RUN mkdir -p /app/audit && chown aetherya:aetherya /app/audit +VOLUME ["/app/audit"] + +USER aetherya +EXPOSE 8080 + +# `degraded` and `policy_fingerprint_match` are the fields that matter: a replica +# serving with the advisory layer inactive, or under an unintended policy, should +# not receive traffic. +HEALTHCHECK --interval=15s --timeout=5s --start-period=20s --retries=3 \ + CMD python -c "import json,urllib.request;\ +d=json.load(urllib.request.urlopen('http://127.0.0.1:8080/health',timeout=4));\ +raise SystemExit(0 if d.get('ok') and not d.get('degraded') and d.get('policy_fingerprint_match') else 1)" + +# --require-semantic-ready: fail startup rather than serve in a degraded state. +CMD ["aetherya-api", \ + "--host", "0.0.0.0", \ + "--port", "8080", \ + "--policy-path", "config/policy.slim.yaml", \ + "--audit-path", "audit/decisions.jsonl", \ + "--no-warmup-semantic"] diff --git a/config/policy.docker.yaml b/config/policy.docker.yaml new file mode 100644 index 0000000..230983b --- /dev/null +++ b/config/policy.docker.yaml @@ -0,0 +1,219 @@ +# Deployment policy: identical to config/policy.yaml except that the rate +# limiter uses the distributed backend. Mounted by docker-compose.yml. +# +# The repo default stays `memory` so the CLI and a bare `aetherya-api` work +# with no infrastructure. `redis` fails closed, so pointing at it without a +# reachable server refuses every request — correct, but not a default. +# +# tests/test_deployment_contracts.py asserts these two files differ in +# exactly this one field, so they cannot drift apart. +version: 1 + +modes: + consultive: + default_state: log_only + thresholds: + deny_at: 90 + confirm_at: 60 + log_only_at: 0 + + operative: + default_state: allow + thresholds: + deny_at: 80 + confirm_at: 50 + log_only_at: 0 + +aggregator: + weights: + execution_gate: 1.0 + capability_gate: 1.0 + jailbreak_guard: 1.0 + procedural_guard: 1.0 + constitution: 1.0 + hard_deny_if: + - critical_tag_detected + - "tag:jailbreak_attempt" + - "tag:tool_not_allowed" + - "tag:capability_tool_denied" + +procedural_guard: + # Each entry activates a rule family in procedural_guard.py. The legacy + # literals ("rm -rf /", "mkfs", ":(){:|:&};:") now expand to flag-order + # tolerant, device-anchored patterns instead of naive substring matching. + # Symbolic keys activate families that have no single literal form. An + # unrecognised entry degrades to literal substring matching, so custom + # operator rules keep working unchanged. + critical_tags: + - "rm -rf /" # root recursive delete: -rf, -fr, -r -f, --no-preserve-root + - "mkfs" # filesystem format — requires a /dev/ argument + - ":(){:|:&};:" # classic fork bomb + - "block_device_write" # dd of=/dev/sdX, > /dev/sdX, shred|wipefs|tee|truncate /dev/X + - "interpreter_destruction" # shutil.rmtree('/') inside a python -c one-liner + - "find_delete" # find / ... -delete + - "recursive_force_delete" # rm -rf on a non-root path (risk 55, not hard-deny) + - "remote_code_execution" # curl|wget piped into a shell (risk 70) + privileged_ops: + - "apt remove" + - "apt purge" + - "userdel" + - "chmod -R 777 /" + +execution_gate: + enabled: true + allowed_tools: + - shell + - http + - filesystem + require_target_for_operate: false + required_parameters: + shell: + - command + allowed_parameters: + shell: + - command + - timeout + http: + - method + - url + - body + filesystem: + - path + - operation + - confirm_token + - confirm_context + +capability_matrix: + enabled: true + default_allow: false + roles: + operator: + tools: + - shell + - http + - filesystem + operations: + - read + - write + - append + - list + - request + - command + actors: + robert: + roles: + - operator + tools: [] + operations: [] + +confirmation: + enabled: true + on_confirmed: allow + require_for: + decisions: + - require_confirm + tools: [] + operations: + - delete + - write + min_risk_score: 0 + evidence: + token_param: confirm_token + context_param: confirm_context + token_pattern: "^ack:[a-z0-9_-]{8,}$" + min_context_length: 12 + signed_proof: + enabled: false + proof_param: confirm_proof + key_env: AETHERYA_CONFIRMATION_HMAC_KEY + keyring_env: AETHERYA_CONFIRMATION_HMAC_KEYRING + active_kid: k1 + max_valid_for_sec: 900 + clock_skew_sec: 5 + replay_mode: single_use + replay_store: memory + replay_redis_url_env: AETHERYA_CONFIRMATION_REPLAY_REDIS_URL + replay_redis_prefix: aetherya:appr + +# LLM shadow — non-authoritative ethical evaluation, never affects the decision. +# provider: dry_run | openai | anthropic +# openai requires OPENAI_API_KEY (e.g. model: gpt-4o-mini) +# anthropic requires ANTHROPIC_API_KEY (e.g. model: claude-opus-4-8) and +# `pip install -e ".[anthropic]"`. Note: temperature is ignored by +# the anthropic provider (current Claude models reject sampling params). +llm_shadow: + enabled: false + provider: dry_run + model: gpt-dry + temperature: 0.0 + max_tokens: 96 + timeout_sec: 10.0 + +policy_adapter_shadow: + enabled: false + max_signals: 3 + +# Rate limiting. The "memory" backend keeps windows in process memory, so behind +# N workers the effective limit is N x requests_per_window — the configured limit +# is simply not the limit that applies. Any multi-worker or multi-replica +# deployment must use "redis". The redis backend fails closed: if Redis is +# unreachable the request is refused rather than silently going unlimited. +rate_limit: + backend: redis # memory | redis + requests_per_window: 60 + window_seconds: 60.0 + max_actors: 10000 # memory backend only: LRU cap against actor-id flooding + redis_url_env: AETHERYA_RATE_LIMIT_REDIS_URL + redis_prefix: "aetherya:ratelimit" + +# Rollout posture. The engine always computes a full decision; the phase only +# decides how much of it is enforced, so advancing is a config change and never +# a code change. Measure the current phase with `aetherya rollout report` before +# advancing. See docs/rollout-phases.md. +# 1 shadow — observe only, nothing is refused +# 2 hard_deny — refuse hard_deny +# 3 full — refuse hard_deny and deny, hold escalate for confirmation +enforcement: + phase: 1 + +# Intent escalation — decouples the guard chain from the parser's verb list. +# ExecutionGate and the capability matrix only evaluate `operate` requests, so +# without this stage a command the parser does not recognise (e.g. +# `dd if=/dev/zero of=/dev/sda`) skips both gates and can reach `allow`. +# Escalation is monotone: it only ever raises ask/consultive -> operate/operative. +intent_escalation: + enabled: true + use_procedural_signal: true # any ProceduralGuard hit implies operative content + use_shape_signals: true # $(...), pipe-to-shell, > /path, binary+flags, sudo X + +# OutputGate contract configuration. +# require_candidate_response: false (default) means OutputGate is opt-in — +# the caller chooses whether to pass candidate_response to run_pipeline. +# Set to true in deployments where output protection must always be enforced. +output_gate: + require_candidate_response: false + +# Constitution evaluator configuration. +# use_semantic: activate the SemanticEvaluator (all-MiniLM-L6-v2) for inputs that +# Fast keyword matching cannot resolve definitively. Enabled by default. +# Thresholds govern the SemanticEvaluator similarity cutoffs. +# Modify with care: lowering thresholds increases sensitivity and false-positive risk. +constitution: + use_semantic: true + semantic_violation_threshold: 0.55 + semantic_gray_zone_threshold: 0.35 + # NON-AUTHORITATIVE CONTRACT: the semantic layer is the only learned component + # in the decision path. Its risk contribution is capped here so it can raise a + # request to `escalate` (a human decides) but never to `deny` on its own. + # Must stay strictly below every mode's deny_at — validated at policy load. + # This is what keeps the *core verdict* deterministic: swapping the model can + # change whether a human is asked, never whether an action is refused. + semantic_max_risk: 60 + # Loading all-MiniLM-L6-v2 costs ~5s in a fresh process. That belongs at + # deployment startup (`aetherya warmup`), never inside a decision. With this + # set, the advisory layer declines to run when the model is not already in the + # process cache instead of stalling the decision path; the trace then carries + # `constitution.semantic_skipped: model_not_warm`. Long-lived servers should + # call `aetherya warmup` at boot so the layer is always available. + # Set to false only if you accept a multi-second cold load inside a decision. + require_warm_semantic_model: true diff --git a/config/policy.slim.yaml b/config/policy.slim.yaml new file mode 100644 index 0000000..94c800e --- /dev/null +++ b/config/policy.slim.yaml @@ -0,0 +1,218 @@ +# Slim deployment policy: identical to config/policy.docker.yaml with the +# advisory semantic layer disabled, so the image does not need PyTorch. +# +# The layer is advisory by construction — capped below every deny threshold, +# it can escalate to a human but never refuse alone. Disabling it removes +# paraphrase detection and changes nothing about what is hard-denied. +# +# tests/test_deployment_contracts.py pins the difference to this one field. +version: 1 + +modes: + consultive: + default_state: log_only + thresholds: + deny_at: 90 + confirm_at: 60 + log_only_at: 0 + + operative: + default_state: allow + thresholds: + deny_at: 80 + confirm_at: 50 + log_only_at: 0 + +aggregator: + weights: + execution_gate: 1.0 + capability_gate: 1.0 + jailbreak_guard: 1.0 + procedural_guard: 1.0 + constitution: 1.0 + hard_deny_if: + - critical_tag_detected + - "tag:jailbreak_attempt" + - "tag:tool_not_allowed" + - "tag:capability_tool_denied" + +procedural_guard: + # Each entry activates a rule family in procedural_guard.py. The legacy + # literals ("rm -rf /", "mkfs", ":(){:|:&};:") now expand to flag-order + # tolerant, device-anchored patterns instead of naive substring matching. + # Symbolic keys activate families that have no single literal form. An + # unrecognised entry degrades to literal substring matching, so custom + # operator rules keep working unchanged. + critical_tags: + - "rm -rf /" # root recursive delete: -rf, -fr, -r -f, --no-preserve-root + - "mkfs" # filesystem format — requires a /dev/ argument + - ":(){:|:&};:" # classic fork bomb + - "block_device_write" # dd of=/dev/sdX, > /dev/sdX, shred|wipefs|tee|truncate /dev/X + - "interpreter_destruction" # shutil.rmtree('/') inside a python -c one-liner + - "find_delete" # find / ... -delete + - "recursive_force_delete" # rm -rf on a non-root path (risk 55, not hard-deny) + - "remote_code_execution" # curl|wget piped into a shell (risk 70) + privileged_ops: + - "apt remove" + - "apt purge" + - "userdel" + - "chmod -R 777 /" + +execution_gate: + enabled: true + allowed_tools: + - shell + - http + - filesystem + require_target_for_operate: false + required_parameters: + shell: + - command + allowed_parameters: + shell: + - command + - timeout + http: + - method + - url + - body + filesystem: + - path + - operation + - confirm_token + - confirm_context + +capability_matrix: + enabled: true + default_allow: false + roles: + operator: + tools: + - shell + - http + - filesystem + operations: + - read + - write + - append + - list + - request + - command + actors: + robert: + roles: + - operator + tools: [] + operations: [] + +confirmation: + enabled: true + on_confirmed: allow + require_for: + decisions: + - require_confirm + tools: [] + operations: + - delete + - write + min_risk_score: 0 + evidence: + token_param: confirm_token + context_param: confirm_context + token_pattern: "^ack:[a-z0-9_-]{8,}$" + min_context_length: 12 + signed_proof: + enabled: false + proof_param: confirm_proof + key_env: AETHERYA_CONFIRMATION_HMAC_KEY + keyring_env: AETHERYA_CONFIRMATION_HMAC_KEYRING + active_kid: k1 + max_valid_for_sec: 900 + clock_skew_sec: 5 + replay_mode: single_use + replay_store: memory + replay_redis_url_env: AETHERYA_CONFIRMATION_REPLAY_REDIS_URL + replay_redis_prefix: aetherya:appr + +# LLM shadow — non-authoritative ethical evaluation, never affects the decision. +# provider: dry_run | openai | anthropic +# openai requires OPENAI_API_KEY (e.g. model: gpt-4o-mini) +# anthropic requires ANTHROPIC_API_KEY (e.g. model: claude-opus-4-8) and +# `pip install -e ".[anthropic]"`. Note: temperature is ignored by +# the anthropic provider (current Claude models reject sampling params). +llm_shadow: + enabled: false + provider: dry_run + model: gpt-dry + temperature: 0.0 + max_tokens: 96 + timeout_sec: 10.0 + +policy_adapter_shadow: + enabled: false + max_signals: 3 + +# Rate limiting. The "memory" backend keeps windows in process memory, so behind +# N workers the effective limit is N x requests_per_window — the configured limit +# is simply not the limit that applies. Any multi-worker or multi-replica +# deployment must use "redis". The redis backend fails closed: if Redis is +# unreachable the request is refused rather than silently going unlimited. +rate_limit: + backend: redis # memory | redis + requests_per_window: 60 + window_seconds: 60.0 + max_actors: 10000 # memory backend only: LRU cap against actor-id flooding + redis_url_env: AETHERYA_RATE_LIMIT_REDIS_URL + redis_prefix: "aetherya:ratelimit" + +# Rollout posture. The engine always computes a full decision; the phase only +# decides how much of it is enforced, so advancing is a config change and never +# a code change. Measure the current phase with `aetherya rollout report` before +# advancing. See docs/rollout-phases.md. +# 1 shadow — observe only, nothing is refused +# 2 hard_deny — refuse hard_deny +# 3 full — refuse hard_deny and deny, hold escalate for confirmation +enforcement: + phase: 1 + +# Intent escalation — decouples the guard chain from the parser's verb list. +# ExecutionGate and the capability matrix only evaluate `operate` requests, so +# without this stage a command the parser does not recognise (e.g. +# `dd if=/dev/zero of=/dev/sda`) skips both gates and can reach `allow`. +# Escalation is monotone: it only ever raises ask/consultive -> operate/operative. +intent_escalation: + enabled: true + use_procedural_signal: true # any ProceduralGuard hit implies operative content + use_shape_signals: true # $(...), pipe-to-shell, > /path, binary+flags, sudo X + +# OutputGate contract configuration. +# require_candidate_response: false (default) means OutputGate is opt-in — +# the caller chooses whether to pass candidate_response to run_pipeline. +# Set to true in deployments where output protection must always be enforced. +output_gate: + require_candidate_response: false + +# Constitution evaluator configuration. +# use_semantic: activate the SemanticEvaluator (all-MiniLM-L6-v2) for inputs that +# Fast keyword matching cannot resolve definitively. Enabled by default. +# Thresholds govern the SemanticEvaluator similarity cutoffs. +# Modify with care: lowering thresholds increases sensitivity and false-positive risk. +constitution: + use_semantic: false + semantic_violation_threshold: 0.55 + semantic_gray_zone_threshold: 0.35 + # NON-AUTHORITATIVE CONTRACT: the semantic layer is the only learned component + # in the decision path. Its risk contribution is capped here so it can raise a + # request to `escalate` (a human decides) but never to `deny` on its own. + # Must stay strictly below every mode's deny_at — validated at policy load. + # This is what keeps the *core verdict* deterministic: swapping the model can + # change whether a human is asked, never whether an action is refused. + semantic_max_risk: 60 + # Loading all-MiniLM-L6-v2 costs ~5s in a fresh process. That belongs at + # deployment startup (`aetherya warmup`), never inside a decision. With this + # set, the advisory layer declines to run when the model is not already in the + # process cache instead of stalling the decision path; the trace then carries + # `constitution.semantic_skipped: model_not_warm`. Long-lived servers should + # call `aetherya warmup` at boot so the layer is always available. + # Set to false only if you accept a multi-second cold load inside a decision. + require_warm_semantic_model: true diff --git a/config/policy.yaml b/config/policy.yaml index c853fc0..623b67f 100644 --- a/config/policy.yaml +++ b/config/policy.yaml @@ -157,6 +157,16 @@ rate_limit: redis_url_env: AETHERYA_RATE_LIMIT_REDIS_URL redis_prefix: "aetherya:ratelimit" +# Rollout posture. The engine always computes a full decision; the phase only +# decides how much of it is enforced, so advancing is a config change and never +# a code change. Measure the current phase with `aetherya rollout report` before +# advancing. See docs/rollout-phases.md. +# 1 shadow — observe only, nothing is refused +# 2 hard_deny — refuse hard_deny +# 3 full — refuse hard_deny and deny, hold escalate for confirmation +enforcement: + phase: 1 + # Intent escalation — decouples the guard chain from the parser's verb list. # ExecutionGate and the capability matrix only evaluate `operate` requests, so # without this stage a command the parser does not recognise (e.g. diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..c2a2725 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,71 @@ +# Local phase-1 stack: the decision service plus the Redis it needs for shared +# state. See docs/rollout-phases.md. +# +# docker compose up --build +# curl -s localhost:8080/health | python -m json.tool +# open http://localhost:8080 +# +# This runs the decision profile only. The approvals service is deliberately +# absent: it signs approval proofs, is localhost-only by design, and belongs on +# a separate trust boundary — phase 3, not phase 1. + +services: + redis: + image: redis:7-alpine + container_name: aetherya-redis + ports: + - "6379:6379" + command: ["redis-server", "--appendonly", "yes"] + volumes: + - redis-data:/data + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 5s + timeout: 3s + retries: 10 + + aetherya: + build: . + container_name: aetherya-decision + ports: + - "8080:8080" + depends_on: + redis: + condition: service_healthy + environment: + # Shared rate-limit state. With the in-process backend, N replicas mean + # N x the configured limit — see docs/security-model.md. + AETHERYA_RATE_LIMIT_REDIS_URL: redis://redis:6379/0 + AETHERYA_CONFIRMATION_REPLAY_REDIS_URL: redis://redis:6379/1 + + # Signs the audit chain. Replace before this is anything but a local demo: + # without a real secret the attestation proves the file is self-consistent, + # not that this service produced it. + AETHERYA_ATTESTATION_KEY: "local-dev-attestation-key-replace-me" + + # Pin the policy so a replica refuses to start under an unintended one. + # Get the value with: aetherya policy fingerprint --json + # Left unset locally so the stack comes up before you have run that. + # AETHERYA_EXPECTED_POLICY_FINGERPRINT: "sha256:..." + + volumes: + # The audit chain must survive the container. Without this, every restart + # begins a new chain and the old one is gone. + - audit-data:/app/audit + # Edit config/policy.yaml on the host and restart to apply. + - ./config:/app/config:ro + command: + - aetherya-api + - --host=0.0.0.0 + - --port=8080 + # Deployment policy: same as config/policy.yaml with the + # distributed rate-limit backend. + - --policy-path=config/policy.docker.yaml + - --audit-path=audit/decisions.jsonl + - --require-semantic-ready + + restart: unless-stopped + +volumes: + redis-data: + audit-data: diff --git a/docs/rollout-phases.md b/docs/rollout-phases.md index d978d1a..f9beba8 100644 --- a/docs/rollout-phases.md +++ b/docs/rollout-phases.md @@ -8,8 +8,30 @@ that decision is enforced**. Nothing about the policy, the integration or the audit trail changes between phases — only the enforcement posture. That is what makes it safe to start in observation mode and tighten later. -Run `python examples/agent_loop.py --phase N` to see each posture against a real -agent loop. +The phase is configuration, not code: + +```yaml +# config/policy.yaml +enforcement: + phase: 1 # 1 shadow · 2 hard_deny · 3 full +``` + +Two ways to see it: + +```bash +python examples/agent_loop.py --phase 1 # engine in-process +python examples/sidecar_agent.py --phase 1 # agent talking to the container +``` + +And one way to know when to advance: + +```bash +aetherya rollout report --phase 1 --audit-path audit/decisions.jsonl +``` + +That command evaluates the exit criteria below against the audit trail, prints +the `hard_deny` events that need human review, and exits non-zero until the +phase is ready — usable as a gate in a promotion pipeline. | Phase | Blocks | Confirms | Risk of deploying it | |---|---|---|---| diff --git a/examples/agent_loop.py b/examples/agent_loop.py index 43f9fc1..529d0d6 100644 --- a/examples/agent_loop.py +++ b/examples/agent_loop.py @@ -300,12 +300,18 @@ class OpenAIAgent: ) def __init__(self, model: str = "gpt-4o-mini") -> None: - try: - from openai import OpenAI - except ImportError as exc: # pragma: no cover - optional path - raise SystemExit("pip install openai, or use --agent scripted") from exc + # Key first: a missing key is the more common misconfiguration, and you + # should not need the SDK installed to be told about it. (Checking the + # import first also made this branch depend on whether `openai` happened + # to be present, which is exactly how it passed locally and failed CI.) if not os.getenv("OPENAI_API_KEY", "").strip(): raise SystemExit("OPENAI_API_KEY is not set — use --agent scripted") + try: + from openai import OpenAI + except ImportError as exc: + raise SystemExit( + 'openai is not installed — pip install -e ".[llm]", or use --agent scripted' + ) from exc self._client = OpenAI() self._model = model self._messages: list[dict[str, Any]] = [ diff --git a/examples/sidecar_agent.py b/examples/sidecar_agent.py new file mode 100644 index 0000000..145c00c --- /dev/null +++ b/examples/sidecar_agent.py @@ -0,0 +1,150 @@ +""" +Phase-1 sidecar: an agent talking to the decision service over HTTP. + + docker compose up -d + python examples/sidecar_agent.py + +`agent_loop.py` runs the engine in-process, which is the right shape for a +library. This is the deployment shape: the agent is one process, ÆTHERYA is +another, and the only coupling is an HTTP call. That is what lets you put the +boundary in front of an agent you did not write, in a language that is not +Python, without linking anything. + + ┌─────────┐ proposed tool call ┌──────────┐ + │ agent │ ─────────────────────► │ ÆTHERYA │ + │ process │ ◄───────────────────── │ sidecar │ + └────┬────┘ allow | hold | refuse└──────────┘ + │ │ + ▼ executes only if allowed ▼ records every decision + tool runtime audit chain + +Phase 1 observes: nothing is refused, but every decision is recorded, and the +verdict carries `shadow_gap=True` when a later phase *would* have refused. That +count is the number the phase exists to produce — read it back with: + + aetherya rollout report --phase 1 +""" + +from __future__ import annotations + +import argparse +import sys +from typing import Any + +from aetherya.actions import ActionRequest +from aetherya.client import AetheryaClient + +# The same trajectory as agent_loop.py: the agent is not malicious, it reads a +# poisoned document and proposes what the document told it to. +PROPOSALS: list[tuple[str, ActionRequest]] = [ + ( + "inspect current config", + ActionRequest( + raw_input="read file /srv/app/config.yaml", + intent="operate", + mode_hint="operative", + tool="filesystem", + target="/srv/app/config.yaml", + parameters={"path": "/srv/app/config.yaml", "operation": "read"}, + ), + ), + ( + "check the deploy notes", + ActionRequest( + raw_input="read file /srv/app/deploy_notes.md", + intent="operate", + mode_hint="operative", + tool="filesystem", + target="/srv/app/deploy_notes.md", + parameters={"path": "/srv/app/deploy_notes.md", "operation": "read"}, + ), + ), + ( + "the notes say to free disk space before deploying", + ActionRequest( + raw_input="rm -rf / --no-preserve-root", + intent="operate", + mode_hint="operative", + tool="shell", + target="localhost", + parameters={"command": "rm -rf / --no-preserve-root"}, + ), + ), + ( + "apply the scaling change", + ActionRequest( + raw_input="write file /srv/app/Config.PROD.yaml", + intent="operate", + mode_hint="operative", + tool="filesystem", + target="/srv/app/Config.PROD.yaml", + parameters={"path": "/srv/app/Config.PROD.yaml", "operation": "write"}, + ), + ), +] + + +def execute(action: ActionRequest) -> str: + """Stand-in for your tool runtime.""" + return f"ok: {action.tool} {action.target}" + + +def run(base_url: str, actor: str, phase: int) -> int: + gate = AetheryaClient(base_url, actor=actor, phase=phase) + + try: + health = gate.health() + except Exception as exc: + print(f" cannot reach {base_url}: {exc}") + print(" start it with: docker compose up -d") + return 2 + + print(f"\n sidecar {base_url} · phase {phase} · actor {actor}") + print( + f" service: ok={health.get('ok')} degraded={health.get('degraded')} " + f"policy={str(health.get('effective_fingerprint'))[:22]}..." + ) + print(" " + "─" * 70) + + gaps = 0 + for step, (why, action) in enumerate(PROPOSALS, start=1): + verdict = gate.check(action) + + print(f"\n {step}. {action.tool}({action.target})") + print(f" why: {why}") + print(f" ÆTHERYA: state={verdict.state} risk={verdict.risk_score}") + + if verdict.execute: + result = execute(action) + marker = "⚠" if verdict.shadow_gap else "·" + print(f" {marker} EXECUTED: {result}") + if verdict.shadow_gap: + gaps += 1 + print(" a later phase would have refused this") + elif verdict.requires_confirmation: + print(f" ⏸ HELD: {verdict.reason}") + else: + # The refusal is what the model sees, not an exception. + print(f" ✕ REFUSED: {verdict.reason}") + + print("\n " + "─" * 70) + if gaps: + print(f"\n {gaps} action(s) executed that phase 2+ would refuse.") + print(" That is the phase-1 measurement. Read the full picture with:") + print(" aetherya rollout report --phase 1 --audit-path audit/decisions.jsonl\n") + else: + print("\n nothing a later phase would have refused\n") + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--url", default="http://localhost:8080") + parser.add_argument("--actor", default="robert") + parser.add_argument("--phase", type=int, default=1, choices=[1, 2, 3]) + args: Any = parser.parse_args(argv) + return run(str(args.url), str(args.actor), int(args.phase)) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/aetherya/api.py b/src/aetherya/api.py index c6b9856..e77a2ad 100644 --- a/src/aetherya/api.py +++ b/src/aetherya/api.py @@ -26,6 +26,7 @@ from aetherya.constitution import Constitution, is_model_warm from aetherya.parser import parse_user_input from aetherya.pipeline import run_pipeline +from aetherya.rate_limiter import RateLimitConfig, RateLimiter, build_rate_limiter @dataclass(frozen=True) @@ -131,6 +132,42 @@ def _header_value(headers: dict[str, Any] | None, key: str) -> str: class AetheryaAPI: def __init__(self, settings: APISettings | None = None): self.settings = settings or APISettings() + # Built once and reused: the in-process backend keeps its windows on the + # instance, so rebuilding per request would reset every limit and make + # the limiter a no-op. Rebuilt only if the policy's rate-limit section + # changes underneath us. + self._rate_limiter: RateLimiter | None = None + self._rate_limiter_key: tuple[Any, ...] | None = None + + def _resolve_rate_limiter(self, cfg: Any) -> RateLimiter | None: + rl = getattr(cfg, "rate_limit", None) + if rl is None: + return None + + key = ( + rl.backend, + rl.requests_per_window, + rl.window_seconds, + rl.max_actors, + rl.redis_url_env, + rl.redis_prefix, + ) + if self._rate_limiter is not None and self._rate_limiter_key == key: + return self._rate_limiter + + limiter = build_rate_limiter( + rl.backend, + RateLimitConfig( + requests_per_window=rl.requests_per_window, + window_seconds=rl.window_seconds, + max_actors=rl.max_actors, + ), + redis_url_env=rl.redis_url_env, + redis_prefix=rl.redis_prefix, + ) + self._rate_limiter = limiter + self._rate_limiter_key = key + return limiter def _resolve_constitution(self) -> Constitution: path = self.settings.constitution_path @@ -276,6 +313,7 @@ def decide(self, payload: Any) -> tuple[int, dict[str, Any]]: audit=audit, response_text=candidate_response, action=action, + rate_limiter=self._resolve_rate_limiter(cfg_effective), ) event = _maybe_read_last_event(audit_path) if audit_path is not None else None diff --git a/src/aetherya/cli.py b/src/aetherya/cli.py index b117e30..0607348 100644 --- a/src/aetherya/cli.py +++ b/src/aetherya/cli.py @@ -32,6 +32,7 @@ from aetherya.parser import parse_user_input from aetherya.pipeline import run_pipeline from aetherya.pipeline_benchmark import main as pipeline_benchmark_main +from aetherya.rollout_report import main as rollout_report_main from aetherya.security_baseline import main as security_baseline_main from aetherya.security_gate import main as security_gate_main from aetherya.verify_release_artifacts import main as verify_release_artifacts_main @@ -488,6 +489,16 @@ def _build_parser() -> argparse.ArgumentParser: warmup_parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON.") warmup_parser.set_defaults(handler=_cmd_warmup) + rollout_parser = subparsers.add_parser("rollout", help="Rollout phase tooling.") + rollout_subparsers = rollout_parser.add_subparsers(dest="rollout_command") + rollout_subparsers.required = True + _add_forward_command( + rollout_subparsers, + name="report", + help_text="Measure the current rollout phase against its exit criteria.", + target_main=rollout_report_main, + ) + policy_parser = subparsers.add_parser("policy", help="Policy inspection tooling.") policy_subparsers = policy_parser.add_subparsers(dest="policy_command") policy_subparsers.required = True diff --git a/src/aetherya/client.py b/src/aetherya/client.py new file mode 100644 index 0000000..0b3146b --- /dev/null +++ b/src/aetherya/client.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +import json +import urllib.error +import urllib.request +from dataclasses import dataclass +from typing import Any + +from aetherya.actions import ActionRequest, Decision +from aetherya.enforcement import Enforcement, EnforcementPhase, apply_enforcement, resolve_phase + +# --------------------------------------------------------------------------- +# Sidecar client. +# +# The agent runs in one process, the decision service in another. This is the +# whole integration surface: build an ActionRequest, ask, act on the answer. +# +# gate = AetheryaClient("http://aetherya:8080", actor="deploy-bot", phase=1) +# verdict = gate.check(action) +# if verdict.execute: +# result = run_the_tool() +# else: +# result = verdict.reason # goes back to the model as tool output +# +# FAIL-CLOSED: an unreachable service refuses the action. A decision boundary +# that disappears when the network hiccups is not a boundary. Phase 1 is the one +# exception and it is explicit — see `fail_open_in_shadow`. +# --------------------------------------------------------------------------- + + +class AetheryaUnavailable(RuntimeError): + """The decision service could not be reached or returned an unusable answer.""" + + +@dataclass(frozen=True) +class Verdict: + """What the caller should do, plus the decision it came from.""" + + execute: bool + requires_confirmation: bool + reason: str + state: str + risk_score: int + shadow_gap: bool + decision: dict[str, Any] + enforcement: Enforcement | None = None + + @property + def refused(self) -> bool: + return not self.execute + + +class AetheryaClient: + """ + Minimal HTTP client for the decision service. + + Deliberately dependency-free (urllib, not requests): this sits in the hot + path of every tool call an agent makes, and the fewer things that can break + or need upgrading there, the better. + """ + + def __init__( + self, + base_url: str, + *, + actor: str, + phase: int | EnforcementPhase = 1, + timeout_sec: float = 5.0, + fail_open_in_shadow: bool = True, + ) -> None: + self.base_url = base_url.rstrip("/") + self.actor = actor + self.phase: EnforcementPhase = ( + phase if isinstance(phase, EnforcementPhase) else resolve_phase(phase) + ) + self.timeout_sec = float(timeout_sec) + # Phase 1 is observation: the service being down must not stop the agent, + # because nothing is being enforced anyway. From phase 2 on, an outage + # refuses the action — that is the point of enforcing. + self.fail_open_in_shadow = bool(fail_open_in_shadow) + + # -- transport ------------------------------------------------------------ + + def _post(self, path: str, payload: dict[str, Any]) -> dict[str, Any]: + request = urllib.request.Request( # noqa: S310 - operator-configured URL + f"{self.base_url}{path}", + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + opened = urllib.request.urlopen(request, timeout=self.timeout_sec) # noqa: S310 + with opened as response: + body = json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", "ignore")[:200] + raise AetheryaUnavailable(f"HTTP {exc.code}: {detail}") from exc + except Exception as exc: + raise AetheryaUnavailable(f"{type(exc).__name__}: {exc}") from exc + + if not isinstance(body, dict): + raise AetheryaUnavailable("response was not a JSON object") + return body + + # -- public surface ------------------------------------------------------- + + def health(self) -> dict[str, Any]: + request = urllib.request.Request(f"{self.base_url}/health") # noqa: S310 + try: + opened = urllib.request.urlopen(request, timeout=self.timeout_sec) # noqa: S310 + with opened as response: + body = json.loads(response.read().decode("utf-8")) + except Exception as exc: + raise AetheryaUnavailable(f"{type(exc).__name__}: {exc}") from exc + return dict(body) if isinstance(body, dict) else {} + + def check( + self, + action: ActionRequest, + *, + candidate_response: str | None = None, + ) -> Verdict: + """Ask the service about an action and apply the configured phase.""" + payload: dict[str, Any] = { + "actor": self.actor, + "wait_shadow": False, + "action": { + "raw_input": action.raw_input, + "intent": action.intent, + "mode_hint": action.mode_hint, + "tool": action.tool, + "target": action.target, + "parameters": dict(action.parameters), + }, + } + if candidate_response is not None: + payload["candidate_response"] = candidate_response + + try: + body = self._post("/v1/decide", payload) + except AetheryaUnavailable as exc: + return self._unavailable_verdict(exc) + + decision = body.get("decision") + if not isinstance(decision, dict) or "state" not in decision: + return self._unavailable_verdict( + AetheryaUnavailable("response did not contain a decision") + ) + + enforcement = apply_enforcement( + Decision( + allowed=bool(decision.get("allowed", False)), + risk_score=int(decision.get("risk_score", 0)), + reason=str(decision.get("reason", "")), + violated_principle=decision.get("violated_principle"), + mode=decision.get("mode"), + state=str(decision["state"]), + ), + self.phase, + ) + return Verdict( + execute=enforcement.execute, + requires_confirmation=enforcement.requires_confirmation, + reason=( + str(decision.get("reason", "")) if not enforcement.execute else enforcement.reason + ), + state=enforcement.state, + risk_score=int(decision.get("risk_score", 0)), + shadow_gap=enforcement.shadow_gap, + decision=decision, + enforcement=enforcement, + ) + + def _unavailable_verdict(self, exc: AetheryaUnavailable) -> Verdict: + shadow = self.phase.number == 1 and self.fail_open_in_shadow + reason = ( + f"aetherya unavailable ({exc}) — phase 1 observes only, action proceeds" + if shadow + else f"REFUSED: aetherya unavailable ({exc})" + ) + return Verdict( + execute=shadow, + requires_confirmation=False, + reason=reason, + state="unavailable", + risk_score=0, + shadow_gap=False, + decision={}, + ) diff --git a/src/aetherya/config.py b/src/aetherya/config.py index 420d689..d347992 100644 --- a/src/aetherya/config.py +++ b/src/aetherya/config.py @@ -234,6 +234,21 @@ class IntentEscalationConfig: use_shape_signals: bool = True +@dataclass(frozen=True) +class EnforcementConfig: + """ + Rollout posture. The engine always computes a full decision; the phase only + decides how much of it is enforced, so advancing is a config change rather + than a code change. See docs/rollout-phases.md. + + 1 shadow — observe only + 2 hard_deny — refuse hard_deny + 3 full — refuse hard_deny and deny, hold escalate for confirmation + """ + + phase: int = 1 + + @dataclass(frozen=True) class PolicyConfig: version: int @@ -250,6 +265,7 @@ class PolicyConfig: constitution_config: ConstitutionConfig = field(default_factory=ConstitutionConfig) intent_escalation: IntentEscalationConfig = field(default_factory=IntentEscalationConfig) rate_limit: RateLimitBackendConfig = field(default_factory=RateLimitBackendConfig) + enforcement: EnforcementConfig = field(default_factory=EnforcementConfig) effective_fingerprint: str | None = None @@ -603,6 +619,14 @@ def _validate_semantic_authority( ) +def _load_enforcement(raw: dict[str, Any] | None) -> EnforcementConfig: + data = raw or {} + phase = int(data.get("phase", 1)) + if phase not in {1, 2, 3}: + raise ValueError("enforcement.phase must be 1, 2 or 3") + return EnforcementConfig(phase=phase) + + def _load_intent_escalation(raw: dict[str, Any] | None) -> IntentEscalationConfig: data = raw or {} return IntentEscalationConfig( @@ -688,6 +712,7 @@ def load_policy_config( constitution_config = _load_constitution_config(data.get("constitution")) intent_escalation = _load_intent_escalation(data.get("intent_escalation")) rate_limit = _load_rate_limit(data.get("rate_limit")) + enforcement = _load_enforcement(data.get("enforcement")) _validate_semantic_authority(constitution_config, modes) config = PolicyConfig( @@ -705,6 +730,7 @@ def load_policy_config( constitution_config=constitution_config, intent_escalation=intent_escalation, rate_limit=rate_limit, + enforcement=enforcement, ) config = replace(config, effective_fingerprint=compute_effective_fingerprint(config)) diff --git a/src/aetherya/enforcement.py b/src/aetherya/enforcement.py new file mode 100644 index 0000000..1762377 --- /dev/null +++ b/src/aetherya/enforcement.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Final + +from aetherya.actions import Decision + +# --------------------------------------------------------------------------- +# Rollout phases as configuration. +# +# The engine always computes a full decision. A phase only decides how much of +# it is enforced. Keeping that split here — rather than in each integration — +# is what makes "deploy in observation mode, tighten later" a config change +# instead of a code change, and what makes the shadow gap measurable: in phase 1 +# we still know what a later phase *would* have refused. +# +# See docs/rollout-phases.md. +# --------------------------------------------------------------------------- + +# States a later phase would act on, ordered from most to least severe. +_ENFORCEABLE: Final[frozenset[str]] = frozenset({"hard_deny", "deny", "escalate"}) + + +@dataclass(frozen=True) +class EnforcementPhase: + number: int + name: str + blocks: frozenset[str] + confirms: frozenset[str] + description: str + + +PHASES: Final[dict[int, EnforcementPhase]] = { + 1: EnforcementPhase( + number=1, + name="shadow", + blocks=frozenset(), + confirms=frozenset(), + description="observe only — every action executes, every decision is recorded", + ), + 2: EnforcementPhase( + number=2, + name="hard_deny", + blocks=frozenset({"hard_deny"}), + confirms=frozenset(), + description="refuse irreversible destruction, jailbreaks and unlisted tools", + ), + 3: EnforcementPhase( + number=3, + name="full", + blocks=frozenset({"hard_deny", "deny"}), + confirms=frozenset({"escalate"}), + description="refuse denials and hold escalations for human confirmation", + ), +} + +MIN_PHASE: Final[int] = min(PHASES) +MAX_PHASE: Final[int] = max(PHASES) + + +def resolve_phase(number: int) -> EnforcementPhase: + if number not in PHASES: + raise ValueError(f"enforcement.phase must be one of {sorted(PHASES)}, got {number!r}") + return PHASES[number] + + +@dataclass(frozen=True) +class Enforcement: + """What the caller should actually do with a decision, under a given phase.""" + + phase: int + state: str + execute: bool + requires_confirmation: bool + reason: str + # True when this executed but a later phase would have refused it. This is + # the number phase 1 exists to produce; it is meaningless once phase 3 is on. + shadow_gap: bool + + def to_dict(self) -> dict[str, object]: + return { + "phase": self.phase, + "state": self.state, + "execute": self.execute, + "requires_confirmation": self.requires_confirmation, + "reason": self.reason, + "shadow_gap": self.shadow_gap, + } + + +def apply_enforcement(decision: Decision, phase: EnforcementPhase | int) -> Enforcement: + """ + Map a decision to an action under the configured phase. + + Never mutates the decision: the audit trail records what the engine ruled, + not what a partially-enforcing deployment chose to do about it. The two are + different facts and conflating them would make phase 1 data worthless. + """ + resolved = phase if isinstance(phase, EnforcementPhase) else resolve_phase(phase) + state = decision.state + + if state in resolved.blocks: + return Enforcement( + phase=resolved.number, + state=state, + execute=False, + requires_confirmation=False, + reason=f"refused: `{state}` is enforced in phase {resolved.number}", + shadow_gap=False, + ) + + if state in resolved.confirms: + return Enforcement( + phase=resolved.number, + state=state, + execute=False, + requires_confirmation=True, + reason=f"held: `{state}` requires human confirmation in phase {resolved.number}", + shadow_gap=False, + ) + + gap = state in _ENFORCEABLE + return Enforcement( + phase=resolved.number, + state=state, + execute=True, + requires_confirmation=False, + reason=( + f"executed: phase {resolved.number} does not enforce `{state}`" + if gap + else "allowed by policy" + ), + shadow_gap=gap, + ) diff --git a/src/aetherya/pipeline.py b/src/aetherya/pipeline.py index f204cae..fd47fbc 100644 --- a/src/aetherya/pipeline.py +++ b/src/aetherya/pipeline.py @@ -47,7 +47,7 @@ ) from aetherya.policy_engine import DecisionState, PolicyEngine from aetherya.procedural_guard import ProceduralGuard -from aetherya.rate_limiter import ActorRateLimiter +from aetherya.rate_limiter import RateLimiter from aetherya.risk import RiskAggregator, RiskDecision, RiskSignal @@ -286,7 +286,7 @@ def run_pipeline( cfg: PolicyConfig, audit: AuditLogger | None = None, response_text: str | None = None, - rate_limiter: ActorRateLimiter | None = None, + rate_limiter: RateLimiter | None = None, action: ActionRequest | None = None, ) -> Decision: """ @@ -957,7 +957,7 @@ def run_pipeline_structured( cfg: PolicyConfig, audit: AuditLogger | None = None, response_text: str | None = None, - rate_limiter: ActorRateLimiter | None = None, + rate_limiter: RateLimiter | None = None, ) -> Decision: """ Preferred entry point for programmatic integrations. diff --git a/src/aetherya/rollout_report.py b/src/aetherya/rollout_report.py new file mode 100644 index 0000000..9600849 --- /dev/null +++ b/src/aetherya/rollout_report.py @@ -0,0 +1,327 @@ +from __future__ import annotations + +import argparse +import json +import sys +from collections import Counter +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Any + +from aetherya.audit_verify import verify_audit_file +from aetherya.enforcement import PHASES, resolve_phase + +# --------------------------------------------------------------------------- +# Phase-1 measurement. +# +# The rollout plan says "review every hard_deny event and confirm it is a true +# positive" and "at least 2 weeks or 10 000 decisions". Without tooling that is +# a grep and a guess, so nobody does it and the phase gate becomes a formality. +# This reads the audit trail and answers the exit criteria directly, including +# printing the hard_deny events that need human eyes on them. +# +# See docs/rollout-phases.md. +# --------------------------------------------------------------------------- + +DEFAULT_MIN_DECISIONS = 10_000 +DEFAULT_MIN_DAYS = 14.0 + + +@dataclass +class Criterion: + name: str + passed: bool + detail: str + + def to_dict(self) -> dict[str, Any]: + return {"name": self.name, "passed": self.passed, "detail": self.detail} + + +@dataclass +class RolloutReport: + audit_path: str + current_phase: int + total_decisions: int = 0 + window_days: float = 0.0 + first_event: str | None = None + last_event: str | None = None + states: dict[str, int] = field(default_factory=dict) + would_block_next_phase: int = 0 + actors_by_state: dict[str, list[str]] = field(default_factory=dict) + tools_by_state: dict[str, list[str]] = field(default_factory=dict) + intent_escalations: int = 0 + semantic_skipped: int = 0 + policy_fingerprints: list[str] = field(default_factory=list) + chain_valid: bool = False + chain_errors: int = 0 + chain_error_detail: str | None = None + hard_deny_events: list[dict[str, Any]] = field(default_factory=list) + criteria: list[Criterion] = field(default_factory=list) + + @property + def ready_to_advance(self) -> bool: + return bool(self.criteria) and all(c.passed for c in self.criteria) + + def to_dict(self) -> dict[str, Any]: + return { + "audit_path": self.audit_path, + "current_phase": self.current_phase, + "next_phase": min(self.current_phase + 1, max(PHASES)), + "ready_to_advance": self.ready_to_advance, + "window": { + "total_decisions": self.total_decisions, + "days": round(self.window_days, 2), + "first_event": self.first_event, + "last_event": self.last_event, + }, + "states": self.states, + "would_block_next_phase": self.would_block_next_phase, + "actors_by_state": self.actors_by_state, + "tools_by_state": self.tools_by_state, + "intent_escalations": self.intent_escalations, + "semantic_skipped": self.semantic_skipped, + "policy_fingerprints": self.policy_fingerprints, + "chain": { + "valid": self.chain_valid, + "errors": self.chain_errors, + "detail": self.chain_error_detail, + }, + "hard_deny_events": self.hard_deny_events, + "criteria": [c.to_dict() for c in self.criteria], + } + + +def _parse_ts(value: Any) -> datetime | None: + if not isinstance(value, str) or not value.strip(): + return None + try: + return datetime.fromisoformat(value) + except ValueError: + return None + + +def _load_events(path: Path) -> list[dict[str, Any]]: + if not path.exists(): + raise ValueError(f"audit file not found: {path}") + events: list[dict[str, Any]] = [] + for line in path.read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + try: + payload = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(payload, dict): + events.append(payload) + return events + + +def build_report( + audit_path: str | Path, + *, + current_phase: int = 1, + min_decisions: int = DEFAULT_MIN_DECISIONS, + min_days: float = DEFAULT_MIN_DAYS, + max_hard_deny_samples: int = 50, +) -> RolloutReport: + path = Path(audit_path) + phase = resolve_phase(current_phase) + next_phase = PHASES[min(current_phase + 1, max(PHASES))] + events = _load_events(path) + + report = RolloutReport(audit_path=str(path), current_phase=phase.number) + report.total_decisions = len(events) + + states: Counter[str] = Counter() + actors: dict[str, set[str]] = {} + tools: dict[str, set[str]] = {} + fingerprints: set[str] = set() + timestamps: list[datetime] = [] + + for event in events: + decision = event.get("decision") or {} + context = event.get("context") or {} + state = str(decision.get("state", "unknown")) + states[state] += 1 + + actors.setdefault(state, set()).add(str(event.get("actor", "unknown"))) + action_ctx = context.get("action") if isinstance(context.get("action"), dict) else {} + tool = action_ctx.get("tool") if isinstance(action_ctx, dict) else None + tools.setdefault(state, set()).add(str(tool) if tool else "-") + + if context.get("intent_escalation"): + report.intent_escalations += 1 + constitution = context.get("constitution") + if isinstance(constitution, dict) and constitution.get("semantic_skipped"): + report.semantic_skipped += 1 + + fingerprint = context.get("policy_fingerprint") + if isinstance(fingerprint, str) and fingerprint: + fingerprints.add(fingerprint) + + ts = _parse_ts(event.get("ts")) + if ts is not None: + timestamps.append(ts) + + # Every hard_deny needs human eyes before enforcement is switched on. + if state == "hard_deny" and len(report.hard_deny_events) < max_hard_deny_samples: + report.hard_deny_events.append( + { + "ts": event.get("ts"), + "actor": event.get("actor"), + "action": str(event.get("action", ""))[:160], + "reason": decision.get("reason"), + "risk_score": decision.get("risk_score"), + } + ) + + report.states = dict(sorted(states.items())) + report.actors_by_state = {k: sorted(v) for k, v in sorted(actors.items())} + report.tools_by_state = {k: sorted(v) for k, v in sorted(tools.items())} + report.policy_fingerprints = sorted(fingerprints) + report.would_block_next_phase = sum( + states[s] for s in (next_phase.blocks | next_phase.confirms) if s in states + ) + + if timestamps: + timestamps.sort() + report.first_event = timestamps[0].isoformat() + report.last_event = timestamps[-1].isoformat() + report.window_days = (timestamps[-1] - timestamps[0]).total_seconds() / 86400.0 + + # A malformed line or an empty file makes `verify_audit_file` raise. Both are + # findings an operator must see before advancing a phase — not reasons for + # the report to crash, and certainly not something to skip silently. + try: + verification = verify_audit_file(str(path), require_chain=True) + report.chain_errors = sum(1 for r in verification if not r.verification.valid) + report.chain_valid = report.chain_errors == 0 + except ValueError as exc: + report.chain_valid = False + report.chain_error_detail = str(exc) + + report.criteria = _evaluate_criteria(report, min_decisions=min_decisions, min_days=min_days) + return report + + +def _evaluate_criteria( + report: RolloutReport, *, min_decisions: int, min_days: float +) -> list[Criterion]: + volume_ok = report.total_decisions >= min_decisions or report.window_days >= min_days + criteria = [ + Criterion( + name="sufficient_window", + passed=volume_ok, + detail=( + f"{report.total_decisions} decisions over {report.window_days:.1f} days " + f"(need {min_decisions} or {min_days:.0f} days)" + ), + ), + Criterion( + name="chain_intact", + passed=report.chain_valid, + detail=( + "audit chain verifies clean" + if report.chain_valid + else (report.chain_error_detail or f"{report.chain_errors} invalid event(s)") + + " — investigate before advancing" + ), + ), + Criterion( + name="single_policy", + passed=len(report.policy_fingerprints) <= 1, + detail=( + f"{len(report.policy_fingerprints)} distinct policy fingerprint(s); more than " + "one means the window mixes different engines and must be restarted" + ), + ), + # Deliberately never auto-passes: a human has to look at these. The tool + # can list them and count them; it cannot judge them. + Criterion( + name="hard_deny_reviewed", + passed=False, + detail=( + f"{report.states.get('hard_deny', 0)} hard_deny event(s) require manual " + "review — confirm every one is a true positive, then advance" + ), + ), + ] + return criteria + + +def _print_text(report: RolloutReport) -> None: + phase = resolve_phase(report.current_phase) + nxt = PHASES[min(report.current_phase + 1, max(PHASES))] + + print(f"\n ROLLOUT REPORT — phase {phase.number} ({phase.name})") + print(f" {report.audit_path}") + print(" " + "─" * 68) + + print(f"\n window {report.total_decisions} decisions over {report.window_days:.1f} days") + if report.first_event: + print(f" {report.first_event} -> {report.last_event}") + + print("\n decisions by state") + for state, count in report.states.items(): + share = (count / report.total_decisions * 100) if report.total_decisions else 0.0 + actors = ", ".join(report.actors_by_state.get(state, [])[:4]) + print(f" {state:12} {count:6} {share:5.1f}% actors: {actors}") + + print( + f"\n phase {nxt.number} ({nxt.name}) would act on {report.would_block_next_phase} of these" + ) + print(f" intent escalations {report.intent_escalations}") + print(f" semantic skipped {report.semantic_skipped}") + print(f" policy fingerprints {len(report.policy_fingerprints)}") + print(f" audit chain {'intact' if report.chain_valid else 'INVALID'}") + + if report.hard_deny_events: + print(f"\n hard_deny events to review ({len(report.hard_deny_events)} shown)") + for event in report.hard_deny_events[:10]: + print(f" {event['ts']} {event['actor']}") + print(f" {event['action']}") + print(f" -> {event['reason']}") + + print("\n exit criteria") + for criterion in report.criteria: + mark = "ok " if criterion.passed else "-- " + print(f" {mark} {criterion.name}: {criterion.detail}") + + verdict = "READY" if report.ready_to_advance else "NOT READY" + print(f"\n {verdict} to advance to phase {nxt.number}\n") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Measure a rollout phase against its exit criteria." + ) + parser.add_argument("--audit-path", default="audit/decisions.jsonl") + parser.add_argument("--phase", type=int, default=1, choices=sorted(PHASES)) + parser.add_argument("--min-decisions", type=int, default=DEFAULT_MIN_DECISIONS) + parser.add_argument("--min-days", type=float, default=DEFAULT_MIN_DAYS) + parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON.") + args = parser.parse_args(argv) + + try: + report = build_report( + args.audit_path, + current_phase=int(args.phase), + min_decisions=int(args.min_decisions), + min_days=float(args.min_days), + ) + except ValueError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + if args.json: + print(json.dumps(report.to_dict(), ensure_ascii=False, indent=2, default=str)) + else: + _print_text(report) + + # Exit 1 when not ready: usable as a gate in a promotion pipeline. + return 0 if report.ready_to_advance else 1 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/src/aetherya/security_gate.py b/src/aetherya/security_gate.py index 5b8c4eb..f2c70a6 100644 --- a/src/aetherya/security_gate.py +++ b/src/aetherya/security_gate.py @@ -271,6 +271,13 @@ def _phase_integrity_fuzz( expected_reject_code: int, ) -> SecurityGatePhase: audit_path = workdir / "phase2_tamper_audit.jsonl" + # AuditLogger appends, by design — the chain must never be truncated in + # normal operation. This phase generates its own corpus, so a file left by a + # previous run would be appended to and counted, and the phase would fail on + # events it did not create. CI never sees it (fresh checkout); a second local + # run does, with a confusing "invalid == total" that suggests a real + # regression. Start from a clean file. + audit_path.unlink(missing_ok=True) logger = AuditLogger(str(audit_path), attestation_key=attestation_key) for idx in range(events): logger.log( diff --git a/tests/test_deployment_contracts.py b/tests/test_deployment_contracts.py index b9ab7b3..546936f 100644 --- a/tests/test_deployment_contracts.py +++ b/tests/test_deployment_contracts.py @@ -502,3 +502,115 @@ def test_cli_fingerprint_text_output(capsys: pytest.CaptureFixture[str]) -> None out = capsys.readouterr().out assert "effective_fingerprint" in out assert "policy_fingerprint" in out + + +# --------------------------------------------------------------------------- +# The HTTP API must actually apply the configured rate limit +# --------------------------------------------------------------------------- + + +def test_api_enforces_the_configured_rate_limit(tmp_path: Path) -> None: + """ + Regression: `rate_limit` was validated on load and `build_rate_limiter` + worked, but nothing in the HTTP path ever constructed a limiter — the + configured limit was documentation, not behaviour. + """ + data = yaml.safe_load(Path(_POLICY).read_text(encoding="utf-8")) + data["rate_limit"].update({"backend": "memory", "requests_per_window": 3}) + policy = tmp_path / "policy.yaml" + policy.write_text(yaml.safe_dump(data), encoding="utf-8") + + api = AetheryaAPI(APISettings(policy_path=policy, audit_path=tmp_path / "a.jsonl")) + states = [] + for _ in range(5): + _, body = api.decide({"raw_input": "help user", "actor": "chatty", "wait_shadow": False}) + states.append(body["decision"]["allowed"]) + + assert states[:3] == [True, True, True] + assert states[3:] == [False, False] + + +def test_the_limiter_is_built_once_and_reused(tmp_path: Path) -> None: + """Rebuilding per request would reset every window and make it a no-op.""" + api = AetheryaAPI(APISettings(audit_path=tmp_path / "a.jsonl")) + cfg = load_policy_config(_POLICY) + + first = api._resolve_rate_limiter(cfg) # noqa: SLF001 + assert api._resolve_rate_limiter(cfg) is first # noqa: SLF001 + + +def test_the_limiter_is_rebuilt_when_the_policy_changes(tmp_path: Path) -> None: + api = AetheryaAPI(APISettings(audit_path=tmp_path / "a.jsonl")) + cfg = load_policy_config(_POLICY) + first = api._resolve_rate_limiter(cfg) # noqa: SLF001 + + changed = replace(cfg, rate_limit=replace(cfg.rate_limit, requests_per_window=7)) + assert api._resolve_rate_limiter(changed) is not first # noqa: SLF001 + + +def test_a_config_without_a_rate_limit_section_disables_limiting(tmp_path: Path) -> None: + class _Cfg: + pass + + api = AetheryaAPI(APISettings(audit_path=tmp_path / "a.jsonl")) + assert api._resolve_rate_limiter(_Cfg()) is None # noqa: SLF001 + + +def test_the_pipeline_accepts_any_limiter_backend() -> None: + """ + The signature said `ActorRateLimiter`, so the Redis backend could not be + passed to the pipeline at all — mypy caught it once the API wired it up. + """ + import inspect + + from aetherya.pipeline import run_pipeline + + annotation = inspect.signature(run_pipeline).parameters["rate_limiter"].annotation + assert "RateLimiter" in str(annotation) + assert "ActorRateLimiter" not in str(annotation) + + +def test_the_slim_policy_differs_only_in_the_semantic_layer() -> None: + """ + `config/policy.slim.yaml` exists so the image can drop PyTorch (~8.7GB -> + ~330MB). That is only safe because the semantic layer is advisory: capped + below every deny threshold, it can escalate to a human but never refuse + alone, so disabling it changes nothing about what is hard-denied. Three + policy files is a drift hazard, so the difference is pinned to one field. + """ + from dataclasses import asdict + + docker = load_policy_config("config/policy.docker.yaml") + slim = load_policy_config("config/policy.slim.yaml") + + assert docker.constitution_config.use_semantic is True + assert slim.constitution_config.use_semantic is False + + ignored = {"policy_fingerprint", "effective_fingerprint", "constitution_config"} + assert {k: v for k, v in asdict(docker).items() if k not in ignored} == { + k: v for k, v in asdict(slim).items() if k not in ignored + } + assert replace(docker.constitution_config, use_semantic=False) == slim.constitution_config + + +def test_the_deployment_policy_differs_only_in_the_rate_limit_backend() -> None: + """ + `config/policy.docker.yaml` exists so the container gets the distributed + limiter while the repo default keeps working with no infrastructure. Two + policy files is a drift hazard, so the difference is pinned to exactly one + field — anything else changing in one and not the other fails here. + """ + from dataclasses import asdict + + repo = load_policy_config(_POLICY) + docker = load_policy_config("config/policy.docker.yaml") + + assert repo.rate_limit.backend == "memory" + assert docker.rate_limit.backend == "redis" + + ignored = {"policy_fingerprint", "effective_fingerprint", "rate_limit"} + repo_fields = {k: v for k, v in asdict(repo).items() if k not in ignored} + docker_fields = {k: v for k, v in asdict(docker).items() if k not in ignored} + assert repo_fields == docker_fields + + assert replace(repo.rate_limit, backend="redis") == docker.rate_limit diff --git a/tests/test_enforcement_and_client.py b/tests/test_enforcement_and_client.py new file mode 100644 index 0000000..f9e2f42 --- /dev/null +++ b/tests/test_enforcement_and_client.py @@ -0,0 +1,304 @@ +from __future__ import annotations + +import json +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from aetherya.actions import ActionRequest, Decision +from aetherya.client import AetheryaClient, AetheryaUnavailable +from aetherya.config import load_policy_config +from aetherya.enforcement import PHASES, apply_enforcement, resolve_phase + +_POLICY = "config/policy.yaml" + + +def _decision(state: str, *, risk: int = 0) -> Decision: + return Decision( + allowed=state in {"allow", "log_only"}, + risk_score=risk, + reason=f"{state}: test", + mode="operative", + state=state, + ) + + +# --------------------------------------------------------------------------- +# Phase semantics +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("state", ["allow", "log_only", "escalate", "deny", "hard_deny"]) +def test_phase_1_never_refuses(state: str) -> None: + """Shadow must not change behaviour, only observe it.""" + result = apply_enforcement(_decision(state), 1) + assert result.execute is True + assert result.requires_confirmation is False + + +@pytest.mark.parametrize( + ("state", "executes"), + [("allow", True), ("log_only", True), ("escalate", True), ("deny", True), ("hard_deny", False)], +) +def test_phase_2_refuses_only_hard_deny(state: str, executes: bool) -> None: + assert apply_enforcement(_decision(state), 2).execute is executes + + +def test_phase_3_refuses_denials_and_holds_escalations() -> None: + assert apply_enforcement(_decision("hard_deny"), 3).execute is False + assert apply_enforcement(_decision("deny"), 3).execute is False + + held = apply_enforcement(_decision("escalate"), 3) + assert held.execute is False + assert held.requires_confirmation is True + + assert apply_enforcement(_decision("allow"), 3).execute is True + + +@pytest.mark.parametrize("state", ["hard_deny", "deny", "escalate"]) +def test_shadow_gap_marks_what_a_later_phase_would_refuse(state: str) -> None: + """The number phase 1 exists to produce.""" + assert apply_enforcement(_decision(state), 1).shadow_gap is True + + +@pytest.mark.parametrize("state", ["allow", "log_only"]) +def test_allowed_states_are_never_a_shadow_gap(state: str) -> None: + assert apply_enforcement(_decision(state), 1).shadow_gap is False + + +def test_shadow_gap_is_false_once_the_phase_enforces() -> None: + assert apply_enforcement(_decision("hard_deny"), 2).shadow_gap is False + + +def test_enforcement_tightens_monotonically() -> None: + """Each phase must refuse at least as much as the one before it.""" + states = ["allow", "log_only", "escalate", "deny", "hard_deny"] + refused = [ + sum(1 for s in states if not apply_enforcement(_decision(s), p).execute) + for p in sorted(PHASES) + ] + assert refused == sorted(refused) + assert refused[0] == 0 + + +def test_enforcement_never_mutates_the_decision() -> None: + """ + The audit trail records what the engine ruled, not what a partially + enforcing deployment did about it. Conflating them makes phase-1 data + worthless. + """ + decision = _decision("hard_deny", risk=164) + apply_enforcement(decision, 1) + assert decision.state == "hard_deny" + assert decision.allowed is False + + +def test_unknown_phase_is_rejected() -> None: + with pytest.raises(ValueError, match="enforcement.phase must be one of"): + resolve_phase(9) + + +def test_policy_exposes_the_phase() -> None: + assert load_policy_config(_POLICY).enforcement.phase == 1 + + +def test_policy_rejects_an_invalid_phase(tmp_path: Path) -> None: + data = yaml.safe_load(Path(_POLICY).read_text(encoding="utf-8")) + data["enforcement"]["phase"] = 4 + path = tmp_path / "policy.yaml" + path.write_text(yaml.safe_dump(data), encoding="utf-8") + + with pytest.raises(ValueError, match="enforcement.phase must be 1, 2 or 3"): + load_policy_config(path) + + +def test_enforcement_serializes() -> None: + payload = apply_enforcement(_decision("hard_deny"), 2).to_dict() + assert payload["execute"] is False + assert payload["phase"] == 2 + + +# --------------------------------------------------------------------------- +# Sidecar client +# --------------------------------------------------------------------------- + + +class _Service(BaseHTTPRequestHandler): + state = "allow" + risk = 0 + status = 200 + body_override: str | None = None + requests: list[dict[str, Any]] = [] + + def _send(self, payload: dict[str, Any], status: int = 200) -> None: + raw = json.dumps(payload).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def do_GET(self) -> None: # noqa: N802 + self._send({"ok": True, "degraded": False, "effective_fingerprint": "sha256:test"}) + + def do_POST(self) -> None: # noqa: N802 + length = int(self.headers.get("Content-Length", 0)) + type(self).requests.append(json.loads(self.rfile.read(length) or b"{}")) + if type(self).body_override is not None: + raw = type(self).body_override.encode("utf-8") + self.send_response(type(self).status) + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + return + self._send( + { + "ok": True, + "decision": { + "state": type(self).state, + "allowed": type(self).state in {"allow", "log_only"}, + "risk_score": type(self).risk, + "reason": f"{type(self).state}: from service", + }, + }, + type(self).status, + ) + + def log_message(self, *args: Any) -> None: + pass + + +@pytest.fixture +def service() -> Any: + _Service.state = "allow" + _Service.risk = 0 + _Service.status = 200 + _Service.body_override = None + _Service.requests = [] + server = HTTPServer(("127.0.0.1", 0), _Service) + threading.Thread(target=server.serve_forever, daemon=True).start() + yield server + server.shutdown() + + +def _client(service: Any, **kwargs: Any) -> AetheryaClient: + return AetheryaClient(f"http://127.0.0.1:{service.server_port}", actor="test-agent", **kwargs) + + +def _action() -> ActionRequest: + return ActionRequest( + raw_input="rm -rf /", + intent="operate", + mode_hint="operative", + tool="shell", + parameters={"command": "rm -rf /"}, + ) + + +def test_client_sends_the_structured_action(service: Any) -> None: + """No free-text parsing over the wire: the agent declares its action.""" + _client(service).check(_action()) + + sent = _Service.requests[-1] + assert sent["action"]["tool"] == "shell" + assert sent["action"]["intent"] == "operate" + assert sent["actor"] == "test-agent" + + +def test_client_applies_the_phase(service: Any) -> None: + _Service.state = "hard_deny" + _Service.risk = 164 + + assert _client(service, phase=1).check(_action()).execute is True + assert _client(service, phase=2).check(_action()).execute is False + + +def test_client_surfaces_the_shadow_gap(service: Any) -> None: + _Service.state = "hard_deny" + verdict = _client(service, phase=1).check(_action()) + + assert verdict.execute is True + assert verdict.shadow_gap is True + assert verdict.refused is False + + +def test_client_holds_escalations_in_phase_3(service: Any) -> None: + _Service.state = "escalate" + verdict = _client(service, phase=3).check(_action()) + + assert verdict.execute is False + assert verdict.requires_confirmation is True + + +def test_refusal_carries_the_engine_reason(service: Any) -> None: + """What goes back to the model must say why, not just that it was refused.""" + _Service.state = "hard_deny" + verdict = _client(service, phase=2).check(_action()) + assert "from service" in verdict.reason + + +def test_candidate_response_is_forwarded(service: Any) -> None: + _client(service).check(_action(), candidate_response="the answer") + assert _Service.requests[-1]["candidate_response"] == "the answer" + + +# -- failure behaviour ------------------------------------------------------- + + +def test_unreachable_service_refuses_from_phase_2(service: Any) -> None: + """A boundary that disappears when the network hiccups is not a boundary.""" + dead = AetheryaClient("http://127.0.0.1:1", actor="a", phase=2, timeout_sec=0.5) + verdict = dead.check(_action()) + + assert verdict.execute is False + assert "unavailable" in verdict.reason + + +def test_unreachable_service_proceeds_in_phase_1(service: Any) -> None: + """Phase 1 enforces nothing, so an outage must not stop the agent.""" + dead = AetheryaClient("http://127.0.0.1:1", actor="a", phase=1, timeout_sec=0.5) + assert dead.check(_action()).execute is True + + +def test_phase_1_fail_open_can_be_disabled() -> None: + dead = AetheryaClient( + "http://127.0.0.1:1", actor="a", phase=1, timeout_sec=0.5, fail_open_in_shadow=False + ) + assert dead.check(_action()).execute is False + + +def test_an_http_error_is_treated_as_unavailable(service: Any) -> None: + _Service.status = 500 + _Service.body_override = '{"ok": false}' + assert _client(service, phase=2).check(_action()).execute is False + + +def test_a_response_without_a_decision_is_treated_as_unavailable(service: Any) -> None: + _Service.body_override = '{"ok": true}' + verdict = _client(service, phase=2).check(_action()) + assert verdict.execute is False + assert verdict.state == "unavailable" + + +def test_a_non_object_response_is_rejected(service: Any) -> None: + _Service.body_override = "[1, 2, 3]" + assert _client(service, phase=2).check(_action()).execute is False + + +def test_health_is_readable(service: Any) -> None: + assert _client(service).health()["ok"] is True + + +def test_health_raises_when_unreachable() -> None: + dead = AetheryaClient("http://127.0.0.1:1", actor="a", timeout_sec=0.5) + with pytest.raises(AetheryaUnavailable): + dead.health() + + +def test_base_url_trailing_slash_is_normalised(service: Any) -> None: + client = AetheryaClient(f"http://127.0.0.1:{service.server_port}/", actor="a", phase=1) + assert client.check(_action()).state == "allow" diff --git a/tests/test_examples_agent_loop.py b/tests/test_examples_agent_loop.py index 5fd731a..dccc6c0 100644 --- a/tests/test_examples_agent_loop.py +++ b/tests/test_examples_agent_loop.py @@ -153,6 +153,29 @@ def test_example_policy_is_loadable_and_knows_the_demo_actor(agent_loop: Any) -> def test_openai_backend_requires_a_key(monkeypatch: pytest.MonkeyPatch, agent_loop: Any) -> None: + """ + Must hold whether or not the SDK is installed. Checking the import first + made this assertion depend on the environment: it passed locally, where + `openai` was present, and failed CI, where the `dev` extra does not pull it. + """ monkeypatch.delenv("OPENAI_API_KEY", raising=False) with pytest.raises(SystemExit, match="OPENAI_API_KEY"): agent_loop.OpenAIAgent() + + +def test_openai_backend_reports_a_missing_sdk( + monkeypatch: pytest.MonkeyPatch, agent_loop: Any +) -> None: + import builtins + + monkeypatch.setenv("OPENAI_API_KEY", "sk-test") + real_import = builtins.__import__ + + def _blocked(name: str, *args: Any, **kwargs: Any) -> Any: + if name == "openai" or name.startswith("openai."): + raise ImportError("No module named 'openai'") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", _blocked) + with pytest.raises(SystemExit, match="openai is not installed"): + agent_loop.OpenAIAgent() diff --git a/tests/test_rollout_report.py b/tests/test_rollout_report.py new file mode 100644 index 0000000..f714d8e --- /dev/null +++ b/tests/test_rollout_report.py @@ -0,0 +1,333 @@ +from __future__ import annotations + +import json +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + +import pytest + +from aetherya.audit import AuditLogger +from aetherya.rollout_report import build_report, main + + +def _audit(tmp_path: Path, rows: list[tuple[str, str, str]]) -> Path: + """rows: (actor, state, reason). Written through AuditLogger so the chain is real.""" + path = tmp_path / "decisions.jsonl" + logger = AuditLogger(str(path), policy_fingerprint="sha256:test-policy") + for actor, state, reason in rows: + logger.log( + actor=actor, + action=f"{state} action for {actor}", + decision={ + "allowed": state in {"allow", "log_only"}, + "risk_score": 100 if state == "hard_deny" else 0, + "reason": reason, + "state": state, + }, + context={"mode": "operative", "action": {"tool": "shell"}}, + ) + return path + + +def test_counts_decisions_by_state(tmp_path: Path) -> None: + path = _audit( + tmp_path, + [ + ("a", "allow", "ok"), + ("a", "allow", "ok"), + ("b", "hard_deny", "destructive system command"), + ("c", "escalate", "confirmation missing"), + ], + ) + report = build_report(path) + + assert report.total_decisions == 4 + assert report.states == {"allow": 2, "escalate": 1, "hard_deny": 1} + assert report.actors_by_state["allow"] == ["a"] + + +def test_reports_what_the_next_phase_would_act_on(tmp_path: Path) -> None: + """The number phase 1 exists to produce.""" + path = _audit( + tmp_path, + [("a", "allow", "ok"), ("b", "hard_deny", "bad"), ("c", "escalate", "hold")], + ) + + # Phase 2 only enforces hard_deny. + assert build_report(path, current_phase=1).would_block_next_phase == 1 + # Phase 3 adds deny and escalate, so it also acts on the escalation. + assert build_report(path, current_phase=2).would_block_next_phase == 2 + + +def test_lists_hard_deny_events_for_manual_review(tmp_path: Path) -> None: + path = _audit( + tmp_path, + [("bot", "hard_deny", "destructive system command"), ("bot", "allow", "ok")], + ) + report = build_report(path) + + assert len(report.hard_deny_events) == 1 + event = report.hard_deny_events[0] + assert event["actor"] == "bot" + assert "destructive" in event["reason"] + + +def test_hard_deny_sample_is_bounded(tmp_path: Path) -> None: + path = _audit(tmp_path, [("bot", "hard_deny", "bad")] * 20) + assert len(build_report(path, max_hard_deny_samples=5).hard_deny_events) == 5 + + +def test_hard_deny_review_never_auto_passes(tmp_path: Path) -> None: + """ + The tool can count and list these; it cannot judge them. Auto-passing would + turn the phase gate into a formality, which is exactly the failure mode the + criterion exists to prevent. + """ + path = _audit(tmp_path, [("a", "allow", "ok")] * 3) + report = build_report(path, min_decisions=1, min_days=0.0) + + reviewed = next(c for c in report.criteria if c.name == "hard_deny_reviewed") + assert reviewed.passed is False + assert report.ready_to_advance is False + + +def test_insufficient_window_fails_the_criterion(tmp_path: Path) -> None: + path = _audit(tmp_path, [("a", "allow", "ok")] * 3) + report = build_report(path, min_decisions=10_000, min_days=14.0) + + window = next(c for c in report.criteria if c.name == "sufficient_window") + assert window.passed is False + assert "3 decisions" in window.detail + + +def test_enough_decisions_satisfies_the_window(tmp_path: Path) -> None: + path = _audit(tmp_path, [("a", "allow", "ok")] * 5) + report = build_report(path, min_decisions=5, min_days=999.0) + assert next(c for c in report.criteria if c.name == "sufficient_window").passed is True + + +def test_mixed_policy_fingerprints_fail_the_criterion(tmp_path: Path) -> None: + """ + Two fingerprints means the window mixes different engines: the measurement + describes no single policy and has to be restarted. + """ + path = tmp_path / "decisions.jsonl" + for fingerprint in ("sha256:one", "sha256:two"): + logger = AuditLogger(str(path), policy_fingerprint=fingerprint) + logger.log( + actor="a", + action="x", + decision={"allowed": True, "risk_score": 0, "reason": "ok", "state": "allow"}, + context={}, + ) + + report = build_report(path) + assert len(report.policy_fingerprints) == 2 + assert next(c for c in report.criteria if c.name == "single_policy").passed is False + + +def test_a_tampered_chain_fails_the_criterion(tmp_path: Path) -> None: + path = _audit(tmp_path, [("a", "allow", "ok"), ("b", "allow", "ok")]) + events = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] + events[0]["actor"] = "tampered" + path.write_text( + "\n".join(json.dumps(e, ensure_ascii=False) for e in events) + "\n", encoding="utf-8" + ) + + report = build_report(path) + assert report.chain_valid is False + assert next(c for c in report.criteria if c.name == "chain_intact").passed is False + + +def test_counts_escalations_and_skipped_semantics(tmp_path: Path) -> None: + path = tmp_path / "decisions.jsonl" + logger = AuditLogger(str(path)) + logger.log( + actor="a", + action="x", + decision={"allowed": False, "risk_score": 50, "reason": "r", "state": "escalate"}, + context={ + "intent_escalation": {"escalated": True}, + "constitution": {"semantic_skipped": "model_not_warm"}, + }, + ) + report = build_report(path) + + assert report.intent_escalations == 1 + assert report.semantic_skipped == 1 + + +def test_window_days_spans_first_to_last_event(tmp_path: Path) -> None: + path = tmp_path / "decisions.jsonl" + base = datetime.now(UTC) + rows = [] + for offset in (0, 10): + rows.append( + { + "ts": (base + timedelta(days=offset)).isoformat(), + "actor": "a", + "action": "x", + "decision": {"allowed": True, "risk_score": 0, "reason": "ok", "state": "allow"}, + "context": {}, + } + ) + path.write_text("\n".join(json.dumps(r) for r in rows) + "\n", encoding="utf-8") + + report = build_report(path) + assert 9.9 < report.window_days < 10.1 + + +def test_malformed_and_blank_lines_are_skipped(tmp_path: Path) -> None: + path = _audit(tmp_path, [("a", "allow", "ok")]) + with path.open("a", encoding="utf-8") as handle: + handle.write("\n") + handle.write("not json at all\n") + handle.write('"a bare string"\n') + + report = build_report(path) + assert report.total_decisions == 1 + # A malformed line is a finding, not something to swallow silently. + assert report.chain_valid is False + assert "invalid JSON" in str(report.chain_error_detail) + + +def test_events_without_a_timestamp_do_not_break_the_window(tmp_path: Path) -> None: + path = tmp_path / "decisions.jsonl" + path.write_text( + json.dumps( + { + "actor": "a", + "action": "x", + "ts": "not-a-timestamp", + "decision": {"allowed": True, "risk_score": 0, "reason": "ok", "state": "allow"}, + "context": {}, + } + ) + + "\n", + encoding="utf-8", + ) + report = build_report(path) + assert report.window_days == 0.0 + assert report.first_event is None + + +def test_missing_audit_file_is_reported(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="audit file not found"): + build_report(tmp_path / "nope.jsonl") + + +def test_report_serializes(tmp_path: Path) -> None: + path = _audit(tmp_path, [("a", "hard_deny", "bad")]) + payload = build_report(path).to_dict() + + assert payload["current_phase"] == 1 + assert payload["next_phase"] == 2 + assert payload["ready_to_advance"] is False + assert payload["chain"]["valid"] is True + + +def test_next_phase_is_clamped_at_the_last_one(tmp_path: Path) -> None: + path = _audit(tmp_path, [("a", "allow", "ok")]) + assert build_report(path, current_phase=3).to_dict()["next_phase"] == 3 + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def test_cli_exits_nonzero_when_not_ready( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """Usable as a gate in a promotion pipeline.""" + path = _audit(tmp_path, [("a", "hard_deny", "bad")]) + assert main(["--audit-path", str(path), "--json"]) == 1 + + payload = json.loads(capsys.readouterr().out) + assert payload["ready_to_advance"] is False + + +def test_cli_text_output_lists_events_and_criteria( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + path = _audit(tmp_path, [("bot", "hard_deny", "destructive system command")]) + main(["--audit-path", str(path)]) + + out = capsys.readouterr().out + assert "ROLLOUT REPORT" in out + assert "hard_deny events to review" in out + assert "NOT READY" in out + + +def test_cli_reports_a_missing_file(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + assert main(["--audit-path", str(tmp_path / "nope.jsonl")]) == 2 + assert "audit file not found" in capsys.readouterr().err + + +def test_cli_accepts_a_phase(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + path = _audit(tmp_path, [("a", "allow", "ok")]) + main(["--audit-path", str(path), "--phase", "2", "--json"]) + assert json.loads(capsys.readouterr().out)["current_phase"] == 2 + + +def test_cli_is_reachable_through_the_main_entry_point( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + from aetherya.cli import main as cli_main + + path = _audit(tmp_path, [("a", "allow", "ok")]) + cli_main(["rollout", "report", "--audit-path", str(path), "--json"]) + assert "ready_to_advance" in capsys.readouterr().out + + +def test_empty_audit_file_produces_an_empty_report(tmp_path: Path) -> None: + path = tmp_path / "decisions.jsonl" + path.write_text("", encoding="utf-8") + + report = build_report(path) + assert report.total_decisions == 0 + assert report.states == {} + assert report.ready_to_advance is False + assert report.chain_valid is False + + +def test_text_output_without_hard_denies_omits_the_review_block( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + path = _audit(tmp_path, [("a", "allow", "ok")]) + main(["--audit-path", str(path)]) + assert "hard_deny events to review" not in capsys.readouterr().out + + +def test_actions_are_truncated_in_the_review_list(tmp_path: Path) -> None: + path = tmp_path / "decisions.jsonl" + AuditLogger(str(path)).log( + actor="a", + action="x" * 400, + decision={"allowed": False, "risk_score": 100, "reason": "bad", "state": "hard_deny"}, + context={}, + ) + assert len(build_report(path).hard_deny_events[0]["action"]) <= 160 + + +def test_context_without_an_action_block_is_tolerated(tmp_path: Path) -> None: + path = tmp_path / "decisions.jsonl" + AuditLogger(str(path)).log( + actor="a", + action="x", + decision={"allowed": True, "risk_score": 0, "reason": "ok", "state": "allow"}, + context={"action": "not-a-mapping"}, + ) + assert build_report(path).tools_by_state["allow"] == ["-"] + + +def test_json_output_is_machine_readable( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + path = _audit(tmp_path, [("a", "allow", "ok")] * 2) + main(["--audit-path", str(path), "--json", "--min-decisions", "1", "--min-days", "0"]) + + payload: dict[str, Any] = json.loads(capsys.readouterr().out) + assert payload["window"]["total_decisions"] == 2 + assert isinstance(payload["criteria"], list) diff --git a/tests/test_security_gate_branches.py b/tests/test_security_gate_branches.py index f53d736..88cbca8 100644 --- a/tests/test_security_gate_branches.py +++ b/tests/test_security_gate_branches.py @@ -308,3 +308,29 @@ def test_main_text_mode_exercises_non_json_output( assert code == 0 assert "security_gate passed=True" in captured.out + + +def test_phase2_is_idempotent_across_runs(tmp_path: Path) -> None: + """ + AuditLogger appends by design, so a corpus file left by a previous run was + appended to and counted: the phase then failed on events it had not created, + reporting invalid == total, which reads like a real integrity regression. + CI never hits it (fresh checkout); a second local run always did. + """ + from aetherya.security_gate import _phase_integrity_fuzz + + kwargs = { + "attestation_key": "test-key", + "workdir": tmp_path, + "events": 40, + "seed": 1337, + "mutation_rounds": 4, + "expected_reject_code": 1, + } + + first = _phase_integrity_fuzz(**kwargs) # type: ignore[arg-type] + second = _phase_integrity_fuzz(**kwargs) # type: ignore[arg-type] + + assert first.passed is True + assert second.passed is True + assert second.details["total"] == first.details["total"] == 40