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
6 changes: 4 additions & 2 deletions agentplane/docs/system_overview.md
Original file line number Diff line number Diff line change
Expand Up @@ -551,7 +551,7 @@ flowchart LR
agent -.->|HTTP subscriptions, inboxes, messages| sidecar
sidecar -.->|Sandbox SVID to destination inside CONNECT| proxy
proxy -.->|opaque TLS| notifications
app -->|gRPC operator watches + health| notifications
app -->|gRPC operator watches + health; optional exact service-SVID reader| notifications
github --> notifications
notifications --> github
notifications -->|Action history: gRPC| actions
Expand All @@ -565,7 +565,9 @@ Sandboxes keep their existing HTTP service URL. Their sidecar or relay recognize
opens CONNECT through the central proxy, and presents the Runner SVID to Notification's shared TLS
RPC listener. Notification resolves the current Sandbox and its ServiceAccount; inbox ownership and
send grants are unchanged. Backend and operator callers retain their method-specific bearer
authorization on that listener.
authorization on that listener. Independently configured service SVIDs may read Session diagnostics
through the read-only operator methods; that grant never supplies an agent Identity or permits
inbox mutation. Deployment settings and callers retain their current authentication until cutover.

The Notification service does not call the integration app, which watches subscription and notice
state over the operator-only `NotificationsOperator` gRPC service. GitHub sends it signed webhooks; it
Expand Down
10 changes: 7 additions & 3 deletions agentplane/notification_service/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,9 @@ Sandbox callers authenticate with the Runner SVID on the shared TLS RPC listener
Egress routes workload gRPC through an opaque central-proxy CONNECT. Notification verifies the
certificate chain, explicit clientAuth EKU, expiry, current Sandbox UID and accepted credential ID,
then resolves the credential's exact live Identity name and UID. Caller headers cannot select another Identity. Native gRPC uses this resolver,
and watch revalidation repeats the live lookup. Backend/operator callers use the same listener
with their audience-scoped tokens and existing authorization.
and watch revalidation repeats the live lookup. Backend and legacy operator callers use the same listener
with their audience-scoped tokens and existing authorization. An exact configured service SVID may call
only the namespace diagnostic read methods; it does not become an agent Identity or own an inbox.

## Running and deployment

