Detect double PoST inclusion malfeasance (#6117)

## Motivation

With ATX merge, a dishonest smesher might try to include his PoST twice for doubled rewards:
- once in a merged ATX (published by another identity)
- once in a self-published ATX

activation/handler_v2.go

@@ -658,6 +658,7 @@ func (h *HandlerV2) checkMalicious(
 	tx *sql.Tx,
 	watx *wire.ActivationTxV2,
 	marrying []marriage,
+	ids []types.NodeID,
 ) error {
 	malicious, err := identities.IsMalicious(tx, watx.SmesherID)
 	if err != nil {
@@ -675,6 +676,14 @@ func (h *HandlerV2) checkMalicious(
 		return nil
 	}
 
+	malicious, err = h.checkDoublePost(ctx, tx, watx, ids)
+	if err != nil {
+		return fmt.Errorf("checking double post: %w", err)
+	}
+	if malicious {
+		return nil
+	}
+
 	// TODO(mafa): contextual validation:
 	// 1. check double-publish = ID contributed post to two ATXs in the same epoch
 	// 2. check previous ATX
@@ -705,6 +714,38 @@ func (h *HandlerV2) checkDoubleMarry(
 	return false, nil
 }
 
+func (h *HandlerV2) checkDoublePost(
+	ctx context.Context,
+	tx *sql.Tx,
+	atx *wire.ActivationTxV2,
+	ids []types.NodeID,
+) (bool, error) {
+	for _, id := range ids {
+		atxids, err := atxs.FindDoublePublish(tx, id, atx.PublishEpoch)
+		switch {
+		case errors.Is(err, sql.ErrNotFound):
+			continue
+		case err != nil:
+			return false, fmt.Errorf("searching for double publish: %w", err)
+		}
+		otherAtxId := slices.IndexFunc(atxids, func(other types.ATXID) bool { return other != atx.ID() })
+		otherAtx := atxids[otherAtxId]
+		h.logger.Debug(
+			"found ID that has already contributed its PoST in this epoch",
+			zap.Stringer("node_id", id),
+			zap.Stringer("atx_id", atx.ID()),
+			zap.Stringer("other_atx_id", otherAtx),
+			zap.Uint32("epoch", atx.PublishEpoch.Uint32()),
+		)
+		// TODO(mafa): finish proof
+		proof := &wire.ATXProof{
+			ProofType: wire.DoublePublish,
+		}
+		return true, h.malPublisher.Publish(ctx, id, proof)
+	}
+	return false, nil
+}
+
 // Store an ATX in the DB.
 func (h *HandlerV2) storeAtx(
 	ctx context.Context,
@@ -752,7 +793,7 @@ func (h *HandlerV2) storeAtx(
 		// TODO(mafa): don't store own ATX if it would mark the node as malicious
 		//    this probably needs to be done by validating and storing own ATXs eagerly and skipping validation in
 		//    the gossip handler (not sync!)
-		err := h.checkMalicious(ctx, tx, watx, marrying)
+		err := h.checkMalicious(ctx, tx, watx, marrying, maps.Keys(units))
 		if err != nil {
 			return fmt.Errorf("check malicious: %w", err)
 		}

activation/handler_v2_test.go

@@ -1730,6 +1730,64 @@ func Test_MarryingMalicious(t *testing.T) {
 	}
 }
 
+func TestContextualValidation_DoublePost(t *testing.T) {
+	t.Parallel()
+	golden := types.RandomATXID()
+	sig, err := signing.NewEdSigner()
+	require.NoError(t, err)
+
+	atxHandler := newV2TestHandler(t, golden)
+
+	// marry
+	otherSig, err := signing.NewEdSigner()
+	require.NoError(t, err)
+	othersAtx := atxHandler.createAndProcessInitial(t, otherSig)
+
+	mATX := newInitialATXv2(t, golden)
+	mATX.Marriages = []wire.MarriageCertificate{
+		{
+			Signature: sig.Sign(signing.MARRIAGE, sig.NodeID().Bytes()),
+		},
+		{
+			ReferenceAtx: othersAtx.ID(),
+			Signature:    otherSig.Sign(signing.MARRIAGE, sig.NodeID().Bytes()),
+		},
+	}
+	mATX.Sign(sig)
+
+	atxHandler.expectInitialAtxV2(mATX)
+	err = atxHandler.processATX(context.Background(), "", mATX, time.Now())
+	require.NoError(t, err)
+
+	// publish merged
+	merged := newSoloATXv2(t, mATX.PublishEpoch+2, mATX.ID(), mATX.ID())
+	post := wire.SubPostV2{
+		MarriageIndex: 1,
+		NumUnits:      othersAtx.TotalNumUnits(),
+		PrevATXIndex:  1,
+	}
+	merged.NiPosts[0].Posts = append(merged.NiPosts[0].Posts, post)
+
+	mATXID := mATX.ID()
+	merged.MarriageATX = &mATXID
+
+	merged.PreviousATXs = []types.ATXID{mATX.ID(), othersAtx.ID()}
+	merged.Sign(sig)
+
+	atxHandler.expectMergedAtxV2(merged, []types.NodeID{sig.NodeID(), otherSig.NodeID()}, []uint64{poetLeaves})
+	err = atxHandler.processATX(context.Background(), "", merged, time.Now())
+	require.NoError(t, err)
+
+	// The otherSig tries to publish alone in the same epoch.
+	// This is malfeasance as it tries include his PoST twice.
+	doubled := newSoloATXv2(t, merged.PublishEpoch, othersAtx.ID(), othersAtx.ID())
+	doubled.Sign(otherSig)
+	atxHandler.expectAtxV2(doubled)
+	atxHandler.mMalPublish.EXPECT().Publish(gomock.Any(), otherSig.NodeID(), gomock.Any())
+	err = atxHandler.processATX(context.Background(), "", doubled, time.Now())
+	require.NoError(t, err)
+}
+
 func Test_CalculatingUnits(t *testing.T) {
 	t.Parallel()
 	t.Run("units on 1 nipost must not overflow", func(t *testing.T) {

sql/atxs/atxs.go

@@ -891,6 +891,38 @@ func Units(db sql.Executor, atxID types.ATXID, nodeID types.NodeID) (uint32, err
 	return units, err
 }
 
+// FindDoublePublish finds 2 distinct ATXIDs that the given identity contributed PoST to in the given epoch.
+//
+// It is guaranteed to return 2 distinct ATXs when the error is nil.
+// It works by finding an ATX in the given epoch that has a PoST contribution from the given identity.
+// - `epoch` is looked up in the `atxs` table by matching atxid.
+func FindDoublePublish(db sql.Executor, nodeID types.NodeID, epoch types.EpochID) ([]types.ATXID, error) {
+	var ids []types.ATXID
+	rows, err := db.Exec(`
+		SELECT p.atxid
+		FROM posts p
+		INNER JOIN atxs a ON p.atxid = a.id
+		WHERE p.pubkey = ?1 AND a.epoch = ?2;`,
+		func(stmt *sql.Statement) {
+			stmt.BindBytes(1, nodeID.Bytes())
+			stmt.BindInt64(2, int64(epoch))
+		},
+		func(stmt *sql.Statement) bool {
+			var id types.ATXID
+			stmt.ColumnBytes(0, id[:])
+			ids = append(ids, id)
+			return len(ids) < 2
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+	if rows != 2 {
+		return nil, sql.ErrNotFound
+	}
+	return ids, nil
+}
+
 func AllUnits(db sql.Executor, id types.ATXID) (map[types.NodeID]uint32, error) {
 	units := make(map[types.NodeID]uint32)
 	rows, err := db.Exec(

sql/atxs/atxs_test.go

@@ -1201,3 +1201,80 @@ func Test_AtxWithPrevious(t *testing.T) {
 		require.Equal(t, atx2.ID(), id)
 	})
 }
+
+func Test_FindDoublePublish(t *testing.T) {
+	t.Parallel()
+	sig, err := signing.NewEdSigner()
+	require.NoError(t, err)
+	t.Run("no atxs", func(t *testing.T) {
+		t.Parallel()
+		db := sql.InMemory()
+		_, err := atxs.FindDoublePublish(db, types.RandomNodeID(), 0)
+		require.ErrorIs(t, err, sql.ErrNotFound)
+	})
+
+	t.Run("no double publish", func(t *testing.T) {
+		t.Parallel()
+		db := sql.InMemory()
+
+		// one atx
+		atx0, blob := newAtx(t, sig, withPublishEpoch(1))
+		require.NoError(t, atxs.Add(db, atx0, blob))
+		require.NoError(t, atxs.SetUnits(db, atx0.ID(), atx0.SmesherID, 10))
+
+		_, err = atxs.FindDoublePublish(db, atx0.SmesherID, atx0.PublishEpoch)
+		require.ErrorIs(t, err, sql.ErrNotFound)
+
+		// two atxs in different epochs
+		atx1, blob := newAtx(t, sig, withPublishEpoch(atx0.PublishEpoch+1))
+		require.NoError(t, atxs.Add(db, atx1, blob))
+		require.NoError(t, atxs.SetUnits(db, atx1.ID(), atx0.SmesherID, 10))
+
+		_, err = atxs.FindDoublePublish(db, atx0.SmesherID, atx0.PublishEpoch)
+		require.ErrorIs(t, err, sql.ErrNotFound)
+	})
+	t.Run("double publish", func(t *testing.T) {
+		t.Parallel()
+		db := sql.InMemory()
+
+		atx0, blob := newAtx(t, sig)
+		require.NoError(t, atxs.Add(db, atx0, blob))
+		require.NoError(t, atxs.SetUnits(db, atx0.ID(), atx0.SmesherID, 10))
+
+		atx1, blob := newAtx(t, sig)
+		require.NoError(t, atxs.Add(db, atx1, blob))
+		require.NoError(t, atxs.SetUnits(db, atx1.ID(), atx0.SmesherID, 10))
+
+		atxids, err := atxs.FindDoublePublish(db, atx0.SmesherID, atx0.PublishEpoch)
+		require.NoError(t, err)
+		require.ElementsMatch(t, []types.ATXID{atx0.ID(), atx1.ID()}, atxids)
+
+		// filters by epoch
+		_, err = atxs.FindDoublePublish(db, atx0.SmesherID, atx0.PublishEpoch+1)
+		require.ErrorIs(t, err, sql.ErrNotFound)
+	})
+	t.Run("double publish different smesher", func(t *testing.T) {
+		t.Parallel()
+		db := sql.InMemory()
+
+		atx0Signer, err := signing.NewEdSigner()
+		require.NoError(t, err)
+
+		atx0, blob := newAtx(t, atx0Signer)
+		require.NoError(t, atxs.Add(db, atx0, blob))
+		require.NoError(t, atxs.SetUnits(db, atx0.ID(), atx0.SmesherID, 10))
+		require.NoError(t, atxs.SetUnits(db, atx0.ID(), sig.NodeID(), 10))
+
+		atx1Signer, err := signing.NewEdSigner()
+		require.NoError(t, err)
+
+		atx1, blob := newAtx(t, atx1Signer)
+		require.NoError(t, atxs.Add(db, atx1, blob))
+		require.NoError(t, atxs.SetUnits(db, atx1.ID(), atx1.SmesherID, 10))
+		require.NoError(t, atxs.SetUnits(db, atx1.ID(), sig.NodeID(), 10))
+
+		atxIDs, err := atxs.FindDoublePublish(db, sig.NodeID(), atx0.PublishEpoch)
+		require.NoError(t, err)
+		require.ElementsMatch(t, []types.ATXID{atx0.ID(), atx1.ID()}, atxIDs)
+	})
+}