From ed8bd64017f5b831b3da70c9dea4f54e9a2d7d84 Mon Sep 17 00:00:00 2001 From: Antonio Navarro Perez Date: Mon, 1 Mar 2021 19:04:59 +0100 Subject: [PATCH] Alpha WebDAV support (#37) --- cmd/distribyted/main.go | 50 +++++-- config/model.go | 21 ++- fs/torrent.go | 7 +- fs/torrent_test.go | 1 + fuse/handler.go | 87 ++++-------- go.mod | 3 +- go.sum | 9 +- templates/config_template.yaml | 14 +- torrent/handler.go | 82 ++++++++++++ webdav/fs.go | 238 +++++++++++++++++++++++++++++++++ webdav/handler.go | 14 ++ webdav/http.go | 14 ++ 12 files changed, 447 insertions(+), 93 deletions(-) create mode 100644 torrent/handler.go create mode 100644 webdav/fs.go create mode 100644 webdav/handler.go create mode 100644 webdav/http.go diff --git a/cmd/distribyted/main.go b/cmd/distribyted/main.go index 44c475d..3e2affa 100644 --- a/cmd/distribyted/main.go +++ b/cmd/distribyted/main.go @@ -14,6 +14,7 @@ import ( "github.com/distribyted/distribyted/http" "github.com/distribyted/distribyted/stats" "github.com/distribyted/distribyted/torrent" + "github.com/distribyted/distribyted/webdav" "github.com/sirupsen/logrus" "github.com/urfave/cli/v2" ) @@ -22,6 +23,7 @@ const ( configFlag = "config" fuseAllowOther = "fuse-allow-other" portFlag = "http-port" + webDAVPortFlag = "webdav-port" ) func main() { @@ -39,7 +41,13 @@ func main() { Name: portFlag, Value: 4444, EnvVars: []string{"DISTRIBYTED_HTTP_PORT"}, - Usage: "HTTP port for web interface", + Usage: "HTTP port for web interface.", + }, + &cli.IntFlag{ + Name: webDAVPortFlag, + Value: 36911, + EnvVars: []string{"DISTRIBYTED_WEBDAV_PORT"}, + Usage: "Port used for WebDAV interface.", }, &cli.BoolFlag{ Name: fuseAllowOther, @@ -50,7 +58,7 @@ func main() { }, Action: func(c *cli.Context) error { - err := load(c.String(configFlag), c.Int(portFlag), c.Bool(fuseAllowOther)) + err := load(c.String(configFlag), c.Int(portFlag), c.Int(webDAVPortFlag), c.Bool(fuseAllowOther)) return err }, @@ -70,7 +78,7 @@ func newCache(folder string) (*filecache.Cache, error) { return filecache.NewCache(folder) } -func load(configPath string, port int, fuseAllowOther bool) error { +func load(configPath string, port, webDAVPort int, fuseAllowOther bool) error { ch := config.NewHandler(configPath) conf, err := ch.Get() @@ -91,32 +99,37 @@ func load(configPath string, port int, fuseAllowOther bool) error { } ss := stats.NewTorrent() - mountService := fuse.NewHandler(c, ss, fuseAllowOther) + + th := torrent.NewHandler(c, ss) + + mh := fuse.NewHandler(fuseAllowOther || conf.AllowOther) sigChan := make(chan os.Signal) signal.Notify(sigChan, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) go func() { <-sigChan - tryClose(c, mountService) + tryClose(c, mh) }() ch.OnReload(func(c *config.Root, ef config.EventFunc) error { ef("unmounting filesystems") - mountService.UnmountAll() + mh.UnmountAll() + th.RemoveAll() ef(fmt.Sprintf("setting cache size to %d MB", c.MaxCacheSize)) fc.SetCapacity(c.MaxCacheSize * 1024 * 1024) for _, mp := range c.MountPoints { - ef(fmt.Sprintf("mounting %v with %d torrents...", mp.Path, len(mp.Torrents))) - if err := mountService.Mount(mp, ef); err != nil { - return fmt.Errorf("error mounting folder %v: %w", mp.Path, err) + ef(fmt.Sprintf("loading %v with %d torrents...", mp.Path, len(mp.Torrents))) + if err := th.Load(mp.Path, mp.Torrents); err != nil { + return fmt.Errorf("error loading folder %v: %w", mp.Path, err) } - ef(fmt.Sprintf("%v mounted", mp.Path)) + ef(fmt.Sprintf("%v loaded", mp.Path)) } - return nil + return mh.MountAll(th.Fileststems(), ef) + }) if err := ch.Reload(nil); err != nil { @@ -124,7 +137,20 @@ func load(configPath string, port int, fuseAllowOther bool) error { } defer func() { - tryClose(c, mountService) + tryClose(c, mh) + }() + + go func() { + if conf.WebDAV != nil { + wdth := torrent.NewHandler(c, ss) + if err := wdth.Load("::/webDAV", conf.WebDAV.Torrents); err != nil { + logrus.WithError(err).Error("error loading torrents for webDAV") + } + + if err := webdav.NewWebDAVServer(wdth.Fileststems(), webDAVPort); err != nil { + logrus.WithError(err).Error("error starting webDAV") + } + } }() err = http.New(fc, ss, ch, port) diff --git a/config/model.go b/config/model.go index 97c1bee..7db840a 100644 --- a/config/model.go +++ b/config/model.go @@ -4,18 +4,25 @@ package config type Root struct { MaxCacheSize int64 `yaml:"max-cache-size,omitempty"` MetadataFolder string `yaml:"metadata-folder-name,omitempty"` + AllowOther bool `yaml:"fuse-allow-other,omitempty"` MountPoints []*MountPoint `yaml:"mountPoints"` + WebDAV *WebDAV `yaml:"webDav"` +} + +type WebDAV struct { + Torrents []*Torrent `yaml:"torrents"` } type MountPoint struct { - AllowOther bool `yaml:"fuse-allow-other,omitempty"` - Path string `yaml:"path"` - Torrents []struct { - MagnetURI string `yaml:"magnetUri,omitempty"` - TorrentPath string `yaml:"torrentPath,omitempty"` - FolderName string `yaml:"folderName,omitempty"` - } `yaml:"torrents"` + Path string `yaml:"path"` + Torrents []*Torrent `yaml:"torrents"` +} + +type Torrent struct { + MagnetURI string `yaml:"magnetUri,omitempty"` + TorrentPath string `yaml:"torrentPath,omitempty"` + FolderName string `yaml:"folderName,omitempty"` } func AddDefaults(r *Root) *Root { diff --git a/fs/torrent.go b/fs/torrent.go index 101411d..9e2e1ec 100644 --- a/fs/torrent.go +++ b/fs/torrent.go @@ -61,8 +61,13 @@ func (d *torrentFile) IsDir() bool { } func (d *torrentFile) Close() error { - err := d.reader.Close() + var err error + if d.reader != nil { + err = d.reader.Close() + } + d.reader = nil + return err } diff --git a/fs/torrent_test.go b/fs/torrent_test.go index 8b8f849..8c02c61 100644 --- a/fs/torrent_test.go +++ b/fs/torrent_test.go @@ -44,4 +44,5 @@ func TestTorrentFilesystem(t *testing.T) { require.NoError(err) require.NotNil(f) require.Equal(f.Size(), int64(1964275)) + require.NoError(f.Close()) } diff --git a/fuse/handler.go b/fuse/handler.go index f4f63f5..c08723d 100644 --- a/fuse/handler.go +++ b/fuse/handler.go @@ -1,98 +1,59 @@ package fuse import ( - "fmt" "os" "path" "runtime" - "github.com/anacrolix/torrent" "github.com/billziss-gh/cgofuse/fuse" "github.com/distribyted/distribyted/config" "github.com/distribyted/distribyted/fs" - "github.com/distribyted/distribyted/stats" log "github.com/sirupsen/logrus" ) type Handler struct { - c *torrent.Client - s *stats.Torrent - fuseAllowOther bool hosts map[string]*fuse.FileSystemHost } -func NewHandler(c *torrent.Client, s *stats.Torrent, fuseAllowOther bool) *Handler { +func NewHandler(fuseAllowOther bool) *Handler { return &Handler{ - c: c, - s: s, fuseAllowOther: fuseAllowOther, hosts: make(map[string]*fuse.FileSystemHost), } } -func (s *Handler) Mount(mpc *config.MountPoint, ef config.EventFunc) error { - var torrents []fs.Filesystem - for _, mpcTorrent := range mpc.Torrents { - var t *torrent.Torrent - var err error - - switch { - case mpcTorrent.MagnetURI != "": - t, err = s.c.AddMagnet(mpcTorrent.MagnetURI) - break - case mpcTorrent.TorrentPath != "": - t, err = s.c.AddTorrentFromFile(mpcTorrent.TorrentPath) - break - default: - err = fmt.Errorf("no magnet URI or torrent path provided") +func (s *Handler) MountAll(fss map[string][]fs.Filesystem, ef config.EventFunc) error { + for p, fss := range fss { + folder := p + // On windows, the folder must don't exist + if runtime.GOOS == "windows" { + folder = path.Dir(folder) } - if err != nil { + if err := os.MkdirAll(folder, 0744); err != nil && !os.IsExist(err) { return err } - // only get info if name is not available - if t.Name() == "" { - ef(fmt.Sprintf("getting torrent info...: %v", t.InfoHash())) - log.WithField("hash", t.InfoHash()).Info("getting torrent info") - <-t.GotInfo() - } + host := fuse.NewFileSystemHost(NewFS(fss)) - s.s.Add(mpc.Path, t) - torrents = append(torrents, fs.NewTorrent(t)) + // TODO improve error handling here + go func() { + var config []string - ef(fmt.Sprintf("torrent %v added to mountpoint", t.Name())) - log.WithField("name", t.Name()).WithField("path", mpc.Path).Info("torrent added to mountpoint") + if s.fuseAllowOther { + config = append(config, "-o", "allow_other") + } + + ok := host.Mount(p, config) + if !ok { + log.WithField("path", p).Error("error trying to mount filesystem") + } + }() + + s.hosts[p] = host } - folder := mpc.Path - // On windows, the folder must don't exist - if runtime.GOOS == "windows" { - folder = path.Dir(folder) - } - if err := os.MkdirAll(folder, 0744); err != nil && !os.IsExist(err) { - return err - } - - host := fuse.NewFileSystemHost(NewFS(torrents)) - - // TODO improve error handling here - go func() { - var config []string - - if mpc.AllowOther || s.fuseAllowOther { - config = append(config, "-o", "allow_other") - } - - ok := host.Mount(mpc.Path, config) - if !ok { - log.WithField("path", mpc.Path).Error("error trying to mount filesystem") - } - }() - - s.hosts[mpc.Path] = host - return nil } @@ -105,6 +66,6 @@ func (s *Handler) UnmountAll() { log.WithField("path", path).Error("unmount failed") } } + s.hosts = make(map[string]*fuse.FileSystemHost) - s.s.RemoveAll() } diff --git a/go.mod b/go.mod index d0066ba..d11794c 100644 --- a/go.mod +++ b/go.mod @@ -31,8 +31,7 @@ require ( github.com/urfave/cli/v2 v2.3.0 github.com/willf/bitset v1.1.11 // indirect golang.org/x/crypto v0.0.0-20201112155050-0c6587e931a9 // indirect - golang.org/x/net v0.0.0-20201110031124-69a78807bb2b - golang.org/x/sys v0.0.0-20201113233024-12cec1faf1ba // indirect + golang.org/x/net v0.0.0-20210119194325-5f4716e94777 golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e // indirect google.golang.org/protobuf v1.25.0 // indirect gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776 diff --git a/go.sum b/go.sum index cd0f851..2b73bfe 100644 --- a/go.sum +++ b/go.sum @@ -666,8 +666,8 @@ golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81R golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= -golang.org/x/net v0.0.0-20201110031124-69a78807bb2b h1:uwuIcX0g4Yl1NC5XAz37xsr2lTtcqevgzYNVt49waME= -golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/net v0.0.0-20210119194325-5f4716e94777 h1:003p0dJM77cxMSyCPFphvZf/Y5/NXf5fzg6ufd1/Oew= +golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20181017192945-9dcd33a902f4/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20181203162652-d668ce993890/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= @@ -721,8 +721,9 @@ golang.org/x/sys v0.0.0-20200519105757-fe76b779f299/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20200724161237-0e2f3a69832c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f h1:+Nyd8tzPX9R7BWHguqsrbFdRx3WQ/1ib8I44HXV5yTA= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20201113233024-12cec1faf1ba h1:xmhUJGQGbxlod18iJGqVEp9cHIPLl7QiX2aA3to708s= -golang.org/x/sys v0.0.0-20201113233024-12cec1faf1ba/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68 h1:nxC68pudNYkKU6jWhgrqdreuFiOQWj1Fs7T3VrH4Pjw= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= diff --git a/templates/config_template.yaml b/templates/config_template.yaml index 31825d7..2c250a1 100644 --- a/templates/config_template.yaml +++ b/templates/config_template.yaml @@ -4,6 +4,9 @@ # max-cache-size: -1 #No limit max-cache-size: 1024 +# Add this flag if you want to allow other users to access this fuse mountpoint. You need to add user_allow_other flag to /etc/fuse.conf file. +# fuse-allow-other: true + # Folder where distribyted metadata will be stored. metadata-folder-name: ./distribyted-data/metadata # List of folders where torrents will be mounted as a filesystem. @@ -11,19 +14,22 @@ mountPoints: # Example mountpoint containing some multimedia files # For windows users: You can set here also a disk letter like X: or Z: - path: ./distribyted-data/mountpoints/multimedia - # Add this flag if you want to allow other users to access this fuse mountpoint. You need to add user_allow_other flag to /etc/fuse.conf file. - # fuse-allow-other: true torrents: # - torrentPath: /path/to/torrent/file.torrent - magnetUri: "magnet:?xt=urn:btih:c9e15763f722f23e98a29decdfae341b98d53056&dn=Cosmos+Laundromat&tr=udp%3A%2F%2Fexplodie.org%3A6969&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.empire-js.us%3A1337&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337&tr=wss%3A%2F%2Ftracker.btorrent.xyz&tr=wss%3A%2F%2Ftracker.fastcast.nz&tr=wss%3A%2F%2Ftracker.openwebtorrent.com&ws=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2F&xs=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2Fcosmos-laundromat.torrent" - magnetUri: "magnet:?xt=urn:btih:dd8255ecdc7ca55fb0bbf81323d87062db1f6d1c&dn=Big+Buck+Bunny&tr=udp%3A%2F%2Fexplodie.org%3A6969&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.empire-js.us%3A1337&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337&tr=wss%3A%2F%2Ftracker.btorrent.xyz&tr=wss%3A%2F%2Ftracker.fastcast.nz&tr=wss%3A%2F%2Ftracker.openwebtorrent.com&ws=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2F&xs=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2Fbig-buck-bunny.torrent" - magnetUri: "magnet:?xt=urn:btih:08ada5a7a6183aae1e09d831df6748d566095a10&dn=Sintel&tr=udp%3A%2F%2Fexplodie.org%3A6969&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.empire-js.us%3A1337&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337&tr=wss%3A%2F%2Ftracker.btorrent.xyz&tr=wss%3A%2F%2Ftracker.fastcast.nz&tr=wss%3A%2F%2Ftracker.openwebtorrent.com&ws=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2F&xs=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2Fsintel.torrent" - - magnetUri: "magnet:?xt=urn:btih:209c8226b299b308beaf2b9cd3fb49212dbd13ec&dn=Tears+of+Steel&tr=udp%3A%2F%2Fexplodie.org%3A6969&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.empire-js.us%3A1337&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337&tr=wss%3A%2F%2Ftracker.btorrent.xyz&tr=wss%3A%2F%2Ftracker.fastcast.nz&tr=wss%3A%2F%2Ftracker.openwebtorrent.com&ws=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2F&xs=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2Ftears-of-steel.torrent" - - magnetUri: "magnet:?xt=urn:btih:a88fda5954e89178c372716a6a78b8180ed4dad3&dn=The+WIRED+CD+-+Rip.+Sample.+Mash.+Share&tr=udp%3A%2F%2Fexplodie.org%3A6969&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.empire-js.us%3A1337&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337&tr=wss%3A%2F%2Ftracker.btorrent.xyz&tr=wss%3A%2F%2Ftracker.fastcast.nz&tr=wss%3A%2F%2Ftracker.openwebtorrent.com&ws=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2F&xs=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2Fwired-cd.torrent" # Example mountpoint containing some datasets, some of them compressed in zip format # - path: ./distribyted-data/mountpoints/datasets # torrents: # - magnetUri: "magnet:?xt=urn:btih:9dea07ba660a722ae1008c4c8afdd303b6f6e53b&tr=http%3A%2F%2Facademictorrents.com%2Fannounce.php&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337%2Fannounce&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969" # - magnetUri: "magnet:?xt=urn:btih:d8b3a315172c8d804528762f37fa67db14577cdb&tr=http%3A%2F%2Facademictorrents.com%2Fannounce.php&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337%2Fannounce&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969" # - magnetUri: "magnet:?xt=urn:btih:1e0a00b9c606cf87c03e676f75929463c7756fb5&tr=http%3A%2F%2Facademictorrents.com%2Fannounce.php&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337%2Fannounce&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969" + +# TODO temporal configuration for new WebDAV feature +webDav: + torrents: + # - torrentPath: /path/to/torrent/file.torrent + - magnetUri: "magnet:?xt=urn:btih:209c8226b299b308beaf2b9cd3fb49212dbd13ec&dn=Tears+of+Steel&tr=udp%3A%2F%2Fexplodie.org%3A6969&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.empire-js.us%3A1337&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337&tr=wss%3A%2F%2Ftracker.btorrent.xyz&tr=wss%3A%2F%2Ftracker.fastcast.nz&tr=wss%3A%2F%2Ftracker.openwebtorrent.com&ws=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2F&xs=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2Ftears-of-steel.torrent" + - magnetUri: "magnet:?xt=urn:btih:a88fda5954e89178c372716a6a78b8180ed4dad3&dn=The+WIRED+CD+-+Rip.+Sample.+Mash.+Share&tr=udp%3A%2F%2Fexplodie.org%3A6969&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.empire-js.us%3A1337&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337&tr=wss%3A%2F%2Ftracker.btorrent.xyz&tr=wss%3A%2F%2Ftracker.fastcast.nz&tr=wss%3A%2F%2Ftracker.openwebtorrent.com&ws=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2F&xs=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2Fwired-cd.torrent" \ No newline at end of file diff --git a/torrent/handler.go b/torrent/handler.go new file mode 100644 index 0000000..57f2209 --- /dev/null +++ b/torrent/handler.go @@ -0,0 +1,82 @@ +package torrent + +import ( + "fmt" + "sync" + + "github.com/anacrolix/torrent" + "github.com/distribyted/distribyted/config" + "github.com/distribyted/distribyted/fs" + "github.com/distribyted/distribyted/stats" + log "github.com/sirupsen/logrus" +) + +type Handler struct { + c *torrent.Client + s *stats.Torrent + + fssMu sync.Mutex + fss map[string][]fs.Filesystem +} + +func NewHandler(c *torrent.Client, s *stats.Torrent) *Handler { + return &Handler{ + c: c, + s: s, + fss: make(map[string][]fs.Filesystem), + } +} + +func (s *Handler) Load(path string, ts []*config.Torrent) error { + var torrents []fs.Filesystem + for _, mpcTorrent := range ts { + var t *torrent.Torrent + var err error + + switch { + case mpcTorrent.MagnetURI != "": + t, err = s.c.AddMagnet(mpcTorrent.MagnetURI) + break + case mpcTorrent.TorrentPath != "": + t, err = s.c.AddTorrentFromFile(mpcTorrent.TorrentPath) + break + default: + err = fmt.Errorf("no magnet URI or torrent path provided") + } + if err != nil { + return err + } + + // only get info if name is not available + if t.Name() == "" { + log.WithField("hash", t.InfoHash()).Info("getting torrent info") + <-t.GotInfo() + } + + s.s.Add(path, t) + torrents = append(torrents, fs.NewTorrent(t)) + + log.WithField("name", t.Name()).WithField("path", path).Info("torrent added to mountpoint") + } + + folder := path + + s.fssMu.Lock() + defer s.fssMu.Unlock() + s.fss[folder] = torrents + + return nil +} + +func (s *Handler) Fileststems() map[string][]fs.Filesystem { + return s.fss +} + +func (s *Handler) RemoveAll() error { + s.fssMu.Lock() + defer s.fssMu.Unlock() + + s.fss = make(map[string][]fs.Filesystem) + s.s.RemoveAll() + return nil +} diff --git a/webdav/fs.go b/webdav/fs.go new file mode 100644 index 0000000..48d7d47 --- /dev/null +++ b/webdav/fs.go @@ -0,0 +1,238 @@ +package webdav + +import ( + "context" + "io" + "os" + "path" + "sync" + "time" + + "github.com/distribyted/distribyted/fs" + "github.com/distribyted/distribyted/iio" + "golang.org/x/net/webdav" +) + +var _ webdav.FileSystem = &WebDAV{} + +type WebDAV struct { + fss []fs.Filesystem +} + +func newFS(mFss map[string][]fs.Filesystem) *WebDAV { + for _, fss := range mFss { + return &WebDAV{fss: fss} + } + + return nil +} + +func (wd *WebDAV) OpenFile(ctx context.Context, name string, flag int, perm os.FileMode) (webdav.File, error) { + p := "/" + name + // TODO handle flag and permissions + f, err := wd.lookupFile(p) + if err != nil { + return nil, err + } + + var dirContent []os.FileInfo + if f.IsDir() { + dir, err := wd.listDir(p) + if err != nil { + return nil, err + } + + dirContent = dir + } + + wdf := newFile(path.Base(p), f, dirContent) + return wdf, nil +} + +func (wd *WebDAV) Stat(ctx context.Context, name string) (os.FileInfo, error) { + p := "/" + name + f, err := wd.lookupFile(p) + if err != nil { + return nil, err + } + fi := newFileInfo(name, f.Size(), f.IsDir()) + return fi, nil +} + +func (wd *WebDAV) Mkdir(ctx context.Context, name string, perm os.FileMode) error { + return webdav.ErrNotImplemented +} + +func (wd *WebDAV) RemoveAll(ctx context.Context, name string) error { + return webdav.ErrNotImplemented +} + +func (wd *WebDAV) Rename(ctx context.Context, oldName, newName string) error { + return webdav.ErrNotImplemented +} + +func (wd *WebDAV) lookupFile(path string) (fs.File, error) { + for _, f := range wd.fss { + file, err := f.Open(path) + if err == os.ErrNotExist { + continue + } + if err != nil { + return nil, err + } + + if file != nil { + return file, nil + } + } + + return nil, os.ErrNotExist +} + +func (wd *WebDAV) listDir(path string) ([]os.FileInfo, error) { + var out []os.FileInfo + for _, ifs := range wd.fss { + files, err := ifs.ReadDir(path) + if err != nil { + return nil, err + } + + for n, f := range files { + out = append(out, newFileInfo(n, f.Size(), f.IsDir())) + } + } + + return out, nil +} + +var _ webdav.File = &webDAVFile{} + +type webDAVFile struct { + iio.Reader + + fi os.FileInfo + + mu sync.Mutex + // dirPos and pos are protected by mu. + dirPos int + pos int64 + dirContent []os.FileInfo +} + +func newFile(name string, f fs.File, dir []os.FileInfo) *webDAVFile { + return &webDAVFile{ + fi: newFileInfo(name, f.Size(), f.IsDir()), + dirContent: dir, + Reader: f, + } +} + +func (wdf *webDAVFile) Readdir(count int) ([]os.FileInfo, error) { + wdf.mu.Lock() + defer wdf.mu.Unlock() + + if !wdf.fi.IsDir() { + return nil, os.ErrInvalid + } + + old := wdf.dirPos + if old >= len(wdf.dirContent) { + // The os.File Readdir docs say that at the end of a directory, + // the error is io.EOF if count > 0 and nil if count <= 0. + if count > 0 { + return nil, io.EOF + } + return nil, nil + } + if count > 0 { + wdf.dirPos += count + if wdf.dirPos > len(wdf.dirContent) { + wdf.dirPos = len(wdf.dirContent) + } + } else { + wdf.dirPos = len(wdf.dirContent) + old = 0 + } + + return wdf.dirContent[old:wdf.dirPos], nil +} + +func (wdf *webDAVFile) Stat() (os.FileInfo, error) { + return wdf.fi, nil +} + +func (wdf *webDAVFile) Read(p []byte) (int, error) { + wdf.mu.Lock() + defer wdf.mu.Unlock() + + n, err := wdf.Reader.ReadAt(p, wdf.pos) + wdf.pos += int64(n) + + return n, err +} + +func (wdf *webDAVFile) Seek(offset int64, whence int) (int64, error) { + wdf.mu.Lock() + defer wdf.mu.Unlock() + + switch whence { + case io.SeekStart: + wdf.pos = offset + break + case io.SeekCurrent: + wdf.pos = wdf.pos + offset + break + case io.SeekEnd: + wdf.pos = wdf.fi.Size() + offset + break + } + + return wdf.pos, nil +} + +func (wdf *webDAVFile) Write(p []byte) (n int, err error) { + return 0, webdav.ErrNotImplemented +} + +type webDAVFileInfo struct { + name string + size int64 + isDir bool +} + +func newFileInfo(name string, size int64, isDir bool) *webDAVFileInfo { + return &webDAVFileInfo{ + name: name, + size: size, + isDir: isDir, + } +} + +func (wdfi *webDAVFileInfo) Name() string { + return wdfi.name +} + +func (wdfi *webDAVFileInfo) Size() int64 { + return wdfi.size +} + +func (wdfi *webDAVFileInfo) Mode() os.FileMode { + if wdfi.isDir { + return 0555 | os.ModeDir + } + + return 0555 +} + +func (wdfi *webDAVFileInfo) ModTime() time.Time { + // TODO fix it + return time.Now() +} + +func (wdfi *webDAVFileInfo) IsDir() bool { + return wdfi.isDir +} + +func (wdfi *webDAVFileInfo) Sys() interface{} { + return nil +} diff --git a/webdav/handler.go b/webdav/handler.go new file mode 100644 index 0000000..86be2c3 --- /dev/null +++ b/webdav/handler.go @@ -0,0 +1,14 @@ +package webdav + +import ( + "github.com/distribyted/distribyted/fs" + "golang.org/x/net/webdav" +) + +func newHandler(fss map[string][]fs.Filesystem) *webdav.Handler { + return &webdav.Handler{ + Prefix: "/", + FileSystem: newFS(fss), + LockSystem: webdav.NewMemLS(), + } +} diff --git a/webdav/http.go b/webdav/http.go new file mode 100644 index 0000000..b2b3f06 --- /dev/null +++ b/webdav/http.go @@ -0,0 +1,14 @@ +package webdav + +import ( + "fmt" + "net/http" + + "github.com/distribyted/distribyted/fs" + "github.com/sirupsen/logrus" +) + +func NewWebDAVServer(fss map[string][]fs.Filesystem, port int) error { + logrus.WithField("host", fmt.Sprintf("0.0.0.0:%d", port)).Info("starting webDAV server") + return http.ListenAndServe(fmt.Sprintf("0.0.0.0:%d", port), newHandler(fss)) +}