ntfy/cmd/subscribe.go

package cmd

import (
	"errors"
	"fmt"
	"github.com/urfave/cli/v2"
	"heckel.io/ntfy/client"
	"heckel.io/ntfy/log"
	"heckel.io/ntfy/util"
	"os"
	"os/exec"
	"os/user"
	"path/filepath"
	"sort"
	"strings"
)

func init() {
	commands = append(commands, cmdSubscribe)
}

const (
	clientRootConfigFileUnixAbsolute    = "/etc/ntfy/client.yml"
	clientUserConfigFileUnixRelative    = "ntfy/client.yml"
	clientUserConfigFileWindowsRelative = "ntfy\\client.yml"
)

var flagsSubscribe = append(
	append([]cli.Flag{}, flagsDefault...),
	&cli.StringFlag{Name: "config", Aliases: []string{"c"}, Usage: "client config file"},
	&cli.StringFlag{Name: "since", Aliases: []string{"s"}, Usage: "return events since `SINCE` (Unix timestamp, or all)"},
	&cli.StringFlag{Name: "user", Aliases: []string{"u"}, EnvVars: []string{"NTFY_USER"}, Usage: "username[:password] used to auth against the server"},
	&cli.StringFlag{Name: "token", Aliases: []string{"k"}, EnvVars: []string{"NTFY_TOKEN"}, Usage: "access token used to auth against the server"},
	&cli.BoolFlag{Name: "from-config", Aliases: []string{"from_config", "C"}, Usage: "read subscriptions from config file (service mode)"},
	&cli.BoolFlag{Name: "poll", Aliases: []string{"p"}, Usage: "return events and exit, do not listen for new events"},
	&cli.BoolFlag{Name: "scheduled", Aliases: []string{"sched", "S"}, Usage: "also return scheduled/delayed events"},
)

var cmdSubscribe = &cli.Command{
	Name:      "subscribe",
	Aliases:   []string{"sub"},
	Usage:     "Subscribe to one or more topics on a ntfy server",
	UsageText: "ntfy subscribe [OPTIONS..] [TOPIC]",
	Action:    execSubscribe,
	Category:  categoryClient,
	Flags:     flagsSubscribe,
	Before:    initLogFunc,
	Description: `Subscribe to a topic from a ntfy server, and either print or execute a command for 
every arriving message. There are 3 modes in which the command can be run:

ntfy subscribe TOPIC
  This prints the JSON representation of every incoming message. It is useful when you
  have a command that wants to stream-read incoming JSON messages. Unless --poll is passed,
  this command stays open forever. 

  Examples:
    ntfy subscribe mytopic            # Prints JSON for incoming messages for ntfy.sh/mytopic
    ntfy sub home.lan/backups         # Subscribe to topic on different server
    ntfy sub --poll home.lan/backups  # Just query for latest messages and exit
    ntfy sub -u phil:mypass secret    # Subscribe with username/password
  
ntfy subscribe TOPIC COMMAND
  This executes COMMAND for every incoming messages. The message fields are passed to the
  command as environment variables:

    Variable        Aliases               Description
    --------------- --------------------- -----------------------------------
    $NTFY_ID        $id                   Unique message ID
    $NTFY_TIME      $time                 Unix timestamp of the message delivery
    $NTFY_TOPIC     $topic                Topic name
    $NTFY_MESSAGE   $message, $m          Message body
    $NTFY_TITLE     $title, $t            Message title
    $NTFY_PRIORITY  $priority, $prio, $p  Message priority (1=min, 5=max)
    $NTFY_TAGS      $tags, $tag, $ta      Message tags (comma separated list)
    $NTFY_RAW       $raw                  Raw JSON message

  Examples:
    ntfy sub mytopic 'notify-send "$m"'    # Execute command for incoming messages
    ntfy sub topic1 myscript.sh            # Execute script for incoming messages

ntfy subscribe --from-config
  Service mode (used in ntfy-client.service). This reads the config file and sets up 
  subscriptions for every topic in the "subscribe:" block (see config file).

  Examples: 
    ntfy sub --from-config                           # Read topics from config file
    ntfy sub --config=myclient.yml --from-config     # Read topics from alternate config file

` + clientCommandDescriptionSuffix,
}

