From 9ddca35277d592f4a40d0a21c8831c8bb9c0c810 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Thu, 28 Oct 2021 04:06:10 +0200 Subject: [PATCH] Transfer: Set max packet size. udt: Improve error messages. Fix bug in udtSocketSend.processDataMsg, the header was not considered in the max size. webapi: New /file/view API which is useful for immediate testing of file transfer. --- Transfer UDT.go | 12 +++- udt/packet/packet.go | 4 +- udt/packet/packet_ack.go | 6 +- udt/packet/packet_data.go | 4 +- udt/packet/packet_handshake.go | 4 +- udt/packet/packet_lightack.go | 4 +- udt/packet/packet_msgdropreq.go | 4 +- udt/packet/packet_nak.go | 2 +- udt/packet/packet_userdefctrl.go | 2 +- udt/udtsocket_send.go | 15 ++--- webapi/API.go | 1 + webapi/File IO.go | 100 +++++++++++++++++++++++++++---- webapi/HTTP Range.go | 15 +++++ 13 files changed, 135 insertions(+), 38 deletions(-) diff --git a/Transfer UDT.go b/Transfer UDT.go index d10dc02..54126a2 100644 --- a/Transfer UDT.go +++ b/Transfer UDT.go @@ -44,13 +44,16 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset, virtualConnection.sequenceNumber = sequenceNumber networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate) + udtConfig := udt.DefaultConfig() + udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize + // start UDT sender - udtConn, err := udt.DialUDT(udt.DefaultConfig(), virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan, true) + udtConn, err := udt.DialUDT(udtConfig, virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan, true) if err != nil { return err } - defer udtConn.Close() // warning: This is currently blocking in case the other side does not call Close(). + defer udtConn.Close() // Start by sending the header: Total File Size and Transfer Size. header := make([]byte, 16) @@ -78,8 +81,11 @@ func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64) } virtualConnection.sequenceNumber = sequence.SequenceNumber + udtConfig := udt.DefaultConfig() + udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize + // start UDT receiver - udtListener := udt.ListenUDT(udt.DefaultConfig(), virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan) + udtListener := udt.ListenUDT(udtConfig, virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan) // request file transfer peer.sendTransfer(nil, protocol.TransferControlRequestStart, virtualConnection.transferProtocol, hash, offset, limit, virtualConnection.sequenceNumber) diff --git a/udt/packet/packet.go b/udt/packet/packet.go index fa02294..bbff770 100644 --- a/udt/packet/packet.go +++ b/udt/packet/packet.go @@ -134,7 +134,7 @@ func (h *ctrlHeader) SetHeader(destSockID uint32, ts uint32) { func (h *ctrlHeader) writeHdrTo(buf []byte, msgType PacketType, info uint32) (uint, error) { l := len(buf) if l < 16 { - return 0, errors.New("packet too small") + return 0, errors.New("ctrl packet too small") } // Sets the flag bit to indicate this is a control packet @@ -151,7 +151,7 @@ func (h *ctrlHeader) writeHdrTo(buf []byte, msgType PacketType, info uint32) (ui func (h *ctrlHeader) readHdrFrom(data []byte) (addtlInfo uint32, err error) { l := len(data) if l < 16 { - return 0, errors.New("packet too small") + return 0, errors.New("ctrl packet too small") } addtlInfo = endianness.Uint32(data[4:8]) h.ts = endianness.Uint32(data[8:12]) diff --git a/udt/packet/packet_ack.go b/udt/packet/packet_ack.go index 881cea2..c5626d9 100644 --- a/udt/packet/packet_ack.go +++ b/udt/packet/packet_ack.go @@ -25,7 +25,7 @@ type AckPacket struct { func (p *AckPacket) WriteTo(buf []byte) (uint, error) { l := len(buf) if l < 32 { - return 0, errors.New("packet too small") + return 0, errors.New("ack packet too small") } if _, err := p.writeHdrTo(buf, ptAck, p.AckSeqNo); err != nil { @@ -38,7 +38,7 @@ func (p *AckPacket) WriteTo(buf []byte) (uint, error) { endianness.PutUint32(buf[28:32], p.BuffAvail) if p.IncludeLink { if l < 40 { - return 0, errors.New("packet too small") + return 0, errors.New("ack packet too small") } endianness.PutUint32(buf[32:36], p.PktRecvRate) endianness.PutUint32(buf[36:40], p.EstLinkCap) @@ -51,7 +51,7 @@ func (p *AckPacket) WriteTo(buf []byte) (uint, error) { func (p *AckPacket) readFrom(data []byte) (err error) { l := len(data) if l < 32 { - return errors.New("packet too small") + return errors.New("ack packet too small") } if p.AckSeqNo, err = p.readHdrFrom(data); err != nil { return err diff --git a/udt/packet/packet_data.go b/udt/packet/packet_data.go index b97423a..2ede2c9 100644 --- a/udt/packet/packet_data.go +++ b/udt/packet/packet_data.go @@ -65,7 +65,7 @@ func (dp *DataPacket) WriteTo(buf []byte) (uint, error) { l := len(buf) ol := 16 + len(dp.Data) if l < ol { - return 0, errors.New("packet too small") + return 0, errors.New("write data packet too small") } endianness.PutUint32(buf[0:4], dp.Seq.Seq&0x7FFFFFFF) endianness.PutUint32(buf[4:8], dp.msg) @@ -79,7 +79,7 @@ func (dp *DataPacket) WriteTo(buf []byte) (uint, error) { func (dp *DataPacket) readFrom(data []byte) (err error) { l := len(data) if l < 16 { - return errors.New("packet too small") + return errors.New("read data packet too small") } //dp.seq = endianness.Uint32(data[0:4]) dp.msg = endianness.Uint32(data[4:8]) diff --git a/udt/packet/packet_handshake.go b/udt/packet/packet_handshake.go index bfa19f0..c452870 100644 --- a/udt/packet/packet_handshake.go +++ b/udt/packet/packet_handshake.go @@ -38,7 +38,7 @@ type HandshakePacket struct { func (p *HandshakePacket) WriteTo(buf []byte) (uint, error) { l := len(buf) if l < 64 { - return 0, errors.New("packet too small") + return 0, errors.New("handshake packet too small") } if _, err := p.writeHdrTo(buf, ptHandshake, 0); err != nil { @@ -64,7 +64,7 @@ func (p *HandshakePacket) WriteTo(buf []byte) (uint, error) { func (p *HandshakePacket) readFrom(data []byte) error { l := len(data) if l < 64 { - return errors.New("packet too small") + return errors.New("handshake packet too small") } if _, err := p.readHdrFrom(data); err != nil { return err diff --git a/udt/packet/packet_lightack.go b/udt/packet/packet_lightack.go index b203f7f..a2e94d0 100644 --- a/udt/packet/packet_lightack.go +++ b/udt/packet/packet_lightack.go @@ -16,7 +16,7 @@ type LightAckPacket struct { func (p *LightAckPacket) WriteTo(buf []byte) (uint, error) { l := len(buf) if l < 20 { - return 0, errors.New("packet too small") + return 0, errors.New("lightack packet too small") } if _, err := p.writeHdrTo(buf, ptAck, 0); err != nil { @@ -31,7 +31,7 @@ func (p *LightAckPacket) WriteTo(buf []byte) (uint, error) { func (p *LightAckPacket) readFrom(data []byte) (err error) { l := len(data) if l < 20 { - return errors.New("packet too small") + return errors.New("lightack packet too small") } if _, err = p.readHdrFrom(data); err != nil { return err diff --git a/udt/packet/packet_msgdropreq.go b/udt/packet/packet_msgdropreq.go index 8839865..780183f 100644 --- a/udt/packet/packet_msgdropreq.go +++ b/udt/packet/packet_msgdropreq.go @@ -18,7 +18,7 @@ type MsgDropReqPacket struct { func (p *MsgDropReqPacket) WriteTo(buf []byte) (uint, error) { l := len(buf) if l < 24 { - return 0, errors.New("packet too small") + return 0, errors.New("drop packet too small") } if _, err := p.writeHdrTo(buf, ptMsgDropReq, p.MsgID); err != nil { @@ -34,7 +34,7 @@ func (p *MsgDropReqPacket) WriteTo(buf []byte) (uint, error) { func (p *MsgDropReqPacket) readFrom(data []byte) (err error) { l := len(data) if l < 24 { - return errors.New("packet too small") + return errors.New("drop packet too small") } if p.MsgID, err = p.readHdrFrom(data); err != nil { return diff --git a/udt/packet/packet_nak.go b/udt/packet/packet_nak.go index fcd3b79..dd6506f 100644 --- a/udt/packet/packet_nak.go +++ b/udt/packet/packet_nak.go @@ -22,7 +22,7 @@ func (p *NakPacket) WriteTo(buf []byte) (uint, error) { l := uint(len(buf)) if l < off+uint(4*len(p.CmpLossInfo)) { - return 0, errors.New("packet too small") + return 0, errors.New("nak packet too small") } for _, elm := range p.CmpLossInfo { diff --git a/udt/packet/packet_userdefctrl.go b/udt/packet/packet_userdefctrl.go index ba132df..e6369b6 100644 --- a/udt/packet/packet_userdefctrl.go +++ b/udt/packet/packet_userdefctrl.go @@ -17,7 +17,7 @@ func (p *UserDefControlPacket) WriteTo(buf []byte) (uint, error) { l := len(buf) ol := 16 + len(p.data) if l < ol { - return 0, errors.New("packet too small") + return 0, errors.New("userdef packet too small") } // Sets the flag bit to indicate this is a control packet diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index 13dfec7..9ba9081 100644 --- a/udt/udtsocket_send.go +++ b/udt/udtsocket_send.go @@ -191,22 +191,17 @@ func (s *udtSocketSend) processDataMsg(isFirst bool, inChan <-chan sendMessage) s.msgSeq++ } - mtu := int(s.socket.maxPacketSize) + mtu := int(s.socket.maxPacketSize) - 16 msgLen := len(partialSend.content) if msgLen >= mtu { // we are full -- send what we can and leave the rest - var dp *packet.DataPacket + dp := &packet.DataPacket{ + Seq: s.sendPktSeq, + Data: partialSend.content[0:mtu], + } if msgLen == mtu { - dp = &packet.DataPacket{ - Seq: s.sendPktSeq, - Data: partialSend.content, - } s.msgPartialSend = nil } else { - dp = &packet.DataPacket{ - Seq: s.sendPktSeq, - Data: partialSend.content[0:mtu], - } s.msgPartialSend = &sendMessage{content: partialSend.content[mtu:], tim: partialSend.tim, ttl: partialSend.ttl} } s.sendPktSeq.Incr() diff --git a/webapi/API.go b/webapi/API.go index bf46a7a..61d6b9d 100644 --- a/webapi/API.go +++ b/webapi/API.go @@ -69,6 +69,7 @@ func Start(ListenAddresses []string, UseSSL bool, CertificateFile, CertificateKe Router.HandleFunc("/warehouse/read", apiWarehouseReadFile).Methods("GET") Router.HandleFunc("/warehouse/delete", apiWarehouseDeleteFile).Methods("GET") Router.HandleFunc("/file/read", apiFileRead).Methods("GET") + Router.HandleFunc("/file/view", apiFileView).Methods("GET") for _, listen := range ListenAddresses { go startWebServer(listen, UseSSL, CertificateFile, CertificateKey, Router, "API", TimeoutRead, TimeoutWrite) diff --git a/webapi/File IO.go b/webapi/File IO.go index 9365eaa..7b7f493 100644 --- a/webapi/File IO.go +++ b/webapi/File IO.go @@ -8,7 +8,6 @@ package webapi import ( "errors" - "fmt" "io" "net/http" "strconv" @@ -81,6 +80,7 @@ func apiFileRead(w http.ResponseWriter, r *http.Request) { return } + // start the reader reader, fileSize, transferSize, err := FileStartReader(peer, fileHash, uint64(offset), uint64(limit)) if reader != nil { defer reader.Close() @@ -89,19 +89,99 @@ func apiFileRead(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNotFound) return } - fmt.Printf("SUCCESS file size is %d transfer size is %d\n", fileSize, transferSize) - // Set the Content-Length header, always to the actual size of transferred data. - w.Header().Set("Content-Length", strconv.FormatUint(transferSize, 10)) + // set the right headers + setContentLengthRangeHeader(w, uint64(offset), transferSize, fileSize, ranges) - // Set the Content-Range header if needed. - if len(ranges) == 1 { - w.Header().Set("Content-Range", "bytes "+strconv.Itoa(offset)+"-"+strconv.Itoa(offset+int(transferSize)-1)+"/"+strconv.FormatUint(fileSize, 10)) - w.WriteHeader(http.StatusPartialContent) - } else { - w.WriteHeader(http.StatusOK) + // Start sending the data! + io.Copy(w, io.LimitReader(reader, int64(transferSize))) +} + +/* +apiFileView is similar to /file/read but but provides a format parameter. It set the Content-Type header and may manipulate the content on the fly. +This endpoint supports the Range, Content-Range and Content-Length headers. Multipart ranges are not supported and result in HTTP 400. +Instead of providing the node ID, the peer ID is also accepted in the &node= parameter. +The default timeout for connecting to the peer is 10 seconds. +Formats: 14 = Video + +Request: GET /file/view?hash=[hash]&node=[node ID]&format=[format] + Optional: &offset=[offset]&limit=[limit] or via Range header. + Optional: &timeout=[seconds] +Response: 200 with the content + 206 with partial content + 400 if the parameters are invalid + 404 if the file was not found or other error on transfer initiate + 502 if unable to find or connect to the remote peer in time +*/ +func apiFileView(w http.ResponseWriter, r *http.Request) { + r.ParseForm() + var err error + + // validate hashes (must be blake3) and other input + fileHash, valid1 := DecodeBlake3Hash(r.Form.Get("hash")) + nodeID, valid2 := DecodeBlake3Hash(r.Form.Get("node")) + publicKey, err3 := core.PublicKeyFromPeerID(r.Form.Get("node")) + if !valid1 || (!valid2 && err3 != nil) { + http.Error(w, "", http.StatusBadRequest) + return } + timeoutSeconds, _ := strconv.Atoi(r.Form.Get("timeout")) + if timeoutSeconds == 0 { + timeoutSeconds = 10 + } + timeout := time.Duration(timeoutSeconds) * time.Second + + offset, _ := strconv.Atoi(r.Form.Get("offset")) + limit, _ := strconv.Atoi(r.Form.Get("limit")) + format, _ := strconv.Atoi(r.Form.Get("format")) + + // Range header? + var ranges []HTTPRange + if ranges, err = ParseRangeHeader(r.Header.Get("Range"), -1, true); err != nil || len(ranges) > 1 { + http.Error(w, "", http.StatusBadRequest) + return + } else if len(ranges) == 1 { + if ranges[0].length != -1 { // if length is not specified, limit remains 0 which is maximum + limit = ranges[0].length + } + offset = ranges[0].start + } + + // try connecting via node ID or peer ID? + var peer *core.PeerInfo + + if valid2 { + peer, err = PeerConnectNode(nodeID, timeout) + } else if err3 == nil { + peer, err = PeerConnectPublicKey(publicKey, timeout) + } + if err != nil { + w.WriteHeader(http.StatusBadGateway) + return + } + + // start the reader + reader, fileSize, transferSize, err := FileStartReader(peer, fileHash, uint64(offset), uint64(limit)) + if reader != nil { + defer reader.Close() + } + if err != nil || reader == nil { + w.WriteHeader(http.StatusNotFound) + return + } + + w.Header().Set("Accept-Ranges", "bytes") // always indicate accepting of Range header + + switch format { + case 14: + // Video: Indicate MP4 always. There are tons of other MIME types that could be used. + w.Header().Set("Content-Type", "video/mp4") + } + + // set the right headers + setContentLengthRangeHeader(w, uint64(offset), transferSize, fileSize, ranges) + // Start sending the data! io.Copy(w, io.LimitReader(reader, int64(transferSize))) } diff --git a/webapi/HTTP Range.go b/webapi/HTTP Range.go index 38ab0be..6dbf473 100644 --- a/webapi/HTTP Range.go +++ b/webapi/HTTP Range.go @@ -8,6 +8,7 @@ package webapi import ( "errors" + "net/http" "net/textproto" "strconv" "strings" @@ -92,3 +93,17 @@ func ParseRangeHeader(s string, size int, noMultiRange bool) ([]HTTPRange, error } return ranges, nil } + +// setContentLengthRangeHeader sets the appropriate Content-Length and Content-Range headers +func setContentLengthRangeHeader(w http.ResponseWriter, offset, transferSize, fileSize uint64, ranges []HTTPRange) { + // Set the Content-Length header, always to the actual size of transferred data. + w.Header().Set("Content-Length", strconv.FormatUint(transferSize, 10)) + + // Set the Content-Range header if needed. + if len(ranges) == 1 { + w.Header().Set("Content-Range", "bytes "+strconv.FormatUint(offset, 10)+"-"+strconv.FormatUint(offset+transferSize-1, 10)+"/"+strconv.FormatUint(fileSize, 10)) + w.WriteHeader(http.StatusPartialContent) + } else { + w.WriteHeader(http.StatusOK) + } +}