package oracle import ( "context" "errors" "mime" "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" "github.com/nspcc-dev/neo-go/pkg/services/oracle/neofs" "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() if !o.running { for _, id := range ids { delete(o.pending, id) } } else { 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 } o.respMtx.Lock() if !o.running { for id, r := range reqs { o.pending[id] = r } o.respMtx.Unlock() return } o.respMtx.Unlock() 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 the 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, Code: transaction.Success} u, err := url.ParseRequestURI(req.Req.URL) if err != nil { o.Log.Warn("malformed oracle request", zap.String("url", req.Req.URL), zap.Error(err)) resp.Code = transaction.ProtocolNotSupported } else { switch u.Scheme { case "https": httpReq, err := http.NewRequest("GET", req.Req.URL, nil) if err != nil { o.Log.Warn("failed to create http request", zap.String("url", req.Req.URL), zap.Error(err)) resp.Code = transaction.Error break } httpReq.Header.Set("User-Agent", "NeoOracleService/3.0") httpReq.Header.Set("Content-Type", "application/json") r, err := o.Client.Do(httpReq) if err != nil { if errors.Is(err, ErrRestrictedRedirect) { resp.Code = transaction.Forbidden } else { resp.Code = transaction.Error } o.Log.Warn("oracle request failed", zap.String("url", req.Req.URL), zap.Error(err), zap.Stringer("code", resp.Code)) break } defer r.Body.Close() switch r.StatusCode { case http.StatusOK: if !checkMediaType(r.Header.Get("Content-Type"), o.MainCfg.AllowedContentTypes) { resp.Code = transaction.ContentTypeNotSupported break } resp.Result, resp.Code = o.readResponse(r.Body, req.Req.URL) case http.StatusForbidden: resp.Code = transaction.Forbidden case http.StatusNotFound: resp.Code = transaction.NotFound case http.StatusRequestTimeout: resp.Code = transaction.Timeout default: resp.Code = transaction.Error } case neofs.URIScheme: ctx, cancel := context.WithTimeout(context.Background(), o.MainCfg.NeoFS.Timeout) defer cancel() index := (int(req.ID) + incTx.attempts) % len(o.MainCfg.NeoFS.Nodes) rc, err := neofs.Get(ctx, priv, u, o.MainCfg.NeoFS.Nodes[index]) if err != nil { resp.Code = transaction.Error o.Log.Warn("failed to perform oracle request", zap.String("url", req.Req.URL), zap.Error(err)) if rc != nil { rc.Close() // intentionally skip the closing error, make it unified with Oracle `https` protocol. } break } resp.Result, resp.Code = o.readResponse(rc, req.Req.URL) rc.Close() // intentionally skip the closing error, make it unified with Oracle `https` protocol. default: resp.Code = transaction.ProtocolNotSupported o.Log.Warn("unknown oracle request scheme", zap.String("url", req.Req.URL)) } } if resp.Code == transaction.Success { resp.Result, err = filterRequest(resp.Result, req.Req) if err != nil { o.Log.Warn("oracle filter failed", zap.Uint64("request", req.ID), zap.Error(err)) resp.Code = transaction.Error } } o.Log.Debug("oracle request processed", zap.String("url", req.Req.URL), zap.Int("code", int(resp.Code)), zap.String("result", string(resp.Result))) currentHeight := o.Chain.BlockHeight() vubInc := o.Chain.GetConfig().MaxValidUntilBlockIncrement _, 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 that it hasn't been persisted from DAO yet. h = currentHeight } h += vubInc // Main tx is only valid for RequestHeight + ValidUntilBlock. tx, err := o.CreateResponseTx(int64(req.Req.GasForResponse), h, resp) if err != nil { return err } for h <= currentHeight { // Backup tx must be valid in any event. h += vubInc } 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(o.Network) txSig := priv.SignHashable(uint32(o.Network), tx) incTx.addResponse(priv.PublicKey(), txSig, false) backupSig := priv.SignHashable(uint32(o.Network), backupTx) 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.ResponseHandler.SendResponse(priv, resp, txSig) if ready { o.sendTx(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.sendTx(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.ResponseHandler.SendResponse(priv, getFailedResponse(req.ID), txSig) if ready { o.sendTx(readyTx) } } func checkMediaType(hdr string, allowed []string) bool { if len(allowed) == 0 { return true } typ, _, err := mime.ParseMediaType(hdr) if err != nil { return false } for _, ct := range allowed { if ct == typ { return true } } return false }