func execSubscribe(c *cli.Context) error {
	// Read config and options
	conf, err := loadConfig(c)
	if err != nil {
		return err
	}
	cl := client.New(conf)
	since := c.String("since")
	user := c.String("user")
	token := c.String("token")
	poll := c.Bool("poll")
	scheduled := c.Bool("scheduled")
	fromConfig := c.Bool("from-config")
	topic := c.Args().Get(0)
	command := c.Args().Get(1)

	// Checks
	if user != "" && token != "" {
		return errors.New("cannot set both --user and --token")
	} else if !topicRegex.MatchString(topic) {
		return fmt.Errorf("topic %s contains invalid characters", topic)
	}

	if !fromConfig {
		conf.Subscribe = nil // wipe if --from-config not passed
	}
	var options []client.SubscribeOption
	if since != "" {
		options = append(options, client.WithSince(since))
	}
	if token != "" {
		options = append(options, client.WithBearerAuth(token))
	} else if user != "" {
		var pass string
		parts := strings.SplitN(user, ":", 2)
		if len(parts) == 2 {
			user = parts[0]
			pass = parts[1]
		} else {
			fmt.Fprint(c.App.ErrWriter, "Enter Password: ")
			p, err := util.ReadPassword(c.App.Reader)
			if err != nil {
				return err
			}
			pass = string(p)
			fmt.Fprintf(c.App.ErrWriter, "\r%s\r", strings.Repeat(" ", 20))
		}
		options = append(options, client.WithBasicAuth(user, pass))
	} else if conf.DefaultToken != "" {
		options = append(options, client.WithBearerAuth(conf.DefaultToken))
	} else if conf.DefaultUser != "" && conf.DefaultPassword != nil {
		options = append(options, client.WithBasicAuth(conf.DefaultUser, *conf.DefaultPassword))
	}
	if scheduled {
		options = append(options, client.WithScheduled())
	}
	if topic == "" && len(conf.Subscribe) == 0 {
		return errors.New("must specify topic, type 'ntfy subscribe --help' for help")
	}

	// Execute poll or subscribe
	if poll {
		return doPoll(c, cl, conf, topic, command, options...)
	}
	return doSubscribe(c, cl, conf, topic, command, options...)
}

func doPoll(c *cli.Context, cl *client.Client, conf *client.Config, topic, command string, options ...client.SubscribeOption) error {
	for _, s := range conf.Subscribe { // may be nil
		if auth := maybeAddAuthHeader(s, conf); auth != nil {
			options = append(options, auth)
		}
		if err := doPollSingle(c, cl, s.Topic, s.Command, options...); err != nil {
			return err
		}
	}
	if topic != "" {
		if err := doPollSingle(c, cl, topic, command, options...); err != nil {
			return err
		}
	}
	return nil
}

func doPollSingle(c *cli.Context, cl *client.Client, topic, command string, options ...client.SubscribeOption) error {
	messages, err := cl.Poll(topic, options...)
	if err != nil {
		return err
	}
	for _, m := range messages {
		printMessageOrRunCommand(c, m, command)
	}
	return nil
}

func doSubscribe(c *cli.Context, cl *client.Client, conf *client.Config, topic, command string, options ...client.SubscribeOption) error {
	cmds := make(map[string]string)    // Subscription ID -> command
	for _, s := range conf.Subscribe { // May be nil
		topicOptions := append(make([]client.SubscribeOption, 0), options...)
		for filter, value := range s.If {
			topicOptions = append(topicOptions, client.WithFilter(filter, value))
		}

		if auth := maybeAddAuthHeader(s, conf); auth != nil {
			topicOptions = append(topicOptions, auth)
		}

		subscriptionID := cl.Subscribe(s.Topic, topicOptions...)
		if s.Command != "" {
			cmds[subscriptionID] = s.Command
		} else if conf.DefaultCommand != "" {
			cmds[subscriptionID] = conf.DefaultCommand
		} else {
			cmds[subscriptionID] = ""
		}
	}
	if topic != "" {
		subscriptionID := cl.Subscribe(topic, options...)
		cmds[subscriptionID] = command
	}
	for m := range cl.Messages {
		cmd, ok := cmds[m.SubscriptionID]
		if !ok {
			continue
		}
		log.Debug("%s Dispatching received message: %s", logMessagePrefix(m), m.Raw)
		printMessageOrRunCommand(c, m, cmd)
	}
	return nil
}

