func (s *Session) Read(w io.Writer, sections []renterhost.RPCReadRequestSection) (err error) {
	defer wrapErr(&err, "Read")
	defer s.collectStats(renterhost.RPCReadID, &err)()
	if !s.isLocked() {
		return ErrNoContractLocked
	} else if !s.isRevisable() {
		return ErrContractFinalized
	} else if len(sections) == 0 {
		return nil
	}
	sectorAccesses := make(map[crypto.Hash]struct{})
	for _, sec := range sections {
		sectorAccesses[sec.MerkleRoot] = struct{}{}
	}
	sectorAccessPrice := s.host.SectorAccessPrice.Mul64(uint64(len(sectorAccesses)))
	var bandwidth uint64
	for _, sec := range sections {
		proofHashes := 2 * bits.Len64(merkle.SegmentsPerSector)
		bandwidth += uint64(sec.Length) + uint64(proofHashes)*crypto.HashSize
	}
	if bandwidth < renterhost.MinMessageSize {
		bandwidth = renterhost.MinMessageSize
	}
	bandwidthPrice := s.host.DownloadBandwidthPrice.Mul64(bandwidth)
	price := s.host.BaseRPCPrice.Add(sectorAccessPrice).Add(bandwidthPrice)
	if !s.sufficientFunds(price) {
		return ErrInsufficientFunds
	}
	rev := s.rev.Revision
	rev.NewRevisionNumber++
	newValid, newMissed := updateRevisionOutputs(&rev, price, types.ZeroCurrency)
	revisionHash := renterhost.HashRevision(rev)
	renterSig := ed25519hash.Sign(s.key, revisionHash)
	uploadBandwidth := 4096 + 4096*uint64(len(sections))
	downloadBandwidth := bandwidth + 4096*uint64(len(sections))
	s.extendBandwidthDeadline(uploadBandwidth, downloadBandwidth)
	req := &renterhost.RPCReadRequest{
		Sections:    sections,
		MerkleProof: true,
		NewRevisionNumber:    rev.NewRevisionNumber,
		NewValidProofValues:  newValid,
		NewMissedProofValues: newMissed,
		Signature:            renterSig,
	}
	if err := s.sess.WriteRequest(renterhost.RPCReadID, req); err != nil {
		return err
	}
	defer s.sess.WriteResponse(&renterhost.RPCReadStop, nil)
	var hostSig []byte
	for _, sec := range sections {
		msgReader, err := s.sess.RawResponse(4096 + uint64(sec.Length))
		if err != nil {
			return wrapResponseErr(err, "couldn't read sector data", "host rejected Read request")
		}
		lenbuf := make([]byte, 8)
		if _, err := io.ReadFull(msgReader, lenbuf); err != nil {
			return fmt.Errorf("couldn't read signature len: %w", err)
		}
		if n := binary.LittleEndian.Uint64(lenbuf); n > 0 {
			hostSig = make([]byte, n)
			if _, err := io.ReadFull(msgReader, hostSig); err != nil {
				return fmt.Errorf("couldn't read signature: %w", err)
			}
		}
		if _, err := io.ReadFull(msgReader, lenbuf); err != nil {
			return fmt.Errorf("couldn't read data len: %w", err)
		} else if binary.LittleEndian.Uint64(lenbuf) != uint64(sec.Length) {
			return errors.New("host sent wrong amount of sector data")
		}
		proofStart := int(sec.Offset) / merkle.SegmentSize
		proofEnd := int(sec.Offset+sec.Length) / merkle.SegmentSize
		rpv := merkle.NewRangeProofVerifier(proofStart, proofEnd)
		tee := io.TeeReader(io.LimitReader(msgReader, int64(sec.Length)), &segWriter{w: w})
		if _, err := rpv.ReadFrom(bufio.NewReaderSize(tee, 1<<16)); err != nil {
			return fmt.Errorf("couldn't stream sector data: %w", err)
		}
		if _, err := io.ReadFull(msgReader, lenbuf); err != nil {
			return fmt.Errorf("couldn't read proof len: %w", err)
		}
		if binary.LittleEndian.Uint64(lenbuf) != uint64(merkle.ProofSize(merkle.SegmentsPerSector, proofStart, proofEnd)) {
			return errors.New("invalid proof size")
		}
		proof := make([]crypto.Hash, binary.LittleEndian.Uint64(lenbuf))
		for i := range proof {
			if _, err := io.ReadFull(msgReader, proof[i][:]); err != nil {
				return fmt.Errorf("couldn't read Merkle proof: %w", err)
			}
		}
		if err := msgReader.VerifyTag(); err != nil {
			return err
		}
		if !rpv.Verify(proof, sec.MerkleRoot) {
			return ErrInvalidMerkleProof
		}
		if len(hostSig) > 0 {
			break
		}
	}
	if len(hostSig) == 0 {
		var resp renterhost.RPCReadResponse
		if err := s.sess.ReadResponse(&resp, 4096); err != nil {
			return wrapResponseErr(err, "couldn't read signature", "host rejected Read request")
		}
		hostSig = resp.Signature
	}
	if !ed25519hash.Verify(s.host.PublicKey.Ed25519(), revisionHash, hostSig) {
		return errors.New("host's signature is invalid")
	}
	s.rev.Revision = rev
	s.rev.Signatures[0].Signature = renterSig
	s.rev.Signatures[1].Signature = hostSig
	return nil
}
Read calls the Read RPC, writing the requested sections of sector data to w.
Merkle proofs are always requested.

