diff --git a/README.md b/README.md index d5db241..03e585a 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,16 @@ that aren't here yet (up to 15 seconds) so they can be shown and tried on. Buyin (`api/service/shop/svc_catalog_purchase.php`) adds them to the person's hub library; IMVU products in the same purchase go on to IMVU without them. +T5DE-hosted rooms (`internal/hosted`, t5de-hub's `docs/hosted-rooms.md`): rooms living on the hub, with instance IDs +`2100000000-` and chat IDs `2100000000+n`. The proxy answers the client's room calls for them (`getRoomCardInfo`, +`getOrMakeChat`, `getRoomState`, `setRoomState`, `getParticipants`, `terminateChat`), serves their IMQ queues +(`/chat/`, `inv:/room/room-`, `room_occupancy/`: subscriptions answered here through `proxy.IMQRouter`, +chat to and from the hub over its WebSocket), adds them to the room list, checks the room's products against its real +owner (`authorizeProducts2`), and answers any other request to IMVU naming one ("not found") rather than send it. + +`T5DE_INSTANCE=` (set by the client's `t5de-instances` patch for a second client run beside the first) keeps +the proxy's data in `%APPDATA%\T5DE-\proxy` and its log in `t5de-proxy-.log`. + For finding out what IMVU's APIs send, `T5DE_PROXY_DUMP=` logs the request and response bodies of requests whose path contains it (test accounts only: bodies hold personal data). In Git Bash, keep the value from starting with `/` (or set `MSYS_NO_PATHCONV=1`), or it's turned into a Windows path. diff --git a/cmd/t5de-proxy/main.go b/cmd/t5de-proxy/main.go index 6617f0c..ad9e2f3 100644 --- a/cmd/t5de-proxy/main.go +++ b/cmd/t5de-proxy/main.go @@ -17,6 +17,7 @@ import ( "crypto/rand" "encoding/hex" "encoding/json" + "errors" "flag" "fmt" "io" @@ -29,12 +30,15 @@ import ( "path/filepath" "runtime" "strings" + "sync" "time" "git.t5de.dev/t5de/t5de-proxy/internal/buildinfo" "git.t5de.dev/t5de/t5de-proxy/internal/control" "git.t5de.dev/t5de/t5de-proxy/internal/fetch" + "git.t5de.dev/t5de/t5de-proxy/internal/hosted" "git.t5de.dev/t5de/t5de-proxy/internal/hub" + "git.t5de.dev/t5de/t5de-proxy/internal/imq" "git.t5de.dev/t5de/t5de-proxy/internal/parent" "git.t5de.dev/t5de/t5de-proxy/internal/products" "git.t5de.dev/t5de/t5de-proxy/internal/proxy" @@ -55,7 +59,7 @@ func main() { trace := flag.Bool("trace", false, "log full URLs and every IMQ frame") packs := flag.String("packs", "", "community product packs: directories of packs, separated by "+string(filepath.ListSeparator)+ " (default: %APPDATA%/T5DE/packs and $T5DE_PACKS)") - data := flag.String("data", "", "where the proxy keeps its state (default: %APPDATA%/T5DE/proxy)") + data := flag.String("data", "", "where the proxy keeps its state (default: %APPDATA%/T5DE/proxy, or T5DE- for $T5DE_INSTANCE)") version := flag.Bool("version", false, "print the version and exit") flag.Parse() if *version { @@ -65,6 +69,10 @@ func main() { var out io.Writer = os.Stderr if *logFile != "" { + if instance := os.Getenv("T5DE_INSTANCE"); instance != "" { + // Another client beside the usual one (its T5DE_INSTANCE): its own log. + *logFile = strings.TrimSuffix(*logFile, ".log") + "-" + instance + ".log" + } f, err := os.Create(*logFile) if err != nil { os.Exit(1) @@ -96,6 +104,9 @@ func dataDir(flagValue string) string { return flagValue } if config, err := os.UserConfigDir(); err == nil { + if instance := os.Getenv("T5DE_INSTANCE"); instance != "" { + return filepath.Join(config, "T5DE-"+instance, "proxy") // the client's own data (its T5DE_INSTANCE) + } return filepath.Join(config, "T5DE", "proxy") } return "." @@ -138,10 +149,25 @@ func run(logger *log.Logger, listen, ready string, parentPID int, trace bool, pa handler.Interceptor, handler.IMQFilter = prods, prods logger.Printf("community products from %s", strings.Join(packs, ", ")) } + // T5DE-hosted rooms (internal/hosted), served with the hub the proxy is signed in to. + hubNow := &hubRef{} + rooms := &hosted.Rooms{Hub: hubNow, Logf: logger.Printf} + if prods != nil { + prods.Hosted = rooms + rooms.User, rooms.LocalProduct, rooms.GlobalProduct = prods.User, prods.LocalID, prods.ProductID + prods.HostedOwner = rooms.Owner + rooms.Deliver = func(c *proxy.IMQConn, m imq.Delivered) { + m.Message = prods.Incoming(c, m) + c.Deliver(m) + } + } // The hub connection, which the client's T5DE settings can sign in and out (internal/control). manager := &control.Manager{Path: filepath.Join(data, "hub.json"), DefaultHub: envOr("T5DE_HUB", defaultHub), Log: logger, OpenBrowser: openBrowser} manager.Attach = func(client *hub.Client) { + hubNow.set(client) + client.OnRoom, client.OnConnect = rooms.FromHub, rooms.Reconnected + rooms.OnJoin, rooms.OnLeave = client.Join, client.Leave fetcher := fetch.New(client, hubProducts, logger.Printf) client.OnLibrary = func() { fetcher.SyncLibrary(context.Background()) } if prods == nil { @@ -168,7 +194,7 @@ func run(logger *log.Logger, listen, ready string, parentPID int, trace bool, pa } } manager.Start() - chain := proxy.Interceptors{manager} + chain := proxy.Interceptors{manager, rooms} // hosted rooms' calls before the products' if prods != nil { chain = append(chain, prods) } @@ -186,6 +212,38 @@ func run(logger *log.Logger, listen, ready string, parentPID int, trace bool, pa return server.Serve(ln) } +// hubRef is the hub client the proxy is signed in with, for the hosted rooms (none while signed out). +type hubRef struct { + mu sync.Mutex + c *hub.Client +} + +func (h *hubRef) set(c *hub.Client) { + h.mu.Lock() + h.c = c + h.mu.Unlock() +} + +func (h *hubRef) get() *hub.Client { + h.mu.Lock() + defer h.mu.Unlock() + return h.c +} + +func (h *hubRef) Send(v any) bool { + if c := h.get(); c != nil { + return c.Send(v) + } + return false +} + +func (h *hubRef) Call(ctx context.Context, method, path string, body, out any) error { + if c := h.get(); c != nil { + return c.Call(ctx, method, path, body, out) + } + return errors.New("not signed in to T5DE's hub") +} + func writeReady(path, address, token string) error { data, err := json.Marshal(map[string]string{"address": address, "token": token}) if err != nil { diff --git a/internal/hosted/hosted.go b/internal/hosted/hosted.go new file mode 100644 index 0000000..8cca9b8 --- /dev/null +++ b/internal/hosted/hosted.go @@ -0,0 +1,827 @@ +// Package hosted runs T5DE-hosted rooms for the client (t5de-hub's docs/hosted-rooms.md): rooms that live on T5DE's +// hub, which the client joins and uses like IMVU's rooms. The proxy answers the client's room calls for them +// (XML-RPC), serves their IMQ queues (subscriptions answered here, messages to and from the hub), and adds them to +// the room list. IMVU never hears of them. +// +// A hosted room has a number n; its instance ID is "2100000000-" and its chat ID 2100000000+n. +package hosted + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "regexp" + "sort" + "strconv" + "strings" + "sync" + "time" + + "git.t5de.dev/t5de/t5de-proxy/internal/imq" + "git.t5de.dev/t5de/t5de-proxy/internal/proxy" + "git.t5de.dev/t5de/t5de-proxy/internal/xmlrpc" +) + +// Base is the first part of hosted rooms' instance IDs, and the start of their chat IDs. +const Base = 2100000000 + +// Hub is what the rooms need of the hub: its realtime connection and its API. +type Hub interface { + Send(v any) bool + Call(ctx context.Context, method, path string, body, out any) error +} + +// Room is a hosted room as the hub lists it. +type Room struct { + N int64 `json:"n"` + InstanceID string `json:"instanceId"` + ChatID int64 `json:"chatId"` + Name string `json:"name"` + Description string `json:"description"` + RoomPid int64 `json:"roomPid"` + Rating string `json:"rating"` + MaxUsers int `json:"maxUsers"` + OwnerImvuID int64 `json:"ownerImvuId"` + OwnerName string `json:"ownerName"` + Occupancy int `json:"occupancy"` +} + +// Participant is someone in a hosted room. +type Participant struct { + ImvuID int64 `json:"imvu_id"` + Seat int `json:"seat"` +} + +// Rooms serves hosted rooms to the client. +type Rooms struct { + Hub Hub + Logf func(format string, args ...any) + // User is the IMVU account the client is signed in as (0 if not yet known). + User func() int64 + // Deliver gives the client a message from someone in a room (through the products' Incoming). + Deliver func(c *proxy.IMQConn, m imq.Delivered) + // OnJoin and OnLeave report being in a room's chat (the room channel, for others' community products). + OnJoin func(chat, room string) + OnLeave func(chat string) + // LocalProduct and GlobalProduct translate scenes' community products between the hub's IDs and the client's + // local IDs (false: not a community product, or not here). + LocalProduct func(hubID string) (int64, bool) + GlobalProduct func(local int64) (string, bool) + + mu sync.Mutex + rooms map[int64]*room + waiting map[int64][]chan joinAnswer + owners map[int64]int64 // room -> its owner's IMVU account, as last looked up + last int64 // the room the client joined last +} + +type room struct { + chat int64 + instance string + participants []Participant + sequence uint64 + conns map[*proxy.IMQConn]bool // subscribed to the room's chat queue +} + +type joinAnswer struct { + Type string `json:"type"` + Room int64 `json:"room"` + Chat int64 `json:"chat"` + Participants []Participant `json:"participants"` + Sequence uint64 `json:"sequence"` + Reason string `json:"reason"` + Message string `json:"message"` +} + +var instancePattern = regexp.MustCompile(`^` + strconv.Itoa(Base) + `-([1-9][0-9]{0,8})$`) + +// RoomNumber is the hosted room an instance ID names. +func RoomNumber(instance string) (int64, bool) { + m := instancePattern.FindStringSubmatch(instance) + if m == nil { + return 0, false + } + n, err := strconv.ParseInt(m[1], 10, 64) + return n, err == nil && n < 2147483647-Base +} + +// ChatRoom is the hosted room a chat ID belongs to. +func ChatRoom(chat int64) (int64, bool) { + if chat <= Base || chat > 2147483647 { + return 0, false + } + return chat - Base, true +} + +func instanceOf(n int64) string { return fmt.Sprintf("%d-%d", Base, n) } + +// queueRoom is the hosted room a queue belongs to, and which of its queues it is: "chat", "node" (the room's +// inv:/room/room-) or "occupancy" (room_occupancy/). +func queueRoom(queue string) (int64, string, bool) { + switch { + case strings.HasPrefix(queue, "/chat/"): + chat, err := strconv.ParseInt(strings.TrimPrefix(queue, "/chat/"), 10, 64) + if err != nil { + return 0, "", false + } + n, ok := ChatRoom(chat) + return n, "chat", ok + case strings.HasPrefix(queue, "inv:/room/room-"): + n, ok := RoomNumber(strings.TrimPrefix(queue, "inv:/room/room-")) + return n, "node", ok + case strings.HasPrefix(queue, "room_occupancy/"): + n, ok := RoomNumber(strings.TrimPrefix(queue, "room_occupancy/")) + return n, "occupancy", ok + } + return 0, "", false +} + +// IsHostedQueue reports whether a queue is a hosted room's. +func IsHostedQueue(queue string) bool { + _, _, ok := queueRoom(queue) + return ok +} + +func (r *Rooms) logf(format string, args ...any) { + if r.Logf != nil { + r.Logf(format, args...) + } +} + +func (r *Rooms) user() int64 { + if r.User == nil { + return 0 + } + return r.User() +} + +// --- the hub's messages --- + +// FromHub handles a "room.*" message from the hub. +func (r *Rooms) FromHub(data []byte) { + var m struct { + joinAnswer + From int64 `json:"from"` + Mount string `json:"mount"` + Message string `json:"message"` + } + if json.Unmarshal(data, &m) != nil { + return + } + switch m.Type { + case "room.joined", "room.error": + m.joinAnswer.Message = m.Message // the outer field takes "message" + r.mu.Lock() + if m.Type == "room.joined" { + rm := r.roomLocked(m.Room) + rm.chat, rm.participants, rm.sequence = m.Chat, m.Participants, m.Sequence + } + waiting := r.waiting[m.Room] + delete(r.waiting, m.Room) + r.mu.Unlock() + for _, w := range waiting { + w <- m.joinAnswer + } + if m.Type == "room.error" { + r.logf("hosted: room %d: %s (%s)", m.Room, m.Reason, m.Message) + } + case "room.participants": + r.mu.Lock() + if rm, ok := r.rooms[m.Room]; ok { + rm.participants = m.Participants + } + r.mu.Unlock() + case "room.message": + r.mu.Lock() + rm, ok := r.rooms[m.Room] + var conns []*proxy.IMQConn + chat := Base + m.Room + if ok { + if m.Sequence > rm.sequence { + rm.sequence = m.Sequence + } + for c := range rm.conns { + conns = append(conns, c) + } + } + r.mu.Unlock() + for _, c := range conns { + d := imq.Delivered{UserID: fmt.Sprintf("/user/%d", m.From), Queue: fmt.Sprintf("/chat/%d", chat), + Mount: m.Mount, Message: []byte(m.Message), Sequence: m.Sequence} + if r.Deliver != nil { + r.Deliver(c, d) + } else { + c.Deliver(d) + } + } + case "room.closed": + r.mu.Lock() + delete(r.rooms, m.Room) + r.mu.Unlock() + if r.OnLeave != nil { + r.OnLeave(strconv.FormatInt(Base+m.Room, 10)) + } + r.logf("hosted: room %d was closed", m.Room) + } +} + +func (r *Rooms) roomLocked(n int64) *room { + if r.rooms == nil { + r.rooms = map[int64]*room{} + } + rm, ok := r.rooms[n] + if !ok { + rm = &room{chat: Base + n, instance: instanceOf(n), conns: map[*proxy.IMQConn]bool{}} + r.rooms[n] = rm + } + return rm +} + +// Reconnected joins the rooms the client is in again (the hub forgets them when the connection drops). +func (r *Rooms) Reconnected() { + r.mu.Lock() + var ns []int64 + for n := range r.rooms { + ns = append(ns, n) + } + r.mu.Unlock() + for _, n := range ns { + r.Hub.Send(map[string]any{"type": "room.join", "room": n}) + } +} + +// join asks the hub to let the client into a room, and waits for the answer. +func (r *Rooms) join(ctx context.Context, n int64) (joinAnswer, error) { + w := make(chan joinAnswer, 1) + r.mu.Lock() + if r.waiting == nil { + r.waiting = map[int64][]chan joinAnswer{} + } + r.waiting[n] = append(r.waiting[n], w) + r.mu.Unlock() + if !r.Hub.Send(map[string]any{"type": "room.join", "room": n}) { + return joinAnswer{}, fmt.Errorf("the hub isn't connected") + } + select { + case a := <-w: + return a, nil + case <-ctx.Done(): + return joinAnswer{}, fmt.Errorf("the hub didn't answer") + } +} + +func (r *Rooms) leave(n int64) { + r.mu.Lock() + _, in := r.rooms[n] + delete(r.rooms, n) + r.mu.Unlock() + if in { + r.Hub.Send(map[string]any{"type": "room.leave", "room": n}) + if r.OnLeave != nil { + r.OnLeave(strconv.FormatInt(Base+n, 10)) + } + } +} + +// --- IMQ --- + +// Route takes hosted rooms' queues out of a subscription and answers the client for them (proxy.IMQRouter). +func (r *Rooms) Route(c *proxy.IMQConn, unsubscribe bool, s imq.Subscription) imq.Subscription { + var rest imq.Subscription + for _, q := range s.Queues { + if !r.routeQueue(c, unsubscribe, q, 0, false) { + rest.Queues = append(rest.Queues, q) + } + } + for _, q := range s.WithResults { + if !r.routeQueue(c, unsubscribe, q.Queue, q.OpID, true) { + rest.WithResults = append(rest.WithResults, q) + } + } + return rest +} + +func (r *Rooms) routeQueue(c *proxy.IMQConn, unsubscribe bool, queue string, opID uint64, withResult bool) bool { + n, kind, ok := queueRoom(queue) + if !ok { + return false + } + me := fmt.Sprintf("/user/%d", r.user()) + if unsubscribe { + if kind == "chat" { + r.mu.Lock() + if rm, ok := r.rooms[n]; ok { + delete(rm.conns, c) + } + r.mu.Unlock() + } + if withResult { + c.Send(imq.G2cResult, imq.Result(opID, 0)) + } + return true + } + switch kind { + case "chat": + r.mu.Lock() + rm := r.roomLocked(n) + rm.conns[c] = true + seq := rm.sequence + r.mu.Unlock() + c.Send(imq.G2cJoinedQueue, imq.JoinedQueue(me, queue, seq)) + c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "control", imq.MessageMount, nil, seq)) + c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "messages", imq.MessageMount, nil, seq)) + c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "participants", imq.StateMount, nil, seq)) + case "node": + c.Send(imq.G2cJoinedQueue, imq.JoinedQueue(me, queue, 0)) + c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "node", imq.MessageMount, nil, 0)) + case "occupancy": + c.Send(imq.G2cJoinedQueue, imq.JoinedQueue(me, queue, 0)) + c.Send(imq.G2cCreateMount, imq.CreateMount(queue, "occupants", imq.StateMount, nil, 0)) + } + if withResult { + c.Send(imq.G2cResult, imq.Result(opID, 0)) + } + return true +} + +// Closed forgets a connection's subscriptions. +func (r *Rooms) Closed(c *proxy.IMQConn) { + r.mu.Lock() + defer r.mu.Unlock() + for _, rm := range r.rooms { + delete(rm.conns, c) + } +} + +// Send sends what the client said or did in a hosted room to the hub (the client's message has already had its +// community products taken out, as in IMVU's rooms). It reports whether the queue was a hosted room's. +func (r *Rooms) Send(queue, mount string, message []byte) bool { + n, kind, ok := queueRoom(queue) + if !ok { + return false + } + if kind != "chat" { + return true // nothing to say on the others + } + if mount == "messages" { + // Sitting down: "*seat " in the chat text. + var m struct { + Message string `json:"message"` + } + if json.Unmarshal(message, &m) == nil && strings.HasPrefix(m.Message, "*seat ") { + if seat, err := strconv.Atoi(strings.TrimSpace(strings.TrimPrefix(m.Message, "*seat "))); err == nil && seat >= 0 && seat < 100 { + r.Hub.Send(map[string]any{"type": "room.seat", "room": n, "seat": seat}) + } + } + } + r.Hub.Send(map[string]any{"type": "room.send", "room": n, "mount": mount, "message": string(message)}) + return true +} + +// --- HTTP: XML-RPC, the participant post and the room list --- + +var participantPath = regexp.MustCompile(`^/chat/chat-([0-9]+)/participants/`) + +// Intercept answers the client's calls about hosted rooms (proxy.Interceptor). It goes before the products'. +func (r *Rooms) Intercept(x *proxy.Exchange) error { + host, path := strings.ToLower(x.URL.Hostname()), x.URL.Path + if !strings.HasSuffix(host, ".imvu.com") { + return nil + } + if host == "api.imvu.com" { + if m := participantPath.FindStringSubmatch(path); m != nil { + if chat, err := strconv.ParseInt(m[1], 10, 64); err == nil { + if _, ok := ChatRoom(chat); ok { + x.Response = proxy.LocalResponse(200, "application/json", []byte(`{"status":"success"}`)) + } + } + } + if x.Response == nil { + r.keepFromIMVU(x) + } + return nil + } + switch path { + case "/api/xmlrpc/chat.php", "/api/xmlrpc/client.php": + if err := r.interceptXMLRPC(x); err != nil || x.Response != nil { + return err + } + case "/api/rooms/rooms_list_paginated.php": + r.addToRoomList(x) + return nil + case "/api/rooms/room_rating.php": + if _, ok := RoomNumber(x.URL.Query().Get("room_instance_id")); ok { + x.Response = proxy.LocalResponse(200, "application/json", []byte(`{"score":0,"result":"success"}`)) + return nil + } + } + r.keepFromIMVU(x) + return nil +} + +// keepFromIMVU answers, rather than sends, any other request to IMVU naming a hosted room (its instance ID, or the +// chat ID of a room the client is in): IMVU doesn't know them, and needn't hear of them. +func (r *Rooms) keepFromIMVU(x *proxy.Exchange) { + text := x.URL.Path + "?" + x.URL.RawQuery + " " + string(x.Body) + named := strings.Contains(text, strconv.Itoa(Base)+"-") + if !named { + r.mu.Lock() + for _, rm := range r.rooms { + if strings.Contains(text, strconv.FormatInt(rm.chat, 10)) { + named = true + break + } + } + r.mu.Unlock() + } + if !named { + return + } + r.logf("hosted: answered a request to %s%s naming a hosted room (not found)", x.URL.Host, x.URL.Path) + x.Response = proxy.LocalResponse(404, "application/json", []byte(`{"status":"failure","error":"NOT_FOUND"}`)) +} + +func xmlrpcAnswer(v xmlrpc.Value) *http.Response { + return proxy.LocalResponse(200, "text/xml", (&xmlrpc.Response{Params: []xmlrpc.Value{v}}).Marshal()) +} + +func xmlrpcFault(message string) *http.Response { + return proxy.LocalResponse(200, "text/xml", (&xmlrpc.Response{Fault: &xmlrpc.Fault{Code: 1, String: message}}).Marshal()) +} + +func member(s xmlrpc.Struct, name string) string { + v, ok := s.Get(name) + if !ok { + return "" + } + switch v := v.(type) { + case string: + return v + case int64: + return strconv.FormatInt(v, 10) + case float64: + return strconv.FormatFloat(v, 'f', -1, 64) + } + return "" +} + +func (r *Rooms) interceptXMLRPC(x *proxy.Exchange) error { + call, err := xmlrpc.ParseCall(x.Body) + if err != nil || len(call.Params) == 0 { + return nil + } + args, _ := call.Params[0].(xmlrpc.Struct) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + switch call.Method { + case "test.getRoomCardInfo", "test.getRoomState", "test.setRoomState": + n, ok := RoomNumber(member(args, "room_instance_id")) + if !ok { + return nil + } + switch call.Method { + case "test.getRoomCardInfo": + x.Response = r.roomCard(ctx, n) + case "test.getRoomState": + x.Response = r.roomState(ctx, n) + default: + x.Response = r.saveRoomState(ctx, n, args) + } + case "chat.getOrMakeChat": + n, ok := RoomNumber(strings.TrimPrefix(member(args, "activity"), "publicroom-")) + if !ok { + return nil + } + x.Response = r.joinChat(ctx, n) + case "chat.getParticipants", "chat.terminateChat": + chat, err := strconv.ParseInt(member(args, "chatId"), 10, 64) + if err != nil { + return nil + } + n, ok := ChatRoom(chat) + if !ok { + return nil + } + if call.Method == "chat.terminateChat" { + r.leave(n) + x.Response = xmlrpcAnswer(true) + return nil + } + x.Response = xmlrpcAnswer(r.others(n)) + } + return nil +} + +func (r *Rooms) getRoom(ctx context.Context, n int64) (*Room, [][3]any, string, error) { + var out struct { + Room Room `json:"room"` + Scene struct { + Contents [][3]any `json:"contents"` + Hash string `json:"hash"` + } `json:"scene"` + } + if err := r.Hub.Call(ctx, "GET", fmt.Sprintf("/api/v1/hosted-rooms/%d", n), nil, &out); err != nil { + return nil, nil, "", err + } + r.mu.Lock() + if r.owners == nil { + r.owners = map[int64]int64{} + } + r.owners[n] = out.Room.OwnerImvuID + r.mu.Unlock() + return &out.Room, out.Scene.Contents, out.Scene.Hash, nil +} + +// Owner is the IMVU account owning the hosted room the client joined last (0 if none, or not known): the client +// checks the room's products against it (products.HostedOwner). +func (r *Rooms) Owner() int64 { + r.mu.Lock() + defer r.mu.Unlock() + if _, in := r.rooms[r.last]; !in { + return 0 + } + return r.owners[r.last] +} + +func (r *Rooms) roomCard(ctx context.Context, n int64) *http.Response { + rm, _, _, err := r.getRoom(ctx, n) + if err != nil { + r.logf("hosted: room %d: %v", n, err) + return xmlrpcFault("This T5DE room isn't available.") + } + adult := rm.Rating == "adult" + return xmlrpcAnswer(xmlrpc.Struct{ + {Name: "owner", Value: plain(rm.OwnerName)}, {Name: "owner_cid", Value: rm.OwnerImvuID}, + {Name: "size", Value: int64(rm.Occupancy)}, {Name: "name", Value: plain(rm.Name)}, + {Name: "lieutenants", Value: []xmlrpc.Value{}}, {Name: "roomOwners", Value: []xmlrpc.Value{}}, + {Name: "allowApProducts", Value: adult}, {Name: "allowRoomShellChange", Value: false}, + {Name: "allowRoomShellOverwrite", Value: false}, {Name: "allowJoinNewRoom", Value: false}, + {Name: "allowLoadNewRoom", Value: false}, {Name: "maxOccupancy", Value: int64(rm.MaxUsers)}, + {Name: "description", Value: plain(rm.Description)}, {Name: "imageUrl", Value: ""}, {Name: "language", Value: ""}, + {Name: "isAP", Value: adult}, {Name: "isVIP", Value: false}, {Name: "allowGames", Value: false}, + {Name: "isFavorite", Value: false}, {Name: "showFavorite", Value: false}, {Name: "showRatings", Value: false}, + {Name: "enableFlagging", Value: false}, {Name: "private", Value: false}, + {Name: "autoBootWhenOwnerLeaves", Value: false}, {Name: "rating", Value: float64(0)}, + {Name: "permission", Value: ""}, {Name: "themedStatus", Value: ""}, {Name: "themeId", Value: ""}, + {Name: "eligibleForThemedRoomSubmission", Value: ""}, {Name: "occupants", Value: []xmlrpc.Value{}}, + }) +} + +func (r *Rooms) roomState(ctx context.Context, n int64) *http.Response { + rm, contents, hash, err := r.getRoom(ctx, n) + if err != nil { + r.logf("hosted: room %d: %v", n, err) + return xmlrpcFault("This T5DE room isn't available.") + } + var items []xmlrpc.Value + for _, it := range contents { + furni, ok1 := it[0].(float64) + placement, ok2 := it[2].(string) + if !ok1 || !ok2 { + continue + } + var product int64 + switch p := it[1].(type) { + case float64: + product = int64(p) + case string: + local, ok := int64(0), false + if r.LocalProduct != nil { + local, ok = r.LocalProduct(p) + } + if !ok { + continue // a community product that isn't here + } + product = local + default: + continue + } + items = append(items, []xmlrpc.Value{int64(furni), product, placement}) + } + if items == nil { + items = []xmlrpc.Value{} + } + return xmlrpcAnswer(xmlrpc.Struct{ + {Name: "room_state", Value: xmlrpc.Struct{ + {Name: "room_info", Value: xmlrpc.Struct{ + {Name: "customers_id", Value: strconv.FormatInt(rm.OwnerImvuID, 10)}, + {Name: "room_instance_id", Value: instanceOf(n)}, + {Name: "room_pid", Value: strconv.FormatInt(rm.RoomPid, 10)}, + }}, + {Name: "room_contents", Value: items}, + }}, + {Name: "room_hash", Value: hash}, + }) +} + +func (r *Rooms) saveRoomState(ctx context.Context, n int64, args xmlrpc.Struct) *http.Response { + v, _ := args.Get("room_contents") + list, _ := v.([]xmlrpc.Value) + contents := [][3]any{} + for _, item := range list { + it, ok := item.([]xmlrpc.Value) + if !ok || len(it) < 3 { + continue + } + furni, ok1 := asInt(it[0]) + product, ok2 := asInt(it[1]) + placement, ok3 := it[2].(string) + if !ok1 || !ok2 || !ok3 { + continue + } + var p any = product + if product >= 2000000000 { + id, ok := "", false + if r.GlobalProduct != nil { + id, ok = r.GlobalProduct(product) + } + if !ok { + continue + } + p = id + } + contents = append(contents, [3]any{furni, p, placement}) + } + if err := r.Hub.Call(ctx, "PUT", fmt.Sprintf("/api/v1/hosted-rooms/%d/scene", n), map[string]any{"contents": contents}, nil); err != nil { + r.logf("hosted: saving room %d: %v", n, err) + return xmlrpcFault("Couldn't save the room on T5DE's hub.") + } + return xmlrpcAnswer(true) +} + +func asInt(v xmlrpc.Value) (int64, bool) { + switch v := v.(type) { + case int64: + return v, true + case float64: + return int64(v), v == float64(int64(v)) + case string: + n, err := strconv.ParseInt(v, 10, 64) + return n, err == nil + } + return 0, false +} + +func (r *Rooms) joinChat(ctx context.Context, n int64) *http.Response { + a, err := r.join(ctx, n) + if err != nil { + r.logf("hosted: joining room %d: %v", n, err) + return xmlrpcFault("T5DE's hub isn't reachable right now.") + } + if a.Type != "room.joined" { + msg := a.Message + if msg == "" { + msg = "You can't join this T5DE room." + } + return xmlrpcFault(msg) + } + me := r.user() + taken := map[int]bool{} + for _, p := range a.Participants { + if p.ImvuID != me { + taken[p.Seat] = true + } + } + seat := 1 + for taken[seat] { + seat++ + } + r.mu.Lock() + r.last = n + _, known := r.owners[n] + r.mu.Unlock() + if !known { + r.getRoom(ctx, n) // its owner, for the room's products + } + if r.OnJoin != nil { + r.OnJoin(strconv.FormatInt(a.Chat, 10), instanceOf(n)) + } + r.logf("hosted: in room %d (chat %d)", n, a.Chat) + return xmlrpcAnswer(xmlrpc.Struct{ + {Name: "waiting", Value: int64(0)}, {Name: "lastMessageId", Value: int64(a.Sequence)}, + {Name: "chatId", Value: a.Chat}, {Name: "partnerId", Value: int64(0)}, {Name: "seat", Value: int64(seat)}, + {Name: "participantUrl", Value: fmt.Sprintf("https://api.imvu.com/chat/chat-%d/participants/user-%d", a.Chat, me)}, + }) +} + +// others is the room's other people, as chat.getParticipants lists them. +func (r *Rooms) others(n int64) []xmlrpc.Value { + me := r.user() + r.mu.Lock() + defer r.mu.Unlock() + out := []xmlrpc.Value{} + if rm, ok := r.rooms[n]; ok { + for _, p := range rm.participants { + if p.ImvuID != me { + out = append(out, xmlrpc.Struct{{Name: "userId", Value: p.ImvuID}, {Name: "seat", Value: int64(p.Seat)}}) + } + } + } + return out +} + +// addToRoomList adds hosted rooms to the client's room list (rooms_list_paginated.php: one array per field): all +// of them to the first page without a search, and those the client was in this run to recently visited rooms. +func (r *Rooms) addToRoomList(x *proxy.Exchange) { + q := x.URL.Query() + if p := q.Get("page"); p != "" && p != "1" && p != "0" { + return + } + if q.Get("keywords") != "" || q.Get("search") != "" { + return + } + recent := q.Get("recent") != "" + x.ModifyResponse = func(resp *http.Response, body []byte) ([]byte, error) { + var list map[string]json.RawMessage + if resp.StatusCode != http.StatusOK || json.Unmarshal(body, &list) != nil { + return body, nil + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + var out struct { + Rooms []Room `json:"rooms"` + } + if err := r.Hub.Call(ctx, "GET", "/api/v1/hosted-rooms", nil, &out); err != nil || len(out.Rooms) == 0 { + return body, nil + } + sort.SliceStable(out.Rooms, func(i, j int) bool { return out.Rooms[i].Occupancy > out.Rooms[j].Occupancy }) + r.mu.Lock() + visited := map[int64]bool{} + for n := range r.owners { + visited[n] = true + } + r.mu.Unlock() + for _, rm := range out.Rooms { + if !recent || visited[rm.N] { + appendRoom(list, rm) + } + } + return json.Marshal(list) + } +} + +// appendRoom adds a room to each of the list's arrays (fields it doesn't know get an empty value of the array's +// kind). +func appendRoom(list map[string]json.RawMessage, rm Room) { + adult := "0" + if rm.Rating == "adult" { + adult = "1" + } + known := map[string]any{ + "customers_room_id": strconv.FormatInt(rm.N, 10), "customers_id": strconv.Itoa(Base), + "room_instance_id": instanceOf(rm.N), "name": plain(rm.Name), "customers_name": plain(rm.OwnerName), + "num_participants": strconv.Itoa(rm.Occupancy), "max_users": strconv.Itoa(rm.MaxUsers), + "description": plain(rm.Description), "image_url": "", "rating": "0", "language": "", "is_ap": adult, + "is_av": "0", "is_vip": "0", "participants": []any{}, + } + for name, raw := range list { + var arr []json.RawMessage + if json.Unmarshal(raw, &arr) != nil { + continue + } + v, ok := known[name] + if !ok { + v = emptyLike(arr) + } else if str, isString := v.(string); isString && len(arr) > 0 && !strings.HasPrefix(strings.TrimSpace(string(arr[0])), "\"") { + // The list has numbers here, not strings. + if n, err := strconv.ParseInt(str, 10, 64); err == nil { + v = n + } + } + b, _ := json.Marshal(v) + arr = append(arr, b) + list[name], _ = json.Marshal(arr) + } +} + +// emptyLike is an empty value of the kind the array holds. +func emptyLike(arr []json.RawMessage) any { + if len(arr) == 0 { + return "" + } + switch b := strings.TrimSpace(string(arr[0])); { + case strings.HasPrefix(b, "["): + return []any{} + case strings.HasPrefix(b, "{"): + return map[string]any{} + case strings.HasPrefix(b, "\""): + return "" + case b == "true" || b == "false": + return false + case b == "null": + return nil + } + return 0 +} + +// plain keeps text to what the client's pages can show however they insert it (as in the products package). +func plain(s string) string { + return strings.Map(func(c rune) rune { + switch c { + case '<', '>', '&', '"', '\'', '`': + return -1 + } + return c + }, s) +} diff --git a/internal/hosted/hosted_test.go b/internal/hosted/hosted_test.go new file mode 100644 index 0000000..4a7765c --- /dev/null +++ b/internal/hosted/hosted_test.go @@ -0,0 +1,259 @@ +package hosted + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync" + "testing" + + "git.t5de.dev/t5de/t5de-proxy/internal/imq" + "git.t5de.dev/t5de/t5de-proxy/internal/proxy" + "git.t5de.dev/t5de/t5de-proxy/internal/xmlrpc" +) + +// fakeHub answers joins at once (through rooms.FromHub) and records what it's sent. +type fakeHub struct { + mu sync.Mutex + rooms *Rooms + sent []map[string]any + scene []any +} + +func (h *fakeHub) Send(v any) bool { + m := v.(map[string]any) + h.mu.Lock() + h.sent = append(h.sent, m) + h.mu.Unlock() + if m["type"] == "room.join" { + n := m["room"].(int64) + var answer string + if n == 9 { + answer = `{"type": "room.error", "room": 9, "reason": "full", "message": "This room is full."}` + } else { + answer = fmt.Sprintf(`{"type": "room.joined", "room": %d, "chat": %d, "sequence": 4, + "participants": [{"imvu_id": 7, "seat": 1}, {"imvu_id": 88, "seat": 2}]}`, n, Base+n) + } + go h.rooms.FromHub([]byte(answer)) + } + return true +} + +func (h *fakeHub) Call(ctx context.Context, method, path string, body, out any) error { + switch { + case method == "GET" && path == "/api/v1/hosted-rooms/3": + return json.Unmarshal([]byte(`{"room": {"n": 3, "name": "Lounge ", "ownerImvuId": 88, "ownerName": "Erin", + "roomPid": 50168116, "rating": "general", "maxUsers": 10, "occupancy": 2}, + "scene": {"contents": [[1, 47477426, "37207-46"], [2, "coatcoat22", "37-11"], [3, "notherenn2", "37-12"]], "hash": "h1"}}`), out) + case method == "GET" && path == "/api/v1/hosted-rooms": + return json.Unmarshal([]byte(`{"rooms": [{"n": 3, "name": "Lounge", "ownerName": "Erin", "maxUsers": 10, "occupancy": 2}]}`), out) + case method == "PUT" && path == "/api/v1/hosted-rooms/3/scene": + h.mu.Lock() + h.scene = append(h.scene, body) + h.mu.Unlock() + return nil + } + return fmt.Errorf("unexpected %s %s", method, path) +} + +func (h *fakeHub) sentOf(kind string) []map[string]any { + h.mu.Lock() + defer h.mu.Unlock() + var out []map[string]any + for _, m := range h.sent { + if m["type"] == kind { + out = append(out, m) + } + } + return out +} + +func setup() (*Rooms, *fakeHub, *[]string) { + h := &fakeHub{} + var events []string + r := &Rooms{Hub: h, User: func() int64 { return 7 }, + OnJoin: func(chat, room string) { events = append(events, "join "+chat+" "+room) }, + OnLeave: func(chat string) { events = append(events, "leave "+chat) }, + LocalProduct: func(id string) (int64, bool) { return 2000000005, id == "coatcoat22" }, + GlobalProduct: func(local int64) (string, bool) { return "coatcoat22", local == 2000000005 }} + h.rooms = r + return r, h, &events +} + +func xmlrpcCall(t *testing.T, r *Rooms, host, method string, args xmlrpc.Struct) (*xmlrpc.Response, bool) { + t.Helper() + body := (&xmlrpc.Call{Method: method, Params: []xmlrpc.Value{args}}).Marshal() + u, _ := url.Parse("https://" + host + "/api/xmlrpc/" + map[bool]string{true: "chat.php", false: "client.php"}[strings.HasPrefix(method, "chat.")]) + req, _ := http.NewRequest("POST", u.String(), bytes.NewReader(body)) + x := &proxy.Exchange{URL: u, Request: req, Body: body} + r.Intercept(x) + if x.Response == nil { + return nil, false + } + b, _ := io.ReadAll(x.Response.Body) + resp, err := xmlrpc.ParseResponse(b) + if err != nil { + t.Fatalf("%s: %v\n%s", method, err, b) + } + return resp, true +} + +func get(v xmlrpc.Value, name string) xmlrpc.Value { + s, _ := v.(xmlrpc.Struct) + out, _ := s.Get(name) + return out +} + +// frames reads the frames the proxy sent the client. +func frames(t *testing.T, b []byte) []imq.Frame { + var out []imq.Frame + for len(b) > 0 { + f, n, err := imq.Next(b) + if err != nil || n == 0 { + t.Fatalf("frames: %v", err) + } + out = append(out, f) + b = b[n:] + } + return out +} + +func TestHostedRoom(t *testing.T) { + r, h, events := setup() + + // IMVU's rooms are left alone. + if _, answered := xmlrpcCall(t, r, "chat.imvu.com", "chat.getOrMakeChat", xmlrpc.Struct{{Name: "activity", Value: "publicroom-170879926-5533"}}); answered { + t.Fatal("answered an IMVU room") + } + + // Joining: the hub lets the client in; it sits in the first free seat. + resp, _ := xmlrpcCall(t, r, "chat.imvu.com", "chat.getOrMakeChat", xmlrpc.Struct{{Name: "activity", Value: "publicroom-2100000000-3"}}) + if get(resp.Params[0], "chatId") != int64(2100000003) || get(resp.Params[0], "seat") != int64(1) || get(resp.Params[0], "lastMessageId") != int64(4) { + t.Fatalf("join answer %v", resp.Params[0]) + } + if len(*events) != 1 || (*events)[0] != "join 2100000003 2100000000-3" { + t.Fatalf("events %v", *events) + } + // A full room: the client is told why. + if resp, _ := xmlrpcCall(t, r, "chat.imvu.com", "chat.getOrMakeChat", xmlrpc.Struct{{Name: "activity", Value: "publicroom-2100000000-9"}}); resp.Fault == nil || resp.Fault.String != "This room is full." { + t.Fatalf("full room: %+v", resp) + } + + // Others in the room, without the client. + resp, _ = xmlrpcCall(t, r, "chat.imvu.com", "chat.getParticipants", xmlrpc.Struct{{Name: "userId", Value: int64(7)}, {Name: "chatId", Value: int64(2100000003)}}) + others := resp.Params[0].([]xmlrpc.Value) + if len(others) != 1 || get(others[0], "userId") != int64(88) { + t.Fatalf("participants %v", others) + } + + // IMQ: the room's queues are answered here; IMVU gets the rest. + var toClient bytes.Buffer + c := proxy.NewIMQConn(1, &toClient) + sub := imq.Subscription{Queues: []string{"/user/7"}, WithResults: []imq.QueueOp{{Queue: "/chat/2100000003", OpID: 5}, {Queue: "/chat/1385863547", OpID: 6}}} + rest := r.Route(c, false, sub) + if fmt.Sprint(rest.All()) != "[/user/7 /chat/1385863547]" { + t.Fatalf("for IMVU: %v", rest.All()) + } + var types []string + for _, f := range frames(t, toClient.Bytes()) { + types = append(types, imq.TypeName(f.Type)) + } + if got := strings.Join(types, " "); got != "G2cJoinedQueue G2cCreateMount G2cCreateMount G2cCreateMount G2cResult" { + t.Fatalf("frames %s", got) + } + + // Chat goes to the hub; sitting down too. + if !r.Send("/chat/2100000003", "messages", []byte(`{"message": "*seat 2", "userId": 7}`)) || r.Send("/chat/1385863547", "messages", []byte(`{}`)) { + t.Fatal("Send's answer") + } + if seats, sends := h.sentOf("room.seat"), h.sentOf("room.send"); len(seats) != 1 || seats[0]["seat"] != 2 || len(sends) != 1 { + t.Fatalf("seats %v, sends %v", seats, sends) + } + + // Others' chat comes back as the gateway would deliver it. + toClient.Reset() + r.FromHub([]byte(`{"type": "room.message", "room": 3, "from": 88, "mount": "messages", "message": "{\"message\":\"hi\",\"userId\":88}", "sequence": 5}`)) + fs := frames(t, toClient.Bytes()) + if len(fs) != 1 { + t.Fatalf("%d frames", len(fs)) + } + d, _ := imq.ParseDelivered(fs[0].Data) + if d.UserID != "/user/88" || d.Queue != "/chat/2100000003" || d.Sequence != 5 || !strings.Contains(string(d.Message), "hi") { + t.Fatalf("delivered %+v", d) + } + + // The room's card and scene come from the hub; the scene's community products as local IDs (missing ones left out). + resp, _ = xmlrpcCall(t, r, "client-dynamic.imvu.com", "test.getRoomCardInfo", xmlrpc.Struct{{Name: "room_instance_id", Value: "2100000000-3"}}) + if get(resp.Params[0], "name") != "Lounge b" || get(resp.Params[0], "owner_cid") != int64(88) { + t.Fatalf("card %v", resp.Params[0]) + } + resp, _ = xmlrpcCall(t, r, "client-dynamic.imvu.com", "test.getRoomState", xmlrpc.Struct{{Name: "room_instance_id", Value: "2100000000-3"}}) + contents := get(get(resp.Params[0], "room_state"), "room_contents").([]xmlrpc.Value) + if len(contents) != 2 || fmt.Sprint(contents[1]) != "[2 2000000005 37-11]" || get(get(get(resp.Params[0], "room_state"), "room_info"), "room_pid") != "50168116" { + t.Fatalf("scene %v", resp.Params[0]) + } + // Saving the scene: community products as the hub's IDs. + xmlrpcCall(t, r, "client-dynamic.imvu.com", "test.setRoomState", xmlrpc.Struct{{Name: "room_instance_id", Value: "2100000000-3"}, + {Name: "room_contents", Value: []xmlrpc.Value{[]xmlrpc.Value{int64(1), int64(47477426), "37-1"}, []xmlrpc.Value{int64(2), int64(2000000005), "37-2"}}}}) + if b, _ := json.Marshal(h.scene); string(b) != `[{"contents":[[1,47477426,"37-1"],[2,"coatcoat22","37-2"]]}]` { + t.Fatalf("saved %s", b) + } + + // The participant post for a hosted chat is answered here. + u, _ := url.Parse("https://api.imvu.com/chat/chat-2100000003/participants/user-7") + req, _ := http.NewRequest("POST", u.String(), nil) + x := &proxy.Exchange{URL: u, Request: req} + r.Intercept(x) + if x.Response == nil || x.Response.StatusCode != 200 { + t.Fatal("participant post not answered") + } + + // Anything else naming a hosted room is answered here, not sent to IMVU. + for _, raw := range []string{"https://api.imvu.com/room/room-2100000000-3", "https://client-dynamic.imvu.com/api/rooms/room_rating.php?room_instance_id=2100000000-3", + "https://client-dynamic.imvu.com/api/whatever.php?chat=2100000003"} { + u, _ := url.Parse(raw) + req, _ := http.NewRequest("GET", u.String(), nil) + x := &proxy.Exchange{URL: u, Request: req} + r.Intercept(x) + if x.Response == nil { + t.Errorf("%s went to IMVU", raw) + } + } + + // Leaving. + xmlrpcCall(t, r, "chat.imvu.com", "chat.terminateChat", xmlrpc.Struct{{Name: "userId", Value: int64(7)}, {Name: "chatId", Value: int64(2100000003)}}) + if len(h.sentOf("room.leave")) != 1 || (*events)[len(*events)-1] != "leave 2100000003" { + t.Fatalf("leave: %v %v", h.sentOf("room.leave"), *events) + } +} + +func TestRoomList(t *testing.T) { + r, _, _ := setup() + u, _ := url.Parse("https://client-dynamic.imvu.com/api/rooms/rooms_list_paginated.php?page=1") + req, _ := http.NewRequest("GET", u.String(), nil) + x := &proxy.Exchange{URL: u, Request: req} + r.Intercept(x) + body := []byte(`{"customers_room_id": ["644"], "customers_id": ["1"], "num_participants": [4], "name": ["A"], "participants": [[]], "is_ap": ["0"], "odd": [{"x": 1}], "number_of_rooms": 1}`) + out, err := x.ModifyResponse(&http.Response{StatusCode: 200}, body) + if err != nil { + t.Fatal(err) + } + var l map[string]any + json.Unmarshal(out, &l) + if fmt.Sprint(l["customers_room_id"]) != "[644 3]" || fmt.Sprint(l["customers_id"]) != "[1 2100000000]" || + fmt.Sprint(l["num_participants"]) != "[4 2]" || fmt.Sprint(l["odd"]) != "[map[x:1] map[]]" || l["number_of_rooms"] != float64(1) { + t.Fatalf("list %s", out) + } + // Other pages and searches are left alone. + u, _ = url.Parse("https://client-dynamic.imvu.com/api/rooms/rooms_list_paginated.php?page=2") + x = &proxy.Exchange{URL: u, Request: req} + r.Intercept(x) + if x.ModifyResponse != nil { + t.Fatal("changed page 2") + } +} diff --git a/internal/hub/hub.go b/internal/hub/hub.go index a098b68..d208c00 100644 --- a/internal/hub/hub.go +++ b/internal/hub/hub.go @@ -176,6 +176,10 @@ type Client struct { OnChat func(chat string, avatars []Avatar) // OnLibrary, if set, is called when connected and whenever the person's library changes. OnLibrary func() + // OnConnect, if set, is called each time the connection to the hub is made (to send state again). + OnConnect func() + // OnRoom, if set, gets the hub's messages about hosted rooms ("room.*"), as sent. + OnRoom func(data []byte) mu sync.Mutex saved *Saved @@ -183,11 +187,13 @@ type Client struct { outfit []Item chats map[string]string // chat ID -> room instance ("" for other chats) wake chan struct{} + outbox chan []byte // messages sent as they come (Send), not state } // 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, chats: map[string]string{}, wake: make(chan struct{}, 1)} + return &Client{path: path, saved: saved, log: logger, chats: map[string]string{}, wake: make(chan struct{}, 1), + outbox: make(chan []byte, 256)} } // Join reports the client is in an IMVU chat (room: the room instance, for a room's chat). @@ -241,6 +247,22 @@ func (c *Client) Outfit(imvuID int64, items []Item) { c.poke() } +// Send sends a message to the hub once (hosted rooms' messages); false if it can't be queued. Messages queued while +// disconnected are dropped on connecting (OnConnect sends what's still true). +func (c *Client) Send(v any) bool { + b, err := json.Marshal(v) + if err != nil { + return false + } + select { + case c.outbox <- b: + return true + default: + c.log.Printf("hub: too many messages waiting; dropped one") + return false + } +} + func (c *Client) poke() { select { case c.wake <- struct{}{}: @@ -289,6 +311,13 @@ func (c *Client) connect(ctx context.Context) error { if c.OnLibrary != nil { go c.OnLibrary() } + for drained := false; !drained; { + select { + case <-c.outbox: + default: + drained = true + } + } errs := make(chan error, 1) go func() { @@ -302,6 +331,17 @@ func (c *Client) connect(ctx context.Context) error { errs <- err return } + var kind struct{ Type string } + if json.Unmarshal(data, &kind) != nil { + continue + } + if strings.HasPrefix(kind.Type, "room.") { + // Hosted rooms' messages, whose fields differ (chat IDs are numbers). + if c.OnRoom != nil { + c.OnRoom(data) + } + continue + } var m struct { Type, Message, Chat string Avatars []Avatar @@ -349,6 +389,7 @@ func (c *Client) connect(ctx context.Context) error { }() sentID, sentOutfit := int64(-1), "" sentChats := map[string]string{} + connected := false for { c.mu.Lock() id, outfit := c.imvuID, append([]Item{}, c.outfit...) // a list, never null @@ -396,12 +437,26 @@ func (c *Client) connect(ctx context.Context) error { delete(sentChats, chat) } } + if !connected { + // After the session (so the hub knows the IMVU account), say what's still true. + connected = true + if c.OnConnect != nil { + go c.OnConnect() + } + } select { case <-ctx.Done(): conn.Close(websocket.StatusNormalClosure, "") return ctx.Err() case err := <-errs: return err + case b := <-c.outbox: + wctx, cancel := context.WithTimeout(ctx, 10*time.Second) + err := conn.Write(wctx, websocket.MessageText, b) + cancel() + if err != nil { + return err + } case <-c.wake: } } diff --git a/internal/imq/frame_test.go b/internal/imq/frame_test.go index 6cd7aba..e70f90b 100644 --- a/internal/imq/frame_test.go +++ b/internal/imq/frame_test.go @@ -81,3 +81,19 @@ func TestSendMessageRoundTrip(t *testing.T) { t.Fatal("a G2cResult parsed as a C2gSendMessage") } } + +func TestSubscriptionOps(t *testing.T) { + // A subscription with a plain queue, and one with an op ID and a property (field 3), which survives re-encoding. + withProps := appendBytesField(appendVarintField(appendBytesField(nil, 1, []byte("/chat/9")), 2, 7), 3, []byte("x")) + data := appendBytesField(appendBytesField(nil, 1, []byte("/user/1")), 2, withProps) + s, err := ParseSubscriptionOps(data) + if err != nil || len(s.Queues) != 1 || len(s.WithResults) != 1 || s.WithResults[0].Queue != "/chat/9" || s.WithResults[0].OpID != 7 { + t.Fatalf("%+v %v", s, err) + } + if string(s.Marshal()) != string(data) { + t.Fatal("re-encoded differently") + } + if all := s.All(); len(all) != 2 || all[1] != "/chat/9" { + t.Fatalf("all %v", all) + } +} diff --git a/internal/imq/messages.go b/internal/imq/messages.go index c7b1d6a..a032a48 100644 --- a/internal/imq/messages.go +++ b/internal/imq/messages.go @@ -9,6 +9,16 @@ const ( C2gSubscribe = 1309 G2cResult = 3102 G2cSendMessage = 3104 + G2cStateChange = 3105 + G2cJoinedQueue = 3106 + G2cLeftQueue = 3107 + G2cCreateMount = 3108 +) + +// Mount types (G2cCreateMount). +const ( + MessageMount = 1 + StateMount = 2 ) // SendMessage is a C2gSendMessage: a message the client sends to a queue's mount. @@ -202,3 +212,105 @@ func (m Delivered) Marshal() []byte { } return b } + +// QueueOp is a queue subscribed to with an operation ID, for a G2cResult. +type QueueOp struct { + Queue string + OpID uint64 + raw []byte // as the client sent it (with any other fields, such as properties) +} + +// Subscription is a C2gSubscribe's or C2gUnsubscribe's data: queues listed plainly, and with operation IDs. +type Subscription struct { + Queues []string + WithResults []QueueOp +} + +// ParseSubscriptionOps reads a C2gSubscribe's or C2gUnsubscribe's data, keeping operation IDs. +func ParseSubscriptionOps(data []byte) (Subscription, error) { + fs, err := fields(data) + if err != nil { + return Subscription{}, err + } + var s Subscription + for _, f := range fs { + switch { + case f.num == 1 && f.isLen: + s.Queues = append(s.Queues, string(f.bytes)) + case f.num == 2 && f.isLen: + sub, err := fields(f.bytes) + if err != nil { + return Subscription{}, err + } + q := QueueOp{raw: f.bytes} + for _, sf := range sub { + switch { + case sf.num == 1 && sf.isLen: + q.Queue = string(sf.bytes) + case sf.num == 2 && !sf.isLen: + q.OpID = sf.varint + } + } + s.WithResults = append(s.WithResults, q) + default: + return Subscription{}, errors.New("imq: unexpected field in a subscription") + } + } + return s, nil +} + +// All is every queue in the subscription. +func (s Subscription) All() []string { + out := append([]string(nil), s.Queues...) + for _, q := range s.WithResults { + out = append(out, q.Queue) + } + return out +} + +// Empty reports whether it names no queues. +func (s Subscription) Empty() bool { return len(s.Queues) == 0 && len(s.WithResults) == 0 } + +// Marshal encodes the subscription's data. +func (s Subscription) Marshal() []byte { + var b []byte + for _, q := range s.Queues { + b = appendBytesField(b, 1, []byte(q)) + } + for _, q := range s.WithResults { + raw := q.raw + if raw == nil { + raw = appendVarintField(appendBytesField(nil, 1, []byte(q.Queue)), 2, q.OpID) + } + b = appendBytesField(b, 2, raw) + } + return b +} + +// Property is a mount's property: a key and a value. +type Property struct{ Key, Value string } + +func appendProperties(b []byte, num uint64, props []Property) []byte { + for _, p := range props { + b = appendBytesField(b, num, appendBytesField(appendBytesField(nil, 1, []byte(p.Key)), 2, []byte(p.Value))) + } + return b +} + +// JoinedQueue encodes a G2cJoinedQueue's (or G2cLeftQueue's) data: a user joined (left) a queue. +func JoinedQueue(userID, queue string, sequence uint64) []byte { + return appendVarintField(appendBytesField(appendBytesField(nil, 1, []byte(userID)), 2, []byte(queue)), 3, sequence) +} + +// CreateMount encodes a G2cCreateMount's data: a queue's mount, of type MessageMount or StateMount. +func CreateMount(queue, mount string, mountType uint64, props []Property, sequence uint64) []byte { + b := appendVarintField(appendBytesField(appendBytesField(nil, 1, []byte(queue)), 2, []byte(mount)), 3, mountType) + return appendVarintField(appendProperties(b, 4, props), 5, sequence) +} + +// StateChange encodes a G2cStateChange's data: a state mount's properties changed (an empty value removes one). +func StateChange(opID uint64, userID, queue, mount string, props []Property, sequence uint64) []byte { + b := appendVarintField(nil, 1, opID) + b = appendBytesField(appendBytesField(appendBytesField(b, 2, []byte(userID)), 3, []byte(queue)), 4, []byte(mount)) + return appendVarintField(appendProperties(b, 5, props), 6, sequence) +} diff --git a/internal/products/intercept.go b/internal/products/intercept.go index 05a6582..e6a0ed3 100644 --- a/internal/products/intercept.go +++ b/internal/products/intercept.go @@ -359,6 +359,13 @@ func (p *Products) authorizeProducts2(x *proxy.Exchange, call *xmlrpc.Call) erro type auth struct { user, product int64 } + // In a room hosted by T5DE's hub, the client asks about the room's products as user 2100000000 (the first part + // of the room's ID): IMVU is asked about the room owner's instead, and answers go back as 2100000000. + owner := int64(0) + if p.HostedOwner != nil { + owner = p.HostedOwner() + } + hostedAsked := map[int64]bool{} var community []auth forward := []xmlrpc.Value{table[0]} asked := map[auth]bool{} @@ -372,6 +379,10 @@ func (p *Products) authorizeProducts2(x *proxy.Exchange, call *xmlrpc.Call) erro if !ok1 || !ok2 { return nil } + if user == HostedOwnerID && owner > 0 { + user, hostedAsked[product] = owner, true + row = []xmlrpc.Value{user, product} + } if !IsCommunity(product) { if !asked[auth{user, product}] { forward, asked[auth{user, product}] = append(forward, row), true @@ -386,9 +397,33 @@ func (p *Products) authorizeProducts2(x *proxy.Exchange, call *xmlrpc.Call) erro } } } - if len(community) == 0 { + if len(community) == 0 && len(hostedAsked) == 0 { return nil } + // hostedBack gives rows asked about as 2100000000 that user ID back. + hostedBack := func(columns []xmlrpc.Value, rows []xmlrpc.Value) { + userCol, productCol := -1, -1 + for i, c := range columns { + switch c { + case "userId": + userCol = i + case "productId": + productCol = i + } + } + if userCol < 0 || productCol < 0 { + return + } + for _, row := range rows { + r, ok := row.([]xmlrpc.Value) + if !ok || len(r) <= userCol || len(r) <= productCol { + continue + } + if product, _ := r[productCol].(int64); r[userCol] == owner && hostedAsked[product] { + r[userCol] = int64(HostedOwnerID) + } + } + } ours := func(columns []xmlrpc.Value) []xmlrpc.Value { var rows []xmlrpc.Value for _, a := range community { @@ -421,6 +456,7 @@ func (p *Products) authorizeProducts2(x *proxy.Exchange, call *xmlrpc.Call) erro columns[i] = c } out := append([]xmlrpc.Value{columns}, ours(columns)...) + hostedBack(columns, out[1:]) x.Response = localResponse(200, "text/xml", (&xmlrpc.Response{Params: []xmlrpc.Value{out}}).Marshal()) return nil } @@ -447,7 +483,9 @@ func (p *Products) authorizeProducts2(x *proxy.Exchange, call *xmlrpc.Call) erro if !ok { return body, errors.New("products: unexpected authorizeProducts2 columns") } - r.Params[0] = append(rows, ours(columns)...) + all := append(rows, ours(columns)...) + hostedBack(columns, all[1:]) + r.Params[0] = all return r.Marshal(), nil } return nil @@ -523,6 +561,15 @@ func (p *Products) interceptBaseLook(x *proxy.Exchange) error { // Outgoing implements proxy.IMQFilter: community products are taken out of outfit chat commands (IMVU and other // people see the defaults), and whole outfits are remembered. func (p *Products) Outgoing(queue, mount string, message []byte) ([]byte, bool) { + out, drop := p.outgoing(queue, mount, message) + // A hosted room's message goes to the hub (without community products, as to IMVU), never to IMVU. + if p.Hosted != nil && !drop && p.Hosted.Send(queue, mount, out) { + return nil, true + } + return out, drop +} + +func (p *Products) outgoing(queue, mount string, message []byte) ([]byte, bool) { if !strings.HasPrefix(queue, "/chat/") || mount != "messages" { return message, false } diff --git a/internal/products/products.go b/internal/products/products.go index fd4bb77..fee2250 100644 --- a/internal/products/products.go +++ b/internal/products/products.go @@ -30,6 +30,8 @@ import ( "git.t5de.dev/t5de/t5de-api/cfl" "git.t5de.dev/t5de/t5de-api/pack" "git.t5de.dev/t5de/t5de-proxy/internal/hub" + "git.t5de.dev/t5de/t5de-proxy/internal/imq" + "git.t5de.dev/t5de/t5de-proxy/internal/proxy" ) // Products is the proxy's community products: an Interceptor for HTTP and an IMQFilter for chat. @@ -67,6 +69,11 @@ type Products struct { // Library, if set, is the person's hub library as last read (for the inventory's Recently Added). Library func() []hub.LibraryEntry + // Hosted, if set, serves T5DE-hosted rooms' queues (the hosted package); HostedOwner is the IMVU account owning + // the hosted room the client is in (0 if none), whose products the room's are. + Hosted HostedRooms + HostedOwner func() int64 + roomMu sync.Mutex room room @@ -99,6 +106,26 @@ func (p *Products) Ready(id string) { } } +// HostedOwnerID is the user ID the client gives hosted rooms' owner: the first part of their instance IDs +// (hosted.Base). +const HostedOwnerID = 2100000000 + +// HostedRooms serves T5DE-hosted rooms' IMQ queues: Route answers subscriptions to them, Send takes the client's +// messages to them (reporting whether the queue was one), Closed forgets a connection. +type HostedRooms interface { + Route(c *proxy.IMQConn, unsubscribe bool, s imq.Subscription) imq.Subscription + Send(queue, mount string, message []byte) bool + Closed(c *proxy.IMQConn) +} + +// Route serves hosted rooms' queues (proxy.IMQRouter). +func (p *Products) Route(c *proxy.IMQConn, unsubscribe bool, s imq.Subscription) imq.Subscription { + if p.Hosted == nil { + return s + } + return p.Hosted.Route(c, unsubscribe, s) +} + // Observer hears about the client's IMVU account, its community outfit, and the chats it's in. type Observer interface { Session(imvuID int64) @@ -371,3 +398,22 @@ func (p *Products) reportOutfit(user string, locals []int64) { } p.Observer.Outfit(id, items) } + +// LocalID is the client's local ID for a product (a hub product ID or a local pack's), if it's here. +func (p *Products) LocalID(id string) (int64, bool) { + if s := p.byProductID(id); s != nil { + return s.local, true + } + return 0, false +} + +// ProductID is the product a local ID was given to. +func (p *Products) ProductID(local int64) (string, bool) { return p.state.productID(local) } + +// User is the IMVU account the client signed in as (0 before it has). +func (p *Products) User() int64 { + p.mu.Lock() + defer p.mu.Unlock() + id, _ := strconv.ParseInt(p.user, 10, 64) + return id +} diff --git a/internal/products/products_test.go b/internal/products/products_test.go index c397648..a9316b3 100644 --- a/internal/products/products_test.go +++ b/internal/products/products_test.go @@ -417,3 +417,24 @@ func TestSavedOutfits(t *testing.T) { t.Fatalf("outfits %s", body) } } + +func TestHostedRoomProducts(t *testing.T) { + h, _ := newHarness(t) + h.do("POST", "https://secure.imvu.com/api/xmlrpc/client.php", call("test.avatarInfoForLogin2"), false, false) + h.p.HostedOwner = func() int64 { return 7 } // the fake IMVU answers as user 7 + auths := xmlrpc.Struct{{Name: "auths", Value: []xmlrpc.Value{ + []xmlrpc.Value{"userId", "productId"}, []xmlrpc.Value{int64(HostedOwnerID), int64(500)}}}} + _, body := h.do("POST", "http://client-dynamic.imvu.com/api/xmlrpc/client.php", call("test.authorizeProducts2", auths), true, false) + asked := h.imvu.asked["test.authorizeProducts2"] + if len(asked) != 1 || strings.Contains(asked[0], "2100000000") { + t.Fatalf("IMVU was asked %q", asked) + } + r, err := xmlrpc.ParseResponse(body) + if err != nil { + t.Fatal(err) + } + rows := r.Params[0].([]xmlrpc.Value) + if first := rows[1].([]xmlrpc.Value); first[0] != int64(HostedOwnerID) || first[2] != int64(500) { + t.Fatalf("answer %v", rows) + } +} diff --git a/internal/products/room.go b/internal/products/room.go index 519bb41..a42e1e1 100644 --- a/internal/products/room.go +++ b/internal/products/room.go @@ -116,6 +116,9 @@ func (p *Products) Unsubscribed(c *proxy.IMQConn, queues []string) { // Closed implements proxy.IMQWatcher: a connection's subscriptions end with it. func (p *Products) Closed(c *proxy.IMQConn) { + if p.Hosted != nil { + p.Hosted.Closed(c) + } p.roomMu.Lock() p.room.init() var queues []string diff --git a/internal/proxy/connect.go b/internal/proxy/connect.go index d538e92..9f7e45d 100644 --- a/internal/proxy/connect.go +++ b/internal/proxy/connect.go @@ -123,6 +123,13 @@ type IMQWatcher interface { Closed(c *IMQConn) } +// IMQRouter is an IMQWatcher that serves some queues itself (T5DE-hosted rooms' queues): it takes them out of the +// client's subscriptions before they go to the gateway, and answers the client for them. +type IMQRouter interface { + // Route returns the part of a subscription (or unsubscription) for the gateway. + Route(c *IMQConn, unsubscribe bool, s imq.Subscription) imq.Subscription +} + // IMQConn is one IMQ connection of the client's. type IMQConn struct { ID int64 @@ -132,6 +139,12 @@ type IMQConn struct { // 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} } +// Send gives the client a frame of the given type as if the gateway had sent it. +func (c *IMQConn) Send(frameType uint64, data []byte) error { + _, err := c.w.Write(imq.EncodeFrame(frameType, data)) + return err +} + // 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())) @@ -225,14 +238,27 @@ 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) - } + sub, err := imq.ParseSubscriptionOps(f.Data) + if err != nil { + return frame } - return frame + if f.Type == imq.C2gSubscribe { + st.watcher.Subscribed(st.conn, sub.All()) + } else { + st.watcher.Unsubscribed(st.conn, sub.All()) + } + router, ok := st.watcher.(IMQRouter) + if !ok { + return frame + } + rest := router.Route(st.conn, f.Type == imq.C2gUnsubscribe, sub) + if rest.Empty() { + return nil + } + if len(rest.All()) == len(sub.All()) { + return frame + } + return imq.EncodeFrame(f.Type, rest.Marshal()) case imq.G2cSendMessage: m, err := imq.ParseDelivered(f.Data) if err != nil {