func maybeAddAuthHeader(s client.Subscribe, conf *client.Config) client.SubscribeOption {
	// check for subscription token then subscription user:pass
	if s.Token != "" {
		return client.WithBearerAuth(s.Token)
	}
	if s.User != "" && s.Password != nil {
		return client.WithBasicAuth(s.User, *s.Password)
	}

	// if no subscription token nor subscription user:pass, check for default token then default user:pass
	if conf.DefaultToken != "" {
		return client.WithBearerAuth(conf.DefaultToken)
	}
	if conf.DefaultUser != "" && conf.DefaultPassword != nil {
		return client.WithBasicAuth(conf.DefaultUser, *conf.DefaultPassword)
	}
	return nil
}

func printMessageOrRunCommand(c *cli.Context, m *client.Message, command string) {
	if command != "" {
		runCommand(c, command, m)
	} else {
		log.Debug("%s Printing raw message", logMessagePrefix(m))
		fmt.Fprintln(c.App.Writer, m.Raw)
	}
}

func runCommand(c *cli.Context, command string, m *client.Message) {
	if err := runCommandInternal(c, command, m); err != nil {
		log.Warn("%s Command failed: %s", logMessagePrefix(m), err.Error())
	}
}

func runCommandInternal(c *cli.Context, script string, m *client.Message) error {
	scriptFile := fmt.Sprintf("%s/ntfy-subscribe-%s.%s", os.TempDir(), util.RandomString(10), scriptExt)
	log.Debug("%s Running command '%s' via temporary script %s", logMessagePrefix(m), script, scriptFile)
	script = scriptHeader + script
	if err := os.WriteFile(scriptFile, []byte(script), 0700); err != nil {
		return err
	}
	defer os.Remove(scriptFile)
	log.Debug("%s Executing script %s", logMessagePrefix(m), scriptFile)
	cmd := exec.Command(scriptLauncher[0], append(scriptLauncher[1:], scriptFile)...)
	cmd.Stdin = c.App.Reader
	cmd.Stdout = c.App.Writer
	cmd.Stderr = c.App.ErrWriter
	cmd.Env = envVars(m)
	return cmd.Run()
}

func envVars(m *client.Message) []string {
	env := make([]string, 0)
	env = append(env, envVar(m.ID, "NTFY_ID", "id")...)
	env = append(env, envVar(m.Topic, "NTFY_TOPIC", "topic")...)
	env = append(env, envVar(fmt.Sprintf("%d", m.Time), "NTFY_TIME", "time")...)
	env = append(env, envVar(m.Message, "NTFY_MESSAGE", "message", "m")...)
	env = append(env, envVar(m.Title, "NTFY_TITLE", "title", "t")...)
	env = append(env, envVar(fmt.Sprintf("%d", m.Priority), "NTFY_PRIORITY", "priority", "prio", "p")...)
	env = append(env, envVar(strings.Join(m.Tags, ","), "NTFY_TAGS", "tags", "tag", "ta")...)
	env = append(env, envVar(m.Raw, "NTFY_RAW", "raw")...)
	sort.Strings(env)
	if log.IsTrace() {
		log.Trace("%s With environment:\n%s", logMessagePrefix(m), strings.Join(env, "\n"))
	}
	return append(os.Environ(), env...)
}

func envVar(value string, vars ...string) []string {
	env := make([]string, 0)
	for _, v := range vars {
		env = append(env, fmt.Sprintf("%s=%s", v, value))
	}
	return env
}

func loadConfig(c *cli.Context) (*client.Config, error) {
	filename := c.String("config")
	if filename != "" {
		return client.LoadConfig(filename)
	}
	configFile := defaultClientConfigFile()
	if s, _ := os.Stat(configFile); s != nil {
		return client.LoadConfig(configFile)
	}
	return client.NewConfig(), nil
}

//lint:ignore U1000 Conditionally used in different builds
func defaultClientConfigFileUnix() string {
	u, _ := user.Current()
	configFile := clientRootConfigFileUnixAbsolute
	if u.Uid != "0" {
		homeDir, _ := os.UserConfigDir()
		return filepath.Join(homeDir, clientUserConfigFileUnixRelative)
	}
	return configFile
}

//lint:ignore U1000 Conditionally used in different builds
func defaultClientConfigFileWindows() string {
	homeDir, _ := os.UserConfigDir()
	return filepath.Join(homeDir, clientUserConfigFileWindowsRelative)
}

func logMessagePrefix(m *client.Message) string {
	return fmt.Sprintf("%s/%s", util.ShortTopicURL(m.TopicURL), m.ID)
}