-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
hor 1204 adds support for ranges ip addresses (#28)
* rebase to main * add support for coverage in make.pyz * create a flags examples * fix mypy config * pre-release changelog * moved load_ips_from_csv definition in luxos.ips * add few extra flags * silence missing imports * fix mypy * fix mypy * fix windows failing test * report session info as debug instead info log level
- Loading branch information
Showing
17 changed files
with
328 additions
and
45 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| """various argparse `type` attributes""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import argparse | ||
| import contextlib | ||
| import datetime | ||
|
|
||
|
|
||
| def type_range(txt: str) -> list[tuple[str, int | None]]: | ||
| """type conversion for ranges | ||
| This will enforce conversion between a string and a ranged object. | ||
| Eg. | ||
| parser.add_argument("--range", type=type_range) | ||
| The --range argument will be: | ||
| 127.0.0.1 # single ip address | ||
| 127.0.0.1:1234 # single ip address with port | ||
| 127.0.0.1-127.0.0.3 # a list of (ip, port) tuple between *.1 and.3 | ||
| """ | ||
| from luxos.ips import iter_ip_ranges | ||
|
|
||
| try: | ||
| return list(iter_ip_ranges(txt)) | ||
| except RuntimeError as exc: | ||
| raise argparse.ArgumentTypeError(f"conversion failed '{txt}': {exc.args[0]}") | ||
| except Exception as exc: | ||
| raise argparse.ArgumentTypeError(f"conversion failed for {txt}") from exc | ||
|
|
||
|
|
||
| def type_hhmm(txt: str): | ||
| """type conversion for ranges | ||
| This will enforce conversion between a string and datetime.time object. | ||
| Eg. | ||
| parser.add_argument("--time", type=type_hhmm) | ||
| The --time format is HH:MM | ||
| """ | ||
| if not txt: | ||
| return | ||
| with contextlib.suppress(ValueError, TypeError): | ||
| hh, _, mm = txt.partition(":") | ||
| hh1 = int(hh) | ||
| mm1 = int(mm) | ||
| return datetime.time(hh1, mm1) | ||
| raise argparse.ArgumentTypeError(f"failed conversion into HH:MM for '{txt}'") | ||
|
|
||
|
|
||
| def add_arguments_rexec(parser: argparse.ArgumentParser) -> None: | ||
| group = parser.add_argument_group( | ||
| "Remote execution", "rexec remote execution limits/timeouts" | ||
| ) | ||
| group.add_argument( | ||
| "--timeout", type=float, default=3.0, help="Timeout for each command" | ||
| ) | ||
| group.add_argument( | ||
| "--max-retries", | ||
| type=int, | ||
| default=3, | ||
| help="Maximum number of retries for each command", | ||
| ) | ||
| group.add_argument( | ||
| "--delay-retry", type=float, default=3.0, help="Delay in s between retries" | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| """ipaddress manipulation""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import ipaddress | ||
| import re | ||
| from pathlib import Path | ||
| from typing import Generator | ||
|
|
||
|
|
||
| def splitip(txt: str) -> tuple[str, int | None]: | ||
| expr = re.compile(r"(?P<ip>\d{1,3}([.]\d{1,3}){3})(:(?P<port>\d+))?") | ||
| if not (match := expr.search(txt)): | ||
| raise RuntimeError(f"invalid ip:port address {txt}") | ||
| return match["ip"], int(match["port"]) if match["port"] is not None else None | ||
|
|
||
|
|
||
| def iter_ip_ranges( | ||
| txt: str, port: int | None = None, rsep: str = "-", gsep: str = "," | ||
| ) -> Generator[tuple[str, int | None], None, None]: | ||
| """iterate over ip ranges. | ||
| The txt string cav have one of these formats: | ||
| 1. a single ip such as '127.0.0.1' or '127.0.0.1:8080' | ||
| 2. an (inclusive) range using two ips separated by `-` | ||
| as '127.0.0.1 - 127.0.0.3' | ||
| 3. a combination of the above `,` separated as | ||
| '127.0.0.1 , 192.168.0.1-192.168.0.10' | ||
| Example: | ||
| ```python | ||
| for ip in iter_ip_ranges("127.0.0.1 , 127.0.0.3-127.0.0.15"): | ||
| print(ip) | ||
| 127.0.0.1 | ||
| 127.0.0.2 | ||
| 127.0.0.3 | ||
| ... | ||
| 127.0.0.15 | ||
| ``` | ||
| """ | ||
| for segment in txt.replace(" ", "").split(gsep): | ||
| start, _, end = segment.partition(rsep) | ||
| if not end: | ||
| start, sport = splitip(start) | ||
| yield (start, sport or port) | ||
| else: | ||
| start, sport = splitip(start) | ||
| end, eport = splitip(end) | ||
| if (sport and eport) and (sport != eport): | ||
| raise RuntimeError(f"invalid range ports in {segment}") | ||
| cur = ipaddress.IPv4Address(start) | ||
| last = ipaddress.IPv4Address(end) | ||
| theport = sport or eport or port | ||
| while cur <= last: | ||
| yield (str(cur), theport) | ||
| cur += 1 | ||
|
|
||
|
|
||
| def load_ips_from_csv(path: Path | str, port: int = 4028) -> list[tuple[str, int]]: | ||
| """loads ip addresses from a csv file | ||
| Example: | ||
| ```python | ||
| foobar.csv contains ranges as parsed by iter_ip_ranges | ||
| 127.0.0.1 # a single address | ||
| 127.0.0.2-127.0.0.10 | ||
| for ip in load_ips_from_csv("foobar.csv"): | ||
| print(ip) | ||
| (127.0.0.1, 4028) | ||
| (127.0.0.2, 4028) | ||
| (127.0.0.3, 4028) | ||
| ... | ||
| (127.0.0.10, 4028) | ||
| ``` | ||
| """ | ||
| result = [] | ||
| for line in Path(path).read_text().split("\n"): | ||
| line = line.partition("#")[0] | ||
| if not line.strip(): | ||
| continue | ||
| for host, port2 in iter_ip_ranges(line): | ||
| result.append((host, port2 or port)) | ||
| return result |
Oops, something went wrong.