diff --git a/core/module/handler/stdHandler.go b/core/module/handler/stdHandler.go index fa47b2472..dccee3557 100644 --- a/core/module/handler/stdHandler.go +++ b/core/module/handler/stdHandler.go @@ -339,6 +339,7 @@ func (h *stdHandler) stepCtx(r *http.Request, rh http.Header) (*model.StepContex Body: body, Role: h.role, SubID: subID, + OwnSubscriberID: h.SubscriberID, RespHeader: rh, ProtocolVersion: protocolVersion, MessageID: messageID, diff --git a/pkg/model/admittednetwork.go b/pkg/model/admittednetwork.go index 09844ae03..105db5a8c 100644 --- a/pkg/model/admittednetwork.go +++ b/pkg/model/admittednetwork.go @@ -1,9 +1,25 @@ package model -import "time" +import ( + "strings" + "time" +) + +// A schema pack names its attributes document; a message is interpreted +// against the context document beside it. +const ( + packFile = "attributes.yaml" + contextFile = "context.jsonld" +) + +// Agreement is what we agreed with another network. A direction is agreed when +// it is present; our own entry, and every provider and upstream, has neither. +type Agreement struct { + Source *FederationDirection `json:"source,omitempty"` + Requester *FederationDirection `json:"requester,omitempty"` +} // AdmittedNetwork is another network we admitted, as our registry holds it. -// Unlike Subscription, it carries what we agreed with that network. type AdmittedNetwork struct { ID string DiscoveryURL string @@ -11,42 +27,73 @@ type AdmittedNetwork struct { // How long this network lets us keep what we crawled. Zero forbids caching. ProjectionTtl time.Duration - // A direction is agreed when it is present. Our own entry, and every - // provider and upstream, carries neither. - Source *FederationDirection - Requester *FederationDirection + Agreement } // FederationDirection is one agreed direction. SchemaPacks is what the // agreement covers: for a source what we take, for a requester what they may ask. type FederationDirection struct { - SchemaPacks []string + SchemaPacks []string `json:"schema_packs,omitempty"` } +// Agreed reports whether this is a network we admitted at all. False is an +// ordinary participant, which no federation rule applies to. +func (a Agreement) Agreed() bool { return a.Source != nil || a.Requester != nil } + // IsSource reports whether we agreed to read from this network. -func (n AdmittedNetwork) IsSource() bool { return n.Source != nil } +func (a Agreement) IsSource() bool { return a.Source != nil } // IsRequester reports whether we agreed this network may read from us. -func (n AdmittedNetwork) IsRequester() bool { return n.Requester != nil } +func (a Agreement) IsRequester() bool { return a.Requester != nil } // SourcePacks is what we take from this network, empty if we read nothing. -func (n AdmittedNetwork) SourcePacks() []string { return n.Source.packs() } +func (a Agreement) SourcePacks() []string { return a.Source.packs() } // RequesterPacks is what this network may ask us for, empty if it may not ask. -func (n AdmittedNetwork) RequesterPacks() []string { return n.Requester.packs() } +func (a Agreement) RequesterPacks() []string { return a.Requester.packs() } // Directions names what we agreed, for a log line. Empty means nothing. -func (n AdmittedNetwork) Directions() []string { +func (a Agreement) Directions() []string { var agreed []string - if n.IsSource() { + if a.IsSource() { agreed = append(agreed, "source") } - if n.IsRequester() { + if a.IsRequester() { agreed = append(agreed, "requester") } return agreed } +// SourceContexts is what we take, as the context documents a discover names. +func (a Agreement) SourceContexts() []string { return contextsOf(a.SourcePacks()) } + +// RequesterContexts is what this network may ask us for, named the same way. +func (a Agreement) RequesterContexts() []string { return contextsOf(a.RequesterPacks()) } + +func contextsOf(packs []string) []string { + contexts := make([]string, 0, len(packs)) + for _, pack := range packs { + if trimmed := strings.TrimSpace(pack); trimmed != "" { + contexts = append(contexts, contextOf(trimmed)) + } + } + return contexts +} + +// contextOf is the context document beside a pack. A pack URL names the +// attributes document, but an agreement recorded with the context document +// already is taken as it stands rather than having a second one appended. +func contextOf(pack string) string { + if strings.HasSuffix(pack, contextFile) { + return pack + } + base := strings.TrimSuffix(pack, packFile) + if !strings.HasSuffix(base, "/") { + base += "/" + } + return base + contextFile +} + func (d *FederationDirection) packs() []string { if d == nil { return nil diff --git a/pkg/model/admittednetwork_test.go b/pkg/model/admittednetwork_test.go index 6b6b7002e..383266c97 100644 --- a/pkg/model/admittednetwork_test.go +++ b/pkg/model/admittednetwork_test.go @@ -1,6 +1,9 @@ package model -import "testing" +import ( + "slices" + "testing" +) func TestADirectionIsAgreedWhenItIsPresent(t *testing.T) { t.Parallel() @@ -12,9 +15,9 @@ func TestADirectionIsAgreedWhenItIsPresent(t *testing.T) { isSource bool isRequester bool }{ - "source only": {AdmittedNetwork{Source: packs}, true, false}, - "requester only": {AdmittedNetwork{Requester: packs}, false, true}, - "both": {AdmittedNetwork{Source: packs, Requester: packs}, true, true}, + "source only": {AdmittedNetwork{Agreement: Agreement{Source: packs}}, true, false}, + "requester only": {AdmittedNetwork{Agreement: Agreement{Requester: packs}}, false, true}, + "both": {AdmittedNetwork{Agreement: Agreement{Source: packs, Requester: packs}}, true, true}, "neither": {AdmittedNetwork{}, false, false}, } { t.Run(name, func(t *testing.T) { @@ -31,10 +34,10 @@ func TestADirectionIsAgreedWhenItIsPresent(t *testing.T) { func TestThePacksWeTakeComeFromTheSourceDirection(t *testing.T) { t.Parallel() - network := AdmittedNetwork{ + network := AdmittedNetwork{Agreement: Agreement{ Source: &FederationDirection{SchemaPacks: []string{"mandi"}}, Requester: &FederationDirection{SchemaPacks: []string{"weather"}}, - } + }} if got := network.SourcePacks(); len(got) != 1 || got[0] != "mandi" { t.Errorf("SourcePacks() = %v, want what we take, not what we give", got) @@ -47,9 +50,90 @@ func TestThePacksWeTakeComeFromTheSourceDirection(t *testing.T) { func TestAnUnagreedDirectionHasNoPacks(t *testing.T) { t.Parallel() - network := AdmittedNetwork{Source: &FederationDirection{SchemaPacks: []string{"mandi"}}} + network := AdmittedNetwork{Agreement: Agreement{Source: &FederationDirection{SchemaPacks: []string{"mandi"}}}} if got := network.RequesterPacks(); got != nil { t.Errorf("RequesterPacks() = %v, want none -- they may not read from us at all", got) } } + +func TestAnAgreementReadsAsTheContextsADiscoverNames(t *testing.T) { + t.Parallel() + + a := Agreement{ + Source: &FederationDirection{SchemaPacks: []string{"https://x/MandiPrice/v0.1/attributes.yaml"}}, + Requester: &FederationDirection{SchemaPacks: []string{"https://x/Weather/v0.1/attributes.yaml", " "}}, + } + + if got := a.SourceContexts(); len(got) != 1 || got[0] != "https://x/MandiPrice/v0.1/context.jsonld" { + t.Errorf("SourceContexts() = %v, want the context document beside the pack", got) + } + if got := a.RequesterContexts(); len(got) != 1 || got[0] != "https://x/Weather/v0.1/context.jsonld" { + t.Errorf("RequesterContexts() = %v, want the blank entry dropped", got) + } +} + +func TestAnUnagreedDirectionNamesNoContexts(t *testing.T) { + t.Parallel() + + if got := (Agreement{}).RequesterContexts(); len(got) != 0 { + t.Errorf("RequesterContexts() = %v, want none", got) + } +} + +func TestOnlyAnAdmittedNetworkHasAnAgreement(t *testing.T) { + t.Parallel() + + packs := &FederationDirection{SchemaPacks: []string{"mandi"}} + + for name, tc := range map[string]struct { + agreement Agreement + agreed bool + directions []string + }{ + "source only": {Agreement{Source: packs}, true, []string{"source"}}, + "requester only": {Agreement{Requester: packs}, true, []string{"requester"}}, + "both": {Agreement{Source: packs, Requester: packs}, true, []string{"source", "requester"}}, + "a provider": {Agreement{}, false, nil}, + } { + t.Run(name, func(t *testing.T) { + if got := tc.agreement.Agreed(); got != tc.agreed { + t.Errorf("Agreed() = %v, want %v", got, tc.agreed) + } + if got := tc.agreement.Directions(); !slices.Equal(got, tc.directions) { + t.Errorf("Directions() = %v, want %v", got, tc.directions) + } + }) + } +} + +func TestAPackAlreadyNamingItsContextIsLeftAlone(t *testing.T) { + t.Parallel() + + for name, tc := range map[string]struct{ pack, want string }{ + "attributes": { + "https://x/MandiPrice/v0.1/attributes.yaml", + "https://x/MandiPrice/v0.1/context.jsonld", + }, + "already the context": { + "https://x/MandiPrice/v0.1/context.jsonld", + "https://x/MandiPrice/v0.1/context.jsonld", + }, + "trailing slash": { + "https://x/MandiPrice/v0.1/", + "https://x/MandiPrice/v0.1/context.jsonld", + }, + "no filename": { + "https://x/MandiPrice/v0.1", + "https://x/MandiPrice/v0.1/context.jsonld", + }, + } { + t.Run(name, func(t *testing.T) { + a := Agreement{Source: &FederationDirection{SchemaPacks: []string{tc.pack}}} + got := a.SourceContexts() + if len(got) != 1 || got[0] != tc.want { + t.Errorf("SourceContexts(%q) = %v, want [%s]", tc.pack, got, tc.want) + } + }) + } +} diff --git a/pkg/model/model.go b/pkg/model/model.go index f45208ab6..faa279f32 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -33,6 +33,10 @@ type Subscription struct { Updated time.Time `json:"updated,omitzero" format:"date-time"` Nonce string `json:"nonce,omitzero"` NetworkMemberships []string `json:"network_memberships,omitempty"` + + // Agreement is set only on a network we admitted. It says which way round we + // agreed and what the agreement covers; every other participant has none. + Agreement Agreement `json:"agreement,omitzero"` } // nonUsableKeySubscriptionStatuses are the Subscription.Status values that mean a @@ -390,10 +394,14 @@ type Keyset struct { // StepContext holds context information for a request processing step. type StepContext struct { context.Context - Request *http.Request - Body []byte - Route *Route - SubID string + Request *http.Request + Body []byte + Route *Route + SubID string + + // OwnSubscriberID is who THIS module is, from its own config. SubID comes + // from the body, so on a cross-network request SubID is THEM. + OwnSubscriberID string Role Role RespHeader http.Header ProtocolVersion string // Protocol version parsed from context.version (e.g. "2.0.0") diff --git a/pkg/plugin/implementation/AgricultureFacility/live_test.go b/pkg/plugin/implementation/AgricultureFacility/live_test.go index 90534d7e7..788e00ead 100644 --- a/pkg/plugin/implementation/AgricultureFacility/live_test.go +++ b/pkg/plugin/implementation/AgricultureFacility/live_test.go @@ -208,7 +208,7 @@ func TestLiveAgainstPocra(t *testing.T) { for _, facilityType := range governedTypes { t.Run(facilityType, func(t *testing.T) { captured = nil - stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(liveRequest(t, facilityType))} + stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(liveRequest(t, facilityType))} started := time.Now() if err := step.Run(stepCtx); err != nil { t.Fatalf("Run() returned an unexpected error: %v", err) @@ -360,7 +360,7 @@ func TestLiveCategoryLeakIsFiltered(t *testing.T) { } captured = nil - stepCtx := &model.StepContext{Context: t.Context(), Body: body} + stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: body} if err := step.Run(stepCtx); err != nil { t.Fatalf("%s: Run() returned an unexpected error: %v", facilityType, err) } diff --git a/pkg/plugin/implementation/AgricultureFacility/mappings_test.go b/pkg/plugin/implementation/AgricultureFacility/mappings_test.go index d8d8526d7..1ca85bd32 100644 --- a/pkg/plugin/implementation/AgricultureFacility/mappings_test.go +++ b/pkg/plugin/implementation/AgricultureFacility/mappings_test.go @@ -78,7 +78,7 @@ const declaredContext = "https://raw.githubusercontent.com/OpenAgriNet/network-s const callerMessageID = "a1b2c3d4-e5f6-4789-abcd-ef1234567890" const selectRequest = `{ - "context": { "version": "2.0.0", "action": "select", + "context": { "version": "2.0.0", "action": "select", "receiverId": "oan.test", "networkId": "oan-dev", "transactionId": "f9e8d7c6-b5a4-4321-9876-543210fedcba", "messageId": "a1b2c3d4-e5f6-4789-abcd-ef1234567890", @@ -268,7 +268,7 @@ func runSelect(t *testing.T, request string) (sent map[string]any, answer map[st } defer closeStep() - stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(request)} + stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(request)} runErr = step.Run(stepCtx) if runErr != nil { return sent, nil, runErr @@ -922,7 +922,7 @@ func runAgainst(t *testing.T, request, providerBody string) map[string]any { } defer closeStep() - stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(request)} + stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(request)} if err := step.Run(stepCtx); err != nil { t.Fatalf("Run() returned an unexpected error: %v", err) } @@ -1405,7 +1405,7 @@ func runSplitSearch(t *testing.T, types []string, byCode map[string]string) (map t.Fatalf("could not rebuild the payload: %v", err) } - stepCtx := &model.StepContext{Context: t.Context(), Body: body} + stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: body} if err := step.Run(stepCtx); err != nil { t.Fatalf("Run() returned an unexpected error: %v", err) } diff --git a/pkg/plugin/implementation/AgricultureFacility/search_test.go b/pkg/plugin/implementation/AgricultureFacility/search_test.go index 33bcb9be1..551953235 100644 --- a/pkg/plugin/implementation/AgricultureFacility/search_test.go +++ b/pkg/plugin/implementation/AgricultureFacility/search_test.go @@ -77,7 +77,7 @@ func runSearchTimed(t *testing.T, types []string, cfg *AgricultureFacility.Confi t.Fatalf("could not rebuild the payload: %v", err) } - return step.Run(&model.StepContext{Context: t.Context(), Body: body}) + return step.Run(&model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: body}) } // A search's calls are sequential unless the deployment raised the limit. diff --git a/pkg/plugin/implementation/KnowledgeAdvisory/live_test.go b/pkg/plugin/implementation/KnowledgeAdvisory/live_test.go index 70bf2f259..4837b635e 100644 --- a/pkg/plugin/implementation/KnowledgeAdvisory/live_test.go +++ b/pkg/plugin/implementation/KnowledgeAdvisory/live_test.go @@ -120,7 +120,7 @@ func TestLiveEndToEnd(t *testing.T) { // Two requests: the second must reuse the token rather than exchange again. for i := 1; i <= 2; i++ { - ctx := &model.StepContext{Context: t.Context(), Body: []byte(selectRequest)} + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(selectRequest)} if err := step.Run(ctx); err != nil { t.Fatalf("request %d: %v", i, err) } diff --git a/pkg/plugin/implementation/KnowledgeAdvisory/mappings_test.go b/pkg/plugin/implementation/KnowledgeAdvisory/mappings_test.go index 0637ca7c6..d89d7b104 100644 --- a/pkg/plugin/implementation/KnowledgeAdvisory/mappings_test.go +++ b/pkg/plugin/implementation/KnowledgeAdvisory/mappings_test.go @@ -43,7 +43,7 @@ const shippedBindingKey = "knowledge-provider|" + shippedCapability // topics to search for and the languages it can read, and no advisory of its // own -- the pack's OnDemand branch requires exactly those two. const selectRequest = `{ - "context": { "version": "2.0.0", "action": "select", + "context": { "version": "2.0.0", "action": "select", "receiverId": "oan.test", "networkId": "oan-dev", "transactionId": "9f2c1a8e-4b70-4d31-9c55-6f2e0b1d7a44", "messageId": "7d41b9e0-52a6-4c18-8b73-1e9f0a4c6d22", @@ -226,7 +226,7 @@ func runShippedWith(t *testing.T, request, providerBody string) (map[string]any, } defer closeStep() - stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(request)} + stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(request)} if err := step.Run(stepCtx); err != nil { t.Fatalf("Run() returned an unexpected error: %v", err) } diff --git a/pkg/plugin/implementation/MandiPrice/mappings_test.go b/pkg/plugin/implementation/MandiPrice/mappings_test.go index c60299513..dc902b48a 100644 --- a/pkg/plugin/implementation/MandiPrice/mappings_test.go +++ b/pkg/plugin/implementation/MandiPrice/mappings_test.go @@ -49,7 +49,7 @@ const shippedBindingKey = "agmarknet|" + shippedCapability const selectRequest = `{ "context": { "version": "2.0.0", - "action": "select", + "action": "select", "receiverId": "oan.test", "networkId": "oan-dev", "transactionId": "9f2c1a8e-4b70-4d31-9c55-6f2e0b1d7a44", "messageId": "7d41b9e0-52a6-4c18-8b73-1e9f0a4c6d22", @@ -207,7 +207,7 @@ func runShippedWith(t *testing.T, request, providerBody string) (url.Values, map } defer closeStep() - stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(request)} + stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(request)} if err := step.Run(stepCtx); err != nil { t.Fatalf("Run() returned an unexpected error: %v", err) } @@ -468,7 +468,7 @@ func TestShippedMappingRefusesWhatItCannotServe(t *testing.T) { } defer closeStep() - stepCtx := &model.StepContext{Context: t.Context(), Body: body} + stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: body} if err := step.Run(stepCtx); err == nil { t.Fatal("expected an unserviceable payload to be refused") } else if !strings.Contains(err.Error(), tc.expect) { @@ -784,7 +784,7 @@ func TestShippedMappingRefusesPayloadsItCannotAnswer(t *testing.T) { } defer closeStep() - stepCtx := &model.StepContext{Context: t.Context(), Body: body} + stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: body} if err := step.Run(stepCtx); err == nil { t.Fatal("expected a payload this upstream cannot answer to be refused") } else if !strings.Contains(err.Error(), tc.expect) { diff --git a/pkg/plugin/implementation/WeatherObservation/mappings_test.go b/pkg/plugin/implementation/WeatherObservation/mappings_test.go index ac1d5d8d5..13b7de9e2 100644 --- a/pkg/plugin/implementation/WeatherObservation/mappings_test.go +++ b/pkg/plugin/implementation/WeatherObservation/mappings_test.go @@ -47,7 +47,7 @@ const shippedMapping = "weather-observation.select.yaml" // selectRequest is the verbatim /select captured from the network. const selectRequest = `{ - "context": { "version": "2.0.0", "action": "select", + "context": { "version": "2.0.0", "action": "select", "receiverId": "oan.test", "networkId": "da.gov.in/vistaar", "bapId": "seeker-network-vistaar.da.gov.in", "bapUri": "https://seeker-network-vistaar.da.gov.in/beckn", @@ -159,7 +159,7 @@ func TestShippedMappingsServeARealSelect(t *testing.T) { } defer closeStep() - stepCtx := &model.StepContext{Context: t.Context(), Body: []byte(selectRequest)} + stepCtx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "provider-network-vistaar.da.gov.in", Body: []byte(selectRequest)} if err := step.Run(stepCtx); err != nil { t.Fatalf("Run() returned an unexpected error: %v", err) } diff --git a/pkg/plugin/implementation/catalogcrawler/discover.go b/pkg/plugin/implementation/catalogcrawler/discover.go index 0d7a5d9ce..ba74f844a 100644 --- a/pkg/plugin/implementation/catalogcrawler/discover.go +++ b/pkg/plugin/implementation/catalogcrawler/discover.go @@ -11,7 +11,6 @@ import ( "log/slog" "net/http" "net/url" - "path" "strings" "time" @@ -23,9 +22,6 @@ import ( const ( protocolVersion = "2.0.0" discoverAction = "discover" - - packFile = "attributes.yaml" - contextFile = "context.jsonld" ) // discoverBody builds the discover we send to one network. We name both ends; the @@ -42,7 +38,7 @@ func discoverBody(network model.AdmittedNetwork, ourNetworkID, intent string, "transactionId": uuid.NewString(), "timestamp": time.Now().UTC().Format(time.RFC3339), } - if contexts := schemaContexts(network.SourcePacks()); len(contexts) > 0 { + if contexts := network.SourceContexts(); len(contexts) > 0 { beckn["schemaContext"] = contexts } @@ -75,26 +71,14 @@ func parseIntent(intent string) (map[string]any, error) { return asked, nil } -// schemaContexts turns the packs we take into the context documents a discover -// filters on. Discovery matches any of them, so three asked for is three back. -func schemaContexts(packs []string) []string { - contexts := make([]string, 0, len(packs)) - for _, pack := range packs { - if trimmed := strings.TrimSpace(pack); trimmed != "" { - contexts = append(contexts, strings.TrimSuffix(trimmed, packFile)+contextFile) - } - } - return contexts -} - -// discoverURL appends the action to the network's base. A network publishes the base -// its network module is mounted at; the adapter takes the action from the URL. +// discoverURL is the address a network published, called as it stands. +// services.discovery names the discover endpoint itself, not a mount to append +// to, so a network is free to serve it from any path. func discoverURL(base string) (string, error) { parsed, err := url.Parse(strings.TrimSpace(base)) if err != nil || parsed.Scheme == "" || parsed.Host == "" { return "", fmt.Errorf("catalogcrawler: %q is not an address a discover can be sent to", base) } - parsed.Path = path.Join(parsed.Path, discoverAction) return parsed.String(), nil } diff --git a/pkg/plugin/implementation/catalogcrawler/discover_test.go b/pkg/plugin/implementation/catalogcrawler/discover_test.go index b1137ea31..3527c4f08 100644 --- a/pkg/plugin/implementation/catalogcrawler/discover_test.go +++ b/pkg/plugin/implementation/catalogcrawler/discover_test.go @@ -24,7 +24,7 @@ func targetNetwork() model.AdmittedNetwork { return model.AdmittedNetwork{ ID: "mahavistara.oan.local", DiscoveryURL: "https://mahavistara.oan.local/network", - Source: &model.FederationDirection{SchemaPacks: []string{weatherPack}}, + Agreement: model.Agreement{Source: &model.FederationDirection{SchemaPacks: []string{weatherPack}}}, } } @@ -133,14 +133,14 @@ func TestDiscoverBodyRefusesAnUnreadableIntent(t *testing.T) { } } -func TestDiscoverURLAppendsTheAction(t *testing.T) { +func TestDiscoverURLIsTheAddressAsPublished(t *testing.T) { t.Parallel() for name, tc := range map[string]struct{ base, want string }{ - "plain": {"https://network.oan.local/network", "https://network.oan.local/network/discover"}, - "trailing": {"https://network.oan.local/network/", "https://network.oan.local/network/discover"}, - "padded": {" https://network.oan.local/network ", "https://network.oan.local/network/discover"}, - "root": {"https://network.oan.local", "https://network.oan.local/discover"}, + "as published": {"https://network.oan.local/federation/discovery", "https://network.oan.local/federation/discovery"}, + "our mount": {"https://network.oan.local/network/discover", "https://network.oan.local/network/discover"}, + "padded": {" https://network.oan.local/network/discover ", "https://network.oan.local/network/discover"}, + "root": {"https://network.oan.local", "https://network.oan.local"}, } { t.Run(name, func(t *testing.T) { got, err := discoverURL(tc.base) @@ -256,22 +256,24 @@ func TestNetworkCallSignsAsUs(t *testing.T) { } } -func TestNetworkCallSendsToTheActionEndpoint(t *testing.T) { +func TestNetworkCallSendsToTheAddressPublished(t *testing.T) { t.Parallel() + // Any path: services.discovery names the endpoint, so a network may serve + // it from somewhere that looks nothing like our own mount. var path string caller, url := testCaller(t, func(w http.ResponseWriter, r *http.Request) { path = r.URL.Path answerCatalogs(0)(w, r) }) network := targetNetwork() - network.DiscoveryURL = url + "/network" + network.DiscoveryURL = url + "/federation/discovery" if _, err := collectDiscover(caller, network); err != nil { t.Fatalf("discover() returned an unexpected error: %v", err) } - if path != "/network/discover" { - t.Errorf("called %q, want /network/discover", path) + if path != "/federation/discovery" { + t.Errorf("called %q, want /federation/discovery", path) } } diff --git a/pkg/plugin/implementation/internal/common/common.go b/pkg/plugin/implementation/internal/common/common.go index 9390a9373..d8ffd132d 100644 --- a/pkg/plugin/implementation/internal/common/common.go +++ b/pkg/plugin/implementation/internal/common/common.go @@ -22,6 +22,7 @@ import ( "strings" "github.com/beckn-one/beckn-onix/pkg/log" + "github.com/beckn-one/beckn-onix/pkg/model" "github.com/beckn-one/beckn-onix/pkg/plugin/definition" "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/common/util" ) @@ -82,6 +83,7 @@ type Step struct { paths Paths prerequisites Prerequisites registry definition.ProviderRecordLookup + networks admittedLookup mapper definition.Mapper httpClient *http.Client @@ -90,6 +92,12 @@ type Step struct { auth map[string]*authenticator } +// admittedLookup lists the networks this one admitted. Narrowed from the +// registry, so a plugin needs no dependency only the federated path uses. +type admittedLookup interface { + AdmittedNetworks(ctx context.Context) ([]model.AdmittedNetwork, error) +} + // New creates the step. func New(ctx context.Context, registry definition.ProviderRecordLookup, mapper definition.Mapper, prerequisites Prerequisites, cfg *Config) (*Step, func() error, error) { @@ -121,6 +129,12 @@ func New(ctx context.Context, registry definition.ProviderRecordLookup, mapper d httpClient: &http.Client{CheckRedirect: util.RefuseOffHostRedirect}, auth: make(map[string]*authenticator, len(cfg.AuthByProvider)), } + // Absent is a deployment that federates with nobody. Only a request + // addressed elsewhere needs it, and that path refuses rather than guessing. + if networks, ok := registry.(admittedLookup); ok { + step.networks = networks + } + for provider, profile := range cfg.AuthByProvider { step.auth[provider] = &authenticator{cfg: *profile} } diff --git a/pkg/plugin/implementation/internal/common/common_test.go b/pkg/plugin/implementation/internal/common/common_test.go index 2a18fbfe6..47a69c474 100644 --- a/pkg/plugin/implementation/internal/common/common_test.go +++ b/pkg/plugin/implementation/internal/common/common_test.go @@ -21,8 +21,12 @@ import ( "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/common/util" ) +// testSubscriberID is who this deployment is; a request names it as receiver. +const testSubscriberID = "oan.test" + const selectBody = `{ - "context": { "version": "2.0.0", "action": "select", "transactionId": "txn-1" }, + "context": { "version": "2.0.0", "action": "select", "transactionId": "txn-1", + "receiverId": "oan.test" }, "message": { "contract": { "commitments": [{ "resources": [{ "resourceAttributes": { "@type": "openagrinet:WeatherObservation", @@ -161,7 +165,7 @@ func newStep(t *testing.T, registry definition.ProviderRecordLookup, mapper defi func runStep(t *testing.T, step *Step, body string) (*model.StepContext, error) { t.Helper() - ctx := &model.StepContext{Context: t.Context(), Body: []byte(body)} + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: "oan.test", Body: []byte(body)} return ctx, step.Run(ctx) } @@ -231,7 +235,7 @@ func TestNewValidatesTheAuthScheme(t *testing.T) { func TestRunReportsAnUnservablePayloadAsABadRequest(t *testing.T) { t.Parallel() - twoCommitments := `{"context":{"action":"select"},"message":{"contract":{"commitments":[ + twoCommitments := `{"context":{"action":"select","receiverId":"oan.test"},"message":{"contract":{"commitments":[ {"offer":{"provider":{"id":"mausamgram"}}, "resources":[{"resourceAttributes":{"@type":"openagrinet:WeatherObservation"}}]}, {"offer":{"provider":{"id":"mausamgram"}}, @@ -449,7 +453,7 @@ func TestRunReportsACancelDuringBackoffAsTheContextsError(t *testing.T) { mapper := &stubMapper{requestResult: []byte(`{}`), responseResult: []byte(`{}`)} step := newStep(t, &stubRegistry{plan: plan}, mapper) - err := step.Run(&model.StepContext{Context: ctx, Body: []byte(selectBody)}) + err := step.Run(&model.StepContext{Context: ctx, OwnSubscriberID: "oan.test", Body: []byte(selectBody)}) if !errors.Is(err, context.Canceled) { t.Fatalf("error %v does not carry context.Canceled", err) } @@ -483,7 +487,7 @@ func TestRunKeepsTheProvidersAnswerWhenTheCallerLeavesDuringBackoff(t *testing.T mapper := &stubMapper{requestResult: []byte(`{}`), responseResult: []byte(`{}`)} step := newStep(t, &stubRegistry{plan: plan}, mapper) - err := step.Run(&model.StepContext{Context: ctx, Body: []byte(selectBody)}) + err := step.Run(&model.StepContext{Context: ctx, OwnSubscriberID: "oan.test", Body: []byte(selectBody)}) if !errors.Is(err, context.Canceled) { t.Fatalf("error %v does not carry context.Canceled", err) } @@ -516,7 +520,7 @@ func TestRunReportsAPermanentAnswerThatArrivesAsTheCallerLeaves(t *testing.T) { mapper := &stubMapper{requestResult: []byte(`{}`), responseResult: []byte(`{}`)} step := newStep(t, &stubRegistry{plan: plan}, mapper) - err := step.Run(&model.StepContext{Context: ctx, Body: []byte(selectBody)}) + err := step.Run(&model.StepContext{Context: ctx, OwnSubscriberID: "oan.test", Body: []byte(selectBody)}) var coded *model.CodedErr if !errors.As(err, &coded) || coded.HTTPStatus() != http.StatusBadGateway { t.Errorf("status for a 400 answer = %v, want 502", err) @@ -559,7 +563,7 @@ func TestRunDoesNotCountAnAttemptThatNeverRan(t *testing.T) { mapper := &stubMapper{requestResult: []byte(`{}`), responseResult: []byte(`{}`)} step := newStep(t, &stubRegistry{plan: plan}, mapper) - err := step.Run(&model.StepContext{Context: ctx, Body: []byte(selectBody)}) + err := step.Run(&model.StepContext{Context: ctx, OwnSubscriberID: "oan.test", Body: []byte(selectBody)}) if err == nil { t.Fatal("expected the ended request to be reported") } @@ -588,7 +592,7 @@ func TestRunStopsWhenTheCallerHasGone(t *testing.T) { cancelled, cancel := context.WithCancel(context.Background()) cancel() - stepCtx := &model.StepContext{Context: cancelled, Body: []byte(selectBody)} + stepCtx := &model.StepContext{Context: cancelled, OwnSubscriberID: "oan.test", Body: []byte(selectBody)} if err := step.Run(stepCtx); err == nil { t.Fatal("expected a cancelled request to be reported") } @@ -944,7 +948,7 @@ func TestRunPassesThroughWhatIsNotItsCapability(t *testing.T) { testCases := []struct{ name, body string }{ {"another provider", strings.Replace(selectBody, `"id": "mausamgram"`, `"id": "agmarknet"`, 1)}, {"another capability", strings.Replace(selectBody, "openagrinet:WeatherObservation", "openagrinet:MarketPrice", 1)}, - {"a payload with no binding at all", `{"context":{"action":"select"},"message":{}}`}, + {"a payload with no binding at all", `{"context":{"action":"select","receiverId":"oan.test"},"message":{}}`}, } for _, tc := range testCases { @@ -1312,7 +1316,7 @@ func TestRunReadsTheBindingKeyFromOverriddenPaths(t *testing.T) { c.CapabilityCodeAt = "what[].type" }) - ctx, err := runStep(t, step, `{"context":{"action":"select"},"who":{"provider":"agmarknet"},"what":[{"type":"openagrinet:MandiPrice"}]}`) + ctx, err := runStep(t, step, `{"context":{"action":"select","receiverId":"oan.test"},"who":{"provider":"agmarknet"},"what":[{"type":"openagrinet:MandiPrice"}]}`) if err != nil { t.Fatalf("Run() returned an unexpected error: %v", err) } @@ -1457,7 +1461,8 @@ func selectWithLocation(t *testing.T, geometry string) string { location = `"location": ` + geometry + `,` } return `{ - "context": { "version": "2.0.0", "action": "select", "transactionId": "txn-1" }, + "context": { "version": "2.0.0", "action": "select", "transactionId": "txn-1", + "receiverId": "oan.test" }, "message": { "contract": { "commitments": [ { "resources": [ { "id": "res:x", "resourceAttributes": { ` + location + ` diff --git a/pkg/plugin/implementation/internal/common/crossnetwork_test.go b/pkg/plugin/implementation/internal/common/crossnetwork_test.go new file mode 100644 index 000000000..033e154f8 --- /dev/null +++ b/pkg/plugin/implementation/internal/common/crossnetwork_test.go @@ -0,0 +1,277 @@ +package common_test + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/beckn-one/beckn-onix/pkg/model" + "github.com/beckn-one/beckn-onix/pkg/plugin/definition" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/common" + "github.com/beckn-one/beckn-onix/pkg/plugin/implementation/internal/common/util" +) + +const ( + us = "bharatvistar.oan.local" + them = "mahavistara.oan.local" + + // Both networks front the same upstream, so they hold the SAME binding key. + sharedBinding = "agmarknet.bharatvistar|openagrinet:MandiPrice" +) + +// crossRequest names a receiver, an address and an action -- the three fields +// the decision rests on. The capability half stays constant. +func crossRequest(receiver, bppURI, action string) []byte { + field := func(name, value string) string { + if value == "" { + return "" + } + return fmt.Sprintf(`, %q: %q`, name, value) + } + return []byte(fmt.Sprintf(`{ + "context": { "version": "2.0.0"%s%s%s }, + "message": { "contract": { "commitments": [ { + "resources": [ { "resourceAttributes": { "@type": "openagrinet:MandiPrice" } } ], + "offer": { "provider": { "id": "agmarknet.bharatvistar" } } + } ] } } +}`, field("action", action), field("receiverId", receiver), field("bppUri", bppURI))) +} + +// networkStub answers both lookups the step makes: the call plan, and the +// networks we admitted. One registry client serves both in the adapter too. +type networkStub struct { + plan *model.ProviderRecord + networks []model.AdmittedNetwork + err error +} + +func (s *networkStub) ProviderRecord(context.Context, string) (*model.ProviderRecord, error) { + return s.plan, nil +} + +func (s *networkStub) AdmittedNetworks(context.Context) ([]model.AdmittedNetwork, error) { + return s.networks, s.err +} + +// agreedPacks is what a direction covers. Any pack will do here; these tests +// are about the direction, not the subset. +func agreedPacks() *model.FederationDirection { + return &model.FederationDirection{SchemaPacks: []string{"https://x/MandiPrice/v0.1/attributes.yaml"}} +} + +// admittedAs is the step with one network on the books in the given direction. +func admittedAs(t *testing.T, agreed model.Agreement) definition.Step { + t.Helper() + return crossStepWith(t, "http://unused.invalid", &networkStub{ + plan: &model.ProviderRecord{BindingKey: sharedBinding}, + networks: []model.AdmittedNetwork{{ID: them, Agreement: agreed}}, + }) +} + +func crossStep(t *testing.T, upstreamURL string) definition.Step { + t.Helper() + plan := &model.ProviderRecord{ + BindingKey: sharedBinding, + BaseURL: upstreamURL, + Actions: map[string]model.ActionPlan{ + "select": {Method: http.MethodGet, Path: "/x", Mappings: dispatchMappingRef, RetryMax: 1}, + }, + } + return crossStepWith(t, upstreamURL, &networkStub{ + plan: plan, + networks: []model.AdmittedNetwork{ + {ID: them, Agreement: model.Agreement{Source: agreedPacks(), Requester: agreedPacks()}}, + }, + }) +} + +func crossStepWith(t *testing.T, upstreamURL string, registry definition.ProviderRecordLookup) definition.Step { + t.Helper() + step, closer, err := common.New(context.Background(), + registry, + fixedMapper{answer: `{"served":"locally"}`}, + nil, + &common.Config{ + BindingKeys: []string{sharedBinding}, + AuthByProvider: map[string]*common.AuthProfile{ + "agmarknet.bharatvistar": {Scheme: util.AuthSchemeNone}, + }, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = closer() }) + return step +} + +func TestRequestForUsIsServedHere(t *testing.T) { + var called bool + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + called = true + fmt.Fprint(w, `{"from":"upstream"}`) + })) + defer upstream.Close() + + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: us, Body: crossRequest(us, "https://"+them+"/provider", "select")} + if err := crossStep(t, upstream.URL).Run(ctx); err != nil { + t.Fatal(err) + } + if !called { + t.Error("a request addressed to us did not reach our upstream") + } + if ctx.Route != nil { + t.Errorf("Route = %#v, want nil -- our own request was sent away", ctx.Route) + } +} + +func TestRequestForAnotherNetworkIsSentThere(t *testing.T) { + var called bool + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + called = true + })) + defer upstream.Close() + + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: us, Body: crossRequest(them, "https://"+them+"/provider", "select")} + if err := crossStep(t, upstream.URL).Run(ctx); err != nil { + t.Fatal(err) + } + if called { + t.Error("another network's request was answered with our data") + } + if ctx.Route == nil { + t.Fatal("Route = nil, want the network's address") + } + if got := ctx.Route.URL.String(); got != "https://"+them+"/provider/select" { + t.Errorf("Route.URL = %q, want the network's address with the action on it", got) + } + if ctx.SubID != us { + t.Errorf("SubID = %q, want %q -- the forwarded call must be signed as us", ctx.SubID, us) + } +} + +func TestEveryActionCrossesTheSameWay(t *testing.T) { + for _, action := range []string{"select", "init", "confirm", "status"} { + t.Run(action, func(t *testing.T) { + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: us, Body: crossRequest(them, "https://"+them+"/provider", action)} + if err := crossStep(t, "http://unused.invalid").Run(ctx); err != nil { + t.Fatal(err) + } + if ctx.Route == nil { + t.Fatal("Route = nil, want the network's address") + } + if got := ctx.Route.URL.String(); got != "https://"+them+"/provider/"+action { + t.Errorf("Route.URL = %q, want the action appended", got) + } + }) + } +} + +func TestRequestNamingNoReceiverIsRefused(t *testing.T) { + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: us, Body: crossRequest("", "", "select")} + err := crossStep(t, "http://unused.invalid").Run(ctx) + if err == nil { + t.Fatal("a request naming no receiver was accepted; it must not be assumed ours") + } + if !strings.Contains(err.Error(), "receiverId") { + t.Errorf("error = %v, want it to name the field to set", err) + } +} + +func TestRequestForAnotherNetworkWithNoAddressIsRefused(t *testing.T) { + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: us, Body: crossRequest(them, "", "select")} + err := crossStep(t, "http://unused.invalid").Run(ctx) + if err == nil { + t.Fatal("a request for another network with no bppUri was accepted") + } + if !strings.Contains(err.Error(), "bppUri") { + t.Errorf("error = %v, want it to name the missing address", err) + } +} + +func TestStepWithoutAnIdentityRefusesToDecide(t *testing.T) { + ctx := &model.StepContext{Context: t.Context(), Body: crossRequest(them, "https://"+them+"/provider", "select")} + err := crossStep(t, "http://unused.invalid").Run(ctx) + if err == nil { + t.Fatal("a step with no subscriberId decided whose request this was") + } + if !strings.Contains(err.Error(), "subscriberId") { + t.Errorf("error = %v, want it to name the missing identity", err) + } +} + +func TestASecondStepLeavesAnAlreadyRoutedRequestAlone(t *testing.T) { + var called bool + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + called = true + })) + defer upstream.Close() + + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: us, Body: crossRequest(us, "https://"+them+"/provider", "select")} + ctx.Route = &model.Route{TargetType: "url"} + if err := crossStep(t, upstream.URL).Run(ctx); err != nil { + t.Fatal(err) + } + if called { + t.Error("a step called its upstream for a request already routed away") + } +} + +func TestForwardingNeedsTheReceiverAdmittedAsASource(t *testing.T) { + for _, tc := range []struct { + role string + agreed model.Agreement + forwarded bool + }{ + {"source", model.Agreement{Source: agreedPacks()}, true}, + {"both", model.Agreement{Source: agreedPacks(), Requester: agreedPacks()}, true}, + {"requester", model.Agreement{Requester: agreedPacks()}, false}, + } { + t.Run(tc.role, func(t *testing.T) { + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: us, + Body: crossRequest(them, "https://"+them+"/provider", "select")} + err := admittedAs(t, tc.agreed).Run(ctx) + if tc.forwarded { + if err != nil { + t.Fatalf("a %s network was not forwarded to: %v", tc.role, err) + } + if ctx.Route == nil { + t.Error("Route = nil, want the network's address") + } + return + } + if err == nil { + t.Fatalf("a %s-only network was forwarded to; we never agreed to read from them", tc.role) + } + if ctx.Route != nil { + t.Errorf("Route = %#v, want nil", ctx.Route) + } + }) + } +} + +func TestForwardingRefusesANetworkWeNeverAdmitted(t *testing.T) { + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: us, + Body: crossRequest("stranger.oan.local", "https://stranger.oan.local/provider", "select")} + err := crossStepWith(t, "http://unused.invalid", &networkStub{ + plan: &model.ProviderRecord{BindingKey: sharedBinding}}).Run(ctx) + if err == nil { + t.Fatal("a request for a network we never admitted was forwarded") + } + if !strings.Contains(err.Error(), "stranger.oan.local") { + t.Errorf("error = %v, want it to name the network", err) + } +} + +func TestForwardingRefusesWhenTheRegistryCannotBeAsked(t *testing.T) { + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: us, + Body: crossRequest(them, "https://"+them+"/provider", "select")} + err := crossStepWith(t, "http://unused.invalid", &networkStub{ + plan: &model.ProviderRecord{BindingKey: sharedBinding}, + err: fmt.Errorf("registry unreachable")}).Run(ctx) + if err == nil { + t.Fatal("a network was forwarded to without checking we admitted them") + } +} diff --git a/pkg/plugin/implementation/internal/common/dispatch_test.go b/pkg/plugin/implementation/internal/common/dispatch_test.go index 727288014..3dbb2052f 100644 --- a/pkg/plugin/implementation/internal/common/dispatch_test.go +++ b/pkg/plugin/implementation/internal/common/dispatch_test.go @@ -22,7 +22,7 @@ const dispatchMappingRef = "https://m.example.com/mausamgram/weather-observation // Kept minimal on purpose: this test is about which step claims a request, not // about what any of them would do with it. const dispatchRequest = `{ - "context": { "version": "2.0.0", "action": "select" }, + "context": { "version": "2.0.0", "action": "select", "receiverId": "bharatvistar.oan.local" }, "message": { "contract": { "commitments": [ { "resources": [ { "resourceAttributes": { "@type": "openagrinet:WeatherObservation" } } ], "offer": { "provider": { "id": "mausamgram" } } @@ -102,7 +102,7 @@ func TestTwoProviderStepsDispatchByBindingKey(t *testing.T) { // A request for the FIRST capability, run through both steps in order, as a // pipeline would. - ctx := &model.StepContext{Context: t.Context(), Body: []byte(dispatchRequest)} + ctx := &model.StepContext{Context: t.Context(), OwnSubscriberID: us, Body: []byte(dispatchRequest)} for i, step := range steps { if err := step.Run(ctx); err != nil { t.Fatalf("step %d: %v", i, err) diff --git a/pkg/plugin/implementation/internal/common/perprovider_test.go b/pkg/plugin/implementation/internal/common/perprovider_test.go index f9e14e0d8..4881d8891 100644 --- a/pkg/plugin/implementation/internal/common/perprovider_test.go +++ b/pkg/plugin/implementation/internal/common/perprovider_test.go @@ -18,7 +18,8 @@ import ( // two binding keys can be sent a request for either. func bodyFor(provider, capability string) string { return fmt.Sprintf(`{ - "context": { "version": "2.0.0", "action": "select", "transactionId": "txn-1" }, + "context": { "version": "2.0.0", "action": "select", "transactionId": "txn-1", + "receiverId": "oan.test" }, "message": { "contract": { "commitments": [{ "resources": [{ "resourceAttributes": { "@type": %q } }], "offer": { "provider": { "id": %q } } diff --git a/pkg/plugin/implementation/internal/common/serve.go b/pkg/plugin/implementation/internal/common/serve.go index 6d34eca71..e840dd736 100644 --- a/pkg/plugin/implementation/internal/common/serve.go +++ b/pkg/plugin/implementation/internal/common/serve.go @@ -7,6 +7,8 @@ import ( "encoding/json" "errors" "fmt" + "net/url" + "path" "slices" "strings" @@ -22,6 +24,18 @@ import ( // recognises its own work, so adding a provider is one config entry rather than // a routing-table change. func (s *Step) Run(ctx *model.StepContext) error { + // An earlier step already routed this. The binding would still match, so + // without this we would call the upstream for an answer about to be discarded. + if ctx.Route != nil { + return nil + } + + // Is this ours to answer at all? Asked BEFORE the binding -- see routeElsewhere. + routed, err := s.routeElsewhere(ctx) + if err != nil || routed { + return err + } + binding, err := BindingFrom(s.paths, ctx.Body) if errors.Is(err, errNoBinding) { return nil @@ -199,3 +213,135 @@ func decodeBody(body []byte) (any, error) { } return decoded, nil } + +// routeElsewhere sends a request addressed to another network to that network. +// Before the binding check: a binding key names the upstream, not the network. +func (s *Step) routeElsewhere(ctx *model.StepContext) (bool, error) { + becknContext, err := contextOf(ctx.Body) + if err != nil { + return false, err + } + + // Refused rather than assumed. Guessing "ours" is what answers another + // network's request with our data. + receiver := contextValue(becknContext, "bpp_id", "bppId", "receiverId") + if receiver == "" { + return false, model.NewBadReqErr("", fmt.Errorf( + "the request names no receiver, so there is nothing to decide by: "+ + "set context.receiverId to the participant that should answer")) + } + + if ctx.OwnSubscriberID == "" { + return false, fmt.Errorf( + "the request is addressed to %s, but this module has no subscriberId to compare it against", + receiver) + } + if strings.EqualFold(receiver, ctx.OwnSubscriberID) { + return false, nil // ours: carry on into the binding check + } + + if err := s.agreedToReadFrom(ctx, receiver); err != nil { + return false, err + } + + target, err := destinationOf(becknContext, receiver, ctx.OwnSubscriberID) + if err != nil { + return false, err + } + + // The network signs, not the participant the request names. Without this + // sign would look up the other network's keyset and fail. + ctx.SubID = ctx.OwnSubscriberID + ctx.Route = &model.Route{TargetType: "url", URL: target} + + log.Infof(ctx, "%s is not %s; sending it to %s", receiver, ctx.OwnSubscriberID, target) + return true, nil +} + +// agreedToReadFrom refuses a receiver we did not admit as a source. Without it +// a field in the request decides where a signed call of ours goes. +func (s *Step) agreedToReadFrom(ctx *model.StepContext, receiver string) error { + if s.networks == nil { + return fmt.Errorf( + "the request is for %s, but this registry cannot list the networks we admitted", + receiver) + } + + admitted, err := s.networks.AdmittedNetworks(ctx) + if err != nil { + return fmt.Errorf("could not check whether %s is a network we admitted: %w", receiver, err) + } + + for _, network := range admitted { + if !strings.EqualFold(network.ID, receiver) { + continue + } + if !network.IsSource() { + return model.NewBadReqErr("", fmt.Errorf( + "we agreed %v with %s, which does not include reading from them", + network.Directions(), receiver)) + } + log.Infof(ctx, "%s is admitted, agreed %v; forwarding", receiver, network.Directions()) + return nil + } + + return model.NewNotFoundErr("", fmt.Errorf( + "%s is not a network we admitted, so this is not ours to send on", receiver)) +} + +// destinationOf builds the URL a request for another participant is sent to. +// The address is context.bppUri, so the caller names where we send it. +func destinationOf(becknContext map[string]any, receiver, us string) (*url.URL, error) { + raw := contextValue(becknContext, "bpp_uri", "bppUri", "receiverUri") + if raw == "" { + return nil, model.NewBadReqErr("", fmt.Errorf( + "the request is for %s rather than %s, but names no context.bppUri, "+ + "so there is no address to send it to", receiver, us)) + } + + target, err := url.Parse(strings.TrimSpace(raw)) + if err != nil || target.Host == "" { + return nil, model.NewBadReqErr("SCH_INVALID_FORMAT", fmt.Errorf( + "context.bppUri %q is not a URL this request can be sent to", raw)) + } + + // The proxy replaces the request URL outright, so a bare bppUri would + // arrive at the other network's mount with no action left on it. + action := contextValue(becknContext, "action") + if action == "" { + return nil, model.NewBadReqErr("", fmt.Errorf( + "the request is for %s rather than %s, but names no context.action, "+ + "so there is no endpoint to send it to", receiver, us)) + } + if target.Path == "" { + target.Path = "/" + } + target.Path = path.Join(target.Path, action) + return target, nil +} + +// contextOf reads the request's Beckn context. An unreadable body is reported +// as that, not as "names no receiver" -- this is the first step to see it. +func contextOf(body []byte) (map[string]any, error) { + var payload struct { + Context map[string]any `json:"context"` + } + if err := json.Unmarshal(body, &payload); err != nil { + return nil, model.NewBadReqErr("SCH_INVALID_FORMAT", + fmt.Errorf("payload could not be read: %w", err)) + } + return payload.Context, nil +} + +// contextValue reads the first of the named context fields carrying a value, +// snake_case then camelCase then Beckn v2, the order the adapter uses. +func contextValue(becknContext map[string]any, keys ...string) string { + for _, key := range keys { + if value, ok := becknContext[key].(string); ok { + if trimmed := strings.TrimSpace(value); trimmed != "" { + return trimmed + } + } + } + return "" +} diff --git a/pkg/plugin/implementation/sunbirdRegistry/admittednetworks.go b/pkg/plugin/implementation/sunbirdRegistry/admittednetworks.go index 77ac6f365..08c1fa022 100644 --- a/pkg/plugin/implementation/sunbirdRegistry/admittednetworks.go +++ b/pkg/plugin/implementation/sunbirdRegistry/admittednetworks.go @@ -107,8 +107,7 @@ func admittedFrom(record participant) (model.AdmittedNetwork, bool) { ID: record.ParticipantID, DiscoveryURL: strings.TrimSpace(record.BaseURL), ProjectionTtl: time.Duration(record.ProjectionTtl) * time.Second, - Source: record.FederationRole.Source.agreed(), - Requester: record.FederationRole.Requester.agreed(), + Agreement: record.FederationRole.agreed(), } usable := strings.EqualFold(record.Status, statusActive) && record.Role.has(roleNetwork) && diff --git a/pkg/plugin/implementation/sunbirdRegistry/cmd/plugin.go b/pkg/plugin/implementation/sunbirdRegistry/cmd/plugin.go index 5992bf4bd..7fd9d2ec7 100644 --- a/pkg/plugin/implementation/sunbirdRegistry/cmd/plugin.go +++ b/pkg/plugin/implementation/sunbirdRegistry/cmd/plugin.go @@ -128,6 +128,16 @@ func (o sunbirdRegistryProvider) parseConfig(config map[string]string) (*sunbird cfg.RetryWaitMax = retryWaitMax } + // Parse crossNetworkCheck. Absent means off, so a deployment that + // federates with nobody behaves exactly as before. + if checkStr, exists := config["crossNetworkCheck"]; exists && checkStr != "" { + check, err := strconv.ParseBool(checkStr) + if err != nil { + return nil, fmt.Errorf("invalid crossNetworkCheck value '%s': %w", checkStr, err) + } + cfg.CrossNetworkCheck = check + } + if cfg.RetryWaitMin > cfg.RetryWaitMax { return nil, fmt.Errorf("retry_wait_min (%v) must not exceed retry_wait_max (%v)", cfg.RetryWaitMin, cfg.RetryWaitMax) } diff --git a/pkg/plugin/implementation/sunbirdRegistry/cmd/plugin_test.go b/pkg/plugin/implementation/sunbirdRegistry/cmd/plugin_test.go index fb2dfec55..624801eff 100644 --- a/pkg/plugin/implementation/sunbirdRegistry/cmd/plugin_test.go +++ b/pkg/plugin/implementation/sunbirdRegistry/cmd/plugin_test.go @@ -247,3 +247,37 @@ func TestProviderSatisfiesTheInterface(t *testing.T) { var _ definition.RegistryLookupProvider = Provider } + +func TestParseConfigReadsCrossNetworkCheck(t *testing.T) { + provider := sunbirdRegistryProvider{} + + for _, tc := range []struct { + name string + value string + want bool + }{ + {"absent means off", "", false}, + {"true", "true", true}, + {"false", "false", false}, + } { + t.Run(tc.name, func(t *testing.T) { + config := map[string]string{"url": "http://registry.invalid/api/v1"} + if tc.value != "" { + config["crossNetworkCheck"] = tc.value + } + cfg, err := provider.parseConfig(config) + if err != nil { + t.Fatalf("parseConfig() returned an unexpected error: %v", err) + } + if cfg.CrossNetworkCheck != tc.want { + t.Errorf("CrossNetworkCheck = %v, want %v", cfg.CrossNetworkCheck, tc.want) + } + }) + } + + if _, err := provider.parseConfig(map[string]string{ + "url": "http://registry.invalid/api/v1", "crossNetworkCheck": "yes please", + }); err == nil { + t.Error("an unparseable crossNetworkCheck was accepted") + } +} diff --git a/pkg/plugin/implementation/sunbirdRegistry/crossnetwork_test.go b/pkg/plugin/implementation/sunbirdRegistry/crossnetwork_test.go new file mode 100644 index 000000000..29e6e3eed --- /dev/null +++ b/pkg/plugin/implementation/sunbirdRegistry/crossnetwork_test.go @@ -0,0 +1,279 @@ +package sunbirdRegistry + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/beckn-one/beckn-onix/pkg/model" +) + +const testAgreedPack = "https://x/MandiPrice/v0.1/attributes.yaml" +const testAgreedContext = "https://x/MandiPrice/v0.1/context.jsonld" + +// admittedRecord is a network we admitted, in the given directions. A pack list +// is required on a real record, so each direction carries one. +func admittedRecord(agreed federationRole) participant { + record := activeRecord() + record.FederationRole = agreed + return record +} + +func agreedPacks() *federationDirection { + return &federationDirection{SchemaPacks: []string{testAgreedPack}} +} + +// askedFor is the context a served discover carries: who signed it, and the +// capabilities their body named. +func askedFor(caller string, contexts ...string) context.Context { + return actionAskedFor(caller, discoverAction, contexts...) +} + +func actionAskedFor(caller, action string, contexts ...string) context.Context { + body := []byte(`{"context":{"action":"` + action + `","senderId":"` + caller + `"}}`) + if contexts != nil { + quoted := make([]string, 0, len(contexts)) + for _, c := range contexts { + quoted = append(quoted, `"`+c+`"`) + } + body = []byte(`{"context":{"action":"` + action + `","senderId":"` + caller + + `","schemaContext":[` + strings.Join(quoted, ",") + `]}}`) + } + return &model.StepContext{Context: calledBy(caller), Body: body} +} + +// calledBy is the context the adapter serves a request in: reqpreprocessor has +// already resolved who signed it. +func calledBy(caller string) context.Context { + return context.WithValue(context.Background(), model.ContextKeyRemoteID, caller) +} + +func lookupAs(t *testing.T, c *Client, ctx context.Context) ([]model.Subscription, error) { + t.Helper() + return c.Lookup(ctx, &model.Subscription{ + Subscriber: model.Subscriber{SubscriberID: testParticipantID}, + KeyID: testOSID, + }) +} + +func crossNetworkClient(t *testing.T, role federationRole, on bool) *Client { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, recordJSON(t, admittedRecord(role))) + })) + t.Cleanup(srv.Close) + return newTestClient(t, srv.URL, nil, func(cfg *Config) { + cfg.CrossNetworkCheck = on + }) +} + +func TestADirectionIsOnlyHonouredWhenTheCheckIsOn(t *testing.T) { + t.Parallel() + + results, err := lookupAs(t, crossNetworkClient(t, federationRole{Source: agreedPacks()}, false), + calledBy(testParticipantID)) + if err != nil { + t.Fatalf("with the check off a lookup must behave as before: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected the key to be returned, got %d results", len(results)) + } +} + +func TestACallerWeAgreedToGiveToResolves(t *testing.T) { + t.Parallel() + + for name, role := range map[string]federationRole{ + "requester": {Requester: agreedPacks()}, + "both": {Source: agreedPacks(), Requester: agreedPacks()}, + } { + t.Run(name, func(t *testing.T) { + results, err := lookupAs(t, crossNetworkClient(t, role, true), + askedFor(testParticipantID, testAgreedContext)) + if err != nil { + t.Fatalf("a %s caller asking within the agreement was refused: %v", name, err) + } + if len(results) != 1 { + t.Fatalf("expected the key to be returned, got %d results", len(results)) + } + }) + } +} + +func TestACallerWeOnlyAgreedToReadFromIsRefused(t *testing.T) { + t.Parallel() + + results, err := lookupAs(t, crossNetworkClient(t, federationRole{Source: agreedPacks()}, true), + calledBy(testParticipantID)) + if err == nil { + t.Fatal("a network we only agreed to read from was given a key to call us with") + } + if len(results) != 0 { + t.Errorf("expected no key, got %d results", len(results)) + } + if !strings.Contains(err.Error(), testParticipantID) { + t.Errorf("error = %v, want it to name the caller", err) + } +} + +func TestARecordWithNoDirectionIsLeftAlone(t *testing.T) { + t.Parallel() + + results, err := lookupAs(t, crossNetworkClient(t, federationRole{}, true), calledBy(testParticipantID)) + if err != nil { + t.Fatalf("a record with no federationRole was refused: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected the key to be returned, got %d results", len(results)) + } +} + +func TestAKeyResolvedOutsideARequestIsNotJudged(t *testing.T) { + t.Parallel() + + results, err := lookupAs(t, crossNetworkClient(t, federationRole{Source: agreedPacks()}, true), + context.Background()) + if err != nil { + t.Fatalf("resolving a key outside a request was refused: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected the key to be returned, got %d results", len(results)) + } +} + +func TestOnlyTheCallerIsJudged(t *testing.T) { + t.Parallel() + + results, err := lookupAs(t, crossNetworkClient(t, federationRole{Source: agreedPacks()}, true), + calledBy("somebody-else.oan.local")) + if err != nil { + t.Fatalf("a lookup for a party other than the caller was refused: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected the key to be returned, got %d results", len(results)) + } +} + +// memoryCache is a real round trip, because this test is about what a cached +// answer skips. +type memoryCache struct { + mu sync.Mutex + items map[string]string +} + +func (c *memoryCache) Get(_ context.Context, key string) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + if v, ok := c.items[key]; ok { + return v, nil + } + return "", fmt.Errorf("miss") +} + +func (c *memoryCache) Set(_ context.Context, key, value string, _ time.Duration) error { + c.mu.Lock() + defer c.mu.Unlock() + if c.items == nil { + c.items = map[string]string{} + } + c.items[key] = value + return nil +} + +func (c *memoryCache) Delete(_ context.Context, key string) error { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.items, key) + return nil +} + +func (c *memoryCache) Clear(_ context.Context) error { return nil } + +func TestACachedAnswerIsJudgedToo(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, recordJSON(t, admittedRecord(federationRole{Source: agreedPacks()}))) + })) + defer srv.Close() + + client := newTestClient(t, srv.URL, &memoryCache{}, func(cfg *Config) { + cfg.CrossNetworkCheck = true + cfg.CacheTTL = time.Minute + }) + + if _, err := lookupAs(t, client, context.Background()); err != nil { + t.Fatalf("warming the cache failed: %v", err) + } + if _, err := lookupAs(t, client, calledBy(testParticipantID)); err == nil { + t.Fatal("a cached record skipped the direction check") + } +} + +func TestADiscoverNamingNoCapabilityIsRefused(t *testing.T) { + t.Parallel() + + for name, ctx := range map[string]context.Context{ + "no schemaContext": askedFor(testParticipantID), + "empty list": askedFor(testParticipantID, []string{}...), + "blank entry": askedFor(testParticipantID, " "), + } { + t.Run(name, func(t *testing.T) { + results, err := lookupAs(t, crossNetworkClient(t, federationRole{Requester: agreedPacks()}, true), ctx) + if err == nil { + t.Fatal("a discover naming no capability asks for everything and must be refused") + } + if len(results) != 0 { + t.Errorf("expected no key, got %d results", len(results)) + } + }) + } +} + +func TestADiscoverAskingOutsideTheAgreementIsRefused(t *testing.T) { + t.Parallel() + + results, err := lookupAs(t, crossNetworkClient(t, federationRole{Requester: agreedPacks()}, true), + askedFor(testParticipantID, testAgreedContext, "https://x/Weather/v0.1/context.jsonld")) + if err == nil { + t.Fatal("a capability outside the agreement was given a key") + } + if len(results) != 0 { + t.Errorf("expected no key, got %d results", len(results)) + } + if strings.Contains(err.Error(), "Weather") { + t.Errorf("error = %v, want it not to name what we hold", err) + } +} + +func TestOnlyADiscoverIsCheckedAgainstTheAgreedSubset(t *testing.T) { + t.Parallel() + + for _, action := range []string{"select", "init", "confirm", "status"} { + t.Run(action, func(t *testing.T) { + results, err := lookupAs(t, crossNetworkClient(t, federationRole{Requester: agreedPacks()}, true), + actionAskedFor(testParticipantID, action)) + if err != nil { + t.Fatalf("a %s from a network we agreed to give to was refused: %v", action, err) + } + if len(results) != 1 { + t.Fatalf("expected the key to be returned, got %d results", len(results)) + } + }) + } +} + +func TestASelectFromANetworkWeOnlyReadFromIsStillRefused(t *testing.T) { + t.Parallel() + + _, err := lookupAs(t, crossNetworkClient(t, federationRole{Source: agreedPacks()}, true), + actionAskedFor(testParticipantID, "select")) + if err == nil { + t.Fatal("the direction still governs an action that names no capability") + } +} diff --git a/pkg/plugin/implementation/sunbirdRegistry/sunbirdRegistry.go b/pkg/plugin/implementation/sunbirdRegistry/sunbirdRegistry.go index 522064b79..d0359b6c6 100644 --- a/pkg/plugin/implementation/sunbirdRegistry/sunbirdRegistry.go +++ b/pkg/plugin/implementation/sunbirdRegistry/sunbirdRegistry.go @@ -12,6 +12,7 @@ import ( "fmt" "net/http" "net/url" + "slices" "strings" "time" @@ -101,6 +102,7 @@ const ( outcomeInactive = "inactive" outcomeKeyInactive = "key_inactive" outcomeNoKey = "no_key" + outcomeRefused = "not_a_requester" outcomeTimeout = "timeout" outcomeRegistryError = "registry_error" outcomeDecodeError = "decode_error" @@ -156,6 +158,10 @@ type Config struct { // truncated, because half a JSON document fails to decode with an error // that says nothing about the cause. MaxResponseBytes int64 `yaml:"maxResponseBytes" json:"maxResponseBytes"` + + // CrossNetworkCheck refuses a key to a network we did not agree to give to. + // Off by default, so a deployment that federates with nobody is unchanged. + CrossNetworkCheck bool `yaml:"crossNetworkCheck" json:"crossNetworkCheck"` } // Client resolves participants from the registry. It is safe for concurrent @@ -167,6 +173,7 @@ type Client struct { cache definition.Cache cacheTTL time.Duration maxResponseBytes int64 + crossNetworkCheck bool } // participant is the subset of a registry record this plugin reads. The @@ -241,6 +248,15 @@ func (d *federationDirection) agreed() *model.FederationDirection { return &model.FederationDirection{SchemaPacks: d.SchemaPacks} } +// agreed is what this record says we agreed, in the shape the rest of the +// adapter reads it in. +func (f federationRole) agreed() model.Agreement { + return model.Agreement{ + Source: f.Source.agreed(), + Requester: f.Requester.agreed(), + } +} + // key is one published key. A participant publishes several -- separate signing // and encryption keys, and more than one signing key while a rotation is in // flight -- so a key is identified by its own OSID rather than by its position. @@ -383,6 +399,7 @@ func New(ctx context.Context, cache definition.Cache, cfg *Config) (*Client, fun cache: cache, cacheTTL: cfg.CacheTTL, maxResponseBytes: maxResponseBytes, + crossNetworkCheck: cfg.CrossNetworkCheck, } closer := func() error { @@ -408,6 +425,7 @@ func New(ctx context.Context, cache definition.Cache, cfg *Config) (*Client, fun // distinguishable instead of collapsing into the same empty result. func (c *Client) Lookup(ctx context.Context, req *model.Subscription) ([]model.Subscription, error) { start := time.Now() + asked := servedRequestFrom(ctx) tracer := otel.Tracer(telemetry.ScopeName, trace.WithInstrumentationVersion(telemetry.ScopeVersion)) ctx, span := tracer.Start(ctx, "registry lookup") defer span.End() @@ -424,6 +442,11 @@ func (c *Client) Lookup(ctx context.Context, req *model.Subscription) ([]model.S cacheKey := fmt.Sprintf("registry_lookup_%s_%s", req.SubscriberID, req.KeyID) if cached, ok := c.cached(ctx, tracer, cacheKey); ok { log.Debugf(ctx, "OAN registry lookup cache hit for key: %s", cacheKey) + if err := c.agreedToGiveTo(ctx, asked, cached); err != nil { + span.SetAttributes(telemetry.AttrErrorType.String(outcomeRefused)) + c.emitMetrics(ctx, start, operationLookup, outcomeRefused) + return nil, err + } span.SetAttributes(telemetry.AttrErrorType.String(outcomeCacheHit)) c.emitMetrics(ctx, start, operationLookup, outcomeCacheHit) return cached, nil @@ -453,6 +476,12 @@ func (c *Client) Lookup(ctx context.Context, req *model.Subscription) ([]model.S status, outcome := resolveStatus(found, matched) results := []model.Subscription{toSubscription(found, matched, status)} + if err := c.agreedToGiveTo(ctx, asked, results); err != nil { + span.SetAttributes(telemetry.AttrErrorType.String(outcomeRefused)) + c.emitMetrics(ctx, start, operationLookup, outcomeRefused) + return nil, err + } + // The header's algorithm was already validated upstream, so a disagreement // here cannot let a bad signature through -- but it means the record and the // caller disagree about the key, which is worth surfacing before it becomes a @@ -615,6 +644,7 @@ func toSubscription(p participant, k key, status string) model.Subscription { ValidFrom: validFrom, ValidUntil: validUntil, Status: status, + Agreement: p.FederationRole.agreed(), } } @@ -729,3 +759,102 @@ func (c *Client) emitMetrics(ctx context.Context, start time.Time, operation, ou m.PluginErrorsTotal.Add(ctx, 1, attrs) } } + +// discoverAction is the only action that names capabilities, and so the only +// one the agreed subset can be checked against. +const discoverAction = "discover" + +// servedRequest is who signed the request we are serving, what they are asking +// for, and which capabilities their body named. Read before the span wraps the +// context, which would otherwise hide the StepContext behind a plain one. +type servedRequest struct { + caller string + action string + capabilities []string +} + +func servedRequestFrom(ctx context.Context) servedRequest { + asked := servedRequest{caller: remoteIDOf(ctx)} + sc, ok := ctx.(*model.StepContext) + if !ok || len(sc.Body) == 0 { + return asked + } + var payload struct { + Context map[string]any `json:"context"` + } + if err := json.Unmarshal(sc.Body, &payload); err != nil || payload.Context == nil { + return asked + } + if asked.caller == "" { + asked.caller = model.ResolveCallerID(payload.Context, model.RoleBPP) + } + asked.action, _ = payload.Context["action"].(string) + asked.capabilities = schemaContextOf(payload.Context) + return asked +} + +func remoteIDOf(ctx context.Context) string { + v, _ := ctx.Value(model.ContextKeyRemoteID).(string) + return v +} + +func schemaContextOf(beckn map[string]any) []string { + listed, _ := beckn["schemaContext"].([]any) + capabilities := make([]string, 0, len(listed)) + for _, entry := range listed { + if name, ok := entry.(string); ok && strings.TrimSpace(name) != "" { + capabilities = append(capabilities, name) + } + } + return capabilities +} + +// agreedToGiveTo refuses a key to a network we never agreed to give to, which +// stops the request at validateSign. Only the caller is judged. +func (c *Client) agreedToGiveTo(ctx context.Context, asked servedRequest, results []model.Subscription) error { + if !c.crossNetworkCheck || len(results) == 0 || asked.caller == "" { + return nil + } + + for _, sub := range results { + if !strings.EqualFold(sub.SubscriberID, asked.caller) { + continue + } + // Our own participants agreed nothing, so there is nothing to honour. + if !sub.Agreement.Agreed() { + return nil + } + if !sub.Agreement.IsRequester() { + log.Infof(ctx, "OAN registry refusing %s: we agreed %v, which is not giving to them", + asked.caller, sub.Agreement.Directions()) + return fmt.Errorf("%s is not a network we agreed to give to", asked.caller) + } + // Only a discover names capabilities. Every other action is governed by + // the direction alone, which we just checked. + if asked.action != discoverAction { + log.Infof(ctx, "OAN registry %s may call us for %q", asked.caller, asked.action) + return nil + } + return agreedToGiveThese(ctx, asked, sub.Agreement.RequesterContexts()) + } + return nil +} + +// agreedToGiveThese refuses a caller asking beyond what the agreement covers. +// The refusal names nothing: listing what we hold but will not share would tell +// a caller the shape of our catalogue. +func agreedToGiveThese(ctx context.Context, asked servedRequest, agreed []string) error { + if len(asked.capabilities) == 0 { + log.Infof(ctx, "OAN registry refusing %s: named no capability, which asks for everything", asked.caller) + return fmt.Errorf("%s named no capability", asked.caller) + } + for _, want := range asked.capabilities { + if !slices.Contains(agreed, want) { + log.Infof(ctx, "OAN registry refusing %s: asked for %s, which we did not agree to give", + asked.caller, want) + return fmt.Errorf("%s asked for a capability we did not agree to give", asked.caller) + } + } + log.Infof(ctx, "OAN registry %s may call us, and asked only within what we agreed", asked.caller) + return nil +}