path: root/rhimport/fetcher.go

//
//  rhimportd
//
//  The Radio Helsinki Rivendell Import Daemon
//
//
//  Copyright (C) 2015-2016 Christian Pointner <equinox@helsinki.at>
//
//  This file is part of rhimportd.
//
//  rhimportd is free software: you can redistribute it and/or modify
//  it under the terms of the GNU General Public License as published by
//  the Free Software Foundation, either version 3 of the License, or
//  any later version.
//
//  rhimportd is distributed in the hope that it will be useful,
//  but WITHOUT ANY WARRANTY; without even the implied warranty of
//  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//  GNU General Public License for more details.
//
//  You should have received a copy of the GNU General Public License
//  along with rhimportd. If not, see <http://www.gnu.org/licenses/>.
//

package rhimport

import (
	"fmt"
	"io/ioutil"
	"mime"
	"net/http"
	"net/url"
	"os"
	"os/user"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/andelf/go-curl"
)

type FetcherCurlCBData struct {
	ctx        *Context
	res        *Result
	basepath   string
	filename   string
	remotename string
	file       *os.File
}

func (self *FetcherCurlCBData) Cleanup() {
	if self.file != nil {
		self.file.Close()
	}
}

func curlHeaderCallback(ptr []byte, userdata interface{}) bool {
	hdr := fmt.Sprintf("%s", ptr)
	data := userdata.(*FetcherCurlCBData)

	if strings.HasPrefix(hdr, "Content-Disposition:") {
		if mediatype, params, err := mime.ParseMediaType(strings.TrimPrefix(hdr, "Content-Disposition:")); err == nil {
			if mediatype == "attachment" {
				data.filename = data.basepath + "/" + params["filename"]
			}
		}
	}
	return true
}

