feat(capability): add release scoped shadow bindings
CI / test (push) Failing after 35s

This commit is contained in:
m
2026-08-31 16:28:57 +03:00
parent d1710907ec
commit dec19e873d
36 changed files with 1078 additions and 50 deletions
@@ -0,0 +1,41 @@
package main
import (
"fmt"
"os"
"git.cin.su/m/Assistant/contracts/capability"
"git.cin.su/m/Assistant/contracts/identity"
)
// candidateReleaseContextFromRuntime accepts only deployment-owned metadata.
// It never derives a candidate slot from an HTTP request or notification body.
func candidateReleaseContextFromRuntime() (*capability.CandidateReleaseContext, error) {
const prefix = "CANDIDATE_"
release := os.Getenv(prefix + "RELEASE_SET_ID")
if release == "" {
return nil, nil
}
context := &capability.CandidateReleaseContext{
ReleaseSetID: release,
Slot: os.Getenv(prefix + "SLOT"),
TaskID: os.Getenv(prefix + "TASK_ID"),
PrincipalID: identity.PrincipalID(os.Getenv(prefix + "PRINCIPAL_ID")),
CapabilityID: capability.CapabilityID(os.Getenv(prefix + "CAPABILITY_ID")),
ResourceScope: os.Getenv(prefix + "RESOURCE_SCOPE"),
Mode: capability.CandidateBindingMode(os.Getenv(prefix + "BINDING_MODE")),
TargetKind: os.Getenv(prefix + "TARGET_KIND"),
TargetRef: os.Getenv(prefix + "TARGET_REF"),
TargetDigest: os.Getenv(prefix + "TARGET_DIGEST"),
AuthorityGrantID: os.Getenv(prefix + "AUTHORITY_GRANT_ID"),
}
var revision int64
if _, err := fmt.Sscan(os.Getenv(prefix+"TASK_REVISION"), &revision); err != nil {
return nil, fmt.Errorf("candidate task revision: %w", err)
}
context.TaskRevision = revision
if context.Slot == "" || context.TaskID == "" || context.TaskRevision < 1 || context.PrincipalID == "" || context.CapabilityID == "" || context.ResourceScope == "" || context.Mode == "" || context.TargetKind == "" || context.TargetRef == "" || context.TargetDigest == "" || context.AuthorityGrantID == "" {
return nil, fmt.Errorf("incomplete candidate release metadata")
}
return context, nil
}
@@ -0,0 +1,25 @@
package main
import "testing"
func TestCandidateReleaseContextRequiresDeploymentMetadata(t *testing.T) {
t.Setenv("CANDIDATE_RELEASE_SET_ID", "assistant-slot-b")
if _, err := candidateReleaseContextFromRuntime(); err == nil {
t.Fatal("partial candidate metadata was accepted")
}
t.Setenv("CANDIDATE_SLOT", "b")
t.Setenv("CANDIDATE_TASK_ID", "GEN2-CAPABILITY-001")
t.Setenv("CANDIDATE_TASK_REVISION", "7")
t.Setenv("CANDIDATE_PRINCIPAL_ID", "domovoy-creator-principal")
t.Setenv("CANDIDATE_CAPABILITY_ID", "prospective-notification-delivery")
t.Setenv("CANDIDATE_RESOURCE_SCOPE", "synthetic-shadow-notification-v1")
t.Setenv("CANDIDATE_BINDING_MODE", "SHADOW")
t.Setenv("CANDIDATE_TARGET_KIND", "NOTIFICATION_SHADOW_RECEIVER")
t.Setenv("CANDIDATE_TARGET_REF", "notification-receiver-candidate")
t.Setenv("CANDIDATE_TARGET_DIGEST", "f9f9fa49a0727444c280c092d5bb340f3a8f33cf325c5396f5ae177a6666da35")
t.Setenv("CANDIDATE_AUTHORITY_GRANT_ID", "grant")
context, err := candidateReleaseContextFromRuntime()
if err != nil || context.ReleaseSetID != "assistant-slot-b" || context.TaskRevision != 7 {
t.Fatalf("context=%+v err=%v", context, err)
}
}
@@ -12,6 +12,7 @@ import (
"time"
"git.cin.su/m/Assistant/contracts/authority"
"git.cin.su/m/Assistant/contracts/capability"
"git.cin.su/m/Assistant/contracts/identity"
"git.cin.su/m/Assistant/platform/ids"
)
@@ -22,6 +23,25 @@ type delegatedAuthorityFake struct {
revoked bool
}
func TestCandidateBindingResolutionUsesServiceCredential(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/v1/internal/capability-bindings/candidate/resolve" || r.Header.Get("Authorization") != "Bearer service-token" {
t.Fatalf("unexpected candidate resolution request: %s %q", r.URL.Path, r.Header.Get("Authorization"))
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"reason":"EXACT_CONTEXT"}`))
}))
defer server.Close()
client, err := newCreatorCoreClient(server.URL, server.Client())
if err != nil {
t.Fatal(err)
}
client.serviceCredential = func() (string, error) { return "service-token", nil }
if _, err := client.ResolveCandidateCapabilityBinding(context.Background(), capability.CandidateReleaseContext{}); err != nil {
t.Fatalf("resolve candidate binding: %v", err)
}
}
func TestDelegatedAuthorityCredentialRotationRejectsOldToken(t *testing.T) {
root := t.TempDir()
tokenPath := filepath.Join(root, "control_plane_service_token")
+55 -2
View File
@@ -25,8 +25,9 @@ var (
)
type creatorCoreClient struct {
baseURL *url.URL
client *http.Client
baseURL *url.URL
client *http.Client
serviceCredential func() (string, error)
}
type creatorContinuityClient struct {
@@ -121,6 +122,58 @@ func (c creatorCoreClient) CapabilityBinding(ctx context.Context, id capability.
return binding, nil
}
func (c creatorCoreClient) ResolveCandidateCapabilityBinding(ctx context.Context, context capability.CandidateReleaseContext) (capability.CandidateBindingResolution, error) {
var resolved capability.CandidateBindingResolution
status, err := c.postServiceAuthenticated(ctx, "/v1/internal/capability-bindings/candidate/resolve", context, &resolved)
if err != nil || status != http.StatusOK {
return capability.CandidateBindingResolution{}, errCreatorQueryUnavailable
}
return resolved, nil
}
func (c creatorCoreClient) postServiceAuthenticated(ctx context.Context, path string, value any, destination any) (int, error) {
if c.serviceCredential == nil {
return 0, errCreatorQueryUnavailable
}
credential, err := c.serviceCredential()
if err != nil || credential == "" {
return 0, errCreatorQueryUnavailable
}
return c.postWithAuthorization(ctx, path, value, destination, "Bearer "+credential)
}
func (c creatorCoreClient) post(ctx context.Context, path string, value any, destination any) (int, error) {
return c.postWithAuthorization(ctx, path, value, destination, "")
}
func (c creatorCoreClient) postWithAuthorization(ctx context.Context, path string, value any, destination any, authorization string) (int, error) {
target := *c.baseURL
target.Path, target.RawQuery = path, ""
body, err := json.Marshal(value)
if err != nil {
return 0, err
}
request, err := http.NewRequestWithContext(ctx, http.MethodPost, target.String(), strings.NewReader(string(body)))
if err != nil {
return 0, err
}
request.Header.Set("Content-Type", "application/json")
if authorization != "" {
request.Header.Set("Authorization", authorization)
}
response, err := c.client.Do(request)
if err != nil {
return 0, err
}
defer response.Body.Close()
if response.StatusCode == http.StatusOK && destination != nil {
if err := json.NewDecoder(io.LimitReader(response.Body, 32<<10)).Decode(destination); err != nil {
return 0, err
}
}
return response.StatusCode, nil
}
func (c creatorCoreClient) get(ctx context.Context, path, authorization string, destination any) (int, error) {
target := *c.baseURL
target.Path = path
+9 -1
View File
@@ -24,6 +24,7 @@ type executionRepository interface {
LoadLatestAttempt(context.Context, commitcontract.AuthorizedActionID) (commitcontract.ActionAttempt, error)
BeginAttempt(context.Context, commitcontract.ActionAttempt) error
FinishAttempt(context.Context, commitcontract.ActionAttempt) error
SaveExecutionReceipt(context.Context, commitcontract.ExecutionReceipt) error
}
type attemptRequest struct {
@@ -94,10 +95,17 @@ func executeAuthorized(ctx context.Context, repository executionRepository, exec
if err := repository.BeginAttempt(ctx, begun); err != nil {
return commitcontract.ActionAttempt{}, err
}
attempt, operationErr := commitdomain.Execute(action, attemptID, executor)
attempt, result, operationErr := commitdomain.ExecuteWithResult(action, attemptID, executor)
if err := repository.FinishAttempt(ctx, attempt); err != nil {
return commitcontract.ActionAttempt{}, err
}
if result.Receipt != nil {
receipt := *result.Receipt
receipt.ActionAttemptRef = attempt.ID
if err := repository.SaveExecutionReceipt(ctx, receipt); err != nil {
return commitcontract.ActionAttempt{}, err
}
}
return attempt, operationErr
}
+11 -1
View File
@@ -16,6 +16,12 @@ import (
type fakeExecutionRepository struct {
action commitcontract.AuthorizedAction
attempts []commitcontract.ActionAttempt
receipt commitcontract.ExecutionReceipt
}
func (f *fakeExecutionRepository) SaveExecutionReceipt(_ context.Context, receipt commitcontract.ExecutionReceipt) error {
f.receipt = receipt
return nil
}
func (f *fakeExecutionRepository) LoadAuthorized(context.Context, commitcontract.AuthorizedActionID) (commitcontract.AuthorizedAction, error) {
@@ -47,7 +53,8 @@ type knownExecutor struct{ calls int }
func (e *knownExecutor) Execute(commitcontract.AuthorizedAction) ports.ExecutionResult {
e.calls++
return ports.ExecutionResult{Known: true, Succeeded: true}
completed := int64(1)
return ports.ExecutionResult{Known: true, Succeeded: true, Receipt: &commitcontract.ExecutionReceipt{ID: "receipt-1", ExecutorImplementationRef: "test-executor", ExternalOperationRef: "SHADOW_CAPTURED", RequestDigest: "request", ExecutorResult: commitcontract.ExecutorReportedSuccess, ReconciliationKey: "receipt-1", DispatchAtUnixMillis: 1, CompletedAtUnixMillis: &completed}}
}
func (e *knownExecutor) Reconcile(commitcontract.AuthorizedAction) ports.ExecutionResult {
return ports.ExecutionResult{Known: true, Succeeded: true}
@@ -66,6 +73,9 @@ func TestExecutionIsDurablyBegunAndCannotBeBlindlyRepeated(t *testing.T) {
if result.Code != http.StatusOK || executor.calls != 1 || repository.attempts[0].State != commitcontract.ActionSucceeded {
t.Fatalf("status=%d calls=%d attempts=%#v", result.Code, executor.calls, repository.attempts)
}
if repository.receipt.ActionAttemptRef != repository.attempts[0].ID {
t.Fatalf("receipt attempt reference=%q want %q", repository.receipt.ActionAttemptRef, repository.attempts[0].ID)
}
repeated := httptest.NewRequest(http.MethodPost, "/v1/internal/authorized-actions/authorized-1/execute", strings.NewReader(`{"AttemptID":"attempt-2"}`))
repeated.Header.Set("Authorization", "Bearer secret")
+26 -5
View File
@@ -73,15 +73,18 @@ func main() {
if err != nil {
log.Fatal(err)
}
notificationSecret, err := readCommitSecretFile("NOTIFICATION_WRITE_TOKEN_FILE")
candidateContext, err := candidateReleaseContextFromRuntime()
if err != nil {
log.Fatal(err)
}
notificationSecretRef := os.Getenv("NOTIFICATION_WRITE_SECRET_REF")
if notificationSecretRef == "" {
log.Fatal("NOTIFICATION_WRITE_SECRET_REF is required")
notificationSecretRef, notificationSecret, err := notificationCredentialsFromRuntime(candidateContext != nil)
if err != nil {
log.Fatal(err)
}
dispatcher := operationDispatcher{continuityExecutor: continuityExecutor, continuityReconciler: continuityExecutor, bindings: creatorCore, secretRef: notificationSecretRef, secret: notificationSecret, client: creatorHTTPClient}
if candidateContext != nil {
creatorCore.serviceCredential = delegatedControlPlaneCredential
}
dispatcher := operationDispatcher{continuityExecutor: continuityExecutor, continuityReconciler: continuityExecutor, bindings: creatorCore, secretRef: notificationSecretRef, secret: notificationSecret, client: creatorHTTPClient, candidateContext: candidateContext}
mux := http.NewServeMux()
mux.Handle("/healthz", health.Handler{Runtime: "commit-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1})
readinessClient := &http.Client{Timeout: 2 * time.Second}
@@ -162,6 +165,24 @@ func main() {
}
}
// notificationCredentialsFromRuntime keeps candidate shadow execution
// physically separate from the production notification credential mount.
func notificationCredentialsFromRuntime(candidate bool) (string, string, error) {
fileName, refName := "NOTIFICATION_WRITE_TOKEN_FILE", "NOTIFICATION_WRITE_SECRET_REF"
if candidate {
fileName, refName = "CANDIDATE_NOTIFICATION_WRITE_TOKEN_FILE", "CANDIDATE_NOTIFICATION_WRITE_SECRET_REF"
}
secret, err := readCommitSecretFile(fileName)
if err != nil {
return "", "", err
}
ref := os.Getenv(refName)
if ref == "" {
return "", "", fmt.Errorf("%s is required", refName)
}
return ref, secret, nil
}
func readCommitSecretFile(name string) (string, error) {
path := os.Getenv(name)
if path == "" {
@@ -14,6 +14,7 @@ import (
type notificationBindingSource interface {
ActiveCapabilityBinding(context.Context, capability.CapabilityID, identity.PrincipalID) (capability.Binding, error)
CapabilityBinding(context.Context, capability.CapabilityBindingID, int64) (capability.Binding, error)
ResolveCandidateCapabilityBinding(context.Context, capability.CandidateReleaseContext) (capability.CandidateBindingResolution, error)
}
type operationDispatcher struct {
@@ -23,6 +24,7 @@ type operationDispatcher struct {
secretRef string
secret string
client *http.Client
candidateContext *capability.CandidateReleaseContext
}
func (d operationDispatcher) Execute(action commitcontract.AuthorizedAction) ports.ExecutionResult {
@@ -30,6 +32,20 @@ func (d operationDispatcher) Execute(action commitcontract.AuthorizedAction) por
return d.continuityExecutor.Execute(action)
}
subject := action.Proposal.NotificationSubject
if d.candidateContext != nil {
resolved, err := d.bindings.ResolveCandidateCapabilityBinding(context.Background(), *d.candidateContext)
if err != nil || resolved.Mode != capability.CandidateBindingModeShadow || resolved.Binding.Context != *d.candidateContext {
return ports.ExecutionResult{Known: true}
}
if subject.BindingID != resolved.Binding.BaseActiveBindingID || subject.BindingRevision != resolved.Binding.BaseActiveBindingRevision {
return ports.ExecutionResult{Known: true}
}
executor, err := notificationexecutor.NewShadow(resolved.Binding, d.secret, d.client)
if err != nil {
return ports.ExecutionResult{Known: true}
}
return executor.Execute(action)
}
exact, err := d.bindings.CapabilityBinding(context.Background(), subject.BindingID, subject.BindingRevision)
if err != nil || exact.SecretRef != d.secretRef {
return ports.ExecutionResult{Known: true}
@@ -49,6 +65,11 @@ func (d operationDispatcher) Reconcile(action commitcontract.AuthorizedAction) p
if action.Proposal.Operation != commitcontract.OperationDeliverProspectiveNotification {
return d.continuityReconciler.Reconcile(action)
}
if d.candidateContext != nil {
// A candidate shadow delivery is never reconciled against the ACTIVE
// production target. Its typed receipt is the sole candidate evidence.
return ports.ExecutionResult{Known: true}
}
subject := action.Proposal.NotificationSubject
exact, err := d.bindings.CapabilityBinding(context.Background(), subject.BindingID, subject.BindingRevision)
if err != nil || exact.SecretRef != d.secretRef {
@@ -3,21 +3,43 @@ package main
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"os"
"testing"
"git.cin.su/m/Assistant/adapters/notificationexecutor"
"git.cin.su/m/Assistant/contracts/capability"
commitcontract "git.cin.su/m/Assistant/contracts/commit"
"git.cin.su/m/Assistant/contracts/identity"
capabilitydomain "git.cin.su/m/Assistant/modules/capability/domain"
)
func TestCandidateNotificationCredentialsDoNotFallbackToProductionCredential(t *testing.T) {
path := t.TempDir() + "/candidate-token"
if err := os.WriteFile(path, []byte("candidate-only-token"), 0o600); err != nil {
t.Fatal(err)
}
t.Setenv("CANDIDATE_NOTIFICATION_WRITE_TOKEN_FILE", path)
t.Setenv("CANDIDATE_NOTIFICATION_WRITE_SECRET_REF", "credential://domovoy/candidate/notification-write/v1")
ref, secret, err := notificationCredentialsFromRuntime(true)
if err != nil || ref == "" || secret != "candidate-only-token" {
t.Fatalf("candidate credential resolution failed: ref=%q err=%v", ref, err)
}
t.Setenv("CANDIDATE_NOTIFICATION_WRITE_TOKEN_FILE", "")
if _, _, err := notificationCredentialsFromRuntime(true); err == nil {
t.Fatal("candidate path fell back to production notification credential")
}
}
type fakeNotificationBindings struct {
exact capability.Binding
active capability.Binding
exactErr error
activeErr error
exact capability.Binding
active capability.Binding
exactErr error
activeErr error
candidate capability.CandidateBindingResolution
candidateErr error
}
func (f fakeNotificationBindings) CapabilityBinding(context.Context, capability.CapabilityBindingID, int64) (capability.Binding, error) {
@@ -26,6 +48,9 @@ func (f fakeNotificationBindings) CapabilityBinding(context.Context, capability.
func (f fakeNotificationBindings) ActiveCapabilityBinding(context.Context, capability.CapabilityID, identity.PrincipalID) (capability.Binding, error) {
return f.active, f.activeErr
}
func (f fakeNotificationBindings) ResolveCandidateCapabilityBinding(context.Context, capability.CandidateReleaseContext) (capability.CandidateBindingResolution, error) {
return f.candidate, f.candidateErr
}
func dispatcherAction(t *testing.T, binding capability.Binding) commitcontract.AuthorizedAction {
t.Helper()
@@ -81,3 +106,28 @@ func TestOperationDispatcherReconcilesRetiredHistoricalBinding(t *testing.T) {
t.Fatalf("reconcile=%+v", result)
}
}
func TestCandidateDispatcherUsesShadowBindingWithoutActiveFallback(t *testing.T) {
var action commitcontract.AuthorizedAction
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("X-Candidate-Release") != "assistant-slot-b" || r.Header.Get("X-Candidate-Task") != "GEN2-CAPABILITY-001" {
t.Fatalf("candidate headers missing")
}
_ = json.NewEncoder(w).Encode(notificationexecutor.Receipt{ReceiptID: "shadow-1", ActionIntentID: action.Proposal.IntentID, RequestDigest: action.Proposal.NotificationSubject.RequestDigest, AcceptedAt: 1, DeliveryMode: "SHADOW"})
}))
defer server.Close()
stable := dispatcherBinding("https://stable.invalid")
action = dispatcherAction(t, stable)
context := capability.CandidateReleaseContext{ReleaseSetID: "assistant-slot-b", Slot: "b", TaskID: "GEN2-CAPABILITY-001", TaskRevision: 7, PrincipalID: stable.OwnerPrincipalID, CapabilityID: stable.CapabilityID, ResourceScope: "synthetic-shadow-notification-v1", Mode: capability.CandidateBindingModeShadow, TargetKind: "NOTIFICATION_SHADOW_RECEIVER", TargetRef: "notification-receiver-candidate", AuthorityGrantID: "grant"}
candidate := capability.CandidateBinding{ID: "candidate-1", Revision: 1, BaseActiveBindingID: stable.ID, BaseActiveBindingRevision: stable.Revision, Context: context, TargetOrigin: server.URL, DeliveryPath: "/delivery/v1/deliveries", ReceiptPath: "/delivery/v1/receipts", Protocol: capability.ProtocolIdempotentReceiptV1, CreatedAtUnixMillis: 1, ExpiresAtUnixMillis: 100, Status: capability.CandidateBindingActive}
digest, err := capabilitydomain.CandidateTargetDigest(candidate)
if err != nil {
t.Fatal(err)
}
context.TargetDigest, candidate.Context.TargetDigest = digest, digest
dispatcher := operationDispatcher{continuityExecutor: &knownExecutor{}, continuityReconciler: &knownExecutor{}, bindings: fakeNotificationBindings{activeErr: errors.New("stable fallback must not run"), candidate: capability.CandidateBindingResolution{Binding: candidate, Mode: capability.CandidateBindingModeShadow}}, secret: "shadow-secret", client: server.Client(), candidateContext: &context}
result := dispatcher.Execute(action)
if !result.Known || !result.Succeeded || result.Receipt == nil || result.Receipt.DeliveryMode != "SHADOW" {
t.Fatalf("candidate result=%+v", result)
}
}