| 1 | package commands |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "net/http" |
| 9 | "slices" |
| 10 | |
| 11 | "github.com/ipfs/go-datastore" |
| 12 | "github.com/ipfs/go-datastore/query" |
| 13 | cmds "github.com/ipfs/go-ipfs-cmds" |
| 14 | cmdenv "github.com/ipfs/kubo/core/commands/cmdenv" |
| 15 | options "github.com/ipfs/kubo/core/coreiface/options" |
| 16 | "github.com/ipfs/kubo/core/node/libp2p" |
| 17 | "github.com/libp2p/go-libp2p/core/peer" |
| 18 | mbase "github.com/multiformats/go-multibase" |
| 19 | ) |
| 20 | |
| 21 | var PubsubCmd = &cmds.Command{ |
| 22 | Status: cmds.Experimental, |
| 23 | Helptext: cmds.HelpText{ |
| 24 | Tagline: "An experimental publish-subscribe system on ipfs.", |
| 25 | ShortDescription: ` |
| 26 | ipfs pubsub allows you to publish messages to a given topic, and also to |
| 27 | subscribe to new messages on a given topic. |
| 28 | |
| 29 | EXPERIMENTAL FEATURE |
| 30 | |
| 31 | This is an opt-in feature optimized for IPNS over PubSub |
| 32 | (https://specs.ipfs.tech/ipns/ipns-pubsub-router/). |
| 33 | |
| 34 | The default message validator is designed for IPNS record protocol. |
| 35 | For custom pubsub applications requiring different validation logic, |
| 36 | use go-libp2p-pubsub (https://github.com/libp2p/go-libp2p-pubsub) |
| 37 | directly in a dedicated binary. |
| 38 | |
| 39 | To enable, set 'Pubsub.Enabled' config to true. |
| 40 | `, |
| 41 | }, |
| 42 | Subcommands: map[string]*cmds.Command{ |
| 43 | "pub": PubsubPubCmd, |
| 44 | "sub": PubsubSubCmd, |
| 45 | "ls": PubsubLsCmd, |
| 46 | "peers": PubsubPeersCmd, |
| 47 | "reset": PubsubResetCmd, |
| 48 | }, |
| 49 | } |
| 50 | |
| 51 | type pubsubMessage struct { |
| 52 | From string `json:"from,omitempty"` |
| 53 | Data string `json:"data,omitempty"` |
| 54 | Seqno string `json:"seqno,omitempty"` |
| 55 | TopicIDs []string `json:"topicIDs,omitempty"` |
| 56 | } |
| 57 | |
| 58 | var PubsubSubCmd = &cmds.Command{ |
| 59 | Status: cmds.Experimental, |
| 60 | Helptext: cmds.HelpText{ |
| 61 | Tagline: "Subscribe to messages on a given topic.", |
| 62 | ShortDescription: ` |
| 63 | ipfs pubsub sub subscribes to messages on a given topic. |
| 64 | |
| 65 | EXPERIMENTAL FEATURE |
| 66 | |
| 67 | This is an opt-in feature optimized for IPNS over PubSub |
| 68 | (https://specs.ipfs.tech/ipns/ipns-pubsub-router/). |
| 69 | |
| 70 | To enable, set 'Pubsub.Enabled' config to true. |
| 71 | |
| 72 | PEER ENCODING |
| 73 | |
| 74 | Peer IDs in From fields are encoded using the default text representation |
| 75 | from go-libp2p. This ensures the same string values as in 'ipfs pubsub peers'. |
| 76 | |
| 77 | TOPIC AND DATA ENCODING |
| 78 | |
| 79 | Topics, Data and Seqno are binary data. To ensure all bytes are transferred |
| 80 | correctly the RPC client and server will use multibase encoding behind |
| 81 | the scenes. |
| 82 | |
| 83 | You can inspect the format by passing --enc=json. The ipfs multibase commands |
| 84 | can be used for encoding/decoding multibase strings in the userland. |
| 85 | `, |
| 86 | }, |
| 87 | Arguments: []cmds.Argument{ |
| 88 | cmds.StringArg("topic", true, false, "Name of topic to subscribe to (multibase encoded when sent over HTTP RPC)."), |
| 89 | }, |
| 90 | PreRun: urlArgsEncoder, |
| 91 | Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error { |
| 92 | api, err := cmdenv.GetApi(env, req) |
| 93 | if err != nil { |
| 94 | return err |
| 95 | } |
| 96 | if err := urlArgsDecoder(req, env); err != nil { |
| 97 | return err |
| 98 | } |
| 99 | |
| 100 | topic := req.Arguments[0] |
| 101 | |
| 102 | sub, err := api.PubSub().Subscribe(req.Context, topic) |
| 103 | if err != nil { |
| 104 | return err |
| 105 | } |
| 106 | defer sub.Close() |
| 107 | |
| 108 | if f, ok := res.(http.Flusher); ok { |
| 109 | f.Flush() |
| 110 | } |
| 111 | |
| 112 | for { |
| 113 | msg, err := sub.Next(req.Context) |
| 114 | if err == io.EOF || err == context.Canceled { |
| 115 | return nil |
| 116 | } else if err != nil { |
| 117 | return err |
| 118 | } |
| 119 | |
| 120 | // turn bytes into strings |
| 121 | encoder, _ := mbase.EncoderByName("base64url") |
| 122 | psm := pubsubMessage{ |
| 123 | Data: encoder.Encode(msg.Data()), |
| 124 | From: msg.From().String(), |
| 125 | Seqno: encoder.Encode(msg.Seq()), |
| 126 | } |
| 127 | for _, topic := range msg.Topics() { |
| 128 | psm.TopicIDs = append(psm.TopicIDs, encoder.Encode([]byte(topic))) |
| 129 | } |
| 130 | if err := res.Emit(&psm); err != nil { |
| 131 | return err |
| 132 | } |
| 133 | } |
| 134 | }, |
| 135 | Encoders: cmds.EncoderMap{ |
| 136 | cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, psm *pubsubMessage) error { |
| 137 | _, dec, err := mbase.Decode(psm.Data) |
| 138 | if err != nil { |
| 139 | return err |
| 140 | } |
| 141 | _, err = w.Write(dec) |
| 142 | return err |
| 143 | }), |
| 144 | // DEPRECATED, undocumented format we used in tests, but not anymore |
| 145 | // <message.payload>\n<message.payload>\n |
| 146 | "ndpayload": cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, psm *pubsubMessage) error { |
| 147 | return errors.New("--enc=ndpayload was removed, use --enc=json instead") |
| 148 | }), |
| 149 | // DEPRECATED, uncodumented format we used in tests, but not anymore |
| 150 | // <varint-len><message.payload><varint-len><message.payload> |
| 151 | "lenpayload": cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, psm *pubsubMessage) error { |
| 152 | return errors.New("--enc=lenpayload was removed, use --enc=json instead") |
| 153 | }), |
| 154 | }, |
| 155 | Type: pubsubMessage{}, |
| 156 | } |
| 157 | |
| 158 | var PubsubPubCmd = &cmds.Command{ |
| 159 | Status: cmds.Experimental, |
| 160 | Helptext: cmds.HelpText{ |
| 161 | Tagline: "Publish data to a given pubsub topic.", |
| 162 | ShortDescription: ` |
| 163 | ipfs pubsub pub publishes a message to a specified topic. |
| 164 | It reads binary data from stdin or a file. |
| 165 | |
| 166 | EXPERIMENTAL FEATURE |
| 167 | |
| 168 | This is an opt-in feature optimized for IPNS over PubSub |
| 169 | (https://specs.ipfs.tech/ipns/ipns-pubsub-router/). |
| 170 | |
| 171 | To enable, set 'Pubsub.Enabled' config to true. |
| 172 | |
| 173 | HTTP RPC ENCODING |
| 174 | |
| 175 | The data to be published is sent in HTTP request body as multipart/form-data. |
| 176 | |
| 177 | Topic names are binary data too. To ensure all bytes are transferred |
| 178 | correctly via URL params, the RPC client and server will use multibase |
| 179 | encoding behind the scenes. |
| 180 | |
| 181 | `, |
| 182 | }, |
| 183 | Arguments: []cmds.Argument{ |
| 184 | cmds.StringArg("topic", true, false, "Topic to publish to (multibase encoded when sent over HTTP RPC)."), |
| 185 | cmds.FileArg("data", true, false, "The data to be published.").EnableStdin(), |
| 186 | }, |
| 187 | PreRun: urlArgsEncoder, |
| 188 | Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error { |
| 189 | api, err := cmdenv.GetApi(env, req) |
| 190 | if err != nil { |
| 191 | return err |
| 192 | } |
| 193 | if err := urlArgsDecoder(req, env); err != nil { |
| 194 | return err |
| 195 | } |
| 196 | |
| 197 | topic := req.Arguments[0] |
| 198 | |
| 199 | // read data passed as a file |
| 200 | file, err := cmdenv.GetFileArg(req.Files.Entries()) |
| 201 | if err != nil { |
| 202 | return err |
| 203 | } |
| 204 | defer file.Close() |
| 205 | data, err := io.ReadAll(file) |
| 206 | if err != nil { |
| 207 | return err |
| 208 | } |
| 209 | |
| 210 | // publish |
| 211 | return api.PubSub().Publish(req.Context, topic, data) |
| 212 | }, |
| 213 | } |
| 214 | |
| 215 | var PubsubLsCmd = &cmds.Command{ |
| 216 | Status: cmds.Experimental, |
| 217 | Helptext: cmds.HelpText{ |
| 218 | Tagline: "List subscribed topics by name.", |
| 219 | ShortDescription: ` |
| 220 | ipfs pubsub ls lists out the names of topics you are currently subscribed to. |
| 221 | |
| 222 | EXPERIMENTAL FEATURE |
| 223 | |
| 224 | This is an opt-in feature optimized for IPNS over PubSub |
| 225 | (https://specs.ipfs.tech/ipns/ipns-pubsub-router/). |
| 226 | |
| 227 | To enable, set 'Pubsub.Enabled' config to true. |
| 228 | |
| 229 | TOPIC ENCODING |
| 230 | |
| 231 | Topic names are a binary data. To ensure all bytes are transferred |
| 232 | correctly RPC client and server will use multibase encoding behind |
| 233 | the scenes. |
| 234 | |
| 235 | You can inspect the format by passing --enc=json. ipfs multibase commands |
| 236 | can be used for encoding/decoding multibase strings in the userland. |
| 237 | `, |
| 238 | }, |
| 239 | Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error { |
| 240 | api, err := cmdenv.GetApi(env, req) |
| 241 | if err != nil { |
| 242 | return err |
| 243 | } |
| 244 | |
| 245 | l, err := api.PubSub().Ls(req.Context) |
| 246 | if err != nil { |
| 247 | return err |
| 248 | } |
| 249 | |
| 250 | // emit topics encoded in multibase |
| 251 | encoder, _ := mbase.EncoderByName("base64url") |
| 252 | for n, topic := range l { |
| 253 | l[n] = encoder.Encode([]byte(topic)) |
| 254 | } |
| 255 | |
| 256 | return cmds.EmitOnce(res, stringList{l}) |
| 257 | }, |
| 258 | Type: stringList{}, |
| 259 | Encoders: cmds.EncoderMap{ |
| 260 | cmds.Text: cmds.MakeTypedEncoder(multibaseDecodedStringListEncoder), |
| 261 | }, |
| 262 | } |
| 263 | |
| 264 | func multibaseDecodedStringListEncoder(req *cmds.Request, w io.Writer, list *stringList) error { |
| 265 | for n, mb := range list.Strings { |
| 266 | _, data, err := mbase.Decode(mb) |
| 267 | if err != nil { |
| 268 | return err |
| 269 | } |
| 270 | list.Strings[n] = string(data) |
| 271 | } |
| 272 | return safeTextListEncoder(req, w, list) |
| 273 | } |
| 274 | |
| 275 | // converts list of strings to text representation where each string is placed |
| 276 | // in separate line with non-printable/unsafe characters escaped |
| 277 | // (this protects terminal output from being mangled by non-ascii topic names) |
| 278 | func safeTextListEncoder(req *cmds.Request, w io.Writer, list *stringList) error { |
| 279 | for _, str := range list.Strings { |
| 280 | _, err := fmt.Fprintf(w, "%s\n", cmdenv.EscNonPrint(str)) |
| 281 | if err != nil { |
| 282 | return err |
| 283 | } |
| 284 | } |
| 285 | return nil |
| 286 | } |
| 287 | |
| 288 | var PubsubPeersCmd = &cmds.Command{ |
| 289 | Status: cmds.Experimental, |
| 290 | Helptext: cmds.HelpText{ |
| 291 | Tagline: "List peers we are currently pubsubbing with.", |
| 292 | ShortDescription: ` |
| 293 | ipfs pubsub peers with no arguments lists out the pubsub peers you are |
| 294 | currently connected to. If given a topic, it will list connected peers who are |
| 295 | subscribed to the named topic. |
| 296 | |
| 297 | EXPERIMENTAL FEATURE |
| 298 | |
| 299 | This is an opt-in feature optimized for IPNS over PubSub |
| 300 | (https://specs.ipfs.tech/ipns/ipns-pubsub-router/). |
| 301 | |
| 302 | To enable, set 'Pubsub.Enabled' config to true. |
| 303 | |
| 304 | TOPIC AND DATA ENCODING |
| 305 | |
| 306 | Topic names are a binary data. To ensure all bytes are transferred |
| 307 | correctly RPC client and server will use multibase encoding behind |
| 308 | the scenes. |
| 309 | |
| 310 | You can inspect the format by passing --enc=json. ipfs multibase commands |
| 311 | can be used for encoding/decoding multibase strings in the userland. |
| 312 | `, |
| 313 | }, |
| 314 | Arguments: []cmds.Argument{ |
| 315 | cmds.StringArg("topic", false, false, "Topic to list connected peers of."), |
| 316 | }, |
| 317 | PreRun: urlArgsEncoder, |
| 318 | Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error { |
| 319 | api, err := cmdenv.GetApi(env, req) |
| 320 | if err != nil { |
| 321 | return err |
| 322 | } |
| 323 | if err := urlArgsDecoder(req, env); err != nil { |
| 324 | return err |
| 325 | } |
| 326 | |
| 327 | var topic string |
| 328 | if len(req.Arguments) == 1 { |
| 329 | topic = req.Arguments[0] |
| 330 | } |
| 331 | |
| 332 | peers, err := api.PubSub().Peers(req.Context, options.PubSub.Topic(topic)) |
| 333 | if err != nil { |
| 334 | return err |
| 335 | } |
| 336 | |
| 337 | list := &stringList{make([]string, 0, len(peers))} |
| 338 | |
| 339 | for _, peer := range peers { |
| 340 | list.Strings = append(list.Strings, peer.String()) |
| 341 | } |
| 342 | slices.Sort(list.Strings) |
| 343 | return cmds.EmitOnce(res, list) |
| 344 | }, |
| 345 | Type: stringList{}, |
| 346 | Encoders: cmds.EncoderMap{ |
| 347 | cmds.Text: cmds.MakeTypedEncoder(safeTextListEncoder), |
| 348 | }, |
| 349 | } |
| 350 | |
| 351 | // TODO: move to cmdenv? |
| 352 | // Encode binary data to be passed as multibase string in URL arguments. |
| 353 | // (avoiding issues described in https://github.com/ipfs/kubo/issues/7939) |
| 354 | func urlArgsEncoder(req *cmds.Request, env cmds.Environment) error { |
| 355 | encoder, _ := mbase.EncoderByName("base64url") |
| 356 | for n, arg := range req.Arguments { |
| 357 | req.Arguments[n] = encoder.Encode([]byte(arg)) |
| 358 | } |
| 359 | return nil |
| 360 | } |
| 361 | |
| 362 | // Decode binary data passed as multibase string in URL arguments. |
| 363 | // (avoiding issues described in https://github.com/ipfs/kubo/issues/7939) |
| 364 | func urlArgsDecoder(req *cmds.Request, env cmds.Environment) error { |
| 365 | for n, arg := range req.Arguments { |
| 366 | encoding, data, err := mbase.Decode(arg) |
| 367 | if err != nil { |
| 368 | return fmt.Errorf("URL arg must be multibase encoded: %w", err) |
| 369 | } |
| 370 | |
| 371 | // Enforce URL-safe encoding is used for data passed via URL arguments |
| 372 | // - without this we get data corruption similar to https://github.com/ipfs/kubo/issues/7939 |
| 373 | // - we can't just deny base64, because there may be other bases that |
| 374 | // are not URL-safe – better to force base64url which is known to be |
| 375 | // safe in URL context |
| 376 | if encoding != mbase.Base64url { |
| 377 | return errors.New("URL arg must be base64url encoded") |
| 378 | } |
| 379 | |
| 380 | req.Arguments[n] = string(data) |
| 381 | } |
| 382 | return nil |
| 383 | } |
| 384 | |
| 385 | type pubsubResetResult struct { |
| 386 | Deleted int64 `json:"deleted"` |
| 387 | } |
| 388 | |
| 389 | var PubsubResetCmd = &cmds.Command{ |
| 390 | Status: cmds.Experimental, |
| 391 | Helptext: cmds.HelpText{ |
| 392 | Tagline: "Reset pubsub validator state.", |
| 393 | ShortDescription: ` |
| 394 | Clears persistent sequence number state used by the pubsub validator. |
| 395 | |
| 396 | WARNING: FOR TESTING ONLY - DO NOT USE IN PRODUCTION |
| 397 | |
| 398 | Resets validator state that protects against replay attacks. After reset, |
| 399 | previously seen messages may be accepted again until their sequence numbers |
| 400 | are re-learned. |
| 401 | |
| 402 | Use cases: |
| 403 | - Testing pubsub functionality |
| 404 | - Recovery from a peer sending artificially high sequence numbers |
| 405 | (which would cause subsequent messages from that peer to be rejected) |
| 406 | |
| 407 | The --peer flag limits the reset to a specific peer's state. |
| 408 | Without --peer, all validator state is cleared. |
| 409 | |
| 410 | NOTE: This only resets the persistent seqno validator state. The in-memory |
| 411 | seen messages cache (Pubsub.SeenMessagesTTL) auto-expires and can only be |
| 412 | fully cleared by restarting the daemon. |
| 413 | `, |
| 414 | }, |
| 415 | Options: []cmds.Option{ |
| 416 | cmds.StringOption(peerOptionName, "p", "Only reset state for this peer ID"), |
| 417 | }, |
| 418 | Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error { |
| 419 | n, err := cmdenv.GetNode(env) |
| 420 | if err != nil { |
| 421 | return err |
| 422 | } |
| 423 | |
| 424 | ds := n.Repo.Datastore() |
| 425 | ctx := req.Context |
| 426 | |
| 427 | peerOpt, _ := req.Options[peerOptionName].(string) |
| 428 | |
| 429 | var deleted int64 |
| 430 | if peerOpt != "" { |
| 431 | // Reset specific peer |
| 432 | pid, err := peer.Decode(peerOpt) |
| 433 | if err != nil { |
| 434 | return fmt.Errorf("invalid peer ID: %w", err) |
| 435 | } |
| 436 | key := datastore.NewKey(libp2p.SeqnoStorePrefix + pid.String()) |
| 437 | exists, err := ds.Has(ctx, key) |
| 438 | if err != nil { |
| 439 | return fmt.Errorf("failed to check seqno state: %w", err) |
| 440 | } |
| 441 | if exists { |
| 442 | if err := ds.Delete(ctx, key); err != nil { |
| 443 | return fmt.Errorf("failed to delete seqno state: %w", err) |
| 444 | } |
| 445 | deleted = 1 |
| 446 | } |
| 447 | } else { |
| 448 | // Reset all peers using batched delete for efficiency |
| 449 | q := query.Query{ |
| 450 | Prefix: libp2p.SeqnoStorePrefix, |
| 451 | KeysOnly: true, |
| 452 | } |
| 453 | results, err := ds.Query(ctx, q) |
| 454 | if err != nil { |
| 455 | return fmt.Errorf("failed to query seqno state: %w", err) |
| 456 | } |
| 457 | defer results.Close() |
| 458 | |
| 459 | batch, err := ds.Batch(ctx) |
| 460 | if err != nil { |
| 461 | return fmt.Errorf("failed to create batch: %w", err) |
| 462 | } |
| 463 | |
| 464 | for result := range results.Next() { |
| 465 | if result.Error != nil { |
| 466 | return fmt.Errorf("query error: %w", result.Error) |
| 467 | } |
| 468 | if err := batch.Delete(ctx, datastore.NewKey(result.Key)); err != nil { |
| 469 | return fmt.Errorf("failed to batch delete key %s: %w", result.Key, err) |
| 470 | } |
| 471 | deleted++ |
| 472 | } |
| 473 | |
| 474 | if err := batch.Commit(ctx); err != nil { |
| 475 | return fmt.Errorf("failed to commit batch delete: %w", err) |
| 476 | } |
| 477 | } |
| 478 | |
| 479 | // Sync to ensure deletions are persisted |
| 480 | if err := ds.Sync(ctx, datastore.NewKey(libp2p.SeqnoStorePrefix)); err != nil { |
| 481 | return fmt.Errorf("failed to sync datastore: %w", err) |
| 482 | } |
| 483 | |
| 484 | return cmds.EmitOnce(res, &pubsubResetResult{Deleted: deleted}) |
| 485 | }, |
| 486 | Type: pubsubResetResult{}, |
| 487 | Encoders: cmds.EncoderMap{ |
| 488 | cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, result *pubsubResetResult) error { |
| 489 | peerOpt, _ := req.Options[peerOptionName].(string) |
| 490 | if peerOpt != "" { |
| 491 | if result.Deleted == 0 { |
| 492 | _, err := fmt.Fprintf(w, "No validator state found for peer %s\n", peerOpt) |
| 493 | return err |
| 494 | } |
| 495 | _, err := fmt.Fprintf(w, "Reset validator state for peer %s\n", peerOpt) |
| 496 | return err |
| 497 | } |
| 498 | _, err := fmt.Fprintf(w, "Reset validator state for %d peer(s)\n", result.Deleted) |
| 499 | return err |
| 500 | }), |
| 501 | }, |
| 502 | } |