// Package sftpserver implements an embedded, per-tenant SFTP server for // archivdms. Instead of provisioning real OS users + OpenSSH // ChrootDirectory per tenant, the server runs inside the archivdms binary // and enforces tenant isolation entirely in software: // // - Authentication is checked against the `sftp_credentials` table // (internal/storage/sftp_credentials.go), a narrow, independently // revocable credential — not a full user login. // - Once authenticated, a tenant is virtually "locked" into // `/inbox//`: the SFTP handlers only ever // resolve paths relative to that directory and reject any path that // would escape it (no OS-level chroot, no setuid, no real filesystem // jail — just careful path handling). // - A polling watcher goroutine picks up files dropped into that // directory and feeds them through the exact same // inbox->hash->store->OCR->DB pipeline as the HTTP upload endpoint // (see internal/api/document_handlers.go storeUploadedFile, exposed // here via the UploadFunc callback to avoid an import cycle between // internal/api and internal/sftpserver). package sftpserver import ( "context" "crypto/rand" "crypto/rsa" "crypto/x509" "encoding/pem" "errors" "fmt" "io" "log/slog" "net" "os" "path/filepath" "strconv" "strings" "sync" "time" "github.com/pkg/sftp" "golang.org/x/crypto/ssh" "archivdms/config" "archivdms/internal/audit" "archivdms/internal/storage" ) // pollInterval is how often the watcher scans inbox directories for new // files dropped over SFTP. No fsnotify dependency — a simple polling loop is // good enough for this volume/latency profile (analogous to the project's // existing cron-style background jobs). const pollInterval = 5 * time.Second // UploadFunc is the shared upload-pipeline entry point, implemented by // internal/api.Server.StoreUploadedFile. Kept as a function value (rather // than importing internal/api directly) to avoid an import cycle: // internal/api already imports internal/storage and internal/audit, and // wiring happens the other way around in cmd/archivdms/main.go. type UploadFunc func(ctx context.Context, tenantID int64, title, docType, correspondent string, file io.Reader, filename, contentType string) (*storage.Document, string, error) // Server is the embedded per-tenant SFTP server. type Server struct { cfg config.SFTPConfig storageCfg config.StorageConfig store *storage.Store audlog *audit.Logger logger *slog.Logger upload UploadFunc listener net.Listener sshCfg *ssh.ServerConfig stopOnce sync.Once stopCh chan struct{} } // New constructs an SFTP server. Call Start to begin listening and Stop to // shut down. func New(cfg config.SFTPConfig, storageCfg config.StorageConfig, store *storage.Store, audlog *audit.Logger, logger *slog.Logger, upload UploadFunc) *Server { return &Server{ cfg: cfg, storageCfg: storageCfg, store: store, audlog: audlog, logger: logger, upload: upload, stopCh: make(chan struct{}), } } // Start loads/generates the host key, opens the listener, and launches the // accept loop plus the inbox watcher as background goroutines. It returns // once the listener is up (or an error occurred setting it up); the accept // loop itself keeps running in the background. func (s *Server) Start(ctx context.Context) error { signer, err := s.loadOrCreateHostKey() if err != nil { return fmt.Errorf("sftpserver: host key: %w", err) } s.sshCfg = &ssh.ServerConfig{ PasswordCallback: s.passwordCallback, } s.sshCfg.AddHostKey(signer) bind := s.cfg.ResolvedBind() ln, err := net.Listen("tcp", bind) if err != nil { return fmt.Errorf("sftpserver: listen %s: %w", bind, err) } s.listener = ln s.logger.Info("sftp server listening", "addr", bind) go s.acceptLoop() go s.watchLoop(ctx) return nil } // Stop closes the listener, ending the accept loop, and signals the watcher // to exit. func (s *Server) Stop() { s.stopOnce.Do(func() { close(s.stopCh) if s.listener != nil { _ = s.listener.Close() } }) } // --- host key bootstrap --- func (s *Server) loadOrCreateHostKey() (ssh.Signer, error) { path := s.cfg.ResolvedHostKeyPath(s.storageCfg.BasePath) if data, err := os.ReadFile(path); err == nil { return ssh.ParsePrivateKey(data) } else if !os.IsNotExist(err) { return nil, err } s.logger.Info("sftp host key not found, generating a new one", "path", path) key, err := rsa.GenerateKey(rand.Reader, 4096) if err != nil { return nil, fmt.Errorf("generate host key: %w", err) } der := x509.MarshalPKCS1PrivateKey(key) block := &pem.Block{Type: "RSA PRIVATE KEY", Bytes: der} if dir := filepath.Dir(path); dir != "" && dir != "." { if err := os.MkdirAll(dir, 0o700); err != nil { return nil, fmt.Errorf("create host key dir: %w", err) } } if err := os.WriteFile(path, pem.EncodeToMemory(block), 0o600); err != nil { return nil, fmt.Errorf("write host key: %w", err) } return ssh.NewSignerFromKey(key) } // --- authentication --- func (s *Server) passwordCallback(meta ssh.ConnMetadata, password []byte) (*ssh.Permissions, error) { username := meta.User() ctx := context.Background() cred, err := s.store.VerifySFTPLogin(ctx, username, string(password)) success := err == nil detail := "" if err != nil { detail = err.Error() } var tenantID *int64 if cred != nil { tenantID = &cred.TenantID } if s.audlog != nil { s.audlog.Log(audit.Entry{ EventType: audit.EventSFTPLogin, Username: username, IPAddress: remoteIPFromConn(meta.RemoteAddr()), TenantID: tenantID, Success: success, Detail: detail, }) } if !success { return nil, fmt.Errorf("sftpserver: authentication failed") } _ = s.store.TouchSFTPLastLogin(ctx, cred.ID) return &ssh.Permissions{ Extensions: map[string]string{ "tenant_id": strconv.FormatInt(cred.TenantID, 10), "username": username, }, }, nil } func remoteIPFromConn(addr net.Addr) string { if addr == nil { return "" } host, _, err := net.SplitHostPort(addr.String()) if err != nil { return addr.String() } return host } // --- accept loop --- func (s *Server) acceptLoop() { for { conn, err := s.listener.Accept() if err != nil { select { case <-s.stopCh: return default: s.logger.Warn("sftp accept error", "err", err) continue } } go s.handleConn(conn) } } func (s *Server) handleConn(conn net.Conn) { sshConn, chans, reqs, err := ssh.NewServerConn(conn, s.sshCfg) if err != nil { s.logger.Warn("sftp handshake failed", "err", err, "remote", conn.RemoteAddr()) return } defer sshConn.Close() tenantIDStr := sshConn.Permissions.Extensions["tenant_id"] tenantID, err := strconv.ParseInt(tenantIDStr, 10, 64) if err != nil { s.logger.Error("sftp connection missing tenant_id extension", "err", err) return } go ssh.DiscardRequests(reqs) for newChan := range chans { if newChan.ChannelType() != "session" { _ = newChan.Reject(ssh.UnknownChannelType, "unsupported channel type") continue } channel, requests, err := newChan.Accept() if err != nil { s.logger.Warn("sftp channel accept failed", "err", err) continue } go s.handleSession(channel, requests, tenantID) } } func (s *Server) handleSession(channel ssh.Channel, requests <-chan *ssh.Request, tenantID int64) { defer channel.Close() for req := range requests { ok := req.Type == "subsystem" && string(req.Payload[4:]) == "sftp" if req.WantReply { _ = req.Reply(ok, nil) } if !ok { continue } root := filepath.Join(s.storageCfg.InboxPath(), strconv.FormatInt(tenantID, 10)) if err := os.MkdirAll(root, 0o750); err != nil { s.logger.Error("sftp: create tenant inbox dir failed", "tenant_id", tenantID, "err", err) return } fs := &tenantFS{root: root} handlers := sftp.Handlers{ FileGet: fs, FilePut: fs, FileCmd: fs, FileList: fs, } server := sftp.NewRequestServer(channel, handlers) if err := server.Serve(); err != nil && err != io.EOF { s.logger.Warn("sftp session ended with error", "tenant_id", tenantID, "err", err) } _ = server.Close() return } } // --- watcher: picks up files dropped into inbox// and feeds them // through the shared upload pipeline --- func (s *Server) watchLoop(ctx context.Context) { ticker := time.NewTicker(pollInterval) defer ticker.Stop() for { select { case <-s.stopCh: return case <-ctx.Done(): return case <-ticker.C: s.scanInbox(ctx) } } } func (s *Server) scanInbox(ctx context.Context) { base := s.storageCfg.InboxPath() tenantDirs, err := os.ReadDir(base) if err != nil { if !os.IsNotExist(err) { s.logger.Warn("sftp watcher: read inbox root failed", "err", err) } return } for _, td := range tenantDirs { if !td.IsDir() { continue } tenantID, err := strconv.ParseInt(td.Name(), 10, 64) if err != nil { continue // not a tenant directory (e.g. stray file), skip } s.scanTenantInbox(ctx, tenantID, filepath.Join(base, td.Name())) } } func (s *Server) scanTenantInbox(ctx context.Context, tenantID int64, dir string) { entries, err := os.ReadDir(dir) if err != nil { s.logger.Warn("sftp watcher: read tenant inbox failed", "tenant_id", tenantID, "err", err) return } for _, e := range entries { if e.IsDir() { continue } path := filepath.Join(dir, e.Name()) s.processInboxFile(ctx, tenantID, path, e.Name()) } } func (s *Server) processInboxFile(ctx context.Context, tenantID int64, path, filename string) { // Skip files still being written (e.g. an in-progress SFTP PUT). A // simple heuristic: if the file's mtime is very recent, give the next // poll cycle a chance to see it settle instead of processing a partial // upload. info, err := os.Stat(path) if err != nil { return // vanished since ReadDir, e.g. concurrent processing } if time.Since(info.ModTime()) < pollInterval { return } f, err := os.Open(path) if err != nil { s.logger.Warn("sftp watcher: open inbox file failed", "path", path, "err", err) return } title := strings.TrimSuffix(filename, filepath.Ext(filename)) doc, warn, err := s.upload(ctx, tenantID, title, "", "", f, filename, "") f.Close() if err != nil { if errors.Is(err, storage.ErrDuplicateContentHash) { s.logger.Info("sftp watcher: duplicate content, discarding", "path", path) } else { s.logger.Error("sftp watcher: upload pipeline failed", "path", path, "err", err) if s.audlog != nil { s.audlog.Log(audit.Entry{EventType: audit.EventDocumentCreate, Username: "sftp:tenant-" + strconv.FormatInt(tenantID, 10), TenantID: &tenantID, Success: false, Detail: err.Error()}) } return // leave the file in place for a retry on the next cycle } } else { s.logger.Info("sftp watcher: document created", "document_id", doc.ID, "path", path) if warn != "" && s.logger != nil { s.logger.Warn("sftp watcher: upload succeeded with warning", "document_id", doc.ID, "warn", warn) } } // Remove the original SFTP-dropped file: storeUploadedFile writes its own // copy into inbox//. and moves *that* into store/, so // this original drop file is no longer needed either way (processed or // confirmed duplicate). if err := os.Remove(path); err != nil { s.logger.Warn("sftp watcher: cleanup of inbox file failed", "path", path, "err", err) } }