add Neutralizer

Closes: #6

cmd/server/config.go

@@ -12,12 +12,16 @@ import (
 	"github.com/pelletier/go-toml"
 )
 
+type configDataProviderNeutralizerMinResultsSize struct {
+	MinimumResultsSize uint
+}
 type configDataProviderFileLoader struct {
 	Path string
 }
 type configDataProvider struct {
-	FileLoader *configDataProviderFileLoader
-	Random     *struct{}
+	FileLoader  *configDataProviderFileLoader
+	Random      *struct{}
+	Neutralizer *configDataProviderNeutralizerMinResultsSize
 }
 type config struct {
 	Address onet_network.Address

cmd/server/main.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"strconv"
 	"strings"
 
 	kyber_encoding "go.dedis.ch/kyber/v3/util/encoding"
@@ -14,8 +15,9 @@ import (
 	onet_network "go.dedis.ch/onet/v3/network"
 
 	"github.com/ldsec/drynx/lib"
-	provider "github.com/ldsec/drynx/lib/provider"
-	loaders "github.com/ldsec/drynx/lib/provider/loaders"
+	"github.com/ldsec/drynx/lib/provider"
+	"github.com/ldsec/drynx/lib/provider/loaders"
+	"github.com/ldsec/drynx/lib/provider/neutralizers"
 	drynx_services "github.com/ldsec/drynx/services"
 
 	"github.com/pelletier/go-toml"
@@ -78,6 +80,31 @@ func dataProviderNewRandom(c *cli.Context) error {
 	return conf.writeTo(os.Stdout)
 }
 
+func dataProviderSetNeutralizerMinimumResultsSize(c *cli.Context) error {
+	args := c.Args()
+	if len(args) != 1 {
+		return errors.New("need a minimum")
+	}
+	minimum, err := strconv.ParseUint(args[0], 10, 0)
+	if err != nil {
+		return err
+	}
+
+	conf, err := readConfigFrom(os.Stdin)
+	if err != nil {
+		return err
+	}
+
+	if conf.DataProvider == nil {
+		return errors.New("not on data-provider stream")
+	}
+	conf.DataProvider.Neutralizer = &configDataProviderNeutralizerMinResultsSize{
+		MinimumResultsSize: uint(minimum),
+	}
+
+	return conf.writeTo(os.Stdout)
+}
+
 func gen(c *cli.Context) error {
 	args := c.Args()
 	if len(args) != 2 {
@@ -127,9 +154,14 @@ func run(c *cli.Context) error {
 		}
 	}
 
+	var neutralizer provider.Neutralizer
+	if c := conf.DataProvider.Neutralizer; c != nil {
+		neutralizer = neutralizers.NewMinimumResultsSize(c.MinimumResultsSize)
+	}
+
 	drynx_services.NewBuilder().
 		WithComputingNode().
-		WithDataProvider(loader).
+		WithDataProvider(loader, neutralizer).
 		WithVerifyingNode().
 		Start()
 
@@ -178,6 +210,7 @@ func main() {
 	if you want to generate a server config, use something like
 		%[1]s new {1,2}.drynx.c4dt.org |
 			%[1]s data-provider new file-loader $my_data |
+				%[1]s data-provider set-neutralizer minimum-results-size 3 |
 			%[1]s computing-node new |
 			%[1]s verifying-node new >
 			$my_node_config
@@ -213,6 +246,13 @@ func main() {
 				Name:   "random",
 				Action: dataProviderNewRandom,
 			}},
+		}, {
+			Name:  "set-neutralizer",
+			Usage: "on a data-provider config stream, set the neutralizer to use",
+			Subcommands: []cli.Command{{
+				Name:   "minimum-results-size",
+				Action: dataProviderSetNeutralizerMinimumResultsSize,
+			}},
 		}}}, {
 		Name:  "verifying-node",
 		Usage: "verifying-node configuration",

lib/operations/min.go

@@ -0,0 +1,62 @@
+package operations
+
+import (
+	"errors"
+)
+
+const minInputSize = 1
+const minEncodedSize = 1
+
+// Min computes the accumulation of values in a column.
+type Min struct{ Range }
+
+// MarshalID is the Operation's ID.
+func (Min) MarshalID() [8]byte {
+	ret := [8]byte{}
+	copy(ret[:], []byte("dr.op.mi"))
+	return ret
+}
+
+// MarshalBinary returns nil.
+func (Min) MarshalBinary() ([]byte, error) {
+	return nil, nil
+}
+
+// UnmarshalBinary does nothing.
+func (Min) UnmarshalBinary([]byte) error {
+	return nil
+}
+
+// ExecuteOnProvider encodes.
+func (s Min) ExecuteOnProvider(loaded [][]float64) ([]float64, error) {
+	if len(loaded) != minInputSize {
+		return nil, errors.New("unexpected number of columns")
+	}
+
+	// TODO
+	//converted := floatsToInts(loaded[0])
+	//min := libdrynxencoding.ExecuteMinOnProvider(converted)
+
+	return []float64{float64(0)}, nil
+}
+
+// ExecuteOnClient decodes.
+func (s Min) ExecuteOnClient(aggregated []float64) ([]float64, error) {
+	if len(aggregated) != minEncodedSize {
+		return nil, errors.New("unexpected size of aggregated vector")
+	}
+
+	// TODO
+	//return []float64{float64(libdrynxencoding.ExecuteMinOnClient(floatsToInts(aggregated)))}, nil
+	return nil, nil
+}
+
+// GetInputSize returns 1.
+func (Min) GetInputSize() uint {
+	return minInputSize
+}
+
+// GetEncodedSize returns 1.
+func (Min) GetEncodedSize() uint {
+	return minEncodedSize
+}

lib/provider/index.go

@@ -8,3 +8,9 @@ type Loader interface {
 	// Returns a matrix of len Query.Operation.NbrInput
 	Provide(libdrynx.Query) ([][]float64, error)
 }
+
+// Neutralizer decides to release or not the results of a query.
+type Neutralizer interface {
+	// Vet checks if the results can be safely released.
+	Vet(libdrynx.Query, [][]float64) bool
+}

lib/provider/neutralizers/results_size.go

@@ -0,0 +1,19 @@
+package neutralizers
+
+import (
+	"github.com/ldsec/drynx/lib"
+	"github.com/ldsec/drynx/lib/provider"
+)
+
+type minimumResultsSize struct {
+	minimum uint
+}
+
+// NewMinimumResultsSize creates a Neutralizer vetting only when len(results) >= minimum
+func NewMinimumResultsSize(minimum uint) provider.Neutralizer {
+	return minimumResultsSize{minimum}
+}
+
+func (rs minimumResultsSize) Vet(_ libdrynx.Query, results [][]float64) bool {
+	return uint(len(results)) >= rs.minimum
+}

protocols/data_collection_protocol.go

@@ -78,15 +78,19 @@ type DataCollectionProtocol struct {
 	// Protocol proof data
 	MapPIs map[string]onet.ProtocolInstance
 
-	// how to get data locally, TODO own by ServiceDrynx
+	// how to get data locally
 	loader provider.Loader
+
+	// when to refuse to release results
+	neutralizer provider.Neutralizer
 }
 
 // NewDataCollectionProtocol constructs a DataCollection protocol instance
-func NewDataCollectionProtocol(loader provider.Loader) DataCollectionProtocol {
+func NewDataCollectionProtocol(loader provider.Loader, neutralizer provider.Neutralizer) DataCollectionProtocol {
 	return DataCollectionProtocol{
 		FeedbackChannel: make(chan map[string]libunlynx.CipherVector),
 		loader:          loader,
+		neutralizer:     neutralizer,
 	}
 }
 
@@ -170,6 +174,21 @@ func (p *DataCollectionProtocol) Dispatch() error {
 // Support Functions
 //______________________________________________________________________________________________________________________
 
+func generateNeutralResponse(survey SurveyToDP, groupsStrings []string) libdrynx.ResponseDPBytes {
+	encrypted := make(libunlynx.CipherVector, survey.Query.Operation.NbrOutput)
+	for i := range encrypted {
+		encrypted[i] = *libunlynx.EncryptInt(survey.Aggregate, 0)
+	}
+	raw, _, _ := encrypted.ToBytes()
+
+	grouped := make(map[string][]byte, len(groupsStrings))
+	for _, g := range groupsStrings {
+		grouped[g] = raw
+	}
+
+	return libdrynx.ResponseDPBytes{Data: grouped, Len: len(groupsStrings)}
+}
+
 // GenerateData is used to generate data at DPs, this is more for simulation's purposes
 func (p *DataCollectionProtocol) GenerateData() libdrynx.ResponseDPBytes {
 	// Prepare the generation of all possible groups with the query information.
@@ -194,21 +213,17 @@ func (p *DataCollectionProtocol) GenerateData() libdrynx.ResponseDPBytes {
 		}
 	}
 
-	// generate fake random data depending on the operation
+	// load wanted data
 	providedData, err := p.loader.Provide(p.Survey.Query)
 	if err != nil {
 		log.Errorf("unable to provide using loader: %v", err)
+		return generateNeutralResponse(p.Survey, groupsString)
+	}
 
-		encrypted := libunlynx.CipherVector{
-			*libunlynx.EncryptInt(p.Survey.Aggregate, 0)}
-		raw, _, _ := encrypted.ToBytes()
-
-		grouped := make(map[string][]byte, len(groupsString))
-		for _, g := range groupsString {
-			grouped[g] = raw
-		}
-
-		return libdrynx.ResponseDPBytes{Data: grouped, Len: 1}
+	// vet results
+	if n := p.neutralizer; n != nil && !n.Vet(p.Survey.Query, providedData) {
+		log.Warn("results neutralized")
+		return generateNeutralResponse(p.Survey, groupsString)
 	}
 
 	// logistic regression specific

protocols/data_collection_protocol_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/ldsec/drynx/lib"
 	"github.com/ldsec/drynx/lib/provider/loaders"
+	"github.com/ldsec/drynx/lib/provider/neutralizers"
 	"github.com/ldsec/drynx/protocols"
 	"github.com/ldsec/unlynx/lib"
 	"github.com/stretchr/testify/assert"
@@ -99,7 +100,7 @@ func NewDataCollectionTest(tni *onet.TreeNodeInstance) (onet.ProtocolInstance, e
 		return nil, err
 	}
 
-	dcp := protocols.NewDataCollectionProtocol(loader)
+	dcp := protocols.NewDataCollectionProtocol(loader, neutralizers.NewMinimumResultsSize(0))
 	dcp.Survey = query
 	dcp.ProtocolRegister(tni)
 	return &dcp, nil

services/builder.go

@@ -24,7 +24,8 @@ func init() {
 }
 
 type builderDataProvider struct {
-	loader provider.Loader
+	loader      provider.Loader
+	neutralizer provider.Neutralizer
 }
 
 // Builder is the state of node creation.
@@ -62,21 +63,21 @@ func (b Builder) WithComputingNode() Builder {
 }
 
 // WithDataProvider add support for running as a Data Provider.
-func (b Builder) WithDataProvider(loader provider.Loader) Builder {
+func (b Builder) WithDataProvider(loader provider.Loader, neutralizer provider.Neutralizer) Builder {
 	if loader == nil {
 		panic("WithDataProvider: loader == nil")
 	}
 
 	onet_network.RegisterMessage(protocols.AnnouncementDCMessage{})
 	onet_network.RegisterMessage(protocols.DataCollectionMessage{})
 
-	dcp := protocols.NewDataCollectionProtocol(loader)
+	dcp := protocols.NewDataCollectionProtocol(loader, neutralizer)
 	_, err := onet.GlobalProtocolRegister(protocols.DataCollectionProtocolName, dcp.ProtocolRegister)
 	if err != nil {
 		log.Fatal("Error registering <DataCollectionProtocol>:", err)
 	}
 
-	b.dataProvider = &builderDataProvider{loader}
+	b.dataProvider = &builderDataProvider{loader, neutralizer}
 	return b
 }
 
@@ -91,6 +92,10 @@ func (b Builder) Start() {
 	if b.dataProvider != nil {
 		loader = b.dataProvider.loader
 	}
+	var neutralizer provider.Neutralizer
+	if b.dataProvider != nil {
+		neutralizer = b.dataProvider.neutralizer
+	}
 
 	_, err := onet.RegisterNewService(ServiceName, func(c *onet.Context) (onet.Service, error) {
 		if loader == nil {
@@ -101,6 +106,7 @@ func (b Builder) Start() {
 			Survey:           concurrent.NewConcurrentMap(),
 			Mutex:            &sync.Mutex{},
 			loader:           loader,
+			neutralizer:      neutralizer,
 		}
 
 		registerHandler := func(handler interface{}) {

services/service.go

@@ -88,7 +88,8 @@ type ServiceDrynx struct {
 	// -------------------------
 
 	// ---- Data Provider ----
-	loader provider.Loader
+	loader      provider.Loader
+	neutralizer provider.Neutralizer
 	// -------------------------
 
 	// ---- Verifying Nodes ----
@@ -382,7 +383,8 @@ func (s *ServiceDrynx) NewProtocol(tn *onet.TreeNodeInstance, conf *onet.Generic
 	case protocols.ProofCollectionProtocolName:
 		return s.NewProofCollectionProtocolInstance(tn, target)
 	case protocols.DataCollectionProtocolName:
-		dcp := protocols.NewDataCollectionProtocol(s.loader)
+		dcp := protocols.NewDataCollectionProtocol(s.loader, s.neutralizer)
+		// TODO can't we remove this registering?
 		dcp.ProtocolRegister(tn)
 
 		if !tn.IsRoot() {

services/service_test.go

@@ -5,6 +5,7 @@ import (
 	"github.com/ldsec/drynx/lib"
 	"github.com/ldsec/drynx/lib/encoding"
 	"github.com/ldsec/drynx/lib/provider/loaders"
+	"github.com/ldsec/drynx/lib/provider/neutralizers"
 	"github.com/ldsec/drynx/lib/range"
 	"github.com/ldsec/drynx/services"
 	"github.com/ldsec/unlynx/lib"
@@ -34,7 +35,7 @@ func generateNodes(local *onet.LocalTest, nbrServers int, nbrDPs int, nbrVNs int
 	}
 	services.NewBuilder().
 		WithComputingNode().
-		WithDataProvider(loader).
+		WithDataProvider(loader, neutralizers.NewMinimumResultsSize(0)).
 		WithVerifyingNode().
 		Start()
 
@@ -729,7 +730,7 @@ func TestServiceDrynxLogisticRegression(t *testing.T) {
 	}
 	services.NewBuilder().
 		WithComputingNode().
-		WithDataProvider(loader).
+		WithDataProvider(loader, neutralizers.NewMinimumResultsSize(0)).
 		WithVerifyingNode().
 		Start()