Merge pull request #1427 from nspcc-dev/oracle/module

network: implement Oracle module

cli/executor_test.go

@@ -80,7 +80,7 @@ func newTestChain(t *testing.T, f func(*config.Config)) (*core.Blockchain, *serv
 	netSrv, err := network.NewServer(serverConfig, chain, zap.NewNop())
 	require.NoError(t, err)
 	go netSrv.Start(make(chan error, 1))
-	rpcServer := server.New(chain, cfg.ApplicationConfiguration.RPC, netSrv, logger)
+	rpcServer := server.New(chain, cfg.ApplicationConfiguration.RPC, netSrv, nil, logger)
 	errCh := make(chan error, 2)
 	rpcServer.Start(errCh)
 

cli/server/server.go

@@ -319,7 +319,7 @@ func startServer(ctx *cli.Context) error {
 	if err != nil {
 		return cli.NewExitError(fmt.Errorf("failed to create network server: %w", err), 1)
 	}
-	rpcServer := server.New(chain, cfg.ApplicationConfiguration.RPC, serv, log)
+	rpcServer := server.New(chain, cfg.ApplicationConfiguration.RPC, serv, serv.GetOracle(), log)
 	errChan := make(chan error)
 
 	go serv.Start(errChan)

config/protocol.mainnet.yml

@@ -48,6 +48,9 @@ ApplicationConfiguration:
   MaxPeers: 100
   AttemptConnPeers: 20
   MinPeers: 5
+  Oracle:
+    Enabled: false
+
   RPC:
     Enabled: true
     MaxGasInvoke: 15

config/protocol.privnet.docker.four.yml

@@ -44,6 +44,17 @@ ApplicationConfiguration:
   MaxPeers: 10
   AttemptConnPeers: 5
   MinPeers: 3
+  Oracle:
+    Enabled: false
+    Nodes:
+      - 172.200.0.1:30333
+      - 172.200.0.2:30334
+      - 172.200.0.3:30335
+      - 172.200.0.4:30336
+    RequestTimeout: 5s
+    UnlockWallet:
+      Path: "/wallet4.json"
+      Password: "four"
   RPC:
     Enabled: true
     MaxGasInvoke: 15

config/protocol.privnet.docker.one.yml

@@ -44,6 +44,17 @@ ApplicationConfiguration:
   MaxPeers: 10
   AttemptConnPeers: 5
   MinPeers: 3
+  Oracle:
+    Enabled: false
+    Nodes:
+      - 172.200.0.1:30333
+      - 172.200.0.2:30334
+      - 172.200.0.3:30335
+      - 172.200.0.4:30336
+    RequestTimeout: 5s
+    UnlockWallet:
+      Path: "/wallet1.json"
+      Password: "one"
   RPC:
     Enabled: true
     MaxGasInvoke: 15

config/protocol.privnet.docker.single.yml

@@ -38,6 +38,14 @@ ApplicationConfiguration:
   MaxPeers: 10
   AttemptConnPeers: 5
   MinPeers: 0
+  Oracle:
+    Enabled: false
+    Nodes:
+      - 172.200.0.1:30333
+    RequestTimeout: 5s
+    UnlockWallet:
+      Path: "/wallet1_solo.json"
+      Password: "one"
   RPC:
     Enabled: true
     EnableCORSWorkaround: false

config/protocol.privnet.docker.three.yml

@@ -44,6 +44,17 @@ ApplicationConfiguration:
   MaxPeers: 10
   AttemptConnPeers: 5
   MinPeers: 3
+  Oracle:
+    Enabled: false
+    Nodes:
+      - 172.200.0.1:30333
+      - 172.200.0.2:30334
+      - 172.200.0.3:30335
+      - 172.200.0.4:30336
+    RequestTimeout: 5s
+    UnlockWallet:
+      Path: "/wallet3.json"
+      Password: "three"
   RPC:
     Enabled: true
     MaxGasInvoke: 15

config/protocol.privnet.docker.two.yml

@@ -44,6 +44,17 @@ ApplicationConfiguration:
   MaxPeers: 10
   AttemptConnPeers: 5
   MinPeers: 3
+  Oracle:
+    Enabled: false
+    Nodes:
+      - 172.200.0.1:30333
+      - 172.200.0.2:30334
+      - 172.200.0.3:30335
+      - 172.200.0.4:30336
+    RequestTimeout: 5s
+    UnlockWallet:
+      Path: "/wallet2.json"
+      Password: "two"
   RPC:
     Enabled: true
     MaxGasInvoke: 15

config/protocol.testnet.yml

@@ -48,6 +48,9 @@ ApplicationConfiguration:
   MaxPeers: 100
   AttemptConnPeers: 20
   MinPeers: 5
+  Oracle:
+    Enabled: false
+
   RPC:
     Enabled: true
     MaxGasInvoke: 15

go.mod

@@ -1,6 +1,7 @@
 module github.com/nspcc-dev/neo-go
 
 require (
+	github.com/PaesslerAG/jsonpath v0.1.1
 	github.com/Workiva/go-datastructures v1.0.50
 	github.com/abiosoft/readline v0.0.0-20180607040430-155bce2042db
 	github.com/alicebob/miniredis v2.5.0+incompatible

go.sum

@@ -8,6 +8,11 @@ github.com/DataDog/zstd v1.4.1 h1:3oxKN3wbHibqx897utPC2LTQU4J+IHWWJO+glkAkpFM=
 github.com/DataDog/zstd v1.4.1/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo=
 github.com/OneOfOne/xxhash v1.2.2 h1:KMrpdQIwFcEqXDklaen+P1axHaj9BSKzvpUUfnHldSE=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
+github.com/PaesslerAG/gval v1.0.0 h1:GEKnRwkWDdf9dOmKcNrar9EA1bz1z9DqPIO1+iLzhd8=
+github.com/PaesslerAG/gval v1.0.0/go.mod h1:y/nm5yEyTeX6av0OfKJNp9rBNj2XrGhAf5+v24IBN1I=
+github.com/PaesslerAG/jsonpath v0.1.0/go.mod h1:4BzmtoM/PI8fPO4aQGIusjGxGir2BzcV0grWtFzq1Y8=
+github.com/PaesslerAG/jsonpath v0.1.1 h1:c1/AToHQMVsduPAa4Vh6xp2U0evy4t8SWp8imEsylIk=
+github.com/PaesslerAG/jsonpath v0.1.1/go.mod h1:lVboNxFGal/VwW6d9JzIy56bUsYAP6tH/x80vjnCseY=
 github.com/Workiva/go-datastructures v1.0.50 h1:slDmfW6KCHcC7U+LP3DDBbm4fqTwZGn1beOFPfGaLvo=
 github.com/Workiva/go-datastructures v1.0.50/go.mod h1:Z+F2Rca0qCsVYDS8z7bAGm8f3UkzuWYS/oBZz5a7VVA=
 github.com/abiosoft/ishell v2.0.0+incompatible h1:zpwIuEHc37EzrsIYah3cpevrIc8Oma7oZPxr03tlmmw=

pkg/config/application_config.go

@@ -26,4 +26,5 @@ type ApplicationConfiguration struct {
 	Relay             bool                    `yaml:"Relay"`
 	RPC               rpc.Config              `yaml:"RPC"`
 	UnlockWallet      Wallet                  `yaml:"UnlockWallet"`
+	Oracle            OracleConfiguration     `yaml:"Oracle"`
 }

pkg/config/oracle_config.go

@@ -0,0 +1,16 @@
+package config
+
+import "time"
+
+// OracleConfiguration is a config for the oracle module.
+type OracleConfiguration struct {
+	Enabled               bool          `yaml:"Enabled"`
+	AllowPrivateHost      bool          `yaml:"AllowPrivateHost"`
+	Nodes                 []string      `yaml:"Nodes"`
+	MaxTaskTimeout        time.Duration `yaml:"MaxTaskTimeout"`
+	RefreshInterval       time.Duration `yaml:"RefreshInterval"`
+	MaxConcurrentRequests int           `yaml:"MaxConcurrentRequests"`
+	RequestTimeout        time.Duration `yaml:"RequestTimeout"`
+	ResponseTimeout       time.Duration `yaml:"ResponseTimeout"`
+	UnlockWallet          Wallet        `yaml:"UnlockWallet"`
+}

pkg/core/blockchain.go

@@ -14,6 +14,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/config"
 	"github.com/nspcc-dev/neo-go/pkg/core/block"
 	"github.com/nspcc-dev/neo-go/pkg/core/blockchainer"
+	"github.com/nspcc-dev/neo-go/pkg/core/blockchainer/services"
 	"github.com/nspcc-dev/neo-go/pkg/core/dao"
 	"github.com/nspcc-dev/neo-go/pkg/core/interop"
 	"github.com/nspcc-dev/neo-go/pkg/core/interop/contract"
@@ -34,7 +35,6 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/smartcontract/trigger"
 	"github.com/nspcc-dev/neo-go/pkg/util"
 	"github.com/nspcc-dev/neo-go/pkg/vm"
-	"github.com/nspcc-dev/neo-go/pkg/vm/emit"
 	"github.com/nspcc-dev/neo-go/pkg/vm/stackitem"
 	"go.uber.org/zap"
 )
@@ -192,6 +192,13 @@ func NewBlockchain(s storage.Store, cfg config.ProtocolConfiguration, log *zap.L
 	return bc, nil
 }
 
+// SetOracle sets oracle module. It doesn't protected by mutex and
+// must be called before `bc.Run()` to avoid data race.
+func (bc *Blockchain) SetOracle(mod services.Oracle) {
+	bc.contracts.Oracle.Module.Store(mod)
+	bc.contracts.Designate.OracleService.Store(mod)
+}
+
 func (bc *Blockchain) init() error {
 	// If we could not find the version in the Store, we know that there is nothing stored.
 	ver, err := bc.dao.GetVersion()
@@ -1654,6 +1661,7 @@ var (
 
 // initVerificationVM initializes VM for witness check.
 func (bc *Blockchain) initVerificationVM(ic *interop.Context, hash util.Uint160, witness *transaction.Witness) error {
+	isNative := false
 	v := ic.VM
 	if len(witness.VerificationScript) != 0 {
 		if witness.ScriptHash() != hash {
@@ -1677,18 +1685,22 @@ func (bc *Blockchain) initVerificationVM(ic *interop.Context, hash util.Uint160,
 		v.Context().NEF = &cs.NEF
 		v.Jump(v.Context(), md.Offset)
 
-		if cs.ID <= 0 {
+		isNative = cs.ID <= 0
-			w := io.NewBufBinWriter()
+		if !isNative && initMD != nil {
-			emit.String(w.BinWriter, manifest.MethodVerify)
-			if w.Err != nil {
-				return w.Err
-			}
-			v.LoadScript(w.Bytes())
-		} else if initMD != nil {
 			v.Call(v.Context(), initMD.Offset)
 		}
 	}
-	v.LoadScript(witness.InvocationScript)
+	if len(witness.InvocationScript) != 0 {
+		v.LoadScript(witness.InvocationScript)
+		if isNative {
+			if err := v.StepOut(); err != nil {
+				return err
+			}
+		}
+	}
+	if isNative {
+		v.Estack().PushVal(manifest.MethodVerify)
+	}
 	return nil
 }
 

pkg/core/blockchain_test.go

@@ -973,7 +973,6 @@ func TestVerifyTx(t *testing.T) {
 				transaction.NotaryServiceFeePerKey + // fee for Notary attribute
 				fee.Opcode(bc.GetBaseExecFee(), // Notary verification script
 					opcode.PUSHDATA1, opcode.RET, // invocation script
-					opcode.PUSHDATA1, opcode.RET, // arguments for native verification call
 					opcode.PUSHINT8, opcode.SYSCALL, opcode.RET) + // Neo.Native.Call
 				native.NotaryVerificationPrice // Notary witness verification price
 			tx.Scripts = []transaction.Witness{

pkg/core/blockchainer/blockchainer.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/nspcc-dev/neo-go/pkg/config"
 	"github.com/nspcc-dev/neo-go/pkg/core/block"
+	"github.com/nspcc-dev/neo-go/pkg/core/blockchainer/services"
 	"github.com/nspcc-dev/neo-go/pkg/core/mempool"
 	"github.com/nspcc-dev/neo-go/pkg/core/state"
 	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
@@ -57,6 +58,7 @@ type Blockchainer interface {
 	GetStorageItems(id int32) (map[string]*state.StorageItem, error)
 	GetTestVM(t trigger.Type, tx *transaction.Transaction, b *block.Block) *vm.VM
 	GetTransaction(util.Uint256) (*transaction.Transaction, uint32, error)
+	SetOracle(service services.Oracle)
 	mempool.Feer // fee interface
 	ManagementContractHash() util.Uint160
 	PoolTx(t *transaction.Transaction, pools ...*mempool.Pool) error

pkg/core/blockchainer/services/oracle.go

@@ -0,0 +1,20 @@
+package services
+
+import (
+	"github.com/nspcc-dev/neo-go/pkg/core/state"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
+)
+
+// Oracle specifies oracle service interface.
+type Oracle interface {
+	// AddRequests processes new requests.
+	AddRequests(map[uint64]*state.OracleRequest)
+	// RemoveRequests removes already processed requests.
+	RemoveRequests([]uint64)
+	// UpdateOracleNodes updates oracle nodes.
+	UpdateOracleNodes(keys.PublicKeys)
+	// Run runs oracle module. Must be invoked in a separate goroutine.
+	Run()
+	// Shutdown shutdowns oracle module.
+	Shutdown()
+}

pkg/core/helper_test.go

@@ -51,6 +51,12 @@ func newTestChainWithCustomCfg(t *testing.T, f func(*config.Config)) *Blockchain
 }
 
 func newTestChainWithCustomCfgAndStore(t *testing.T, st storage.Store, f func(*config.Config)) *Blockchain {
+	chain := initTestChain(t, st, f)
+	go chain.Run()
+	return chain
+}
+
+func initTestChain(t *testing.T, st storage.Store, f func(*config.Config)) *Blockchain {
 	unitTestNetCfg, err := config.Load("../../config", testchain.Network())
 	require.NoError(t, err)
 	if f != nil {
@@ -61,7 +67,6 @@ func newTestChainWithCustomCfgAndStore(t *testing.T, st storage.Store, f func(*c
 	}
 	chain, err := NewBlockchain(st, unitTestNetCfg.ProtocolConfiguration, zaptest.NewLogger(t))
 	require.NoError(t, err)
-	go chain.Run()
 	return chain
 }
 

pkg/core/native/designate.go

@@ -7,6 +7,7 @@ import (
 	"sort"
 	"sync/atomic"
 
+	"github.com/nspcc-dev/neo-go/pkg/core/blockchainer/services"
 	"github.com/nspcc-dev/neo-go/pkg/core/dao"
 	"github.com/nspcc-dev/neo-go/pkg/core/interop"
 	"github.com/nspcc-dev/neo-go/pkg/core/interop/runtime"
@@ -32,6 +33,8 @@ type Designate struct {
 
 	// p2pSigExtensionsEnabled defines whether the P2P signature extensions logic is relevant.
 	p2pSigExtensionsEnabled bool
+
+	OracleService atomic.Value
 }
 
 type oraclesData struct {
@@ -117,6 +120,9 @@ func (s *Designate) PostPersist(ic *interop.Context) error {
 		height: height,
 	}
 	s.oracles.Store(od)
+	if orc, _ := s.OracleService.Load().(services.Oracle); orc != nil {
+		orc.UpdateOracleNodes(od.nodes.Copy())
+	}
 	s.rolesChangedFlag.Store(false)
 	return nil
 }

pkg/core/native/oracle.go

@@ -7,7 +7,9 @@ import (
 	"math"
 	"math/big"
 	"strings"
+	"sync/atomic"
 
+	"github.com/nspcc-dev/neo-go/pkg/core/blockchainer/services"
 	"github.com/nspcc-dev/neo-go/pkg/core/dao"
 	"github.com/nspcc-dev/neo-go/pkg/core/interop"
 	"github.com/nspcc-dev/neo-go/pkg/core/interop/contract"
@@ -36,6 +38,11 @@ type Oracle struct {
 
 	Desig        *Designate
 	oracleScript []byte
+
+	// Module is an oracle module capable of talking with the external world.
+	Module atomic.Value
+	// newRequests contains new requests created during current block.
+	newRequests map[uint64]*state.OracleRequest
 }
 
 const (
@@ -69,8 +76,7 @@ func newOracle() *Oracle {
 	o := &Oracle{ContractMD: *interop.NewContractMD(nativenames.Oracle, oracleContractID)}
 
 	w := io.NewBufBinWriter()
-	emit.Int(w.BinWriter, 0)
+	emit.Opcodes(w.BinWriter, opcode.NEWARRAY0)
-	emit.Opcodes(w.BinWriter, opcode.NEWARRAY)
 	emit.Int(w.BinWriter, int64(callflag.All))
 	emit.String(w.BinWriter, "finish")
 	emit.Bytes(w.BinWriter, o.Hash.BytesBE())
@@ -113,7 +119,11 @@ func (o *Oracle) GetOracleResponseScript() []byte {
 
 // OnPersist implements Contract interface.
 func (o *Oracle) OnPersist(ic *interop.Context) error {
-	return nil
+	var err error
+	if o.newRequests == nil {
+		o.newRequests, err = o.getRequests(ic.DAO)
+	}
+	return err
 }
 
 // PostPersist represents `postPersist` method.
@@ -121,6 +131,9 @@ func (o *Oracle) PostPersist(ic *interop.Context) error {
 	var nodes keys.PublicKeys
 	var reward []big.Int
 	single := new(big.Int).SetUint64(oracleRequestPrice)
+	var removedIDs []uint64
+
+	orc, _ := o.Module.Load().(services.Oracle)
 	for _, tx := range ic.Block.Transactions {
 		resp := getResponse(tx)
 		if resp == nil {
@@ -134,6 +147,9 @@ func (o *Oracle) PostPersist(ic *interop.Context) error {
 		if err := ic.DAO.DeleteStorageItem(o.ContractID, reqKey); err != nil {
 			return err
 		}
+		if orc != nil {
+			removedIDs = append(removedIDs, resp.ID)
+		}
 
 		idKey := makeIDListKey(req.URL)
 		idList := new(IDList)
@@ -171,7 +187,11 @@ func (o *Oracle) PostPersist(ic *interop.Context) error {
 	for i := range reward {
 		o.GAS.mint(ic, nodes[i].GetScriptHash(), &reward[i], false)
 	}
-	return nil
+
+	if len(removedIDs) != 0 && orc != nil {
+		orc.RemoveRequests(removedIDs)
+	}
+	return o.updateCache(ic.DAO)
 }
 
 // Metadata returns contract metadata.
@@ -339,6 +359,7 @@ func (o *Oracle) PutRequestInternal(id uint64, req *state.OracleRequest, d dao.D
 	if err := d.PutStorageItem(o.ContractID, reqKey, reqItem); err != nil {
 		return err
 	}
+	o.newRequests[id] = req
 
 	// Add request ID to the id list.
 	lst := new(IDList)
@@ -394,6 +415,29 @@ func (o *Oracle) getOriginalTxID(d dao.DAO, tx *transaction.Transaction) util.Ui
 	return tx.Hash()
 }
 
+// getRequests returns all requests which have not been finished yet.
+func (o *Oracle) getRequests(d dao.DAO) (map[uint64]*state.OracleRequest, error) {
+	m, err := d.GetStorageItemsWithPrefix(o.ContractID, prefixRequest)
+	if err != nil {
+		return nil, err
+	}
+	reqs := make(map[uint64]*state.OracleRequest, len(m))
+	for k, si := range m {
+		if len(k) != 8 {
+			return nil, errors.New("invalid request ID")
+		}
+		r := io.NewBinReaderFromBuf(si.Value)
+		req := new(state.OracleRequest)
+		req.DecodeBinary(r)
+		if r.Err != nil {
+			return nil, r.Err
+		}
+		id := binary.LittleEndian.Uint64([]byte(k))
+		reqs[id] = req
+	}
+	return reqs, nil
+}
+
 func makeRequestKey(id uint64) []byte {
 	k := make([]byte, 9)
 	k[0] = prefixRequest[0]
@@ -408,3 +452,22 @@ func makeIDListKey(url string) []byte {
 func (o *Oracle) getSerializableFromDAO(d dao.DAO, key []byte, item io.Serializable) error {
 	return getSerializableFromDAO(o.ContractID, d, key, item)
 }
+
+// updateCache updates cached Oracle values if they've been changed
+func (o *Oracle) updateCache(d dao.DAO) error {
+	orc, _ := o.Module.Load().(services.Oracle)
+	if orc == nil {
+		return nil
+	}
+
+	reqs := o.newRequests
+	o.newRequests = make(map[uint64]*state.OracleRequest)
+	for id := range reqs {
+		key := makeRequestKey(id)
+		if si := d.GetStorageItem(o.ContractID, key); si == nil { // tx has failed
+			delete(reqs, id)
+		}
+	}
+	orc.AddRequests(reqs)
+	return nil
+}

pkg/core/oracle_test.go

@@ -0,0 +1,388 @@
+package core
+
+import (
+	"bytes"
+	"errors"
+	gio "io"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"path"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nspcc-dev/neo-go/pkg/config"
+	"github.com/nspcc-dev/neo-go/pkg/config/netmode"
+	"github.com/nspcc-dev/neo-go/pkg/core/native"
+	"github.com/nspcc-dev/neo-go/pkg/core/state"
+	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
+	"github.com/nspcc-dev/neo-go/pkg/services/oracle"
+	"github.com/nspcc-dev/neo-go/pkg/util"
+	"github.com/nspcc-dev/neo-go/pkg/wallet"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zaptest"
+)
+
+const oracleModulePath = "../services/oracle/"
+
+func getOracleConfig(t *testing.T, bc *Blockchain, w, pass string) oracle.Config {
+	return oracle.Config{
+		Log:     zaptest.NewLogger(t),
+		Network: netmode.UnitTestNet,
+		MainCfg: config.OracleConfiguration{
+			RefreshInterval: time.Second,
+			UnlockWallet: config.Wallet{
+				Path:     path.Join(oracleModulePath, w),
+				Password: pass,
+			},
+		},
+		Chain:          bc,
+		Client:         newDefaultHTTPClient(),
+		OracleScript:   bc.contracts.Oracle.NEF.Script,
+		OracleResponse: bc.contracts.Oracle.GetOracleResponseScript(),
+		OracleHash:     bc.contracts.Oracle.Hash,
+	}
+}
+
+func getTestOracle(t *testing.T, bc *Blockchain, walletPath, pass string) (
+	*wallet.Account,
+	*oracle.Oracle,
+	map[uint64]*responseWithSig,
+	chan *transaction.Transaction) {
+	m := make(map[uint64]*responseWithSig)
+	ch := make(chan *transaction.Transaction, 5)
+	orcCfg := getOracleConfig(t, bc, walletPath, pass)
+	orcCfg.ResponseHandler = saveToMapBroadcaster{m}
+	orcCfg.OnTransaction = saveTxToChan(ch)
+	orcCfg.URIValidator = func(u *url.URL) error {
+		if strings.HasPrefix(u.Host, "private") {
+			return errors.New("private network")
+		}
+		return nil
+	}
+	orc, err := oracle.NewOracle(orcCfg)
+	require.NoError(t, err)
+
+	w, err := wallet.NewWalletFromFile(path.Join(oracleModulePath, walletPath))
+	require.NoError(t, err)
+	require.NoError(t, w.Accounts[0].Decrypt(pass))
+	return w.Accounts[0], orc, m, ch
+}
+
+// Compatibility test from C# code.
+// https://github.com/neo-project/neo-modules/blob/master/tests/Neo.Plugins.OracleService.Tests/UT_OracleService.cs#L61
+func TestCreateResponseTx(t *testing.T) {
+	bc := newTestChain(t)
+	defer bc.Close()
+
+	require.Equal(t, int64(30), bc.GetBaseExecFee())
+	require.Equal(t, int64(1000), bc.FeePerByte())
+	acc, orc, _, _ := getTestOracle(t, bc, "./testdata/oracle1.json", "one")
+	req := &state.OracleRequest{
+		OriginalTxID:     util.Uint256{},
+		GasForResponse:   100000000,
+		URL:              "https://127.0.0.1/test",
+		Filter:           new(string),
+		CallbackContract: util.Uint160{},
+		CallbackMethod:   "callback",
+		UserData:         []byte{},
+	}
+	resp := &transaction.OracleResponse{
+		ID:     1,
+		Code:   transaction.Success,
+		Result: []byte{0},
+	}
+	require.NoError(t, bc.contracts.Oracle.PutRequestInternal(1, req, bc.dao))
+	orc.UpdateOracleNodes(keys.PublicKeys{acc.PrivateKey().PublicKey()})
+	tx, err := orc.CreateResponseTx(int64(req.GasForResponse), 1, resp)
+	require.NoError(t, err)
+	assert.Equal(t, 167, tx.Size())
+	assert.Equal(t, int64(2216640), tx.NetworkFee)
+	assert.Equal(t, int64(97783360), tx.SystemFee)
+}
+
+func TestOracle_InvalidWallet(t *testing.T) {
+	bc := newTestChain(t)
+
+	_, err := oracle.NewOracle(getOracleConfig(t, bc, "./testdata/oracle1.json", "invalid"))
+	require.Error(t, err)
+
+	_, err = oracle.NewOracle(getOracleConfig(t, bc, "./testdata/oracle1.json", "one"))
+	require.NoError(t, err)
+}
+
+func TestOracle(t *testing.T) {
+	bc := newTestChain(t)
+	defer bc.Close()
+
+	oracleCtr := bc.contracts.Oracle
+	acc1, orc1, m1, ch1 := getTestOracle(t, bc, "./testdata/oracle1.json", "one")
+	acc2, orc2, m2, ch2 := getTestOracle(t, bc, "./testdata/oracle2.json", "two")
+	oracleNodes := keys.PublicKeys{acc1.PrivateKey().PublicKey(), acc2.PrivateKey().PublicKey()}
+	// Must be set in native contract for tx verification.
+	bc.setNodesByRole(t, true, native.RoleOracle, oracleNodes)
+	orc1.UpdateOracleNodes(oracleNodes.Copy())
+	orc2.UpdateOracleNodes(oracleNodes.Copy())
+
+	cs := getOracleContractState(bc.contracts.Oracle.Hash)
+	require.NoError(t, bc.contracts.Management.PutContractState(bc.dao, cs))
+
+	putOracleRequest(t, cs.Hash, bc, "http://get.1234", nil, "handle", []byte{}, 10_000_000)
+	putOracleRequest(t, cs.Hash, bc, "http://get.1234", nil, "handle", []byte{}, 10_000_000)
+	putOracleRequest(t, cs.Hash, bc, "http://get.timeout", nil, "handle", []byte{}, 10_000_000)
+	putOracleRequest(t, cs.Hash, bc, "http://get.notfound", nil, "handle", []byte{}, 10_000_000)
+	putOracleRequest(t, cs.Hash, bc, "http://get.forbidden", nil, "handle", []byte{}, 10_000_000)
+	putOracleRequest(t, cs.Hash, bc, "http://private.url", nil, "handle", []byte{}, 10_000_000)
+	putOracleRequest(t, cs.Hash, bc, "http://get.big", nil, "handle", []byte{}, 10_000_000)
+	putOracleRequest(t, cs.Hash, bc, "http://get.maxallowed", nil, "handle", []byte{}, 10_000_000)
+	putOracleRequest(t, cs.Hash, bc, "http://get.maxallowed", nil, "handle", []byte{}, 100_000_000)
+
+	flt := "Values[1]"
+	putOracleRequest(t, cs.Hash, bc, "http://get.filter", &flt, "handle", []byte{}, 10_000_000)
+	putOracleRequest(t, cs.Hash, bc, "http://get.filterinv", &flt, "handle", []byte{}, 10_000_000)
+
+	checkResp := func(t *testing.T, id uint64, resp *transaction.OracleResponse) *state.OracleRequest {
+		req, err := oracleCtr.GetRequestInternal(bc.dao, id)
+		require.NoError(t, err)
+
+		reqs := map[uint64]*state.OracleRequest{id: req}
+		orc1.ProcessRequestsInternal(reqs)
+		require.NotNil(t, m1[id])
+		require.Equal(t, resp, m1[id].resp)
+		require.Empty(t, ch1)
+		return req
+	}
+
+	// Checks if tx is ready and valid.
+	checkEmitTx := func(t *testing.T, ch chan *transaction.Transaction) {
+		require.Len(t, ch, 1)
+		tx := <-ch
+		require.NoError(t, bc.verifyAndPoolTx(tx, bc.GetMemPool(), bc))
+	}
+
+	t.Run("NormalRequest", func(t *testing.T) {
+		resp := &transaction.OracleResponse{
+			ID:     1,
+			Code:   transaction.Success,
+			Result: []byte{1, 2, 3, 4},
+		}
+		req := checkResp(t, 1, resp)
+
+		reqs := map[uint64]*state.OracleRequest{1: req}
+		orc2.ProcessRequestsInternal(reqs)
+		require.Equal(t, resp, m2[1].resp)
+		require.Empty(t, ch2)
+
+		t.Run("InvalidSignature", func(t *testing.T) {
+			orc1.AddResponse(acc2.PrivateKey().PublicKey(), m2[1].resp.ID, []byte{1, 2, 3})
+			require.Empty(t, ch1)
+		})
+		orc1.AddResponse(acc2.PrivateKey().PublicKey(), m2[1].resp.ID, m2[1].txSig)
+		checkEmitTx(t, ch1)
+
+		t.Run("FirstOtherThenMe", func(t *testing.T) {
+			const reqID = 2
+
+			resp := &transaction.OracleResponse{
+				ID:     reqID,
+				Code:   transaction.Success,
+				Result: []byte{1, 2, 3, 4},
+			}
+			req := checkResp(t, reqID, resp)
+			orc2.AddResponse(acc1.PrivateKey().PublicKey(), reqID, m1[reqID].txSig)
+			require.Empty(t, ch2)
+
+			reqs := map[uint64]*state.OracleRequest{reqID: req}
+			orc2.ProcessRequestsInternal(reqs)
+			require.Equal(t, resp, m2[reqID].resp)
+			checkEmitTx(t, ch2)
+		})
+	})
+	t.Run("Invalid", func(t *testing.T) {
+		t.Run("Timeout", func(t *testing.T) {
+			checkResp(t, 3, &transaction.OracleResponse{
+				ID:   3,
+				Code: transaction.Timeout,
+			})
+		})
+		t.Run("NotFound", func(t *testing.T) {
+			checkResp(t, 4, &transaction.OracleResponse{
+				ID:   4,
+				Code: transaction.NotFound,
+			})
+		})
+		t.Run("Forbidden", func(t *testing.T) {
+			checkResp(t, 5, &transaction.OracleResponse{
+				ID:   5,
+				Code: transaction.Forbidden,
+			})
+		})
+		t.Run("PrivateNetwork", func(t *testing.T) {
+			checkResp(t, 6, &transaction.OracleResponse{
+				ID:   6,
+				Code: transaction.Forbidden,
+			})
+		})
+		t.Run("Big", func(t *testing.T) {
+			checkResp(t, 7, &transaction.OracleResponse{
+				ID:   7,
+				Code: transaction.ResponseTooLarge,
+			})
+		})
+		t.Run("MaxAllowedSmallGAS", func(t *testing.T) {
+			checkResp(t, 8, &transaction.OracleResponse{
+				ID:   8,
+				Code: transaction.InsufficientFunds,
+			})
+		})
+	})
+	t.Run("MaxAllowedEnoughGAS", func(t *testing.T) {
+		checkResp(t, 9, &transaction.OracleResponse{
+			ID:     9,
+			Code:   transaction.Success,
+			Result: make([]byte, transaction.MaxOracleResultSize),
+		})
+	})
+	t.Run("WithFilter", func(t *testing.T) {
+		checkResp(t, 10, &transaction.OracleResponse{
+			ID:     10,
+			Code:   transaction.Success,
+			Result: []byte(`[2]`),
+		})
+		t.Run("invalid response", func(t *testing.T) {
+			checkResp(t, 11, &transaction.OracleResponse{
+				ID:   11,
+				Code: transaction.Error,
+			})
+		})
+	})
+}
+
+func TestOracleFull(t *testing.T) {
+	bc := initTestChain(t, nil, nil)
+	acc, orc, _, _ := getTestOracle(t, bc, "./testdata/oracle2.json", "two")
+	mp := bc.GetMemPool()
+	orc.OnTransaction = func(tx *transaction.Transaction) { _ = mp.Add(tx, bc) }
+	bc.SetOracle(orc)
+
+	cs := getOracleContractState(bc.contracts.Oracle.Hash)
+	require.NoError(t, bc.contracts.Management.PutContractState(bc.dao, cs))
+
+	go bc.Run()
+	defer bc.Close()
+	go orc.Run()
+	defer orc.Shutdown()
+
+	bc.setNodesByRole(t, true, native.RoleOracle, keys.PublicKeys{acc.PrivateKey().PublicKey()})
+	putOracleRequest(t, cs.Hash, bc, "http://get.1234", new(string), "handle", []byte{}, 10_000_000)
+
+	require.Eventually(t, func() bool { return mp.Count() == 1 },
+		time.Second*2, time.Millisecond*200)
+
+	txes := mp.GetVerifiedTransactions()
+	require.Len(t, txes, 1)
+	require.True(t, txes[0].HasAttribute(transaction.OracleResponseT))
+}
+
+type saveToMapBroadcaster struct {
+	m map[uint64]*responseWithSig
+}
+
+func (b saveToMapBroadcaster) SendResponse(_ *keys.PrivateKey, resp *transaction.OracleResponse, txSig []byte) {
+	b.m[resp.ID] = &responseWithSig{
+		resp:  resp,
+		txSig: txSig,
+	}
+}
+func (saveToMapBroadcaster) Run()      {}
+func (saveToMapBroadcaster) Shutdown() {}
+
+type responseWithSig struct {
+	resp  *transaction.OracleResponse
+	txSig []byte
+}
+
+func saveTxToChan(ch chan *transaction.Transaction) oracle.TxCallback {
+	return func(tx *transaction.Transaction) {
+		ch <- tx
+	}
+}
+
+type (
+	// httpClient implements oracle.HTTPClient with
+	// mocked URL or responses.
+	httpClient struct {
+		responses map[string]testResponse
+	}
+
+	testResponse struct {
+		code int
+		body []byte
+	}
+)
+
+// Get implements oracle.HTTPClient interface.
+func (c *httpClient) Get(url string) (*http.Response, error) {
+	resp, ok := c.responses[url]
+	if ok {
+		return &http.Response{
+			StatusCode: resp.code,
+			Body:       newResponseBody(resp.body),
+		}, nil
+	}
+	return nil, errors.New("error during request")
+}
+
+func newDefaultHTTPClient() oracle.HTTPClient {
+	return &httpClient{
+		responses: map[string]testResponse{
+			"http://get.1234": {
+				code: http.StatusOK,
+				body: []byte{1, 2, 3, 4},
+			},
+			"http://get.4321": {
+				code: http.StatusOK,
+				body: []byte{4, 3, 2, 1},
+			},
+			"http://get.timeout": {
+				code: http.StatusRequestTimeout,
+				body: []byte{},
+			},
+			"http://get.notfound": {
+				code: http.StatusNotFound,
+				body: []byte{},
+			},
+			"http://get.forbidden": {
+				code: http.StatusForbidden,
+				body: []byte{},
+			},
+			"http://private.url": {
+				code: http.StatusOK,
+				body: []byte("passwords"),
+			},
+			"http://get.big": {
+				code: http.StatusOK,
+				body: make([]byte, transaction.MaxOracleResultSize+1),
+			},
+			"http://get.maxallowed": {
+				code: http.StatusOK,
+				body: make([]byte, transaction.MaxOracleResultSize),
+			},
+			"http://get.filter": {
+				code: http.StatusOK,
+				body: []byte(`{"Values":["one", 2, 3],"Another":null}`),
+			},
+			"http://get.filterinv": {
+				code: http.StatusOK,
+				body: []byte{0xFF},
+			},
+		},
+	}
+}
+
+func newResponseBody(resp []byte) gio.ReadCloser {
+	return ioutil.NopCloser(bytes.NewReader(resp))
+}

pkg/network/helper_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/config/netmode"
 	"github.com/nspcc-dev/neo-go/pkg/core/block"
 	"github.com/nspcc-dev/neo-go/pkg/core/blockchainer"
+	"github.com/nspcc-dev/neo-go/pkg/core/blockchainer/services"
 	"github.com/nspcc-dev/neo-go/pkg/core/interop"
 	"github.com/nspcc-dev/neo-go/pkg/core/mempool"
 	"github.com/nspcc-dev/neo-go/pkg/core/native"
@@ -278,7 +279,9 @@ func (chain testChain) ManagementContractHash() util.Uint160 {
 func (chain *testChain) PoolTx(tx *transaction.Transaction, _ ...*mempool.Pool) error {
 	return chain.poolTx(tx)
 }
-
+func (chain testChain) SetOracle(services.Oracle) {
+	panic("TODO")
+}
 func (chain *testChain) SubscribeForBlocks(ch chan<- *block.Block) {
 	chain.blocksCh = append(chain.blocksCh, ch)
 }

pkg/network/server.go

@@ -20,6 +20,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
 	"github.com/nspcc-dev/neo-go/pkg/network/capability"
 	"github.com/nspcc-dev/neo-go/pkg/network/payload"
+	"github.com/nspcc-dev/neo-go/pkg/services/oracle"
 	"github.com/nspcc-dev/neo-go/pkg/util"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
@@ -82,6 +83,8 @@ type (
 
 		consensusStarted *atomic.Bool
 
+		oracle *oracle.Oracle
+
 		log *zap.Logger
 	}
 
@@ -142,6 +145,29 @@ func newServerFromConstructors(config ServerConfig, chain blockchainer.Blockchai
 		}
 	})
 
+	if config.OracleCfg.Enabled {
+		orcCfg := oracle.Config{
+			Log:     log,
+			Network: config.Net,
+			MainCfg: config.OracleCfg,
+			Chain:   chain,
+		}
+		orc, err := oracle.NewOracle(orcCfg)
+		if err != nil {
+			return nil, fmt.Errorf("can't initialize Oracle module: %w", err)
+		}
+		orc.SetOnTransaction(func(tx *transaction.Transaction) {
+			r := s.RelayTxn(tx)
+			if r != RelaySucceed {
+				orc.Log.Error("can't pool oracle tx",
+					zap.String("hash", tx.Hash().StringLE()),
+					zap.Uint8("reason", byte(r)))
+			}
+		})
+		s.oracle = orc
+		chain.SetOracle(orc)
+	}
+
 	srv, err := newConsensus(consensus.Config{
 		Logger:    log,
 		Broadcast: s.handleNewPayload,
@@ -203,6 +229,9 @@ func (s *Server) Start(errChan chan error) {
 	s.initStaleMemPools()
 
 	go s.broadcastTxLoop()
+	if s.oracle != nil {
+		go s.oracle.Run()
+	}
 	go s.relayBlocksLoop()
 	go s.bQueue.run()
 	go s.transport.Accept()
@@ -222,9 +251,17 @@ func (s *Server) Shutdown() {
 		p.Disconnect(errServerShutdown)
 	}
 	s.bQueue.discard()
+	if s.oracle != nil {
+		s.oracle.Shutdown()
+	}
 	close(s.quit)
 }
 
+// GetOracle returns oracle module instance.
+func (s *Server) GetOracle() *oracle.Oracle {
+	return s.oracle
+}
+
 // UnconnectedPeers returns a list of peers that are in the discovery peer list
 // but are not connected to the server.
 func (s *Server) UnconnectedPeers() []string {

pkg/network/server_config.go

@@ -66,6 +66,9 @@ type (
 
 		// TimePerBlock is an interval which should pass between two successive blocks.
 		TimePerBlock time.Duration
+
+		// OracleCfg is oracle module configuration.
+		OracleCfg config.OracleConfiguration
 	}
 )
 
@@ -96,5 +99,6 @@ func NewServerConfig(cfg config.Config) ServerConfig {
 		MinPeers:          appConfig.MinPeers,
 		Wallet:            wc,
 		TimePerBlock:      time.Duration(protoConfig.SecondsPerBlock) * time.Second,
+		OracleCfg:         appConfig.Oracle,
 	}
 }

pkg/rpc/client/rpc.go

@@ -491,6 +491,12 @@ func (c *Client) SubmitBlock(b block.Block) (util.Uint256, error) {
 	return resp.Hash, nil
 }
 
+// SubmitRawOracleResponse submits raw oracle response to the oracle node.
+// Raw params are used to avoid excessive marshalling.
+func (c *Client) SubmitRawOracleResponse(ps request.RawParams) error {
+	return c.performRequest("submitoracleresponse", ps, new(result.RelayResult))
+}
+
 // SignAndPushInvocationTx signs and pushes given script as an invocation
 // transaction  using given wif to sign it and spending the amount of gas
 // specified. It returns a hash of the invocation transaction and an error.

pkg/rpc/server/server.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"context"
+	"crypto/elliptic"
 	"encoding/binary"
 	"encoding/hex"
 	"encoding/json"
@@ -23,10 +24,13 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/core/mpt"
 	"github.com/nspcc-dev/neo-go/pkg/core/state"
 	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/hash"
 	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
 	"github.com/nspcc-dev/neo-go/pkg/encoding/address"
 	"github.com/nspcc-dev/neo-go/pkg/io"
 	"github.com/nspcc-dev/neo-go/pkg/network"
+	"github.com/nspcc-dev/neo-go/pkg/services/oracle"
+	"github.com/nspcc-dev/neo-go/pkg/services/oracle/broadcaster"
 	"github.com/nspcc-dev/neo-go/pkg/rpc"
 	"github.com/nspcc-dev/neo-go/pkg/rpc/request"
 	"github.com/nspcc-dev/neo-go/pkg/rpc/response"
@@ -47,6 +51,7 @@ type (
 		network          netmode.Magic
 		stateRootEnabled bool
 		coreServer       *network.Server
+		oracle           *oracle.Oracle
 		log              *zap.Logger
 		https            *http.Server
 		shutdown         chan struct{}
@@ -116,6 +121,7 @@ var rpcHandlers = map[string]func(*Server, request.Params) (interface{}, *respon
 	"invokecontractverify":   (*Server).invokeContractVerify,
 	"sendrawtransaction":     (*Server).sendrawtransaction,
 	"submitblock":            (*Server).submitBlock,
+	"submitoracleresponse":   (*Server).submitOracleResponse,
 	"validateaddress":        (*Server).validateAddress,
 	"verifyproof":            (*Server).verifyProof,
 }
@@ -134,7 +140,8 @@ var invalidBlockHeightError = func(index int, height int) *response.Error {
 var upgrader = websocket.Upgrader{}
 
 // New creates a new Server struct.
-func New(chain blockchainer.Blockchainer, conf rpc.Config, coreServer *network.Server, log *zap.Logger) Server {
+func New(chain blockchainer.Blockchainer, conf rpc.Config, coreServer *network.Server,
+	orc *oracle.Oracle, log *zap.Logger) Server {
 	httpServer := &http.Server{
 		Addr: conf.Address + ":" + strconv.FormatUint(uint64(conf.Port), 10),
 	}
@@ -146,6 +153,9 @@ func New(chain blockchainer.Blockchainer, conf rpc.Config, coreServer *network.S
 		}
 	}
 
+	if orc != nil {
+		orc.SetBroadcaster(broadcaster.New(orc.MainCfg, log))
+	}
 	return Server{
 		Server:           httpServer,
 		chain:            chain,
@@ -154,6 +164,7 @@ func New(chain blockchainer.Blockchainer, conf rpc.Config, coreServer *network.S
 		stateRootEnabled: chain.GetConfig().StateRootInHeader,
 		coreServer:       coreServer,
 		log:              log,
+		oracle:           orc,
 		https:            tlsServer,
 		shutdown:         make(chan struct{}),
 
@@ -1223,6 +1234,38 @@ func (s *Server) submitBlock(reqParams request.Params) (interface{}, *response.E
 	}, nil
 }
 
+func (s *Server) submitOracleResponse(ps request.Params) (interface{}, *response.Error) {
+	if s.oracle == nil {
+		return nil, response.NewInternalServerError("oracle is not enabled", nil)
+	}
+	var pub *keys.PublicKey
+	pubBytes, err := ps.Value(0).GetBytesBase64()
+	if err == nil {
+		pub, err = keys.NewPublicKeyFromBytes(pubBytes, elliptic.P256())
+	}
+	if err != nil {
+		return nil, response.NewInvalidParamsError("public key is missing", err)
+	}
+	reqID, err := ps.Value(1).GetInt()
+	if err != nil {
+		return nil, response.NewInvalidParamsError("request ID is missing", err)
+	}
+	txSig, err := ps.Value(2).GetBytesBase64()
+	if err != nil {
+		return nil, response.NewInvalidParamsError("tx signature is missing", err)
+	}
+	msgSig, err := ps.Value(3).GetBytesBase64()
+	if err != nil {
+		return nil, response.NewInvalidParamsError("msg signature is missing", err)
+	}
+	data := broadcaster.GetMessage(pubBytes, uint64(reqID), txSig)
+	if !pub.Verify(msgSig, hash.Sha256(data).BytesBE()) {
+		return nil, response.NewRPCError("Invalid sign", "", nil)
+	}
+	s.oracle.AddResponse(pub, uint64(reqID), txSig)
+	return json.RawMessage([]byte("{}")), nil
+}
+
 func (s *Server) sendrawtransaction(reqParams request.Params) (interface{}, *response.Error) {
 	var resultsErr *response.Error
 	var results interface{}

pkg/rpc/server/server_helper_test.go

@@ -16,13 +16,14 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/core/storage"
 	"github.com/nspcc-dev/neo-go/pkg/io"
 	"github.com/nspcc-dev/neo-go/pkg/network"
+	"github.com/nspcc-dev/neo-go/pkg/services/oracle"
 	"github.com/nspcc-dev/neo-go/pkg/util"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
 )
 
-func getUnitTestChain(t *testing.T) (*core.Blockchain, config.Config, *zap.Logger) {
+func getUnitTestChain(t *testing.T, enableOracle bool) (*core.Blockchain, *oracle.Oracle, config.Config, *zap.Logger) {
 	net := netmode.UnitTestNet
 	configPath := "../../../config"
 	cfg, err := config.Load(configPath, net)
@@ -33,9 +34,26 @@ func getUnitTestChain(t *testing.T) (*core.Blockchain, config.Config, *zap.Logge
 	chain, err := core.NewBlockchain(memoryStore, cfg.ProtocolConfiguration, logger)
 	require.NoError(t, err, "could not create chain")
 
+	var orc *oracle.Oracle
+	if enableOracle {
+		cfg.ApplicationConfiguration.Oracle.Enabled = true
+		cfg.ApplicationConfiguration.Oracle.UnlockWallet = config.Wallet{
+			Path:     "../../services/oracle/testdata/oracle1.json",
+			Password: "one",
+		}
+		orc, err = oracle.NewOracle(oracle.Config{
+			Log:     logger,
+			Network: netmode.UnitTestNet,
+			MainCfg: cfg.ApplicationConfiguration.Oracle,
+			Chain:   chain,
+		})
+		require.NoError(t, err)
+		chain.SetOracle(orc)
+	}
+
 	go chain.Run()
 
-	return chain, cfg, logger
+	return chain, orc, cfg, logger
 }
 
 func getTestBlocks(t *testing.T) []*block.Block {
@@ -61,13 +79,13 @@ func getTestBlocks(t *testing.T) []*block.Block {
 	return blocks
 }
 
-func initClearServerWithInMemoryChain(t *testing.T) (*core.Blockchain, *Server, *httptest.Server) {
+func initClearServerWithOracle(t *testing.T, needOracle bool) (*core.Blockchain, *Server, *httptest.Server) {
-	chain, cfg, logger := getUnitTestChain(t)
+	chain, orc, cfg, logger := getUnitTestChain(t, needOracle)
 
 	serverConfig := network.NewServerConfig(cfg)
 	server, err := network.NewServer(serverConfig, chain, logger)
 	require.NoError(t, err)
-	rpcServer := New(chain, cfg.ApplicationConfiguration.RPC, server, logger)
+	rpcServer := New(chain, cfg.ApplicationConfiguration.RPC, server, orc, logger)
 	errCh := make(chan error, 2)
 	rpcServer.Start(errCh)
 
@@ -77,6 +95,10 @@ func initClearServerWithInMemoryChain(t *testing.T) (*core.Blockchain, *Server,
 	return chain, &rpcServer, srv
 }
 
+func initClearServerWithInMemoryChain(t *testing.T) (*core.Blockchain, *Server, *httptest.Server) {
+	return initClearServerWithOracle(t, false)
+}
+
 func initServerWithInMemoryChain(t *testing.T) (*core.Blockchain, *Server, *httptest.Server) {
 	chain, rpcServer, srv := initClearServerWithInMemoryChain(t)
 

pkg/rpc/server/server_test.go

@@ -25,8 +25,10 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/core/fee"
 	"github.com/nspcc-dev/neo-go/pkg/core/state"
 	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
 	"github.com/nspcc-dev/neo-go/pkg/encoding/address"
 	"github.com/nspcc-dev/neo-go/pkg/io"
+	rpc2 "github.com/nspcc-dev/neo-go/pkg/services/oracle/broadcaster"
 	"github.com/nspcc-dev/neo-go/pkg/rpc/response"
 	"github.com/nspcc-dev/neo-go/pkg/rpc/response/result"
 	"github.com/nspcc-dev/neo-go/pkg/smartcontract/trigger"
@@ -885,6 +887,13 @@ var rpcTestCases = map[string][]rpcTestCase{
 			fail:   true,
 		},
 	},
+	"submitoracleresponse": {
+		{
+			name:   "no params",
+			params: `[]`,
+			fail:   true,
+		},
+	},
 	"validateaddress": {
 		{
 			name:   "positive",
@@ -920,6 +929,39 @@ func TestRPC(t *testing.T) {
 	})
 }
 
+func TestSubmitOracle(t *testing.T) {
+	chain, rpcSrv, httpSrv := initClearServerWithOracle(t, true)
+	defer chain.Close()
+	defer rpcSrv.Shutdown()
+
+	rpc := `{"jsonrpc": "2.0", "id": 1, "method": "submitoracleresponse", "params": %s}`
+	runCase := func(t *testing.T, fail bool, params ...string) func(t *testing.T) {
+		return func(t *testing.T) {
+			ps := `[` + strings.Join(params, ",") + `]`
+			req := fmt.Sprintf(rpc, ps)
+			body := doRPCCallOverHTTP(req, httpSrv.URL, t)
+			checkErrGetResult(t, body, fail)
+		}
+	}
+	t.Run("MissingKey", runCase(t, true))
+	t.Run("InvalidKey", runCase(t, true, `"1234"`))
+
+	priv, err := keys.NewPrivateKey()
+	require.NoError(t, err)
+	pubStr := `"` + base64.StdEncoding.EncodeToString(priv.PublicKey().Bytes()) + `"`
+	t.Run("InvalidReqID", runCase(t, true, pubStr, `"notanumber"`))
+	t.Run("InvalidTxSignature", runCase(t, true, pubStr, `1`, `"qwerty"`))
+
+	txSig := priv.Sign([]byte{1, 2, 3})
+	txSigStr := `"` + base64.StdEncoding.EncodeToString(txSig) + `"`
+	t.Run("MissingMsgSignature", runCase(t, true, pubStr, `1`, txSigStr))
+	t.Run("InvalidMsgSignature", runCase(t, true, pubStr, `1`, txSigStr, `"0123"`))
+
+	msg := rpc2.GetMessage(priv.PublicKey().Bytes(), 1, txSig)
+	msgSigStr := `"` + base64.StdEncoding.EncodeToString(priv.Sign(msg)) + `"`
+	t.Run("Valid", runCase(t, false, pubStr, `1`, txSigStr, msgSigStr))
+}
+
 // testRPCProtocol runs a full set of tests using given callback to make actual
 // calls. Some tests change the chain state, thus we reinitialize the chain from
 // scratch here.

pkg/services/oracle/broadcaster/client.go

@@ -0,0 +1,58 @@
+package broadcaster
+
+import (
+	"context"
+	"time"
+
+	"github.com/nspcc-dev/neo-go/pkg/rpc/client"
+	"github.com/nspcc-dev/neo-go/pkg/rpc/request"
+	"go.uber.org/zap"
+)
+
+type oracleClient struct {
+	client      *client.Client
+	addr        string
+	close       chan struct{}
+	responses   chan request.RawParams
+	log         *zap.Logger
+	sendTimeout time.Duration
+}
+
+func (r *rpcBroascaster) newOracleClient(addr string, timeout time.Duration, ch chan request.RawParams) *oracleClient {
+	return &oracleClient{
+		addr:        addr,
+		close:       r.close,
+		responses:   ch,
+		log:         r.log.With(zap.String("address", addr)),
+		sendTimeout: timeout,
+	}
+}
+
+func (c *oracleClient) run() {
+	// We ignore error as not every node can be available on startup.
+	c.client, _ = client.New(context.Background(), "http://"+c.addr, client.Options{
+		DialTimeout:    c.sendTimeout,
+		RequestTimeout: c.sendTimeout,
+	})
+	for {
+		select {
+		case <-c.close:
+			return
+		case ps := <-c.responses:
+			if c.client == nil {
+				var err error
+				c.client, err = client.New(context.Background(), "http://"+c.addr, client.Options{
+					DialTimeout:    c.sendTimeout,
+					RequestTimeout: c.sendTimeout,
+				})
+				if err != nil {
+					continue
+				}
+			}
+			err := c.client.SubmitRawOracleResponse(ps)
+			if err != nil {
+				c.log.Error("error while submitting oracle response", zap.Error(err))
+			}
+		}
+	}
+}

pkg/services/oracle/broadcaster/oracle.go

@@ -0,0 +1,96 @@
+package broadcaster
+
+import (
+	"encoding/base64"
+	"encoding/binary"
+	"time"
+
+	"github.com/nspcc-dev/neo-go/pkg/config"
+	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
+	"github.com/nspcc-dev/neo-go/pkg/rpc/request"
+	"github.com/nspcc-dev/neo-go/pkg/services/oracle"
+	"go.uber.org/zap"
+)
+
+type rpcBroascaster struct {
+	clients map[string]*oracleClient
+	log     *zap.Logger
+
+	close       chan struct{}
+	responses   chan request.RawParams
+	sendTimeout time.Duration
+}
+
+const (
+	defaultSendTimeout = time.Second * 4
+
+	defaultChanCapacity = 16
+)
+
+// New returns new struct capable of broadcasting oracle responses.
+func New(cfg config.OracleConfiguration, log *zap.Logger) oracle.Broadcaster {
+	if cfg.ResponseTimeout == 0 {
+		cfg.ResponseTimeout = defaultSendTimeout
+	}
+	r := &rpcBroascaster{
+		clients:     make(map[string]*oracleClient, len(cfg.Nodes)),
+		log:         log,
+		close:       make(chan struct{}),
+		responses:   make(chan request.RawParams),
+		sendTimeout: cfg.ResponseTimeout,
+	}
+	for i := range cfg.Nodes {
+		r.clients[cfg.Nodes[i]] = r.newOracleClient(cfg.Nodes[i], cfg.ResponseTimeout, make(chan request.RawParams, defaultChanCapacity))
+	}
+	return r
+}
+
+// Run implements oracle.Broadcaster.
+func (r *rpcBroascaster) Run() {
+	for _, c := range r.clients {
+		go c.run()
+	}
+	for {
+		select {
+		case <-r.close:
+			return
+		case ps := <-r.responses:
+			for _, c := range r.clients {
+				select {
+				case c.responses <- ps:
+				default:
+					c.log.Error("can't send response, channel is full")
+				}
+			}
+		}
+	}
+}
+
+// Shutdown implements oracle.Broadcaster.
+func (r *rpcBroascaster) Shutdown() {
+	close(r.close)
+}
+
+// SendResponse implements interfaces.Broadcaster.
+func (r *rpcBroascaster) SendResponse(priv *keys.PrivateKey, resp *transaction.OracleResponse, txSig []byte) {
+	pub := priv.PublicKey()
+	data := GetMessage(pub.Bytes(), resp.ID, txSig)
+	msgSig := priv.Sign(data)
+	params := request.NewRawParams(
+		base64.StdEncoding.EncodeToString(pub.Bytes()),
+		resp.ID,
+		base64.StdEncoding.EncodeToString(txSig),
+		base64.StdEncoding.EncodeToString(msgSig),
+	)
+	r.responses <- params
+}
+
+// GetMessage returns data which is signed upon sending response by RPC.
+func GetMessage(pubBytes []byte, reqID uint64, txSig []byte) []byte {
+	data := make([]byte, len(pubBytes)+8+len(txSig))
+	copy(data, pubBytes)
+	binary.LittleEndian.PutUint64(data[len(pubBytes):], uint64(reqID))
+	copy(data[len(pubBytes)+8:], txSig)
+	return data
+}

pkg/services/oracle/filter.go

@@ -0,0 +1,24 @@
+package oracle
+
+import (
+	"encoding/json"
+	"errors"
+	"unicode/utf8"
+
+	"github.com/PaesslerAG/jsonpath"
+)
+
+func filter(value []byte, path string) ([]byte, error) {
+	if !utf8.Valid(value) {
+		return nil, errors.New("not an UTF-8")
+	}
+	var v interface{}
+	if err := json.Unmarshal(value, &v); err != nil {
+		return nil, err
+	}
+	result, err := jsonpath.Get(path, v)
+	if err != nil {
+		return nil, err
+	}
+	return json.Marshal([]interface{}{result})
+}

pkg/services/oracle/filter_test.go

@@ -0,0 +1,50 @@
+package oracle
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestFilter(t *testing.T) {
+	js := `{
+  	"Stores": [ "Lambton Quay",	"Willis Street" ],
+  	"Manufacturers": [
+		{
+			"Name": "Acme Co",
+			"Products": [
+		        { "Name": "Anvil", "Price": 50 }
+      		]
+    	},
+    	{
+      		"Name": "Contoso",
+      		"Products": [
+        		{ "Name": "Elbow Grease", "Price": 99.95 },
+        		{ "Name": "Headlight Fluid", "Price": 4 }
+      		]
+    	}
+  	]
+}`
+
+	testCases := []struct {
+		result, path string
+	}{
+		{`["Acme Co"]`, "Manufacturers[0].Name"},
+		{`[50]`, "Manufacturers[0].Products[0].Price"},
+		{`["Elbow Grease"]`, "Manufacturers[1].Products[0].Name"},
+		{`[{"Name":"Elbow Grease","Price":99.95}]`, "Manufacturers[1].Products[0]"},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.path, func(t *testing.T) {
+			actual, err := filter([]byte(js), tc.path)
+			require.NoError(t, err)
+			require.Equal(t, tc.result, string(actual))
+		})
+	}
+
+	t.Run("not an UTF-8", func(t *testing.T) {
+		_, err := filter([]byte{0xFF}, "Manufacturers[0].Name")
+		require.Error(t, err)
+	})
+}

pkg/services/oracle/network.go

@@ -0,0 +1,56 @@
+package oracle
+
+import (
+	"errors"
+	"net"
+	"net/url"
+)
+
+// reservedCIDRs is a list of ip addresses for private networks.
+// https://tools.ietf.org/html/rfc6890
+var reservedCIDRs = []string{
+	// IPv4
+	"10.0.0.0/8",
+	"100.64.0.0/10",
+	"172.16.0.0/12",
+	"192.0.0.0/24",
+	"192.168.0.0/16",
+	"198.18.0.0/15",
+	// IPv6
+	"fc00::/7",
+}
+
+var privateNets = make([]net.IPNet, 0, len(reservedCIDRs))
+
+func init() {
+	for i := range reservedCIDRs {
+		_, ipNet, err := net.ParseCIDR(reservedCIDRs[i])
+		if err != nil {
+			panic(err)
+		}
+		privateNets = append(privateNets, *ipNet)
+	}
+}
+
+func defaultURIValidator(u *url.URL) error {
+	ip, err := net.ResolveIPAddr("ip", u.Hostname())
+	if err != nil {
+		return err
+	}
+	if isReserved(ip.IP) {
+		return errors.New("IP is not global unicast")
+	}
+	return nil
+}
+
+func isReserved(ip net.IP) bool {
+	if !ip.IsGlobalUnicast() {
+		return true
+	}
+	for i := range privateNets {
+		if privateNets[i].Contains(ip) {
+			return true
+		}
+	}
+	return false
+}

pkg/services/oracle/network_test.go

@@ -0,0 +1,18 @@
+package oracle
+
+import (
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestIsReserved(t *testing.T) {
+	require.True(t, isReserved(net.IPv4zero))
+	require.True(t, isReserved(net.IPv4(10, 0, 0, 1)))
+	require.True(t, isReserved(net.IPv4(192, 168, 0, 1)))
+	require.True(t, isReserved(net.IPv6interfacelocalallnodes))
+	require.True(t, isReserved(net.IPv6loopback))
+
+	require.False(t, isReserved(net.IPv4(8, 8, 8, 8)))
+}

pkg/services/oracle/nodes.go

@@ -0,0 +1,69 @@
+package oracle
+
+import (
+	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
+	"github.com/nspcc-dev/neo-go/pkg/encoding/address"
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract"
+	"github.com/nspcc-dev/neo-go/pkg/wallet"
+	"go.uber.org/zap"
+)
+
+// UpdateOracleNodes updates oracle nodes list.
+func (o *Oracle) UpdateOracleNodes(oracleNodes keys.PublicKeys) {
+	o.accMtx.Lock()
+	defer o.accMtx.Unlock()
+
+	old := o.oracleNodes
+	if isEqual := len(old) == len(oracleNodes); isEqual {
+		for i := range old {
+			if !old[i].Equal(oracleNodes[i]) {
+				isEqual = false
+				break
+			}
+		}
+		if isEqual {
+			return
+		}
+	}
+
+	var acc *wallet.Account
+	for i := range oracleNodes {
+		acc = o.wallet.GetAccount(oracleNodes[i].GetScriptHash())
+		if acc != nil {
+			if acc.PrivateKey() != nil {
+				break
+			}
+			err := acc.Decrypt(o.MainCfg.UnlockWallet.Password)
+			if err != nil {
+				o.Log.Error("can't unlock account",
+					zap.String("address", address.Uint160ToString(acc.Contract.ScriptHash())),
+					zap.Error(err))
+				o.currAccount = nil
+				return
+			}
+			break
+		}
+	}
+
+	o.currAccount = acc
+	o.oracleSignContract, _ = smartcontract.CreateDefaultMultiSigRedeemScript(oracleNodes)
+	o.oracleNodes = oracleNodes
+}
+
+func (o *Oracle) getAccount() *wallet.Account {
+	o.accMtx.RLock()
+	defer o.accMtx.RUnlock()
+	return o.currAccount
+}
+
+func (o *Oracle) getOracleNodes() keys.PublicKeys {
+	o.accMtx.RLock()
+	defer o.accMtx.RUnlock()
+	return o.oracleNodes
+}
+
+func (o *Oracle) getOracleSignContract() []byte {
+	o.accMtx.RLock()
+	defer o.accMtx.RUnlock()
+	return o.oracleSignContract
+}

pkg/services/oracle/oracle.go

@@ -0,0 +1,241 @@
+package oracle
+
+import (
+	"errors"
+	"net/http"
+	"net/url"
+	"sync"
+	"time"
+
+	"github.com/nspcc-dev/neo-go/pkg/config"
+	"github.com/nspcc-dev/neo-go/pkg/config/netmode"
+	"github.com/nspcc-dev/neo-go/pkg/core/blockchainer"
+	"github.com/nspcc-dev/neo-go/pkg/core/state"
+	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
+	"github.com/nspcc-dev/neo-go/pkg/util"
+	"github.com/nspcc-dev/neo-go/pkg/wallet"
+	"go.uber.org/zap"
+)
+
+type (
+	// Oracle represents oracle module capable of talking
+	// with the external world.
+	Oracle struct {
+		Config
+
+		// mtx protects setting callbacks.
+		mtx sync.RWMutex
+
+		// accMtx protects account and oracle nodes.
+		accMtx             sync.RWMutex
+		currAccount        *wallet.Account
+		oracleNodes        keys.PublicKeys
+		oracleSignContract []byte
+
+		close      chan struct{}
+		requestCh  chan request
+		requestMap chan map[uint64]*state.OracleRequest
+
+		// respMtx protects responses map.
+		respMtx   sync.RWMutex
+		responses map[uint64]*incompleteTx
+		// removed contains ids of requests which won't be processed further due to expiration.
+		removed map[uint64]bool
+
+		wallet *wallet.Wallet
+	}
+
+	// Config contains oracle module parameters.
+	Config struct {
+		Log             *zap.Logger
+		Network         netmode.Magic
+		MainCfg         config.OracleConfiguration
+		Client          HTTPClient
+		Chain           blockchainer.Blockchainer
+		ResponseHandler Broadcaster
+		OnTransaction   TxCallback
+		URIValidator    URIValidator
+		OracleScript    []byte
+		OracleResponse  []byte
+		OracleHash      util.Uint160
+	}
+
+	// HTTPClient is an interface capable of doing oracle requests.
+	HTTPClient interface {
+		Get(string) (*http.Response, error)
+	}
+
+	// Broadcaster broadcasts oracle responses.
+	Broadcaster interface {
+		SendResponse(priv *keys.PrivateKey, resp *transaction.OracleResponse, txSig []byte)
+		Run()
+		Shutdown()
+	}
+
+	defaultResponseHandler struct{}
+
+	// TxCallback executes on new transactions when they are ready to be pooled.
+	TxCallback = func(tx *transaction.Transaction)
+	// URIValidator is used to check if provided URL is valid.
+	URIValidator = func(*url.URL) error
+)
+
+const (
+	// defaultRequestTimeout is default request timeout.
+	defaultRequestTimeout = time.Second * 5
+
+	// defaultMaxTaskTimeout is default timeout for the request to be dropped if it can't be processed.
+	defaultMaxTaskTimeout = time.Hour
+
+	// defaultRefreshInterval is default timeout for the failed request to be reprocessed.
+	defaultRefreshInterval = time.Minute * 3
+)
+
+// NewOracle returns new oracle instance.
+func NewOracle(cfg Config) (*Oracle, error) {
+	o := &Oracle{
+		Config: cfg,
+
+		close:      make(chan struct{}),
+		requestMap: make(chan map[uint64]*state.OracleRequest, 1),
+		responses:  make(map[uint64]*incompleteTx),
+		removed:    make(map[uint64]bool),
+	}
+	if o.MainCfg.RequestTimeout == 0 {
+		o.MainCfg.RequestTimeout = defaultRequestTimeout
+	}
+	if o.MainCfg.MaxConcurrentRequests == 0 {
+		o.MainCfg.MaxConcurrentRequests = defaultMaxConcurrentRequests
+	}
+	o.requestCh = make(chan request, o.MainCfg.MaxConcurrentRequests)
+	if o.MainCfg.MaxTaskTimeout == 0 {
+		o.MainCfg.MaxTaskTimeout = defaultMaxTaskTimeout
+	}
+	if o.MainCfg.RefreshInterval == 0 {
+		o.MainCfg.RefreshInterval = defaultRefreshInterval
+	}
+
+	var err error
+	w := cfg.MainCfg.UnlockWallet
+	if o.wallet, err = wallet.NewWalletFromFile(w.Path); err != nil {
+		return nil, err
+	}
+
+	haveAccount := false
+	for _, acc := range o.wallet.Accounts {
+		if err := acc.Decrypt(w.Password); err == nil {
+			haveAccount = true
+			break
+		}
+	}
+	if !haveAccount {
+		return nil, errors.New("no wallet account could be unlocked")
+	}
+
+	if o.Client == nil {
+		var client http.Client
+		client.Transport = &http.Transport{DisableKeepAlives: true}
+		client.Timeout = o.MainCfg.RequestTimeout
+		o.Client = &client
+	}
+	if o.ResponseHandler == nil {
+		o.ResponseHandler = defaultResponseHandler{}
+	}
+	if o.OnTransaction == nil {
+		o.OnTransaction = func(*transaction.Transaction) {}
+	}
+	if o.URIValidator == nil {
+		o.URIValidator = defaultURIValidator
+	}
+	return o, nil
+}
+
+// Shutdown shutdowns Oracle.
+func (o *Oracle) Shutdown() {
+	close(o.close)
+	o.getBroadcaster().Shutdown()
+}
+
+// Run runs must be executed in a separate goroutine.
+func (o *Oracle) Run() {
+	for i := 0; i < o.MainCfg.MaxConcurrentRequests; i++ {
+		go o.runRequestWorker()
+	}
+
+	tick := time.NewTicker(o.MainCfg.RefreshInterval)
+	for {
+		select {
+		case <-o.close:
+			tick.Stop()
+			return
+		case <-tick.C:
+			var reprocess []uint64
+			o.respMtx.RLock()
+			o.removed = make(map[uint64]bool)
+			for id, incTx := range o.responses {
+				incTx.RLock()
+				since := time.Since(incTx.time)
+				if since > o.MainCfg.MaxTaskTimeout {
+					o.removed[id] = true
+				} else if since > o.MainCfg.RefreshInterval {
+					reprocess = append(reprocess, id)
+				}
+				incTx.RUnlock()
+			}
+			for id := range o.removed {
+				delete(o.responses, id)
+			}
+			o.respMtx.Unlock()
+
+			for _, id := range reprocess {
+				o.requestCh <- request{ID: id}
+			}
+		case reqs := <-o.requestMap:
+			for id, req := range reqs {
+				o.requestCh <- request{
+					ID:  id,
+					Req: req,
+				}
+			}
+		}
+	}
+}
+
+func (o *Oracle) getOnTransaction() TxCallback {
+	o.mtx.RLock()
+	defer o.mtx.RUnlock()
+	return o.OnTransaction
+}
+
+// SetOnTransaction sets callback to pool and broadcast tx.
+func (o *Oracle) SetOnTransaction(cb TxCallback) {
+	o.mtx.Lock()
+	defer o.mtx.Unlock()
+	o.OnTransaction = cb
+}
+
+func (o *Oracle) getBroadcaster() Broadcaster {
+	o.mtx.RLock()
+	defer o.mtx.RUnlock()
+	return o.ResponseHandler
+}
+
+// SetBroadcaster sets callback to broadcast response.
+func (o *Oracle) SetBroadcaster(b Broadcaster) {
+	o.mtx.Lock()
+	defer o.mtx.Unlock()
+	o.ResponseHandler.Shutdown()
+	o.ResponseHandler = b
+	go b.Run()
+}
+
+// SendResponse implements Broadcaster interface.
+func (defaultResponseHandler) SendResponse(*keys.PrivateKey, *transaction.OracleResponse, []byte) {
+}
+
+// Run implements Broadcaster interface.
+func (defaultResponseHandler) Run() {}
+
+// Shutdown implements Broadcaster interface.
+func (defaultResponseHandler) Shutdown() {}

pkg/services/oracle/request.go

@@ -0,0 +1,216 @@
+package oracle
+
+import (
+	"errors"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/nspcc-dev/neo-go/pkg/core/state"
+	"github.com/nspcc-dev/neo-go/pkg/core/storage"
+	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
+	"go.uber.org/zap"
+)
+
+const defaultMaxConcurrentRequests = 10
+
+type request struct {
+	ID  uint64
+	Req *state.OracleRequest
+}
+
+func (o *Oracle) runRequestWorker() {
+	for {
+		select {
+		case <-o.close:
+			return
+		case req := <-o.requestCh:
+			acc := o.getAccount()
+			if acc == nil {
+				continue
+			}
+			err := o.processRequest(acc.PrivateKey(), req)
+			if err != nil {
+				o.Log.Debug("can't process request", zap.Uint64("id", req.ID), zap.Error(err))
+			}
+		}
+	}
+}
+
+// RemoveRequests removes all data associated with requests
+// which have been processed by oracle contract.
+func (o *Oracle) RemoveRequests(ids []uint64) {
+	o.respMtx.Lock()
+	defer o.respMtx.Unlock()
+	for _, id := range ids {
+		delete(o.responses, id)
+	}
+}
+
+// AddRequests saves all requests in-fly for further processing.
+func (o *Oracle) AddRequests(reqs map[uint64]*state.OracleRequest) {
+	if len(reqs) == 0 {
+		return
+	}
+
+	select {
+	case o.requestMap <- reqs:
+	default:
+		select {
+		case old := <-o.requestMap:
+			for id, r := range old {
+				reqs[id] = r
+			}
+		default:
+		}
+		o.requestMap <- reqs
+	}
+}
+
+// ProcessRequestsInternal processes provided requests synchronously.
+func (o *Oracle) ProcessRequestsInternal(reqs map[uint64]*state.OracleRequest) {
+	acc := o.getAccount()
+	if acc == nil {
+		return
+	}
+
+	// Process actual requests.
+	for id, req := range reqs {
+		if err := o.processRequest(acc.PrivateKey(), request{ID: id, Req: req}); err != nil {
+			o.Log.Debug("can't process request", zap.Error(err))
+		}
+	}
+}
+
+func (o *Oracle) processRequest(priv *keys.PrivateKey, req request) error {
+	if req.Req == nil {
+		o.processFailedRequest(priv, req)
+		return nil
+	}
+
+	incTx := o.getResponse(req.ID, true)
+	if incTx == nil {
+		return nil
+	}
+	resp := &transaction.OracleResponse{ID: req.ID}
+	u, err := url.ParseRequestURI(req.Req.URL)
+	if err == nil && !o.MainCfg.AllowPrivateHost {
+		err = o.URIValidator(u)
+	}
+	if err != nil {
+		resp.Code = transaction.Forbidden
+	} else if u.Scheme == "http" {
+		r, err := o.Client.Get(req.Req.URL)
+		switch {
+		case err != nil:
+			resp.Code = transaction.Error
+		case r.StatusCode == http.StatusOK:
+			result, err := readResponse(r.Body, transaction.MaxOracleResultSize)
+			if err != nil {
+				if errors.Is(err, ErrResponseTooLarge) {
+					resp.Code = transaction.ResponseTooLarge
+				} else {
+					resp.Code = transaction.Error
+				}
+				break
+			}
+			if req.Req.Filter != nil {
+				res, err := filter(result, *req.Req.Filter)
+				if err != nil {
+					resp.Code = transaction.Error
+					break
+				}
+				result = res
+			}
+			resp.Code = transaction.Success
+			resp.Result = result
+		case r.StatusCode == http.StatusForbidden:
+			resp.Code = transaction.Forbidden
+		case r.StatusCode == http.StatusNotFound:
+			resp.Code = transaction.NotFound
+		case r.StatusCode == http.StatusRequestTimeout:
+			resp.Code = transaction.Timeout
+		default:
+			resp.Code = transaction.Error
+		}
+	}
+
+	currentHeight := o.Chain.BlockHeight()
+	_, h, err := o.Chain.GetTransaction(req.Req.OriginalTxID)
+	if err != nil {
+		if !errors.Is(err, storage.ErrKeyNotFound) {
+			return err
+		}
+		// The only reason tx can be not found is if it wasn't yet persisted from DAO.
+		h = currentHeight
+	}
+	tx, err := o.CreateResponseTx(int64(req.Req.GasForResponse), h, resp)
+	if err != nil {
+		return err
+	}
+	backupTx, err := o.CreateResponseTx(int64(req.Req.GasForResponse), h, &transaction.OracleResponse{
+		ID:   req.ID,
+		Code: transaction.ConsensusUnreachable,
+	})
+	if err != nil {
+		return err
+	}
+
+	incTx.Lock()
+	incTx.request = req.Req
+	incTx.tx = tx
+	incTx.backupTx = backupTx
+	incTx.reverifyTx()
+
+	txSig := priv.Sign(tx.GetSignedPart())
+	incTx.addResponse(priv.PublicKey(), txSig, false)
+
+	backupSig := priv.Sign(backupTx.GetSignedPart())
+	incTx.addResponse(priv.PublicKey(), backupSig, true)
+
+	readyTx, ready := incTx.finalize(o.getOracleNodes(), false)
+	if ready {
+		ready = !incTx.isSent
+		incTx.isSent = true
+	}
+	incTx.time = time.Now()
+	incTx.attempts++
+	incTx.Unlock()
+
+	o.getBroadcaster().SendResponse(priv, resp, txSig)
+	if ready {
+		o.getOnTransaction()(readyTx)
+	}
+	return nil
+}
+
+func (o *Oracle) processFailedRequest(priv *keys.PrivateKey, req request) {
+	// Request is being processed again.
+	incTx := o.getResponse(req.ID, false)
+	if incTx == nil {
+		// Request was processed by other oracle nodes.
+		return
+	} else if incTx.isSent {
+		// Tx was sent but not yet persisted. Try to pool it again.
+		o.getOnTransaction()(incTx.tx)
+		return
+	}
+
+	// Don't process request again, fallback to backup tx.
+	incTx.Lock()
+	readyTx, ready := incTx.finalize(o.getOracleNodes(), true)
+	if ready {
+		ready = !incTx.isSent
+		incTx.isSent = true
+	}
+	incTx.time = time.Now()
+	incTx.attempts++
+	txSig := incTx.backupSigs[string(priv.PublicKey().Bytes())].sig
+	incTx.Unlock()
+
+	o.getBroadcaster().SendResponse(priv, getFailedResponse(req.ID), txSig)
+	if ready {
+		o.getOnTransaction()(readyTx)
+	}
+}

pkg/services/oracle/response.go

@@ -0,0 +1,163 @@
+package oracle
+
+import (
+	"encoding/hex"
+	"errors"
+	gio "io"
+
+	"github.com/nspcc-dev/neo-go/pkg/core/fee"
+	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/hash"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
+	"github.com/nspcc-dev/neo-go/pkg/io"
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract/callflag"
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract/manifest"
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract/trigger"
+	"github.com/nspcc-dev/neo-go/pkg/vm"
+	"go.uber.org/zap"
+)
+
+func (o *Oracle) getResponse(reqID uint64, create bool) *incompleteTx {
+	o.respMtx.Lock()
+	defer o.respMtx.Unlock()
+	incTx, ok := o.responses[reqID]
+	if !ok && create && !o.removed[reqID] {
+		incTx = newIncompleteTx()
+		o.responses[reqID] = incTx
+	}
+	return incTx
+}
+
+// AddResponse processes oracle response from node pub.
+// sig is response transaction signature.
+func (o *Oracle) AddResponse(pub *keys.PublicKey, reqID uint64, txSig []byte) {
+	incTx := o.getResponse(reqID, true)
+	if incTx == nil {
+		return
+	}
+
+	incTx.Lock()
+	isBackup := false
+	if incTx.tx != nil {
+		ok := pub.Verify(txSig, incTx.tx.GetSignedHash().BytesBE())
+		if !ok {
+			ok = pub.Verify(txSig, incTx.backupTx.GetSignedHash().BytesBE())
+			if !ok {
+				o.Log.Debug("invalid response signature",
+					zap.String("pub", hex.EncodeToString(pub.Bytes())))
+				incTx.Unlock()
+				return
+			}
+			isBackup = true
+		}
+	}
+	incTx.addResponse(pub, txSig, isBackup)
+	readyTx, ready := incTx.finalize(o.getOracleNodes(), false)
+	if ready {
+		ready = !incTx.isSent
+		incTx.isSent = true
+	}
+	incTx.Unlock()
+
+	if ready {
+		o.getOnTransaction()(readyTx)
+	}
+}
+
+// ErrResponseTooLarge is returned when response exceeds max allowed size.
+var ErrResponseTooLarge = errors.New("too big response")
+
+func readResponse(rc gio.ReadCloser, limit int) ([]byte, error) {
+	defer rc.Close()
+
+	buf := make([]byte, limit+1)
+	n, err := gio.ReadFull(rc, buf)
+	if err == gio.ErrUnexpectedEOF && n <= limit {
+		return buf[:n], nil
+	}
+	if err == nil || n > limit {
+		return nil, ErrResponseTooLarge
+	}
+	return nil, err
+}
+
+// CreateResponseTx creates unsigned oracle response transaction.
+func (o *Oracle) CreateResponseTx(gasForResponse int64, height uint32, resp *transaction.OracleResponse) (*transaction.Transaction, error) {
+	tx := transaction.New(o.Network, o.OracleResponse, 0)
+	tx.Nonce = uint32(resp.ID)
+	tx.ValidUntilBlock = height + transaction.MaxValidUntilBlockIncrement
+	tx.Attributes = []transaction.Attribute{{
+		Type:  transaction.OracleResponseT,
+		Value: resp,
+	}}
+
+	oracleSignContract := o.getOracleSignContract()
+	tx.Signers = []transaction.Signer{
+		{
+			Account: o.OracleHash,
+			Scopes:  transaction.None,
+		},
+		{
+			Account: hash.Hash160(oracleSignContract),
+			Scopes:  transaction.None,
+		},
+	}
+	tx.Scripts = []transaction.Witness{
+		{}, // native contract witness is fixed, second witness is set later.
+	}
+
+	// Calculate network fee.
+	size := io.GetVarSize(tx)
+	tx.Scripts = append(tx.Scripts, transaction.Witness{VerificationScript: oracleSignContract})
+
+	gasConsumed, ok := o.testVerify(tx)
+	if !ok {
+		return nil, errors.New("can't verify transaction")
+	}
+	tx.NetworkFee += gasConsumed
+
+	netFee, sizeDelta := fee.Calculate(o.Chain.GetPolicer().GetBaseExecFee(), tx.Scripts[1].VerificationScript)
+	tx.NetworkFee += netFee
+	size += sizeDelta
+
+	currNetFee := tx.NetworkFee + int64(size)*o.Chain.FeePerByte()
+	if currNetFee > gasForResponse {
+		attrSize := io.GetVarSize(tx.Attributes)
+		resp.Code = transaction.InsufficientFunds
+		resp.Result = nil
+		size = size - attrSize + io.GetVarSize(tx.Attributes)
+	}
+	tx.NetworkFee += int64(size) * o.Chain.FeePerByte() // 233
+
+	// Calculate system fee.
+	tx.SystemFee = gasForResponse - tx.NetworkFee
+	return tx, nil
+}
+
+func (o *Oracle) testVerify(tx *transaction.Transaction) (int64, bool) {
+	v := o.Chain.GetTestVM(trigger.Verification, tx, nil)
+	v.GasLimit = o.Chain.GetPolicer().GetMaxVerificationGAS()
+	v.LoadScriptWithHash(o.OracleScript, o.OracleHash, callflag.ReadStates)
+	v.Estack().PushVal(manifest.MethodVerify)
+
+	ok := isVerifyOk(v)
+	return v.GasConsumed(), ok
+}
+
+func isVerifyOk(v *vm.VM) bool {
+	if err := v.Run(); err != nil {
+		return false
+	}
+	if v.Estack().Len() != 1 {
+		return false
+	}
+	ok, err := v.Estack().Pop().Item().TryBool()
+	return err == nil && ok
+}
+
+func getFailedResponse(id uint64) *transaction.OracleResponse {
+	return &transaction.OracleResponse{
+		ID:   id,
+		Code: transaction.Error,
+	}
+}

pkg/services/oracle/testdata/oracle1.json

@@ -0,0 +1 @@
+{"version":"3.0","accounts":[{"address":"NMy1PN9GCdGc26YFG7JruYg7UBStw2pPKN","key":"6PYML6dDTMXJBD7ywRwiCAhseCPToWkMfvPUViuxiXM6s5oi7ggf4ho3AK","label":"","contract":{"script":"DCEDNxK01e1DnGA+TiGU3H4DKUuGliSz89/NuZCbVvA2u0wLQZVEDXg=","parameters":[{"name":"parameter0","type":"Signature"}],"deployed":false},"lock":false,"isdefault":false}],"scrypt":{"n":16384,"r":8,"p":8},"extra":{"Tokens":null}}

pkg/services/oracle/testdata/oracle2.json

@@ -0,0 +1 @@
+{"version":"3.0","accounts":[{"address":"NU7QxQXULbmZU7kaWUaeF3r9v3zimU42bV","key":"6PYKv77p5wihN64XaPB5Nbci1sCLV5CrzSu8GKv7UHXHRtytfLt8zfrMgT","label":"","contract":{"script":"DCEDEXzwIl4Jhvsj98GYIPFFiedeb1QdP8T79uSBSDNsiswLQZVEDXg=","parameters":[{"name":"parameter0","type":"Signature"}],"deployed":false},"lock":false,"isdefault":false}],"scrypt":{"n":16384,"r":8,"p":8},"extra":{"Tokens":null}}

pkg/services/oracle/transaction.go

@@ -0,0 +1,129 @@
+package oracle
+
+import (
+	"sync"
+	"time"
+
+	"github.com/nspcc-dev/neo-go/pkg/core/state"
+	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
+	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
+	"github.com/nspcc-dev/neo-go/pkg/io"
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract"
+	"github.com/nspcc-dev/neo-go/pkg/vm/emit"
+)
+
+type (
+	incompleteTx struct {
+		sync.RWMutex
+		// isSent is true tx was already broadcasted.
+		isSent bool
+		// attempts is how many times request was processed.
+		attempts int
+		// time is the time when request was last processed.
+		time time.Time
+		// request is oracle request.
+		request *state.OracleRequest
+		// tx is oracle response transaction.
+		tx *transaction.Transaction
+		// sigs contains signature from every oracle node.
+		sigs map[string]*txSignature
+		// backupTx is backup transaction.
+		backupTx *transaction.Transaction
+		// backupSigs contains signatures of backup tx.
+		backupSigs map[string]*txSignature
+	}
+
+	txSignature struct {
+		// pub is cached public key.
+		pub *keys.PublicKey
+		// ok is true if signature was verified.
+		ok bool
+		// sig is tx signature.
+		sig []byte
+	}
+)
+
+func newIncompleteTx() *incompleteTx {
+	return &incompleteTx{
+		sigs:       make(map[string]*txSignature),
+		backupSigs: make(map[string]*txSignature),
+	}
+}
+
+func (t *incompleteTx) reverifyTx() {
+	txHash := t.tx.GetSignedHash()
+	backupHash := t.backupTx.GetSignedHash()
+	for pub, sig := range t.sigs {
+		if !sig.ok {
+			sig.ok = sig.pub.Verify(sig.sig, txHash.BytesBE())
+			if !sig.ok && sig.pub.Verify(sig.sig, backupHash.BytesBE()) {
+				t.backupSigs[pub] = &txSignature{
+					pub: sig.pub,
+					ok:  true,
+					sig: sig.sig,
+				}
+			}
+		}
+	}
+}
+
+func (t *incompleteTx) addResponse(pub *keys.PublicKey, sig []byte, isBackup bool) {
+	tx, sigs := t.tx, t.sigs
+	if isBackup {
+		tx, sigs = t.backupTx, t.backupSigs
+	}
+	sigs[string(pub.Bytes())] = &txSignature{
+		pub: pub,
+		ok:  tx != nil,
+		sig: sig,
+	}
+
+}
+
+// finalize checks is either main or backup tx has sufficient number of signatures and returns
+// tx and bool value indicating if it is ready to be broadcasted.
+func (t *incompleteTx) finalize(oracleNodes keys.PublicKeys, backupOnly bool) (*transaction.Transaction, bool) {
+	if !backupOnly && finalizeTx(oracleNodes, t.tx, t.sigs) {
+		return t.tx, true
+	}
+	return t.backupTx, finalizeTx(oracleNodes, t.backupTx, t.backupSigs)
+}
+
+func finalizeTx(oracleNodes keys.PublicKeys, tx *transaction.Transaction, txSigs map[string]*txSignature) bool {
+	if tx == nil {
+		return false
+	}
+	m := smartcontract.GetDefaultHonestNodeCount(len(oracleNodes))
+	sigs := make([][]byte, 0, m)
+	for _, pub := range oracleNodes {
+		sig, ok := txSigs[string(pub.Bytes())]
+		if ok && sig.ok {
+			sigs = append(sigs, sig.sig)
+			if len(sigs) == m {
+				break
+			}
+		}
+	}
+	if len(sigs) != m {
+		return false
+	}
+
+	w := io.NewBufBinWriter()
+	for i := range sigs {
+		emit.Bytes(w.BinWriter, sigs[i])
+	}
+	tx.Scripts[1].InvocationScript = w.Bytes()
+	return true
+}
+
+func (t *incompleteTx) getRequest() *state.OracleRequest {
+	t.RLock()
+	defer t.RUnlock()
+	return t.request
+}
+
+func (t *incompleteTx) getTime() time.Time {
+	t.RLock()
+	defer t.RUnlock()
+	return t.time
+}