Note that sector data is streamed to w before it has been validated. Callers
MUST check the returned error, and discard any data written to w if the error
is non-nil. Failure to do so may allow an attacker to inject malicious data.
package proto
import (
	"bufio"
	"crypto/ed25519"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"math/bits"
	"net"
	"sort"
	"time"
	"go.sia.tech/siad/crypto"
	"go.sia.tech/siad/modules"
	"go.sia.tech/siad/types"
	"lukechampine.com/us/ed25519hash"
	"lukechampine.com/us/hostdb"
	"lukechampine.com/us/merkle"
	"lukechampine.com/us/renterhost"
)
var (
	ErrInsufficientFunds = errors.New("insufficient funds")
	ErrInvalidMerkleProof = errors.New("host supplied invalid Merkle proof")
	ErrContractLocked = errors.New("contract is locked by another party")
	ErrNoContractLocked = errors.New("no contract locked")
	ErrContractFinalized = errors.New("contract cannot be revised further")
)
func wrapResponseErr(err error, readCtx, rejectCtx string) error {
	if errors.As(err, new(*renterhost.RPCError)) {
		return fmt.Errorf("%s: %w", rejectCtx, err)
	}
	if err != nil {
		return fmt.Errorf("%s: %w", readCtx, err)
	}
	return nil
}
type statsConn struct {
	net.Conn
	r, w uint64
}
func (sc *statsConn) Read(p []byte) (int, error) {
	n, err := sc.Conn.Read(p)
	sc.r += uint64(n)
	return n, err
}
func (sc *statsConn) Write(p []byte) (int, error) {
	n, err := sc.Conn.Write(p)
	sc.w += uint64(n)
	return n, err
}
type Session struct {
	sess        *renterhost.Session
	conn        *statsConn
	appendRoots []crypto.Hash
	latency       time.Duration
	readDeadline  time.Duration
	writeDeadline time.Duration
	stats         RPCStatsRecorder
	host   hostdb.ScannedHost
	height types.BlockHeight
	rev    ContractRevision
	key    ed25519.PrivateKey
}
func (s *Session) HostKey() hostdb.HostPublicKey { return s.host.PublicKey }
func (s *Session) Revision() ContractRevision { return s.rev }
func (s *Session) IsClosed() bool { return s.sess.IsClosed() }
func (s *Session) SetLatency(d time.Duration) { s.latency = d }
func (s *Session) SetReadDeadline(d time.Duration) { s.readDeadline = d }
func (s *Session) SetWriteDeadline(d time.Duration) { s.writeDeadline = d }
func (s *Session) extendDeadline(d time.Duration) {
	_ = s.conn.SetDeadline(time.Now().Add(s.latency + d))
}
func (s *Session) extendBandwidthDeadline(up, down uint64) {
	if up < renterhost.MinMessageSize {
		up = renterhost.MinMessageSize
	}
	if down < renterhost.MinMessageSize {
		down = renterhost.MinMessageSize
	}
	s.extendDeadline(s.writeDeadline*time.Duration(up) + s.readDeadline*time.Duration(down))
}
func (s *Session) SetRPCStatsRecorder(stats RPCStatsRecorder) { s.stats = stats }
func (s *Session) collectStats(id renterhost.Specifier, err *error) (record func()) {
	if s.stats == nil {
		return func() {}
	}
	stats := RPCStats{
		Host:      s.host.PublicKey,
		Contract:  s.rev.Revision.ID(),
		RPC:       id,
		Timestamp: time.Now(),
	}
	var startFunds types.Currency
	if s.rev.IsValid() {
		startFunds = s.rev.RenterFunds()
	}
	oldW, oldR := s.conn.w, s.conn.r
	return func() {
		stats.Err = *err
		stats.Elapsed = time.Since(stats.Timestamp)
		stats.Uploaded = s.conn.w - oldW
		stats.Downloaded = s.conn.r - oldR
		if s.rev.IsValid() && startFunds.Cmp(s.rev.RenterFunds()) > 0 {
			stats.Cost = startFunds.Sub(s.rev.RenterFunds())
		}
		s.stats.RecordRPCStats(stats)
	}
}
func (s *Session) call(rpcID renterhost.Specifier, req, resp renterhost.ProtocolObject) error {
	if err := s.sess.WriteRequest(rpcID, req); err != nil {
		return err
	}
	err := s.sess.ReadResponse(resp, 4096)
	return wrapResponseErr(err, fmt.Sprintf("couldn't read %v response", rpcID), fmt.Sprintf("host rejected %v request", rpcID))
}
func (s *Session) isLocked() bool    { return s.rev.IsValid() }
func (s *Session) isRevisable() bool { return s.rev.Revision.NewRevisionNumber < math.MaxUint64 }
func (s *Session) sufficientFunds(price types.Currency) bool {
	if !s.rev.IsValid() {
		panic("sufficientFunds called with invalid revision")
	}
	renewPrice := s.host.BaseRPCPrice.Mul64(5)
	return s.rev.RenterFunds().Cmp(price.Add(renewPrice)) >= 0
}
func (s *Session) Lock(id types.FileContractID, key ed25519.PrivateKey, timeout time.Duration) (err error) {
	defer wrapErr(&err, "Lock")
	defer s.collectStats(renterhost.RPCLockID, &err)()
	req := &renterhost.RPCLockRequest{
		ContractID: id,
		Signature:  s.sess.SignChallenge(key),
		Timeout:    uint64(timeout.Milliseconds()),
	}
	s.extendDeadline(time.Duration(req.Timeout) * time.Millisecond)
	var resp renterhost.RPCLockResponse
	if err := s.call(renterhost.RPCLockID, req, &resp); err != nil {
		return err
	}
	s.sess.SetChallenge(resp.NewChallenge)
	if len(resp.Signatures) != 2 {
		return fmt.Errorf("host returned wrong number of signatures (expected 2, got %v)", len(resp.Signatures))
	}
	revHash := renterhost.HashRevision(resp.Revision)
	if !ed25519hash.Verify(ed25519hash.ExtractPublicKey(key), revHash, resp.Signatures[0].Signature) {
		return errors.New("renter's signature on claimed revision is invalid")
	} else if !ed25519hash.Verify(s.host.PublicKey.Ed25519(), revHash, resp.Signatures[1].Signature) {
		return errors.New("host's signature on claimed revision is invalid")
	}
	if !resp.Acquired {
		return ErrContractLocked
	}
	s.rev = ContractRevision{
		Revision:   resp.Revision,
		Signatures: [2]types.TransactionSignature{resp.Signatures[0], resp.Signatures[1]},
	}
	s.key = key
	if s.rev.Revision.NewRevisionNumber == math.MaxUint64 {
		return ErrContractFinalized
	}
	return nil
}
func (s *Session) Unlock() (err error) {
	defer wrapErr(&err, "Unlock")
	defer s.collectStats(renterhost.RPCUnlockID, &err)()
	if s.key == nil {
		return errors.New("no contract locked")
	}
	s.extendBandwidthDeadline(renterhost.MinMessageSize, renterhost.MinMessageSize)
	if err := s.sess.WriteRequest(renterhost.RPCUnlockID, nil); err != nil {
		return err
	}
	s.rev = ContractRevision{}
	s.key = nil
	return nil
}
func (s *Session) Settings() (_ hostdb.HostSettings, err error) {
	defer wrapErr(&err, "Settings")
	defer s.collectStats(renterhost.RPCSettingsID, &err)()
	s.extendBandwidthDeadline(renterhost.MinMessageSize, renterhost.MinMessageSize)
	var resp renterhost.RPCSettingsResponse
	if err := s.call(renterhost.RPCSettingsID, nil, &resp); err != nil {
		return hostdb.HostSettings{}, err
	} else if err := json.Unmarshal(resp.Settings, &s.host.HostSettings); err != nil {
		return hostdb.HostSettings{}, fmt.Errorf("couldn't unmarshal json: %w", err)
	}
	return s.host.HostSettings, nil
}
func (s *Session) SectorRoots(offset, n int) (_ []crypto.Hash, err error) {
	defer wrapErr(&err, "SectorRoots")
	defer s.collectStats(renterhost.RPCSectorRootsID, &err)()
	if !s.isLocked() {
		return nil, ErrNoContractLocked
	} else if !s.isRevisable() {
		return nil, ErrContractFinalized
	} else if offset < 0 || n < 0 || offset+n > s.rev.NumSectors() {
		return nil, errors.New("requested range is out-of-bounds")
	} else if n == 0 {
		return nil, nil
	}
	proofHashes := merkle.ProofSize(s.rev.NumSectors(), offset, offset+n)
	downloadBandwidth := uint64(proofHashes+n) * crypto.HashSize
	if downloadBandwidth < renterhost.MinMessageSize {
		downloadBandwidth = renterhost.MinMessageSize
	}
	bandwidthPrice := s.host.DownloadBandwidthPrice.Mul64(downloadBandwidth)
	price := s.host.BaseRPCPrice.Add(bandwidthPrice)
	if !s.sufficientFunds(price) {
		return nil, ErrInsufficientFunds
	}
	rev := s.rev.Revision
	rev.NewRevisionNumber++
	newValid, newMissed := updateRevisionOutputs(&rev, price, types.ZeroCurrency)
	revisionHash := renterhost.HashRevision(rev)
	s.extendBandwidthDeadline(renterhost.MinMessageSize, downloadBandwidth)
	req := &renterhost.RPCSectorRootsRequest{
		RootOffset: uint64(offset),
		NumRoots:   uint64(n),
		NewRevisionNumber:    rev.NewRevisionNumber,
		NewValidProofValues:  newValid,
		NewMissedProofValues: newMissed,
		Signature:            ed25519hash.Sign(s.key, revisionHash),
	}
	var resp renterhost.RPCSectorRootsResponse
	if err := s.sess.WriteRequest(renterhost.RPCSectorRootsID, req); err != nil {
		return nil, err
	}
	if err := s.sess.ReadResponse(&resp, uint64(4096+32*n)); err != nil {
		readCtx := fmt.Sprintf("couldn't read %v response", renterhost.RPCSectorRootsID)
		rejectCtx := fmt.Sprintf("host rejected %v request", renterhost.RPCSectorRootsID)
		return nil, wrapResponseErr(err, readCtx, rejectCtx)
	}
	if !ed25519hash.Verify(s.host.PublicKey.Ed25519(), revisionHash, resp.Signature) {
		return nil, errors.New("host's signature is invalid")
	}
	s.rev.Revision = rev
	s.rev.Signatures[0].Signature = req.Signature
	s.rev.Signatures[1].Signature = resp.Signature
	if !merkle.VerifySectorRangeProof(resp.MerkleProof, resp.SectorRoots, offset, offset+n, s.rev.NumSectors(), rev.NewFileMerkleRoot) {
		return nil, ErrInvalidMerkleProof
	}
	return resp.SectorRoots, nil
}
type segWriter struct {
	w   io.Writer
	buf [merkle.SegmentSize * 64]byte
	len int
}
func (sw *segWriter) Write(p []byte) (int, error) {
	lenp := len(p)
	for len(p) > 0 {
		n := copy(sw.buf[sw.len:], p)
		sw.len += n
		p = p[n:]
		segs := sw.buf[:sw.len-(sw.len%merkle.SegmentSize)]
		if _, err := sw.w.Write(segs); err != nil {
			return 0, err
		}
		sw.len = copy(sw.buf[:], sw.buf[len(segs):sw.len])
	}
	return lenp, nil
}