Merge pull request #272 from marcelamelara/add-scai-bindings

Add Go and Python bindings for SCAI predicate

Author: pxp928

go/predicates/scai/v0/scai.go

@@ -0,0 +1,53 @@
+/*
+Wrapper APIs for SCAI AttributeAssertion and AttributeReport protos.
+*/
+
+package v0
+
+import "fmt"
+
+func (a *AttributeAssertion) Validate() error {
+	// at least the attribute field is required
+	if a.GetAttribute() == "" {
+		return fmt.Errorf("The attribute field is required")
+	}
+
+	// check target and evidence are valid ResourceDescriptors
+	if a.GetTarget() != nil {
+		if err := a.GetTarget().Validate(); err != nil {
+			return fmt.Errorf("Target validation failed with an error: %w", err)
+		}
+	}
+
+	if a.GetEvidence() != nil {
+		if err := a.GetEvidence().Validate(); err != nil {
+			return fmt.Errorf("Evidence validation failed with an error: %w", err)
+		}
+	}
+
+	return nil
+}
+
+func (r *AttributeReport) Validate() error {
+	// at least the attributes field is required
+	attrs := r.GetAttributes()
+	if attrs == nil || len(attrs) == 0 {
+		return fmt.Errorf("At least one AttributeAssertion is required")
+	}
+
+	// ensure all AttributeAssertions are valid
+	for _, a := range attrs {
+		if err := a.Validate(); err != nil {
+			return fmt.Errorf("AttributeAssertion validation failed with an error: %w", err)
+		}
+	}
+
+	// ensure the producer is a valid ResourceDescriptor
+	if r.GetProducer() != nil {
+		if err := r.GetProducer().Validate(); err != nil {
+			return fmt.Errorf("Producer validation failed with an error: %w", err)
+		}
+	}
+
+	return nil
+}

python/in_toto_attestation/predicates/scai/v0/scai.py

@@ -0,0 +1,70 @@
+# Wrapper class for in-toto attestation SCAI predicate protos.
+
+import in_toto_attestation.predicates.scai.v0.scai_pb2 as scaipb
+from in_toto_attestation.v1.resource_descriptor import ResourceDescriptor
+
+SCAI_PREDICATE_TYPE = 'https://in-toto.io/attestation/scai/attribute-report/'
+SCAI_PREDICATE_VERSION = 'v0.2'
+
+class AttributeAssertion:
+    def __init__(self, attribute, target=None, conditions=None, evidence=None) -> None:
+        self.pb = scaipb.AttributeAssertion()
+        self.pb.attribute = attribute
+        if target:
+            self.pb.target.CopyFrom(target)
+        if conditions:
+            self.pb.conditions.update(conditions)
+        if evidence:
+            self.pb.evidence.CopyFrom(evidence)
+
+    @staticmethod
+    def copy_from_pb(proto: type[scaipb.AttributeAssertion]) -> 'AttributeAssertion':
+        assertion = AttributeAssertion('tmp-attr')
+        assertion.pb.CopyFrom(proto)
+        return assertion
+
+    def validate(self) -> None:
+        if len(self.pb.attribute) == 0:
+            raise ValueError('The attribute field is required')
+
+        # check any resource descriptors are valid
+        if self.pb.target.ByteSize() > 0:
+            rd = ResourceDescriptor.copy_from_pb(self.pb.target)
+            rd.validate()
+
+        if self.pb.evidence.ByteSize() > 0:
+            rd = ResourceDescriptor.copy_from_pb(self.pb.evidence)
+            rd.validate()
+
+class AttributeReport:
+    def __init__(self, attributes: list, producer=None) -> None:
+        self.pb = scaipb.AttributeReport()
+        self.pb.attributes.extend(attributes)
+        if producer:
+            self.pb.producer.CopyFrom(producer)
+
+    @staticmethod
+    def copy_from_pb(proto: type[scaipb.AttributeReport]) -> 'AttributeReport':
+        report = AttributeReport([None])
+        report.pb.CopyFrom(proto)
+        return report
+
+    def validate(self) -> None:
+        if len(self.pb.attributes) == 0:
+            raise ValueError('The attributes field is required')
+
+        # check any attribute assertions are valid
+        attributes = self.pb.attributes
+        for i, attrpb in enumerate(attributes):
+            assertion = AttributeAssertion.copy_from_pb(attrpb)
+
+            try:
+                assertion.validate()
+            except ValueError as e:
+                # return index in the attributes list in case of failure:
+                # can't assume any other fields in attribute assertion are set
+                raise ValueError('Invalid attributes field (index: {0})'.format(i)) from e
+
+        if self.pb.producer.ByteSize() > 0:
+            rd = ResourceDescriptor.copy_from_pb(self.pb.producer)
+            rd.validate()