client.go 50 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030
  1. package sftp
  2. import (
  3. "bytes"
  4. "encoding/binary"
  5. "errors"
  6. "fmt"
  7. "io"
  8. "math"
  9. "os"
  10. "path"
  11. "sync"
  12. "sync/atomic"
  13. "syscall"
  14. "time"
  15. "github.com/kr/fs"
  16. "golang.org/x/crypto/ssh"
  17. )
  18. var (
  19. // ErrInternalInconsistency indicates the packets sent and the data queued to be
  20. // written to the file don't match up. It is an unusual error and usually is
  21. // caused by bad behavior server side or connection issues. The error is
  22. // limited in scope to the call where it happened, the client object is still
  23. // OK to use as long as the connection is still open.
  24. ErrInternalInconsistency = errors.New("internal inconsistency")
  25. // InternalInconsistency alias for ErrInternalInconsistency.
  26. //
  27. // Deprecated: please use ErrInternalInconsistency
  28. InternalInconsistency = ErrInternalInconsistency
  29. )
  30. // A ClientOption is a function which applies configuration to a Client.
  31. type ClientOption func(*Client) error
  32. // MaxPacketChecked sets the maximum size of the payload, measured in bytes.
  33. // This option only accepts sizes servers should support, ie. <= 32768 bytes.
  34. //
  35. // If you get the error "failed to send packet header: EOF" when copying a
  36. // large file, try lowering this number.
  37. //
  38. // The default packet size is 32768 bytes.
  39. func MaxPacketChecked(size int) ClientOption {
  40. return func(c *Client) error {
  41. if size < 1 {
  42. return errors.New("size must be greater or equal to 1")
  43. }
  44. if size > 32768 {
  45. return errors.New("sizes larger than 32KB might not work with all servers")
  46. }
  47. c.maxPacket = size
  48. return nil
  49. }
  50. }
  51. // MaxPacketUnchecked sets the maximum size of the payload, measured in bytes.
  52. // It accepts sizes larger than the 32768 bytes all servers should support.
  53. // Only use a setting higher than 32768 if your application always connects to
  54. // the same server or after sufficiently broad testing.
  55. //
  56. // If you get the error "failed to send packet header: EOF" when copying a
  57. // large file, try lowering this number.
  58. //
  59. // The default packet size is 32768 bytes.
  60. func MaxPacketUnchecked(size int) ClientOption {
  61. return func(c *Client) error {
  62. if size < 1 {
  63. return errors.New("size must be greater or equal to 1")
  64. }
  65. c.maxPacket = size
  66. return nil
  67. }
  68. }
  69. // MaxPacket sets the maximum size of the payload, measured in bytes.
  70. // This option only accepts sizes servers should support, ie. <= 32768 bytes.
  71. // This is a synonym for MaxPacketChecked that provides backward compatibility.
  72. //
  73. // If you get the error "failed to send packet header: EOF" when copying a
  74. // large file, try lowering this number.
  75. //
  76. // The default packet size is 32768 bytes.
  77. func MaxPacket(size int) ClientOption {
  78. return MaxPacketChecked(size)
  79. }
  80. // MaxConcurrentRequestsPerFile sets the maximum concurrent requests allowed for a single file.
  81. //
  82. // The default maximum concurrent requests is 64.
  83. func MaxConcurrentRequestsPerFile(n int) ClientOption {
  84. return func(c *Client) error {
  85. if n < 1 {
  86. return errors.New("n must be greater or equal to 1")
  87. }
  88. c.maxConcurrentRequests = n
  89. return nil
  90. }
  91. }
  92. // UseConcurrentWrites allows the Client to perform concurrent Writes.
  93. //
  94. // Using concurrency while doing writes, requires special consideration.
  95. // A write to a later offset in a file after an error,
  96. // could end up with a file length longer than what was successfully written.
  97. //
  98. // When using this option, if you receive an error during `io.Copy` or `io.WriteTo`,
  99. // you may need to `Truncate` the target Writer to avoid “holes” in the data written.
  100. func UseConcurrentWrites(value bool) ClientOption {
  101. return func(c *Client) error {
  102. c.useConcurrentWrites = value
  103. return nil
  104. }
  105. }
  106. // UseConcurrentReads allows the Client to perform concurrent Reads.
  107. //
  108. // Concurrent reads are generally safe to use and not using them will degrade
  109. // performance, so this option is enabled by default.
  110. //
  111. // When enabled, WriteTo will use Stat/Fstat to get the file size and determines
  112. // how many concurrent workers to use.
  113. // Some "read once" servers will delete the file if they receive a stat call on an
  114. // open file and then the download will fail.
  115. // Disabling concurrent reads you will be able to download files from these servers.
  116. // If concurrent reads are disabled, the UseFstat option is ignored.
  117. func UseConcurrentReads(value bool) ClientOption {
  118. return func(c *Client) error {
  119. c.disableConcurrentReads = !value
  120. return nil
  121. }
  122. }
  123. // UseFstat sets whether to use Fstat or Stat when File.WriteTo is called
  124. // (usually when copying files).
  125. // Some servers limit the amount of open files and calling Stat after opening
  126. // the file will throw an error From the server. Setting this flag will call
  127. // Fstat instead of Stat which is suppose to be called on an open file handle.
  128. //
  129. // It has been found that that with IBM Sterling SFTP servers which have
  130. // "extractability" level set to 1 which means only 1 file can be opened at
  131. // any given time.
  132. //
  133. // If the server you are working with still has an issue with both Stat and
  134. // Fstat calls you can always open a file and read it until the end.
  135. //
  136. // Another reason to read the file until its end and Fstat doesn't work is
  137. // that in some servers, reading a full file will automatically delete the
  138. // file as some of these mainframes map the file to a message in a queue.
  139. // Once the file has been read it will get deleted.
  140. func UseFstat(value bool) ClientOption {
  141. return func(c *Client) error {
  142. c.useFstat = value
  143. return nil
  144. }
  145. }
  146. // Client represents an SFTP session on a *ssh.ClientConn SSH connection.
  147. // Multiple Clients can be active on a single SSH connection, and a Client
  148. // may be called concurrently from multiple Goroutines.
  149. //
  150. // Client implements the github.com/kr/fs.FileSystem interface.
  151. type Client struct {
  152. clientConn
  153. ext map[string]string // Extensions (name -> data).
  154. maxPacket int // max packet size read or written.
  155. maxConcurrentRequests int
  156. nextid uint32
  157. // write concurrency is… error prone.
  158. // Default behavior should be to not use it.
  159. useConcurrentWrites bool
  160. useFstat bool
  161. disableConcurrentReads bool
  162. }
  163. // NewClient creates a new SFTP client on conn, using zero or more option
  164. // functions.
  165. func NewClient(conn *ssh.Client, opts ...ClientOption) (*Client, error) {
  166. s, err := conn.NewSession()
  167. if err != nil {
  168. return nil, err
  169. }
  170. if err := s.RequestSubsystem("sftp"); err != nil {
  171. return nil, err
  172. }
  173. pw, err := s.StdinPipe()
  174. if err != nil {
  175. return nil, err
  176. }
  177. pr, err := s.StdoutPipe()
  178. if err != nil {
  179. return nil, err
  180. }
  181. return NewClientPipe(pr, pw, opts...)
  182. }
  183. // NewClientPipe creates a new SFTP client given a Reader and a WriteCloser.
  184. // This can be used for connecting to an SFTP server over TCP/TLS or by using
  185. // the system's ssh client program (e.g. via exec.Command).
  186. func NewClientPipe(rd io.Reader, wr io.WriteCloser, opts ...ClientOption) (*Client, error) {
  187. sftp := &Client{
  188. clientConn: clientConn{
  189. conn: conn{
  190. Reader: rd,
  191. WriteCloser: wr,
  192. },
  193. inflight: make(map[uint32]chan<- result),
  194. closed: make(chan struct{}),
  195. },
  196. ext: make(map[string]string),
  197. maxPacket: 1 << 15,
  198. maxConcurrentRequests: 64,
  199. }
  200. for _, opt := range opts {
  201. if err := opt(sftp); err != nil {
  202. wr.Close()
  203. return nil, err
  204. }
  205. }
  206. if err := sftp.sendInit(); err != nil {
  207. wr.Close()
  208. return nil, fmt.Errorf("error sending init packet to server: %w", err)
  209. }
  210. if err := sftp.recvVersion(); err != nil {
  211. wr.Close()
  212. return nil, fmt.Errorf("error receiving version packet from server: %w", err)
  213. }
  214. sftp.clientConn.wg.Add(1)
  215. go func() {
  216. defer sftp.clientConn.wg.Done()
  217. if err := sftp.clientConn.recv(); err != nil {
  218. sftp.clientConn.broadcastErr(err)
  219. }
  220. }()
  221. return sftp, nil
  222. }
  223. // Create creates the named file mode 0666 (before umask), truncating it if it
  224. // already exists. If successful, methods on the returned File can be used for
  225. // I/O; the associated file descriptor has mode O_RDWR. If you need more
  226. // control over the flags/mode used to open the file see client.OpenFile.
  227. //
  228. // Note that some SFTP servers (eg. AWS Transfer) do not support opening files
  229. // read/write at the same time. For those services you will need to use
  230. // `client.OpenFile(os.O_WRONLY|os.O_CREATE|os.O_TRUNC)`.
  231. func (c *Client) Create(path string) (*File, error) {
  232. return c.open(path, flags(os.O_RDWR|os.O_CREATE|os.O_TRUNC))
  233. }
  234. const sftpProtocolVersion = 3 // https://filezilla-project.org/specs/draft-ietf-secsh-filexfer-02.txt
  235. func (c *Client) sendInit() error {
  236. return c.clientConn.conn.sendPacket(&sshFxInitPacket{
  237. Version: sftpProtocolVersion, // https://filezilla-project.org/specs/draft-ietf-secsh-filexfer-02.txt
  238. })
  239. }
  240. // returns the next value of c.nextid
  241. func (c *Client) nextID() uint32 {
  242. return atomic.AddUint32(&c.nextid, 1)
  243. }
  244. func (c *Client) recvVersion() error {
  245. typ, data, err := c.recvPacket(0)
  246. if err != nil {
  247. if err == io.EOF {
  248. return fmt.Errorf("server unexpectedly closed connection: %w", io.ErrUnexpectedEOF)
  249. }
  250. return err
  251. }
  252. if typ != sshFxpVersion {
  253. return &unexpectedPacketErr{sshFxpVersion, typ}
  254. }
  255. version, data, err := unmarshalUint32Safe(data)
  256. if err != nil {
  257. return err
  258. }
  259. if version != sftpProtocolVersion {
  260. return &unexpectedVersionErr{sftpProtocolVersion, version}
  261. }
  262. for len(data) > 0 {
  263. var ext extensionPair
  264. ext, data, err = unmarshalExtensionPair(data)
  265. if err != nil {
  266. return err
  267. }
  268. c.ext[ext.Name] = ext.Data
  269. }
  270. return nil
  271. }
  272. // HasExtension checks whether the server supports a named extension.
  273. //
  274. // The first return value is the extension data reported by the server
  275. // (typically a version number).
  276. func (c *Client) HasExtension(name string) (string, bool) {
  277. data, ok := c.ext[name]
  278. return data, ok
  279. }
  280. // Walk returns a new Walker rooted at root.
  281. func (c *Client) Walk(root string) *fs.Walker {
  282. return fs.WalkFS(root, c)
  283. }
  284. // ReadDir reads the directory named by dirname and returns a list of
  285. // directory entries.
  286. func (c *Client) ReadDir(p string) ([]os.FileInfo, error) {
  287. handle, err := c.opendir(p)
  288. if err != nil {
  289. return nil, err
  290. }
  291. defer c.close(handle) // this has to defer earlier than the lock below
  292. var attrs []os.FileInfo
  293. var done = false
  294. for !done {
  295. id := c.nextID()
  296. typ, data, err1 := c.sendPacket(nil, &sshFxpReaddirPacket{
  297. ID: id,
  298. Handle: handle,
  299. })
  300. if err1 != nil {
  301. err = err1
  302. done = true
  303. break
  304. }
  305. switch typ {
  306. case sshFxpName:
  307. sid, data := unmarshalUint32(data)
  308. if sid != id {
  309. return nil, &unexpectedIDErr{id, sid}
  310. }
  311. count, data := unmarshalUint32(data)
  312. for i := uint32(0); i < count; i++ {
  313. var filename string
  314. filename, data = unmarshalString(data)
  315. _, data = unmarshalString(data) // discard longname
  316. var attr *FileStat
  317. attr, data = unmarshalAttrs(data)
  318. if filename == "." || filename == ".." {
  319. continue
  320. }
  321. attrs = append(attrs, fileInfoFromStat(attr, path.Base(filename)))
  322. }
  323. case sshFxpStatus:
  324. // TODO(dfc) scope warning!
  325. err = normaliseError(unmarshalStatus(id, data))
  326. done = true
  327. default:
  328. return nil, unimplementedPacketErr(typ)
  329. }
  330. }
  331. if err == io.EOF {
  332. err = nil
  333. }
  334. return attrs, err
  335. }
  336. func (c *Client) opendir(path string) (string, error) {
  337. id := c.nextID()
  338. typ, data, err := c.sendPacket(nil, &sshFxpOpendirPacket{
  339. ID: id,
  340. Path: path,
  341. })
  342. if err != nil {
  343. return "", err
  344. }
  345. switch typ {
  346. case sshFxpHandle:
  347. sid, data := unmarshalUint32(data)
  348. if sid != id {
  349. return "", &unexpectedIDErr{id, sid}
  350. }
  351. handle, _ := unmarshalString(data)
  352. return handle, nil
  353. case sshFxpStatus:
  354. return "", normaliseError(unmarshalStatus(id, data))
  355. default:
  356. return "", unimplementedPacketErr(typ)
  357. }
  358. }
  359. // Stat returns a FileInfo structure describing the file specified by path 'p'.
  360. // If 'p' is a symbolic link, the returned FileInfo structure describes the referent file.
  361. func (c *Client) Stat(p string) (os.FileInfo, error) {
  362. fs, err := c.stat(p)
  363. if err != nil {
  364. return nil, err
  365. }
  366. return fileInfoFromStat(fs, path.Base(p)), nil
  367. }
  368. // Lstat returns a FileInfo structure describing the file specified by path 'p'.
  369. // If 'p' is a symbolic link, the returned FileInfo structure describes the symbolic link.
  370. func (c *Client) Lstat(p string) (os.FileInfo, error) {
  371. id := c.nextID()
  372. typ, data, err := c.sendPacket(nil, &sshFxpLstatPacket{
  373. ID: id,
  374. Path: p,
  375. })
  376. if err != nil {
  377. return nil, err
  378. }
  379. switch typ {
  380. case sshFxpAttrs:
  381. sid, data := unmarshalUint32(data)
  382. if sid != id {
  383. return nil, &unexpectedIDErr{id, sid}
  384. }
  385. attr, _ := unmarshalAttrs(data)
  386. return fileInfoFromStat(attr, path.Base(p)), nil
  387. case sshFxpStatus:
  388. return nil, normaliseError(unmarshalStatus(id, data))
  389. default:
  390. return nil, unimplementedPacketErr(typ)
  391. }
  392. }
  393. // ReadLink reads the target of a symbolic link.
  394. func (c *Client) ReadLink(p string) (string, error) {
  395. id := c.nextID()
  396. typ, data, err := c.sendPacket(nil, &sshFxpReadlinkPacket{
  397. ID: id,
  398. Path: p,
  399. })
  400. if err != nil {
  401. return "", err
  402. }
  403. switch typ {
  404. case sshFxpName:
  405. sid, data := unmarshalUint32(data)
  406. if sid != id {
  407. return "", &unexpectedIDErr{id, sid}
  408. }
  409. count, data := unmarshalUint32(data)
  410. if count != 1 {
  411. return "", unexpectedCount(1, count)
  412. }
  413. filename, _ := unmarshalString(data) // ignore dummy attributes
  414. return filename, nil
  415. case sshFxpStatus:
  416. return "", normaliseError(unmarshalStatus(id, data))
  417. default:
  418. return "", unimplementedPacketErr(typ)
  419. }
  420. }
  421. // Link creates a hard link at 'newname', pointing at the same inode as 'oldname'
  422. func (c *Client) Link(oldname, newname string) error {
  423. id := c.nextID()
  424. typ, data, err := c.sendPacket(nil, &sshFxpHardlinkPacket{
  425. ID: id,
  426. Oldpath: oldname,
  427. Newpath: newname,
  428. })
  429. if err != nil {
  430. return err
  431. }
  432. switch typ {
  433. case sshFxpStatus:
  434. return normaliseError(unmarshalStatus(id, data))
  435. default:
  436. return unimplementedPacketErr(typ)
  437. }
  438. }
  439. // Symlink creates a symbolic link at 'newname', pointing at target 'oldname'
  440. func (c *Client) Symlink(oldname, newname string) error {
  441. id := c.nextID()
  442. typ, data, err := c.sendPacket(nil, &sshFxpSymlinkPacket{
  443. ID: id,
  444. Linkpath: newname,
  445. Targetpath: oldname,
  446. })
  447. if err != nil {
  448. return err
  449. }
  450. switch typ {
  451. case sshFxpStatus:
  452. return normaliseError(unmarshalStatus(id, data))
  453. default:
  454. return unimplementedPacketErr(typ)
  455. }
  456. }
  457. func (c *Client) setfstat(handle string, flags uint32, attrs interface{}) error {
  458. id := c.nextID()
  459. typ, data, err := c.sendPacket(nil, &sshFxpFsetstatPacket{
  460. ID: id,
  461. Handle: handle,
  462. Flags: flags,
  463. Attrs: attrs,
  464. })
  465. if err != nil {
  466. return err
  467. }
  468. switch typ {
  469. case sshFxpStatus:
  470. return normaliseError(unmarshalStatus(id, data))
  471. default:
  472. return unimplementedPacketErr(typ)
  473. }
  474. }
  475. // setstat is a convience wrapper to allow for changing of various parts of the file descriptor.
  476. func (c *Client) setstat(path string, flags uint32, attrs interface{}) error {
  477. id := c.nextID()
  478. typ, data, err := c.sendPacket(nil, &sshFxpSetstatPacket{
  479. ID: id,
  480. Path: path,
  481. Flags: flags,
  482. Attrs: attrs,
  483. })
  484. if err != nil {
  485. return err
  486. }
  487. switch typ {
  488. case sshFxpStatus:
  489. return normaliseError(unmarshalStatus(id, data))
  490. default:
  491. return unimplementedPacketErr(typ)
  492. }
  493. }
  494. // Chtimes changes the access and modification times of the named file.
  495. func (c *Client) Chtimes(path string, atime time.Time, mtime time.Time) error {
  496. type times struct {
  497. Atime uint32
  498. Mtime uint32
  499. }
  500. attrs := times{uint32(atime.Unix()), uint32(mtime.Unix())}
  501. return c.setstat(path, sshFileXferAttrACmodTime, attrs)
  502. }
  503. // Chown changes the user and group owners of the named file.
  504. func (c *Client) Chown(path string, uid, gid int) error {
  505. type owner struct {
  506. UID uint32
  507. GID uint32
  508. }
  509. attrs := owner{uint32(uid), uint32(gid)}
  510. return c.setstat(path, sshFileXferAttrUIDGID, attrs)
  511. }
  512. // Chmod changes the permissions of the named file.
  513. //
  514. // Chmod does not apply a umask, because even retrieving the umask is not
  515. // possible in a portable way without causing a race condition. Callers
  516. // should mask off umask bits, if desired.
  517. func (c *Client) Chmod(path string, mode os.FileMode) error {
  518. return c.setstat(path, sshFileXferAttrPermissions, toChmodPerm(mode))
  519. }
  520. // Truncate sets the size of the named file. Although it may be safely assumed
  521. // that if the size is less than its current size it will be truncated to fit,
  522. // the SFTP protocol does not specify what behavior the server should do when setting
  523. // size greater than the current size.
  524. func (c *Client) Truncate(path string, size int64) error {
  525. return c.setstat(path, sshFileXferAttrSize, uint64(size))
  526. }
  527. // Open opens the named file for reading. If successful, methods on the
  528. // returned file can be used for reading; the associated file descriptor
  529. // has mode O_RDONLY.
  530. func (c *Client) Open(path string) (*File, error) {
  531. return c.open(path, flags(os.O_RDONLY))
  532. }
  533. // OpenFile is the generalized open call; most users will use Open or
  534. // Create instead. It opens the named file with specified flag (O_RDONLY
  535. // etc.). If successful, methods on the returned File can be used for I/O.
  536. func (c *Client) OpenFile(path string, f int) (*File, error) {
  537. return c.open(path, flags(f))
  538. }
  539. func (c *Client) open(path string, pflags uint32) (*File, error) {
  540. id := c.nextID()
  541. typ, data, err := c.sendPacket(nil, &sshFxpOpenPacket{
  542. ID: id,
  543. Path: path,
  544. Pflags: pflags,
  545. })
  546. if err != nil {
  547. return nil, err
  548. }
  549. switch typ {
  550. case sshFxpHandle:
  551. sid, data := unmarshalUint32(data)
  552. if sid != id {
  553. return nil, &unexpectedIDErr{id, sid}
  554. }
  555. handle, _ := unmarshalString(data)
  556. return &File{c: c, path: path, handle: handle}, nil
  557. case sshFxpStatus:
  558. return nil, normaliseError(unmarshalStatus(id, data))
  559. default:
  560. return nil, unimplementedPacketErr(typ)
  561. }
  562. }
  563. // close closes a handle handle previously returned in the response
  564. // to SSH_FXP_OPEN or SSH_FXP_OPENDIR. The handle becomes invalid
  565. // immediately after this request has been sent.
  566. func (c *Client) close(handle string) error {
  567. id := c.nextID()
  568. typ, data, err := c.sendPacket(nil, &sshFxpClosePacket{
  569. ID: id,
  570. Handle: handle,
  571. })
  572. if err != nil {
  573. return err
  574. }
  575. switch typ {
  576. case sshFxpStatus:
  577. return normaliseError(unmarshalStatus(id, data))
  578. default:
  579. return unimplementedPacketErr(typ)
  580. }
  581. }
  582. func (c *Client) stat(path string) (*FileStat, error) {
  583. id := c.nextID()
  584. typ, data, err := c.sendPacket(nil, &sshFxpStatPacket{
  585. ID: id,
  586. Path: path,
  587. })
  588. if err != nil {
  589. return nil, err
  590. }
  591. switch typ {
  592. case sshFxpAttrs:
  593. sid, data := unmarshalUint32(data)
  594. if sid != id {
  595. return nil, &unexpectedIDErr{id, sid}
  596. }
  597. attr, _ := unmarshalAttrs(data)
  598. return attr, nil
  599. case sshFxpStatus:
  600. return nil, normaliseError(unmarshalStatus(id, data))
  601. default:
  602. return nil, unimplementedPacketErr(typ)
  603. }
  604. }
  605. func (c *Client) fstat(handle string) (*FileStat, error) {
  606. id := c.nextID()
  607. typ, data, err := c.sendPacket(nil, &sshFxpFstatPacket{
  608. ID: id,
  609. Handle: handle,
  610. })
  611. if err != nil {
  612. return nil, err
  613. }
  614. switch typ {
  615. case sshFxpAttrs:
  616. sid, data := unmarshalUint32(data)
  617. if sid != id {
  618. return nil, &unexpectedIDErr{id, sid}
  619. }
  620. attr, _ := unmarshalAttrs(data)
  621. return attr, nil
  622. case sshFxpStatus:
  623. return nil, normaliseError(unmarshalStatus(id, data))
  624. default:
  625. return nil, unimplementedPacketErr(typ)
  626. }
  627. }
  628. // StatVFS retrieves VFS statistics from a remote host.
  629. //
  630. // It implements the statvfs@openssh.com SSH_FXP_EXTENDED feature
  631. // from http://www.opensource.apple.com/source/OpenSSH/OpenSSH-175/openssh/PROTOCOL?txt.
  632. func (c *Client) StatVFS(path string) (*StatVFS, error) {
  633. // send the StatVFS packet to the server
  634. id := c.nextID()
  635. typ, data, err := c.sendPacket(nil, &sshFxpStatvfsPacket{
  636. ID: id,
  637. Path: path,
  638. })
  639. if err != nil {
  640. return nil, err
  641. }
  642. switch typ {
  643. // server responded with valid data
  644. case sshFxpExtendedReply:
  645. var response StatVFS
  646. err = binary.Read(bytes.NewReader(data), binary.BigEndian, &response)
  647. if err != nil {
  648. return nil, errors.New("can not parse reply")
  649. }
  650. return &response, nil
  651. // the resquest failed
  652. case sshFxpStatus:
  653. return nil, normaliseError(unmarshalStatus(id, data))
  654. default:
  655. return nil, unimplementedPacketErr(typ)
  656. }
  657. }
  658. // Join joins any number of path elements into a single path, adding a
  659. // separating slash if necessary. The result is Cleaned; in particular, all
  660. // empty strings are ignored.
  661. func (c *Client) Join(elem ...string) string { return path.Join(elem...) }
  662. // Remove removes the specified file or directory. An error will be returned if no
  663. // file or directory with the specified path exists, or if the specified directory
  664. // is not empty.
  665. func (c *Client) Remove(path string) error {
  666. err := c.removeFile(path)
  667. // some servers, *cough* osx *cough*, return EPERM, not ENODIR.
  668. // serv-u returns ssh_FX_FILE_IS_A_DIRECTORY
  669. // EPERM is converted to os.ErrPermission so it is not a StatusError
  670. if err, ok := err.(*StatusError); ok {
  671. switch err.Code {
  672. case sshFxFailure, sshFxFileIsADirectory:
  673. return c.RemoveDirectory(path)
  674. }
  675. }
  676. if os.IsPermission(err) {
  677. return c.RemoveDirectory(path)
  678. }
  679. return err
  680. }
  681. func (c *Client) removeFile(path string) error {
  682. id := c.nextID()
  683. typ, data, err := c.sendPacket(nil, &sshFxpRemovePacket{
  684. ID: id,
  685. Filename: path,
  686. })
  687. if err != nil {
  688. return err
  689. }
  690. switch typ {
  691. case sshFxpStatus:
  692. return normaliseError(unmarshalStatus(id, data))
  693. default:
  694. return unimplementedPacketErr(typ)
  695. }
  696. }
  697. // RemoveDirectory removes a directory path.
  698. func (c *Client) RemoveDirectory(path string) error {
  699. id := c.nextID()
  700. typ, data, err := c.sendPacket(nil, &sshFxpRmdirPacket{
  701. ID: id,
  702. Path: path,
  703. })
  704. if err != nil {
  705. return err
  706. }
  707. switch typ {
  708. case sshFxpStatus:
  709. return normaliseError(unmarshalStatus(id, data))
  710. default:
  711. return unimplementedPacketErr(typ)
  712. }
  713. }
  714. // Rename renames a file.
  715. func (c *Client) Rename(oldname, newname string) error {
  716. id := c.nextID()
  717. typ, data, err := c.sendPacket(nil, &sshFxpRenamePacket{
  718. ID: id,
  719. Oldpath: oldname,
  720. Newpath: newname,
  721. })
  722. if err != nil {
  723. return err
  724. }
  725. switch typ {
  726. case sshFxpStatus:
  727. return normaliseError(unmarshalStatus(id, data))
  728. default:
  729. return unimplementedPacketErr(typ)
  730. }
  731. }
  732. // PosixRename renames a file using the posix-rename@openssh.com extension
  733. // which will replace newname if it already exists.
  734. func (c *Client) PosixRename(oldname, newname string) error {
  735. id := c.nextID()
  736. typ, data, err := c.sendPacket(nil, &sshFxpPosixRenamePacket{
  737. ID: id,
  738. Oldpath: oldname,
  739. Newpath: newname,
  740. })
  741. if err != nil {
  742. return err
  743. }
  744. switch typ {
  745. case sshFxpStatus:
  746. return normaliseError(unmarshalStatus(id, data))
  747. default:
  748. return unimplementedPacketErr(typ)
  749. }
  750. }
  751. // RealPath can be used to have the server canonicalize any given path name to an absolute path.
  752. //
  753. // This is useful for converting path names containing ".." components,
  754. // or relative pathnames without a leading slash into absolute paths.
  755. func (c *Client) RealPath(path string) (string, error) {
  756. id := c.nextID()
  757. typ, data, err := c.sendPacket(nil, &sshFxpRealpathPacket{
  758. ID: id,
  759. Path: path,
  760. })
  761. if err != nil {
  762. return "", err
  763. }
  764. switch typ {
  765. case sshFxpName:
  766. sid, data := unmarshalUint32(data)
  767. if sid != id {
  768. return "", &unexpectedIDErr{id, sid}
  769. }
  770. count, data := unmarshalUint32(data)
  771. if count != 1 {
  772. return "", unexpectedCount(1, count)
  773. }
  774. filename, _ := unmarshalString(data) // ignore attributes
  775. return filename, nil
  776. case sshFxpStatus:
  777. return "", normaliseError(unmarshalStatus(id, data))
  778. default:
  779. return "", unimplementedPacketErr(typ)
  780. }
  781. }
  782. // Getwd returns the current working directory of the server. Operations
  783. // involving relative paths will be based at this location.
  784. func (c *Client) Getwd() (string, error) {
  785. return c.RealPath(".")
  786. }
  787. // Mkdir creates the specified directory. An error will be returned if a file or
  788. // directory with the specified path already exists, or if the directory's
  789. // parent folder does not exist (the method cannot create complete paths).
  790. func (c *Client) Mkdir(path string) error {
  791. id := c.nextID()
  792. typ, data, err := c.sendPacket(nil, &sshFxpMkdirPacket{
  793. ID: id,
  794. Path: path,
  795. })
  796. if err != nil {
  797. return err
  798. }
  799. switch typ {
  800. case sshFxpStatus:
  801. return normaliseError(unmarshalStatus(id, data))
  802. default:
  803. return unimplementedPacketErr(typ)
  804. }
  805. }
  806. // MkdirAll creates a directory named path, along with any necessary parents,
  807. // and returns nil, or else returns an error.
  808. // If path is already a directory, MkdirAll does nothing and returns nil.
  809. // If path contains a regular file, an error is returned
  810. func (c *Client) MkdirAll(path string) error {
  811. // Most of this code mimics https://golang.org/src/os/path.go?s=514:561#L13
  812. // Fast path: if we can tell whether path is a directory or file, stop with success or error.
  813. dir, err := c.Stat(path)
  814. if err == nil {
  815. if dir.IsDir() {
  816. return nil
  817. }
  818. return &os.PathError{Op: "mkdir", Path: path, Err: syscall.ENOTDIR}
  819. }
  820. // Slow path: make sure parent exists and then call Mkdir for path.
  821. i := len(path)
  822. for i > 0 && path[i-1] == '/' { // Skip trailing path separator.
  823. i--
  824. }
  825. j := i
  826. for j > 0 && path[j-1] != '/' { // Scan backward over element.
  827. j--
  828. }
  829. if j > 1 {
  830. // Create parent
  831. err = c.MkdirAll(path[0 : j-1])
  832. if err != nil {
  833. return err
  834. }
  835. }
  836. // Parent now exists; invoke Mkdir and use its result.
  837. err = c.Mkdir(path)
  838. if err != nil {
  839. // Handle arguments like "foo/." by
  840. // double-checking that directory doesn't exist.
  841. dir, err1 := c.Lstat(path)
  842. if err1 == nil && dir.IsDir() {
  843. return nil
  844. }
  845. return err
  846. }
  847. return nil
  848. }
  849. // RemoveAll delete files recursively in the directory and Recursively delete subdirectories.
  850. // An error will be returned if no file or directory with the specified path exists
  851. func (c *Client) RemoveAll(path string) error {
  852. // Get the file/directory information
  853. fi, err := c.Stat(path)
  854. if err != nil {
  855. return err
  856. }
  857. if fi.IsDir() {
  858. // Delete files recursively in the directory
  859. files, err := c.ReadDir(path)
  860. if err != nil {
  861. return err
  862. }
  863. for _, file := range files {
  864. if file.IsDir() {
  865. // Recursively delete subdirectories
  866. err = c.RemoveAll(path + "/" + file.Name())
  867. if err != nil {
  868. return err
  869. }
  870. } else {
  871. // Delete individual files
  872. err = c.Remove(path + "/" + file.Name())
  873. if err != nil {
  874. return err
  875. }
  876. }
  877. }
  878. }
  879. return c.Remove(path)
  880. }
  881. // File represents a remote file.
  882. type File struct {
  883. c *Client
  884. path string
  885. handle string
  886. mu sync.Mutex
  887. offset int64 // current offset within remote file
  888. }
  889. // Close closes the File, rendering it unusable for I/O. It returns an
  890. // error, if any.
  891. func (f *File) Close() error {
  892. return f.c.close(f.handle)
  893. }
  894. // Name returns the name of the file as presented to Open or Create.
  895. func (f *File) Name() string {
  896. return f.path
  897. }
  898. // Read reads up to len(b) bytes from the File. It returns the number of bytes
  899. // read and an error, if any. Read follows io.Reader semantics, so when Read
  900. // encounters an error or EOF condition after successfully reading n > 0 bytes,
  901. // it returns the number of bytes read.
  902. //
  903. // To maximise throughput for transferring the entire file (especially
  904. // over high latency links) it is recommended to use WriteTo rather
  905. // than calling Read multiple times. io.Copy will do this
  906. // automatically.
  907. func (f *File) Read(b []byte) (int, error) {
  908. f.mu.Lock()
  909. defer f.mu.Unlock()
  910. n, err := f.ReadAt(b, f.offset)
  911. f.offset += int64(n)
  912. return n, err
  913. }
  914. // readChunkAt attempts to read the whole entire length of the buffer from the file starting at the offset.
  915. // It will continue progressively reading into the buffer until it fills the whole buffer, or an error occurs.
  916. func (f *File) readChunkAt(ch chan result, b []byte, off int64) (n int, err error) {
  917. for err == nil && n < len(b) {
  918. id := f.c.nextID()
  919. typ, data, err := f.c.sendPacket(ch, &sshFxpReadPacket{
  920. ID: id,
  921. Handle: f.handle,
  922. Offset: uint64(off) + uint64(n),
  923. Len: uint32(len(b) - n),
  924. })
  925. if err != nil {
  926. return n, err
  927. }
  928. switch typ {
  929. case sshFxpStatus:
  930. return n, normaliseError(unmarshalStatus(id, data))
  931. case sshFxpData:
  932. sid, data := unmarshalUint32(data)
  933. if id != sid {
  934. return n, &unexpectedIDErr{id, sid}
  935. }
  936. l, data := unmarshalUint32(data)
  937. n += copy(b[n:], data[:l])
  938. default:
  939. return n, unimplementedPacketErr(typ)
  940. }
  941. }
  942. return
  943. }
  944. func (f *File) readAtSequential(b []byte, off int64) (read int, err error) {
  945. for read < len(b) {
  946. rb := b[read:]
  947. if len(rb) > f.c.maxPacket {
  948. rb = rb[:f.c.maxPacket]
  949. }
  950. n, err := f.readChunkAt(nil, rb, off+int64(read))
  951. if n < 0 {
  952. panic("sftp.File: returned negative count from readChunkAt")
  953. }
  954. if n > 0 {
  955. read += n
  956. }
  957. if err != nil {
  958. return read, err
  959. }
  960. }
  961. return read, nil
  962. }
  963. // ReadAt reads up to len(b) byte from the File at a given offset `off`. It returns
  964. // the number of bytes read and an error, if any. ReadAt follows io.ReaderAt semantics,
  965. // so the file offset is not altered during the read.
  966. func (f *File) ReadAt(b []byte, off int64) (int, error) {
  967. if len(b) <= f.c.maxPacket {
  968. // This should be able to be serviced with 1/2 requests.
  969. // So, just do it directly.
  970. return f.readChunkAt(nil, b, off)
  971. }
  972. if f.c.disableConcurrentReads {
  973. return f.readAtSequential(b, off)
  974. }
  975. // Split the read into multiple maxPacket-sized concurrent reads bounded by maxConcurrentRequests.
  976. // This allows writes with a suitably large buffer to transfer data at a much faster rate
  977. // by overlapping round trip times.
  978. cancel := make(chan struct{})
  979. concurrency := len(b)/f.c.maxPacket + 1
  980. if concurrency > f.c.maxConcurrentRequests || concurrency < 1 {
  981. concurrency = f.c.maxConcurrentRequests
  982. }
  983. resPool := newResChanPool(concurrency)
  984. type work struct {
  985. id uint32
  986. res chan result
  987. b []byte
  988. off int64
  989. }
  990. workCh := make(chan work)
  991. // Slice: cut up the Read into any number of buffers of length <= f.c.maxPacket, and at appropriate offsets.
  992. go func() {
  993. defer close(workCh)
  994. b := b
  995. offset := off
  996. chunkSize := f.c.maxPacket
  997. for len(b) > 0 {
  998. rb := b
  999. if len(rb) > chunkSize {
  1000. rb = rb[:chunkSize]
  1001. }
  1002. id := f.c.nextID()
  1003. res := resPool.Get()
  1004. f.c.dispatchRequest(res, &sshFxpReadPacket{
  1005. ID: id,
  1006. Handle: f.handle,
  1007. Offset: uint64(offset),
  1008. Len: uint32(chunkSize),
  1009. })
  1010. select {
  1011. case workCh <- work{id, res, rb, offset}:
  1012. case <-cancel:
  1013. return
  1014. }
  1015. offset += int64(len(rb))
  1016. b = b[len(rb):]
  1017. }
  1018. }()
  1019. type rErr struct {
  1020. off int64
  1021. err error
  1022. }
  1023. errCh := make(chan rErr)
  1024. var wg sync.WaitGroup
  1025. wg.Add(concurrency)
  1026. for i := 0; i < concurrency; i++ {
  1027. // Map_i: each worker gets work, and then performs the Read into its buffer from its respective offset.
  1028. go func() {
  1029. defer wg.Done()
  1030. for packet := range workCh {
  1031. var n int
  1032. s := <-packet.res
  1033. resPool.Put(packet.res)
  1034. err := s.err
  1035. if err == nil {
  1036. switch s.typ {
  1037. case sshFxpStatus:
  1038. err = normaliseError(unmarshalStatus(packet.id, s.data))
  1039. case sshFxpData:
  1040. sid, data := unmarshalUint32(s.data)
  1041. if packet.id != sid {
  1042. err = &unexpectedIDErr{packet.id, sid}
  1043. } else {
  1044. l, data := unmarshalUint32(data)
  1045. n = copy(packet.b, data[:l])
  1046. // For normal disk files, it is guaranteed that this will read
  1047. // the specified number of bytes, or up to end of file.
  1048. // This implies, if we have a short read, that means EOF.
  1049. if n < len(packet.b) {
  1050. err = io.EOF
  1051. }
  1052. }
  1053. default:
  1054. err = unimplementedPacketErr(s.typ)
  1055. }
  1056. }
  1057. if err != nil {
  1058. // return the offset as the start + how much we read before the error.
  1059. errCh <- rErr{packet.off + int64(n), err}
  1060. return
  1061. }
  1062. }
  1063. }()
  1064. }
  1065. // Wait for long tail, before closing results.
  1066. go func() {
  1067. wg.Wait()
  1068. close(errCh)
  1069. }()
  1070. // Reduce: collect all the results into a relevant return: the earliest offset to return an error.
  1071. firstErr := rErr{math.MaxInt64, nil}
  1072. for rErr := range errCh {
  1073. if rErr.off <= firstErr.off {
  1074. firstErr = rErr
  1075. }
  1076. select {
  1077. case <-cancel:
  1078. default:
  1079. // stop any more work from being distributed. (Just in case.)
  1080. close(cancel)
  1081. }
  1082. }
  1083. if firstErr.err != nil {
  1084. // firstErr.err != nil if and only if firstErr.off > our starting offset.
  1085. return int(firstErr.off - off), firstErr.err
  1086. }
  1087. // As per spec for io.ReaderAt, we return nil error if and only if we read everything.
  1088. return len(b), nil
  1089. }
  1090. // writeToSequential implements WriteTo, but works sequentially with no parallelism.
  1091. func (f *File) writeToSequential(w io.Writer) (written int64, err error) {
  1092. b := make([]byte, f.c.maxPacket)
  1093. ch := make(chan result, 1) // reusable channel
  1094. for {
  1095. n, err := f.readChunkAt(ch, b, f.offset)
  1096. if n < 0 {
  1097. panic("sftp.File: returned negative count from readChunkAt")
  1098. }
  1099. if n > 0 {
  1100. f.offset += int64(n)
  1101. m, err := w.Write(b[:n])
  1102. written += int64(m)
  1103. if err != nil {
  1104. return written, err
  1105. }
  1106. }
  1107. if err != nil {
  1108. if err == io.EOF {
  1109. return written, nil // return nil explicitly.
  1110. }
  1111. return written, err
  1112. }
  1113. }
  1114. }
  1115. // WriteTo writes the file to the given Writer.
  1116. // The return value is the number of bytes written.
  1117. // Any error encountered during the write is also returned.
  1118. //
  1119. // This method is preferred over calling Read multiple times
  1120. // to maximise throughput for transferring the entire file,
  1121. // especially over high latency links.
  1122. func (f *File) WriteTo(w io.Writer) (written int64, err error) {
  1123. f.mu.Lock()
  1124. defer f.mu.Unlock()
  1125. if f.c.disableConcurrentReads {
  1126. return f.writeToSequential(w)
  1127. }
  1128. // For concurrency, we want to guess how many concurrent workers we should use.
  1129. var fileStat *FileStat
  1130. if f.c.useFstat {
  1131. fileStat, err = f.c.fstat(f.handle)
  1132. } else {
  1133. fileStat, err = f.c.stat(f.path)
  1134. }
  1135. if err != nil {
  1136. return 0, err
  1137. }
  1138. fileSize := fileStat.Size
  1139. if fileSize <= uint64(f.c.maxPacket) || !isRegular(fileStat.Mode) {
  1140. // only regular files are guaranteed to return (full read) xor (partial read, next error)
  1141. return f.writeToSequential(w)
  1142. }
  1143. concurrency64 := fileSize/uint64(f.c.maxPacket) + 1 // a bad guess, but better than no guess
  1144. if concurrency64 > uint64(f.c.maxConcurrentRequests) || concurrency64 < 1 {
  1145. concurrency64 = uint64(f.c.maxConcurrentRequests)
  1146. }
  1147. // Now that concurrency64 is saturated to an int value, we know this assignment cannot possibly overflow.
  1148. concurrency := int(concurrency64)
  1149. chunkSize := f.c.maxPacket
  1150. pool := newBufPool(concurrency, chunkSize)
  1151. resPool := newResChanPool(concurrency)
  1152. cancel := make(chan struct{})
  1153. var wg sync.WaitGroup
  1154. defer func() {
  1155. // Once the writing Reduce phase has ended, all the feed work needs to unconditionally stop.
  1156. close(cancel)
  1157. // We want to wait until all outstanding goroutines with an `f` or `f.c` reference have completed.
  1158. // Just to be sure we don’t orphan any goroutines any hanging references.
  1159. wg.Wait()
  1160. }()
  1161. type writeWork struct {
  1162. b []byte
  1163. off int64
  1164. err error
  1165. next chan writeWork
  1166. }
  1167. writeCh := make(chan writeWork)
  1168. type readWork struct {
  1169. id uint32
  1170. res chan result
  1171. off int64
  1172. cur, next chan writeWork
  1173. }
  1174. readCh := make(chan readWork)
  1175. // Slice: hand out chunks of work on demand, with a `cur` and `next` channel built-in for sequencing.
  1176. go func() {
  1177. defer close(readCh)
  1178. off := f.offset
  1179. cur := writeCh
  1180. for {
  1181. id := f.c.nextID()
  1182. res := resPool.Get()
  1183. next := make(chan writeWork)
  1184. readWork := readWork{
  1185. id: id,
  1186. res: res,
  1187. off: off,
  1188. cur: cur,
  1189. next: next,
  1190. }
  1191. f.c.dispatchRequest(res, &sshFxpReadPacket{
  1192. ID: id,
  1193. Handle: f.handle,
  1194. Offset: uint64(off),
  1195. Len: uint32(chunkSize),
  1196. })
  1197. select {
  1198. case readCh <- readWork:
  1199. case <-cancel:
  1200. return
  1201. }
  1202. off += int64(chunkSize)
  1203. cur = next
  1204. }
  1205. }()
  1206. wg.Add(concurrency)
  1207. for i := 0; i < concurrency; i++ {
  1208. // Map_i: each worker gets readWork, and does the Read into a buffer at the given offset.
  1209. go func() {
  1210. defer wg.Done()
  1211. for readWork := range readCh {
  1212. var b []byte
  1213. var n int
  1214. s := <-readWork.res
  1215. resPool.Put(readWork.res)
  1216. err := s.err
  1217. if err == nil {
  1218. switch s.typ {
  1219. case sshFxpStatus:
  1220. err = normaliseError(unmarshalStatus(readWork.id, s.data))
  1221. case sshFxpData:
  1222. sid, data := unmarshalUint32(s.data)
  1223. if readWork.id != sid {
  1224. err = &unexpectedIDErr{readWork.id, sid}
  1225. } else {
  1226. l, data := unmarshalUint32(data)
  1227. b = pool.Get()[:l]
  1228. n = copy(b, data[:l])
  1229. b = b[:n]
  1230. }
  1231. default:
  1232. err = unimplementedPacketErr(s.typ)
  1233. }
  1234. }
  1235. writeWork := writeWork{
  1236. b: b,
  1237. off: readWork.off,
  1238. err: err,
  1239. next: readWork.next,
  1240. }
  1241. select {
  1242. case readWork.cur <- writeWork:
  1243. case <-cancel:
  1244. return
  1245. }
  1246. if err != nil {
  1247. return
  1248. }
  1249. }
  1250. }()
  1251. }
  1252. // Reduce: serialize the results from the reads into sequential writes.
  1253. cur := writeCh
  1254. for {
  1255. packet, ok := <-cur
  1256. if !ok {
  1257. return written, errors.New("sftp.File.WriteTo: unexpectedly closed channel")
  1258. }
  1259. // Because writes are serialized, this will always be the last successfully read byte.
  1260. f.offset = packet.off + int64(len(packet.b))
  1261. if len(packet.b) > 0 {
  1262. n, err := w.Write(packet.b)
  1263. written += int64(n)
  1264. if err != nil {
  1265. return written, err
  1266. }
  1267. }
  1268. if packet.err != nil {
  1269. if packet.err == io.EOF {
  1270. return written, nil
  1271. }
  1272. return written, packet.err
  1273. }
  1274. pool.Put(packet.b)
  1275. cur = packet.next
  1276. }
  1277. }
  1278. // Stat returns the FileInfo structure describing file. If there is an
  1279. // error.
  1280. func (f *File) Stat() (os.FileInfo, error) {
  1281. fs, err := f.c.fstat(f.handle)
  1282. if err != nil {
  1283. return nil, err
  1284. }
  1285. return fileInfoFromStat(fs, path.Base(f.path)), nil
  1286. }
  1287. // Write writes len(b) bytes to the File. It returns the number of bytes
  1288. // written and an error, if any. Write returns a non-nil error when n !=
  1289. // len(b).
  1290. //
  1291. // To maximise throughput for transferring the entire file (especially
  1292. // over high latency links) it is recommended to use ReadFrom rather
  1293. // than calling Write multiple times. io.Copy will do this
  1294. // automatically.
  1295. func (f *File) Write(b []byte) (int, error) {
  1296. f.mu.Lock()
  1297. defer f.mu.Unlock()
  1298. n, err := f.WriteAt(b, f.offset)
  1299. f.offset += int64(n)
  1300. return n, err
  1301. }
  1302. func (f *File) writeChunkAt(ch chan result, b []byte, off int64) (int, error) {
  1303. typ, data, err := f.c.sendPacket(ch, &sshFxpWritePacket{
  1304. ID: f.c.nextID(),
  1305. Handle: f.handle,
  1306. Offset: uint64(off),
  1307. Length: uint32(len(b)),
  1308. Data: b,
  1309. })
  1310. if err != nil {
  1311. return 0, err
  1312. }
  1313. switch typ {
  1314. case sshFxpStatus:
  1315. id, _ := unmarshalUint32(data)
  1316. err := normaliseError(unmarshalStatus(id, data))
  1317. if err != nil {
  1318. return 0, err
  1319. }
  1320. default:
  1321. return 0, unimplementedPacketErr(typ)
  1322. }
  1323. return len(b), nil
  1324. }
  1325. // writeAtConcurrent implements WriterAt, but works concurrently rather than sequentially.
  1326. func (f *File) writeAtConcurrent(b []byte, off int64) (int, error) {
  1327. // Split the write into multiple maxPacket sized concurrent writes
  1328. // bounded by maxConcurrentRequests. This allows writes with a suitably
  1329. // large buffer to transfer data at a much faster rate due to
  1330. // overlapping round trip times.
  1331. cancel := make(chan struct{})
  1332. type work struct {
  1333. id uint32
  1334. res chan result
  1335. off int64
  1336. }
  1337. workCh := make(chan work)
  1338. concurrency := len(b)/f.c.maxPacket + 1
  1339. if concurrency > f.c.maxConcurrentRequests || concurrency < 1 {
  1340. concurrency = f.c.maxConcurrentRequests
  1341. }
  1342. pool := newResChanPool(concurrency)
  1343. // Slice: cut up the Read into any number of buffers of length <= f.c.maxPacket, and at appropriate offsets.
  1344. go func() {
  1345. defer close(workCh)
  1346. var read int
  1347. chunkSize := f.c.maxPacket
  1348. for read < len(b) {
  1349. wb := b[read:]
  1350. if len(wb) > chunkSize {
  1351. wb = wb[:chunkSize]
  1352. }
  1353. id := f.c.nextID()
  1354. res := pool.Get()
  1355. off := off + int64(read)
  1356. f.c.dispatchRequest(res, &sshFxpWritePacket{
  1357. ID: id,
  1358. Handle: f.handle,
  1359. Offset: uint64(off),
  1360. Length: uint32(len(wb)),
  1361. Data: wb,
  1362. })
  1363. select {
  1364. case workCh <- work{id, res, off}:
  1365. case <-cancel:
  1366. return
  1367. }
  1368. read += len(wb)
  1369. }
  1370. }()
  1371. type wErr struct {
  1372. off int64
  1373. err error
  1374. }
  1375. errCh := make(chan wErr)
  1376. var wg sync.WaitGroup
  1377. wg.Add(concurrency)
  1378. for i := 0; i < concurrency; i++ {
  1379. // Map_i: each worker gets work, and does the Write from each buffer to its respective offset.
  1380. go func() {
  1381. defer wg.Done()
  1382. for work := range workCh {
  1383. s := <-work.res
  1384. pool.Put(work.res)
  1385. err := s.err
  1386. if err == nil {
  1387. switch s.typ {
  1388. case sshFxpStatus:
  1389. err = normaliseError(unmarshalStatus(work.id, s.data))
  1390. default:
  1391. err = unimplementedPacketErr(s.typ)
  1392. }
  1393. }
  1394. if err != nil {
  1395. errCh <- wErr{work.off, err}
  1396. }
  1397. }
  1398. }()
  1399. }
  1400. // Wait for long tail, before closing results.
  1401. go func() {
  1402. wg.Wait()
  1403. close(errCh)
  1404. }()
  1405. // Reduce: collect all the results into a relevant return: the earliest offset to return an error.
  1406. firstErr := wErr{math.MaxInt64, nil}
  1407. for wErr := range errCh {
  1408. if wErr.off <= firstErr.off {
  1409. firstErr = wErr
  1410. }
  1411. select {
  1412. case <-cancel:
  1413. default:
  1414. // stop any more work from being distributed. (Just in case.)
  1415. close(cancel)
  1416. }
  1417. }
  1418. if firstErr.err != nil {
  1419. // firstErr.err != nil if and only if firstErr.off >= our starting offset.
  1420. return int(firstErr.off - off), firstErr.err
  1421. }
  1422. return len(b), nil
  1423. }
  1424. // WriteAt writes up to len(b) byte to the File at a given offset `off`. It returns
  1425. // the number of bytes written and an error, if any. WriteAt follows io.WriterAt semantics,
  1426. // so the file offset is not altered during the write.
  1427. func (f *File) WriteAt(b []byte, off int64) (written int, err error) {
  1428. if len(b) <= f.c.maxPacket {
  1429. // We can do this in one write.
  1430. return f.writeChunkAt(nil, b, off)
  1431. }
  1432. if f.c.useConcurrentWrites {
  1433. return f.writeAtConcurrent(b, off)
  1434. }
  1435. ch := make(chan result, 1) // reusable channel
  1436. chunkSize := f.c.maxPacket
  1437. for written < len(b) {
  1438. wb := b[written:]
  1439. if len(wb) > chunkSize {
  1440. wb = wb[:chunkSize]
  1441. }
  1442. n, err := f.writeChunkAt(ch, wb, off+int64(written))
  1443. if n > 0 {
  1444. written += n
  1445. }
  1446. if err != nil {
  1447. return written, err
  1448. }
  1449. }
  1450. return len(b), nil
  1451. }
  1452. // ReadFromWithConcurrency implements ReaderFrom,
  1453. // but uses the given concurrency to issue multiple requests at the same time.
  1454. //
  1455. // Giving a concurrency of less than one will default to the Client’s max concurrency.
  1456. //
  1457. // Otherwise, the given concurrency will be capped by the Client's max concurrency.
  1458. func (f *File) ReadFromWithConcurrency(r io.Reader, concurrency int) (read int64, err error) {
  1459. // Split the write into multiple maxPacket sized concurrent writes.
  1460. // This allows writes with a suitably large reader
  1461. // to transfer data at a much faster rate due to overlapping round trip times.
  1462. cancel := make(chan struct{})
  1463. type work struct {
  1464. id uint32
  1465. res chan result
  1466. off int64
  1467. }
  1468. workCh := make(chan work)
  1469. type rwErr struct {
  1470. off int64
  1471. err error
  1472. }
  1473. errCh := make(chan rwErr)
  1474. if concurrency > f.c.maxConcurrentRequests || concurrency < 1 {
  1475. concurrency = f.c.maxConcurrentRequests
  1476. }
  1477. pool := newResChanPool(concurrency)
  1478. // Slice: cut up the Read into any number of buffers of length <= f.c.maxPacket, and at appropriate offsets.
  1479. go func() {
  1480. defer close(workCh)
  1481. b := make([]byte, f.c.maxPacket)
  1482. off := f.offset
  1483. for {
  1484. n, err := r.Read(b)
  1485. if n > 0 {
  1486. read += int64(n)
  1487. id := f.c.nextID()
  1488. res := pool.Get()
  1489. f.c.dispatchRequest(res, &sshFxpWritePacket{
  1490. ID: id,
  1491. Handle: f.handle,
  1492. Offset: uint64(off),
  1493. Length: uint32(n),
  1494. Data: b[:n],
  1495. })
  1496. select {
  1497. case workCh <- work{id, res, off}:
  1498. case <-cancel:
  1499. return
  1500. }
  1501. off += int64(n)
  1502. }
  1503. if err != nil {
  1504. if err != io.EOF {
  1505. errCh <- rwErr{off, err}
  1506. }
  1507. return
  1508. }
  1509. }
  1510. }()
  1511. var wg sync.WaitGroup
  1512. wg.Add(concurrency)
  1513. for i := 0; i < concurrency; i++ {
  1514. // Map_i: each worker gets work, and does the Write from each buffer to its respective offset.
  1515. go func() {
  1516. defer wg.Done()
  1517. for work := range workCh {
  1518. s := <-work.res
  1519. pool.Put(work.res)
  1520. err := s.err
  1521. if err == nil {
  1522. switch s.typ {
  1523. case sshFxpStatus:
  1524. err = normaliseError(unmarshalStatus(work.id, s.data))
  1525. default:
  1526. err = unimplementedPacketErr(s.typ)
  1527. }
  1528. }
  1529. if err != nil {
  1530. errCh <- rwErr{work.off, err}
  1531. }
  1532. }
  1533. }()
  1534. }
  1535. // Wait for long tail, before closing results.
  1536. go func() {
  1537. wg.Wait()
  1538. close(errCh)
  1539. }()
  1540. // Reduce: Collect all the results into a relevant return: the earliest offset to return an error.
  1541. firstErr := rwErr{math.MaxInt64, nil}
  1542. for rwErr := range errCh {
  1543. if rwErr.off <= firstErr.off {
  1544. firstErr = rwErr
  1545. }
  1546. select {
  1547. case <-cancel:
  1548. default:
  1549. // stop any more work from being distributed.
  1550. close(cancel)
  1551. }
  1552. }
  1553. if firstErr.err != nil {
  1554. // firstErr.err != nil if and only if firstErr.off is a valid offset.
  1555. //
  1556. // firstErr.off will then be the lesser of:
  1557. // * the offset of the first error from writing,
  1558. // * the last successfully read offset.
  1559. //
  1560. // This could be less than the last successfully written offset,
  1561. // which is the whole reason for the UseConcurrentWrites() ClientOption.
  1562. //
  1563. // Callers are responsible for truncating any SFTP files to a safe length.
  1564. f.offset = firstErr.off
  1565. // ReadFrom is defined to return the read bytes, regardless of any writer errors.
  1566. return read, firstErr.err
  1567. }
  1568. f.offset += read
  1569. return read, nil
  1570. }
  1571. // ReadFrom reads data from r until EOF and writes it to the file. The return
  1572. // value is the number of bytes read. Any error except io.EOF encountered
  1573. // during the read is also returned.
  1574. //
  1575. // This method is preferred over calling Write multiple times
  1576. // to maximise throughput for transferring the entire file,
  1577. // especially over high-latency links.
  1578. func (f *File) ReadFrom(r io.Reader) (int64, error) {
  1579. f.mu.Lock()
  1580. defer f.mu.Unlock()
  1581. if f.c.useConcurrentWrites {
  1582. var remain int64
  1583. switch r := r.(type) {
  1584. case interface{ Len() int }:
  1585. remain = int64(r.Len())
  1586. case interface{ Size() int64 }:
  1587. remain = r.Size()
  1588. case *io.LimitedReader:
  1589. remain = r.N
  1590. case interface{ Stat() (os.FileInfo, error) }:
  1591. info, err := r.Stat()
  1592. if err == nil {
  1593. remain = info.Size()
  1594. }
  1595. }
  1596. if remain < 0 {
  1597. // We can strongly assert that we want default max concurrency here.
  1598. return f.ReadFromWithConcurrency(r, f.c.maxConcurrentRequests)
  1599. }
  1600. if remain > int64(f.c.maxPacket) {
  1601. // Otherwise, only use concurrency, if it would be at least two packets.
  1602. // This is the best reasonable guess we can make.
  1603. concurrency64 := remain/int64(f.c.maxPacket) + 1
  1604. // We need to cap this value to an `int` size value to avoid overflow on 32-bit machines.
  1605. // So, we may as well pre-cap it to `f.c.maxConcurrentRequests`.
  1606. if concurrency64 > int64(f.c.maxConcurrentRequests) {
  1607. concurrency64 = int64(f.c.maxConcurrentRequests)
  1608. }
  1609. return f.ReadFromWithConcurrency(r, int(concurrency64))
  1610. }
  1611. }
  1612. ch := make(chan result, 1) // reusable channel
  1613. b := make([]byte, f.c.maxPacket)
  1614. var read int64
  1615. for {
  1616. n, err := r.Read(b)
  1617. if n < 0 {
  1618. panic("sftp.File: reader returned negative count from Read")
  1619. }
  1620. if n > 0 {
  1621. read += int64(n)
  1622. m, err2 := f.writeChunkAt(ch, b[:n], f.offset)
  1623. f.offset += int64(m)
  1624. if err == nil {
  1625. err = err2
  1626. }
  1627. }
  1628. if err != nil {
  1629. if err == io.EOF {
  1630. return read, nil // return nil explicitly.
  1631. }
  1632. return read, err
  1633. }
  1634. }
  1635. }
  1636. // Seek implements io.Seeker by setting the client offset for the next Read or
  1637. // Write. It returns the next offset read. Seeking before or after the end of
  1638. // the file is undefined. Seeking relative to the end calls Stat.
  1639. func (f *File) Seek(offset int64, whence int) (int64, error) {
  1640. f.mu.Lock()
  1641. defer f.mu.Unlock()
  1642. switch whence {
  1643. case io.SeekStart:
  1644. case io.SeekCurrent:
  1645. offset += f.offset
  1646. case io.SeekEnd:
  1647. fi, err := f.Stat()
  1648. if err != nil {
  1649. return f.offset, err
  1650. }
  1651. offset += fi.Size()
  1652. default:
  1653. return f.offset, unimplementedSeekWhence(whence)
  1654. }
  1655. if offset < 0 {
  1656. return f.offset, os.ErrInvalid
  1657. }
  1658. f.offset = offset
  1659. return f.offset, nil
  1660. }
  1661. // Chown changes the uid/gid of the current file.
  1662. func (f *File) Chown(uid, gid int) error {
  1663. return f.c.Chown(f.path, uid, gid)
  1664. }
  1665. // Chmod changes the permissions of the current file.
  1666. //
  1667. // See Client.Chmod for details.
  1668. func (f *File) Chmod(mode os.FileMode) error {
  1669. return f.c.setfstat(f.handle, sshFileXferAttrPermissions, toChmodPerm(mode))
  1670. }
  1671. // Sync requests a flush of the contents of a File to stable storage.
  1672. //
  1673. // Sync requires the server to support the fsync@openssh.com extension.
  1674. func (f *File) Sync() error {
  1675. id := f.c.nextID()
  1676. typ, data, err := f.c.sendPacket(nil, &sshFxpFsyncPacket{
  1677. ID: id,
  1678. Handle: f.handle,
  1679. })
  1680. switch {
  1681. case err != nil:
  1682. return err
  1683. case typ == sshFxpStatus:
  1684. return normaliseError(unmarshalStatus(id, data))
  1685. default:
  1686. return &unexpectedPacketErr{want: sshFxpStatus, got: typ}
  1687. }
  1688. }
  1689. // Truncate sets the size of the current file. Although it may be safely assumed
  1690. // that if the size is less than its current size it will be truncated to fit,
  1691. // the SFTP protocol does not specify what behavior the server should do when setting
  1692. // size greater than the current size.
  1693. // We send a SSH_FXP_FSETSTAT here since we have a file handle
  1694. func (f *File) Truncate(size int64) error {
  1695. return f.c.setfstat(f.handle, sshFileXferAttrSize, uint64(size))
  1696. }
  1697. // normaliseError normalises an error into a more standard form that can be
  1698. // checked against stdlib errors like io.EOF or os.ErrNotExist.
  1699. func normaliseError(err error) error {
  1700. switch err := err.(type) {
  1701. case *StatusError:
  1702. switch err.Code {
  1703. case sshFxEOF:
  1704. return io.EOF
  1705. case sshFxNoSuchFile:
  1706. return os.ErrNotExist
  1707. case sshFxPermissionDenied:
  1708. return os.ErrPermission
  1709. case sshFxOk:
  1710. return nil
  1711. default:
  1712. return err
  1713. }
  1714. default:
  1715. return err
  1716. }
  1717. }
  1718. // flags converts the flags passed to OpenFile into ssh flags.
  1719. // Unsupported flags are ignored.
  1720. func flags(f int) uint32 {
  1721. var out uint32
  1722. switch f & os.O_WRONLY {
  1723. case os.O_WRONLY:
  1724. out |= sshFxfWrite
  1725. case os.O_RDONLY:
  1726. out |= sshFxfRead
  1727. }
  1728. if f&os.O_RDWR == os.O_RDWR {
  1729. out |= sshFxfRead | sshFxfWrite
  1730. }
  1731. if f&os.O_APPEND == os.O_APPEND {
  1732. out |= sshFxfAppend
  1733. }
  1734. if f&os.O_CREATE == os.O_CREATE {
  1735. out |= sshFxfCreat
  1736. }
  1737. if f&os.O_TRUNC == os.O_TRUNC {
  1738. out |= sshFxfTrunc
  1739. }
  1740. if f&os.O_EXCL == os.O_EXCL {
  1741. out |= sshFxfExcl
  1742. }
  1743. return out
  1744. }
  1745. // toChmodPerm converts Go permission bits to POSIX permission bits.
  1746. //
  1747. // This differs from fromFileMode in that we preserve the POSIX versions of
  1748. // setuid, setgid and sticky in m, because we've historically supported those
  1749. // bits, and we mask off any non-permission bits.
  1750. func toChmodPerm(m os.FileMode) (perm uint32) {
  1751. const mask = os.ModePerm | s_ISUID | s_ISGID | s_ISVTX
  1752. perm = uint32(m & mask)
  1753. if m&os.ModeSetuid != 0 {
  1754. perm |= s_ISUID
  1755. }
  1756. if m&os.ModeSetgid != 0 {
  1757. perm |= s_ISGID
  1758. }
  1759. if m&os.ModeSticky != 0 {
  1760. perm |= s_ISVTX
  1761. }
  1762. return perm
  1763. }