diff --git a/README.md b/README.md index 56274cb..8e639c8 100644 --- a/README.md +++ b/README.md @@ -60,8 +60,13 @@ leaves the machine. API): it shows a code to enter on the hub's website (`/device`, OAuth 2.0 device authorization) and saves the session token it gets in `%APPDATA%\T5DE\proxy\hub.json`, encrypted for the Windows user (DPAPI). From then on the running proxy keeps a WebSocket to the hub (`internal/hub`) with the token as a bearer token, reporting which IMVU -account the client signed in as and the community products it wears. If the hub ends the session (signed out -everywhere, banned), the proxy stops connecting until `signin` is run again. +account the client signed in as, the community products it wears, and the IMVU chats it's in. If the hub ends the +session (signed out everywhere, banned), the proxy stops connecting until `signin` is run again. + +The room channel (`internal/products/room.go`): the hub tells the proxy which other T5DE users are in the same chat +and what they wear; the proxy gives the client those of their products it has packs for, as `*use`/`*remove` lines +from them, and adds them to the whole outfits they send. It also keeps community products out of the outfit the +client posts to IMVU's chat participant endpoint. ## Building diff --git a/cmd/t5de-proxy/main.go b/cmd/t5de-proxy/main.go index 2eb18b9..a597a71 100644 --- a/cmd/t5de-proxy/main.go +++ b/cmd/t5de-proxy/main.go @@ -136,6 +136,7 @@ func run(logger *log.Logger, listen, ready string, parentPID int, trace bool, pa client := hub.New(filepath.Join(data, "hub.json"), saved, logger) if prods != nil { prods.Observer = client + client.OnChat = prods.Others } go client.Run(context.Background()) } else if !errors.Is(err, hub.ErrSignedOut) { diff --git a/internal/hub/hub.go b/internal/hub/hub.go index e7bd134..3a1ed52 100644 --- a/internal/hub/hub.go +++ b/internal/hub/hub.go @@ -1,6 +1,7 @@ // Package hub connects the proxy to T5DE's hub (git.t5de.dev/t5de/t5de-hub): signing in with a code the person // enters on the hub's website (OAuth 2.0 device authorization, by the hub's Better Auth), and a live connection -// reporting which IMVU account the client is using and the community products it wears. +// reporting which IMVU account the client is using, the community products it wears, and the IMVU chats it's in; +// the hub answers with the other T5DE users in those chats and what they wear (the room channel). // // Signing in gives a session token, kept encrypted for the Windows user and sent as a bearer token. The hub extends // the session while it's used, and ends it when the person signs their computers out or is banned. @@ -25,8 +26,15 @@ import ( // Item is a community product worn. type Item struct { - ID string `json:"id"` - Name string `json:"name"` + ID string `json:"id"` + Name string `json:"name"` + Adult bool `json:"adult"` +} + +// Avatar is another T5DE user in a chat, and what they wear. +type Avatar struct { + IMVUID int64 `json:"imvu_id"` + Items []Item `json:"items"` } // Saved is the proxy's sign-in, in /hub.json. The session token is encrypted for the Windows user. @@ -162,16 +170,46 @@ type Client struct { path string log *log.Logger + // OnChat, if set, hears the other T5DE users in a chat the client is in, whenever that changes (and with none + // when the connection to the hub is lost). + OnChat func(chat string, avatars []Avatar) + mu sync.Mutex saved *Saved imvuID int64 outfit []Item + chats map[string]string // chat ID -> room instance ("" for other chats) wake chan struct{} } // New makes a client for the sign-in saved at path. func New(path string, saved *Saved, logger *log.Logger) *Client { - return &Client{path: path, saved: saved, log: logger, wake: make(chan struct{}, 1)} + return &Client{path: path, saved: saved, log: logger, chats: map[string]string{}, wake: make(chan struct{}, 1)} +} + +// Join reports the client is in an IMVU chat (room: the room instance, for a room's chat). +func (c *Client) Join(chat, room string) { + c.mu.Lock() + old, had := c.chats[chat] + c.chats[chat] = room + c.mu.Unlock() + if !had || old != room { + c.poke() + } +} + +// Leave reports the client left an IMVU chat. +func (c *Client) Leave(chat string) { + c.mu.Lock() + _, had := c.chats[chat] + delete(c.chats, chat) + c.mu.Unlock() + if had { + c.poke() + } + if c.OnChat != nil { + c.OnChat(chat, nil) + } } // Session reports the IMVU account the client signed in as. @@ -258,9 +296,23 @@ func (c *Client) connect(ctx context.Context) error { errs <- err return } - var m struct{ Type, Message string } - if json.Unmarshal(data, &m) == nil && m.Type == "error" { + var m struct { + Type, Message, Chat string + Avatars []Avatar + } + if json.Unmarshal(data, &m) != nil { + continue + } + switch m.Type { + case "error": c.log.Printf("hub: %s", m.Message) + case "chat": + c.mu.Lock() + _, in := c.chats[m.Chat] + c.mu.Unlock() + if in && c.OnChat != nil { + c.OnChat(m.Chat, m.Avatars) + } } } }() @@ -270,10 +322,30 @@ func (c *Client) connect(ctx context.Context) error { defer cancel() return conn.Write(wctx, websocket.MessageText, b) } + // Without the hub, nobody else's community products are known. + defer func() { + if c.OnChat == nil { + return + } + c.mu.Lock() + var chats []string + for chat := range c.chats { + chats = append(chats, chat) + } + c.mu.Unlock() + for _, chat := range chats { + c.OnChat(chat, nil) + } + }() sentID, sentOutfit := int64(-1), "" + sentChats := map[string]string{} for { c.mu.Lock() id, outfit := c.imvuID, append([]Item{}, c.outfit...) // a list, never null + chats := make(map[string]string, len(c.chats)) + for k, v := range c.chats { + chats[k] = v + } c.mu.Unlock() if id != sentID && id != 0 { if err := send(map[string]any{"type": "session", "imvu_id": id}); err != nil { @@ -287,6 +359,33 @@ func (c *Client) connect(ctx context.Context) error { } sentOutfit = string(b) } + for chat, room := range chats { + if r, ok := sentChats[chat]; ok && r == room { + continue + } + m := map[string]any{"type": "join", "chat": chat} + if room != "" { + m["room"] = room + } + if err := send(m); err != nil { + return err + } + if room != "" { + c.log.Printf("hub: in chat %s (room %s)", chat, room) + } else { + c.log.Printf("hub: in chat %s", chat) + } + sentChats[chat] = room + } + for chat := range sentChats { + if _, ok := chats[chat]; !ok { + if err := send(map[string]any{"type": "leave", "chat": chat}); err != nil { + return err + } + c.log.Printf("hub: left chat %s", chat) + delete(sentChats, chat) + } + } select { case <-ctx.Done(): conn.Close(websocket.StatusNormalClosure, "") diff --git a/internal/imq/messages.go b/internal/imq/messages.go index 91696e0..c7b1d6a 100644 --- a/internal/imq/messages.go +++ b/internal/imq/messages.go @@ -5,7 +5,10 @@ import "errors" // Message types the proxy reads or writes. const ( C2gSendMessage = 1304 + C2gUnsubscribe = 1308 + C2gSubscribe = 1309 G2cResult = 3102 + G2cSendMessage = 3104 ) // SendMessage is a C2gSendMessage: a message the client sends to a queue's mount. @@ -118,3 +121,84 @@ func appendBytesField(b []byte, num uint64, v []byte) []byte { b = appendVarint(appendVarint(b, num<<3|2), uint64(len(v))) return append(b, v...) } + +// ParseSubscription reads a C2gSubscribe's or C2gUnsubscribe's data: the queues (listed plainly, or with an op ID +// for a result). +func ParseSubscription(data []byte) ([]string, error) { + fs, err := fields(data) + if err != nil { + return nil, err + } + var queues []string + for _, f := range fs { + switch { + case f.num == 1 && f.isLen: + queues = append(queues, string(f.bytes)) + case f.num == 2 && f.isLen: + sub, err := fields(f.bytes) + if err != nil { + return nil, err + } + for _, sf := range sub { + if sf.num == 1 && sf.isLen { + queues = append(queues, string(sf.bytes)) + } + } + default: + return nil, errors.New("imq: unexpected field in a subscription") + } + } + return queues, nil +} + +// Delivered is a G2cSendMessage: a message the gateway delivers to the client, from a user (user_id). +type Delivered struct { + UserID string + Queue string + Mount string + Message []byte + Sequence uint64 // 0: none (the client only tracks sequences it's given) +} + +// ParseDelivered reads a G2cSendMessage's data. +func ParseDelivered(data []byte) (Delivered, error) { + fs, err := fields(data) + if err != nil { + return Delivered{}, err + } + var m Delivered + seen := 0 + for _, f := range fs { + switch { + case f.num == 1 && f.isLen: + m.UserID, seen = string(f.bytes), seen|1 + case f.num == 2 && f.isLen: + m.Queue, seen = string(f.bytes), seen|2 + case f.num == 3 && f.isLen: + m.Mount, seen = string(f.bytes), seen|4 + case f.num == 4 && f.isLen: + m.Message, seen = f.bytes, seen|8 + case f.num == 5 && !f.isLen: + m.Sequence = f.varint + default: + return Delivered{}, errors.New("imq: unexpected field in G2cSendMessage") + } + } + if seen != 15 { + return Delivered{}, errors.New("imq: G2cSendMessage is missing fields") + } + return m, nil +} + +// Marshal encodes the message's data, fields in the gateway's order. +func (m Delivered) Marshal() []byte { + var b []byte + b = appendBytesField(b, 1, []byte(m.UserID)) + b = appendBytesField(b, 2, []byte(m.Queue)) + b = appendBytesField(b, 3, []byte(m.Mount)) + b = appendBytesField(b, 4, m.Message) + if m.Sequence != 0 { + b = appendVarintField(b, 5, m.Sequence) + } + return b +} diff --git a/internal/products/intercept.go b/internal/products/intercept.go index 9fdf1d1..5cc5488 100644 --- a/internal/products/intercept.go +++ b/internal/products/intercept.go @@ -27,7 +27,12 @@ func (p *Products) Intercept(x *proxy.Exchange) error { if !strings.HasSuffix(host, ".imvu.com") { return nil } + if host == "api.imvu.com" && participantPath.MatchString(path) { + return p.interceptParticipant(x) + } switch path { + case "/api/xmlrpc/chat.php": + return p.interceptChat(x) case "/api/xmlrpc/client.php": return p.interceptXMLRPC(x) case "/api/shop/product.php": diff --git a/internal/products/products.go b/internal/products/products.go index f11b9b6..b0ba9c3 100644 --- a/internal/products/products.go +++ b/internal/products/products.go @@ -44,15 +44,20 @@ type Products struct { byID map[string]*served problems map[string]string // pack directory -> last error logged - // Observer, if set, hears which IMVU account the client signs in as and the community products it wears (the - // hub client). + // Observer, if set, hears which IMVU account the client signs in as, the community products it wears, and the + // chats it's in (the hub client). Observer Observer + + roomMu sync.Mutex + room room } -// Observer hears about the client's IMVU account and its community outfit. +// Observer hears about the client's IMVU account, its community outfit, and the chats it's in. type Observer interface { Session(imvuID int64) Outfit(imvuID int64, items []hub.Item) + Join(chat, room string) + Leave(chat string) } // served is a pack ready to serve. @@ -305,7 +310,7 @@ func (p *Products) reportOutfit(user string, locals []int64) { items := []hub.Item{} for _, local := range locals { if s := p.lookup(local); s != nil { - items = append(items, hub.Item{ID: s.pack.Manifest.ID, Name: s.pack.Manifest.Name}) + items = append(items, hub.Item{ID: s.pack.Manifest.ID, Name: s.pack.Manifest.Name, Adult: s.pack.Manifest.Rating == "adult"}) } } p.Observer.Outfit(id, items) diff --git a/internal/products/products_test.go b/internal/products/products_test.go index 79b37ea..8cfe440 100644 --- a/internal/products/products_test.go +++ b/internal/products/products_test.go @@ -67,7 +67,16 @@ func (f *fakeIMVU) RoundTrip(r *http.Request) (*http.Response, error) { f.t.Errorf("%s: bad signature", r.URL) } } + if participantPath.MatchString(r.URL.Path) { + checkSigned(false) + f.record("participant", body) + return reply(`{"status": "success"}`) + } switch r.URL.Path { + case "/api/xmlrpc/chat.php": + f.record("chat", body) + return reply(string((&xmlrpc.Response{Params: []xmlrpc.Value{xmlrpc.Struct{ + {Name: "chatId", Value: int64(555)}, {Name: "participantUrl", Value: "https://api.imvu.com/chat/chat-555/participants/user-7"}}}}).Marshal())) case "/api/base_look.php": if len(body) > 0 { checkSigned(true) diff --git a/internal/products/room.go b/internal/products/room.go new file mode 100644 index 0000000..204b369 --- /dev/null +++ b/internal/products/room.go @@ -0,0 +1,353 @@ +package products + +// The room channel: other T5DE users' community products, on their avatars in this client. +// +// IMVU never sees community products, so other people's clients get each avatar's outfit without them. The hub +// tells the proxy which T5DE users are in the same chat and what they wear (by product ID); for products the proxy +// has, it delivers "*use" and "*remove" lines from those users to the client, as if they had sent them, and adds the +// products to the whole outfits those users send (which would otherwise put IMVU's products back in their place). +// The client then asks for the products like any other, and the proxy answers for them. + +import ( + "bytes" + "encoding/json" + "net/http" + "regexp" + "slices" + "strconv" + "strings" + + "git.t5de.dev/t5de/t5de-proxy/internal/hub" + "git.t5de.dev/t5de/t5de-proxy/internal/imq" + "git.t5de.dev/t5de/t5de-proxy/internal/proxy" + "git.t5de.dev/t5de/t5de-proxy/internal/xmlrpc" +) + +// room is the room channel's state, under Products.roomMu. +type room struct { + subs map[*proxy.IMQConn]map[string]bool // chat IDs each IMQ connection is subscribed to + rooms map[string]string // chat ID -> room instance, from getOrMakeChat + applied map[string]map[int64][]int64 // chat -> user -> local IDs the client was given for them + userIDs map[int64]string // user -> the gateway's user_id for them, as last seen + missing map[string]bool // products others wear that aren't in the packs (logged once) +} + +func (r *room) init() { + if r.subs == nil { + r.subs = map[*proxy.IMQConn]map[string]bool{} + r.rooms = map[string]string{} + r.applied = map[string]map[int64][]int64{} + r.userIDs = map[int64]string{} + r.missing = map[string]bool{} + } +} + +// chatOf is the chat ID of an IMQ chat queue ("/chat/"), or "". +func chatOf(queue string) string { + id, ok := strings.CutPrefix(queue, "/chat/") + if !ok || id == "" || strings.Trim(id, "0123456789") != "" { + return "" + } + return id +} + +// inChat reports whether any connection is subscribed to chat (under roomMu). +func (r *room) inChat(chat string) bool { + for _, chats := range r.subs { + if chats[chat] { + return true + } + } + return false +} + +// Subscribed implements proxy.IMQWatcher: chats the client joins are reported to the hub. +func (p *Products) Subscribed(c *proxy.IMQConn, queues []string) { + p.roomMu.Lock() + p.room.init() + var joined []string + var rooms []string + for _, q := range queues { + chat := chatOf(q) + if chat == "" { + continue + } + if !p.room.inChat(chat) { + joined, rooms = append(joined, chat), append(rooms, p.room.rooms[chat]) + } + if p.room.subs[c] == nil { + p.room.subs[c] = map[string]bool{} + } + p.room.subs[c][chat] = true + } + p.roomMu.Unlock() + if p.Observer != nil { + for i, chat := range joined { + p.Observer.Join(chat, rooms[i]) + } + } +} + +// Unsubscribed implements proxy.IMQWatcher. +func (p *Products) Unsubscribed(c *proxy.IMQConn, queues []string) { + p.roomMu.Lock() + p.room.init() + var left []string + for _, q := range queues { + if chat := chatOf(q); chat != "" && p.room.subs[c][chat] { + delete(p.room.subs[c], chat) + if !p.room.inChat(chat) { + left = append(left, chat) + delete(p.room.applied, chat) + } + } + } + p.roomMu.Unlock() + if p.Observer != nil { + for _, chat := range left { + p.Observer.Leave(chat) + } + } +} + +// Closed implements proxy.IMQWatcher: a connection's subscriptions end with it. +func (p *Products) Closed(c *proxy.IMQConn) { + p.roomMu.Lock() + p.room.init() + var queues []string + for chat := range p.room.subs[c] { + queues = append(queues, "/chat/"+chat) + } + p.roomMu.Unlock() + p.Unsubscribed(c, queues) + p.roomMu.Lock() + delete(p.room.subs, c) + p.roomMu.Unlock() +} + +// userOf is the user ID in the gateway's user_id ("123" or ".../123"). +func userOf(userID string) int64 { + i := strings.LastIndexByte(userID, '/') + id, err := strconv.ParseInt(userID[i+1:], 10, 64) + if err != nil { + return 0 + } + return id +} + +// Incoming implements proxy.IMQWatcher: a T5DE user's whole outfit gets their community products added. +func (p *Products) Incoming(c *proxy.IMQConn, m imq.Delivered) []byte { + chat := chatOf(m.Queue) + if chat == "" || m.Mount != "messages" { + return m.Message + } + user := userOf(m.UserID) + if user == 0 { + return m.Message + } + p.roomMu.Lock() + p.room.init() + p.room.userIDs[user] = m.UserID + locals := slices.Clone(p.room.applied[chat][user]) + p.roomMu.Unlock() + if len(locals) == 0 { + return m.Message + } + msg, ok := parseChatMessage(m.Message) + if !ok { + return m.Message + } + fields := strings.Fields(msg.text) + if len(fields) < 2 || fields[0] != "*use" || !slices.Contains(fields[1:], strconv.Itoa(bodyProduct)) { + return m.Message + } + for _, id := range locals { + if s := strconv.FormatInt(id, 10); !slices.Contains(fields, s) { + fields = append(fields, s) + } + } + return msg.withText(strings.Join(fields, " ")) +} + +// Others is the hub's word on the other T5DE users in a chat and what they wear (none: forget them). The client is +// given the community products it has packs for, and told to take off the ones no longer worn. +func (p *Products) Others(chat string, avatars []hub.Avatar) { + type line struct { + user int64 + text string + } + want := map[int64][]int64{} + for _, a := range avatars { + var locals []int64 + for _, it := range a.Items { + if s := p.byProductID(it.ID); s != nil { + locals = append(locals, s.local) + continue + } + p.roomMu.Lock() + p.room.init() + first := !p.room.missing[it.ID] + p.room.missing[it.ID] = true + p.roomMu.Unlock() + if first { + p.logf("products: someone in the room wears %q (%s), which isn't in your packs", it.Name, it.ID) + } + } + slices.Sort(locals) + want[a.IMVUID] = locals + } + + p.roomMu.Lock() + p.room.init() + if avatars == nil { + delete(p.room.applied, chat) + p.roomMu.Unlock() + return + } + had := p.room.applied[chat] + var lines []line + for user, old := range had { + if _, still := want[user]; !still && len(old) > 0 { + lines = append(lines, line{user, "*remove " + joinIDs(old)}) + } + } + for user, locals := range want { + old := had[user] + var gone, added []int64 + for _, id := range old { + if !slices.Contains(locals, id) { + gone = append(gone, id) + } + } + for _, id := range locals { + if !slices.Contains(old, id) { + added = append(added, id) + } + } + if len(gone) > 0 { + lines = append(lines, line{user, "*remove " + joinIDs(gone)}) + } + if len(added) > 0 { + lines = append(lines, line{user, "*use " + joinIDs(added)}) + } + } + p.room.applied[chat] = want + var conns []*proxy.IMQConn + for c, chats := range p.room.subs { + if chats[chat] { + conns = append(conns, c) + } + } + userIDs := map[int64]string{} + for _, l := range lines { + if id, ok := p.room.userIDs[l.user]; ok { + userIDs[l.user] = id + } else { + userIDs[l.user] = strconv.FormatInt(l.user, 10) + } + } + p.roomMu.Unlock() + + chatID, _ := strconv.ParseInt(chat, 10, 64) + for _, l := range lines { + message, _ := json.Marshal(map[string]any{"userId": l.user, "chatId": chatID, "message": l.text, "to": 0}) + for _, c := range conns { + if err := c.Deliver(imq.Delivered{UserID: userIDs[l.user], Queue: "/chat/" + chat, Mount: "messages", Message: message}); err != nil { + p.logf("products: giving the client %d's community products: %v", l.user, err) + } + } + } + if len(lines) > 0 { + p.logf("products: %d community product changes for others in chat %s", len(lines), chat) + } +} + +func joinIDs(ids []int64) string { + s := make([]string, len(ids)) + for i, id := range ids { + s[i] = strconv.FormatInt(id, 10) + } + return strings.Join(s, " ") +} + +// interceptChat learns which room a chat is (chat.getOrMakeChat: asked with activity "publicroom-", +// answered with the chat ID). +func (p *Products) interceptChat(x *proxy.Exchange) error { + call, err := xmlrpc.ParseCall(x.Body) + if err != nil || call.Method != "chat.getOrMakeChat" || len(call.Params) != 1 { + return nil + } + args, ok := call.Params[0].(xmlrpc.Struct) + if !ok { + return nil + } + activity, _ := args.Get("activity") + a, _ := activity.(string) + inst, ok := strings.CutPrefix(a, "publicroom-") + if !ok || !roomInstance.MatchString(inst) { + return nil + } + x.ModifyResponse = func(resp *http.Response, body []byte) ([]byte, error) { + r, err := xmlrpc.ParseResponse(body) + if err != nil || r.Fault != nil || len(r.Params) != 1 { + return body, nil + } + res, ok := r.Params[0].(xmlrpc.Struct) + if !ok { + return body, nil + } + v, _ := res.Get("chatId") + var chat string + switch v := v.(type) { + case int64: + chat = strconv.FormatInt(v, 10) + case string: + chat = v + } + if chat != "" && strings.Trim(chat, "0123456789") == "" { + p.roomMu.Lock() + p.room.init() + p.room.rooms[chat] = inst + p.roomMu.Unlock() + } + return body, nil + } + return nil +} + +var roomInstance = regexp.MustCompile(`^[0-9]{1,12}-[0-9]{1,12}$`) + +// participantPath is where the client posts its outfit and seat to IMVU while in a chat. +var participantPath = regexp.MustCompile(`^/chat/chat-[0-9]+/participants/user-[0-9]+$`) + +// interceptParticipant keeps community products out of the outfit the client posts to IMVU's chat participant +// endpoint ({"legacy_outfit_message": "*use ...", ...}). +func (p *Products) interceptParticipant(x *proxy.Exchange) error { + if len(x.Body) == 0 || !bytes.Contains(x.Body, []byte("legacy_outfit_message")) { + return nil + } + var post map[string]json.RawMessage + if json.Unmarshal(x.Body, &post) != nil { + return nil + } + var text string + if json.Unmarshal(post["legacy_outfit_message"], &text) != nil { + return nil + } + r := stripCommand(text, IsCommunity) + if !r.changed { + return nil + } + if r.empty { + delete(post, "legacy_outfit_message") + } else { + post["legacy_outfit_message"], _ = json.Marshal(r.text) + } + body, _ := json.Marshal(post) + if !p.keys.resign(x.Request.Header, x.Body, x.URL.RawQuery, body, x.URL.RawQuery) { + return errNoKey + } + p.logf("products: kept %d community products out of the outfit posted to IMVU's chat", len(r.products)) + x.Body = body + return nil +} diff --git a/internal/products/room_test.go b/internal/products/room_test.go new file mode 100644 index 0000000..a595724 --- /dev/null +++ b/internal/products/room_test.go @@ -0,0 +1,126 @@ +package products + +import ( + "bytes" + "encoding/json" + "fmt" + "strings" + "sync" + "testing" + + "git.t5de.dev/t5de/t5de-proxy/internal/hub" + "git.t5de.dev/t5de/t5de-proxy/internal/imq" + "git.t5de.dev/t5de/t5de-proxy/internal/proxy" + "git.t5de.dev/t5de/t5de-proxy/internal/xmlrpc" +) + +// fakeObserver records what the hub client would be told. +type fakeObserver struct { + mu sync.Mutex + events []string +} + +func (f *fakeObserver) add(e string) { + f.mu.Lock() + defer f.mu.Unlock() + f.events = append(f.events, e) +} +func (f *fakeObserver) Session(int64) {} +func (f *fakeObserver) Outfit(int64, []hub.Item) {} +func (f *fakeObserver) Join(chat, room string) { f.add("join " + chat + " " + room) } +func (f *fakeObserver) Leave(chat string) { f.add("leave " + chat) } + +// delivered reads the chat lines delivered to a client: "from: text". +func delivered(t *testing.T, b *bytes.Buffer) []string { + t.Helper() + var lines []string + for b.Len() > 0 { + f, n, err := imq.Next(b.Bytes()) + if err != nil || n == 0 { + t.Fatalf("bad frames: %v", err) + } + b.Next(n) + m, err := imq.ParseDelivered(f.Data) + if err != nil || m.Queue != "/chat/555" || m.Mount != "messages" { + t.Fatalf("delivered %+v %v", m, err) + } + var msg struct { + Message string + UserID int64 `json:"userId"` + } + json.Unmarshal(m.Message, &msg) + lines = append(lines, fmt.Sprintf("%s(%d): %s", m.UserID, msg.UserID, msg.Message)) + } + return lines +} + +func TestRoomChannel(t *testing.T) { + h, _ := newHarness(t) + obs := &fakeObserver{} + h.p.Observer = obs + local := h.p.byProductID("blue-coat").local + + // Joining a room: getOrMakeChat says which chat it is, then the client subscribes to its queue. + h.do("POST", "http://chat.imvu.com/api/xmlrpc/chat.php", call("chat.getOrMakeChat", + xmlrpc.Struct{{Name: "userId", Value: int64(7)}, {Name: "activity", Value: "publicroom-12-3"}}), false, false) + var toClient bytes.Buffer + c := proxy.NewIMQConn(1, &toClient) + h.p.Subscribed(c, []string{"/user/7", "/chat/555"}) + if got := strings.Join(obs.events, "; "); got != "join 555 12-3" { + t.Fatalf("observer heard %q", got) + } + + // Someone else wears the coat (and something not in the packs): the client is given it from them. + h.p.Incoming(c, imq.Delivered{UserID: "/user/88", Queue: "/chat/555", Mount: "messages", Message: []byte(`{"message":"hi","userId":88}`)}) + h.p.Others("555", []hub.Avatar{{IMVUID: 88, Items: []hub.Item{{ID: "blue-coat", Name: "Coat"}, {ID: "elsewhere", Name: "Hat"}}}}) + if got := delivered(t, &toClient); len(got) != 1 || got[0] != fmt.Sprintf("/user/88(88): *use %d", local) { + t.Fatalf("delivered %q", got) + } + // Told again with no change: nothing new. + h.p.Others("555", []hub.Avatar{{IMVUID: 88, Items: []hub.Item{{ID: "blue-coat", Name: "Coat"}}}}) + if toClient.Len() != 0 { + t.Fatal("delivered again") + } + + // Their whole outfit (which IMVU sends without it) gets it added; other lines don't. + whole := []byte(`{"message":"*use 80 1 2","userId":88,"chatId":555,"to":0}`) + out := h.p.Incoming(c, imq.Delivered{UserID: "/user/88", Queue: "/chat/555", Mount: "messages", Message: whole}) + if !bytes.Contains(out, []byte(fmt.Sprintf(`"message":"*use 80 1 2 %d"`, local))) || !bytes.Contains(out, []byte(`"chatId":555`)) { + t.Fatalf("whole outfit %s", out) + } + part := []byte(`{"message":"*use 3","userId":88}`) + if out := h.p.Incoming(c, imq.Delivered{UserID: "/user/88", Queue: "/chat/555", Mount: "messages", Message: part}); !bytes.Equal(out, part) { + t.Fatalf("a single product changed: %s", out) + } + other := []byte(`{"message":"*use 80 1","userId":99}`) + if out := h.p.Incoming(c, imq.Delivered{UserID: "/user/99", Queue: "/chat/555", Mount: "messages", Message: other}); !bytes.Equal(out, other) { + t.Fatalf("someone else's outfit changed: %s", out) + } + + // They take it off, then leave: the client is told to take it off once. + h.p.Others("555", []hub.Avatar{{IMVUID: 88}}) + h.p.Others("555", []hub.Avatar{}) + if got := delivered(t, &toClient); len(got) != 1 || got[0] != fmt.Sprintf("/user/88(88): *remove %d", local) { + t.Fatalf("delivered %q", got) + } + + // The connection closing leaves the chat. + h.p.Closed(c) + if got := obs.events[len(obs.events)-1]; got != "leave 555" { + t.Fatalf("last event %q", got) + } +} + +func TestParticipantPostKeepsCommunityProducts(t *testing.T) { + h, _ := newHarness(t) + local := h.p.byProductID("blue-coat").local + h.do("POST", "https://secure.imvu.com/api/xmlrpc/client.php", call("test.avatarInfoForLogin2"), false, false) + body := []byte(fmt.Sprintf(`{"legacy_outfit_message": "*use 80 1 %d", "seat_number": 2}`, local)) + if status, _ := h.do("POST", "https://api.imvu.com/chat/chat-555/participants/user-7", body, true, false); status != 200 { + t.Fatalf("status %d", status) + } + got := h.imvu.asked["participant"] + if len(got) != 1 || strings.Contains(got[0], fmt.Sprint(local)) || !strings.Contains(got[0], `"legacy_outfit_message":"*use 80 1"`) || !strings.Contains(got[0], `"seat_number":2`) { + t.Fatalf("IMVU was sent %q", got) + } +} diff --git a/internal/proxy/connect.go b/internal/proxy/connect.go index 3a1c362..584bf9c 100644 --- a/internal/proxy/connect.go +++ b/internal/proxy/connect.go @@ -70,6 +70,12 @@ func (s *Server) serveConnect(w http.ResponseWriter, r *http.Request) { toClient := &lockedWriter{w: conn} down := &stream{s: s, id: id, name: "g2c"} up := &stream{s: s, id: id, name: "c2g", filter: s.IMQFilter, reply: toClient} + if w, ok := s.IMQFilter.(IMQWatcher); ok { + c := &IMQConn{ID: id, w: toClient} + down.watcher, down.conn = w, c + up.watcher, up.conn = w, c + defer w.Closed(c) + } // IMQ is a session: when either side closes, it's over, so both connections close. var once sync.Once closeBoth := func() { @@ -101,6 +107,34 @@ type IMQFilter interface { Outgoing(queue, mount string, message []byte) (out []byte, drop bool) } +// IMQWatcher is an IMQFilter that also follows each IMQ connection: the queues the client subscribes to, the +// messages the gateway delivers (which it can change), and it can deliver messages of its own. +type IMQWatcher interface { + IMQFilter + // Subscribed and Unsubscribed see the client's subscriptions on a connection. + Subscribed(c *IMQConn, queues []string) + Unsubscribed(c *IMQConn, queues []string) + // Incoming returns the message to deliver instead of m's (m.Message itself to leave it). + Incoming(c *IMQConn, m imq.Delivered) []byte + // Closed: the connection is over (and with it, its subscriptions). + Closed(c *IMQConn) +} + +// IMQConn is one IMQ connection of the client's. +type IMQConn struct { + ID int64 + w io.Writer +} + +// NewIMQConn is a connection writing frames to w (for tests of watchers). +func NewIMQConn(id int64, w io.Writer) *IMQConn { return &IMQConn{ID: id, w: w} } + +// Deliver gives the client a message as if the gateway had delivered it. +func (c *IMQConn) Deliver(m imq.Delivered) error { + _, err := c.w.Write(imq.EncodeFrame(imq.G2cSendMessage, m.Marshal())) + return err +} + // lockedWriter serializes writes, each a whole number of frames. type lockedWriter struct { mu sync.Mutex @@ -125,6 +159,8 @@ type stream struct { pending []byte filter IMQFilter // outgoing only reply io.Writer // outgoing only: to the client + watcher IMQWatcher + conn *IMQConn } func (st *stream) copy(dst io.Writer, src io.Reader) { @@ -183,6 +219,30 @@ func (st *stream) pass(dst io.Writer, data []byte) error { // filtered is the frame to send on: as is, changed by the filter, or none (answered here). func (st *stream) filtered(f imq.Frame, frame []byte) []byte { + if st.watcher != nil { + switch f.Type { + case imq.C2gSubscribe, imq.C2gUnsubscribe: + if queues, err := imq.ParseSubscription(f.Data); err == nil { + if f.Type == imq.C2gSubscribe { + st.watcher.Subscribed(st.conn, queues) + } else { + st.watcher.Unsubscribed(st.conn, queues) + } + } + return frame + case imq.G2cSendMessage: + m, err := imq.ParseDelivered(f.Data) + if err != nil { + return frame + } + out := st.watcher.Incoming(st.conn, m) + if bytes.Equal(out, m.Message) { + return frame + } + m.Message = out + return imq.EncodeFrame(imq.G2cSendMessage, m.Marshal()) + } + } if st.filter == nil || f.Type != imq.C2gSendMessage { return frame } diff --git a/internal/proxy/proxy_test.go b/internal/proxy/proxy_test.go index 91296d4..0f9db71 100644 --- a/internal/proxy/proxy_test.go +++ b/internal/proxy/proxy_test.go @@ -474,3 +474,102 @@ func TestInterceptor(t *testing.T) { t.Fatalf("changed: %q, server got %q, %v", b, gotBody, resp.Header) } } + +// testWatcher records subscriptions, uppercases delivered messages, and delivers "welcome" on each subscription. +type testWatcher struct { + suffixFilter + mu sync.Mutex + events []string +} + +func (w *testWatcher) add(e string) { + w.mu.Lock() + defer w.mu.Unlock() + w.events = append(w.events, e) +} +func (w *testWatcher) Subscribed(c *IMQConn, queues []string) { + w.add("sub " + strings.Join(queues, ",")) + c.Deliver(imq.Delivered{UserID: "9", Queue: queues[0], Mount: "messages", Message: []byte("welcome")}) +} +func (w *testWatcher) Unsubscribed(c *IMQConn, queues []string) { + w.add("unsub " + strings.Join(queues, ",")) +} +func (w *testWatcher) Incoming(c *IMQConn, m imq.Delivered) []byte { + return bytes.ToUpper(m.Message) +} +func (w *testWatcher) Closed(c *IMQConn) { w.add("closed") } + +func TestConnectWatchesSubscriptionsAndDeliveries(t *testing.T) { + gateway := httptest.NewUnstartedServer(nil) + gateway.StartTLS() + gateway.Close() + ln, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{Certificates: gateway.TLS.Certificates}) + if err != nil { + t.Fatal(err) + } + defer ln.Close() + delivered := imq.EncodeFrame(imq.G2cSendMessage, imq.Delivered{UserID: "/user/5", Queue: "/chat/1", Mount: "messages", Message: []byte("hello"), Sequence: 3}.Marshal()) + go func() { + c, err := ln.Accept() + if err != nil { + return + } + defer c.Close() + buf := make([]byte, 4096) + c.Read(buf) // the subscription + c.Write(delivered) + c.SetReadDeadline(time.Now().Add(2 * time.Second)) + c.Read(buf) + }() + roots := gateway.Client().Transport.(*http.Transport).TLSClientConfig.RootCAs + w := &testWatcher{} + p, logs := newProxy(t, &Server{TLSConfig: &tls.Config{RootCAs: roots}, IMQFilter: w, + Dial: func(ctx context.Context, network, addr string) (net.Conn, error) { + return net.Dial(network, ln.Addr().String()) + }}) + conn, err := net.Dial("tcp", p.Listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + fmt.Fprintf(conn, "CONNECT example.com:443 HTTP/1.1\r\nProxy-authorization: %s\r\n\r\n", auth) + r := bufio.NewReader(conn) + for line, _ := r.ReadString('\n'); line != "\r\n"; line, _ = r.ReadString('\n') { + if line == "" { + t.Fatalf("no response; log %s", logs) + } + } + // queues: field 1; queues_with_results: field 2 (a Subscription, name in field 1). + sub := []byte{0x0a, 7} + sub = append(sub, "/chat/1"...) + sub = append(sub, 0x12, 9, 0x0a, 7) + sub = append(sub, "/chat/2"...) + conn.Write(imq.EncodeFrame(imq.C2gSubscribe, sub)) + + want := append( + imq.EncodeFrame(imq.G2cSendMessage, imq.Delivered{UserID: "9", Queue: "/chat/1", Mount: "messages", Message: []byte("welcome")}.Marshal()), + imq.EncodeFrame(imq.G2cSendMessage, imq.Delivered{UserID: "/user/5", Queue: "/chat/1", Mount: "messages", Message: []byte("HELLO"), Sequence: 3}.Marshal())...) + conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + got := make([]byte, len(want)) + if _, err := io.ReadFull(r, got); err != nil { + t.Fatalf("%v; log %s", err, logs) + } + if !bytes.Equal(got, want) { + t.Fatalf("client got %q, want %q", got, want) + } + conn.Close() + for i := 0; i < 200; i++ { + w.mu.Lock() + n := len(w.events) + w.mu.Unlock() + if n == 2 { + break + } + time.Sleep(10 * time.Millisecond) + } + w.mu.Lock() + defer w.mu.Unlock() + if strings.Join(w.events, "; ") != "sub /chat/1,/chat/2; closed" { + t.Fatalf("events %q", w.events) + } +}