Expand Down Expand Up @@ -130,7 +131,10 @@ that fails validation stops startup without printing its value.
The native gRPC listener (`rpc.listen_address`) serves `Notifications` and `grpc.health.v1`.
`rpc.tls` supplies its server certificate and key, and `rpc.auth` explicitly admits SPIFFE and/or
Kubernetes bearer authentication. Both modes retain the same method and inbox ownership policy.
A certificate combined with bearer or delegation metadata is refused.
A certificate combined with bearer or delegation metadata is refused. `operator_reader_services`
admits exact service SPIFFE IDs to the five diagnostic read methods; it requires SPIFFE client trust
and can coexist with `operator_reader_account` during a rollout. A service SVID grants neither an
agent Identity nor inbox ownership, and is not admitted to mutations or other RPCs.
The HTTP listener (`host`, `http_port`) serves only `POST /v1/webhooks/github` when GitHub is configured; all other
paths return not found. It has no workload HTTP API or documentation routes. The generated App browser facade dials
the gRPC listener, and the CLI embeds the workflow guides. Health checks take no bearer: service `""` is readiness (the LISTEN
Expand Down
27 changes: 19 additions & 8 deletions agentplane/notification_service/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"strings"
"time"

"github.com/agentydragon/ducktape/agentplane/identity/auth"
"github.com/agentydragon/ducktape/agentplane/identity/auth/listener"
"github.com/agentydragon/ducktape/agentplane/net/peertls"
"github.com/agentydragon/ducktape/agentplane/notification_service/internal/sources/github/settings"
Expand All @@ -36,14 +37,16 @@ type config struct {
Namespace string `mapstructure:"namespace"`
TokenAudience string `mapstructure:"token_audience"`
// OperatorReaderAccount is the ServiceAccount allowed to inspect Session diagnostics across the namespace.
OperatorReaderAccount string `mapstructure:"operator_reader_account"`
Quotas store.Quotas `mapstructure:"quotas"`
worker.Settings `mapstructure:",squash"`
GitHubRetention settings.Retention `mapstructure:",squash"`
Actions actionPeer `mapstructure:"actions"`
RunnerGateway runnerGateway `mapstructure:"runner_gateway"`
HistoryService peer `mapstructure:"history_service"`
SessionDispatcher peer `mapstructure:"session_dispatcher"`
OperatorReaderAccount string `mapstructure:"operator_reader_account"`
// OperatorReaderServices are exact service SVIDs admitted to the same read-only diagnostic methods.
OperatorReaderServices []string `mapstructure:"operator_reader_services"`
Quotas store.Quotas `mapstructure:"quotas"`
worker.Settings `mapstructure:",squash"`
GitHubRetention settings.Retention `mapstructure:",squash"`
Actions actionPeer `mapstructure:"actions"`
RunnerGateway runnerGateway `mapstructure:"runner_gateway"`
HistoryService peer `mapstructure:"history_service"`
SessionDispatcher peer `mapstructure:"session_dispatcher"`
// GitHub is on when any of its settings is set.
GitHub settings.Settings `mapstructure:"github"`
// RequestTimeoutSeconds bounds a call's backend work and each write to a watch stream.
Expand Down Expand Up @@ -230,6 +233,14 @@ func (c *config) validate() error {
if err := c.RPC.Validate(); err != nil {
return fmt.Errorf("rpc listener: %w", err)
}
if err := auth.ValidateServiceCallers(c.OperatorReaderServices); err != nil {
return fmt.Errorf("invalid operator_reader_services: %w", err)
}
if len(c.OperatorReaderServices) > 0 {
if !c.RPC.Has(listener.SPIFFE) {
return errors.New("operator_reader_services requires rpc spiffe client trust")
}
}
if !c.RPC.AllowsBearerAudience(c.TokenAudience) {
return errors.New("rpc.auth must allow the configured token_audience")
}
Expand Down
16 changes: 16 additions & 0 deletions agentplane/notification_service/config_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,22 @@ func TestTheSettingsTheDeploymentManifestsSetLoadWithTheirSecretsFromTheEnvironm
}
}

func TestOperatorReaderAccountAndServiceSVIDCanCoexist(t *testing.T) {
file := strings.Replace(
deploymentFile,
"operator_reader_account: test-app\n",
"operator_reader_account: test-app\noperator_reader_services: [spiffe://agentplane/app]\n",
1,
)
c, err := loadTestConfig(t, file, webhookSecretEnv, privateKeyEnv)
if err != nil {
t.Fatal(err)
}
if c.OperatorReaderAccount != "test-app" || len(c.OperatorReaderServices) != 1 || c.OperatorReaderServices[0] != "spiffe://agentplane/app" {
t.Fatalf("operator readers were not retained together: account=%q services=%v", c.OperatorReaderAccount, c.OperatorReaderServices)
}
}

func TestWhatADeploymentLeavesUnsetTakesItsDefault(t *testing.T) {
c, err := loadTestConfig(t, requiredFile)
if err != nil {
Expand Down
13 changes: 7 additions & 6 deletions agentplane/notification_service/docs/api.md
Original file line number Diff line number Diff line change
Expand Up @@ -47,9 +47,9 @@ for the exact original Sandbox UID. Unknown or inconsistent owner identity pause
or idempotently cancel. Cancelling never cancels an Action or erases accepted entries.
- `GET /v1/inboxes`: the Identity's qualified session inboxes.
- `GET /v1/sessions/{session_id}/inbox/entries?after_cursor=0&limit=128`: non-destructive, contiguous paging, prefix
metadata, and the latest notice's admission/confirmation state. The configured app account may inspect
retained entries by Session ID through its authenticated operator facade, including after the original
Sandbox disappears. It cannot acknowledge or retire them. Payloads are stored provider event snapshots,
metadata, and the latest notice's admission/confirmation state. The configured operator account and exact
service SVID readers may inspect retained entries by Session ID, including after the original Sandbox
disappears. Diagnostic readers cannot acknowledge or retire another caller's inbox. Payloads are stored provider event snapshots,
not fetched at read time or promoted to operator instructions.
- `PUT /v1/sessions/{session_id}/inbox/acknowledgement` with `through_cursor`: monotonically acknowledge the handled
prefix. Future cursors are refused. Delivery and reads never acknowledge.
Expand Down Expand Up @@ -303,9 +303,10 @@ lifecycle fields; it never returns provider payloads. `WatchSubscriptions` repor
state. Each watch starts with current state; entry and subscription watches then send changes.
A missing inbox is a state, since inboxes are created lazily. Reconnecting starts from fresh state.
Watches send heartbeats and revalidate authorization; they do not end on a fixed lease. Reading
never acknowledges entries. Workloads can read their own Sandbox incarnation's inbox; the
configured `operator_reader_account` can inspect Session diagnostics in the namespace but cannot
read full inbox payloads or mutate another account's inbox.
never acknowledges entries. Workloads can read their own Sandbox incarnation's inbox; the configured
`operator_reader_account` and exact `operator_reader_services` SVIDs can inspect Session diagnostics
in the namespace. Service SVID admission is restricted to these read-only methods and never asserts
an agent Identity. Diagnostic readers cannot mutate another caller's inbox.

A Session has at most one inbox. The database migration refuses existing collisions rather than
choosing or discarding records; reconcile any collisions before admitting writes with the new schema.
Expand Down
48 changes: 44 additions & 4 deletions agentplane/notification_service/internal/rpc/fixture_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -298,14 +298,16 @@ type serverConfig struct {
debounce store.Debounce
githubDisabled bool
// operatorReader is the account allowed to inspect namespace Session diagnostics.
operatorReader string
operatorReader string
operatorReaderServices []string
// messageBodyLimit is in UTF-8 bytes.
messageBodyLimit int
}

type testServer struct {
notifications *Notifications
client notificationspb.NotificationsClient
serviceClient notificationspb.NotificationsClient
clientRoutes *routedClientConn
store *store.Store
hub *wakeups.Hub
Expand All @@ -321,6 +323,7 @@ type testServer struct {
workload *workloadauthtesting.IdentityWorkload
ca *transporttesting.CA
identityAddress string
serviceAddress string
// kube holds the Sandboxes the service reads.
kube client.Client
config serverConfig
Expand Down Expand Up @@ -442,7 +445,8 @@ func newServer(t *testing.T, configure ...func(*serverConfig)) testServer {
options := Options{
Store: server.store, Wakeups: hub, Namespace: testNamespace, Sandboxes: sandboxes,
Runner: server.runner, Actions: server.actions, OperatorReader: config.operatorReader,
SendPolicy: server.policy, History: server.history, Sessions: server.openings, MessageBodyLimit: config.messageBodyLimit,
OperatorReaderServices: config.operatorReaderServices,
SendPolicy: server.policy, History: server.history, Sessions: server.openings, MessageBodyLimit: config.messageBodyLimit,
Debounce: config.debounce, RequestTimeout: config.requestTimeout,
}
if !config.githubDisabled {
Expand All @@ -453,7 +457,11 @@ func newServer(t *testing.T, configure ...func(*serverConfig)) testServer {
if err != nil {
t.Fatal(err)
}
grpcServer := grpc.NewServer(Methods(options, resolver).ServerOptions(config.requestTimeout)...)
methods, err := Methods(options, resolver)
if err != nil {
t.Fatal(err)
}
grpcServer := grpc.NewServer(methods.ServerOptions(config.requestTimeout)...)
notificationspb.RegisterNotificationsServer(grpcServer, server.notifications)
go func() { _ = grpcServer.Serve(listener) }()
t.Cleanup(grpcServer.Stop)
Expand Down Expand Up @@ -483,16 +491,48 @@ func newServer(t *testing.T, configure ...func(*serverConfig)) testServer {
if err != nil {
t.Fatal(err)
}
identityGRPCServer := grpc.NewServer(append(Methods(options, identityResolver).ServerOptions(config.requestTimeout), grpc.Creds(credentials.NewTLS(serverTLS)))...)
identityMethods, err := Methods(options, identityResolver)
if err != nil {
t.Fatal(err)
}
identityGRPCServer := grpc.NewServer(append(identityMethods.ServerOptions(config.requestTimeout), grpc.Creds(credentials.NewTLS(serverTLS)))...)
notificationspb.RegisterNotificationsServer(identityGRPCServer, server.notifications)
go func() { _ = identityGRPCServer.Serve(identityListener) }()
t.Cleanup(identityGRPCServer.Stop)
identityConn := connectIdentityClient(t, server.ca, identityWorkload.Workload, identityListener.Addr().String())
server.identityAddress = identityListener.Addr().String()
server.clientRoutes = &routedClientConn{account: connection, identity: identityConn}
var serviceConn *grpc.ClientConn
if len(config.operatorReaderServices) > 0 {
serviceTLS, err := serving.WorkloadServerTLSConfigWithServices(peertls.NotificationsURI, true)
if err != nil {
t.Fatal(err)
}
serviceListener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
server.serviceAddress = serviceListener.Addr().String()
serviceServer := grpc.NewServer(append(methods.ServerOptions(config.requestTimeout), grpc.Creds(credentials.NewTLS(serviceTLS)))...)
notificationspb.RegisterNotificationsServer(serviceServer, server.notifications)
go func() { _ = serviceServer.Serve(serviceListener) }()
t.Cleanup(serviceServer.Stop)
serviceCredentials, err := server.ca.Issue(t, peertls.AppURI).ClientCredentials(peertls.NotificationsURI)
if err != nil {
t.Fatal(err)
}
serviceConn, err = grpc.NewClient(serviceListener.Addr().String(), grpc.WithTransportCredentials(serviceCredentials))
if err != nil {
t.Fatal(err)
}
server.serviceClient = notificationspb.NewNotificationsClient(serviceConn)
}
t.Cleanup(func() {
_ = connection.Close()
_ = identityConn.Close()
if serviceConn != nil {
_ = serviceConn.Close()
}
})
server.client = notificationspb.NewNotificationsClient(server.clientRoutes)
return server
Expand Down
5 changes: 4 additions & 1 deletion agentplane/notification_service/internal/rpc/methods_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,10 @@ func TestEveryRPCAdmitsOnlyItsCallers(t *testing.T) {
{"Notifications", operatorName, notificationspb.Notifications_ServiceDesc, workloads},
} {
t.Run(test.name, func(t *testing.T) {
methods := Methods(Options{Namespace: testNamespace, OperatorReader: test.reader}, resolver)
methods, err := Methods(Options{Namespace: testNamespace, OperatorReader: test.reader}, resolver)
if err != nil {
t.Fatal(err)
}
connection := grpcmethods.Serve(t, methods, registerStandIns)
for _, method := range grpcmethods.Of(test.desc) {
t.Run(method.Name, func(t *testing.T) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -36,18 +36,23 @@ type Notifications struct {
timeout time.Duration
wakeups *wakeups.Hub
operatorReader string
operatorServices map[string]struct{}
debounce store.Debounce
draining context.Context
drain context.CancelFunc
}

func NewNotifications(options Options) *Notifications {
draining, drain := context.WithCancel(context.Background())
operatorServices := make(map[string]struct{}, len(options.OperatorReaderServices))
for _, service := range options.OperatorReaderServices {
operatorServices[service] = struct{}{}
}
return &Notifications{
store: options.Store, namespace: options.Namespace, sandboxes: options.Sandboxes,
runner: options.Runner, actions: options.Actions, github: options.GitHub, policy: options.SendPolicy,
history: options.History, sessions: options.Sessions, messageBodyLimit: options.MessageBodyLimit, timeout: options.RequestTimeout,
wakeups: options.Wakeups, operatorReader: options.OperatorReader, debounce: options.Debounce,
wakeups: options.Wakeups, operatorReader: options.OperatorReader, operatorServices: operatorServices, debounce: options.Debounce,
draining: draining, drain: drain,
}
}
Expand Down
41 changes: 41 additions & 0 deletions agentplane/notification_service/internal/rpc/operator_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,11 @@ import (
"strings"
"testing"

"github.com/agentydragon/ducktape/agentplane/net/peertls"
pb "github.com/agentydragon/ducktape/agentplane/notification_service/api/notifications/notificationspb"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)

func TestOperatorInspectionDoesNotGrantInboxMutation(t *testing.T) {
Expand Down Expand Up @@ -61,6 +64,44 @@ func TestOperatorPayloadReadRequiresConfiguredAccount(t *testing.T) {
requireCode(t, err, codes.NotFound)
}

func TestAppServiceSVIDCanReadDiagnosticsWithoutBecomingAnIdentity(t *testing.T) {
s := newServer(t, func(config *serverConfig) {
config.operatorReaderServices = []string{peertls.AppURI}
})
subscription := s.subscribe(t, "app-svid-reader")
s.record(t, subscription, 1)

page, err := s.serviceClient.ReadInbox(bearing(t, ""), &pb.ReadInboxRequest{SessionId: testSession})
if err != nil || len(page.GetEntries()) != 1 || page.GetInbox().GetAcknowledged() != 0 {
t.Fatalf("service SVID read = %v, %v", page, err)
}
_, err = s.serviceClient.AcknowledgeInbox(bearing(t, ""), &pb.AcknowledgeInboxRequest{
SessionId: testSession, ThroughCursor: 1,
})
requireCode(t, err, codes.PermissionDenied)
_, err = s.serviceClient.ListInboxes(bearing(t, ""), &pb.ListInboxesRequest{})
requireCode(t, err, codes.PermissionDenied)

legacyPage, err := s.client.ReadInbox(bearing(t, operatorToken), &pb.ReadInboxRequest{SessionId: testSession})
if err != nil || len(legacyPage.GetEntries()) != 1 {
t.Fatalf("legacy account reader stopped working during overlap = %v, %v", legacyPage, err)
}

wrongCredentials, err := s.ca.Issue(t, peertls.ActionsURI).ClientCredentials(peertls.NotificationsURI)
if err != nil {
t.Fatal(err)
}
wrongConnection, err := grpc.NewClient(s.serviceAddress, grpc.WithTransportCredentials(wrongCredentials))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = wrongConnection.Close() })
_, err = pb.NewNotificationsClient(wrongConnection).ReadInbox(bearing(t, ""), &pb.ReadInboxRequest{SessionId: testSession})
if got := status.Code(err); got != codes.PermissionDenied {
t.Fatalf("unallowlisted service SVID read status = %v, want %v (%v)", got, codes.PermissionDenied, err)
}
}

func TestSessionInboxReadsAreAuthorizedAndDoNotAcknowledge(t *testing.T) {
s := newServer(t)
sub := s.subscribe(t, "session-status")
Expand Down
6 changes: 5 additions & 1 deletion agentplane/notification_service/internal/rpc/ownership.go
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ func (s *Notifications) currentWorkload(ctx context.Context) (workload, error) {
}
}

// operator reports whether the admitted caller is the configured operator reader, a ServiceAccount.
// operator reports whether the admitted caller is a configured ServiceAccount or service-SVID diagnostic reader.
func (s *Notifications) operator(ctx context.Context) (bool, error) {
admission, err := auth.Admitted(ctx)
if err != nil {
Expand All @@ -57,6 +57,10 @@ func (s *Notifications) operator(ctx context.Context) (bool, error) {
if _, ok := admission.(auth.IdentityAdmission); ok {
return false, nil
}
if service, ok := admission.(auth.ServiceAdmission); ok {
_, allowed := s.operatorServices[service.Caller.SPIFFEID]
return allowed, nil
}
caller, err := s.caller(ctx)
if err != nil {
return false, err
Expand Down
Loading
Loading