tools: Implement tool to send a bundle via MTCP

This is handy e.g. for testing custom source EIDs, not requiring an
additional BPA instance to be run for injecting bundles.

Signed-off-by: Felix Walter <felix.walter@d3tn.com>

.gitlab-ci.yml

@@ -415,6 +415,9 @@ mtcp-test:
         - while ! nc -z localhost 4224; do sleep 0.1; done
         - python tools/cla/mtcp_test.py -t mtcp --payload TESTPAYLOAD &
         - timeout -v 5 python tools/cla/mtcp_sink.py --count 1 --verify-pl TESTPAYLOAD
+        - sleep 2
+        - (sleep 1 && python tools/cla/mtcp_send_bundle.py dtn://ud3tn.dtn/sink hello) &
+        - timeout -v 3 aap2-receive -a sink -c 1 --verify-pl hello
         - sleep 0.2
         - kill -TERM $UD3TN_PID
         - echo "Waiting for uD3TN to exit gracefully - if it doesn't, check for sanitizer warnings."

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - Implement `ipn` scheme update (RFC 9758)
+- Implement a tool to inject bundles via MTCP (`tools/cla/mtcp_send_bundle.py`)
 
 ### Deprecated
 

tools/cla/mtcp_send_bundle.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: BSD-3-Clause OR Apache-2.0
+# encoding: utf-8
+
+"""
+A tool to send custom bundles via the MTCP CLA to a specified DTN node.
+"""
+
+import argparse
+import sys
+
+from pyd3tn.bundle7 import serialize_bundle7, BundleProcFlag
+from pyd3tn.bundle6 import serialize_bundle6, RFC5050Flag
+from pyd3tn.eid import validate_eid
+from pyd3tn.mtcp import MTCPConnection
+
+# The MUST_NOT_BE_FRAGMENTED flag is required if the source EID is dtn:none.
+BPV7_FLAGS = BundleProcFlag.MUST_NOT_BE_FRAGMENTED
+BPV6_FLAGS = RFC5050Flag.DEFAULT_OUTGOING | RFC5050Flag.MUST_NOT_BE_FRAGMENTED
+
+
+def _argparse_non_empty_str(value):
+    """An argparse type validation function to exclude the empty string."""
+    if value is None or value.strip() == "":
+        raise argparse.ArgumentTypeError("must not be empty")
+    return value
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="send a bundle via uD3TN's MTCP interface",
+    )
+    parser.add_argument(
+        "dst_eid",
+        type=_argparse_non_empty_str,
+        help="the destination EID of the created bundle",
+    )
+    parser.add_argument(
+        "PAYLOAD",
+        default=None,
+        nargs="?",
+        help="the payload of the created bundle, (default: read from STDIN)",
+    )
+    parser.add_argument(
+        "-l", "--host",
+        default="127.0.0.1",
+        help="MTCP host or IP address to connect to (default: 127.0.0.1)",
+    )
+    parser.add_argument(
+        "-p", "--port",
+        type=int,
+        default=4224,
+        help="MTCP port to connect to (default: 4224)",
+    )
+    parser.add_argument(
+        "-s", "--src-eid",
+        default="dtn:none",
+        help="EID of sender (default: dtn:none)",
+    )
+    parser.add_argument(
+        "-b", "--bundle-version",
+        default="7",
+        choices=["6", "7"],
+        help="Version of the bundle protocol to use (defaults to 7)",
+    )
+    parser.add_argument(
+        "--timeout",
+        type=int, default=3000,
+        help="TCP timeout in ms (default: 3000)",
+    )
+
+    args = parser.parse_args()
+
+    serialize_bundle = {
+        "6": serialize_bundle6,
+        "7": serialize_bundle7,
+    }[args.bundle_version]
+
+    if args.PAYLOAD:
+        payload = args.PAYLOAD.encode("utf-8")
+    else:
+        payload = sys.stdin.buffer.read()
+        sys.stdin.buffer.close()
+
+    validate_eid(args.dst_eid)
+
+    with MTCPConnection(args.host, args.port, timeout=args.timeout) as conn:
+        conn.send_bundle(serialize_bundle(
+            args.src_eid,
+            args.dst_eid,
+            payload,
+            flags=(
+                BPV7_FLAGS
+                if args.bundle_version == "7"
+                else BPV6_FLAGS
+            ),
+        ))
+
+
+if __name__ == "__main__":
+    main()