From c281fb572cf8c0921184bad8d5fc1214eee9c380 Mon Sep 17 00:00:00 2001 From: dhkatz Date: Thu, 29 Oct 2026 18:56:26 -0700 Subject: [PATCH] Get T5 products from the hub's catalogs The proxy keeps the person's library in a cache folder (hub-products in its data folder), one pack per product, served like any other pack: the hub says which revision to load and which catalogs host it, each file comes from a catalog and is checked against the hub's sha256 (files are kept by hash, so a new revision only downloads what changed), and adult files are fetched with the hub's access token. Products someone wears in a room that aren't here yet are fetched too, and put on them when they arrive. Co-Authored-By: Claude Opus 5.5 --- cmd/t5de-proxy/main.go | 12 +- internal/fetch/fetch.go | 293 ++++++++++++++++++++++++++++++++++ internal/fetch/fetch_test.go | 143 +++++++++++++++++ internal/hub/hub.go | 95 +++++++++++ internal/products/products.go | 30 +++- internal/products/room.go | 9 +- 6 files changed, 579 insertions(+), 3 deletions(-) create mode 100644 internal/fetch/fetch.go create mode 100644 internal/fetch/fetch_test.go diff --git a/cmd/t5de-proxy/main.go b/cmd/t5de-proxy/main.go index a597a71..5c28e45 100644 --- a/cmd/t5de-proxy/main.go +++ b/cmd/t5de-proxy/main.go @@ -33,6 +33,7 @@ import ( "time" "git.t5de.dev/t5de/t5de-proxy/internal/buildinfo" + "git.t5de.dev/t5de/t5de-proxy/internal/fetch" "git.t5de.dev/t5de/t5de-proxy/internal/hub" "git.t5de.dev/t5de/t5de-proxy/internal/parent" "git.t5de.dev/t5de/t5de-proxy/internal/products" @@ -125,7 +126,12 @@ func run(logger *log.Logger, listen, ready string, parentPID int, trace bool, pa }() } handler := &proxy.Server{Token: token, Log: logger, Trace: trace} - prods, err := products.New(packs, data, logger.Printf) + // Products from the hub's catalogs are kept as packs in their own folder, served like the others. + hubProducts := filepath.Join(data, "hub-products") + if err := os.MkdirAll(hubProducts, 0o755); err != nil { + logger.Printf("hub products: %v", err) + } + prods, err := products.New(append(packs, hubProducts), data, logger.Printf) if err != nil { logger.Printf("community products are off: %v", err) } else { @@ -134,10 +140,14 @@ func run(logger *log.Logger, listen, ready string, parentPID int, trace bool, pa } if saved, err := hub.Load(filepath.Join(data, "hub.json")); err == nil { client := hub.New(filepath.Join(data, "hub.json"), saved, logger) + fetcher := fetch.New(client, hubProducts, logger.Printf) if prods != nil { prods.Observer = client client.OnChat = prods.Others + fetcher.OnReady = prods.Ready + prods.Want = fetcher.Want } + client.OnLibrary = func() { fetcher.SyncLibrary(context.Background()) } go client.Run(context.Background()) } else if !errors.Is(err, hub.ErrSignedOut) { logger.Printf("hub: %v", err) diff --git a/internal/fetch/fetch.go b/internal/fetch/fetch.go new file mode 100644 index 0000000..71c7c1c --- /dev/null +++ b/internal/fetch/fetch.go @@ -0,0 +1,293 @@ +// Package fetch gets T5 products from the hub's catalogs (t5de-hub's docs/products.md): the hub says which revision +// to load and which catalogs host it; the files come from a catalog, each checked against the hub's sha256, and the +// product is written as a pack in the cache folder, where the products package serves it like any other pack. +package fetch + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "regexp" + "strings" + "sync" + "time" + "unicode/utf8" + + "git.t5de.dev/t5de/t5de-proxy/internal/hub" +) + +// Hub is what the fetcher asks the hub. +type Hub interface { + ProductFiles(ctx context.Context, id string) (*hub.Files, error) + AccessToken(ctx context.Context) (string, error) + Library(ctx context.Context) ([]hub.LibraryEntry, error) +} + +// IsHubID reports whether id is a hub product ID (10 of a-z2-7), as opposed to a local pack's. +var IsHubID = regexp.MustCompile(`^[a-z2-7]{10}$`).MatchString + +// Fetcher keeps hub products in a cache folder, one pack folder per product. +type Fetcher struct { + hub Hub + dir string + logf func(format string, args ...any) + http *http.Client + + // OnReady, if set, is called when a product was written (or updated) in the cache. + OnReady func(id string) + + mu sync.Mutex + inflight map[string]bool + failed map[string]time.Time // products that failed lately (not retried for a while) + access string + accessAt time.Time +} + +// New keeps products in dir. +func New(h Hub, dir string, logf func(format string, args ...any)) *Fetcher { + if logf == nil { + logf = func(string, ...any) {} + } + return &Fetcher{hub: h, dir: dir, logf: logf, http: &http.Client{Timeout: 2 * time.Minute}, + inflight: map[string]bool{}, failed: map[string]time.Time{}} +} + +// Dir is the cache folder (a pack root). +func (f *Fetcher) Dir() string { return f.dir } + +// SyncLibrary makes sure every product in the person's library is in the cache, at the revision they use. +func (f *Fetcher) SyncLibrary(ctx context.Context) { + lib, err := f.hub.Library(ctx) + if err != nil { + f.logf("fetch: reading the library: %v", err) + return + } + for _, e := range lib { + if err := f.Ensure(ctx, e.Product.ID); err != nil { + f.logf("fetch: %s (%s): %v", e.Product.ID, e.Product.Name, err) + } + } +} + +// Want gets a product in the background, once at a time, and not again soon after a failure. +func (f *Fetcher) Want(id string) { + if !IsHubID(id) { + return + } + f.mu.Lock() + if f.inflight[id] || time.Since(f.failed[id]) < 5*time.Minute { + f.mu.Unlock() + return + } + f.inflight[id] = true + f.mu.Unlock() + go func() { + err := f.Ensure(context.Background(), id) + f.mu.Lock() + delete(f.inflight, id) + if err != nil { + f.failed[id] = time.Now() + } + f.mu.Unlock() + if err != nil { + f.logf("fetch: %s: %v", id, err) + } + }() +} + +// manifest is the pack manifest written for a product (t5de-api's pack format 1). +type manifest struct { + Format int `json:"format"` + ID string `json:"id"` + Revision int `json:"revision"` + Name string `json:"name"` + Creator string `json:"creator,omitempty"` + Rating string `json:"rating"` +} + +// Ensure puts the product's current revision in the cache, unless it's there already. +func (f *Fetcher) Ensure(ctx context.Context, id string) error { + if !IsHubID(id) { + return fmt.Errorf("not a hub product ID: %q", id) + } + info, err := f.hub.ProductFiles(ctx, id) + if err != nil { + return err + } + final := filepath.Join(f.dir, id) + if b, err := os.ReadFile(filepath.Join(final, "t5de-product.json")); err == nil { + var m manifest + if json.Unmarshal(b, &m) == nil && m.Revision == info.Revision { + return nil + } + } + if len(info.Catalogs) == 0 { + return errors.New("no catalog hosts it") + } + for _, fe := range info.Files { + if fe.Name == "" || fe.Name == "." || fe.Name == ".." || filepath.Base(fe.Name) != fe.Name || strings.ContainsAny(fe.Name, `/\:`) { + return fmt.Errorf("bad file name %q", fe.Name) + } + } + token := "" + if info.Rating == "adult" { + if token, err = f.accessToken(ctx); err != nil { + return err + } + } + for _, fe := range info.Files { + if err := f.blob(ctx, info.Catalogs, fe, token); err != nil { + return fmt.Errorf("%s: %w", fe.Name, err) + } + } + + // The pack, in a temporary folder, then in place. + tmp, err := os.MkdirTemp(f.dir, "."+id+"-") + if err != nil { + return err + } + defer os.RemoveAll(tmp) + if err := os.Mkdir(filepath.Join(tmp, "files"), 0o755); err != nil { + return err + } + for _, fe := range info.Files { + if err := copyFile(f.blobPath(fe.SHA256), filepath.Join(tmp, "files", fe.Name)); err != nil { + return err + } + } + m := manifest{Format: 1, ID: id, Revision: info.Revision, Name: clip(info.Name, "Untitled"), Creator: clip(info.Creator, ""), Rating: info.Rating} + b, _ := json.MarshalIndent(m, "", " ") + if err := os.WriteFile(filepath.Join(tmp, "t5de-product.json"), b, 0o644); err != nil { + return err + } + old := final + ".old" + os.RemoveAll(old) + if _, err := os.Stat(final); err == nil { + if err := os.Rename(final, old); err != nil { + return err + } + } + if err := os.Rename(tmp, final); err != nil { + os.Rename(old, final) + return err + } + os.RemoveAll(old) + f.logf("fetch: %s revision %d (%s) is ready", id, info.Revision, m.Name) + if f.OnReady != nil { + f.OnReady(id) + } + return nil +} + +// clip makes a name fit the pack format (1-64 characters, no control characters). +func clip(s, empty string) string { + s = strings.Map(func(r rune) rune { + if r < 0x20 || r == 0x7f { + return -1 + } + return r + }, strings.TrimSpace(s)) + for utf8.RuneCountInString(s) > 64 { + _, n := utf8.DecodeLastRuneInString(s) + s = s[:len(s)-n] + } + if s == "" { + return empty + } + return s +} + +func (f *Fetcher) blobPath(sum string) string { return filepath.Join(f.dir, ".blobs", sum) } + +// blob gets a file into the blob cache (by sha256), from the first catalog that has it. +func (f *Fetcher) blob(ctx context.Context, catalogs []string, fe hub.FileEntry, token string) error { + path := f.blobPath(fe.SHA256) + if st, err := os.Stat(path); err == nil && st.Size() == fe.Size { + return nil + } + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + var last error + for _, c := range catalogs { + if last = f.download(ctx, strings.TrimSuffix(c, "/")+"/api/v1/files/"+fe.SHA256, fe, token, path); last == nil { + return nil + } + } + return last +} + +func (f *Fetcher) download(ctx context.Context, url string, fe hub.FileEntry, token, path string) error { + req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + if err != nil { + return err + } + req.Header.Set("User-Agent", hub.UserAgent) + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + resp, err := f.http.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("%s answered %d", url, resp.StatusCode) + } + tmp, err := os.CreateTemp(filepath.Dir(path), ".dl-") + if err != nil { + return err + } + defer os.Remove(tmp.Name()) + h := sha256.New() + n, err := io.Copy(io.MultiWriter(tmp, h), io.LimitReader(resp.Body, fe.Size+1)) + tmp.Close() + if err != nil { + return err + } + if n != fe.Size || hex.EncodeToString(h.Sum(nil)) != fe.SHA256 { + return fmt.Errorf("%s sent a different file than the hub listed", url) + } + return os.Rename(tmp.Name(), path) +} + +func (f *Fetcher) accessToken(ctx context.Context) (string, error) { + f.mu.Lock() + tok, at := f.access, f.accessAt + f.mu.Unlock() + if tok != "" && time.Since(at) < 50*time.Minute { + return tok, nil + } + tok, err := f.hub.AccessToken(ctx) + if err != nil { + return "", err + } + f.mu.Lock() + f.access, f.accessAt = tok, time.Now() + f.mu.Unlock() + return tok, nil +} + +func copyFile(src, dst string) error { + in, err := os.Open(src) + if err != nil { + return err + } + defer in.Close() + out, err := os.Create(dst) + if err != nil { + return err + } + if _, err := io.Copy(out, in); err != nil { + out.Close() + return err + } + return out.Close() +} diff --git a/internal/fetch/fetch_test.go b/internal/fetch/fetch_test.go new file mode 100644 index 0000000..8f1d913 --- /dev/null +++ b/internal/fetch/fetch_test.go @@ -0,0 +1,143 @@ +package fetch + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + + "git.t5de.dev/t5de/t5de-api/pack" + "git.t5de.dev/t5de/t5de-proxy/internal/hub" +) + +const id = "abcdefgh23" + +func indexXML(rating string) []byte { + return []byte(``) +} + +func entry(name string, data []byte) hub.FileEntry { + sum := sha256.Sum256(data) + return hub.FileEntry{Name: name, SHA256: hex.EncodeToString(sum[:]), Size: int64(len(data))} +} + +// fakeHub answers for one product. +type fakeHub struct { + mu sync.Mutex + files hub.Files +} + +func (h *fakeHub) ProductFiles(ctx context.Context, pid string) (*hub.Files, error) { + h.mu.Lock() + defer h.mu.Unlock() + f := h.files + return &f, nil +} +func (h *fakeHub) AccessToken(ctx context.Context) (string, error) { return "adult-token", nil } +func (h *fakeHub) Library(ctx context.Context) ([]hub.LibraryEntry, error) { + var e hub.LibraryEntry + e.Product.ID = id + return []hub.LibraryEntry{e}, nil +} + +// catalog serves blobs by sha256; adult ones need the token. +func catalog(t *testing.T, blobs map[string][]byte, adult map[string]bool, served *int) *httptest.Server { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + sum := strings.TrimPrefix(r.URL.Path, "/api/v1/files/") + b, ok := blobs[sum] + if !ok { + http.NotFound(w, r) + return + } + if adult[sum] && r.Header.Get("Authorization") != "Bearer adult-token" { + w.WriteHeader(http.StatusUnauthorized) + return + } + *served++ + w.Write(b) + })) + t.Cleanup(srv.Close) + return srv +} + +func TestEnsureWritesAPack(t *testing.T) { + index, tex := indexXML("GA"), []byte("texture one") + tex2 := []byte("texture two") + blobs := map[string][]byte{} + for _, b := range [][]byte{index, tex, tex2} { + blobs[entry("", b).SHA256] = b + } + served := 0 + cat := catalog(t, blobs, nil, &served) + h := &fakeHub{files: hub.Files{Product: id, Revision: 1, Rating: "general", Name: "Blue Coat", Creator: "maker", + Files: []hub.FileEntry{entry("index.xml", index), entry("tex.png", tex)}, Catalogs: []string{"http://127.0.0.1:1", cat.URL}}} + dir := t.TempDir() + f := New(h, dir, t.Logf) + var ready []string + f.OnReady = func(id string) { ready = append(ready, id) } + + f.SyncLibrary(context.Background()) + p, err := pack.Load(filepath.Join(dir, id)) + if err != nil { + t.Fatal(err) + } + if p.Manifest.Name != "Blue Coat" || p.Manifest.Revision != 1 || len(p.Files) != 2 || len(ready) != 1 { + t.Fatalf("pack %+v, ready %v", p.Manifest, ready) + } + // Already there: nothing fetched again. + if err := f.Ensure(context.Background(), id); err != nil || served != 2 || len(ready) != 1 { + t.Fatalf("again: %v, served %d", err, served) + } + + // A new revision: only the changed file is downloaded; the pack is replaced. + h.files.Revision, h.files.Files = 2, []hub.FileEntry{entry("index.xml", index), entry("tex.png", tex2)} + if err := f.Ensure(context.Background(), id); err != nil { + t.Fatal(err) + } + if b, _ := os.ReadFile(filepath.Join(dir, id, "files", "tex.png")); string(b) != "texture two" || served != 3 { + t.Fatalf("tex %q, served %d", b, served) + } + entries, _ := os.ReadDir(dir) + for _, e := range entries { + if e.Name() != id && e.Name() != ".blobs" { + t.Fatalf("left behind %s", e.Name()) + } + } +} + +func TestEnsureChecksFilesAndAdultTokens(t *testing.T) { + index := indexXML("AP") + blobs := map[string][]byte{entry("", index).SHA256: index} + adult := map[string]bool{entry("", index).SHA256: true} + served := 0 + cat := catalog(t, blobs, adult, &served) + h := &fakeHub{files: hub.Files{Product: id, Revision: 1, Rating: "adult", Name: "Lace", + Files: []hub.FileEntry{entry("index.xml", index)}, Catalogs: []string{cat.URL}}} + f := New(h, t.TempDir(), t.Logf) + if err := f.Ensure(context.Background(), id); err != nil { + t.Fatal(err) + } + + // A catalog sending something else than the hub listed is refused. + h.files.Revision = 2 + h.files.Files = []hub.FileEntry{{Name: "index.xml", SHA256: strings.Repeat("0", 64), Size: int64(len(index))}} + blobs[strings.Repeat("0", 64)] = []byte(strings.Repeat("x", len(index))) + if err := f.Ensure(context.Background(), id); err == nil || !strings.Contains(err.Error(), "different file") { + t.Fatalf("got %v", err) + } + // File names that would leave the pack's folder are refused. + h.files.Files = []hub.FileEntry{{Name: `..\evil`, SHA256: strings.Repeat("0", 64), Size: 1}} + if err := f.Ensure(context.Background(), id); err == nil || !strings.Contains(err.Error(), "bad file name") { + t.Fatalf("got %v", err) + } + if err := f.Ensure(context.Background(), "../etc"); err == nil { + t.Fatal("bad ID accepted") + } +} diff --git a/internal/hub/hub.go b/internal/hub/hub.go index 3a1ed52..e65d922 100644 --- a/internal/hub/hub.go +++ b/internal/hub/hub.go @@ -173,6 +173,8 @@ type Client struct { // 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) + // OnLibrary, if set, is called when connected and whenever the person's library changes. + OnLibrary func() mu sync.Mutex saved *Saved @@ -283,6 +285,9 @@ func (c *Client) connect(ctx context.Context) error { } defer conn.CloseNow() c.log.Printf("hub: connected to %s", c.saved.URL) + if c.OnLibrary != nil { + go c.OnLibrary() + } errs := make(chan error, 1) go func() { @@ -306,6 +311,10 @@ func (c *Client) connect(ctx context.Context) error { switch m.Type { case "error": c.log.Printf("hub: %s", m.Message) + case "library": + if c.OnLibrary != nil { + go c.OnLibrary() + } case "chat": c.mu.Lock() _, in := c.chats[m.Chat] @@ -396,3 +405,89 @@ func (c *Client) connect(ctx context.Context) error { } } } + +// Call makes a request to the hub's API as the signed-in person: path under the hub's URL, body (if any) sent as +// JSON, the answer decoded into out. +func (c *Client) Call(ctx context.Context, method, path string, body, out any) error { + var r io.Reader + if body != nil { + b, _ := json.Marshal(body) + r = strings.NewReader(string(b)) + } + req, err := http.NewRequestWithContext(ctx, method, c.saved.URL+path, r) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+c.saved.Token) + req.Header.Set("User-Agent", UserAgent) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + data, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<20)) + if resp.StatusCode/100 != 2 { + var e struct{ Error, Message string } + json.Unmarshal(data, &e) + return fmt.Errorf("the hub answered %d %s", resp.StatusCode, e.Error) + } + if out == nil { + return nil + } + return json.Unmarshal(data, out) +} + +// FileEntry is one of a revision's files. +type FileEntry struct { + Name string `json:"name"` + SHA256 string `json:"sha256"` + Size int64 `json:"size"` +} + +// Files is the revision of a product the person uses, and where to get it (GET /api/v1/products/:id/files). +type Files struct { + Product string `json:"product"` + Revision int `json:"revision"` + Rating string `json:"rating"` + Parent int64 `json:"parent"` + Name string `json:"name"` + Creator string `json:"creator"` + Files []FileEntry `json:"files"` + Catalogs []string `json:"catalogs"` + Token string `json:"token"` +} + +// LibraryEntry is a product in the person's library. +type LibraryEntry struct { + Product struct { + ID string `json:"id"` + Name string `json:"name"` + } `json:"product"` + StayOn *int `json:"stayOn"` +} + +// Library is the person's library. +func (c *Client) Library(ctx context.Context) ([]LibraryEntry, error) { + var out struct{ Library []LibraryEntry } + err := c.Call(ctx, "GET", "/api/v1/library", nil, &out) + return out.Library, err +} + +// ProductFiles is the revision of a product to load. +func (c *Client) ProductFiles(ctx context.Context, id string) (*Files, error) { + var out Files + if err := c.Call(ctx, "GET", "/api/v1/products/"+id+"/files", nil, &out); err != nil { + return nil, err + } + return &out, nil +} + +// AccessToken is a token for catalogs' adult files (an hour long). +func (c *Client) AccessToken(ctx context.Context) (string, error) { + var out struct{ Token string } + err := c.Call(ctx, "POST", "/api/v1/access", nil, &out) + return out.Token, err +} diff --git a/internal/products/products.go b/internal/products/products.go index b0ba9c3..5e47f8d 100644 --- a/internal/products/products.go +++ b/internal/products/products.go @@ -48,10 +48,38 @@ type Products struct { // chats it's in (the hub client). Observer Observer + // Want, if set, is asked for products someone in a room wears that aren't here (the fetcher gets hub products). + Want func(id string) + roomMu sync.Mutex room room } +// Ready is told a product arrived (the fetcher wrote it): it's served from now on, and given to the client for +// anyone in a room who wears it. +func (p *Products) Ready(id string) { + p.refresh(true) + p.roomMu.Lock() + p.room.init() + var chats []string + for chat, avatars := range p.room.last { + for _, a := range avatars { + for _, it := range a.Items { + if it.ID == id { + chats = append(chats, chat) + } + } + } + } + p.roomMu.Unlock() + for _, chat := range chats { + p.roomMu.Lock() + avatars := p.room.last[chat] + p.roomMu.Unlock() + p.Others(chat, avatars) + } +} + // Observer hears about the client's IMVU account, its community outfit, and the chats it's in. type Observer interface { Session(imvuID int64) @@ -120,7 +148,7 @@ func (p *Products) refresh(force bool) { } for _, e := range entries { dir := filepath.Join(root, e.Name()) - if !e.IsDir() { + if !e.IsDir() || strings.HasPrefix(e.Name(), ".") { // hidden: the fetcher's work in progress continue } if _, err := os.Stat(filepath.Join(dir, "t5de-product.json")); err != nil { diff --git a/internal/products/room.go b/internal/products/room.go index 204b369..7535e0d 100644 --- a/internal/products/room.go +++ b/internal/products/room.go @@ -30,6 +30,7 @@ type room struct { 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) + last map[string][]hub.Avatar // chat -> what the hub last said about the others in it } func (r *room) init() { @@ -39,6 +40,7 @@ func (r *room) init() { r.applied = map[string]map[int64][]int64{} r.userIDs = map[int64]string{} r.missing = map[string]bool{} + r.last = map[string][]hub.Avatar{} } } @@ -184,13 +186,16 @@ func (p *Products) Others(chat string, avatars []hub.Avatar) { locals = append(locals, s.local) continue } + if p.Want != nil { + p.Want(it.ID) // given to the client when it arrives (Ready) + } 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) + p.logf("products: someone in the room wears %q (%s), which isn't here yet", it.Name, it.ID) } } slices.Sort(locals) @@ -201,9 +206,11 @@ func (p *Products) Others(chat string, avatars []hub.Avatar) { p.room.init() if avatars == nil { delete(p.room.applied, chat) + delete(p.room.last, chat) p.roomMu.Unlock() return } + p.room.last[chat] = avatars had := p.room.applied[chat] var lines []line for user, old := range had {