Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,8 @@ dependencies = [
"pyyaml>=6.0.2",
"textual>=0.50.0",
"pydantic>=2.10.0",
"mcp>=1.6.0",
# <2.0: mcp 2.0.0 removes mcp.server.fastmcp, which aegis_memory/mcp_server.py is built on.
"mcp>=1.6.0,<2.0",
# Transitive security floors (OpenSSF Scorecard / OSV) — pulled in transitively
# (idna<-httpx, pygments<-rich); loose upstream bounds otherwise let known-vulnerable
# versions resolve. Pinned to the first patched release of each advisory.
Expand Down
9 changes: 8 additions & 1 deletion server/api/routers/decay.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,13 @@
"""

from fastapi import APIRouter, Depends
from memory_authz import effective_agent_id
from memory_repository import MemoryRepository
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession
from temporal_decay import DEFAULT_HALF_LIFE, HALF_LIVES

from api.dependencies.auth import check_rate_limit
from api.dependencies.auth import AuthContext, check_rate_limit, get_auth_context
from api.dependencies.database import get_db

router = APIRouter()
Expand Down Expand Up @@ -69,6 +70,7 @@ async def get_decay_config(project_id: str = Depends(check_rate_limit)):
async def archive_stale_memories(
body: ArchiveRequest,
project_id: str = Depends(check_rate_limit),
auth: AuthContext = Depends(get_auth_context),
db: AsyncSession = Depends(get_db),
):
"""
Expand All @@ -77,13 +79,18 @@ async def archive_stale_memories(
Uses the existing is_deprecated / deprecated_at soft-delete columns.
Set dry_run=True to preview how many memories would be archived without
actually modifying any rows.

A bound key acts as its agent only, so its sweep is confined to that agent's
memories; an unbound key represents the application and sweeps the project.
"""
acting_agent_id = effective_agent_id(auth, None)
archived = await MemoryRepository.archive_stale(
db,
project_id=project_id,
namespace=body.namespace,
threshold=body.threshold,
dry_run=body.dry_run,
agent_id=acting_agent_id,
)
return ArchiveResponse(
archived=archived,
Expand Down
25 changes: 20 additions & 5 deletions server/api/routers/handoffs.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,21 @@
"""
Handoff Router (~60 lines)
Handoff Router (~80 lines)

Handles: /memories/handoff
"""

from api.dependencies.auth import check_rate_limit
from api.dependencies.auth import AuthContext, check_rate_limit, get_auth_context
from api.dependencies.database import get_db
from config import get_settings
from embedding_service import get_embedding_service
from fastapi import APIRouter, BackgroundTasks, Depends
from memory_authz import effective_agent_id, read_scope_restriction
from memory_repository import MemoryRepository
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter()
_settings = get_settings()


class HandoffRequest(BaseModel):
Expand Down Expand Up @@ -43,21 +46,33 @@ async def handoff(
body: HandoffRequest,
background_tasks: BackgroundTasks,
project_id: str = Depends(check_rate_limit),
auth: AuthContext = Depends(get_auth_context),
db: AsyncSession = Depends(get_db),
):
"""Generate a structured handoff baton for agent-to-agent state transfer."""
"""Generate a structured handoff baton for agent-to-agent state transfer.

The baton carries the source agent's memories — including agent-private ones — as
``key_facts``, so a handoff is a push by the source, never a pull by an arbitrary key
naming a source: a bound key may only hand off its own memories, while an unbound key
represents the whole application and may orchestrate handoffs between any of its agents.
"""
source_agent_id = effective_agent_id(auth, body.source_agent_id)
# Same principal-trust ceiling as the query routes: this is a bulk read that never
# passes through authorize_read.
scope_filter = read_scope_restriction(auth, enforce_principal_trust=_settings.enable_trust_levels)
embed_service = get_embedding_service()
task_embedding = None
if body.task_context:
task_embedding = await embed_service.embed_single(body.task_context, db)
results = await MemoryRepository.get_agent_memories_for_handoff(
db, project_id=project_id, source_agent_id=body.source_agent_id,
db, project_id=project_id, source_agent_id=source_agent_id,
namespace=body.namespace, user_id=body.user_id,
task_embedding=task_embedding, max_memories=body.max_memories,
requested_scope=scope_filter,
)
memories = [mem for mem, _ in results]
return HandoffBaton(
source_agent_id=body.source_agent_id, target_agent_id=body.target_agent_id,
source_agent_id=source_agent_id, target_agent_id=body.target_agent_id,
namespace=body.namespace, user_id=body.user_id, task_context=body.task_context,
summary=None, active_tasks=[], blocked_on=[], recent_decisions=[],
key_facts=[mem.content for mem in memories],
Expand Down
14 changes: 14 additions & 0 deletions server/memory_repository.py
Original file line number Diff line number Diff line change
Expand Up @@ -583,12 +583,16 @@ async def get_agent_memories_for_handoff(
user_id: str | None = None,
task_embedding: list[float] | None = None,
max_memories: int = 20,
requested_scope: str | None = None,
) -> list[tuple[Memory, float | None]]:
"""
Get memories for agent handoff.

If task_embedding provided, rank by relevance.
Otherwise, return most recent.

``requested_scope`` is the caller's principal-trust ceiling (``read_scope_restriction``):
a NULL scope means agent-private, so restricted principals never match those rows.
"""
conditions = [
Memory.project_id == project_id,
Expand All @@ -603,6 +607,9 @@ async def get_agent_memories_for_handoff(
if user_id is not None:
conditions.append(Memory.user_id == user_id)

if requested_scope is not None:
conditions.append(Memory.scope == requested_scope)

if task_embedding is not None:
# Rank by semantic similarity to task
distance_expr = Memory.embedding.cosine_distance(task_embedding)
Expand Down Expand Up @@ -729,13 +736,17 @@ async def archive_stale(
threshold: float = 0.1,
batch_size: int = 500,
dry_run: bool = False,
agent_id: str | None = None,
) -> int:
"""
Soft-deprecate active memories whose relevance_score falls below threshold.

Uses existing is_deprecated / deprecated_at columns — no new schema needed.
Relevance is computed in Python on fetched batches (chunked approach).

``agent_id`` restricts the sweep to one agent's rows — set for bound keys, which act
as their agent only. None sweeps the whole project (unbound/application keys).

Returns count of memories archived (or that would be archived for dry_run).
"""
now = datetime.now(timezone.utc)
Expand All @@ -745,6 +756,9 @@ async def archive_stale(
not_(Memory.is_deprecated),
]

if agent_id is not None:
conditions.append(Memory.agent_id == agent_id)

stmt = (
select(Memory)
.where(and_(*conditions))
Expand Down
189 changes: 189 additions & 0 deletions tests/test_authz_bypass.py
Original file line number Diff line number Diff line change
Expand Up @@ -435,6 +435,45 @@ def test_every_router_that_writes_memories_screens_and_authorizes(self):
"these routers write to memory without the full gate set: " + "; ".join(offenders)
)

def test_every_router_that_touches_the_repository_authorizes_an_identity(self):
"""The read-side generalization of the ace_delta bug.

The write sweep above keys on ``MemoryRepository.add`` and so covered only writers.
``handoffs.py`` — a pure read sink — inherited nothing from the authorization pass:
it took ``source_agent_id`` from the request body with no auth context and returned
that agent's memories, private ones included, as ``key_facts`` to any project key.
``decay.py`` was the state-change analogue (project-wide archive with no agent authz).

So enumerate every router that touches ``MemoryRepository`` at all — read, write, or
sweep — and require it to resolve an authorized identity via a ``memory_authz`` helper
or sit behind the admin gate. Source-derived, not hand-maintained, so a new router
fails here on the day it is added.
"""
import inspect
import pkgutil
import importlib
from api import routers as routers_pkg

gates = (
"effective_agent_id(", "authorize_read(", "authorize_write(",
"authorize_delete(", "read_scope_restriction(", "require_admin",
)
offenders = []
for mod_info in pkgutil.iter_modules(routers_pkg.__path__):
mod = importlib.import_module(f"api.routers.{mod_info.name}")
try:
src = inspect.getsource(mod)
except OSError:
continue
if "MemoryRepository." not in src:
continue
if not any(gate in src for gate in gates):
offenders.append(f"{mod_info.name}.py")
assert not offenders, (
"these routers reach MemoryRepository without resolving an authorized identity: "
+ "; ".join(offenders)
)

def test_authz_helpers_are_actually_called_by_the_routers(self):
"""Guards against the exact original failure: implemented, exported, never invoked."""
import inspect
Expand Down Expand Up @@ -598,3 +637,153 @@ def test_omitted_agent_id_is_pinned_to_the_key_not_left_open(self, monkeypatch):
assert canary.get("requesting_agent_id") == "agent-1", (
"omitting agent_id left the ACL identity unset, widening the query beyond the key"
)


class TestHandoffHttpBypass:
"""``POST /memories/handoff`` attempted the way an attacker would.

This route was missed by the first authorization pass because the source-derived sweep
keyed on ``MemoryRepository.add`` — a handoff writes nothing. Pre-fix, it took
``source_agent_id`` from the request body with no auth context, and the repository applied
no scope filter, so any project key received another agent's memories — agent-private ones
included — as ``key_facts`` in one call.
"""

@staticmethod
def _client(monkeypatch, bound_agent_id: str | None, canary: dict):
from fastapi import FastAPI
from fastapi.testclient import TestClient

from api.dependencies.auth import AuthContext, check_rate_limit, get_auth_context
from api.dependencies.database import get_db
from api.routers import handoffs
from memory_repository import MemoryRepository

app = FastAPI()
# Same prefix as production (api/app.py:133) so the path under test is the real one.
app.include_router(handoffs.router, prefix="/memories")

async def _fake_db():
yield None

app.dependency_overrides[get_auth_context] = lambda: AuthContext(
project_id="proj-1", trust_level="internal", bound_agent_id=bound_agent_id
)
app.dependency_overrides[check_rate_limit] = lambda: "proj-1"
app.dependency_overrides[get_db] = _fake_db

async def _canary(*args, **kwargs):
canary["called"] = True
canary["source_agent_id"] = kwargs.get("source_agent_id")
return []

monkeypatch.setattr(
MemoryRepository, "get_agent_memories_for_handoff", staticmethod(_canary)
)
return TestClient(app, raise_server_exceptions=False)

def test_bound_key_cannot_pull_another_agents_baton(self, monkeypatch):
"""The exfiltration path: agent-2's key names agent-1 as the handoff source.

Against the pre-fix code this returned 200 with agent-1's private memories in
key_facts.
"""
canary: dict = {}
client = self._client(monkeypatch, "agent-2", canary)

resp = client.post("/memories/handoff", json={
"source_agent_id": "agent-1", "target_agent_id": "agent-2",
})

assert resp.status_code == 403, (
f"expected 403 for a handoff pull naming another agent, got {resp.status_code}. "
f"A handoff is a push by the source, never a pull by an arbitrary key."
)
assert not canary.get("called"), (
"authorization ran after the read: the memories were already fetched by the time "
"the request was denied"
)

def test_source_agent_may_hand_off_its_own_memories(self, monkeypatch):
"""Positive control, and pins the identity handed to the repository to the key's."""
canary: dict = {}
client = self._client(monkeypatch, "agent-1", canary)

resp = client.post("/memories/handoff", json={
"source_agent_id": "agent-1", "target_agent_id": "agent-2",
})

assert resp.status_code == 200, f"a legitimate self-handoff was blocked: {resp.text}"
assert canary.get("source_agent_id") == "agent-1"

def test_unbound_key_may_orchestrate_handoffs(self, monkeypatch):
"""Documented posture: an unbound project key represents the whole application."""
canary: dict = {}
client = self._client(monkeypatch, None, canary)

resp = client.post("/memories/handoff", json={
"source_agent_id": "agent-1", "target_agent_id": "agent-2",
})

assert resp.status_code == 200
assert canary.get("source_agent_id") == "agent-1"


class TestDecayArchiveScoping:
"""``POST /memories/decay/archive`` — a bound key's sweep must not touch other agents' rows.

The state-change analogue of the handoff read gap: pre-fix the route archived project-wide
with no agent authorization, so a key bound to one agent could soft-deprecate every other
agent's memories.
"""

@staticmethod
def _client(monkeypatch, bound_agent_id: str | None, canary: dict):
from fastapi import FastAPI
from fastapi.testclient import TestClient

from api.dependencies.auth import AuthContext, check_rate_limit, get_auth_context
from api.dependencies.database import get_db
from api.routers import decay
from memory_repository import MemoryRepository

app = FastAPI()
app.include_router(decay.router, prefix="/memories/decay")

async def _fake_db():
yield None

app.dependency_overrides[get_auth_context] = lambda: AuthContext(
project_id="proj-1", trust_level="internal", bound_agent_id=bound_agent_id
)
app.dependency_overrides[check_rate_limit] = lambda: "proj-1"
app.dependency_overrides[get_db] = _fake_db

async def _canary(*args, **kwargs):
canary["called"] = True
canary["agent_id"] = kwargs.get("agent_id")
return 0

monkeypatch.setattr(MemoryRepository, "archive_stale", staticmethod(_canary))
return TestClient(app, raise_server_exceptions=False)

def test_bound_key_sweep_is_confined_to_its_agent(self, monkeypatch):
canary: dict = {}
client = self._client(monkeypatch, "agent-1", canary)

resp = client.post("/memories/decay/archive", json={})

assert resp.status_code == 200
assert canary.get("agent_id") == "agent-1", (
"a bound key's archive sweep ran project-wide instead of being confined to its agent"
)

def test_unbound_key_sweep_covers_the_project(self, monkeypatch):
canary: dict = {}
client = self._client(monkeypatch, None, canary)

resp = client.post("/memories/decay/archive", json={})

assert resp.status_code == 200
assert canary.get("called")
assert canary.get("agent_id") is None
Loading