mirror of
https://gitlab.com/d3tn/ud3tn.git
synced 2026-08-13 12:33:27 +02:00
afdd9fbdfb
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>
101 lines
2.7 KiB
Python
101 lines
2.7 KiB
Python
#!/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()
|