From 24de0413d172a9fac246e75b39a874aeda51005b Mon Sep 17 00:00:00 2001 From: Kent Bull Date: Wed, 6 May 2026 16:38:13 -0600 Subject: [PATCH] feat: SSE AgentSignals for agent events Signed-off-by: Kent Bull --- .github/workflows/test.yaml | 4 +- scripts/sync_integration_deps.py | 4 +- src/signify/app/clienting.py | 5 + src/signify/app/signaling.py | 39 +++++++ tests/app/test_clienting.py | 10 ++ tests/app/test_signaling.py | 82 ++++++++++++++ tests/integration/README.md | 2 +- tests/integration/_services/keria_server.py | 43 ++++++++ tests/integration/_services/witness_server.py | 2 +- tests/integration/dependencies.py | 4 +- tests/integration/test_signaling.py | 102 ++++++++++++++++++ 11 files changed, 290 insertions(+), 7 deletions(-) create mode 100644 src/signify/app/signaling.py create mode 100644 tests/app/test_signaling.py create mode 100644 tests/integration/test_signaling.py diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index a49e4e1..41c0ff6 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -14,7 +14,7 @@ env: # deliberate and reviewable instead of being buried inside individual steps. PYTHON_VERSION: '3.12.6' KERIPY_REF: '4ee02c0213770d25a0114fe7ebd7ab4ab5500cde' - KERIA_REF: '9e2461550f373ad7bdbe7eebeaceac689cb15397' + KERIA_REF: '5b703bd8a60fab68a6476819626b22784317bf14' VLEI_REF: 'f514b9431c5f965b5f7f64a8693e19df2f181564' jobs: @@ -109,7 +109,7 @@ jobs: - name: Stage 1 - Check out KERIA uses: actions/checkout@v4.2.2 with: - repository: WebOfTrust/keria + repository: kentbull/keria ref: ${{ env.KERIA_REF }} path: keria diff --git a/scripts/sync_integration_deps.py b/scripts/sync_integration_deps.py index f0a6904..c85f9f8 100644 --- a/scripts/sync_integration_deps.py +++ b/scripts/sync_integration_deps.py @@ -25,8 +25,10 @@ def sync_repo(root: Path, repo: str, ref: str) -> None: if not root.exists(): root.parent.mkdir(parents=True, exist_ok=True) run("git", "clone", repo, str(root)) + else: + run("git", "-C", str(root), "remote", "set-url", "origin", repo) - run("git", "-C", str(root), "fetch", "--tags", "origin") + run("git", "-C", str(root), "fetch", "--tags", "--force", "origin") run("git", "-C", str(root), "fetch", "origin") run("git", "-C", str(root), "checkout", "--detach", ref) diff --git a/src/signify/app/clienting.py b/src/signify/app/clienting.py index 2cd7479..13b144c 100644 --- a/src/signify/app/clienting.py +++ b/src/signify/app/clienting.py @@ -419,6 +419,11 @@ def schemas(self): from signify.app.schemas import Schemas return Schemas(client=self) + def signals(self): + """Return the generic KERIA agent signaling resource wrapper.""" + from signify.app.signaling import AgentSignals + return AgentSignals(client=self) + def config(self): """Return the agent-configuration read resource wrapper.""" from signify.app.coring import Config diff --git a/src/signify/app/signaling.py b/src/signify/app/signaling.py new file mode 100644 index 0000000..4af1d1c --- /dev/null +++ b/src/signify/app/signaling.py @@ -0,0 +1,39 @@ +# -*- encoding: utf-8 -*- +"""Generic KERIA agent signaling helpers.""" + +from keri import kering +from keri.core import indexing, serdering + + +class AgentSignals: + """Generic signed event stream for one connected KERIA agent.""" + + def __init__(self, client): + self.client = client + + def stream(self): + """Open the authenticated generic agent SSE stream.""" + return self.client.stream( + "/signals/stream", + headers={"Accept": "text/event-stream"}, + ) + + def verifyReplyEnvelope(self, envelope, route=None): + """Verify one KERIA agent-signed KERI ``rpy`` envelope.""" + if self.client.agent is None: + raise kering.ConfigurationError("client must be connected before verification") + + rserder = serdering.SerderKERI(sad=envelope["rpy"]) + if route is not None and rserder.ked.get("r") != route: + return False + + data = rserder.ked.get("a", {}) + if data.get("agent") != self.client.agent.pre: + return False + + sigs = envelope.get("sigs") or [] + if not sigs: + return False + + siger = indexing.Siger(qb64=sigs[0]) + return self.client.agent.verfer.verify(sig=siger.raw, ser=rserder.raw) diff --git a/tests/app/test_clienting.py b/tests/app/test_clienting.py index 462a940..b9895a2 100644 --- a/tests/app/test_clienting.py +++ b/tests/app/test_clienting.py @@ -837,6 +837,16 @@ def test_signify_client_schemas(make_signify_client): assert out.client == client +def test_signify_client_signals(make_signify_client): + client = make_signify_client() + + out = client.signals() + + from signify.app.signaling import AgentSignals + assert type(out) is AgentSignals + assert out.client == client + + def test_signify_client_config(make_signify_client): client = make_signify_client() diff --git a/tests/app/test_signaling.py b/tests/app/test_signaling.py new file mode 100644 index 0000000..e9c55f4 --- /dev/null +++ b/tests/app/test_signaling.py @@ -0,0 +1,82 @@ +# -*- encoding: utf-8 -*- +""" +SIGNIFY generic agent signaling helper tests. +""" + +import pytest +from keri import kering +from keri.core import eventing, signing +from mockito import expect, mock, verifyNoUnwantedInteractions, unstub + +from signify.app.signaling import AgentSignals + + +def signed_envelope(route="/test/signals/request", agent="agent-aid", signer=None): + signer = signer if signer is not None else signing.Salter(raw=b"0123456789abcdef").signer() + rpy = eventing.reply(route=route, data={"agent": agent, "payload": "value"}) + sig = signer.sign(ser=rpy.raw, index=0) + return signer, {"rpy": rpy.ked, "sigs": [sig.qb64]} + + +def test_agent_signals_stream_uses_generic_endpoint(): + client = mock(strict=True) + stream = iter([]) + expect(client, times=1).stream( + "/signals/stream", + headers={"Accept": "text/event-stream"}, + ).thenReturn(stream) + + assert AgentSignals(client).stream() is stream + verifyNoUnwantedInteractions() + unstub() + + +def test_agent_signals_verify_agent_signed_reply_envelope(): + signer, envelope = signed_envelope() + client = mock({"agent": mock({"pre": "agent-aid", "verfer": signer.verfer})}) + + assert AgentSignals(client).verifyReplyEnvelope( + envelope, + route="/test/signals/request", + ) is True + + +def test_agent_signals_verify_rejects_wrong_route(): + signer, envelope = signed_envelope(route="/wrong") + client = mock({"agent": mock({"pre": "agent-aid", "verfer": signer.verfer})}) + + assert AgentSignals(client).verifyReplyEnvelope( + envelope, + route="/test/signals/request", + ) is False + + +def test_agent_signals_verify_rejects_wrong_agent_payload(): + signer, envelope = signed_envelope(agent="wrong-agent") + client = mock({"agent": mock({"pre": "agent-aid", "verfer": signer.verfer})}) + + assert AgentSignals(client).verifyReplyEnvelope(envelope) is False + + +def test_agent_signals_verify_rejects_missing_signatures(): + signer, envelope = signed_envelope() + envelope["sigs"] = [] + client = mock({"agent": mock({"pre": "agent-aid", "verfer": signer.verfer})}) + + assert AgentSignals(client).verifyReplyEnvelope(envelope) is False + + +def test_agent_signals_verify_rejects_bad_signature(): + signer, envelope = signed_envelope() + other = signing.Salter(raw=b"fedcba9876543210").signer() + client = mock({"agent": mock({"pre": "agent-aid", "verfer": other.verfer})}) + + assert AgentSignals(client).verifyReplyEnvelope(envelope) is False + + +def test_agent_signals_verify_requires_connected_agent(): + _, envelope = signed_envelope() + client = mock({"agent": None}) + + with pytest.raises(kering.ConfigurationError, match="client must be connected"): + AgentSignals(client).verifyReplyEnvelope(envelope) diff --git a/tests/integration/README.md b/tests/integration/README.md index ab11956..06b23a0 100644 --- a/tests/integration/README.md +++ b/tests/integration/README.md @@ -56,7 +56,7 @@ The current stack pins those source dependencies to explicit compatibility SHAs: - `keripy`: `4ee02c0213770d25a0114fe7ebd7ab4ab5500cde` (tag `1.2.12`) -- `keria`: `9e2461550f373ad7bdbe7eebeaceac689cb15397` +- `keria`: `5b703bd8a60fab68a6476819626b22784317bf14` from `kentbull/keria` - `vLEI`: `f514b9431c5f965b5f7f64a8693e19df2f181564` (tag `1.0.2`) The CI runtime also constrains `hio` to `0.6.14` across the live stack. diff --git a/tests/integration/_services/keria_server.py b/tests/integration/_services/keria_server.py index dd534b6..f308985 100644 --- a/tests/integration/_services/keria_server.py +++ b/tests/integration/_services/keria_server.py @@ -6,6 +6,8 @@ from pathlib import Path import signal +import falcon + def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) @@ -26,11 +28,52 @@ def configure_temp_log_root(config_dir: str) -> None: ogling.Ogler.TempHeadDir = str(temp_head_dir) +class TestSignalsEnd: + """Harness-only route for exercising KERIA's generic SSE signal channel.""" + + def __init__(self, streaming): + self.streaming = streaming + + def on_get(self, req, rep): + agent = req.context.agent + rep.status = falcon.HTTP_200 + rep.media = {"subscribers": len(agent.sseBroadcaster.subscribers)} + + def on_post(self, req, rep): + agent = req.context.agent + body = req.media or {} + self.streaming.enqueueSignedReplyCue( + agent.signalCues, + event=body.get("event", "agent.signal.test"), + route=body.get("route", "/test/signals/request"), + payload=body.get("payload", {}), + event_id=body.get("event_id", "test-signal"), + ) + rep.status = falcon.HTTP_202 + rep.media = {"queued": True} + + +def install_test_signal_route(agenting): + """Register an authenticated test-only signal trigger route on KERIA admin.""" + original_create_admin_server_doer = agenting.createAdminServerDoer + + def create_admin_server_doer(config, agency): + admin_app, admin_server_doer = original_create_admin_server_doer(config, agency) + from keria.app import streaming + + admin_app.add_route("/test/signals", TestSignalsEnd(streaming)) + return admin_app, admin_server_doer + + agenting.createAdminServerDoer = create_admin_server_doer + + def main() -> None: args = parse_args() configure_temp_log_root(args.config_dir) from keria.app import agenting + install_test_signal_route(agenting) + config = agenting.KERIAServerConfig( name="keria", base="", diff --git a/tests/integration/_services/witness_server.py b/tests/integration/_services/witness_server.py index 902e6b1..d6a1fac 100644 --- a/tests/integration/_services/witness_server.py +++ b/tests/integration/_services/witness_server.py @@ -46,7 +46,7 @@ def create_loopback_http_server( indirecting.createHttpServer = create_loopback_http_server class NoopQueryEnd: - def __init__(self, hab): + def __init__(self, hab, **_kwa): self.hab = hab def on_get(self, req, rep): diff --git a/tests/integration/dependencies.py b/tests/integration/dependencies.py index a685a53..6c61f30 100644 --- a/tests/integration/dependencies.py +++ b/tests/integration/dependencies.py @@ -26,8 +26,8 @@ class IntegrationDependency: KERIA = IntegrationDependency( name="KERIA", - repo="https://github.com/WebOfTrust/keria.git", - ref="9e2461550f373ad7bdbe7eebeaceac689cb15397", + repo="https://github.com/kentbull/keria.git", + ref="5b703bd8a60fab68a6476819626b22784317bf14", path_name="keria", env_root="SIGNIFYPY_INTEGRATION_KERIA_ROOT", env_ref="SIGNIFYPY_INTEGRATION_KERIA_REF", diff --git a/tests/integration/test_signaling.py b/tests/integration/test_signaling.py new file mode 100644 index 0000000..5f7775e --- /dev/null +++ b/tests/integration/test_signaling.py @@ -0,0 +1,102 @@ +"""Live integration coverage for generic KERIA AgentSignals SSE.""" + +from __future__ import annotations + +import json +import queue +import threading +import time + +import pytest + +from signify.app.clienting import SignifyClient +from tests.integration.helpers import POLL_INTERVAL, poll_until + + +pytestmark = pytest.mark.integration + + +def _same_agent_client(client: SignifyClient) -> SignifyClient: + """Connect a second HTTP session to the same already-booted KERIA agent.""" + control = SignifyClient( + passcode=client.bran, + url=client.url, + boot_url=client.boot_url, + ) + control.connect() + return control + + +def _wait_for_event(events: queue.Queue, errors: queue.Queue, *, timeout=30): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if not errors.empty(): + raise errors.get() + try: + return events.get(timeout=0.1) + except queue.Empty: + continue + + if not errors.empty(): + raise errors.get() + raise TimeoutError("timed out waiting for SSE signal event") + + +def test_agent_signals_stream_receives_signed_agent_event(client_factory): + stream_client = client_factory() + control_client = _same_agent_client(stream_client) + assert control_client.agent.pre == stream_client.agent.pre + + stream = stream_client.signals().stream() + events = queue.Queue() + errors = queue.Queue() + + def read_first_data_event(): + try: + for event in stream: + if event.data: + events.put(event) + return + except Exception as err: # pragma: no cover - surfaced by main thread + errors.put(err) + + reader = threading.Thread(target=read_first_data_event, daemon=True) + reader.start() + + try: + poll_until( + lambda: control_client.get("/test/signals").json(), + ready=lambda body: body["subscribers"] >= 1, + timeout=20, + interval=POLL_INTERVAL, + describe="SSE subscription registration", + ) + + signal = { + "event": "agent.signal.test", + "event_id": "test-signal-1", + "route": "/test/signals/request", + "payload": {"subject": "sse-integration"}, + } + response = control_client.post("/test/signals", json=signal) + assert response.status_code == 202 + + event = _wait_for_event(events, errors) + assert event.event == signal["event"] + assert event.id == signal["event_id"] + + envelope = json.loads(event.data) + rpy = envelope["rpy"] + assert rpy["r"] == signal["route"] + assert rpy["a"]["subject"] == signal["payload"]["subject"] + assert rpy["a"]["agent"] == stream_client.agent.pre + assert stream_client.signals().verifyReplyEnvelope( + envelope, + route=signal["route"], + ) is True + finally: + try: + stream.close() + except Exception: + pass + reader.join(timeout=5)