Get Exploit

2026-06-04 11:31:03 +02:00
parent d19ff1f8b2
commit c726361deb
2 changed files with 94 additions and 0 deletions
@@ -1492,6 +1492,99 @@ func (w *Workflow) detectIsolatedProcessings() []IntegrityViolation {
 	return violations
 }

+// ---------------------------------------------------------------------------
+// AE validation helpers — centralised so both oc-scheduler and oc-schedulerd
+// share the same logic without code duplication.
+// ---------------------------------------------------------------------------
+
+// BuildResourceIDSet constructs the per-type resource-ID map and the flat
+// coupling-membership set used by ValidateWorkflowAE.
+//
+// selectedEmbeddedStorages and selectedInstances come from the scheduling
+// request (WorkflowSchedule) or from the WorkflowExecution at launch time.
+// Embedded storages are NOT stored in Workflow.Storages (they are inside
+// ComputeResourceInstance.AvailableStorages), so they must be resolved here
+// to make them visible to the AE coupling check.
+func (w *Workflow) BuildResourceIDSet(
+	selectedEmbeddedStorages map[string]*resources.EmbeddedStorageSelection,
+	selectedInstances ConfigItem,
+) (map[tools.DataType][]string, map[string]struct{}) {
+	resourcesByType := map[tools.DataType][]string{
+		tools.DATA_RESOURCE:       w.Datas,
+		tools.PROCESSING_RESOURCE: w.Processings,
+		tools.STORAGE_RESOURCE:    append([]string{}, w.Storages...),
+		tools.COMPUTE_RESOURCE:    w.Computes,
+		tools.WORKFLOW_RESOURCE:   w.Workflows,
+		tools.SERVICE_RESOURCE:    w.Services,
+	}
+	idSet := map[string]struct{}{}
+	for _, ids := range resourcesByType {
+		for _, id := range ids {
+			idSet[id] = struct{}{}
+		}
+	}
+	for graphItemID, sel := range selectedEmbeddedStorages {
+		if sel == nil {
+			continue
+		}
+		c, ok := w.Graph.Items[graphItemID]
+		if !ok {
+			continue
+		}
+		_, computeRes := c.GetResource()
+		computeResource, ok := computeRes.(*resources.ComputeResource)
+		if !ok {
+			continue
+		}
+		computeIdx := 0
+		if d := selectedInstances.Get(computeResource.GetID()); d != nil {
+			computeIdx = *d
+		}
+		if computeIdx >= len(computeResource.Instances) {
+			continue
+		}
+		computeInst := computeResource.Instances[computeIdx]
+		if sel.StorageIndex >= len(computeInst.AvailableStorages) {
+			continue
+		}
+		storageID := computeInst.AvailableStorages[sel.StorageIndex].GetID()
+		if storageID == "" {
+			continue
+		}
+		idSet[storageID] = struct{}{}
+		resourcesByType[tools.STORAGE_RESOURCE] = append(resourcesByType[tools.STORAGE_RESOURCE], storageID)
+	}
+	return resourcesByType, idSet
+}
+
+// ValidateWorkflowAE checks the ExploitationAuthorizations of every resource
+// referenced in resourcesByType against the coupling/peer/workflow constraints.
+//
+// loadResource is injected by the caller to avoid a circular import
+// (oc-lib/models/resources → oclib → oc-lib/models → resources).
+// A nil return from loadResource means "resource not found — skip".
+func (w *Workflow) ValidateWorkflowAE(
+	workflowID, consumerPeerID string,
+	resourcesByType map[tools.DataType][]string,
+	idSet map[string]struct{},
+	loadResource func(tools.DataType, string) resources.ResourceInterface,
+) []resources.AEViolation {
+	now := time.Now().UTC()
+	var violations []resources.AEViolation
+	for dt, ids := range resourcesByType {
+		for _, id := range ids {
+			res := loadResource(dt, id)
+			if res == nil {
+				continue
+			}
+			for _, ae := range res.GetExploitationAuthorizations() {
+				violations = append(violations, ae.CheckAE(id, workflowID, consumerPeerID, idSet, now)...)
+			}
+		}
+	}
+	return violations
+}
+
 // detectOrphanedStorages warns when a storage node is not linked to any
 // processing node — it contributes no data flow to the workflow.
 func (w *Workflow) detectOrphanedStorages() []IntegrityViolation {