func curlWriteCallback(ptr []byte, userdata interface{}) bool {
	data := userdata.(*FetcherCurlCBData)
	if data.file == nil {
		if data.filename == "" {
			data.filename = data.basepath + "/" + data.remotename
		}
		fp, err := os.OpenFile(data.filename, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
		if err != nil {
			rhl.Printf("Unable to create file %s: %s", data.filename, err)
			return false
		}
		data.file = fp
	}
	if _, err := data.file.Write(ptr); err != nil {
		rhl.Printf("Unable to write file %s: %s", data.filename, err)
		return false
	}
	return true
}

func curlProgressCallback(dltotal, dlnow, ultotal, ulnow float64, userdata interface{}) bool {
	data := userdata.(*FetcherCurlCBData)

	if data.ctx.Cancel != nil && len(data.ctx.Cancel) > 0 {
		rhl.Printf("downloading '%s' got canceled", data.ctx.SourceUri)
		data.res.ResponseCode = http.StatusNoContent
		data.res.ErrorString = "canceled"
		return false
	}

	if data.ctx.ProgressCallBack != nil {
		if keep := data.ctx.ProgressCallBack(1, "downloading", dlnow/dltotal, data.ctx.ProgressCallBackData); !keep {
			data.ctx.ProgressCallBack = nil
		}
	}
	return true
}

func fetchFileCurl(ctx *Context, res *Result, uri *url.URL) (err error) {
	rhl.Printf("curl-based fetcher called for '%s'", ctx.SourceUri)

	easy := curl.EasyInit()
	if easy != nil {
		defer easy.Cleanup()

		easy.Setopt(curl.OPT_FOLLOWLOCATION, true)
		easy.Setopt(curl.OPT_URL, ctx.SourceUri)

		cbdata := &FetcherCurlCBData{ctx: ctx, res: res, remotename: path.Base(uri.Path)}
		defer cbdata.Cleanup()
		if cbdata.basepath, err = ioutil.TempDir(ctx.conf.TempDir, "rhimportd-"); err != nil {
			return
		}

		easy.Setopt(curl.OPT_HEADERFUNCTION, curlHeaderCallback)
		easy.Setopt(curl.OPT_HEADERDATA, cbdata)

		easy.Setopt(curl.OPT_WRITEFUNCTION, curlWriteCallback)
		easy.Setopt(curl.OPT_WRITEDATA, cbdata)

		easy.Setopt(curl.OPT_NOPROGRESS, false)
		easy.Setopt(curl.OPT_PROGRESSFUNCTION, curlProgressCallback)
		easy.Setopt(curl.OPT_PROGRESSDATA, cbdata)

		if err = easy.Perform(); err != nil {
			if cbdata.file != nil {
				rhdl.Printf("Removing stale file: %s", cbdata.filename)
				os.Remove(cbdata.filename)
				os.Remove(path.Dir(cbdata.filename))
			}
			if res.ResponseCode == http.StatusNoContent {
				err = nil
			} else {
				err = fmt.Errorf("curl-fetcher('%s'): %s", ctx.SourceUri, err)
			}
			return
		}

		ctx.SourceFile = cbdata.filename
		if ctx.SourceFilePolicy == Auto {
			ctx.DeleteSourceFile = true
			ctx.DeleteSourceDir = true
		}
	} else {
		err = fmt.Errorf("Error initializing libcurl")
	}

	return
}

var months = [...]string{
	"Jänner",
	"Februar",
	"März",
	"April",
	"Mai",
	"Juni",
	"Juli",
	"August",
	"September",
	"Oktober",
	"November",
	"Dezember",
}

var weekdays = [...]string{
	"Sonntag",
	"Montag",
	"Dienstag",
	"Mittwoch",
	"Donnerstag",
	"Freitag",
	"Samstag",
}

func generateArchivFilePath(uri *url.URL) (file, path string, err error) {
	var t time.Time

	if t, err = time.Parse("2006/01/02/15", fmt.Sprintf("%s%s", uri.Host, uri.Path)); err != nil {
		return
	}

	file = t.Format("2006-01-02-15") + "00.ogg"
	// TODO: make basepath configurable
	path = fmt.Sprintf("/srv/archiv/%04d/%02d-%s/%02d-%s", t.Year(), t.Month(), months[t.Month()-1], t.Day(), weekdays[t.Weekday()])

	return
}

func fetchFileArchiv(ctx *Context, res *Result, uri *url.URL) (err error) {
	rhdl.Printf("archiv fetcher called for '%s'", ctx.SourceUri)

	var srcfile, srcpath string
	if srcfile, srcpath, err = generateArchivFilePath(uri); err != nil {
		res.ResponseCode = http.StatusBadRequest
		res.ErrorString = fmt.Sprintf("date/time is invalid: %s", err)
		return nil
	}

	easy := curl.EasyInit()
	if easy != nil {
		defer easy.Cleanup()

		// TODO: make user and host configurable
		scpuri := fmt.Sprintf("sftp://archiv.helsinki.at%s/%s", srcpath, srcfile)
		easy.Setopt(curl.OPT_URL, scpuri)
		easy.Setopt(curl.OPT_USERNAME, "rhimport")

		u, _ := user.Current()
		easy.Setopt(curl.OPT_SSH_PUBLIC_KEYFILE, fmt.Sprintf("%s/.ssh/id_rsa.pub", u.HomeDir))
		easy.Setopt(curl.OPT_SSH_PRIVATE_KEYFILE, fmt.Sprintf("%s/.ssh/id_rsa", u.HomeDir))

		cbdata := &FetcherCurlCBData{ctx: ctx, res: res}
		defer cbdata.Cleanup()
		var destpath string
		if destpath, err = ioutil.TempDir(ctx.conf.TempDir, "rhimportd-"); err != nil {
			return
		}
		cbdata.filename = fmt.Sprintf("%s/%s", destpath, srcfile)

		easy.Setopt(curl.OPT_WRITEFUNCTION, curlWriteCallback)
		easy.Setopt(curl.OPT_WRITEDATA, cbdata)

		easy.Setopt(curl.OPT_NOPROGRESS, false)
		easy.Setopt(curl.OPT_PROGRESSFUNCTION, curlProgressCallback)
		easy.Setopt(curl.OPT_PROGRESSDATA, cbdata)

		rhdl.Printf("importing archiv file from %s", scpuri)
		if err = easy.Perform(); err != nil {
			if cbdata.file != nil {
				rhdl.Printf("Removing stale file: %s", cbdata.filename)
				os.Remove(cbdata.filename)
				os.Remove(path.Dir(cbdata.filename))
			}
			if res.ResponseCode == http.StatusNoContent {
				err = nil
			} else {
				err = fmt.Errorf("archiv fetcher('%s'): %s", ctx.SourceUri, err)
			}
			return
		}

		ctx.SourceFile = cbdata.filename
		if ctx.SourceFilePolicy == Auto {
			ctx.DeleteSourceFile = true
			ctx.DeleteSourceDir = true
		}
	} else {
		err = fmt.Errorf("Error initializing libcurl")
	}

	return nil
}

func fetchFileLocal(ctx *Context, res *Result, uri *url.URL) (err error) {
	rhl.Printf("Local fetcher called for '%s'", ctx.SourceUri)
	if ctx.ProgressCallBack != nil {
		if keep := ctx.ProgressCallBack(1, "fetching", 0.0, ctx.ProgressCallBackData); !keep {
			ctx.ProgressCallBack = nil
		}
	}

	ctx.SourceFile = filepath.Join(ctx.conf.LocalFetchDir, path.Clean("/"+uri.Path))
	var src *os.File
	if src, err = os.Open(ctx.SourceFile); err != nil {
		res.ResponseCode = http.StatusBadRequest
		res.ErrorString = fmt.Sprintf("local-file open(): %s", err)
		return nil
	}
	if info, err := src.Stat(); err != nil {
		res.ResponseCode = http.StatusBadRequest
		res.ErrorString = fmt.Sprintf("local-file stat(): %s", err)
		return nil
	} else {
		if info.IsDir() {
			res.ResponseCode = http.StatusBadRequest
			res.ErrorString = fmt.Sprintf("'%s' is a directory", ctx.SourceFile)
			return nil
		}
	}
	src.Close()
	if ctx.ProgressCallBack != nil {
		if keep := ctx.ProgressCallBack(1, "fetching", 1.0, ctx.ProgressCallBackData); !keep {
			ctx.ProgressCallBack = nil
		}
	}
	if ctx.SourceFilePolicy == Auto {
		ctx.DeleteSourceFile = false
		ctx.DeleteSourceDir = false
	}
	return
}

func fetchFileFake(ctx *Context, res *Result, uri *url.URL) error {
	rhdl.Printf("Fake fetcher for '%s'", ctx.SourceUri)

	if duration, err := strconv.ParseUint(uri.Host, 10, 32); err != nil {
		err = nil
		res.ResponseCode = http.StatusBadRequest
		res.ErrorString = "invalid duration (must be a positive integer)"
	} else {
		for i := uint(0); i < uint(duration); i++ {
			if ctx.Cancel != nil && len(ctx.Cancel) > 0 {
				rhl.Printf("faking got canceled")
				res.ResponseCode = http.StatusNoContent
				res.ErrorString = "canceled"
				return nil
			}
			if ctx.ProgressCallBack != nil {
				if keep := ctx.ProgressCallBack(1, "faking", float64(i)/float64(duration), ctx.ProgressCallBackData); !keep {
					ctx.ProgressCallBack = nil
				}
			}
			time.Sleep(100 * time.Millisecond)
		}
		if ctx.ProgressCallBack != nil {
			if keep := ctx.ProgressCallBack(1, "faking", 1.0, ctx.ProgressCallBackData); !keep {
				ctx.ProgressCallBack = nil
			}
		}
		ctx.SourceFile = "/nonexistend/fake.mp3"
		if ctx.SourceFilePolicy == Auto {
			ctx.DeleteSourceFile = false
			ctx.DeleteSourceDir = false
		}
	}
	return nil
}

type FetchFunc func(*Context, *Result, *url.URL) (err error)

// TODO: implement fetchers for:
//         public://
//         home:// ?????
var (
	fetchers = map[string]FetchFunc{
		"local": fetchFileLocal,
		"fake":  fetchFileFake,
	}
	curlProtos = map[string]bool{
		"http": false, "https": false,
		"ftp": false, "ftps": false,
	}
)

func fetcherInit() {
	archiveEnabled := false
	info := curl.VersionInfo(curl.VERSION_FIRST)
	protos := info.Protocols
	for _, proto := range protos {
		if proto == "sftp" {
			rhdl.Printf("curl: enabling protocol %s", proto)
			fetchers["archiv"] = fetchFileArchiv
			archiveEnabled = true
		} else if _, ok := curlProtos[proto]; ok {
			rhdl.Printf("curl: enabling protocol %s", proto)
			fetchers[proto] = fetchFileCurl
			curlProtos[proto] = true
		} else {
			rhdl.Printf("curl: ignoring protocol %s", proto)
		}
	}
	for proto, enabled := range curlProtos {
		if !enabled {
			rhl.Printf("curl: protocol %s is disabled because the installed library version doesn't support it!", proto)
		}
	}
	if !archiveEnabled {
		rhl.Printf("archiv: fetcher is disabled because the installed curl library version doesn't support sFTP!")
	}
}

func checkPassword(ctx *Context, res *Result) (err error) {
	ok := false
	if ok, err = ctx.CheckPassword(); err != nil {
		return
	}
	if !ok {
		res.ResponseCode = http.StatusUnauthorized
		res.ErrorString = "invalid username and/or password"
	}
	return
}

func FetchFile(ctx *Context) (res *Result, err error) {
	res = &Result{ResponseCode: http.StatusOK}

	var uri *url.URL
	if uri, err = url.Parse(ctx.SourceUri); err != nil {
		res.ResponseCode = http.StatusBadRequest
		res.ErrorString = fmt.Sprintf("parsing uri: %s", err)
		return res, nil
	}

	if !ctx.Trusted {
		if err = checkPassword(ctx, res); err != nil || res.ResponseCode != http.StatusOK {
			return
		}
	}

	if fetcher, ok := fetchers[uri.Scheme]; ok {
		err = fetcher(ctx, res, uri)
	} else {
		err = fmt.Errorf("No fetcher for uri scheme '%s' found.", uri.Scheme)
	}

	switch ctx.SourceFilePolicy {
	case Keep:
		ctx.DeleteSourceFile = false
		ctx.DeleteSourceDir = false
	case DeleteWithDir:
		ctx.DeleteSourceDir = true
		fallthrough
	case Delete:
		ctx.DeleteSourceFile = true
	}

	return
}