diff --git a/cmd/swarm/main.go b/cmd/swarm/main.go index 886895852f..4233c7c982 100644 --- a/cmd/swarm/main.go +++ b/cmd/swarm/main.go @@ -560,7 +560,7 @@ func registerBzzService(bzzconfig *bzzapi.Config, ctx *cli.Context, stack *node. } // In production, mockStore must be always nil. - return swarm.NewSwarm(ctx, swapClient, ensClient, bzzconfig, bzzconfig.SwapEnabled, bzzconfig.SyncEnabled, bzzconfig.Cors, bzzconfig.PssEnabled, nil) + return swarm.NewSwarm(ctx, swapClient, ensClient, bzzconfig, bzzconfig.SwapEnabled, bzzconfig.SyncEnabled, bzzconfig.Cors, bzzconfig.PssEnabled, bzzconfig.ResourceEnabled, nil) } //register within the ethereum node if err := stack.Register(boot); err != nil { diff --git a/ethclient/ethclient.go b/ethclient/ethclient.go index 87a912901a..4b53e785d6 100644 --- a/ethclient/ethclient.go +++ b/ethclient/ethclient.go @@ -23,6 +23,7 @@ import ( "errors" "fmt" "math/big" + "strconv" "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" @@ -76,6 +77,15 @@ type rpcBlock struct { UncleHashes []common.Hash `json:"uncles"` } +func (ec *Client) BlockNumber(ctx context.Context) (uint64, error) { + var number string + err := ec.c.CallContext(ctx, &number, "eth_blockNumber") + if err != nil { + return 0, err + } + return strconv.ParseUint(number, 10, 64) +} + func (ec *Client) getBlock(ctx context.Context, method string, args ...interface{}) (*types.Block, error) { var raw json.RawMessage err := ec.c.CallContext(ctx, &raw, method, args...) diff --git a/swarm/api/api.go b/swarm/api/api.go index cfc26aeff5..0e11028ec4 100644 --- a/swarm/api/api.go +++ b/swarm/api/api.go @@ -365,13 +365,13 @@ func (self *Api) BuildDirectoryTree(mhash string, nameresolver bool) (key storag } // Look up mutable resource updates at specific periods and versions -func (self *Api) DbLookup(key storage.Key, name string, period uint32, version uint32) (io.ReadSeeker, error) { +func (self *Api) DbLookup(key storage.Key, name string, period uint32, version uint32) (storage.Key, io.ReadSeeker, int, error) { var err error if version != 0 { if period == 0 { currentblocknumber, err := self.resource.GetBlock() if err != nil { - return nil, fmt.Errorf("Could not determine latest block: %v", err) + return nil, nil, 0, fmt.Errorf("Could not determine latest block: %v", err) } period = self.resource.BlockToPeriod(name, currentblocknumber) } @@ -382,13 +382,13 @@ func (self *Api) DbLookup(key storage.Key, name string, period uint32, version u _, err = self.resource.LookupLatestByName(name, true) } if err != nil { - return nil, err + return nil, nil, 0, err } - data, err := self.resource.GetData(name) + key, data, err := self.resource.GetContent(name) if err != nil { - return nil, err + return nil, nil, 0, err } - return bytes.NewReader(data), nil + return key, bytes.NewReader(data), len(data), nil } func (self *Api) DbCreate(name string, frequency uint64) (err error) { @@ -406,3 +406,7 @@ func (self *Api) DbUpdate(name string, data []byte) (storage.Key, uint32, uint32 func (self *Api) DbHashSize() int { return self.resource.HashSize() } + +func (self *Api) DbIsValidated() bool { + return self.resource.IsValidated() +} diff --git a/swarm/api/config.go b/swarm/api/config.go index d4dba36094..31281f9bff 100644 --- a/swarm/api/config.go +++ b/swarm/api/config.go @@ -46,22 +46,23 @@ type Config struct { *network.HiveParams Swap *swap.SwapParams //*network.SyncParams - Contract common.Address - EnsRoot common.Address - EnsApi string - Path string - ListenAddr string - Port string - PublicKey string - BzzKey string - NetworkId uint64 - SwapEnabled bool - SyncEnabled bool - PssEnabled bool - SwapApi string - Cors string - BzzAccount string - BootNodes string + Contract common.Address + EnsRoot common.Address + EnsApi string + Path string + ListenAddr string + Port string + PublicKey string + BzzKey string + NetworkId uint64 + SwapEnabled bool + SyncEnabled bool + PssEnabled bool + ResourceEnabled bool + SwapApi string + Cors string + BzzAccount string + BootNodes string } //create a default config with all parameters to set to defaults @@ -72,18 +73,19 @@ func NewConfig() (self *Config) { ChunkerParams: storage.NewChunkerParams(), HiveParams: network.NewHiveParams(), //SyncParams: network.NewDefaultSyncParams(), - Swap: swap.NewDefaultSwapParams(), - ListenAddr: DefaultHTTPListenAddr, - Port: DefaultHTTPPort, - Path: node.DefaultDataDir(), - EnsApi: node.DefaultIPCEndpoint("geth"), - EnsRoot: ens.TestNetAddress, - NetworkId: network.NetworkID, - SwapEnabled: false, - SyncEnabled: true, - PssEnabled: true, - SwapApi: "", - BootNodes: "", + Swap: swap.NewDefaultSwapParams(), + ListenAddr: DefaultHTTPListenAddr, + Port: DefaultHTTPPort, + Path: node.DefaultDataDir(), + EnsApi: node.DefaultIPCEndpoint("geth"), + EnsRoot: ens.TestNetAddress, + NetworkId: network.NetworkID, + SwapEnabled: false, + SyncEnabled: true, + PssEnabled: true, + ResourceEnabled: true, + SwapApi: "", + BootNodes: "", } return diff --git a/swarm/api/http/server.go b/swarm/api/http/server.go index 9e3c7711c1..26c59c313e 100644 --- a/swarm/api/http/server.go +++ b/swarm/api/http/server.go @@ -327,7 +327,7 @@ func (s *Server) HandlePostDb(w http.ResponseWriter, r *Request) { func (s *Server) HandleGetDb(w http.ResponseWriter, r *Request) { w.Header().Set("Content-Type", "application/octet-stream") - key, err := s.api.Resolve(r.uri) + rootKey, err := s.api.Resolve(r.uri) if err != nil { s.Error(w, r, fmt.Errorf("error resolving %s: %s", r.uri.Addr, err)) return @@ -337,12 +337,15 @@ func (s *Server) HandleGetDb(w http.ResponseWriter, r *Request) { if len(r.uri.Path) > 0 { params = strings.Split(r.uri.Path, "/") } + var updateKey storage.Key var period uint64 var version uint64 var data io.ReadSeeker + var dataLength int + now := time.Now() switch len(params) { case 0: - data, err = s.api.DbLookup(key, r.uri.Addr, 0, 0) + updateKey, data, dataLength, err = s.api.DbLookup(rootKey, r.uri.Addr, 0, 0) break case 2: version, err = strconv.ParseUint(params[1], 10, 32) @@ -354,7 +357,7 @@ func (s *Server) HandleGetDb(w http.ResponseWriter, r *Request) { if err != nil { break } - data, err = s.api.DbLookup(key, r.uri.Addr, uint32(period), uint32(version)) + updateKey, data, dataLength, err = s.api.DbLookup(rootKey, r.uri.Addr, uint32(period), uint32(version)) break default: w.WriteHeader(http.StatusBadRequest) @@ -365,9 +368,32 @@ func (s *Server) HandleGetDb(w http.ResponseWriter, r *Request) { return } if !r.uri.DbRaw() { - + entry := api.ManifestEntry{ + Hash: rootKey.Hex(), + Path: updateKey.Hex(), + ContentType: api.DbManifestType, + Size: int64(dataLength), + ModTime: now, + Status: http.StatusOK, + } + mode := (6 << 2) | (4 << 1) | 4 + if s.api.DbIsValidated() { + mode |= 2 << 1 + } + entry.Mode = int64(mode) + manifest := api.Manifest{ + Entries: []api.ManifestEntry{ + entry, + }, + } + manifestJson, err := json.Marshal(manifest) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + data = bytes.NewReader(manifestJson) } - http.ServeContent(w, &r.Request, "", time.Now(), data) + http.ServeContent(w, &r.Request, "", now, data) } // HandleGet handles a GET request to diff --git a/swarm/api/http/server_test.go b/swarm/api/http/server_test.go index 888445d88d..e2165f1d61 100644 --- a/swarm/api/http/server_test.go +++ b/swarm/api/http/server_test.go @@ -78,6 +78,16 @@ func TestBzzGetDb(t *testing.T) { } b, err = ioutil.ReadAll(resp.Body) fmt.Printf("Get: %s : %s\n", resp.Status, b) + + url = fmt.Sprintf("%s/bzz-db:/%s", srv.URL, fmt.Sprintf("%x", keybytes)) + resp, err = http.Get(url) + if err != nil { + fmt.Printf("err: %v\n", err) + return + } + b, err = ioutil.ReadAll(resp.Body) + fmt.Printf("Get: %s : %s\n", resp.Status, b) + } func TestBzzGetPath(t *testing.T) { diff --git a/swarm/api/manifest.go b/swarm/api/manifest.go index 685a300fca..46b55b3b7a 100644 --- a/swarm/api/manifest.go +++ b/swarm/api/manifest.go @@ -33,7 +33,8 @@ import ( ) const ( - ManifestType = "application/bzz-manifest+json" + ManifestType = "application/bzz-manifest+json" + DbManifestType = "application/bzz-db-manifest+json" ) // Manifest represents a swarm manifest diff --git a/swarm/storage/resource.go b/swarm/storage/resource.go index b76117b336..23e1a6336b 100644 --- a/swarm/storage/resource.go +++ b/swarm/storage/resource.go @@ -1,10 +1,10 @@ package storage import ( + "context" "encoding/binary" "fmt" "path/filepath" - "strconv" "sync" "time" @@ -12,8 +12,8 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethclient" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/rpc" ) const ( @@ -37,6 +37,7 @@ type resource struct { nameHash common.Hash startBlock uint64 lastPeriod uint32 + lastKey Key frequency uint64 version uint32 data []byte @@ -114,26 +115,30 @@ type ResourceValidator interface { // stored using a separate store, and forwarding/syncing protocols carry per-chunk // flags to tell whether the chunk can be validated or not; if not it is to be // treated as a resource update chunk. +// +// TODO: Include modtime in chunk data + signature type ResourceHandler struct { ChunkStore validator ResourceValidator - rpcClient *rpc.Client + ethClient *ethclient.Client resources map[string]*resource hashLock sync.Mutex resourceLock sync.RWMutex hasher SwarmHash nameHash nameHashFunc storeTimeout time.Duration + ctx context.Context + cancelFunc func() } // Create or open resource update chunk store // // If validator is nil, signature and access validation will be deactivated -func NewResourceHandler(datadir string, cloudStore CloudStore, rpcClient *rpc.Client, validator ResourceValidator) (*ResourceHandler, error) { +func NewResourceHandler(datadir string, cloudStore CloudStore, ethClient *ethclient.Client, validator ResourceValidator) (*ResourceHandler, error) { hashfunc := MakeHashFunc(SHA3Hash) - path := filepath.Join(datadir, dbDirName) + path := filepath.Join(datadir, DbDirName) dbStore, err := NewDbStore(datadir, hashfunc, singletonSwarmDbCapacity, 0) if err != nil { return nil, err @@ -143,6 +148,7 @@ func NewResourceHandler(datadir string, cloudStore CloudStore, rpcClient *rpc.Cl DbStore: dbStore, } + ctx, cancel := context.WithCancel(context.Background()) rh := &ResourceHandler{ ChunkStore: newResourceChunkStore(path, hashfunc, localStore, cloudStore), rpcClient: rpcClient, @@ -150,6 +156,8 @@ func NewResourceHandler(datadir string, cloudStore CloudStore, rpcClient *rpc.Cl hasher: hashfunc(), validator: validator, storeTimeout: defaultStoreTimeout, + ctx: ctx, + cancelFunc: cancel, } if rh.validator != nil { @@ -167,17 +175,22 @@ func NewResourceHandler(datadir string, cloudStore CloudStore, rpcClient *rpc.Cl return rh, nil } +func (self *ResourceHandler) IsValidated() bool { + return self.validator == nil +} + func (self *ResourceHandler) HashSize() int { return self.validator.hashSize() } // get data from current resource -func (self *ResourceHandler) GetData(name string) ([]byte, error) { + +func (self *ResourceHandler) GetContent(name string) (Key, []byte, error) { rsrc := self.getResource(name) if rsrc == nil || !rsrc.isSynced() { - return nil, fmt.Errorf("Resource does not exist or is not synced") + return nil, nil, fmt.Errorf("Resource does not exist or is not synced") } - return rsrc.data, nil + return rsrc.lastKey, rsrc.data, nil } func (self *ResourceHandler) GetLastPeriod(name string) (uint32, error) { @@ -430,6 +443,7 @@ func (self *ResourceHandler) updateResourceIndex(rsrc *resource, chunk *Chunk) ( if *rsrc.name != name { return nil, fmt.Errorf("Update belongs to '%s', but have '%s'", name, *rsrc.name) } + log.Trace("update", "name", *rsrc.name, "rootkey", rsrc.nameHash, "updatekey", chunk.Key, "period", period, "version", version) // only check signature if validator is present if self.validator != nil { digest := self.keyDataHash(chunk.Key, data) @@ -440,6 +454,7 @@ func (self *ResourceHandler) updateResourceIndex(rsrc *resource, chunk *Chunk) ( } // update our rsrcs entry map + rsrc.lastKey = chunk.Key rsrc.lastPeriod = period rsrc.version = version rsrc.updated = time.Now() @@ -582,20 +597,22 @@ func (self *ResourceHandler) Update(name string, data []byte) (Key, error) { // Closes the datastore. // Always call this at shutdown to avoid data corruption. func (self *ResourceHandler) Close() { + self.cancelFunc() self.ChunkStore.Close() } func (self *ResourceHandler) GetBlock() (uint64, error) { + return self.ethClient.BlockNumber(self.ctx) // get the block height and convert to uint64 - var currentblock string - err := self.rpcClient.Call(¤tblock, "eth_blockNumber") - if err != nil { - return 0, err - } - if currentblock == "0x0" { - return 0, nil - } - return strconv.ParseUint(currentblock, 10, 64) + // var currentblock string + // err := self.rpcClient.Call(¤tblock, "eth_blockNumber") + // if err != nil { + // return 0, err + // } + // if currentblock == "0x0" { + // return 0, nil + // } + // return strconv.ParseUint(currentblock, 10, 64) } // Calculate the period index (aka major version number) from a given block number diff --git a/swarm/storage/resource_sign.go b/swarm/storage/resource_sign.go new file mode 100644 index 0000000000..840e705ac1 --- /dev/null +++ b/swarm/storage/resource_sign.go @@ -0,0 +1,20 @@ +package storage + +import ( + "crypto/ecdsa" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" +) + +// matches the SignFunc type +func NewGenericResourceSigner(privKey *ecdsa.PrivateKey) SignFunc { + return func(data common.Hash) (signature Signature, err error) { + signaturebytes, err := crypto.Sign(data.Bytes(), privKey) + if err != nil { + return + } + copy(signature[:], signaturebytes) + return + } +} diff --git a/swarm/storage/resource_test.go b/swarm/storage/resource_test.go index d130709428..2da9b7704a 100644 --- a/swarm/storage/resource_test.go +++ b/swarm/storage/resource_test.go @@ -22,6 +22,7 @@ import ( "github.com/ethereum/go-ethereum/contracts/ens/contract" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethclient" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/rpc" ) @@ -211,8 +212,8 @@ func TestResourceHandler(t *testing.T) { // it will match on second iteration startblocknumber + (resourceFrequency * 3) fwdBlocks(int(resourceFrequency*2)-1, backend) - rh2, err := NewResourceHandler(datadir, &testCloudStore{}, rh.rpcClient, nil) - _, err = rh2.LookupLatest(safeName, true) + rh2, err := NewResourceHandler(datadir, &testCloudStore{}, rh.ethClient, nil) + _, err = rh2.LookupLatestByName(safeName, true) if err != nil { teardownTest(t, err) } @@ -230,7 +231,7 @@ func TestResourceHandler(t *testing.T) { log.Debug("Latest lookup", "period", rh2.resources[safeName].lastPeriod, "version", rh2.resources[safeName].version, "data", rh2.resources[safeName].data) // specific block, latest version - rsrc, err := rh2.LookupHistorical(safeName, 3, true) + rsrc, err := rh2.LookupHistoricalByName(safeName, 3, true) if err != nil { teardownTest(t, err) } @@ -241,7 +242,7 @@ func TestResourceHandler(t *testing.T) { log.Debug("Historical lookup", "period", rh2.resources[safeName].lastPeriod, "version", rh2.resources[safeName].version, "data", rh2.resources[safeName].data) // specific block, specific version - rsrc, err = rh2.LookupVersion(safeName, 3, 1, true) + rsrc, err = rh2.LookupVersionByName(safeName, 3, 1, true) if err != nil { teardownTest(t, err) } @@ -365,12 +366,14 @@ func setupTest(contractbackend bind.ContractBackend, validator ResourceValidator } // connect to fake rpc - rpcclient, err := rpc.Dial(ipcpath) + rpcClient, err := rpc.Dial(ipcpath) if err != nil { return } - rh, err = NewResourceHandler(datadir, &testCloudStore{}, rpcclient, validator) + ethClient := ethclient.NewClient(rpcClient) + + rh, err = NewResourceHandler(datadir, &testCloudStore{}, ethClient, validator) teardown = func(t *testing.T, err error) { cleanF() if err != nil { @@ -426,8 +429,9 @@ func setupENS(addr common.Address, transactOpts *bind.TransactOpts, sub string, // implementation of an external signer to pass to validator type testSigner struct { - privKey *ecdsa.PrivateKey - hasher SwarmHash + privKey *ecdsa.PrivateKey + hasher SwarmHash + signContent SignFunc } func newTestSigner() (*testSigner, error) { @@ -436,21 +440,12 @@ func newTestSigner() (*testSigner, error) { return nil, err } return &testSigner{ - privKey: privKey, - hasher: testHasher, + privKey: privKey, + hasher: testHasher, + signContent: NewGenericResourceSigner(privKey), }, nil } -// matches the SignFunc type -func (self *testSigner) signContent(data common.Hash) (signature Signature, err error) { - signaturebytes, err := crypto.Sign(data.Bytes(), self.privKey) - if err != nil { - return - } - copy(signature[:], signaturebytes) - return -} - type testCloudStore struct { } diff --git a/swarm/swarm.go b/swarm/swarm.go index 14826bb021..766c210d3d 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -22,6 +22,7 @@ import ( "crypto/ecdsa" "fmt" "net" + "path/filepath" "github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/common" @@ -82,7 +83,8 @@ func (self *Swarm) API() *SwarmAPI { // implements node.Service // If mockStore is not nil, it will be used as the storage for chunk data. // MockStore should be used only for testing. -func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *ethclient.Client, config *api.Config, swapEnabled, syncEnabled bool, cors string, pssEnabled bool, mockStore *mock.NodeStore) (self *Swarm, err error) { +func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *ethclient.Client, config *api.Config, swapEnabled, syncEnabled bool, cors string, pssEnabled bool, resourceEnabled bool, mockStore *mock.NodeStore) (self *Swarm, err error) { + if bytes.Equal(common.FromHex(config.PublicKey), storage.ZeroKey) { return nil, fmt.Errorf("empty public key") } @@ -158,7 +160,23 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e } log.Debug(fmt.Sprintf("-> Swarm Domain Name Registrar @ address %v", config.EnsRoot.Hex())) - self.api = api.NewApi(self.dpa, self.dns) + var resourceHandler *storage.ResourceHandler + // if use resource updates + if resourceEnabled { + var resourceValidator storage.ResourceValidator + if self.dns != nil { + resourceValidator, err = storage.NewENSValidator(config.EnsRoot, ensClient, transactOpts, storage.NewGenericResourceSigner(self.privateKey)) + if err != nil { + return nil, err + } + } + resourceHandler, err = storage.NewResourceHandler(filepath.Join(self.config.Path, storage.DbDirName), self.cloud, ensClient, resourceValidator) + if err != nil { + return nil, err + } + } + + self.api = api.NewApi(self.dpa, self.dns, resourceHandler) // Manifests for Smart Hosting log.Debug(fmt.Sprintf("-> Web3 virtual server API")) @@ -360,7 +378,7 @@ func NewLocalSwarm(datadir, port string) (self *Swarm, err error) { } self = &Swarm{ - api: api.NewApi(dpa, nil), + api: api.NewApi(dpa, nil, nil), config: config, } diff --git a/swarm/testutil/http.go b/swarm/testutil/http.go index dcba90c9c5..4cec635936 100644 --- a/swarm/testutil/http.go +++ b/swarm/testutil/http.go @@ -25,6 +25,7 @@ import ( "strconv" "testing" + "github.com/ethereum/go-ethereum/ethclient" "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/api" httpapi "github.com/ethereum/go-ethereum/swarm/api/http" @@ -55,30 +56,32 @@ func NewTestSwarmServer(t *testing.T) *TestSwarmServer { dpa.Start() // mutable resources test setup - resourcedir, err := ioutil.TempDir("", "swarm-resource-test") + resourceDir, err := ioutil.TempDir("", "swarm-resource-test") if err != nil { t.Fatal(err) } - ipcpath := filepath.Join(resourcedir, "test.ipc") - ipcl, err := rpc.CreateIPCListener(ipcpath) + ipcPath := filepath.Join(resourceDir, "test.ipc") + ipcl, err := rpc.CreateIPCListener(ipcPath) if err != nil { t.Fatal(err) } - rpcserver := rpc.NewServer() - rpcserver.RegisterName("eth", &FakeRPC{}) + rpcServer := rpc.NewServer() + rpcServer.RegisterName("eth", &FakeRPC{}) go func() { - rpcserver.ServeListener(ipcl) + rpcServer.ServeListener(ipcl) }() rpcClean := func() { - rpcserver.Stop() + rpcServer.Stop() } // connect to fake rpc - rpcclient, err := rpc.Dial(ipcpath) + rpcClient, err := rpc.Dial(ipcPath) if err != nil { t.Fatal(err) } - rh, err := storage.NewResourceHandler(resourcedir, &testCloudStore{}, rpcclient, nil) + ethClient := ethclient.NewClient(rpcClient) + + rh, err := storage.NewResourceHandler(resourceDir, &testCloudStore{}, ethClient, nil) if err != nil { t.Fatal(err) } @@ -94,7 +97,7 @@ func NewTestSwarmServer(t *testing.T) *TestSwarmServer { rh.Close() rpcClean() os.RemoveAll(dir) - os.RemoveAll(resourcedir) + os.RemoveAll(resourceDir) }, } }