diff --git a/icmplib/models.py b/icmplib/models.py index c055fdf..90711e6 100644 --- a/icmplib/models.py +++ b/icmplib/models.py @@ -26,6 +26,7 @@ . ''' +from typing import List, Optional from .exceptions import * from .utils import random_byte_message @@ -74,8 +75,16 @@ class ICMPRequest: __slots__ = '_destination', '_id', '_sequence', '_payload', \ '_payload_size', '_ttl', '_traffic_class', '_time' - def __init__(self, destination, id, sequence, payload=None, - payload_size=56, ttl=64, traffic_class=0): + def __init__( + self, + destination: str, + id: int, + sequence: int, + payload: Optional[bytes] = None, + payload_size: int = 56, + ttl: int = 64, + traffic_class: int = 0 + ): if payload: payload_size = len(payload) @@ -199,8 +208,17 @@ class ICMPReply: __slots__ = '_source', '_family', '_id', '_sequence', '_type', \ '_code', '_bytes_received', '_time' - def __init__(self, source, family, id, sequence, type, code, - bytes_received, time): + def __init__( + self, + source: str, + family: int, + id: int, + sequence: int, + type: int, + code: int, + bytes_received: int, + time: float + ): self._source = source self._family = family @@ -323,7 +341,12 @@ class Host: ''' __slots__ = '_address', '_packets_sent', '_rtts' - def __init__(self, address, packets_sent, rtts): + def __init__( + self, + address: str, + packets_sent: int, + rtts: List[float] + ): self._address = address self._packets_sent = packets_sent self._rtts = rtts @@ -471,7 +494,13 @@ class Hop(Host): ''' __slots__ = '_address', '_packets_sent', '_rtts', '_distance' - def __init__(self, address, packets_sent, rtts, distance): + def __init__( + self, + address: str, + packets_sent: int, + rtts: List[float], + distance: int + ): super().__init__(address, packets_sent, rtts) self._distance = distance diff --git a/icmplib/multiping.py b/icmplib/multiping.py index 0705b28..efbecdd 100644 --- a/icmplib/multiping.py +++ b/icmplib/multiping.py @@ -27,13 +27,22 @@ ''' import asyncio +from typing import List, Optional from .ping import async_ping -async def async_multiping(addresses, count=2, interval=0.5, timeout=2, - concurrent_tasks=50, source=None, family=None, privileged=True, - **kwargs): +async def async_multiping( + addresses: List[str], + count: int = 2, + interval: float = 0.5, + timeout: float = 2, + concurrent_tasks: int = 50, + source: Optional[str] = None, + family: Optional[int] = None, + privileged: Optional[bool] = True, + **kwargs + ): ''' Send ICMP Echo Request packets to several network hosts. @@ -163,9 +172,17 @@ async def async_multiping(addresses, count=2, interval=0.5, timeout=2, return [task.result() for task in tasks] -def multiping(addresses, count=2, interval=0.5, timeout=2, - concurrent_tasks=50, source=None, family=None, privileged=True, - **kwargs): +def multiping( + addresses: List[str], + count: int = 2, + interval: float = 0.5, + timeout: float = 2, + concurrent_tasks: int = 50, + source: Optional[str] = None, + family: Optional[int] = None, + privileged: Optional[bool] = True, + **kwargs + ): ''' Send ICMP Echo Request packets to several network hosts. diff --git a/icmplib/ping.py b/icmplib/ping.py index 5ba1efc..c8ddf71 100644 --- a/icmplib/ping.py +++ b/icmplib/ping.py @@ -28,6 +28,7 @@ import asyncio from time import sleep +from typing import Optional from .sockets import ICMPv4Socket, ICMPv6Socket, AsyncSocket from .models import ICMPRequest, Host @@ -35,8 +36,17 @@ from .utils import * -def ping(address, count=4, interval=1, timeout=2, id=None, source=None, - family=None, privileged=True, **kwargs): +def ping( + address: str, + count: int = 4, + interval: float = 1, + timeout: float = 2, + id: Optional[int] = None, + source: Optional[str] = None, + family: Optional[int] = None, + privileged: Optional[bool] = True, + **kwargs + ): ''' Send ICMP Echo Request packets to a network host. @@ -165,8 +175,10 @@ def ping(address, count=4, interval=1, timeout=2, id=None, source=None, return Host(address, packets_sent, rtts) -async def async_ping(address, count=4, interval=1, timeout=2, id=None, - source=None, family=None, privileged=True, **kwargs): +async def async_ping(address: str, count: int = 4, interval: float = 1, + timeout: float = 2, id: Optional[int] = None, + source: Optional[str] = None, family: Optional[int] = None, + privileged: Optional[bool] = True, **kwargs): ''' Send ICMP Echo Request packets to a network host. diff --git a/icmplib/sockets.py b/icmplib/sockets.py index 130f55a..8b2768b 100644 --- a/icmplib/sockets.py +++ b/icmplib/sockets.py @@ -29,8 +29,9 @@ import socket, asyncio from struct import pack, unpack from time import time +from typing import Optional -from .models import ICMPReply +from .models import ICMPReply, ICMPRequest from .exceptions import * from .utils import PLATFORM_LINUX, PLATFORM_MACOS, PLATFORM_WINDOWS @@ -74,7 +75,7 @@ class ICMPSocket: _ICMP_ECHO_REQUEST = -1 _ICMP_ECHO_REPLY = -1 - def __init__(self, address=None, privileged=True): + def __init__(self, address: Optional[str] = None, privileged: Optional[bool] = True): self._sock = None self._address = address @@ -233,7 +234,7 @@ def _parse_reply(self, packet, source, current_time): bytes_received=bytes_received, time=current_time) - def send(self, request): + def send(self, request: ICMPRequest): ''' Send an ICMP request message over the network to a remote host. @@ -287,7 +288,7 @@ def send(self, request): except OSError as err: raise ICMPSocketError(str(err)) - def receive(self, request=None, timeout=2): + def receive(self, request: Optional[ICMPRequest] = None, timeout: float = 2): ''' Receive an ICMP reply message from the socket. @@ -532,7 +533,7 @@ def broadcast(self): socket.SO_BROADCAST) > 0 @broadcast.setter - def broadcast(self, enable): + def broadcast(self, enable: bool): self._sock.setsockopt( socket.SOL_SOCKET, socket.SO_BROADCAST, @@ -666,7 +667,7 @@ class AsyncSocket: ''' __slots__ = '_icmp_sock' - def __init__(self, icmp_sock): + def __init__(self, icmp_sock: ICMPSocket): self._icmp_sock = icmp_sock self._icmp_sock.blocking = False @@ -701,7 +702,7 @@ def __del__(self): ''' self.close() - async def receive(self, request=None, timeout=2): + async def receive(self, request: Optional[ICMPRequest] = None, timeout: float = 2): ''' Receive an ICMP reply message from the socket. diff --git a/icmplib/traceroute.py b/icmplib/traceroute.py index 7dd45f8..a55c515 100644 --- a/icmplib/traceroute.py +++ b/icmplib/traceroute.py @@ -27,6 +27,7 @@ ''' from time import sleep +from typing import Optional from .sockets import ICMPv4Socket, ICMPv6Socket from .models import ICMPRequest, Hop @@ -34,9 +35,19 @@ from .utils import * -def traceroute(address, count=2, interval=0.05, timeout=2, first_hop=1, - max_hops=30, fast=False, id=None, source=None, family=None, - **kwargs): +def traceroute( + address: str, + count: int = 2, + interval: float = 0.05, + timeout: float = 2, + first_hop: int = 1, + max_hops: int = 30, + fast: Optional[bool] = False, + id: Optional[int] = None, + source: Optional[str] = None, + family: Optional[int] = None, + **kwargs + ): ''' Determine the route to a destination host. diff --git a/icmplib/utils.py b/icmplib/utils.py index 00e2664..bfebe86 100644 --- a/icmplib/utils.py +++ b/icmplib/utils.py @@ -33,6 +33,7 @@ from os import getpid from re import match from random import choices +from typing import Optional from .exceptions import NameLookupError @@ -46,7 +47,7 @@ _current_id = PID -def random_byte_message(size): +def random_byte_message(size: int): ''' Generate a random byte sequence of the specified size. @@ -74,7 +75,7 @@ def unique_identifier(): return _current_id -def resolve(name, family=None): +def resolve(name, family: Optional[int] = None): ''' Resolve a hostname or FQDN to an IP address. Depending on the name specified in parameters, several IP addresses may be returned. @@ -120,7 +121,7 @@ def resolve(name, family=None): raise NameLookupError(name) -async def async_resolve(name, family=None): +async def async_resolve(name, family: Optional[int] = None): ''' Resolve a hostname or FQDN to an IP address. Depending on the name specified in parameters, several IP addresses may be returned. @@ -170,7 +171,7 @@ async def async_resolve(name, family=None): raise NameLookupError(name) -def is_hostname(name): +def is_hostname(name: str): ''' Indicate whether the specified name is a hostname or an FQDN. Return a `boolean`. @@ -180,7 +181,7 @@ def is_hostname(name): return match(pattern, name) is not None -def is_ipv4_address(address): +def is_ipv4_address(address: str): ''' Indicate whether the specified address is an IPv4 address. Return a `boolean`. @@ -190,7 +191,7 @@ def is_ipv4_address(address): return match(pattern, address) is not None -def is_ipv6_address(address): +def is_ipv6_address(address: str): ''' Indicate whether the specified address is an IPv6 address. Return a `boolean`.