Stricter validations for centralized registry workflows #23013
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| package standardcapabilities | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func Test_parseTenantID(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| tests := []struct { | ||
| name string | ||
| config string | ||
| want uint64 | ||
| wantOK bool | ||
| }{ | ||
| {name: "empty config", config: "", want: 0, wantOK: false}, | ||
| {name: "no tenantID", config: `{"schedule":"1s"}`, want: 0, wantOK: false}, | ||
| {name: "zero tenantID", config: `{"tenantID":0}`, want: 0, wantOK: false}, | ||
| {name: "tenantID set", config: `{"tenantID":1}`, want: 1, wantOK: true}, | ||
| {name: "tenantID with other fields", config: `{"foo":"bar","tenantID":42}`, want: 42, wantOK: true}, | ||
| {name: "invalid json", config: `{not json`, want: 0, wantOK: false}, | ||
| } | ||
|
|
||
| for _, tt := range tests { | ||
| t.Run(tt.name, func(t *testing.T) { | ||
| t.Parallel() | ||
| got, ok := parseTenantID(tt.config) | ||
| assert.Equal(t, tt.wantOK, ok) | ||
| assert.Equal(t, tt.want, got) | ||
| }) | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -7,7 +7,10 @@ import ( | |
| "fmt" | ||
| "io" | ||
| "maps" | ||
| "strconv" | ||
| "strings" | ||
| "sync" | ||
| "sync/atomic" | ||
| "time" | ||
|
|
||
| "go.opentelemetry.io/otel" | ||
|
|
@@ -89,7 +92,11 @@ type eventHandler struct { | |
| workflowDonSubscriber capabilities.DonSubscriber | ||
| billingClient metering.BillingClient | ||
| orgResolver orgresolver.OrgResolver | ||
| // tenantID is the tenant numeric id for the CRE environment. It is sourced from a | ||
| // job spec (see SetTenantID, wired from the consensus OCR2 plugin) and may be | ||
| // updated at runtime, so access is atomic. Zero falls back to defaultTenantID. | ||
| tenantID atomic.Uint64 | ||
| secretsFetcher v2.SecretsFetcher | ||
| // localSecretOverrides is keyed by owner address; values are secret id -> secret value | ||
| localSecretOverrides map[string]map[string]string | ||
|
|
||
|
|
@@ -180,6 +187,21 @@ func WithOrgResolver(orgResolver orgresolver.OrgResolver) func(*eventHandler) { | |
| } | ||
| } | ||
|
|
||
| // WithTenantID sets the tenant numeric id for the CRE environment. A zero value | ||
| // falls back to defaultTenantID. | ||
| func WithTenantID(tenantID uint64) func(*eventHandler) { | ||
| return func(e *eventHandler) { | ||
| e.tenantID.Store(tenantID) | ||
| } | ||
| } | ||
|
|
||
| // SetTenantID updates the tenant numeric id for the CRE environment. It is safe for | ||
| // concurrent use and is called at runtime when the owning job spec (consensus OCR2 | ||
| // plugin) is applied. | ||
| func (h *eventHandler) SetTenantID(tenantID uint64) { | ||
| h.tenantID.Store(tenantID) | ||
| } | ||
|
|
||
| // WithDebugMode enables OTel tracing when debugMode is true. | ||
| // When disabled (default), a noop tracer is used for zero overhead. | ||
| // The debugMode is also propagated to workflow engines created by this handler. | ||
|
|
@@ -640,6 +662,24 @@ func (h *eventHandler) createWorkflowSpec(ctx context.Context, payload WorkflowR | |
| } | ||
| ctx = contexts.WithCRE(ctx, contexts.CRE{Org: orgID, Owner: owner, Workflow: wfID}) | ||
|
|
||
| // Defense-in-depth against centralized workflow source owner spoofing: | ||
| // on-chain sources enforce ownership via msg.sender, but centralized (off-chain) | ||
| // sources accept an owner claim without any owner-signed proof. For centralized | ||
| // sources, verify that the claimed owner is the address deterministically derived | ||
| // from the organization ID that the Linking Service maps it to. A mismatch means | ||
| // the registry is asserting an owner it cannot prove ownership of (e.g. an on-chain | ||
| // EOA), so we log a critical error and reject the workflow (fail-closed). | ||
| if isCentralizedWorkflowSource(payload.Source) { | ||
| if gateErr := h.engineLimiters.CentralizedWorkflowOwnerVerificationEnabled.AllowErr(ctx); gateErr == nil { | ||
| if verr := h.verifyCentralizedOwnerOrgMapping(payload.Source, owner, orgID); verr != nil { | ||
| return nil, verr | ||
| } | ||
| } else if !errors.Is(gateErr, limits.ErrorNotAllowed{}) { | ||
| h.lggr.Warnw("failed to evaluate limit CentralizedWorkflowOwnerVerificationEnabled", "error", gateErr) | ||
| return nil, gateErr | ||
| } | ||
| } | ||
|
|
||
| // With Workflow Registry contract v2 the BinaryURL and ConfigURL are expected to be identifiers that put through the Storage Service. | ||
| decodedBinary, config, err := h.workflowArtifactsStore.FetchWorkflowArtifacts(ctx, wfID, payload.BinaryURL, payload.ConfigURL) | ||
| if err != nil { | ||
|
|
@@ -670,6 +710,87 @@ func (h *eventHandler) createWorkflowSpec(ctx context.Context, payload WorkflowR | |
| return entry, nil | ||
| } | ||
|
|
||
| // defaultTenantID is the fallback tenant numeric id for the CRE environment when none | ||
| // is configured on the job spec. It matches defaultJWTAuthJobSpecTenantID in | ||
| // core/capabilities/vault and cre-platform-graphql's account service. | ||
| const defaultTenantID uint64 = 1 | ||
|
Contributor
There was a problem hiding this comment. I think we should be very careful when using defaults - if the tenant ID ever disappears from the job spec by mistake, DONs will start rejecting requests. I would rather enforce it in the job spec if that's a possibility.
Contributor
Author
There was a problem hiding this comment. We're using the same pattern in other vault checks too, to default to a tenantID of 1 if not provided. I could also make this an explicit failure if tenantID wasn't set. But that can only be done after we first fix all job-specs and deploy them across all environments. Otherwise this change will start breaking rollouts immediately.
Contributor
There was a problem hiding this comment. But if things break, isn't it harder for you to fix an invalid tenant ID because you have to go through the process of job spec rollout for the entire DON and the resolution time depends on how quickly a NOP can accept the updated job spec? I understand that you can't get rid of defaults immediately due to what you have described above, but just asking to consider the risks one more time (I'm unfamiliar with this codebase, so I can't tell you if this is a bad or a good idea, but I'll defer to @cedric-cordenier to weigh in).
Contributor
Author
There was a problem hiding this comment. My idea was to set the right tenantID on all environments job-specs, and push them out. |
||
|
|
||
| // ErrCentralizedOwnerOrgMismatch is returned when a workflow owner claimed by a | ||
| // centralized (off-chain) source does not match the owner deterministically derived | ||
| // from its Linking Service organization ID. The workflow is rejected: admitting it | ||
| // would let a centralized registry impersonate an owner (e.g. an on-chain EOA) and | ||
| // exfiltrate that owner's Vault secrets. | ||
| var ErrCentralizedOwnerOrgMismatch = errors.New("centralized workflow owner does not match owner derived from its organization ID") | ||
|
|
||
| // tenantIDOrDefault returns the tenant id for the CRE environment, falling back to | ||
| // defaultTenantID when the job spec does not configure one. | ||
| func (h *eventHandler) tenantIDOrDefault() uint64 { | ||
| if v := h.tenantID.Load(); v != 0 { | ||
| return v | ||
| } | ||
| return defaultTenantID | ||
| } | ||
|
|
||
| // isCentralizedWorkflowSource reports whether workflow metadata originated from a | ||
| // centralized / off-chain source (the gRPC workflow registry or a local file source), | ||
| // as opposed to the on-chain workflow registry contract (source prefix "contract:"). | ||
| // On-chain ownership is enforced by msg.sender; off-chain sources accept an owner | ||
| // claim without on-chain proof and therefore require independent owner<->orgID | ||
| // verification before the workflow is admitted. | ||
| func isCentralizedWorkflowSource(source string) bool { | ||
| return strings.HasPrefix(source, "grpc:") || strings.HasPrefix(source, "file:") | ||
|
Contributor
There was a problem hiding this comment. Where does this source string come from?
Contributor
Author
There was a problem hiding this comment. Here: And that field is set here: I will also ask @cedric-cordenier to confirm again, that this is the right authoritative place to read the source from, and isn't spoofable.
Contributor
There was a problem hiding this comment. Okay, I'm asking because I'm pretty sure that the workflow registry does not provide this field, so the workflow metadata must have been augmented during processing on the node side. |
||
| } | ||
|
|
||
| // verifyCentralizedOwnerOrgMapping checks that a workflow owner claimed by a | ||
| // centralized source is consistent with the organization the Linking Service maps it | ||
| // to. Legitimate centralized CRE workflow owners are deterministically derived from | ||
| // (tenantID, orgID) via workflows.GenerateWorkflowOwnerAddress. This mirrors the | ||
| // derivation in core/capabilities/vault/workflow_owner_derivation.go | ||
| // (DeriveJWTAuthorizedVaultWorkflowOwner) and cre-platform-graphql | ||
| // account_service.GetCreOrganizationInfo, which is also the source of truth the Vault | ||
| // DON authorizes secret access against. A mismatch means the centralized registry is | ||
| // asserting an owner it cannot prove ownership of (for example an on-chain EOA), | ||
| // which indicates data corruption or a malicious registry attempting to hijack | ||
| // another owner's identity and secrets. | ||
| // | ||
| // On a mismatch it logs a critical error and returns ErrCentralizedOwnerOrgMismatch | ||
| // so the caller rejects the workflow (fail-closed). If the orgID cannot be resolved | ||
| // or the derivation fails, verification is skipped (returns nil) rather than blocking | ||
| // legitimate workflows on a transient Linking Service outage. | ||
| func (h *eventHandler) verifyCentralizedOwnerOrgMapping(source, ownerHex, orgID string) error { | ||
|
Contributor
There was a problem hiding this comment. Did you consider putting this logic in the GRPC contract source instead? Feels like that is the more natural place for it -- we should check that we can re-derive the owner from the org ID + tenant ID there |
||
| if orgID == "" { | ||
| // Without an orgID we cannot derive the expected owner; the resolution failure | ||
| // was already logged by fetchOrganizationID. | ||
| h.lggr.Warnw("skipping centralized workflow owner/orgID verification: no organization ID resolved", | ||
|
Contributor
There was a problem hiding this comment. Are we sure we don't want to treat this as an error instead? What happens when you can't resolve an org, do you usually just proceed?
Contributor
Author
There was a problem hiding this comment. Yes, in my understanding too, if we cannot fetch orgID, we should likely fail even running any workflow, not just this specific check. I didn't explicitly force that for this check, because it means all workflows will start failing without orgID, which currently isn't so. @cedric-cordenier do you know if there's a harm in enforcing strict orgID presence?
Contributor
There was a problem hiding this comment. Atm we fail open, so I wouldn't start enforcing this as part of this PR |
||
| "source", source, "workflowOwner", ownerHex) | ||
| return nil | ||
| } | ||
|
|
||
| tenantID := h.tenantIDOrDefault() | ||
| derived, err := pkgworkflows.GenerateWorkflowOwnerAddress(strconv.FormatUint(tenantID, 10), orgID) | ||
| if err != nil { | ||
| h.lggr.Errorw("failed to derive expected workflow owner for centralized source", | ||
| "source", source, "workflowOwner", ownerHex, "organizationID", orgID, "error", err) | ||
| return nil | ||
| } | ||
| derivedHex := hex.EncodeToString(derived) | ||
|
|
||
| claimedOwner := strings.ToLower(strings.TrimPrefix(strings.TrimSpace(ownerHex), "0x")) | ||
| if strings.ToLower(derivedHex) != claimedOwner { | ||
| logger.Sugared(h.lggr).Criticalw( | ||
| "centralized workflow owner does not match owner derived from its organization ID: possible data corruption or malicious workflow registry", | ||
| "source", source, | ||
| "claimedWorkflowOwner", ownerHex, | ||
| "organizationID", orgID, | ||
| "derivedWorkflowOwner", derivedHex, | ||
| "tenantID", tenantID, | ||
| ) | ||
| return fmt.Errorf("%w: source=%s claimedOwner=%s organizationID=%s derivedOwner=%s", | ||
| ErrCentralizedOwnerOrgMismatch, source, ownerHex, orgID, derivedHex) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // fetchOrganizationID fetches the organization ID for the given workflow owner using the OrgResolver | ||
| func (h *eventHandler) fetchOrganizationID(ctx context.Context, workflowOwner string) (string, error) { | ||
| if h.orgResolver == nil { | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
:( This feels pretty wrong tbh -- what happens if multiple capabilities have different tenant IDs? Atm the last one ones and we're probably not guaranteed that it will consistently win