saltyim is the Go library and reference client and broker implementation for Salty IM it contains a command-line client (cli), a terminal user interface (tui), builtin server/broker and a Mobile / Desktop App PWA (progressive web app) https://salty.im/
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
saltyim/client.go

473 lines
12 KiB

package saltyim
import (
"bytes"
"context"
"crypto/sha256"
"encoding/json"
"errors"
"fmt"
"net/http"
"net/url"
"os"
"path"
"strings"
"time"
"git.mills.io/prologic/msgbus"
msgbus_client "git.mills.io/prologic/msgbus/client"
"github.com/keys-pub/keys"
log "github.com/sirupsen/logrus"
"go.yarn.social/lextwt"
"go.mills.io/salty"
"go.mills.io/saltyim/internal/exec"
)
const (
DefaultEnvPath = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
ServiceUser = "salty"
)
var (
ErrNoMessages = errors.New("error: no messages found")
ErrNoSender = errors.New("error: no sender configured")
ErrNotConnected = errors.New("error: client not connected")
ErrMissingIdentity = errors.New("error: missing identity")
)
type addrCache map[string]*Addr
type Message struct {
Text string
Key *keys.EdX25519PublicKey
}
// TODO: Support shell quoting and escapes?
func parseExtraEnvs(extraenvs string) map[string]string {
env := make(map[string]string)
for _, extraenv := range strings.Split(extraenvs, " ") {
tokens := strings.SplitN(extraenv, "=", 2)
switch len(tokens) {
case 1:
env[tokens[0]] = ""
case 2:
env[tokens[0]] = tokens[1]
}
}
return env
}
// PackMessage formats an outgoing message in the Message Format
// <timestamp>\t(<sender>) <message>
func PackMessage(me *Addr, msg string) []byte {
//log.Debug("pack: ", me.Formatted(), msg)
return []byte(
fmt.Sprint(
time.Now().UTC().Format(time.RFC3339), "\t",
me.Formatted(), "\t",
strings.TrimSpace(msg), "\n",
),
)
}
// PackMessageTime formats an incoming message in the Message Format using the existing timestamp
// <timestamp>\t(<sender>) <message>
func PackMessageTime(me *Addr, msg string, t *lextwt.DateTime) []byte {
//log.Debug("pack: ", me.Formatted(), msg)
return []byte(
fmt.Sprint(
t.Literal(), "\t",
me.Formatted(), "\t",
strings.TrimSpace(msg), "\n",
),
)
}
// Client is a Salty IM client that handles talking to a Salty IM Broker
// and Sedngina and Receiving messages to/from Salty IM Users.
type Client struct {
me *Addr
id *Identity
key *keys.EdX25519Key
cache addrCache
state *State
lookup Lookuper
send Sender
}
// ClientOption is a function that configures a client
type ClientOption func(cli *Client) error
// NewClient reeturns a new Salty IM client for sending and receiving
// encrypted messages to other Salty IM users as well as decrypting
// and displaying messages of the user's own inbox.
func NewClient(me *Addr, options ...ClientOption) (*Client, error) {
cli := &Client{
me: me,
cache: make(addrCache),
lookup: &DirectLookup{},
send: &DirectSend{},
}
for _, option := range options {
if err := option(cli); err != nil {
return nil, fmt.Errorf("error configuring client: %w", err)
}
}
if cli.state == nil {
log.Debugf("no state loaded, using an empty state")
cli.state = NewState()
}
if cli.id == nil || cli.id.key == nil {
return nil, ErrMissingIdentity
}
if cli.me == nil || cli.me.IsZero() {
cli.me = cli.id.addr
}
if err := cli.me.Refresh(); err != nil {
log.WithError(err).Debug("error looking up user endpoint")
}
if cli.me == nil || cli.me.IsZero() {
return nil, fmt.Errorf("unable to find your user address in %s", cli.id.Source())
}
log.Debugf("Using identity %s with public key %s", cli.id.Source(), cli.id.key)
log.Debugf("Salty Addr is %s", cli.me)
log.Debugf("Endpoint is %s", cli.me.Endpoint())
return cli, nil
}
func (cli *Client) getAddr(user string) (*Addr, error) {
addr, ok := cli.cache[user]
if ok {
return addr, nil
}
addr, err := cli.lookup.LookupAddr(user)
if err != nil {
return nil, fmt.Errorf("error: failed to lookup user %s: %w", user, err)
}
cli.cache[user] = addr
return addr, nil
}
func (cli *Client) processMessage(msg *msgbus.Message, extraenvs, prehook, posthook string) (Message, error) {
var data []byte
defer func() {
if posthook != "" {
out, err := exec.RunCmd(exec.DefaultRunCmdTimeout, cli.Env(extraenvs), posthook, bytes.NewBuffer(data))
if err != nil {
log.WithError(err).Debugf("error running post-hook %s", posthook)
}
log.Debugf("post-hook: %q", out)
}
}()
if prehook != "" {
out, err := exec.RunCmd(exec.DefaultRunCmdTimeout, cli.Env(extraenvs), prehook, bytes.NewBuffer(msg.Payload))
if err != nil {
log.WithError(err).Debugf("error running pre-hook %s", prehook)
}
log.Debugf("pre-hook: %q", out)
}
unencrypted, senderKey, err := salty.Decrypt(cli.key, msg.Payload)
if err != nil {
return Message{}, fmt.Errorf("error decrypting message: %w", err)
}
data = unencrypted[:]
return Message{Text: string(data), Key: senderKey}, nil
}
func (cli *Client) messageHandler(extraenvs, prehook, posthook string, msgs chan Message) msgbus.HandlerFunc {
return func(msg *msgbus.Message) error {
message, err := cli.processMessage(msg, extraenvs, prehook, posthook)
if err != nil {
return fmt.Errorf("error processing message: %w", err)
}
cli.state.SetIndex(msg.Topic.Name, msg.ID)
msgs <- message
return nil
}
}
func (cli *Client) Me() *Addr { return cli.me }
func (cli *Client) Key() *keys.EdX25519PublicKey { return cli.key.PublicKey() }
func (cli *Client) State() *State { return cli.state }
func (cli *Client) Env(extraenvs string) []string {
Path := DefaultEnvPath
GoPath := os.Getenv("GOPATH")
if GoPath != "" {
Path = fmt.Sprintf("%s/bin:%s", GoPath, Path)
}
env := []string{
fmt.Sprintf("PATH=%s", Path),
fmt.Sprintf("PWD=%s", os.Getenv("PWD")),
fmt.Sprintf("HOME=%s", os.Getenv("HOME")),
fmt.Sprintf("SALTY_USER=%s", cli.me.String()),
fmt.Sprintf("SALTY_IDENTITY=%s", cli.id.Source()),
}
for key, val := range parseExtraEnvs(extraenvs) {
log.Debugf("key: %q", key)
log.Debugf("val: %q", val)
val = os.ExpandEnv(val)
if val == "" {
val = os.Getenv(key)
}
if val != "" {
env = append(env, fmt.Sprintf("%s=%s", key, val))
}
}
log.Debugf("env: #%v", env)
return env
}
func (cli *Client) Outbox() *url.URL {
// use url struct copy to avoid modifying cli.me.Endpoint().Path
// https://github.com/golang/go/issues/38351
ep := *cli.me.Endpoint()
ep.Path = path.Join(
path.Dir(ep.Path),
fmt.Sprintf("%x", sha256.Sum256(cli.id.key.Private())),
)
return &ep
}
func (cli *Client) OutboxAddr(to *Addr) *Addr {
return &Addr{
User: to.User,
Domain: to.Domain,
endpoint: cli.Outbox(),
key: cli.me.key,
capabilities: cli.me.capabilities,
}
}
func (cli *Client) OutboxClient(to *Addr) *Client {
if to == nil {
to = cli.me
}
return &Client{
me: &Addr{
User: to.User,
Domain: to.Domain,
key: cli.me.key,
endpoint: cli.Outbox(),
discoveredDomain: cli.me.discoveredDomain,
avatar: cli.me.avatar,
capabilities: cli.me.capabilities,
checkedAvatar: cli.me.checkedAvatar,
},
id: &Identity{
addr: &Addr{
User: to.User,
Domain: to.Domain,
key: cli.me.key,
endpoint: cli.Outbox(),
discoveredDomain: cli.me.discoveredDomain,
avatar: cli.me.avatar,
capabilities: cli.me.capabilities,
checkedAvatar: cli.me.checkedAvatar,
},
key: cli.key,
},
key: cli.key,
cache: cli.cache,
state: cli.state,
lookup: cli.lookup,
send: cli.send,
}
}
func (cli *Client) String() string {
b := &bytes.Buffer{}
fmt.Fprintln(b, "Me: ", cli.me)
fmt.Fprintln(b, "Endpoint: ", cli.me.Endpoint())
fmt.Fprintln(b, "Outbox: ", cli.Outbox())
fmt.Fprintln(b, "Key: ", cli.key)
return b.String()
}
// SetLookup sets the internal lookup interface to use (Lookuper) for looking
// up Salty Addresses. By default the DirectLookup implementation is used.
func (cli *Client) SetLookup(lookup Lookuper) {
cli.lookup = lookup
}
// SetSend sets the internal send interface to use (Sender) for sending
// messages to endpoints. By default the DirectSend implementation is used.
func (cli *Client) SetSend(send Sender) {
cli.send = send
}
// Read reads a single message from this user's inbox
func (cli *Client) Read(extraenvs, prehook, posthook string) (Message, error) {
if cli.me.Endpoint() == nil {
return Message{}, ErrNotConnected
}
uri, inbox := SplitInbox(cli.me.Endpoint().String())
bus := msgbus_client.NewClient(uri, nil)
msg, err := bus.Pull(inbox)
if err != nil {
return Message{}, fmt.Errorf("error reading inbox: %w", err)
}
if msg == nil {
return Message{}, ErrNoMessages
}
return cli.processMessage(msg, extraenvs, prehook, posthook)
}
// Subscribe subscribers to this user's inbox for new messages
func (cli *Client) Subscribe(ctx context.Context, extraenvs, prehook, posthook string) chan Message {
if cli.me.Endpoint() == nil {
return nil
}
uri, inbox := SplitInbox(cli.me.Endpoint().String())
bus := msgbus_client.NewClient(uri, nil)
msgs := make(chan Message)
index := cli.state.GetIndex(inbox) + 1 // +1 to skip over the last seen message
log.Debugf("streaming inbox %s from %d ...", inbox, index)
s := bus.Subscribe(inbox, index, cli.messageHandler(extraenvs, prehook, posthook, msgs))
go s.Run(ctx)
log.Debugf("Connected to %s/%s", uri, inbox)
go func() {
<-ctx.Done()
close(msgs)
}()
return msgs
}
// Lookup performs a lookup for a user's address and config
// If the user has an address already cached, the cached addr
// is returned, otherwise a full lookup is done.
func (cli *Client) Lookup(user string) (*Addr, error) {
return cli.getAddr(user)
}
// Send sends an encrypted message to the specified user
func (cli *Client) Send(user, msg string) error {
if cli.me.Endpoint() == nil {
return fmt.Errorf("unable to find your endpoint for %s", cli.me.String())
}
addr, err := cli.getAddr(user)
if err != nil {
return fmt.Errorf("error looking up user %s: %w", user, err)
}
err = cli.SendToAddr(addr, msg)
if err != nil {
return err
}
return cli.OutboxClient(addr).SendToAddr(cli.OutboxAddr(addr), msg)
}
func (cli *Client) SendToAddr(addr *Addr, msg string) error {
if cli.me == nil || cli.me.IsZero() {
return ErrNoSender
}
b, err := salty.Encrypt(cli.key, PackMessage(cli.me, msg), []string{addr.key.ID().String()})
if err != nil {
return fmt.Errorf("error encrypting message to %s: %w", addr, err)
}
endpoint := addr.Endpoint().String()
log.Debugf("sending message to %s", endpoint)
if err := cli.send.Send(cli.key, endpoint, string(b), addr.Cap()); err != nil {
return fmt.Errorf("error sending message to %s: %w", addr, err)
}
return nil
}
// Register sends a registration request to the service user of a Salty Broker
func (cli *Client) Register(brokerURI string) error {
if brokerURI == "" {
log.Debugf("Looking up SRV record for _salty._tcp.%s", cli.Me().Domain)
target, err := resolver.LookupSRV("salty", "tcp", cli.Me().Domain)
if err != nil {
return fmt.Errorf("unable to find broker for %s: %w", cli.Me(), err)
}
brokerURI = fmt.Sprintf("https://%s", target)
}
u, err := url.Parse(brokerURI)
if err != nil {
return fmt.Errorf("error parsing broker uri %s: %w", brokerURI, err)
}
u.Path = "/api/v1/register"
req := RegisterRequest{Hash: cli.Me().Hash(), Key: cli.Key().ID().String()}
data, err := json.Marshal(req)
if err != nil {
return fmt.Errorf("error serializing register request: %w", err)
}
signed, err := salty.Sign(cli.key, data)
if err != nil {
return fmt.Errorf("error signing register request: %w", err)
}
_, err = Request(http.MethodPost, u.String(), nil, bytes.NewBuffer(signed))
if err != nil {
return fmt.Errorf("error registering address: %w", err)
}
return nil
}
// SetAvatar creates or updates an avatar for a user with a broker
func (cli *Client) SetAvatar(content []byte) error {
// TODO: Verify cli.Me().Domain has valid SRV records
req := AvatarRequest{Addr: cli.Me(), Content: content}
data, err := json.Marshal(req)
if err != nil {
return fmt.Errorf("error serializing avatar request: %w", err)
}
signed, err := salty.Sign(cli.key, data)
if err != nil {
return fmt.Errorf("error signing avatar request: %w", err)
}
// TODO: Automatically work out the URI based on SRV lookups of the user's address
u := cli.Me().Endpoint()
u.Path = "/api/v1/avatar"
_, err = Request(http.MethodPost, u.String(), nil, bytes.NewBuffer(signed))
if err != nil {
return fmt.Errorf("error updating avatar: %w", err)
}
return nil
}