diff --git a/pkgs/additional/sane-scripts/src/lib/__pycache__/sane_ssdp.cpython-310.pyc b/pkgs/additional/sane-scripts/src/lib/__pycache__/sane_ssdp.cpython-310.pyc new file mode 100644 index 00000000..fd3b52ce Binary files /dev/null and b/pkgs/additional/sane-scripts/src/lib/__pycache__/sane_ssdp.cpython-310.pyc differ diff --git a/pkgs/additional/sane-scripts/src/lib/sane_ssdp.py b/pkgs/additional/sane-scripts/src/lib/sane_ssdp.py index b86e6d48..6633cc03 100644 --- a/pkgs/additional/sane-scripts/src/lib/sane_ssdp.py +++ b/pkgs/additional/sane-scripts/src/lib/sane_ssdp.py @@ -66,16 +66,20 @@ def get_wan_from_location(location: str): """ location = URI from the Location header, e.g. http://10.78.79.1:2189/rootDesc.xml """ # get connection [s]tatus - res = subprocess.run(["upnpc", "-u", location, "-s"], capture_output=True) - res.check_returncode() + cmd = ["upnpc", "-u", location, "-s"] + res = subprocess.run(cmd, capture_output=True) + if res.returncode != 0: + logger.info(f"get_wan_from_location failed: {cmd!r}\n{res.stderr}") + return None status = res.stdout.decode("utf-8") - logger.info(f"got status: {status}") + logger.debug(f"got status: {status}") for line in [l.strip() for l in status.split("\n")]: sentinel = "ExternalIPAddress =" if line.startswith(sentinel): ip = line[len(sentinel):].strip() + logger.info(f"got ExternalIPAddress = {ip} from {location}") return ip def get_any_wan(): diff --git a/pkgs/additional/sane-scripts/src/sane-ip-port-forward b/pkgs/additional/sane-scripts/src/sane-ip-port-forward index f4bc6661..b2bc2cf7 100755 --- a/pkgs/additional/sane-scripts/src/sane-ip-port-forward +++ b/pkgs/additional/sane-scripts/src/sane-ip-port-forward @@ -1,5 +1,5 @@ #!/usr/bin/env nix-shell -#!nix-shell -i python3 -p "python3.withPackages (ps: [ ])" -p miniupnpc +#!nix-shell -i python3 -p "python3.withPackages (ps: [ ])" -p inetutils -p miniupnpc ''' USAGE: sane-ip-port-forward [options] [proto:port]* @@ -8,12 +8,15 @@ options: -v: verbose (show info messages) -vv: more verbose (show debug messages) -h: show this help messages + -d : lease for the given duration in seconds (default: {DEFAULT_LEASE_SEC}) proto:port: proto is `udp` or `tcp` (case insensitive) port is any integer 1-65535 inclusive ''' +DEFAULT_LEASE_SEC = 86400 + import logging import subprocess import sys @@ -24,7 +27,7 @@ from lib.sane_ssdp import get_any_wan, forward_port class BadCliArgs(Exception): def __init__(self, msg: str = None): - helpstr = __doc__.strip() + helpstr = __doc__.format(DEFAULT_LEASE_SEC=DEFAULT_LEASE_SEC).strip() if msg: super().__init__(f"{msg}\n\n{helpstr}") else: @@ -45,25 +48,40 @@ def try_parse_port(s: str): pass def parse_args(argv: "List[str]") -> "List[('udp'|'tcp', port)]": + """ + returns (list of forwards, lease duration) + where: + - list of forwards is [('udp'|'tcp', port: int)] + - lease duration is seconds: int + """ forwards = [] - for arg in sys.argv[1:]: + duration = DEFAULT_LEASE_SEC + unparsed = sys.argv[1:][::-1] + while unparsed: + arg = unparsed.pop() if arg == "-h": raise BadCliArgs() if arg == "-v": logging.getLogger().setLevel(logging.INFO) elif arg == "-vv": logging.getLogger().setLevel(logging.DEBUG) + elif arg == "-d" and unparsed: + d = unparsed.pop() + try: + duration = int(d) + except Exception: + raise BadCliArgs(f"invalid CLI argument: -d {d!r}") elif try_parse_port(arg): forwards.append(try_parse_port(arg)) else: - raise BadCliArgs(f"invalid CLI argument {arg!r}") - return forwards + raise BadCliArgs(f"invalid CLI argument: {arg!r}") + return forwards, duration if __name__ == '__main__': logging.basicConfig() try: - forwards = parse_args(sys.argv) + forwards, duration = parse_args(sys.argv) except BadCliArgs as e: print(e) sys.exit(1) @@ -71,4 +89,4 @@ if __name__ == '__main__': root_device, _wan = get_any_wan() hostname = subprocess.check_output(["hostname"]).decode("utf-8").strip() for (proto, port) in forwards: - forward_port(root_device, proto, port, f"colin-{hostname}") + forward_port(root_device, proto, port, f"colin-{hostname}", duration=duration)