From a064cf8917eb89a77c513ba79d34ab187ba2b3d1 Mon Sep 17 00:00:00 2001 From: Baraa Abuarra Date: Mon, 8 Jun 2026 01:20:14 +0200 Subject: [PATCH] Add tenant-scoped kill switch --- api/nexus/authorization/http_models.py | 10 ++++ api/nexus/authorization/models.py | 2 + api/nexus/authorization/routes.py | 40 ++++++++++++- api/nexus/authorization/service.py | 26 ++++++++- api/nexus/main.py | 3 + api/nexus/repositories/kill_switch.py | 46 +++++++++++++++ api/tests/test_authorization_service.py | 40 +++++++++++++ api/tests/test_authorize_endpoint.py | 78 ++++++++++++++++++++++++- migrations/0015_kill_switch.sql | 52 +++++++++++++++++ 9 files changed, 293 insertions(+), 4 deletions(-) create mode 100644 api/nexus/repositories/kill_switch.py create mode 100644 migrations/0015_kill_switch.sql diff --git a/api/nexus/authorization/http_models.py b/api/nexus/authorization/http_models.py index 24af745..94cb686 100644 --- a/api/nexus/authorization/http_models.py +++ b/api/nexus/authorization/http_models.py @@ -18,3 +18,13 @@ class AuthorizeResponse(BaseModel): effect: DecisionEffect fail_mode_applied: bool reason: str + + +class KillSwitchRequest(BaseModel): + halted: bool + agent_key: str = Field(default="*", min_length=1, max_length=128) + + +class KillSwitchResponse(BaseModel): + agent_key: str + halted: bool diff --git a/api/nexus/authorization/models.py b/api/nexus/authorization/models.py index 9cbd30a..0be65ca 100644 --- a/api/nexus/authorization/models.py +++ b/api/nexus/authorization/models.py @@ -10,10 +10,12 @@ class DecisionEffect(StrEnum): ALLOW = "allow" DENY = "deny" + class DecisionReason(StrEnum): DEFAULT_ALLOW = "default_allow" FAIL_CLOSED = "fail_closed" STATIC_DENY_TEST_TOOL = "denied: static deny test tool" + KILL_SWITCH_ACTIVE = "denied: kill switch active" @dataclass(frozen=True) diff --git a/api/nexus/authorization/routes.py b/api/nexus/authorization/routes.py index 985bbd4..f0a30a1 100644 --- a/api/nexus/authorization/routes.py +++ b/api/nexus/authorization/routes.py @@ -7,6 +7,8 @@ from nexus.authorization.http_models import ( AuthorizeRequest, AuthorizeResponse, + KillSwitchRequest, + KillSwitchResponse, ) from nexus.authorization.models import AuthorizationRequest from nexus.authorization.service import AuthorizationService @@ -15,6 +17,7 @@ router = APIRouter(prefix="/v1") CONTRACT_HEADER = "x-nexus-contract-version" + def _error(status: int, code: str, message: str, details: list | None = None): body = {"error": {"code": code, "message": message}} if details is not None: @@ -23,7 +26,10 @@ def _error(status: int, code: str, message: str, details: list | None = None): def _authorization_service(request: Request) -> AuthorizationService: - return AuthorizationService(request.app.state.decision_repo) + return AuthorizationService( + request.app.state.decision_repo, + request.app.state.kill_switch_repo, + ) @router.post("/authorize", response_model=AuthorizeResponse) @@ -78,3 +84,35 @@ async def authorize(request: Request): fail_mode_applied=decision.fail_mode_applied, reason=decision.reason.value, ) + + +@router.post("/kill-switch", response_model=KillSwitchResponse) +async def set_kill_switch(request: Request): + try: + body = await request.json() + except Exception: + return _error(400, "malformed_request", "request body is not valid JSON") + + try: + payload = KillSwitchRequest.model_validate(body) + except PydanticValidationError as e: + details = [ + {"field": ".".join(str(p) for p in err["loc"]), "issue": err["msg"]} + for err in e.errors() + ] + return _error( + 422, + "validation_failed", + "request failed schema validation", + details, + ) + + await request.app.state.kill_switch_repo.set_halted( + payload.halted, + payload.agent_key, + ) + + return KillSwitchResponse( + agent_key=payload.agent_key, + halted=payload.halted, + ) diff --git a/api/nexus/authorization/service.py b/api/nexus/authorization/service.py index f473f2d..241c7f7 100644 --- a/api/nexus/authorization/service.py +++ b/api/nexus/authorization/service.py @@ -21,15 +21,37 @@ async def record( """Persist the authorization decision before it is returned.""" +class KillSwitchReader(Protocol): + async def is_halted(self, agent_key: str) -> bool: + """Return whether authorization must be denied by kill switch.""" + + class AuthorizationService: - def __init__(self, writer: DecisionLogWriter): + def __init__( + self, + writer: DecisionLogWriter, + kill_switch: KillSwitchReader | None = None, + ): self._writer = writer + self._kill_switch = kill_switch async def authorize( self, request: AuthorizationRequest, ) -> AuthorizationDecision: - if request.tool_name == "__nexus_deny_test__": + kill_switch_active = False + if self._kill_switch is not None: + kill_switch_active = await self._kill_switch.is_halted(request.agent_key) + + if kill_switch_active: + decision = AuthorizationDecision( + decision_id=uuid4(), + effect=DecisionEffect.DENY, + fail_mode_applied=False, + reason=DecisionReason.KILL_SWITCH_ACTIVE, + decided_at=datetime.now(UTC), + ) + elif request.tool_name == "__nexus_deny_test__": decision = AuthorizationDecision( decision_id=uuid4(), effect=DecisionEffect.DENY, diff --git a/api/nexus/main.py b/api/nexus/main.py index 8293956..670cb02 100644 --- a/api/nexus/main.py +++ b/api/nexus/main.py @@ -34,6 +34,8 @@ from nexus.repositories.agent_run import AgentRunRepository from nexus.tenancy.boot_assertion import assert_safe_role from nexus.read.routes import router as read_router +from nexus.repositories.kill_switch import KillSwitchRepository + def create_app(settings: Settings | None = None) -> FastAPI: settings = settings or Settings.from_env() @@ -51,6 +53,7 @@ async def lifespan(app: FastAPI): limits = Limits.from_env() app.state.agent_run_repo = AgentRunRepository(pool, crypto, limits) app.state.decision_repo = DecisionRepository(pool) + app.state.kill_switch_repo = KillSwitchRepository(pool) try: yield finally: diff --git a/api/nexus/repositories/kill_switch.py b/api/nexus/repositories/kill_switch.py new file mode 100644 index 0000000..0aec338 --- /dev/null +++ b/api/nexus/repositories/kill_switch.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from uuid import uuid4 + +from nexus.repositories.base import BaseRepository + + +class KillSwitchRepository(BaseRepository): + async def is_halted(self, agent_key: str) -> bool: + async with self._tx() as conn: + halted = await conn.fetchval( + """ + SELECT halted + FROM kill_switch_state + WHERE agent_key IN ('*', $1) + AND halted = true + LIMIT 1 + """, + agent_key, + ) + return bool(halted) + + async def set_halted(self, halted: bool, agent_key: str = "*") -> None: + async with self._tx() as conn: + await conn.execute( + """ + INSERT INTO kill_switch_state + (tenant_id, agent_key, halted, updated_at) + VALUES (current_setting('app.tenant_id')::uuid, $1, $2, now()) + ON CONFLICT (tenant_id, agent_key) + DO UPDATE SET halted = EXCLUDED.halted, updated_at = now() + """, + agent_key, + halted, + ) + + await conn.execute( + """ + INSERT INTO kill_switch_events + (tenant_id, event_id, agent_key, halted, changed_at) + VALUES (current_setting('app.tenant_id')::uuid, $1, $2, $3, now()) + """, + uuid4(), + agent_key, + halted, + ) diff --git a/api/tests/test_authorization_service.py b/api/tests/test_authorization_service.py index 03749fd..8c665b1 100644 --- a/api/tests/test_authorization_service.py +++ b/api/tests/test_authorization_service.py @@ -29,6 +29,12 @@ class FailingWriter: async def record(self, request, decision): raise RuntimeError("write failed") +class FakeKillSwitch: + def __init__(self, halted: bool): + self.halted = halted + + async def is_halted(self, agent_key: str) -> bool: + return self.halted def make_request(): return AuthorizationRequest( @@ -92,3 +98,37 @@ async def test_authorization_preserves_request_fields_for_logging(): assert logged_decision.effect == decision.effect assert logged_decision.fail_mode_applied == decision.fail_mode_applied assert logged_decision.decided_at == decision.decided_at + + +@pytest.mark.asyncio +async def test_kill_switch_denies_before_default_allow(): + writer = FakeWriter() + service = AuthorizationService( + writer, + FakeKillSwitch(True), + ) + + decision = await service.authorize(make_request()) + + assert decision.effect == DecisionEffect.DENY + assert decision.fail_mode_applied is False + assert decision.reason == DecisionReason.KILL_SWITCH_ACTIVE + + +@pytest.mark.asyncio +async def test_kill_switch_decision_is_logged(): + writer = FakeWriter() + service = AuthorizationService( + writer, + FakeKillSwitch(True), + ) + request = make_request() + + decision = await service.authorize(request) + + logged_request, logged_decision = writer.calls[0] + + assert logged_request == request + assert logged_decision.effect == DecisionEffect.DENY + assert logged_decision.reason == DecisionReason.KILL_SWITCH_ACTIVE + assert logged_decision.decision_id == decision.decision_id diff --git a/api/tests/test_authorize_endpoint.py b/api/tests/test_authorize_endpoint.py index e2a36bf..7e8d26e 100644 --- a/api/tests/test_authorize_endpoint.py +++ b/api/tests/test_authorize_endpoint.py @@ -158,4 +158,80 @@ async def test_authorize_static_deny_is_tenant_isolated(client, two_auth_headers assert r2.status_code == 200, r2.text assert r1.json()["effect"] == "deny" assert r2.json()["effect"] == "deny" - assert r1.json()["decision_id"] != r2.json()["decision_id"] \ No newline at end of file + assert r1.json()["decision_id"] != r2.json()["decision_id"] + + +async def test_kill_switch_halt_forces_deny(client, auth_headers): + r_set = await client.post( + "/v1/kill-switch", + json={"halted": True}, + headers=auth_headers, + ) + + assert r_set.status_code == 200, r_set.text + assert r_set.json()["halted"] is True + + r = await client.post( + "/v1/authorize", + json=_authorize_body(), + headers=auth_headers, + ) + + assert r.status_code == 200, r.text + assert r.json()["effect"] == "deny" + assert r.json()["reason"] == "denied: kill switch active" + + +async def test_kill_switch_clear_restores_default_allow(client, auth_headers): + await client.post( + "/v1/kill-switch", + json={"halted": True}, + headers=auth_headers, + ) + r_clear = await client.post( + "/v1/kill-switch", + json={"halted": False}, + headers=auth_headers, + ) + + assert r_clear.status_code == 200, r_clear.text + assert r_clear.json()["halted"] is False + + r = await client.post( + "/v1/authorize", + json=_authorize_body(), + headers=auth_headers, + ) + + assert r.status_code == 200, r.text + assert r.json()["effect"] == "allow" + assert r.json()["reason"] == "default_allow" + + +async def test_kill_switch_deny_is_idempotent(client, auth_headers): + tool_call_id = str(uuid.uuid4()) + + await client.post( + "/v1/kill-switch", + json={"halted": True}, + headers=auth_headers, + ) + + r1 = await client.post( + "/v1/authorize", + json=_authorize_body(tool_call_id), + headers=auth_headers, + ) + r2 = await client.post( + "/v1/authorize", + json=_authorize_body(tool_call_id), + headers=auth_headers, + ) + + assert r1.status_code == 200, r1.text + assert r2.status_code == 200, r2.text + assert r1.json()["decision_id"] == r2.json()["decision_id"] + assert r1.json()["effect"] == "deny" + assert r2.json()["effect"] == "deny" + assert r1.json()["reason"] == r2.json()["reason"] + assert r1.json()["reason"] == "denied: kill switch active" \ No newline at end of file diff --git a/migrations/0015_kill_switch.sql b/migrations/0015_kill_switch.sql new file mode 100644 index 0000000..b8434c5 --- /dev/null +++ b/migrations/0015_kill_switch.sql @@ -0,0 +1,52 @@ +-- 0015_kill_switch.sql +-- Run as nexus_owner. +-- PR #32: tenant-scoped kill switch v1. +-- +-- Current halt state is mutable. +-- Halt changes are append-only governance records. + +SET search_path = nexus, public; + +CREATE TABLE kill_switch_state ( + tenant_id uuid NOT NULL, + agent_key text NOT NULL DEFAULT '*', + halted boolean NOT NULL, + updated_at timestamptz NOT NULL DEFAULT now(), + + PRIMARY KEY (tenant_id, agent_key) +); + +CREATE TABLE kill_switch_events ( + tenant_id uuid NOT NULL, + event_id uuid NOT NULL, + agent_key text NOT NULL DEFAULT '*', + halted boolean NOT NULL, + changed_at timestamptz NOT NULL DEFAULT now(), + + PRIMARY KEY (tenant_id, event_id) +); + +ALTER TABLE kill_switch_state ENABLE ROW LEVEL SECURITY; +ALTER TABLE kill_switch_state FORCE ROW LEVEL SECURITY; + +CREATE POLICY kill_switch_state_tenant_isolation + ON kill_switch_state + USING (tenant_id = current_setting('app.tenant_id')::uuid) + WITH CHECK (tenant_id = current_setting('app.tenant_id')::uuid); + +ALTER TABLE kill_switch_events ENABLE ROW LEVEL SECURITY; +ALTER TABLE kill_switch_events FORCE ROW LEVEL SECURITY; + +CREATE POLICY kill_switch_events_tenant_isolation + ON kill_switch_events + USING (tenant_id = current_setting('app.tenant_id')::uuid) + WITH CHECK (tenant_id = current_setting('app.tenant_id')::uuid); + +CREATE INDEX kill_switch_events_tenant_time_idx + ON kill_switch_events (tenant_id, changed_at DESC); + +GRANT SELECT, INSERT, UPDATE ON kill_switch_state TO nexus_app; +REVOKE DELETE ON kill_switch_state FROM nexus_app; + +GRANT SELECT, INSERT ON kill_switch_events TO nexus_app; +REVOKE UPDATE, DELETE ON kill_switch_events FROM nexus_app;