diff --git a/Command Line.go b/Command Line.go index aedbef0..167ac53 100644 --- a/Command Line.go +++ b/Command Line.go @@ -21,6 +21,7 @@ import ( "github.com/PeernetOfficial/core" "github.com/PeernetOfficial/core/dht" "github.com/PeernetOfficial/core/protocol" + "github.com/PeernetOfficial/core/webapi" "github.com/btcsuite/btcd/btcec" ) @@ -37,6 +38,7 @@ func showHelp(output io.Writer) { "debug watch searches Watch all outgoing DHT searches\n"+ "debug watch incoming Watch all incoming information requests\n"+ "debug watch Watch packets and info requests for hash\n"+ + "probe file transfer Attempts to transfer and validate a remote file against a local file\n"+ "hash Create blake3 hash of input\n"+ "warehouse get Get data from local warehouse by hash\n"+ "warehouse store Store data into local warehouse\n"+ @@ -350,6 +352,45 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct } else { fmt.Fprintf(output, "The hash was removed from the monitoring list.\n") } + + case "probe file transfer": + fmt.Fprintf(output, "Enter peer ID or node ID to connect:\n") + nodeIDA, _, terminate := getUserOptionString(reader, terminateSignal) + if terminate { + return + } + fmt.Fprintf(output, "Enter file hash:\n") + fileHashA, _, terminate := getUserOptionString(reader, terminateSignal) + if terminate { + return + } + + fileHash, valid1 := webapi.DecodeBlake3Hash(fileHashA) + nodeID, valid2 := webapi.DecodeBlake3Hash(nodeIDA) + publicKey, err3 := core.PublicKeyFromPeerID(nodeIDA) + + if !valid2 && err3 != nil { + fmt.Fprintf(output, "Invalid peer ID or node ID.\n") + break + } else if !valid1 { + fmt.Fprintf(output, "Invalid file hash.\n") + } + + var peer *core.PeerInfo + var err error + timeout := time.Second * 10 + + if valid2 { + peer, err = webapi.PeerConnectNode(nodeID, timeout) + } else if err3 == nil { + peer, err = webapi.PeerConnectPublicKey(publicKey, timeout) + } + if err != nil { + fmt.Fprintf(output, "Could not connect to peer: %s\n", err.Error()) + break + } + + go transferCompareFile(peer, fileHash) } } } diff --git a/File Transfer.go b/File Transfer.go new file mode 100644 index 0000000..2654c37 --- /dev/null +++ b/File Transfer.go @@ -0,0 +1,206 @@ +/* +File Name: File Transfer.go +Copyright: 2021 Peernet Foundation s.r.o. +Author: Peter Kleissner +*/ + +package main + +import ( + "bytes" + "encoding/hex" + "fmt" + "time" + + "github.com/PeernetOfficial/core" + "github.com/PeernetOfficial/core/udt" + "github.com/PeernetOfficial/core/warehouse" +) + +// transferCompareFile downloads a file from a remote peer and compares it with the same file in the local warehouse. +// This function exists to test a file transfer. +// Note: The file MUST be stored locally, otherwise this function fails. +func transferCompareFile(peer *core.PeerInfo, fileHash []byte) { + // check if the file exists locally + _, fileInfo, status, _ := core.UserWarehouse.FileExists(fileHash) + if status != warehouse.StatusOK { + fmt.Printf("File does not exist in local warehouse: %s\n", hex.EncodeToString(fileHash)) + return + } + expectedSize := fileInfo.Size() + + // peer must be connected + if !peer.IsConnectionActive() { + fmt.Printf("Peer has no active connection: %s\n", hex.EncodeToString(peer.NodeID)) + return + } + + fmt.Printf("1. Peer connected: %s\n", hex.EncodeToString(peer.NodeID)) + + // request file transfer + udtConn, virtualConn, err := peer.FileTransferRequestUDT(fileHash, 0, 0) + if err != nil { + fmt.Printf("Error opening UDT connection: %s\n", err) + return + } + defer udtConn.Close() + + fmt.Printf("2. Opened UDT connection for file: %s\n", hex.EncodeToString(fileHash)) + + fileSize, transferSize, err := core.FileTransferReadHeaderUDT(udtConn) + if err != nil { + fmt.Printf("Error reading file transfer header: %s\n", err) + return + } + + if fileSize != uint64(expectedSize) { + fmt.Printf("Error expected local file size %d mismatch with remote file size %d\n", expectedSize, fileSize) + return + } else if fileSize != transferSize { + fmt.Printf("Error remote peer only offering %d of total file size %d\n", transferSize, fileSize) + return + } + + fmt.Printf("3. Matching transfer size %d and file size %d\n", transferSize, expectedSize) + + // Previous: Loop in explicitly 512 bytes (which is the same buffer as io.Copy apparently) and compare with what is expected. + // Now use 4 KB buffer. + fileOffset := 0 + totalRead := 0 + timeStart := time.Now() + timeUpdateLast := time.Now() + dataRemaining := fileSize + + for { + maxSize := uint64(4096) + if dataRemaining < maxSize { + maxSize = dataRemaining + } + + data := make([]byte, maxSize) + n, err := udtConn.Read(data) + + totalRead += n + dataRemaining -= uint64(n) + data = data[:n] + + if err != nil { + fmt.Printf("-- TERMINATE: ERROR READING. Read %d bytes. Total read %d : %v\n", n, fileOffset+n, err) + break + } else if n == 0 { + fmt.Printf("-- TERMINATE: EMPTY READ but no error indicated. Read %d bytes. Total read %d : %v\n", n, fileOffset+n, err) + break + } else if dataRemaining <= 0 { + fmt.Printf("-- TERMINATE: EVERYTHING READ. Read %d bytes. Total read %d : %v\n", n, fileOffset+n, err) + break + } + + // read the exact piece from the local file for comparison + dataCompare := make([]byte, 0, n) + compareBuffer := bytes.NewBuffer(dataCompare) + _, bytesRead, err := core.UserWarehouse.ReadFile(fileHash, int64(fileOffset), int64(n), compareBuffer) + if err != nil { + fmt.Printf("Warehouse error reading at offset %d length %d: %v\n", fileOffset, n, err) + break + } else if int(bytesRead) != n { + fmt.Printf("Warehouse did not read full data. Requested %d, provided %d.\n", n, bytesRead) + break + } + dataCompare = compareBuffer.Bytes() + + // make the comparison + if !bytes.Equal(data, dataCompare) { + fmt.Printf("Offset %08X read %d DATA MISMATCH:\n", fileOffset, n) + fmt.Printf("---- DATA FROM REMOTE:\n%s\n", hex.Dump(data)) + fmt.Printf("---- DATA FROM LOCAL WAREHOUSE:\n%s\n", hex.Dump(dataCompare)) + + break + } + + // status update every few seconds + //fmt.Printf("Offset %08X read %d SUCCESS\n", fileOffset, n) + if time.Now().After(timeUpdateLast.Add(time.Second)) { + speed := float64(totalRead) / time.Since(timeStart).Seconds() / 1024 + fmt.Printf("Offset %08X progress %.2f %% MATCHING. Speed: %.2f KB/s\n", fileOffset, float64((fileOffset+n)*100)/float64(fileSize), speed) + timeUpdateLast = time.Now() + } + + fileOffset += n + } + + fmt.Printf("Terminate reason %d: %s\n", virtualConn.GetTerminateReason(), translateTerminateReason(virtualConn.GetTerminateReason())) + + speed := float64(totalRead) / time.Since(timeStart).Seconds() / 1024 + + fmt.Printf("Transfer took %s. Speed is %.2f KB/s\n", time.Since(timeStart).String(), speed) + + if totalRead != int(expectedSize) { + fmt.Printf("Error transferred data %d mismatch with reported file size %d\n", totalRead, fileSize) + return + } + + fmt.Printf("Finished reading total of %d bytes. Expected %d bytes.\n", totalRead, fileSize) +} + +func translateTerminateReason(reason int) string { + switch reason { + case 0: + return "Virtual connection does not indicated a shutdown." + case 404: + return "Remote peer does not store the file." + case 2: + return "Remote termination signal (upstream)" + case 3: + return "Sequence invalidation or expiration (upstream)" + + case udt.TerminateReasonListenerClosed: + return "Listener: The listener.Close function was called." + case udt.TerminateReasonLingerTimerExpired: + return "Socket: The linger timer expired." + case udt.TerminateReasonConnectTimeout: + return "Socket: The connection timed out when sending the initial handshake." + case udt.TerminateReasonRemoteSentShutdown: + return "Remote peer sent a shutdown message." + case udt.TerminateReasonCannotProcessOutgoing: + return "Send: Cannot process outgoing messages." + case udt.TerminateReasonInvalidPacketIDAck: + return "Send: Invalid packet ID received in ACK message." + case udt.TerminateReasonInvalidPacketIDNak: + return "Send: Invalid packet ID received in NAK message." + case udt.TerminateReasonCorruptPacketNak: + return "Send: Invalid NAK packet received." + default: + return "Unknown." + } +} + +/* +// downloadFile downloads the file from the target peer +func downloadFile(publicKey *btcec.PublicKey, hash []byte) (data []byte, err error) { + peer := core.PeerlistLookup(publicKey) + if peer == nil { + return nil, errors.New("peer not connected") + } + + udtConn, _, err := peer.FileTransferRequestUDT(hash, 0, 0) + if err != nil { + return nil, err + } + defer udtConn.Close() + + fileSize, transferSize, err := core.FileTransferReadHeaderUDT(udtConn) + if err != nil { + return nil, err + } + + fmt.Printf("* Indicated file size = %d. Target transfer size = %d\n", fileSize, transferSize) + + // read all data + data = make([]byte, transferSize) + n, err := udtConn.Read(data) + + fmt.Printf("* Read %d bytes (target %d), error: %v\n", n, transferSize, err) + + return data, err +} +*/ diff --git a/go.mod b/go.mod index 968a06f..bec75b1 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/PeernetOfficial/Cmd go 1.16 require ( - github.com/PeernetOfficial/core v0.0.0-20211022145503-f1e30c942f34 + github.com/PeernetOfficial/core v0.0.0-20211109200406-ce3746170520 github.com/btcsuite/btcd v0.22.0-beta.0.20210625194946-86a17263b0ff github.com/gorilla/websocket v1.4.3-0.20210424162022-e8629af678b7 ) diff --git a/go.sum b/go.sum index e422abd..b31c20a 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -github.com/PeernetOfficial/core v0.0.0-20211022145503-f1e30c942f34 h1:wqkSKcA1zjzwjAfY7mmdxi9OZmoZswXSWtVW9ZrkhR4= -github.com/PeernetOfficial/core v0.0.0-20211022145503-f1e30c942f34/go.mod h1:u5RB1/k3V5C6PwNuVeQLVAkW8agwpjmmiucWZwk08s4= +github.com/PeernetOfficial/core v0.0.0-20211109200406-ce3746170520 h1:r/IwB+hs/Lr8zdDlvXh6XyKu6+gXqobCEOa1M3TJPqA= +github.com/PeernetOfficial/core v0.0.0-20211109200406-ce3746170520/go.mod h1:u5RB1/k3V5C6PwNuVeQLVAkW8agwpjmmiucWZwk08s4= github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII= github.com/akrylysov/pogreb v0.10.1 h1:FqlR8VR7uCbJdfUob916tPM+idpKgeESDXOA1K0DK4w= github.com/akrylysov/pogreb v0.10.1/go.mod h1:pNs6QmpQ1UlTJKDezuRWmaqkgUE2TuU0YTWyqJZ7+lI=