data_socket.go

// Copyright 2018 The goftp Authors. All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.

package ftp

import (
	"crypto/tls"
	"errors"
	"io"
	"net"
	"os"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/globalcyberalliance/ftp-go/ratelimit"
)

type (
	// DataSocket describes a data socket is used to send non-control data between the client and server.
	DataSocket interface {
		Host() string

		Port() int

		// Read implements the standard io.Reader interface.
		Read(p []byte) (n int, err error)

		// ReadFrom implements the standard io.ReaderFrom interface.
		ReadFrom(r io.Reader) (int64, error)

		// Write implements the standard io.Writer interface.
		Write(p []byte) (n int, err error)

		// Close implements the standard io.Closer interface.
		Close() error
	}

	activeSocket struct {
		conn   *net.TCPConn
		reader io.Reader
		writer io.Writer
		sess   *Session
		host   string
		port   int
	}
)

func newActiveSocket(sess *Session, remote string, port int) (DataSocket, error) {
	connectTo := net.JoinHostPort(remote, strconv.Itoa(port))

	sess.log("Opening active data connection to " + connectTo)

	raddr, err := net.ResolveTCPAddr("tcp", connectTo)
	if err != nil {
		sess.log(err)
		return nil, err
	}

	tcpConn, err := net.DialTCP("tcp", nil, raddr)
	if err != nil {
		sess.log(err)
		return nil, err
	}

	socket := new(activeSocket)
	socket.sess = sess
	socket.conn = tcpConn
	socket.reader = ratelimit.Reader(tcpConn, sess.server.rateLimiter)
	socket.writer = ratelimit.Writer(tcpConn, sess.server.rateLimiter)
	socket.host = remote
	socket.port = port

	return socket, nil
}

func (socket *activeSocket) Host() string {
	return socket.host
}

func (socket *activeSocket) Port() int {
	return socket.port
}

func (socket *activeSocket) Read(p []byte) (n int, err error) {
	return socket.reader.Read(p)
}

func (socket *activeSocket) ReadFrom(r io.Reader) (int64, error) {
	return io.Copy(socket.writer, r)
}

func (socket *activeSocket) Write(p []byte) (n int, err error) {
	return socket.writer.Write(p)
}

func (socket *activeSocket) Close() error {
	return socket.conn.Close()
}

type passiveSocket struct {
	conn    net.Conn
	reader  io.Reader
	writer  io.Writer
	err     error
	sess    *Session
	ingress chan []byte
	egress  chan []byte
	host    string
	port    int
	lock    sync.Mutex // protects conn and err
}

// isErrorAddressAlreadyInUse detects if an error is "bind: address already in use"
//
// Originally from https://stackoverflow.com/a/52152912/164234
func isErrorAddressAlreadyInUse(err error) bool {
	errOpError, ok := err.(*net.OpError)
	if !ok {
		return false
	}

	var errSyscallError *os.SyscallError
	if !errors.As(errOpError.Err, &errSyscallError) {
		return false
	}

	var errErrno syscall.Errno
	if !errors.As(errSyscallError.Err, &errErrno) {
		return false
	}

	if errors.Is(errErrno, syscall.EADDRINUSE) {
		return true
	}

	const WSAEADDRINUSE = 10048
	if runtime.GOOS == "windows" && errErrno == WSAEADDRINUSE {
		return true
	}

	return false
}

func (sess *Session) newPassiveSocket() (DataSocket, error) {
	socket := new(passiveSocket)
	socket.ingress = make(chan []byte)
	socket.egress = make(chan []byte)
	socket.sess = sess
	socket.host = sess.passiveListenIP()

	const retries = 10
	var err error
	for i := 1; i <= retries; i++ {
		socket.port = sess.PassivePort()
		err = socket.ListenAndServe()
		if err != nil && socket.port != 0 && isErrorAddressAlreadyInUse(err) {
			// Choose a different port on error already in use.
			continue
		}
		break
	}
	sess.dataConn = socket
	return socket, err
}

func (socket *passiveSocket) Host() string {
	return socket.host
}

func (socket *passiveSocket) Port() int {
	return socket.port
}

func (socket *passiveSocket) Read(p []byte) (n int, err error) {
	socket.lock.Lock()
	defer socket.lock.Unlock()
	if socket.err != nil {
		return 0, socket.err
	}
	return socket.reader.Read(p)
}

func (socket *passiveSocket) ReadFrom(r io.Reader) (int64, error) {
	socket.lock.Lock()
	defer socket.lock.Unlock()
	if socket.err != nil {
		return 0, socket.err
	}

	// For normal TCPConn, this will use sendfile syscall; if not, it will just downgrade to normal read/write
	// procedure.
	return io.Copy(socket.writer, r)
}

func (socket *passiveSocket) Write(p []byte) (n int, err error) {
	socket.lock.Lock()
	defer socket.lock.Unlock()
	if socket.err != nil {
		return 0, socket.err
	}
	return socket.writer.Write(p)
}

func (socket *passiveSocket) Close() error {
	socket.lock.Lock()
	defer socket.lock.Unlock()
	if socket.conn != nil {
		return socket.conn.Close()
	}
	return nil
}

func (socket *passiveSocket) ListenAndServe() (err error) {
	laddr, err := net.ResolveTCPAddr("tcp", net.JoinHostPort("", strconv.Itoa(socket.port)))
	if err != nil {
		socket.sess.log(err)
		return err
	}

	var tcplistener *net.TCPListener
	tcplistener, err = net.ListenTCP("tcp", laddr)
	if err != nil {
		socket.sess.log(err)
		return err
	}

	// The timeout, for a remote client to establish connection with a PASV style data connection.
	const acceptTimeout = 60 * time.Second
	err = tcplistener.SetDeadline(time.Now().Add(acceptTimeout))
	if err != nil {
		socket.sess.log(err)
		return err
	}

	var listener net.Listener = tcplistener
	add := listener.Addr()
	parts := strings.Split(add.String(), ":")
	port, err := strconv.Atoi(parts[len(parts)-1])
	if err != nil {
		socket.sess.log(err)
		return err
	}

	socket.port = port
	if socket.sess.server.tlsConfig != nil {
		listener = tls.NewListener(listener, socket.sess.server.tlsConfig)
	}

	socket.lock.Lock()

	go func() {
		defer socket.lock.Unlock()

		conn, err := listener.Accept()
		if err != nil {
			socket.err = err
			return
		}

		socket.err = nil
		socket.conn = conn
		socket.reader = ratelimit.Reader(socket.conn, socket.sess.server.rateLimiter)
		socket.writer = ratelimit.Writer(socket.conn, socket.sess.server.rateLimiter)

		_ = listener.Close()
	}()

	return nil
}