diff --git a/src/whoseport/cli.py b/src/whoseport/cli.py index 6c65732..bf4fd2f 100644 --- a/src/whoseport/cli.py +++ b/src/whoseport/cli.py @@ -5,7 +5,6 @@ import argparse import json import os -import re import shutil import sys from collections import Counter @@ -13,14 +12,13 @@ import whoseport from whoseport import core, kill from whoseport.core import BackendError, PortUser, UnsupportedPlatformError +from whoseport.tokens import parse_port_tokens EXIT_OK = 0 EXIT_NOTHING_FOUND = 1 EXIT_USAGE = 2 EXIT_KILL_FAILED = 3 -_PORT_TOKEN = re.compile(r"^(\d{1,5})(?:-(\d{1,5}))?$") - # ---------------------------------------------------------------- colors --- @@ -93,22 +91,6 @@ def sym(fancy: str, plain: str) -> str: # ---------------------------------------------------------------- parsing --- -def parse_port_tokens(tokens: list) -> set: - ports = set() - for token in tokens: - m = _PORT_TOKEN.match(token) - if not m: - raise ValueError(f"invalid port or range: {token!r} (examples: 8080, 3000-3005)") - start = int(m.group(1)) - end = int(m.group(2)) if m.group(2) else start - if not (1 <= start <= 65535 and 1 <= end <= 65535): - raise ValueError(f"port out of range 1-65535: {token!r}") - if end < start: - raise ValueError(f"range end before start: {token!r}") - ports.update(range(start, end + 1)) - return ports - - def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="whoseport", diff --git a/src/whoseport/tokens.py b/src/whoseport/tokens.py new file mode 100644 index 0000000..5ec97fa --- /dev/null +++ b/src/whoseport/tokens.py @@ -0,0 +1,61 @@ +"""Port token parsing: ranges, host:port, and URLs.""" + +from __future__ import annotations + +import re +from urllib.parse import urlparse + +_PORT_TOKEN = re.compile(r"^(\d{1,5})(?:-(\d{1,5}))?$") + + +def _strip_port_suffix(port_s: str) -> str: + """Drop path / query / fragment so ``host:8080/health#x`` still parses.""" + return port_s.split("/", 1)[0].split("?", 1)[0].split("#", 1)[0] + + +def port_from_endpoint(token: str) -> int | None: + """Pull a port out of ``host:port``, ``[ipv6]:port`` or a URL. + + All-digit left-hand sides such as ``8080:8090`` stay rejected so a + mistyped range is not silently turned into a single port. + """ + text = token.strip() + if "://" in text: + try: + port = urlparse(text).port + except ValueError: + return None + return int(port) if port else None + if text.startswith("[") and "]:" in text: + port_s = _strip_port_suffix(text.rsplit("]:", 1)[-1]) + return int(port_s) if port_s.isdigit() else None + if ":" in text and not text.startswith(":"): + host, _, port_s = text.rpartition(":") + port_s = _strip_port_suffix(port_s) + if host and not host.isdigit() and port_s.isdigit(): + return int(port_s) + return None + + +def parse_port_tokens(tokens: list) -> set: + ports = set() + for token in tokens: + extracted = port_from_endpoint(token) + if extracted is not None: + if not 1 <= extracted <= 65535: + raise ValueError(f"port out of range 1-65535: {token!r}") + ports.add(extracted) + continue + m = _PORT_TOKEN.match(token) + if not m: + raise ValueError( + f"invalid port or range: {token!r} (examples: 8080, 3000-3005, localhost:8080)" + ) + start = int(m.group(1)) + end = int(m.group(2)) if m.group(2) else start + if not (1 <= start <= 65535 and 1 <= end <= 65535): + raise ValueError(f"port out of range 1-65535: {token!r}") + if end < start: + raise ValueError(f"range end before start: {token!r}") + ports.update(range(start, end + 1)) + return ports diff --git a/tests/test_host_port.py b/tests/test_host_port.py new file mode 100644 index 0000000..9b00335 --- /dev/null +++ b/tests/test_host_port.py @@ -0,0 +1,38 @@ +import json + +from whoseport import cli, core +from whoseport.tokens import parse_port_tokens + + +def test_parse_host_port_and_url(): + assert parse_port_tokens(["localhost:8080"]) == {8080} + assert parse_port_tokens(["127.0.0.1:3000"]) == {3000} + assert parse_port_tokens(["[::1]:5173"]) == {5173} + assert parse_port_tokens(["http://127.0.0.1:4173/app"]) == {4173} + assert parse_port_tokens(["localhost:8080/health"]) == {8080} + assert parse_port_tokens(["[::1]:5173/app"]) == {5173} + assert parse_port_tokens(["localhost:8080#ready"]) == {8080} + assert parse_port_tokens(["[::1]:5173/app?x=1#top"]) == {5173} + + +def test_numeric_colon_range_still_rejected(): + try: + parse_port_tokens(["8080:8090"]) + except ValueError as exc: + assert "invalid" in str(exc) + else: + raise AssertionError("8080:8090 must stay rejected") + + +def test_cli_reexports_token_parser(): + assert cli.parse_port_tokens is parse_port_tokens + assert cli.parse_port_tokens(["localhost:8080", "3000-3001"]) == {8080, 3000, 3001} + + +def test_cli_accepts_host_port(monkeypatch, capsys, make_row): + monkeypatch.setattr(core, "collect", lambda **kw: [make_row(local_port=8080)]) + rc = cli.main(["localhost:8080", "--json"]) + doc = json.loads(capsys.readouterr().out) + assert rc == 0 + assert doc["query"] == [8080] + assert doc["sockets"][0]["local_port"] == 8080