]> git.r.bdr.sh - rbdr/mobius/blob - hotline/server.go
Fix broken io.Reader implementations
[rbdr/mobius] / hotline / server.go
1 package hotline
2
3 import (
4 "bufio"
5 "bytes"
6 "context"
7 "encoding/binary"
8 "errors"
9 "fmt"
10 "github.com/go-playground/validator/v10"
11 "golang.org/x/text/encoding/charmap"
12 "gopkg.in/yaml.v3"
13 "io"
14 "io/fs"
15 "log"
16 "log/slog"
17 "math/big"
18 "math/rand"
19 "net"
20 "os"
21 "path"
22 "path/filepath"
23 "strings"
24 "sync"
25 "time"
26 )
27
28 type contextKey string
29
30 var contextKeyReq = contextKey("req")
31
32 type requestCtx struct {
33 remoteAddr string
34 }
35
36 // Converts bytes from Mac Roman encoding to UTF-8
37 var txtDecoder = charmap.Macintosh.NewDecoder()
38
39 // Converts bytes from UTF-8 to Mac Roman encoding
40 var txtEncoder = charmap.Macintosh.NewEncoder()
41
42 type Server struct {
43 NetInterface string
44 Port int
45 Accounts map[string]*Account
46 Agreement []byte
47 Clients map[uint16]*ClientConn
48 fileTransfers map[[4]byte]*FileTransfer
49
50 Config *Config
51 ConfigDir string
52 Logger *slog.Logger
53 banner []byte
54
55 PrivateChatsMu sync.Mutex
56 PrivateChats map[uint32]*PrivateChat
57
58 NextGuestID *uint16
59 TrackerPassID [4]byte
60
61 StatsMu sync.Mutex
62 Stats *Stats
63
64 FS FileStore // Storage backend to use for File storage
65
66 outbox chan Transaction
67 mux sync.Mutex
68
69 threadedNewsMux sync.Mutex
70 ThreadedNews *ThreadedNews
71
72 flatNewsMux sync.Mutex
73 FlatNews []byte
74
75 banListMU sync.Mutex
76 banList map[string]*time.Time
77 }
78
79 func (s *Server) CurrentStats() Stats {
80 s.StatsMu.Lock()
81 defer s.StatsMu.Unlock()
82
83 stats := s.Stats
84 stats.CurrentlyConnected = len(s.Clients)
85
86 return *stats
87 }
88
89 type PrivateChat struct {
90 Subject string
91 ClientConn map[uint16]*ClientConn
92 }
93
94 func (s *Server) ListenAndServe(ctx context.Context, cancelRoot context.CancelFunc) error {
95 var wg sync.WaitGroup
96
97 wg.Add(1)
98 go func() {
99 ln, err := net.Listen("tcp", fmt.Sprintf("%s:%v", s.NetInterface, s.Port))
100 if err != nil {
101 log.Fatal(err)
102 }
103
104 log.Fatal(s.Serve(ctx, ln))
105 }()
106
107 wg.Add(1)
108 go func() {
109 ln, err := net.Listen("tcp", fmt.Sprintf("%s:%v", s.NetInterface, s.Port+1))
110 if err != nil {
111 log.Fatal(err)
112 }
113
114 log.Fatal(s.ServeFileTransfers(ctx, ln))
115 }()
116
117 wg.Wait()
118
119 return nil
120 }
121
122 func (s *Server) ServeFileTransfers(ctx context.Context, ln net.Listener) error {
123 for {
124 conn, err := ln.Accept()
125 if err != nil {
126 return err
127 }
128
129 go func() {
130 defer func() { _ = conn.Close() }()
131
132 err = s.handleFileTransfer(
133 context.WithValue(ctx, contextKeyReq, requestCtx{
134 remoteAddr: conn.RemoteAddr().String(),
135 }),
136 conn,
137 )
138
139 if err != nil {
140 s.Logger.Error("file transfer error", "reason", err)
141 }
142 }()
143 }
144 }
145
146 func (s *Server) sendTransaction(t Transaction) error {
147 clientID, err := byteToInt(*t.clientID)
148 if err != nil {
149 return fmt.Errorf("invalid client ID: %v", err)
150 }
151
152 s.mux.Lock()
153 client, ok := s.Clients[uint16(clientID)]
154 s.mux.Unlock()
155 if !ok || client == nil {
156 return fmt.Errorf("invalid client id %v", *t.clientID)
157 }
158
159 _, err = io.Copy(client.Connection, &t)
160 if err != nil {
161 return fmt.Errorf("failed to send transaction to client %v: %v", clientID, err)
162 }
163
164 return nil
165 }
166
167 func (s *Server) processOutbox() {
168 for {
169 t := <-s.outbox
170 go func() {
171 if err := s.sendTransaction(t); err != nil {
172 s.Logger.Error("error sending transaction", "err", err)
173 }
174 }()
175 }
176 }
177
178 func (s *Server) Serve(ctx context.Context, ln net.Listener) error {
179 go s.processOutbox()
180
181 for {
182 conn, err := ln.Accept()
183 if err != nil {
184 s.Logger.Error("error accepting connection", "err", err)
185 }
186 connCtx := context.WithValue(ctx, contextKeyReq, requestCtx{
187 remoteAddr: conn.RemoteAddr().String(),
188 })
189
190 go func() {
191 s.Logger.Info("Connection established", "RemoteAddr", conn.RemoteAddr())
192
193 defer conn.Close()
194 if err := s.handleNewConnection(connCtx, conn, conn.RemoteAddr().String()); err != nil {
195 if err == io.EOF {
196 s.Logger.Info("Client disconnected", "RemoteAddr", conn.RemoteAddr())
197 } else {
198 s.Logger.Error("error serving request", "RemoteAddr", conn.RemoteAddr(), "err", err)
199 }
200 }
201 }()
202 }
203 }
204
205 const (
206 agreementFile = "Agreement.txt"
207 )
208
209 // NewServer constructs a new Server from a config dir
210 func NewServer(configDir, netInterface string, netPort int, logger *slog.Logger, fs FileStore) (*Server, error) {
211 server := Server{
212 NetInterface: netInterface,
213 Port: netPort,
214 Accounts: make(map[string]*Account),
215 Config: new(Config),
216 Clients: make(map[uint16]*ClientConn),
217 fileTransfers: make(map[[4]byte]*FileTransfer),
218 PrivateChats: make(map[uint32]*PrivateChat),
219 ConfigDir: configDir,
220 Logger: logger,
221 NextGuestID: new(uint16),
222 outbox: make(chan Transaction),
223 Stats: &Stats{Since: time.Now()},
224 ThreadedNews: &ThreadedNews{},
225 FS: fs,
226 banList: make(map[string]*time.Time),
227 }
228
229 var err error
230
231 // generate a new random passID for tracker registration
232 if _, err := rand.Read(server.TrackerPassID[:]); err != nil {
233 return nil, err
234 }
235
236 server.Agreement, err = os.ReadFile(filepath.Join(configDir, agreementFile))
237 if err != nil {
238 return nil, err
239 }
240
241 if server.FlatNews, err = os.ReadFile(filepath.Join(configDir, "MessageBoard.txt")); err != nil {
242 return nil, err
243 }
244
245 // try to load the ban list, but ignore errors as this file may not be present or may be empty
246 _ = server.loadBanList(filepath.Join(configDir, "Banlist.yaml"))
247
248 if err := server.loadThreadedNews(filepath.Join(configDir, "ThreadedNews.yaml")); err != nil {
249 return nil, fmt.Errorf("error loading threaded news: %w", err)
250 }
251
252 if err := server.loadConfig(filepath.Join(configDir, "config.yaml")); err != nil {
253 return nil, err
254 }
255
256 if err := server.loadAccounts(filepath.Join(configDir, "Users/")); err != nil {
257 return nil, err
258 }
259
260 // If the FileRoot is an absolute path, use it, otherwise treat as a relative path to the config dir.
261 if !filepath.IsAbs(server.Config.FileRoot) {
262 server.Config.FileRoot = filepath.Join(configDir, server.Config.FileRoot)
263 }
264
265 server.banner, err = os.ReadFile(filepath.Join(server.ConfigDir, server.Config.BannerFile))
266 if err != nil {
267 return nil, fmt.Errorf("error opening banner: %w", err)
268 }
269
270 *server.NextGuestID = 1
271
272 if server.Config.EnableTrackerRegistration {
273 server.Logger.Info(
274 "Tracker registration enabled",
275 "frequency", fmt.Sprintf("%vs", trackerUpdateFrequency),
276 "trackers", server.Config.Trackers,
277 )
278
279 go func() {
280 for {
281 tr := &TrackerRegistration{
282 UserCount: server.userCount(),
283 PassID: server.TrackerPassID,
284 Name: server.Config.Name,
285 Description: server.Config.Description,
286 }
287 binary.BigEndian.PutUint16(tr.Port[:], uint16(server.Port))
288 for _, t := range server.Config.Trackers {
289 if err := register(t, tr); err != nil {
290 server.Logger.Error("unable to register with tracker %v", "error", err)
291 }
292 server.Logger.Debug("Sent Tracker registration", "addr", t)
293 }
294
295 time.Sleep(trackerUpdateFrequency * time.Second)
296 }
297 }()
298 }
299
300 // Start Client Keepalive go routine
301 go server.keepaliveHandler()
302
303 return &server, nil
304 }
305
306 func (s *Server) userCount() int {
307 s.mux.Lock()
308 defer s.mux.Unlock()
309
310 return len(s.Clients)
311 }
312
313 func (s *Server) keepaliveHandler() {
314 for {
315 time.Sleep(idleCheckInterval * time.Second)
316 s.mux.Lock()
317
318 for _, c := range s.Clients {
319 c.IdleTime += idleCheckInterval
320 if c.IdleTime > userIdleSeconds && !c.Idle {
321 c.Idle = true
322
323 flagBitmap := big.NewInt(int64(binary.BigEndian.Uint16(c.Flags)))
324 flagBitmap.SetBit(flagBitmap, UserFlagAway, 1)
325 binary.BigEndian.PutUint16(c.Flags, uint16(flagBitmap.Int64()))
326
327 c.sendAll(
328 TranNotifyChangeUser,
329 NewField(FieldUserID, *c.ID),
330 NewField(FieldUserFlags, c.Flags),
331 NewField(FieldUserName, c.UserName),
332 NewField(FieldUserIconID, c.Icon),
333 )
334 }
335 }
336 s.mux.Unlock()
337 }
338 }
339
340 func (s *Server) writeBanList() error {
341 s.banListMU.Lock()
342 defer s.banListMU.Unlock()
343
344 out, err := yaml.Marshal(s.banList)
345 if err != nil {
346 return err
347 }
348 err = os.WriteFile(
349 filepath.Join(s.ConfigDir, "Banlist.yaml"),
350 out,
351 0666,
352 )
353 return err
354 }
355
356 func (s *Server) writeThreadedNews() error {
357 s.threadedNewsMux.Lock()
358 defer s.threadedNewsMux.Unlock()
359
360 out, err := yaml.Marshal(s.ThreadedNews)
361 if err != nil {
362 return err
363 }
364 err = s.FS.WriteFile(
365 filepath.Join(s.ConfigDir, "ThreadedNews.yaml"),
366 out,
367 0666,
368 )
369 return err
370 }
371
372 func (s *Server) NewClientConn(conn io.ReadWriteCloser, remoteAddr string) *ClientConn {
373 s.mux.Lock()
374 defer s.mux.Unlock()
375
376 clientConn := &ClientConn{
377 ID: &[]byte{0, 0},
378 Icon: []byte{0, 0},
379 Flags: []byte{0, 0},
380 UserName: []byte{},
381 Connection: conn,
382 Server: s,
383 Version: []byte{},
384 AutoReply: []byte{},
385 RemoteAddr: remoteAddr,
386 transfers: map[int]map[[4]byte]*FileTransfer{
387 FileDownload: {},
388 FileUpload: {},
389 FolderDownload: {},
390 FolderUpload: {},
391 bannerDownload: {},
392 },
393 }
394
395 *s.NextGuestID++
396 ID := *s.NextGuestID
397
398 binary.BigEndian.PutUint16(*clientConn.ID, ID)
399 s.Clients[ID] = clientConn
400
401 return clientConn
402 }
403
404 // NewUser creates a new user account entry in the server map and config file
405 func (s *Server) NewUser(login, name, password string, access accessBitmap) error {
406 s.mux.Lock()
407 defer s.mux.Unlock()
408
409 account := Account{
410 Login: login,
411 Name: name,
412 Password: hashAndSalt([]byte(password)),
413 Access: access,
414 }
415 out, err := yaml.Marshal(&account)
416 if err != nil {
417 return err
418 }
419
420 // Create account file, returning an error if one already exists.
421 file, err := os.OpenFile(
422 filepath.Join(s.ConfigDir, "Users", path.Join("/", login)+".yaml"),
423 os.O_CREATE|os.O_EXCL|os.O_WRONLY,
424 0644,
425 )
426 if err != nil {
427 return err
428 }
429 defer file.Close()
430
431 _, err = file.Write(out)
432 if err != nil {
433 return fmt.Errorf("error writing account file: %w", err)
434 }
435
436 s.Accounts[login] = &account
437
438 return nil
439 }
440
441 func (s *Server) UpdateUser(login, newLogin, name, password string, access accessBitmap) error {
442 s.mux.Lock()
443 defer s.mux.Unlock()
444
445 // update renames the user login
446 if login != newLogin {
447 err := os.Rename(filepath.Join(s.ConfigDir, "Users", path.Join("/", login)+".yaml"), filepath.Join(s.ConfigDir, "Users", path.Join("/", newLogin)+".yaml"))
448 if err != nil {
449 return fmt.Errorf("unable to rename account: %w", err)
450 }
451 s.Accounts[newLogin] = s.Accounts[login]
452 s.Accounts[newLogin].Login = newLogin
453 delete(s.Accounts, login)
454 }
455
456 account := s.Accounts[newLogin]
457 account.Access = access
458 account.Name = name
459 account.Password = password
460
461 out, err := yaml.Marshal(&account)
462 if err != nil {
463 return err
464 }
465
466 if err := os.WriteFile(filepath.Join(s.ConfigDir, "Users", newLogin+".yaml"), out, 0666); err != nil {
467 return err
468 }
469
470 return nil
471 }
472
473 // DeleteUser deletes the user account
474 func (s *Server) DeleteUser(login string) error {
475 s.mux.Lock()
476 defer s.mux.Unlock()
477
478 err := s.FS.Remove(filepath.Join(s.ConfigDir, "Users", path.Join("/", login)+".yaml"))
479 if err != nil {
480 return err
481 }
482
483 delete(s.Accounts, login)
484
485 return nil
486 }
487
488 func (s *Server) connectedUsers() []Field {
489 s.mux.Lock()
490 defer s.mux.Unlock()
491
492 var connectedUsers []Field
493 for _, c := range sortedClients(s.Clients) {
494 b, err := io.ReadAll(&User{
495 ID: *c.ID,
496 Icon: c.Icon,
497 Flags: c.Flags,
498 Name: string(c.UserName),
499 })
500 if err != nil {
501 return nil
502 }
503 connectedUsers = append(connectedUsers, NewField(FieldUsernameWithInfo, b))
504 }
505 return connectedUsers
506 }
507
508 func (s *Server) loadBanList(path string) error {
509 fh, err := os.Open(path)
510 if err != nil {
511 return err
512 }
513 decoder := yaml.NewDecoder(fh)
514
515 return decoder.Decode(s.banList)
516 }
517
518 // loadThreadedNews loads the threaded news data from disk
519 func (s *Server) loadThreadedNews(threadedNewsPath string) error {
520 fh, err := os.Open(threadedNewsPath)
521 if err != nil {
522 return err
523 }
524 decoder := yaml.NewDecoder(fh)
525
526 return decoder.Decode(s.ThreadedNews)
527 }
528
529 // loadAccounts loads account data from disk
530 func (s *Server) loadAccounts(userDir string) error {
531 matches, err := filepath.Glob(filepath.Join(userDir, "*.yaml"))
532 if err != nil {
533 return err
534 }
535
536 if len(matches) == 0 {
537 return errors.New("no user accounts found in " + userDir)
538 }
539
540 for _, file := range matches {
541 fh, err := s.FS.Open(file)
542 if err != nil {
543 return err
544 }
545
546 account := Account{}
547 decoder := yaml.NewDecoder(fh)
548 if err = decoder.Decode(&account); err != nil {
549 return fmt.Errorf("error loading account %s: %w", file, err)
550 }
551
552 s.Accounts[account.Login] = &account
553 }
554 return nil
555 }
556
557 func (s *Server) loadConfig(path string) error {
558 fh, err := s.FS.Open(path)
559 if err != nil {
560 return err
561 }
562
563 decoder := yaml.NewDecoder(fh)
564 err = decoder.Decode(s.Config)
565 if err != nil {
566 return err
567 }
568
569 validate := validator.New()
570 err = validate.Struct(s.Config)
571 if err != nil {
572 return err
573 }
574 return nil
575 }
576
577 // handleNewConnection takes a new net.Conn and performs the initial login sequence
578 func (s *Server) handleNewConnection(ctx context.Context, rwc io.ReadWriteCloser, remoteAddr string) error {
579 defer dontPanic(s.Logger)
580
581 if err := Handshake(rwc); err != nil {
582 return err
583 }
584
585 // Create a new scanner for parsing incoming bytes into transaction tokens
586 scanner := bufio.NewScanner(rwc)
587 scanner.Split(transactionScanner)
588
589 scanner.Scan()
590
591 // Make a new []byte slice and copy the scanner bytes to it. This is critical to avoid a data race as the
592 // scanner re-uses the buffer for subsequent scans.
593 buf := make([]byte, len(scanner.Bytes()))
594 copy(buf, scanner.Bytes())
595
596 var clientLogin Transaction
597 if _, err := clientLogin.Write(buf); err != nil {
598 return err
599 }
600
601 // check if remoteAddr is present in the ban list
602 if banUntil, ok := s.banList[strings.Split(remoteAddr, ":")[0]]; ok {
603 // permaban
604 if banUntil == nil {
605 t := NewTransaction(
606 TranServerMsg,
607 &[]byte{0, 0},
608 NewField(FieldData, []byte("You are permanently banned on this server")),
609 NewField(FieldChatOptions, []byte{0, 0}),
610 )
611
612 _, err := io.Copy(rwc, t)
613 if err != nil {
614 return err
615 }
616
617 time.Sleep(1 * time.Second)
618 return nil
619 }
620
621 // temporary ban
622 if time.Now().Before(*banUntil) {
623 t := NewTransaction(
624 TranServerMsg,
625 &[]byte{0, 0},
626 NewField(FieldData, []byte("You are temporarily banned on this server")),
627 NewField(FieldChatOptions, []byte{0, 0}),
628 )
629
630 _, err := io.Copy(rwc, t)
631 if err != nil {
632 return err
633 }
634
635 time.Sleep(1 * time.Second)
636 return nil
637 }
638 }
639
640 c := s.NewClientConn(rwc, remoteAddr)
641 defer c.Disconnect()
642
643 encodedLogin := clientLogin.GetField(FieldUserLogin).Data
644 encodedPassword := clientLogin.GetField(FieldUserPassword).Data
645 c.Version = clientLogin.GetField(FieldVersion).Data
646
647 var login string
648 for _, char := range encodedLogin {
649 login += string(rune(255 - uint(char)))
650 }
651 if login == "" {
652 login = GuestAccount
653 }
654
655 c.logger = s.Logger.With("remoteAddr", remoteAddr, "login", login)
656
657 // If authentication fails, send error reply and close connection
658 if !c.Authenticate(login, encodedPassword) {
659 t := c.NewErrReply(&clientLogin, "Incorrect login.")
660
661 _, err := io.Copy(rwc, &t)
662 if err != nil {
663 return err
664 }
665
666 c.logger.Info("Login failed", "clientVersion", fmt.Sprintf("%x", c.Version))
667
668 return nil
669 }
670
671 if clientLogin.GetField(FieldUserIconID).Data != nil {
672 c.Icon = clientLogin.GetField(FieldUserIconID).Data
673 }
674
675 c.Account = c.Server.Accounts[login]
676
677 if clientLogin.GetField(FieldUserName).Data != nil {
678 if c.Authorize(accessAnyName) {
679 c.UserName = clientLogin.GetField(FieldUserName).Data
680 } else {
681 c.UserName = []byte(c.Account.Name)
682 }
683 }
684
685 if c.Authorize(accessDisconUser) {
686 c.Flags = []byte{0, 2}
687 }
688
689 s.outbox <- c.NewReply(&clientLogin,
690 NewField(FieldVersion, []byte{0x00, 0xbe}),
691 NewField(FieldCommunityBannerID, []byte{0, 0}),
692 NewField(FieldServerName, []byte(s.Config.Name)),
693 )
694
695 // Send user access privs so client UI knows how to behave
696 c.Server.outbox <- *NewTransaction(TranUserAccess, c.ID, NewField(FieldUserAccess, c.Account.Access[:]))
697
698 // Accounts with accessNoAgreement do not receive the server agreement on login. The behavior is different between
699 // client versions. For 1.2.3 client, we do not send TranShowAgreement. For other client versions, we send
700 // TranShowAgreement but with the NoServerAgreement field set to 1.
701 if c.Authorize(accessNoAgreement) {
702 // If client version is nil, then the client uses the 1.2.3 login behavior
703 if c.Version != nil {
704 c.Server.outbox <- *NewTransaction(TranShowAgreement, c.ID, NewField(FieldNoServerAgreement, []byte{1}))
705 }
706 } else {
707 c.Server.outbox <- *NewTransaction(TranShowAgreement, c.ID, NewField(FieldData, s.Agreement))
708 }
709
710 // If the client has provided a username as part of the login, we can infer that it is using the 1.2.3 login
711 // flow and not the 1.5+ flow.
712 if len(c.UserName) != 0 {
713 // Add the client username to the logger. For 1.5+ clients, we don't have this information yet as it comes as
714 // part of TranAgreed
715 c.logger = c.logger.With("Name", string(c.UserName))
716
717 c.logger.Info("Login successful", "clientVersion", "Not sent (probably 1.2.3)")
718
719 // Notify other clients on the server that the new user has logged in. For 1.5+ clients we don't have this
720 // information yet, so we do it in TranAgreed instead
721 for _, t := range c.notifyOthers(
722 *NewTransaction(
723 TranNotifyChangeUser, nil,
724 NewField(FieldUserName, c.UserName),
725 NewField(FieldUserID, *c.ID),
726 NewField(FieldUserIconID, c.Icon),
727 NewField(FieldUserFlags, c.Flags),
728 ),
729 ) {
730 c.Server.outbox <- t
731 }
732 }
733
734 c.Server.Stats.ConnectionCounter += 1
735 if len(s.Clients) > c.Server.Stats.ConnectionPeak {
736 c.Server.Stats.ConnectionPeak = len(s.Clients)
737 }
738
739 // Scan for new transactions and handle them as they come in.
740 for scanner.Scan() {
741 // Make a new []byte slice and copy the scanner bytes to it. This is critical to avoid a data race as the
742 // scanner re-uses the buffer for subsequent scans.
743 buf := make([]byte, len(scanner.Bytes()))
744 copy(buf, scanner.Bytes())
745
746 var t Transaction
747 if _, err := t.Write(buf); err != nil {
748 return err
749 }
750
751 if err := c.handleTransaction(t); err != nil {
752 c.logger.Error("Error handling transaction", "err", err)
753 }
754 }
755 return nil
756 }
757
758 func (s *Server) NewPrivateChat(cc *ClientConn) []byte {
759 s.PrivateChatsMu.Lock()
760 defer s.PrivateChatsMu.Unlock()
761
762 randID := make([]byte, 4)
763 rand.Read(randID)
764 data := binary.BigEndian.Uint32(randID)
765
766 s.PrivateChats[data] = &PrivateChat{
767 ClientConn: make(map[uint16]*ClientConn),
768 }
769 s.PrivateChats[data].ClientConn[cc.uint16ID()] = cc
770
771 return randID
772 }
773
774 const dlFldrActionSendFile = 1
775 const dlFldrActionResumeFile = 2
776 const dlFldrActionNextFile = 3
777
778 // handleFileTransfer receives a client net.Conn from the file transfer server, performs the requested transfer type, then closes the connection
779 func (s *Server) handleFileTransfer(ctx context.Context, rwc io.ReadWriter) error {
780 defer dontPanic(s.Logger)
781
782 txBuf := make([]byte, 16)
783 if _, err := io.ReadFull(rwc, txBuf); err != nil {
784 return err
785 }
786
787 var t transfer
788 if _, err := t.Write(txBuf); err != nil {
789 return err
790 }
791
792 defer func() {
793 s.mux.Lock()
794 delete(s.fileTransfers, t.ReferenceNumber)
795 s.mux.Unlock()
796
797 // Wait a few seconds before closing the connection: this is a workaround for problems
798 // observed with Windows clients where the client must initiate close of the TCP connection before
799 // the server does. This is gross and seems unnecessary. TODO: Revisit?
800 time.Sleep(3 * time.Second)
801 }()
802
803 s.mux.Lock()
804 fileTransfer, ok := s.fileTransfers[t.ReferenceNumber]
805 s.mux.Unlock()
806 if !ok {
807 return errors.New("invalid transaction ID")
808 }
809
810 defer func() {
811 fileTransfer.ClientConn.transfersMU.Lock()
812 delete(fileTransfer.ClientConn.transfers[fileTransfer.Type], t.ReferenceNumber)
813 fileTransfer.ClientConn.transfersMU.Unlock()
814 }()
815
816 rLogger := s.Logger.With(
817 "remoteAddr", ctx.Value(contextKeyReq).(requestCtx).remoteAddr,
818 "login", fileTransfer.ClientConn.Account.Login,
819 "Name", string(fileTransfer.ClientConn.UserName),
820 )
821
822 fullPath, err := readPath(s.Config.FileRoot, fileTransfer.FilePath, fileTransfer.FileName)
823 if err != nil {
824 return err
825 }
826
827 switch fileTransfer.Type {
828 case bannerDownload:
829 if _, err := io.Copy(rwc, bytes.NewBuffer(s.banner)); err != nil {
830 return fmt.Errorf("error sending banner: %w", err)
831 }
832 case FileDownload:
833 s.Stats.DownloadCounter += 1
834 s.Stats.DownloadsInProgress += 1
835 defer func() {
836 s.Stats.DownloadsInProgress -= 1
837 }()
838
839 var dataOffset int64
840 if fileTransfer.fileResumeData != nil {
841 dataOffset = int64(binary.BigEndian.Uint32(fileTransfer.fileResumeData.ForkInfoList[0].DataSize[:]))
842 }
843
844 fw, err := newFileWrapper(s.FS, fullPath, 0)
845 if err != nil {
846 return err
847 }
848
849 rLogger.Info("File download started", "filePath", fullPath)
850
851 // if file transfer options are included, that means this is a "quick preview" request from a 1.5+ client
852 if fileTransfer.options == nil {
853 _, err = io.Copy(rwc, fw.ffo)
854 if err != nil {
855 return err
856 }
857 }
858
859 file, err := fw.dataForkReader()
860 if err != nil {
861 return err
862 }
863
864 br := bufio.NewReader(file)
865 if _, err := br.Discard(int(dataOffset)); err != nil {
866 return err
867 }
868
869 if _, err = io.Copy(rwc, io.TeeReader(br, fileTransfer.bytesSentCounter)); err != nil {
870 return err
871 }
872
873 // if the client requested to resume transfer, do not send the resource fork header, or it will be appended into the fileWrapper data
874 if fileTransfer.fileResumeData == nil {
875 err = binary.Write(rwc, binary.BigEndian, fw.rsrcForkHeader())
876 if err != nil {
877 return err
878 }
879 }
880
881 rFile, err := fw.rsrcForkFile()
882 if err != nil {
883 return nil
884 }
885
886 if _, err = io.Copy(rwc, io.TeeReader(rFile, fileTransfer.bytesSentCounter)); err != nil {
887 return err
888 }
889
890 case FileUpload:
891 s.Stats.UploadCounter += 1
892 s.Stats.UploadsInProgress += 1
893 defer func() { s.Stats.UploadsInProgress -= 1 }()
894
895 var file *os.File
896
897 // A file upload has three possible cases:
898 // 1) Upload a new file
899 // 2) Resume a partially transferred file
900 // 3) Replace a fully uploaded file
901 // We have to infer which case applies by inspecting what is already on the filesystem
902
903 // 1) Check for existing file:
904 _, err = os.Stat(fullPath)
905 if err == nil {
906 return errors.New("existing file found at " + fullPath)
907 }
908 if errors.Is(err, fs.ErrNotExist) {
909 // If not found, open or create a new .incomplete file
910 file, err = os.OpenFile(fullPath+incompleteFileSuffix, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
911 if err != nil {
912 return err
913 }
914 }
915
916 f, err := newFileWrapper(s.FS, fullPath, 0)
917 if err != nil {
918 return err
919 }
920
921 rLogger.Info("File upload started", "dstFile", fullPath)
922
923 rForkWriter := io.Discard
924 iForkWriter := io.Discard
925 if s.Config.PreserveResourceForks {
926 rForkWriter, err = f.rsrcForkWriter()
927 if err != nil {
928 return err
929 }
930
931 iForkWriter, err = f.infoForkWriter()
932 if err != nil {
933 return err
934 }
935 }
936
937 if err := receiveFile(rwc, file, rForkWriter, iForkWriter, fileTransfer.bytesSentCounter); err != nil {
938 s.Logger.Error(err.Error())
939 }
940
941 if err := file.Close(); err != nil {
942 return err
943 }
944
945 if err := s.FS.Rename(fullPath+".incomplete", fullPath); err != nil {
946 return err
947 }
948
949 rLogger.Info("File upload complete", "dstFile", fullPath)
950
951 case FolderDownload:
952 s.Stats.DownloadCounter += 1
953 s.Stats.DownloadsInProgress += 1
954 defer func() { s.Stats.DownloadsInProgress -= 1 }()
955
956 // Folder Download flow:
957 // 1. Get filePath from the transfer
958 // 2. Iterate over files
959 // 3. For each fileWrapper:
960 // Send fileWrapper header to client
961 // The client can reply in 3 ways:
962 //
963 // 1. If type is an odd number (unknown type?), or fileWrapper download for the current fileWrapper is completed:
964 // client sends []byte{0x00, 0x03} to tell the server to continue to the next fileWrapper
965 //
966 // 2. If download of a fileWrapper is to be resumed:
967 // client sends:
968 // []byte{0x00, 0x02} // download folder action
969 // [2]byte // Resume data size
970 // []byte fileWrapper resume data (see myField_FileResumeData)
971 //
972 // 3. Otherwise, download of the fileWrapper is requested and client sends []byte{0x00, 0x01}
973 //
974 // When download is requested (case 2 or 3), server replies with:
975 // [4]byte - fileWrapper size
976 // []byte - Flattened File Object
977 //
978 // After every fileWrapper download, client could request next fileWrapper with:
979 // []byte{0x00, 0x03}
980 //
981 // This notifies the server to send the next item header
982
983 basePathLen := len(fullPath)
984
985 rLogger.Info("Start folder download", "path", fullPath)
986
987 nextAction := make([]byte, 2)
988 if _, err := io.ReadFull(rwc, nextAction); err != nil {
989 return err
990 }
991
992 i := 0
993 err = filepath.Walk(fullPath+"/", func(path string, info os.FileInfo, err error) error {
994 s.Stats.DownloadCounter += 1
995 i += 1
996
997 if err != nil {
998 return err
999 }
1000
1001 // skip dot files
1002 if strings.HasPrefix(info.Name(), ".") {
1003 return nil
1004 }
1005
1006 hlFile, err := newFileWrapper(s.FS, path, 0)
1007 if err != nil {
1008 return err
1009 }
1010
1011 subPath := path[basePathLen+1:]
1012 rLogger.Debug("Sending fileheader", "i", i, "path", path, "fullFilePath", fullPath, "subPath", subPath, "IsDir", info.IsDir())
1013
1014 if i == 1 {
1015 return nil
1016 }
1017
1018 fileHeader := NewFileHeader(subPath, info.IsDir())
1019 if _, err := io.Copy(rwc, &fileHeader); err != nil {
1020 return fmt.Errorf("error sending file header: %w", err)
1021 }
1022
1023 // Read the client's Next Action request
1024 if _, err := io.ReadFull(rwc, nextAction); err != nil {
1025 return err
1026 }
1027
1028 rLogger.Debug("Client folder download action", "action", fmt.Sprintf("%X", nextAction[0:2]))
1029
1030 var dataOffset int64
1031
1032 switch nextAction[1] {
1033 case dlFldrActionResumeFile:
1034 // get size of resumeData
1035 resumeDataByteLen := make([]byte, 2)
1036 if _, err := io.ReadFull(rwc, resumeDataByteLen); err != nil {
1037 return err
1038 }
1039
1040 resumeDataLen := binary.BigEndian.Uint16(resumeDataByteLen)
1041 resumeDataBytes := make([]byte, resumeDataLen)
1042 if _, err := io.ReadFull(rwc, resumeDataBytes); err != nil {
1043 return err
1044 }
1045
1046 var frd FileResumeData
1047 if err := frd.UnmarshalBinary(resumeDataBytes); err != nil {
1048 return err
1049 }
1050 dataOffset = int64(binary.BigEndian.Uint32(frd.ForkInfoList[0].DataSize[:]))
1051 case dlFldrActionNextFile:
1052 // client asked to skip this file
1053 return nil
1054 }
1055
1056 if info.IsDir() {
1057 return nil
1058 }
1059
1060 rLogger.Info("File download started",
1061 "fileName", info.Name(),
1062 "TransferSize", fmt.Sprintf("%x", hlFile.ffo.TransferSize(dataOffset)),
1063 )
1064
1065 // Send file size to client
1066 if _, err := rwc.Write(hlFile.ffo.TransferSize(dataOffset)); err != nil {
1067 s.Logger.Error(err.Error())
1068 return err
1069 }
1070
1071 // Send ffo bytes to client
1072 _, err = io.Copy(rwc, hlFile.ffo)
1073 if err != nil {
1074 return err
1075 }
1076
1077 file, err := s.FS.Open(path)
1078 if err != nil {
1079 return err
1080 }
1081
1082 // wr := bufio.NewWriterSize(rwc, 1460)
1083 if _, err = io.Copy(rwc, io.TeeReader(file, fileTransfer.bytesSentCounter)); err != nil {
1084 return err
1085 }
1086
1087 if nextAction[1] != 2 && hlFile.ffo.FlatFileHeader.ForkCount[1] == 3 {
1088 err = binary.Write(rwc, binary.BigEndian, hlFile.rsrcForkHeader())
1089 if err != nil {
1090 return err
1091 }
1092
1093 rFile, err := hlFile.rsrcForkFile()
1094 if err != nil {
1095 return err
1096 }
1097
1098 if _, err = io.Copy(rwc, io.TeeReader(rFile, fileTransfer.bytesSentCounter)); err != nil {
1099 return err
1100 }
1101 }
1102
1103 // Read the client's Next Action request. This is always 3, I think?
1104 if _, err := io.ReadFull(rwc, nextAction); err != nil {
1105 return err
1106 }
1107
1108 return nil
1109 })
1110
1111 if err != nil {
1112 return err
1113 }
1114
1115 case FolderUpload:
1116 s.Stats.UploadCounter += 1
1117 s.Stats.UploadsInProgress += 1
1118 defer func() { s.Stats.UploadsInProgress -= 1 }()
1119 rLogger.Info(
1120 "Folder upload started",
1121 "dstPath", fullPath,
1122 "TransferSize", binary.BigEndian.Uint32(fileTransfer.TransferSize),
1123 "FolderItemCount", fileTransfer.FolderItemCount,
1124 )
1125
1126 // Check if the target folder exists. If not, create it.
1127 if _, err := s.FS.Stat(fullPath); os.IsNotExist(err) {
1128 if err := s.FS.Mkdir(fullPath, 0777); err != nil {
1129 return err
1130 }
1131 }
1132
1133 // Begin the folder upload flow by sending the "next file action" to client
1134 if _, err := rwc.Write([]byte{0, dlFldrActionNextFile}); err != nil {
1135 return err
1136 }
1137
1138 fileSize := make([]byte, 4)
1139
1140 for i := 0; i < fileTransfer.ItemCount(); i++ {
1141 s.Stats.UploadCounter += 1
1142
1143 var fu folderUpload
1144 if _, err := io.ReadFull(rwc, fu.DataSize[:]); err != nil {
1145 return err
1146 }
1147 if _, err := io.ReadFull(rwc, fu.IsFolder[:]); err != nil {
1148 return err
1149 }
1150 if _, err := io.ReadFull(rwc, fu.PathItemCount[:]); err != nil {
1151 return err
1152 }
1153
1154 fu.FileNamePath = make([]byte, binary.BigEndian.Uint16(fu.DataSize[:])-4) // -4 to subtract the path separator bytes
1155
1156 if _, err := io.ReadFull(rwc, fu.FileNamePath); err != nil {
1157 return err
1158 }
1159
1160 rLogger.Info(
1161 "Folder upload continued",
1162 "FormattedPath", fu.FormattedPath(),
1163 "IsFolder", fmt.Sprintf("%x", fu.IsFolder),
1164 "PathItemCount", binary.BigEndian.Uint16(fu.PathItemCount[:]),
1165 )
1166
1167 if fu.IsFolder == [2]byte{0, 1} {
1168 if _, err := os.Stat(filepath.Join(fullPath, fu.FormattedPath())); os.IsNotExist(err) {
1169 if err := os.Mkdir(filepath.Join(fullPath, fu.FormattedPath()), 0777); err != nil {
1170 return err
1171 }
1172 }
1173
1174 // Tell client to send next file
1175 if _, err := rwc.Write([]byte{0, dlFldrActionNextFile}); err != nil {
1176 return err
1177 }
1178 } else {
1179 nextAction := dlFldrActionSendFile
1180
1181 // Check if we have the full file already. If so, send dlFldrAction_NextFile to client to skip.
1182 _, err = os.Stat(filepath.Join(fullPath, fu.FormattedPath()))
1183 if err != nil && !errors.Is(err, fs.ErrNotExist) {
1184 return err
1185 }
1186 if err == nil {
1187 nextAction = dlFldrActionNextFile
1188 }
1189
1190 // Check if we have a partial file already. If so, send dlFldrAction_ResumeFile to client to resume upload.
1191 incompleteFile, err := os.Stat(filepath.Join(fullPath, fu.FormattedPath()+incompleteFileSuffix))
1192 if err != nil && !errors.Is(err, fs.ErrNotExist) {
1193 return err
1194 }
1195 if err == nil {
1196 nextAction = dlFldrActionResumeFile
1197 }
1198
1199 if _, err := rwc.Write([]byte{0, uint8(nextAction)}); err != nil {
1200 return err
1201 }
1202
1203 switch nextAction {
1204 case dlFldrActionNextFile:
1205 continue
1206 case dlFldrActionResumeFile:
1207 offset := make([]byte, 4)
1208 binary.BigEndian.PutUint32(offset, uint32(incompleteFile.Size()))
1209
1210 file, err := os.OpenFile(fullPath+"/"+fu.FormattedPath()+incompleteFileSuffix, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
1211 if err != nil {
1212 return err
1213 }
1214
1215 fileResumeData := NewFileResumeData([]ForkInfoList{*NewForkInfoList(offset)})
1216
1217 b, _ := fileResumeData.BinaryMarshal()
1218
1219 bs := make([]byte, 2)
1220 binary.BigEndian.PutUint16(bs, uint16(len(b)))
1221
1222 if _, err := rwc.Write(append(bs, b...)); err != nil {
1223 return err
1224 }
1225
1226 if _, err := io.ReadFull(rwc, fileSize); err != nil {
1227 return err
1228 }
1229
1230 if err := receiveFile(rwc, file, io.Discard, io.Discard, fileTransfer.bytesSentCounter); err != nil {
1231 s.Logger.Error(err.Error())
1232 }
1233
1234 err = os.Rename(fullPath+"/"+fu.FormattedPath()+".incomplete", fullPath+"/"+fu.FormattedPath())
1235 if err != nil {
1236 return err
1237 }
1238
1239 case dlFldrActionSendFile:
1240 if _, err := io.ReadFull(rwc, fileSize); err != nil {
1241 return err
1242 }
1243
1244 filePath := filepath.Join(fullPath, fu.FormattedPath())
1245
1246 hlFile, err := newFileWrapper(s.FS, filePath, 0)
1247 if err != nil {
1248 return err
1249 }
1250
1251 rLogger.Info("Starting file transfer", "path", filePath, "fileNum", i+1, "fileSize", binary.BigEndian.Uint32(fileSize))
1252
1253 incWriter, err := hlFile.incFileWriter()
1254 if err != nil {
1255 return err
1256 }
1257
1258 rForkWriter := io.Discard
1259 iForkWriter := io.Discard
1260 if s.Config.PreserveResourceForks {
1261 iForkWriter, err = hlFile.infoForkWriter()
1262 if err != nil {
1263 return err
1264 }
1265
1266 rForkWriter, err = hlFile.rsrcForkWriter()
1267 if err != nil {
1268 return err
1269 }
1270 }
1271 if err := receiveFile(rwc, incWriter, rForkWriter, iForkWriter, fileTransfer.bytesSentCounter); err != nil {
1272 return err
1273 }
1274
1275 if err := os.Rename(filePath+".incomplete", filePath); err != nil {
1276 return err
1277 }
1278 }
1279
1280 // Tell client to send next fileWrapper
1281 if _, err := rwc.Write([]byte{0, dlFldrActionNextFile}); err != nil {
1282 return err
1283 }
1284 }
1285 }
1286 rLogger.Info("Folder upload complete")
1287 }
1288
1289 return nil
1290 }